diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index cdae2e037a59..c81e7d84b4bf 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -101,6 +101,23 @@ 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/run_agent/test_request_client_reuse_abort_races.py b/tests/run_agent/test_request_client_reuse_abort_races.py index 73a9c41887cc..76b3bc49946d 100644 --- a/tests/run_agent/test_request_client_reuse_abort_races.py +++ b/tests/run_agent/test_request_client_reuse_abort_races.py @@ -25,7 +25,12 @@ Invariants pinned here: 4. ``run_codex_stream``'s finally must poison the reuse slot when ``event_stream.close()`` fails; otherwise a failed close caches the client with a connection still checked out of the pool. + +5. Relay's managed stream wrapper must preserve the same close failure and + request-client identity rather than masking the signal used to poison the + slot. """ +from contextlib import contextmanager import threading import time from types import SimpleNamespace @@ -49,6 +54,33 @@ def _make_agent(): return agent +@contextmanager +def _managed_relay_turn(agent, tmp_path, monkeypatch): + pytest.importorskip("nemo_relay") + from agent import relay_runtime + + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profile")) + relay_runtime._reset_for_tests() + 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="request-client-reuse-turn", + task_id="request-client-reuse-task", + ) + lease.host.retain_managed_execution("test.request_client_reuse") + try: + yield + finally: + lease.host.release_managed_execution("test.request_client_reuse") + relay_runtime.SESSION_COORDINATOR.end_turn(turn, outcome="success") + relay_runtime.SESSION_COORDINATOR.release_conversation(lease) + relay_runtime._reset_for_tests() + + def _chunk(content=None, finish_reason=None): return SimpleNamespace( choices=[ @@ -154,6 +186,40 @@ def test_worker_interrupt_break_poisons_slot_when_stream_close_fails(): assert "interrupt_stream_close_failed" in abort_reasons +def test_relay_managed_close_failure_poisons_request_client(tmp_path, monkeypatch): + """Relay must preserve close failures needed by the reuse-slot guard.""" + agent = _make_agent() + + def chunks(): + yield _chunk(content="partial ") + agent._interrupt_requested = True + yield _chunk(content="never processed") + + stream = _FakeStream(chunks, close_raises=True) + request_client = _mock_wire_client(stream) + abort_reasons = [] + + with _managed_relay_turn(agent, tmp_path, monkeypatch), patch.object( + agent, "_create_request_openai_client", return_value=request_client + ), patch.object(agent, "_close_request_openai_client"), patch.object( + agent, + "_abort_request_openai_client", + side_effect=lambda client, *, reason: abort_reasons.append( + (client, reason) + ), + ): + with pytest.raises(InterruptedError): + agent._interruptible_streaming_api_call( + { + "model": "test/model", + "messages": [{"role": "user", "content": "hello"}], + } + ) + + assert stream.close_calls == 1 + assert abort_reasons == [(request_client, "interrupt_stream_close_failed")] + + def test_stale_abort_is_atomic_with_holder_read(monkeypatch): """The stranger-thread abort must complete before the worker's finally can pop + cache the client.