"""Remote LlamaAgents bridge tests."""

from __future__ import annotations

from ._llama_agents_bridge_support import (
    AgentTurnInput,
    Any,
    AsyncIterator,
    CancelToken,
    InMemoryRingBuffer,
    LlamaAgentsBridge,
    _AdvanceThenBlockStream,
    _BlockingRemoteStream,
    _HandlerData,
    _InputRequiredEvent,
    _RawStream,
    _recorder,
    _RemoteClient,
    _RemoteEnvelope,
    _RemoteHitlClient,
    _RemoteStream,
    _StopEvent,
    _TextEvent,
    asyncio,
    pytest,
)


class TestRemoteLlamaAgentsBridge:
    @pytest.mark.asyncio
    async def test_streams_remote_workflow_client_events(self, fake_workflows_modules):
        client = _RemoteClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        events = []
        async for event in bridge.invoke(AgentTurnInput.from_text("Ada"), _recorder()):
            events.append(event)

        assert [event.text for event in events if event.kind == "text_delta"] == [
            "remote ",
            "text",
        ]
        assert [event.text for event in events if event.kind == "done"] == ["remote text"]
        start_event = client.run_calls[0]["start_event"]
        assert start_event.message == "Ada"

    @pytest.mark.asyncio
    async def test_remote_human_input_event_uses_send_event(self, fake_workflows_modules):
        client = _RemoteHitlClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        first_turn = []
        async for event in bridge.invoke(AgentTurnInput.from_text("start"), _recorder()):
            first_turn.append(event)

        assert [event.text for event in first_turn if event.kind == "text_delta"] == [
            "Remote prompt"
        ]

        second_turn = []
        async for event in bridge.invoke(AgentTurnInput.from_text("Ada"), _recorder()):
            second_turn.append(event)

        assert client.sent_events[-1][0] == "h1"
        assert client.sent_events[-1][1].response == "Ada"
        assert client.stream_calls[1]["after_sequence"] == 0
        assert [event.text for event in second_turn if event.kind == "done"] == ["Remote done"]

    @pytest.mark.asyncio
    async def test_remote_start_failure_keeps_interruption_note_for_retry(
        self, fake_workflows_modules
    ):
        class _FailOnceClient(_RemoteClient):
            async def run_workflow_nowait(self, workflow_name: str, **kwargs: Any) -> _HandlerData:
                self.run_calls.append({"workflow_name": workflow_name, **kwargs})
                if len(self.run_calls) == 1:
                    raise RuntimeError("remote workflow start failed")
                return _HandlerData("h-retried")

        client = _FailOnceClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")
        bridge.append_interruption_note("retry-note")

        with pytest.raises(RuntimeError, match="remote workflow start failed"):
            async for _ in bridge.invoke(AgentTurnInput.from_text("next"), _recorder()):
                pass

        assert client.run_calls[0]["start_event"].easycat_interruption_note == "retry-note"
        assert bridge._pending_interruption_note == "retry-note"

        async for _ in bridge.invoke(AgentTurnInput.from_text("next"), _recorder()):
            pass

        assert client.run_calls[1]["start_event"].easycat_interruption_note == "retry-note"
        assert bridge._pending_interruption_note is None

    @pytest.mark.asyncio
    async def test_remote_hitl_retry_after_send_failure(self, fake_workflows_modules):
        """A transient send_event failure must keep the bridge waiting for
        input so a retry resends the HumanResponseEvent instead of starting
        a fresh run that leaves the paused workflow stuck."""

        class _FlakySendClient(_RemoteHitlClient):
            def __init__(self) -> None:
                super().__init__()
                self.send_attempts = 0

            async def send_event(
                self, handler_id: str, event: Any, step: str | None = None
            ) -> None:
                self.send_attempts += 1
                if self.send_attempts == 1:
                    raise RuntimeError("transient network error")
                await super().send_event(handler_id, event, step=step)

        client = _FlakySendClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        async for _ in bridge.invoke(AgentTurnInput.from_text("start"), _recorder()):
            pass

        with pytest.raises(RuntimeError, match="transient network error"):
            async for _ in bridge.invoke(AgentTurnInput.from_text("Ada"), _recorder()):
                pass

        retry_turn = []
        async for event in bridge.invoke(AgentTurnInput.from_text("Ada"), _recorder()):
            retry_turn.append(event)

        # Only the initial start spun up a run -- the retry resent the
        # human response to the same paused handler rather than launching
        # a second run_workflow_nowait().
        assert len(client.run_calls) == 1
        assert client.sent_events[-1][0] == "h1"
        assert client.sent_events[-1][1].response == "Ada"
        assert [event.text for event in retry_turn if event.kind == "done"] == ["Remote done"]

    @pytest.mark.asyncio
    async def test_remote_cancellation_during_idle_stream_is_prompt(self, fake_workflows_modules):
        stream = _BlockingRemoteStream()

        class _BlockingRemoteClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> Any:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                return stream

        client = _BlockingRemoteClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")
        token = CancelToken()
        collected: list[Any] = []

        async def _consume() -> None:
            async for event in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder(), token):
                collected.append(event)

        async def _cancel_when_blocked() -> None:
            await stream.streaming_blocked.wait()
            token.cancel()

        await asyncio.wait_for(asyncio.gather(_consume(), _cancel_when_blocked()), timeout=2.0)

        assert client.cancelled == ["h1"]
        assert stream.closed is True
        assert [e.text for e in collected if e.kind == "text_delta"] == ["remote partial"]

    @pytest.mark.asyncio
    async def test_remote_hard_task_cancel_calls_cancel_handler(self, fake_workflows_modules):
        """Cancelling the invoke task (not just the token) must still stop the
        remote handler so it does not keep running server-side."""
        stream = _BlockingRemoteStream()

        class _BlockingRemoteClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> Any:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                return stream

        client = _BlockingRemoteClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        async def _consume() -> None:
            async for _ in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder()):
                pass

        task = asyncio.ensure_future(_consume())
        await asyncio.wait_for(stream.streaming_blocked.wait(), timeout=2.0)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task

        assert client.cancelled == ["h1"]
        assert stream.closed is True

    @pytest.mark.asyncio
    async def test_remote_stream_failure_cancels_handler_and_clears_state(
        self, fake_workflows_modules
    ):
        """A non-cancellation error from the remote event stream (e.g. a
        connection failure that outlived the client's reconnect attempts)
        must stop the server-side handler and drop the handler references so
        the next turn starts fresh instead of resuming an abandoned handler,
        and snapshots must stop advertising the failed handler."""

        class _FailingStream:
            def __init__(self) -> None:
                self.closed = False
                self.last_sequence = -1

            def __aiter__(self) -> AsyncIterator[Any]:
                return self._iterate()

            async def _iterate(self) -> AsyncIterator[Any]:
                self.last_sequence = 0
                yield _RemoteEnvelope(_TextEvent("remote partial"))
                raise RuntimeError("remote stream connection lost")

            async def aclose(self) -> None:
                self.closed = True

        failing_stream = _FailingStream()

        class _FailingStreamClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> Any:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                if len(self.stream_calls) == 1:
                    return failing_stream
                return _RemoteStream([_TextEvent("second turn"), _StopEvent("done")])

        client = _FailingStreamClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        with pytest.raises(RuntimeError, match="remote stream connection lost"):
            async for _ in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder()):
                pass

        # Server-side workflow stopped and the SSE reader closed.
        assert client.cancelled == ["h1"]
        assert failing_stream.closed is True

        # snapshot_state() no longer advertises the failed handler.
        fields = bridge.snapshot_state().fields
        assert fields["active_handler_id"] is None
        assert fields["remote_handler_id"] is None
        assert fields["waiting_for_input"] is False

        # The next turn starts a fresh run rather than resuming the dead
        # (abandoned/contaminated) handler.
        async for _ in bridge.invoke(AgentTurnInput.from_text("again"), _recorder()):
            pass
        assert client.run_calls[1]["handler_id"] is None

    @pytest.mark.asyncio
    async def test_remote_cancelled_handler_continues_next_turn(self, fake_workflows_modules):
        """A barge-in must preserve the server-side handler reference.

        ``cancel_handler`` keeps the handler persisted (purge defaults to
        False), so the next default (preserve_context) turn continues from it
        via ``handler_id`` -- the only server-side state reference, since the
        real ``llama-agents-client`` ``HandlerData`` model exposes no context
        field. Dropping it would silently lose all conversation state after
        every barge-in.
        """
        first_stream = _BlockingRemoteStream()

        class _CancelThenResumeClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> Any:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                if len(self.stream_calls) == 1:
                    return first_stream
                return _RemoteStream([_TextEvent("second turn"), _StopEvent("done")])

        client = _CancelThenResumeClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")
        token = CancelToken()

        async def _consume() -> None:
            async for _ in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder(), token):
                pass

        async def _cancel_when_blocked() -> None:
            await first_stream.streaming_blocked.wait()
            token.cancel()

        await asyncio.wait_for(asyncio.gather(_consume(), _cancel_when_blocked()), timeout=2.0)
        assert client.cancelled == ["h1"]

        async for _ in bridge.invoke(AgentTurnInput.from_text("again"), _recorder()):
            pass

        # The next turn continues from the preserved (terminal but still
        # persisted) handler instead of starting a fresh one and losing all
        # workflow state. The event cursor is carried forward -- exactly like
        # the completed-handler preserve_context path -- so the resumed
        # handler's accumulating event log is not replayed.
        assert client.run_calls[1]["handler_id"] == "h1"
        assert client.stream_calls[1]["after_sequence"] == 0

    @pytest.mark.asyncio
    async def test_remote_cancellation_advances_cursor_past_dropped_event(
        self, fake_workflows_modules
    ):
        """A barge-in that drops an event must still advance the cursor.

        When the stream's ``last_sequence`` has already moved past an event
        that the cancellation race never delivered to the loop body, the next
        preserve_context turn must resume *after* that dropped interrupted
        delta. Otherwise the stale cursor replays it as fresh speech.
        """
        first_stream = _AdvanceThenBlockStream()

        class _CancelThenResumeClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> Any:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                if len(self.stream_calls) == 1:
                    return first_stream
                return _RemoteStream([_TextEvent("second turn"), _StopEvent("done")])

        client = _CancelThenResumeClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")
        token = CancelToken()
        first_turn: list[Any] = []

        async def _consume() -> None:
            async for event in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder(), token):
                first_turn.append(event)

        async def _cancel_when_blocked() -> None:
            await first_stream.first_delivered.wait()
            token.cancel()

        await asyncio.wait_for(asyncio.gather(_consume(), _cancel_when_blocked()), timeout=2.0)

        # Only the delivered event 0 was spoken; the undelivered event 1 was
        # dropped by the cancellation race.
        assert [e.text for e in first_turn if e.kind == "text_delta"] == ["first delta"]
        assert client.cancelled == ["h1"]

        second_turn: list[Any] = []
        async for event in bridge.invoke(AgentTurnInput.from_text("again"), _recorder()):
            second_turn.append(event)

        # The cursor advanced to the stream's last_sequence (1) even though the
        # loop body never ran for event 1, so the resumed handler streams from
        # after the dropped delta instead of replaying it.
        assert client.run_calls[1]["handler_id"] == "h1"
        assert client.stream_calls[1]["after_sequence"] == 1
        assert "interrupted second delta" not in [
            e.text for e in second_turn if e.kind == "text_delta"
        ]

    @pytest.mark.asyncio
    async def test_remote_failed_handler_status_raises(self, fake_workflows_modules):
        class _FailingClient(_RemoteClient):
            async def get_handler(self, handler_id: str) -> _HandlerData:
                return _HandlerData(handler_id, result=None, status="failed", error="boom")

        client = _FailingClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")
        journal = InMemoryRingBuffer(capacity=1000)

        with pytest.raises(RuntimeError, match="failed"):
            async for _ in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder(journal)):
                pass

        names = [record.name for record in journal.read()]
        assert "framework_error" in names
        exit_records = [r for r in journal.read() if r.name == "unit_exited"]
        assert exit_records and exit_records[-1].data.get("exit_reason") == "error"

    @pytest.mark.asyncio
    async def test_remote_extracts_text_from_nested_dict_envelope(self, fake_workflows_modules):
        class _NestedClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> _RemoteStream:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                return _RemoteStream([_StopEvent("ignored")])

            async def get_handler(self, handler_id: str) -> _HandlerData:
                envelope = {"value": {"result": {"message": "Hello from envelope"}}}
                return _HandlerData(handler_id, result=envelope)

        client = _NestedClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        events = []
        async for event in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder()):
            events.append(event)

        assert [event.text for event in events if event.kind == "text_delta"] == [
            "Hello from envelope"
        ]
        assert [event.text for event in events if event.kind == "done"] == ["Hello from envelope"]

    @pytest.mark.asyncio
    async def test_remote_preserve_context_carries_event_cursor(self, fake_workflows_modules):
        """A preserved second turn resumes after the first turn's events.

        With preserve_context the handler (and its server-side event log)
        survives into the next turn, so streaming from -1 again would replay
        the first turn before the new response.
        """

        class _PreservingClient(_RemoteClient):
            async def run_workflow_nowait(self, workflow_name: str, **kwargs: Any) -> _HandlerData:
                self.run_calls.append({"workflow_name": workflow_name, **kwargs})
                return _HandlerData("h1")

            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> _RemoteStream:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                n = len(self.stream_calls)
                return _RemoteStream([_TextEvent(f"turn {n}"), _StopEvent(f"r{n}")])

            async def get_handler(self, handler_id: str) -> _HandlerData:
                return _HandlerData(handler_id, result="ok", context={"saved": True})

        client = _PreservingClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        async for _ in bridge.invoke(AgentTurnInput.from_text("one"), _recorder()):
            pass
        async for _ in bridge.invoke(AgentTurnInput.from_text("two"), _recorder()):
            pass

        # First turn streams from the beginning; the preserved second turn
        # resumes after the first turn's last event instead of replaying it.
        assert client.stream_calls[0]["after_sequence"] == -1
        assert client.stream_calls[1]["after_sequence"] == 1
        # The handler is reused (preserve_context), not recreated.
        assert client.run_calls[1]["handler_id"] == "h1"

    @pytest.mark.asyncio
    async def test_remote_hitl_pause_closes_event_stream(self, fake_workflows_modules):
        """A HITL pause must close the SSE stream, not leak it until cancel."""
        streams: list[_RemoteStream] = []

        class _PausingClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> _RemoteStream:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                stream = _RemoteStream([_InputRequiredEvent(prefix="name?")])
                streams.append(stream)
                return stream

        client = _PausingClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        async for _ in bridge.invoke(AgentTurnInput.from_text("start"), _recorder()):
            pass

        assert streams and streams[0].closed is True

    @pytest.mark.asyncio
    async def test_remote_opaque_stop_subclass_envelope_is_terminal(self, fake_workflows_modules):
        """A server-only StopEvent subclass envelope must not be re-streamed.

        ``load_event()`` fails (the subclass isn't importable client-side) so
        the bridge keeps the envelope; it advertises StopEvent via ``types``
        and must be treated as terminal instead of having its result appended
        as a duplicate text delta after the streamed progress.
        """

        class _OpaqueStopEnvelope:
            type = "acme.CustomStopEvent"
            types = ["workflows.events.StopEvent", "workflows.events.Event"]  # noqa: RUF012 test fake uses shared class fixture
            value = {"result": "the answer"}  # noqa: RUF012 test fake uses shared class fixture
            qualified_name = "acme.CustomStopEvent"

            def load_event(self) -> Any:
                raise RuntimeError("CustomStopEvent not importable on client")

        class _OpaqueClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> Any:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                return _RawStream(
                    [_RemoteEnvelope(_TextEvent("the answer")), _OpaqueStopEnvelope()]
                )

        client = _OpaqueClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        events = []
        async for event in bridge.invoke(AgentTurnInput.from_text("q"), _recorder()):
            events.append(event)

        assert [e.text for e in events if e.kind == "text_delta"] == ["the answer"]
        assert [e.text for e in events if e.kind == "done"] == ["the answer"]

    @pytest.mark.asyncio
    async def test_remote_empty_result_envelope_is_not_spoken(self, fake_workflows_modules):
        """A silent StopEvent() result must not leak envelope repr to TTS."""

        class _EmptyEnvelope:
            type = "StopEvent"
            types = ["workflows.events.StopEvent"]  # noqa: RUF012 test fake uses shared class fixture
            value: dict[str, Any] = {}  # noqa: RUF012 test fake uses shared class fixture
            qualified_name = "workflows.events.StopEvent"

            def load_event(self) -> Any:
                return _StopEvent()

        class _SilentClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> _RemoteStream:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                return _RemoteStream([_StopEvent()])

            async def get_handler(self, handler_id: str) -> _HandlerData:
                return _HandlerData(handler_id, result=_EmptyEnvelope())

        client = _SilentClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        events = []
        async for event in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder()):
            events.append(event)

        assert [e.text for e in events if e.kind == "text_delta"] == []
        assert [e.text for e in events if e.kind == "done"] == [""]

    @pytest.mark.asyncio
    async def test_remote_opaque_input_required_subclass_is_hitl_pause(
        self, fake_workflows_modules
    ):
        """A server-only InputRequiredEvent subclass envelope must pause.

        ``load_event()`` fails (the subclass isn't importable client-side) so
        the bridge keeps the envelope; its ``type`` is the subclass name but
        ``types`` advertises InputRequiredEvent. The bridge must still treat
        it as a HITL prompt and send the next user turn back as a
        HumanResponseEvent instead of starting a fresh workflow run.
        """

        class _OpaqueInputEnvelope:
            type = "acme.CustomInputRequiredEvent"
            types = ["workflows.events.InputRequiredEvent", "workflows.events.Event"]  # noqa: RUF012 test fake uses shared class fixture
            value = {"prefix": "Remote subclass prompt"}  # noqa: RUF012 test fake uses shared class fixture
            qualified_name = "acme.CustomInputRequiredEvent"

            def load_event(self) -> Any:
                raise RuntimeError("CustomInputRequiredEvent not importable on client")

        class _OpaqueInputClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> Any:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                if len(self.stream_calls) == 1:
                    return _RawStream([_OpaqueInputEnvelope()])
                return _RemoteStream(
                    [_TextEvent("Remote "), _TextEvent("done"), _StopEvent("done")]
                )

        client = _OpaqueInputClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        first_turn = []
        async for event in bridge.invoke(AgentTurnInput.from_text("start"), _recorder()):
            first_turn.append(event)

        assert [e.text for e in first_turn if e.kind == "text_delta"] == ["Remote subclass prompt"]
        # The prompt paused the run; no new workflow was started for it.
        assert len(client.run_calls) == 1

        second_turn = []
        async for event in bridge.invoke(AgentTurnInput.from_text("Ada"), _recorder()):
            second_turn.append(event)

        # The pause resumed the same handler with a HumanResponseEvent
        # instead of running the workflow again.
        assert len(client.run_calls) == 1
        assert client.sent_events[-1][0] == "h1"
        assert client.sent_events[-1][1].response == "Ada"
        assert [e.text for e in second_turn if e.kind == "done"] == ["Remote done"]

    @pytest.mark.asyncio
    async def test_remote_result_envelope_unwrapped_for_structured_output(
        self, fake_workflows_modules
    ):
        """structured_output must be StopEvent.result, not the envelope.

        The real WorkflowClient stores the final StopEvent wrapped in an
        EventEnvelope; local mode exposes the StopEvent's raw ``result``.
        Remote mode must match by unwrapping the envelope rather than
        leaking serialization metadata into structured_output.
        """

        class _StructuredClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> _RemoteStream:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                return _RemoteStream([_StopEvent("ignored")])

            async def get_handler(self, handler_id: str) -> _HandlerData:
                envelope = _RemoteEnvelope(_StopEvent(result={"score": 0.9}))
                return _HandlerData(handler_id, result=envelope)

        client = _StructuredClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        done = [
            event
            async for event in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder())
            if event.kind == "done"
        ]

        assert len(done) == 1
        assert done[0].structured_output == {"score": 0.9}

    @pytest.mark.asyncio
    async def test_remote_opaque_stop_subclass_result_unwrapped(self, fake_workflows_modules):
        """structured_output must be the workflow result for an opaque
        StopEvent subclass too.

        ``load_event()`` fails (the subclass is only importable on the
        workflow server) so the envelope is kept; ``_is_stop_event()`` still
        matches it via ``types``. The real return value lives in the
        envelope's serialized ``value['result']`` -- it must be dug out
        instead of leaking the envelope's serialization metadata.
        """

        class _OpaqueStopEnvelope:
            type = "acme.CustomStopEvent"
            types = ["workflows.events.StopEvent", "workflows.events.Event"]  # noqa: RUF012 test fake uses shared class fixture
            value = {"result": {"score": 0.9}}  # noqa: RUF012 test fake uses shared class fixture
            qualified_name = "acme.CustomStopEvent"

            def load_event(self) -> Any:
                raise RuntimeError("CustomStopEvent not importable on client")

        class _OpaqueResultClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> _RemoteStream:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                return _RemoteStream([_StopEvent("ignored")])

            async def get_handler(self, handler_id: str) -> _HandlerData:
                return _HandlerData(handler_id, result=_OpaqueStopEnvelope())

        client = _OpaqueResultClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        done = [
            event
            async for event in bridge.invoke(AgentTurnInput.from_text("hi"), _recorder())
            if event.kind == "done"
        ]

        assert len(done) == 1
        assert done[0].structured_output == {"score": 0.9}

    @pytest.mark.asyncio
    async def test_remote_early_generator_close_cancels_handler(self, fake_workflows_modules):
        """Closing invoke() mid-stream (GeneratorExit) must cancel the remote
        handler so the workflow does not keep running server-side."""
        stream = _BlockingRemoteStream()

        class _BlockingRemoteClient(_RemoteClient):
            def get_workflow_events(self, handler_id: str, **kwargs: Any) -> Any:
                self.stream_calls.append({"handler_id": handler_id, **kwargs})
                return stream

        client = _BlockingRemoteClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        agen = bridge.invoke(AgentTurnInput.from_text("hi"), _recorder())
        first = await agen.__anext__()
        assert first.kind == "text_delta" and first.text == "remote partial"

        await asyncio.wait_for(agen.aclose(), timeout=2.0)

        assert client.cancelled == ["h1"]
        assert stream.closed is True

    @pytest.mark.asyncio
    async def test_reset_cancels_paused_remote_hitl_handler(self, fake_workflows_modules):
        """reset() after a remote HITL pause must cancel the server-side
        handler, not just drop the handler id and leak the paused workflow."""
        client = _RemoteHitlClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        async for _ in bridge.invoke(AgentTurnInput.from_text("start"), _recorder()):
            pass
        assert bridge.snapshot_state().fields["waiting_for_input"] is True

        bridge.reset()
        cleanup_tasks = bridge._reset_cleanup_task_scope.tasks()
        assert [task.get_name() for task in cleanup_tasks] == [
            "easycat-llama-reset-remote-handler"
        ]
        await asyncio.gather(*cleanup_tasks)

        assert client.cancelled == ["h1"]
        assert bridge.snapshot_state().fields["waiting_for_input"] is False

    @pytest.mark.asyncio
    async def test_aclose_cancels_paused_remote_hitl_handler(self, fake_workflows_modules):
        """aclose() (the session stop/shutdown path) must cancel the paused
        server-side handler too, not just reset() between turns."""
        client = _RemoteHitlClient()
        bridge = LlamaAgentsBridge(client=client, workflow_name="greet")

        async for _ in bridge.invoke(AgentTurnInput.from_text("start"), _recorder()):
            pass
        assert bridge.snapshot_state().fields["waiting_for_input"] is True

        await asyncio.wait_for(bridge.aclose(), timeout=2.0)

        assert client.cancelled == ["h1"]
        assert bridge.snapshot_state().fields["waiting_for_input"] is False
