import asyncio
import concurrent.futures
import threading
from unittest.mock import AsyncMock, MagicMock, patch

from behave import given, then, when
from rx.scheduler.eventloop import AsyncIOScheduler

from cleveragents.langgraph.graph import (
    MAX_EXECUTION_HISTORY,
    GraphConfig,
    LangGraph,
)
from cleveragents.langgraph.nodes import Edge, NodeConfig, NodeType
from cleveragents.langgraph.state import GraphState, StateManager
from cleveragents.reactive.stream_router import StreamMessage

# Existing steps cover constructor scheduling, execution dispatch, start validation,
# cycle detection, and node executor state updates.


@given("a langgraph config with layered nodes")
def step_layered_config(context):
    nodes = {
        "a": NodeConfig(name="a", type=NodeType.FUNCTION),
        "b": NodeConfig(name="b", type=NodeType.FUNCTION),
        "c": NodeConfig(name="c", type=NodeType.FUNCTION),
    }
    edges = [Edge(source="a", target="b"), Edge(source="b", target="c")]
    context.graph = _fresh_graph(
        GraphConfig(name="graph-levels", nodes=nodes, edges=edges)
    )


@when("I compute the topological levels")
def step_compute_levels(context):
    context.levels = context.graph._topological_levels()  # pylint: disable=protected-access


@then("the levels should list each layer")
def step_assert_levels(context):
    assert context.levels[0] == {"a"}
    assert context.levels[1] == {"b"}
    assert context.levels[2] == {"c"}


@given("a langgraph instance with recorded history")
def step_history_instance(context):
    config = GraphConfig(name="graph-history")
    context.graph = _fresh_graph(config)
    context.graph.execution_history.extend(["x", "y"])


@when("I request execution history")
def step_get_history(context):
    context.history_result = context.graph.get_execution_history()


@then("it should return a copy of the recorded history")
def step_assert_history_copy(context):
    assert context.history_result == ["x", "y"]
    assert context.history_result is not context.graph.execution_history


@given("a langgraph config with explicit start and end nodes")
def step_explicit_guards(context):
    start = NodeConfig(name="start", type=NodeType.START, metadata={"guard": True})
    end = NodeConfig(name="end", type=NodeType.END, metadata={"guard": True})
    nodes = {
        "start": start,
        "end": end,
        "worker": NodeConfig(name="worker", type=NodeType.FUNCTION),
    }
    context.graph = _fresh_graph(GraphConfig(name="graph-guards", nodes=nodes))


@when("I construct the LangGraph with that config")
def step_construct_with_guards(_context):
    # Construction already performed in the given step
    return


@then("initialization should reuse the provided start and end nodes")
def step_assert_guard_reuse(context):
    assert context.graph.nodes["start"].config is context.graph.config.nodes["start"]
    assert context.graph.nodes["end"].config is context.graph.config.nodes["end"]


@given("a langgraph instance prepared to observe node streams")
def step_prepare_node_stream(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    context.graph = _fresh_graph(GraphConfig(name="graph-observe", nodes=nodes))
    context.worker_stream = context.graph.stream_router.streams[
        f"__{context.graph.name}_node_worker__"
    ]
    context.graph.logger = MagicMock()


@when("the node stream emits next error and completion events")
def step_emit_stream_events(context):
    context.worker_stream.on_next("payload")
    context.worker_stream.on_error(ValueError("boom"))
    context.worker_stream.on_completed()


@then("the subscription handlers should be exercised")
def step_assert_stream_handlers(context):
    context.graph.logger.error.assert_called_once()


@given("a langgraph config with converging branches")
def step_converging_config(context):
    nodes = {
        "a": NodeConfig(name="a", type=NodeType.FUNCTION),
        "b": NodeConfig(name="b", type=NodeType.FUNCTION),
        "c": NodeConfig(name="c", type=NodeType.FUNCTION),
    }
    edges = [
        Edge(source="start", target="a"),
        Edge(source="start", target="b"),
        Edge(source="a", target="c"),
        Edge(source="b", target="c"),
        Edge(source="c", target="end"),
    ]
    context.graph = _fresh_graph(
        GraphConfig(name="graph-parallel", nodes=nodes, edges=edges)
    )


@when("I construct the LangGraph for parallel groups")
def step_construct_parallel(_context):
    return


@then("parallel groups should reflect staged execution order")
def step_assert_parallel_groups(context):
    assert context.graph.parallel_groups[0] == {"start"}
    assert context.graph.parallel_groups[1] == {"a", "b"}
    assert context.graph.parallel_groups[2] == {"c"}
    assert context.graph.parallel_groups[3] == {"end"}


@given("a langgraph instance missing required guard nodes")
def step_missing_guard_nodes(context):
    config = GraphConfig(name="graph-missing-guards")
    context.graph = _fresh_graph(config)
    context.graph.nodes.pop("start", None)


@when("I validate the graph explicitly")
def step_validate_graph(context):
    context.validation_error = None
    try:
        context.graph._validate_graph()  # pylint: disable=protected-access
    except Exception as exc:  # pylint: disable=broad-except
        context.validation_error = exc


@then("a guard validation error should be raised")
def step_assert_guard_validation(context):
    assert isinstance(context.validation_error, ValueError)


@given("a langgraph instance ready to start")
def step_ready_to_start(context):
    config = GraphConfig(name="graph-start")
    context.graph = _fresh_graph(config)
    context.start_stream = f"__{context.graph.name}_node_start__"
    context.graph.stream_router.send_message = MagicMock()


@when("I start without input then stop and fetch state")
def step_start_and_stop(context):
    context.graph.start()
    context.graph.stop()
    context.final_state = context.graph.get_state()


@then("it should use default state and update running flags")
def step_assert_start_stop(context):
    context.graph.stream_router.send_message.assert_called_with(
        context.start_stream, GraphState()
    )
    assert context.graph.is_running is False
    assert isinstance(context.final_state, GraphState)


def _fresh_graph(config: GraphConfig) -> LangGraph:
    """Construct a LangGraph with isolated scheduler and router."""
    return LangGraph(config=config, stream_router=None, scheduler=None)


@given("a langgraph config with only start and end nodes")
def step_minimal_config(context):
    context.graph_config = GraphConfig(name="graph-minimal")


@given("asyncio has no running loop")
def step_no_running_loop(_context):
    try:
        loop = asyncio.get_running_loop()
        if loop.is_running():
            loop.stop()
    except RuntimeError:
        # Expected when no loop is running
        pass


@when("I construct a LangGraph without providing a scheduler")
def step_construct_graph(context):
    context.graph = _fresh_graph(context.graph_config)


@then("it should create an AsyncIOScheduler and initialize streams")
def step_assert_scheduler_and_streams(context):
    assert isinstance(context.graph.scheduler, AsyncIOScheduler)
    start_stream = f"__{context.graph.name}_node_start__"
    assert start_stream in context.graph.stream_router.streams


@given("a langgraph instance with a fresh state manager")
def step_graph_with_state_manager(context):
    config = GraphConfig(name="graph-exec")
    context.graph = _fresh_graph(config)
    context.graph.is_running = True
    context.start_stream = f"__{context.graph.name}_node_start__"
    context.graph.stream_router.send_message = MagicMock()


@when("I execute the graph with input data")
def step_execute_graph(context):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        result = loop.run_until_complete(context.graph.execute({"messages": []}))
        context.execute_result = result
    finally:
        loop.close()
        replacement_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(replacement_loop)
        context.add_cleanup(replacement_loop.close)


@then("the start stream should receive the state and execution returns a GraphState")
def step_assert_execute(context):
    context.graph.stream_router.send_message.assert_called_once()
    called_state = context.graph.stream_router.send_message.call_args.args[1]
    assert isinstance(called_state, GraphState)
    assert isinstance(context.execute_result, GraphState)


@given("a langgraph instance with a removed start stream")
def step_graph_missing_start(context):
    config = GraphConfig(name="graph-missing-start")
    context.graph = _fresh_graph(config)
    start_stream = f"__{context.graph.name}_node_start__"
    context.graph.stream_router.streams.pop(start_stream, None)


@when("I call start on the graph")
def step_call_start(context):
    context.start_error = None
    try:
        context.graph.start()
    except Exception as exc:  # pylint: disable=broad-except
        context.start_error = exc


@then("a start stream error should be raised")
def step_assert_start_error(context):
    assert isinstance(context.start_error, ValueError)


@given("a langgraph config with parallel execution disabled and a cycle")
def step_cycle_config(context):
    nodes = {
        "a": NodeConfig(name="a", type=NodeType.FUNCTION),
    }
    edges = [Edge(source="start", target="a"), Edge(source="a", target="start")]
    context.graph_config = GraphConfig(
        name="graph-cycle",
        nodes=nodes,
        edges=edges,
        parallel_execution=False,
    )


@when("I construct the LangGraph")
def step_construct_cycle_graph(context):
    context.graph = _fresh_graph(context.graph_config)


@then("the graph should report cycles and empty parallel groups")
def step_assert_cycles(context):
    assert context.graph.has_cycles is True
    assert context.graph.parallel_groups == []


@given("a langgraph instance with a patched state manager")
def step_graph_with_executor(context):
    nodes = {
        "worker": NodeConfig(name="worker", type=NodeType.FUNCTION),
    }
    config = GraphConfig(name="graph-executor", nodes=nodes)
    graph = _fresh_graph(config)
    graph.is_running = True
    graph.state_manager.get_state = MagicMock(return_value=GraphState())
    graph.state_manager.update_state = MagicMock()
    graph.nodes["worker"].execute = AsyncMock(return_value={"messages": ["m"]})
    context.executor = graph._node_executors["worker"]  # pylint: disable=protected-access
    context.graph = graph


@when("I invoke the builtin node executor")
def step_invoke_executor(context):
    result = context.executor(StreamMessage(content="x", metadata={}))
    context.executor_result = result


@then("it should update state and record execution history")
def step_assert_executor(context):
    context.graph.state_manager.update_state.assert_called_once()
    # deque does not compare equal to list; convert first
    assert list(context.graph.execution_history) == ["worker"]
    assert isinstance(context.executor_result, StreamMessage)


# ── Node stream on_next wiring tests ──────────────────────────────────


@given("a langgraph instance with a mocked node executor")
def step_prepare_mocked_executor(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    context.graph = _fresh_graph(GraphConfig(name="graph-exec-wired", nodes=nodes))
    context.mock_executor = MagicMock(
        return_value=StreamMessage(content={}, metadata={})
    )
    context.graph._node_executors["worker"] = context.mock_executor  # pylint: disable=protected-access


@when("a message is emitted on the worker node stream")
def step_emit_worker_stream_message(context):
    worker_stream = context.graph.stream_router.streams[
        f"__{context.graph.name}_node_worker__"
    ]
    worker_stream.on_next(StreamMessage(content="test-payload", metadata={}))


@then("the registered executor should be called with the message")
def step_assert_executor_called(context):
    context.mock_executor.assert_called_once()
    call_args = context.mock_executor.call_args
    msg = call_args[0][0]
    assert isinstance(msg, StreamMessage)
    assert msg.content == "test-payload"


@when("a raw value is emitted on the worker node stream")
def step_emit_raw_value_on_stream(context):
    worker_stream = context.graph.stream_router.streams[
        f"__{context.graph.name}_node_worker__"
    ]
    worker_stream.on_next("raw-string-payload")


@then("the executor should receive a StreamMessage wrapping the raw value")
def step_assert_wrapped_message(context):
    context.mock_executor.assert_called_once()
    call_args = context.mock_executor.call_args
    msg = call_args[0][0]
    assert isinstance(msg, StreamMessage)
    assert msg.content == "raw-string-payload"
    assert msg.metadata == {}


@given("a langgraph instance with a removed node executor")
def step_prepare_missing_executor(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    context.graph = _fresh_graph(GraphConfig(name="graph-no-exec", nodes=nodes))
    context.graph._node_executors.pop("worker", None)  # pylint: disable=protected-access
    context.graph.logger = MagicMock()


@when("a message is emitted on the executorless worker stream")
def step_emit_executorless_message(context):
    worker_stream = context.graph.stream_router.streams[
        f"__{context.graph.name}_node_worker__"
    ]
    worker_stream.on_next("test")


@then("a warning about missing executor should be logged")
def step_assert_warning_logged(context):
    context.graph.logger.warning.assert_called_once()
    args, _kwargs = context.graph.logger.warning.call_args
    assert args[0] == "No executor found for node %s"
    assert args[1] == "worker"


@given("a langgraph instance with multiple mocked node executors")
def step_prepare_multi_executors(context):
    nodes = {
        "alpha": NodeConfig(name="alpha", type=NodeType.FUNCTION),
        "beta": NodeConfig(name="beta", type=NodeType.FUNCTION),
    }
    context.graph = _fresh_graph(GraphConfig(name="graph-multi", nodes=nodes))
    context.alpha_executor = MagicMock(
        return_value=StreamMessage(content={}, metadata={})
    )
    context.beta_executor = MagicMock(
        return_value=StreamMessage(content={}, metadata={})
    )
    context.graph._node_executors["alpha"] = context.alpha_executor  # pylint: disable=protected-access
    context.graph._node_executors["beta"] = context.beta_executor  # pylint: disable=protected-access


@when("messages are emitted on both alpha and beta node streams")
def step_emit_both_messages(context):
    alpha_stream = context.graph.stream_router.streams[
        f"__{context.graph.name}_node_alpha__"
    ]
    beta_stream = context.graph.stream_router.streams[
        f"__{context.graph.name}_node_beta__"
    ]
    alpha_stream.on_next(StreamMessage(content="alpha-payload", metadata={}))
    beta_stream.on_next(StreamMessage(content="beta-payload", metadata={}))


@then("each executor should receive its own message independently")
def step_assert_independent_executors(context):
    context.alpha_executor.assert_called_once()
    context.beta_executor.assert_called_once()
    alpha_msg = context.alpha_executor.call_args[0][0]
    beta_msg = context.beta_executor.call_args[0][0]
    assert isinstance(alpha_msg, StreamMessage)
    assert isinstance(beta_msg, StreamMessage)
    assert alpha_msg.content == "alpha-payload"
    assert beta_msg.content == "beta-payload"


# ── Executor exception handling (M3) ─────────────────────────────────


@given("a langgraph instance with a failing node executor")
def step_prepare_failing_executor(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    context.graph = _fresh_graph(GraphConfig(name="graph-fail-exec", nodes=nodes))
    context.graph._node_executors["worker"] = MagicMock(  # pylint: disable=protected-access
        side_effect=RuntimeError("executor boom")
    )
    context.graph.logger = MagicMock()


@when("a message is emitted on the failing worker stream")
def step_emit_failing_worker_message(context):
    worker_stream = context.graph.stream_router.streams[
        f"__{context.graph.name}_node_worker__"
    ]
    context.on_next_error = None
    try:
        worker_stream.on_next(StreamMessage(content="trigger", metadata={}))
    except Exception as exc:  # pylint: disable=broad-except
        context.on_next_error = exc


@then("the exception should be caught and logged without propagating")
def step_assert_exception_caught(context):
    assert context.on_next_error is None, (
        f"on_next raised unexpectedly: {context.on_next_error}"
    )
    context.graph.logger.exception.assert_called_once()
    args, _kwargs = context.graph.logger.exception.call_args
    assert args[0] == "Executor failed for node %s"
    assert args[1] == "worker"


# ── Missing observable branch coverage (m3) ──────────────────────────


@given("a langgraph instance with a removed observable")
def step_prepare_removed_observable(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    context.graph = _fresh_graph(GraphConfig(name="graph-no-obs", nodes=nodes))
    stream_name = f"__{context.graph.name}_node_worker__"
    context.graph.stream_router.observables.pop(stream_name, None)
    context.graph.logger = MagicMock()


@when("node stream subscriptions are set up")
def step_rerun_subscriptions(context):
    # Note: calling _setup_node_stream_subscriptions a second time creates
    # duplicate subscriptions for start/end nodes.  This is expected and
    # harmless in this test context — we only care about the worker branch.
    context.graph._setup_node_stream_subscriptions()  # pylint: disable=protected-access


@then("no executor should be called and no error should occur")
def step_assert_no_error_on_missing_observable(context):
    context.graph.logger.debug.assert_any_call(
        "No observable for stream %s; skipping subscription",
        f"__{context.graph.name}_node_worker__",
    )
    context.graph.logger.error.assert_not_called()
    context.graph.logger.warning.assert_not_called()


# ── Execute raises when start stream is missing (m1) ─────────────────


@given("a langgraph instance with a removed start stream for execute")
def step_prepare_execute_missing_start(context):
    config = GraphConfig(name="graph-exec-no-start")
    context.graph = _fresh_graph(config)
    context.graph.is_running = True
    start_stream = f"__{context.graph.name}_node_start__"
    context.graph.stream_router.streams.pop(start_stream, None)


@when("I execute the graph expecting a start stream error")
def step_execute_expecting_error(context):
    context.execute_error = None
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(context.graph.execute({"messages": []}))
    except Exception as exc:  # pylint: disable=broad-except
        context.execute_error = exc
    finally:
        loop.close()
        replacement_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(replacement_loop)
        context.add_cleanup(replacement_loop.close)


@then("a start stream missing error should be raised from execute")
def step_assert_execute_start_error(context):
    assert isinstance(context.execute_error, ValueError)
    assert "Start stream not initialized" in str(context.execute_error)


# ── run_coroutine_threadsafe path coverage (M5) ──────────────────────


def _cleanup_bg_loop(context):
    """Shut down the background event loop and join the thread."""
    bg_loop = getattr(context, "_bg_loop", None)
    bg_thread = getattr(context, "_bg_thread", None)
    try:
        if bg_loop is not None:
            bg_loop.call_soon_threadsafe(bg_loop.stop)
    except RuntimeError:
        # Loop may already be closed; ensure thread join still happens.
        pass
    if bg_thread is not None:
        bg_thread.join(timeout=5.0)
    if bg_loop is not None:
        bg_loop.close()


@given(
    "a langgraph instance with a scheduler whose loop is running in a background thread"
)
def step_prepare_running_loop_graph(context):
    bg_loop = asyncio.new_event_loop()

    def run_loop() -> None:
        asyncio.set_event_loop(bg_loop)
        bg_loop.run_forever()

    bg_thread = threading.Thread(target=run_loop, daemon=True)
    bg_thread.start()
    context._bg_loop = bg_loop
    context._bg_thread = bg_thread
    # Register cleanup so the background loop is always stopped, even if
    # assertions fail in the @then step.
    context.add_cleanup(_cleanup_bg_loop, context)

    scheduler = AsyncIOScheduler(bg_loop)
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    config = GraphConfig(name="graph-running-loop", nodes=nodes)
    context.graph = LangGraph(config=config, stream_router=None, scheduler=scheduler)
    context.graph.is_running = True
    context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
    context.graph.state_manager.update_state = MagicMock()
    context.graph.nodes["worker"].execute = AsyncMock(return_value={"messages": ["bg"]})


@when("a message is emitted on the worker stream via the running loop path")
def step_emit_via_running_loop(context):
    executor = context.graph._node_executors["worker"]  # pylint: disable=protected-access
    with patch(
        "asyncio.run_coroutine_threadsafe", wraps=asyncio.run_coroutine_threadsafe
    ) as mock_rcts:
        context.executor_result = executor(
            StreamMessage(content="bg-test", metadata={})
        )
        context._mock_run_coroutine_threadsafe = mock_rcts


@then("the executor should complete successfully via run_coroutine_threadsafe")
def step_assert_running_loop_executor(context):
    assert isinstance(context.executor_result, StreamMessage)
    assert context.executor_result.metadata["node"] == "worker"
    context.graph.state_manager.update_state.assert_called_once()
    # Verify that run_coroutine_threadsafe was actually used (not the
    # thread-pool fallback path).  The mock retains call history after
    # the patch context manager exits in the @when step.
    context._mock_run_coroutine_threadsafe.assert_called_once()


# ── Execution history cap test (m5) ──────────────────────────────────


@given("a langgraph instance with execution history at capacity")
def step_prepare_full_history(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    config = GraphConfig(name="graph-history-cap", nodes=nodes)
    context.graph = _fresh_graph(config)
    context.graph.is_running = True
    context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
    context.graph.state_manager.update_state = MagicMock()
    context.graph.nodes["worker"].execute = AsyncMock(return_value={"messages": ["m"]})
    # Pre-fill to capacity
    for i in range(MAX_EXECUTION_HISTORY):
        context.graph.execution_history.append(f"entry-{i}")
    context.oldest_entry = "entry-0"


@when("one more executor invocation is recorded")
def step_invoke_one_more(context):
    executor = context.graph._node_executors["worker"]  # pylint: disable=protected-access
    executor(StreamMessage(content="x", metadata={}))


@then(
    "the history length should remain at the maximum and the oldest entry should be dropped"
)
def step_assert_history_cap(context):
    assert len(context.graph.execution_history) == MAX_EXECUTION_HISTORY
    history_list = list(context.graph.execution_history)
    assert context.oldest_entry not in history_list
    assert history_list[-1] == "worker"


# ── Concurrent state update lock test (m8) ───────────────────────────


@given("a state manager configured for concurrent access")
def step_prepare_concurrent_state_manager(context):
    context.state_manager = StateManager(initial_state=GraphState())
    context.num_threads = 20
    context.updates_per_thread = 10


@when("multiple threads perform state updates simultaneously")
def step_concurrent_updates(context):
    barrier = threading.Barrier(context.num_threads)

    def updater() -> None:
        barrier.wait(timeout=10.0)
        for _ in range(context.updates_per_thread):
            context.state_manager.update_state(
                {"metadata": {"t": threading.current_thread().name}}
            )

    threads = [threading.Thread(target=updater) for _ in range(context.num_threads)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=30.0)
    assert all(not t.is_alive() for t in threads), "Some threads did not complete"


@then("the final execution count should equal the total number of updates")
def step_assert_concurrent_count(context):
    expected = context.num_threads * context.updates_per_thread
    actual = context.state_manager.get_state().execution_count
    assert actual == expected, f"Expected {expected}, got {actual}"


# ── TimeoutError branch coverage (M5) ────────────────────────────────


@given(
    "a langgraph instance with a slow executor on a background loop and a short timeout"
)
def step_prepare_slow_executor_bg_loop(context):
    bg_loop = asyncio.new_event_loop()

    def run_loop() -> None:
        asyncio.set_event_loop(bg_loop)
        bg_loop.run_forever()

    bg_thread = threading.Thread(target=run_loop, daemon=True)
    bg_thread.start()
    context._bg_loop = bg_loop
    context._bg_thread = bg_thread
    context.add_cleanup(_cleanup_bg_loop, context)

    scheduler = AsyncIOScheduler(bg_loop)
    # Use a very short timeout (0.01s) so the test completes quickly.
    nodes = {
        "worker": NodeConfig(name="worker", type=NodeType.FUNCTION, timeout=0.01),
    }
    config = GraphConfig(name="graph-timeout-rcts", nodes=nodes)
    context.graph = LangGraph(config=config, stream_router=None, scheduler=scheduler)
    context.graph.is_running = True
    context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
    context.graph.state_manager.update_state = MagicMock()

    async def slow_execute(state):
        # Use the original (un-patched) asyncio.sleep so the test-infrastructure
        # 10 ms sleep cap does not race with the 0.01 s executor timeout.
        _real_sleep = getattr(asyncio, "_original_sleep", asyncio.sleep)
        await _real_sleep(0.5)
        return {"messages": ["never"]}

    context.graph.nodes["worker"].execute = slow_execute


@when("the slow executor is invoked via the running loop path")
def step_invoke_slow_executor_rcts(context):
    executor = context.graph._node_executors["worker"]  # pylint: disable=protected-access
    context.timeout_error = None
    try:
        executor(StreamMessage(content="slow", metadata={}))
    except TimeoutError as exc:
        context.timeout_error = exc


@then("a TimeoutError should be raised with the node name and timeout duration")
def step_assert_timeout_rcts(context):
    assert context.timeout_error is not None, "Expected TimeoutError was not raised"
    msg = str(context.timeout_error)
    assert "worker" in msg
    assert "0.01" in msg


@given("a langgraph instance with a slow executor and a short timeout")
def step_prepare_slow_executor_tp(context):
    nodes = {
        "worker": NodeConfig(name="worker", type=NodeType.FUNCTION, timeout=0.01),
    }
    config = GraphConfig(name="graph-timeout-tp", nodes=nodes)
    context.graph = _fresh_graph(config)
    context.graph.is_running = True
    context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
    context.graph.state_manager.update_state = MagicMock()

    async def slow_execute(state):
        # Use the original (un-patched) asyncio.sleep so the test-infrastructure
        # 10 ms sleep cap does not race with the 0.01 s executor timeout.
        _real_sleep = getattr(asyncio, "_original_sleep", asyncio.sleep)
        await _real_sleep(0.5)
        return {"messages": ["never"]}

    context.graph.nodes["worker"].execute = slow_execute


@when("the slow executor is invoked via the thread pool path")
def step_invoke_slow_executor_tp(context):
    executor = context.graph._node_executors["worker"]  # pylint: disable=protected-access
    context.timeout_error = None
    try:
        executor(StreamMessage(content="slow", metadata={}))
    except TimeoutError as exc:
        context.timeout_error = exc


@then("a TimeoutError should be raised from the thread pool path with the node name")
def step_assert_timeout_tp(context):
    assert context.timeout_error is not None, "Expected TimeoutError was not raised"
    msg = str(context.timeout_error)
    assert "worker" in msg
    assert "0.01" in msg


# ── Restart-after-stop test (m7) ─────────────────────────────────────


@given("a langgraph instance that has been started and stopped")
def step_prepare_restart_graph(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    config = GraphConfig(name="graph-restart", nodes=nodes)
    context.graph = _fresh_graph(config)
    context.graph.stream_router.send_message = MagicMock()
    context.graph.state_manager.get_state = MagicMock(return_value=GraphState())
    context.graph.state_manager.update_state = MagicMock()
    context.graph.nodes["worker"].execute = AsyncMock(
        return_value={"messages": ["restarted"]}
    )
    context.graph.start()
    context.graph.stop()
    assert context.graph.is_running is False


@when("I restart the graph and invoke an executor")
def step_restart_and_invoke(context):
    context.graph.start()
    # Ensure the pool is cleaned up even if assertions fail later.
    context.add_cleanup(context.graph.stop)
    # Invoke the executor directly to verify the pool is recreated after
    # restart.  This does NOT verify the full stream→executor subscription
    # path; that is covered by the on_next wiring scenarios above.
    executor = context.graph._node_executors["worker"]  # pylint: disable=protected-access
    context.restart_result = executor(
        StreamMessage(content="after-restart", metadata={})
    )


@then("the executor should complete successfully after restart")
def step_assert_restart_success(context):
    assert isinstance(context.restart_result, StreamMessage)
    assert context.restart_result.metadata["node"] == "worker"
    assert context.graph.is_running is True
    # stop() is handled by context.add_cleanup registered in the @when step.


# ── is_running guard test (m4) ───────────────────────────────────────


@given("a langgraph instance that is not running")
def step_prepare_not_running_graph(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    config = GraphConfig(name="graph-not-running", nodes=nodes)
    context.graph = _fresh_graph(config)
    # Ensure the graph is explicitly not running.
    context.graph.is_running = False


@when("I invoke the node executor directly")
def step_invoke_executor_not_running(context):
    executor = context.graph._node_executors["worker"]  # pylint: disable=protected-access
    context.not_running_error = None
    try:
        executor(StreamMessage(content="should-fail", metadata={}))
    except Exception as exc:  # pylint: disable=broad-except
        context.not_running_error = exc


@then("a RuntimeError should be raised indicating the graph is not running")
def step_assert_not_running_error(context):
    assert isinstance(context.not_running_error, RuntimeError)
    assert "not running" in str(context.not_running_error)
    assert "worker" in str(context.not_running_error)


# ── StateManager.replace_state() direct test (m6) ───────────────────


@given("a state manager with an initial state")
def step_prepare_state_manager_for_replace(context):
    context.state_manager = StateManager(initial_state=GraphState())
    context.stream_emissions = []  # list[GraphState]
    context.state_manager.state_stream.subscribe(
        on_next=lambda s: context.stream_emissions.append(s)
    )


@when("I call replace_state with a new state")
def step_call_replace_state(context):
    context.new_state = GraphState(metadata={"replaced": True}, current_node="new-node")
    context.state_manager.replace_state(context.new_state)


@then("get_state should return the new state")
def step_assert_replace_state_get(context):
    retrieved = context.state_manager.get_state()
    assert retrieved.metadata == {"replaced": True}
    assert retrieved.current_node == "new-node"


@then("the state stream should have received a notification")
def step_assert_replace_state_stream(context):
    # BehaviorSubject emits the initial value on subscribe, plus the
    # replace_state emission, so we expect at least 2 emissions.
    assert len(context.stream_emissions) >= 2


@then("the emitted state should be a deep copy not the same object")
def step_assert_replace_state_deep_copy(context):
    # The last emission should be a deep copy, not the same object as
    # the internal state or the new_state we passed in.
    emitted = context.stream_emissions[-1]
    assert emitted is not context.new_state
    assert emitted.metadata == {"replaced": True}


@then("the emitted state should be distinct from the internal state")
def step_assert_emitted_distinct_from_internal(context):
    # After the M1 fix (deep-copy on store), the emitted copy must also
    # be a distinct object from the internal state held by the manager.
    emitted = context.stream_emissions[-1]
    # Direct access intentional: identity check requires the actual
    # internal object, not a get_state() copy.
    assert emitted is not context.state_manager.state


# ── CancelledError path coverage (M3) ───────────────────────────────


@given("a langgraph instance with a cancellable executor task")
def step_prepare_cancellable_executor(context):
    nodes = {
        "worker": NodeConfig(name="worker", type=NodeType.FUNCTION),
    }
    config = GraphConfig(name="graph-cancel", nodes=nodes)
    context.graph = _fresh_graph(config)
    context.graph.is_running = True


@when("the executor future is cancelled before completion")
def step_cancel_executor_future(context):
    context.cancel_error = None
    # Patch the executor pool's submit to return a future that raises
    # CancelledError on result().  This simulates graph shutdown
    # (stop() calls shutdown(wait=True, cancel_futures=True)) cancelling
    # a queued-but-not-yet-started task.
    cancelled_future: concurrent.futures.Future[StreamMessage] = (
        concurrent.futures.Future()
    )
    # cancel() on a fresh Future is sufficient to put it in CANCELLED state.
    cancelled_future.cancel()

    executor_fn = context.graph._node_executors["worker"]  # pylint: disable=protected-access
    with patch.object(
        context.graph._executor_pool,  # pylint: disable=protected-access
        "submit",
        return_value=cancelled_future,
    ):
        try:
            executor_fn(StreamMessage(content="cancel-me", metadata={}))
        except RuntimeError as exc:
            context.cancel_error = exc


@then("a RuntimeError should be raised indicating graph stopping")
def step_assert_cancelled_error(context):
    assert context.cancel_error is not None, "Expected RuntimeError was not raised"
    msg = str(context.cancel_error)
    assert "cancelled" in msg and "stopping" in msg, f"Unexpected error message: {msg}"


# ── __del__ safety net test (m3) ─────────────────────────────────────


@given("a langgraph instance that was never stopped")
def step_prepare_unstopped_graph(context):
    config = GraphConfig(name="graph-del-test")
    context.graph = _fresh_graph(config)
    # Ensure the graph has a live executor pool (never stopped).
    assert context.graph._executor_pool is not None  # pylint: disable=protected-access


@when("__del__ is called on the graph")
def step_call_del(context):
    context.del_error = None
    try:
        context.graph.__del__()
    except Exception as exc:  # pylint: disable=broad-except
        context.del_error = exc


# The @then step "no exception should propagate from __del__" is already
# defined in bridge_coverage_steps.py and is reused here.  It checks
# ``context.del_error is None``.


# ── execute() is_running guard test (n1) ─────────────────────────────


@given("a langgraph instance that is not running for execute")
def step_prepare_not_running_for_execute(context):
    config = GraphConfig(name="graph-exec-not-running")
    context.graph = _fresh_graph(config)
    context.graph.is_running = False


@when("I call execute on the non-running graph")
def step_call_execute_not_running(context):
    context.execute_not_running_error = None
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(context.graph.execute({"messages": []}))
    except Exception as exc:  # pylint: disable=broad-except
        context.execute_not_running_error = exc
    finally:
        loop.close()
        replacement_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(replacement_loop)
        context.add_cleanup(replacement_loop.close)


@then("a RuntimeError should be raised indicating the graph is not running for execute")
def step_assert_execute_not_running_error(context):
    assert isinstance(context.execute_not_running_error, RuntimeError)
    assert "not running" in str(context.execute_not_running_error)


# ── TimeoutError through stream path (M2 cycle 6) ───────────────────


@given("a langgraph instance with an executor that raises TimeoutError")
def step_prepare_timeout_executor_for_stream(context):
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    context.graph = _fresh_graph(GraphConfig(name="graph-timeout-stream", nodes=nodes))
    context.graph._node_executors["worker"] = MagicMock(  # pylint: disable=protected-access
        side_effect=TimeoutError("timed out")
    )
    context.graph.logger = MagicMock()


@when("a message is emitted on the timeout worker stream")
def step_emit_timeout_worker_stream(context):
    worker_stream = context.graph.stream_router.streams[
        f"__{context.graph.name}_node_worker__"
    ]
    context.stream_error = None
    try:
        worker_stream.on_next(StreamMessage(content="trigger-timeout", metadata={}))
    except Exception as exc:  # pylint: disable=broad-except
        context.stream_error = exc


@then("the timeout should be logged at warning level")
def step_assert_timeout_warning_logged(context):
    assert context.stream_error is None, (
        f"on_next raised unexpectedly: {context.stream_error}"
    )
    context.graph.logger.warning.assert_called_once()
    args, _kwargs = context.graph.logger.warning.call_args
    assert args[0] == "Executor timed out for node %s"
    assert args[1] == "worker"


@then("logger exception should not be called for timeout")
def step_assert_no_exception_log_for_timeout(context):
    context.graph.logger.exception.assert_not_called()


# ── CancelledError in run_coroutine_threadsafe path (M1 cycle 6) ────


@given("a langgraph instance with a scheduler loop and a cancellable coroutine future")
def step_prepare_rcts_cancellable(context):
    bg_loop = asyncio.new_event_loop()

    def run_loop() -> None:
        asyncio.set_event_loop(bg_loop)
        bg_loop.run_forever()

    bg_thread = threading.Thread(target=run_loop, daemon=True)
    bg_thread.start()
    context._bg_loop = bg_loop
    context._bg_thread = bg_thread
    context.add_cleanup(_cleanup_bg_loop, context)

    scheduler = AsyncIOScheduler(bg_loop)
    nodes = {"worker": NodeConfig(name="worker", type=NodeType.FUNCTION)}
    config = GraphConfig(name="graph-cancel-rcts", nodes=nodes)
    context.graph = LangGraph(config=config, stream_router=None, scheduler=scheduler)
    context.graph.is_running = True


@when("the coroutine future is cancelled before completion")
def step_cancel_coroutine_future(context):
    context.cancel_rcts_error = None
    cancelled_future: concurrent.futures.Future[StreamMessage] = (
        concurrent.futures.Future()
    )
    # cancel() on a fresh Future is sufficient to put it in CANCELLED state.
    cancelled_future.cancel()

    executor_fn = context.graph._node_executors["worker"]  # pylint: disable=protected-access
    with patch(
        "asyncio.run_coroutine_threadsafe",
        return_value=cancelled_future,
    ):
        try:
            executor_fn(StreamMessage(content="cancel-rcts", metadata={}))
        except RuntimeError as exc:
            context.cancel_rcts_error = exc


@then("a RuntimeError should be raised indicating graph stopping from coroutine path")
def step_assert_rcts_cancelled_error(context):
    assert context.cancel_rcts_error is not None, "Expected RuntimeError was not raised"
    msg = str(context.cancel_rcts_error)
    assert "cancelled" in msg and "stopping" in msg, f"Unexpected error message: {msg}"