From 956fc87eef9c165ebb7976531097b02555821a34 Mon Sep 17 00:00:00 2001 From: Alex Fournier Date: Tue, 28 Jul 2026 08:25:25 -0700 Subject: [PATCH] test(relay): gate native Anthropic streaming in e2e Signed-off-by: Alex Fournier --- .github/workflows/tests.yml | 17 ---- .../e2e/test_relay_native_anthropic_stream.py | 92 +++++++++++++++++++ tests/run_agent/test_streaming.py | 89 ------------------ 3 files changed, 92 insertions(+), 106 deletions(-) create mode 100644 tests/e2e/test_relay_native_anthropic_stream.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index c81e7d84b4b..cdae2e037a5 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -101,23 +101,6 @@ jobs: # re-download, keeping the persisted cache small and fast to restore. run: uv cache prune --ci - - name: Run merge-sensitive provider contracts - if: matrix.slice.index == 1 - # Keep provider concurrency and client-lifecycle contracts visible - # outside the duration-balanced file shards. - run: | - source .venv/bin/activate - python -m pytest -q --tb=short \ - tests/agent/test_relay_llm.py::test_anthropic_stream_callbacks_do_not_reenter_captured_context \ - tests/agent/test_relay_llm.py::test_explicit_stream_close_surfaces_provider_close_failure \ - tests/run_agent/test_streaming.py::TestAnthropicStreamCallbacks::test_anthropic_sdk_stream_runs_through_relay_managed_execution \ - tests/run_agent/test_request_client_reuse_abort_races.py::test_relay_managed_close_failure_poisons_request_client - env: - ANTHROPIC_API_KEY: "" - OPENROUTER_API_KEY: "" - OPENAI_API_KEY: "" - NOUS_API_KEY: "" - - name: Run tests (slice ${{ matrix.slice.index }}/${{ inputs.slice_count }}) # Per-file isolation via scripts/run_tests.sh: each test file runs # in its own freshly-spawned `python -m pytest ` subprocess diff --git a/tests/e2e/test_relay_native_anthropic_stream.py b/tests/e2e/test_relay_native_anthropic_stream.py new file mode 100644 index 00000000000..3f1075071e0 --- /dev/null +++ b/tests/e2e/test_relay_native_anthropic_stream.py @@ -0,0 +1,92 @@ +"""Native Anthropic SDK streaming through Relay's managed execution path.""" + +import pytest + + +@pytest.mark.filterwarnings("ignore:Pydantic serializer warnings:UserWarning") +def test_anthropic_sdk_stream_runs_through_relay_managed_execution( + tmp_path, + monkeypatch, +): + anthropic = pytest.importorskip("anthropic") + httpx = pytest.importorskip("httpx") + pytest.importorskip("nemo_relay") + from agent import relay_runtime + from run_agent import AIAgent + + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes-home")) + response_body = b"""event: message_start +data: {"type":"message_start","message":{"id":"msg_test","type":"message","role":"assistant","content":[],"model":"claude-test","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":1,"output_tokens":0}}} + +event: content_block_start +data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + +event: content_block_delta +data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello"}} + +event: content_block_stop +data: {"type":"content_block_stop","index":0} + +event: message_delta +data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":1}} + +event: message_stop +data: {"type":"message_stop"} + +""" + + def respond(request): + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=response_body, + request=request, + ) + + client = anthropic.Anthropic( + api_key="test-key", + http_client=httpx.Client(transport=httpx.MockTransport(respond)), + ) + relay_runtime._reset_for_tests() + agent = AIAgent( + api_key="test-key", + base_url="https://api.anthropic.com", + provider="anthropic", + model="claude-test", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + agent.api_mode = "anthropic_messages" + agent.session_id = "anthropic-relay-session" + agent._interrupt_requested = False + agent._create_request_anthropic_client = lambda *args, **kwargs: client + lease = relay_runtime.SESSION_COORDINATOR.acquire_conversation( + profile_key=relay_runtime.current_profile_key(), + session_id=agent.session_id, + platform="cli", + ) + turn = relay_runtime.SESSION_COORDINATOR.begin_turn( + lease, + turn_id="anthropic-relay-turn", + task_id="anthropic-relay-task", + ) + lease.host.retain_managed_execution("test.anthropic_relay") + + try: + result = agent._interruptible_streaming_api_call( + { + "model": "claude-test", + "max_tokens": 16, + "messages": [{"role": "user", "content": "hi"}], + } + ) + finally: + lease.host.release_managed_execution("test.anthropic_relay") + relay_runtime.SESSION_COORDINATOR.end_turn(turn, outcome="success") + relay_runtime.SESSION_COORDINATOR.release_conversation(lease) + relay_runtime._reset_for_tests() + client.close() + + assert result.content[0].text == "hello" + assert result.stop_reason == "end_turn" diff --git a/tests/run_agent/test_streaming.py b/tests/run_agent/test_streaming.py index 199d1ae355e..0f2fc27a515 100644 --- a/tests/run_agent/test_streaming.py +++ b/tests/run_agent/test_streaming.py @@ -1313,95 +1313,6 @@ class TestAnthropicStreamCallbacks: assert touch_calls.count("receiving stream response") == len(events) mock_stream.close.assert_called_once() - @pytest.mark.filterwarnings("ignore:Pydantic serializer warnings:UserWarning") - def test_anthropic_sdk_stream_runs_through_relay_managed_execution( - self, - tmp_path, - monkeypatch, - ): - anthropic = pytest.importorskip("anthropic") - httpx = pytest.importorskip("httpx") - pytest.importorskip("nemo_relay") - from agent import relay_runtime - from run_agent import AIAgent - - monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes-home")) - response_body = b"""event: message_start -data: {"type":"message_start","message":{"id":"msg_test","type":"message","role":"assistant","content":[],"model":"claude-test","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":1,"output_tokens":0}}} - -event: content_block_start -data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} - -event: content_block_delta -data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello"}} - -event: content_block_stop -data: {"type":"content_block_stop","index":0} - -event: message_delta -data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":1}} - -event: message_stop -data: {"type":"message_stop"} - -""" - - def respond(request): - return httpx.Response( - 200, - headers={"content-type": "text/event-stream"}, - content=response_body, - request=request, - ) - - client = anthropic.Anthropic( - api_key="test-key", - http_client=httpx.Client(transport=httpx.MockTransport(respond)), - ) - relay_runtime._reset_for_tests() - agent = AIAgent( - api_key="test-key", - base_url="https://api.anthropic.com", - provider="anthropic", - model="claude-test", - quiet_mode=True, - skip_context_files=True, - skip_memory=True, - ) - agent.api_mode = "anthropic_messages" - agent.session_id = "anthropic-relay-session" - agent._interrupt_requested = False - agent._create_request_anthropic_client = lambda *args, **kwargs: client - lease = relay_runtime.SESSION_COORDINATOR.acquire_conversation( - profile_key=relay_runtime.current_profile_key(), - session_id=agent.session_id, - platform="cli", - ) - turn = relay_runtime.SESSION_COORDINATOR.begin_turn( - lease, - turn_id="anthropic-relay-turn", - task_id="anthropic-relay-task", - ) - lease.host.retain_managed_execution("test.anthropic_relay") - - try: - result = agent._interruptible_streaming_api_call( - { - "model": "claude-test", - "max_tokens": 16, - "messages": [{"role": "user", "content": "hi"}], - } - ) - finally: - lease.host.release_managed_execution("test.anthropic_relay") - relay_runtime.SESSION_COORDINATOR.end_turn(turn, outcome="success") - relay_runtime.SESSION_COORDINATOR.release_conversation(lease) - relay_runtime._reset_for_tests() - client.close() - - assert result.content[0].text == "hello" - assert result.stop_reason == "end_turn" - @patch("run_agent.AIAgent._rebuild_anthropic_client") @patch("run_agent.AIAgent._replace_primary_openai_client") def test_anthropic_stream_parser_valueerror_retries_before_delivery(