From 8e5cd079b0bf572d469eafe41a849d9c96ab5882 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 21:11:26 +0000 Subject: [PATCH 1/2] fix(opencode-agents): own the HTTP client Callers pass base_url for a running opencode serve. The handler opens its own client, so a caller cannot pass an httpx client or transport. Co-authored-by: alexis --- packages/opencode-agents/README.md | 2 +- packages/opencode-agents/agents.md | 2 +- .../handler.py | 24 +- .../launchdarkly_ai_opencode_agents/server.py | 11 +- .../opencode-agents/tests/test_handler.py | 343 ++++++++++-------- .../opencode-agents/tests/test_sdk_usage.py | 20 +- 6 files changed, 200 insertions(+), 202 deletions(-) diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index b88866b7..ec941cf1 100644 --- a/packages/opencode-agents/README.md +++ b/packages/opencode-agents/README.md @@ -28,4 +28,4 @@ async def main() -> None: asyncio.run(main()) ``` -`base_url` is a running `opencode serve`. Each call opens its own session, including a judge routed to this handler. A judge runs as a full OpenCode agent: the server's tools, in the server's working directory, apply to the judge prompt. There is no tool-less judge mode. The handler yields AG-UI text and tool events. `config().stream()` yields `chunk` then `done`. +`base_url` is a running `opencode serve`. The handler owns every HTTP call to that server. A caller does not pass an HTTP client or transport. Each call opens its own session, including a judge routed to this handler. A judge runs as a full OpenCode agent: the server's tools, in the server's working directory, apply to the judge prompt. There is no tool-less judge mode. The handler yields AG-UI text and tool events. `config().stream()` yields `chunk` then `done`. diff --git a/packages/opencode-agents/agents.md b/packages/opencode-agents/agents.md index 49457ebd..38721fee 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -30,7 +30,7 @@ A run opens one `invoke_agent` span and sets `launchdarkly.operation.type`. The ## Call shape -`create_opencode_handler(*, base_url, capture_content=None, model=None, agent=None, client=None)` returns a `ProviderHandler`. `base_url` is a running `opencode serve`. Each call opens its own session. +`create_opencode_handler(*, base_url, capture_content=None, model=None, agent=None)` returns a `ProviderHandler`. `base_url` is a running `opencode serve`. The handler owns the HTTP client. Each call opens its own session. Assistant text is emitted only after `message.updated` marks that message `role == "assistant"`. A part snapshot is the full text of that part so far; a later delta appends. A text delta without `partID` is dropped. diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py index f1e35fd3..80ac2701 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -11,8 +11,6 @@ from contextlib import aclosing from typing import Any -import httpx - from launchdarkly_ai_server import ( AiConfigRep, ProviderHandler, @@ -493,17 +491,17 @@ def create_opencode_handler( capture_content: bool | None = None, model: ModelResolver | None = None, agent: str | None = None, - client: httpx.AsyncClient | None = None, ) -> ProviderHandler: """Return a handler that opens one OpenCode session per call. ``provides_for`` is ``('*', 'agent')``. *base_url* is a running - ``opencode serve``. A judge routed here gets its own session and runs as a - full OpenCode agent: the server's tools, in the server's working directory, - apply to the judge prompt too. This package does not offer a tool-less - judge agent. A Vega-local handler was rejected: the session boundary - belongs to this package. *model* overrides provider and model selection. - *client* is for tests. + ``opencode serve``. The handler opens its own HTTP client. A caller does + not pass a client or a transport. A judge routed here gets its own session + and runs as a full OpenCode agent: the server's tools, in the server's + working directory, apply to the judge prompt too. This package does not + offer a tool-less judge agent. A Vega-local handler was rejected: the + session boundary belongs to this package. *model* overrides provider and + model selection. """ if not isinstance(base_url, str) or not base_url: raise TypeError("The OpenCode handler requires a base_url.") @@ -513,7 +511,6 @@ def create_opencode_handler( capture_content=capture_content, model=model or default_model, agent=agent, - client=client, ) @@ -533,7 +530,6 @@ def opencode_agents( raise TypeError("The OpenCode handler requires a base_url.") model = kwargs.pop("model", None) agent = kwargs.pop("agent", None) - client = kwargs.pop("client", None) return _config( key=config_key, handler=_create_opencode_handler( @@ -541,7 +537,6 @@ def opencode_agents( capture_content=capture_content, model=model or default_model, agent=agent, - client=client, ), **kwargs, ).invoke(user_input, context, variables, history) @@ -553,13 +548,11 @@ def _create_opencode_handler( capture_content: bool | None, model: ModelResolver, agent: str | None, - client: httpx.AsyncClient | None, ) -> ProviderHandler: content_override = capture_content bound_url = base_url resolve_model = model bound_agent = agent - bound_client = client async def _stream_impl( config: AiConfigRep, @@ -594,7 +587,6 @@ async def _stream_impl( provider_id=provider_id, model_id=model_id, agent=bound_agent, - client=bound_client, capture=capture, ) ) as reduced: @@ -669,7 +661,6 @@ async def _reduce_session( provider_id: str, model_id: str, agent: str | None, - client: httpx.AsyncClient | None, capture: bool, ) -> AsyncGenerator[dict[str, Any], None]: """Reduce the session ``POST /session`` created. The reducer is unchanged.""" @@ -684,7 +675,6 @@ async def _reduce_session( model_id=model_id, agent=agent, session_ids=session_ids, - client=client, ) ) as raw_events: async for raw in raw_events: diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py index d86f314e..09b02834 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py @@ -60,17 +60,17 @@ async def session_events( model_id: str, agent: str | None, session_ids: list[str], - client: httpx.AsyncClient | None = None, ) -> AsyncGenerator[dict[str, Any], None]: """Open one session and yield its bus events. - The event stream is opened before the message post. The post runs + The handler owns the HTTP client. Callers pass ``base_url`` for a server + that is already running and do not supply a client or a transport. The + event stream is opened before the message post. The post runs concurrently with the read. This session's ``session.idle`` ends the turn. A failed message POST raises without waiting for another bus frame. Leaving early cancels the post and aborts the session. """ - owns_client = client is None - http = client or httpx.AsyncClient( + http = httpx.AsyncClient( timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS, read=MESSAGE_TIMEOUT_SECONDS) ) base = base_url.rstrip("/") @@ -127,8 +127,7 @@ async def session_events( ) except Exception: pass - if owns_client: - await http.aclose() + await http.aclose() async def _read_until_done( diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 9cc37ad7..069f410b 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -4,12 +4,16 @@ import asyncio import json +import select +import socket +import threading +import time from collections.abc import Iterator from contextlib import aclosing +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any from unittest.mock import AsyncMock, MagicMock -import httpx import pytest import launchdarkly_ai_server.lifecycle as lifecycle_module @@ -108,48 +112,110 @@ def assistant_completed( return bus("message.updated", info=info) -class Script: - """One fake `opencode serve` over ``httpx.MockTransport``.""" +class _Loopback(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True - def __init__(self, events: list[dict[str, Any]]) -> None: - self.events = events + +class _Request(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, fmt: str, *args: Any) -> None: + return + + def do_GET(self) -> None: + script: _Serve = self.server.script # type: ignore[attr-defined] + if self.path.split("?", 1)[0] != "/event": + self.send_error(404) + return + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + try: + script.serve_events(self) + self._chunk(b"") + except (BrokenPipeError, ConnectionResetError): + return + + def do_POST(self) -> None: + script: _Serve = self.server.script # type: ignore[attr-defined] + length = int(self.headers.get("Content-Length", "0") or "0") + body = self.rfile.read(length) if length else b"" + status, payload = script.serve_post(self.path.split("?", 1)[0], body) + raw = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def _chunk(self, data: bytes) -> None: + self.wfile.write(f"{len(data):X}\r\n".encode() + data + b"\r\n") + self.wfile.flush() + + +class _Serve: + """A loopback ``opencode serve``. The handler uses its own HTTP client.""" + + def __init__(self) -> None: self.sessions: list[str] = [] self.messages: list[dict[str, Any]] = [] self.aborts: list[str] = [] + self.base_url = "" + self._httpd: _Loopback | None = None + + def start(self) -> str: + httpd = _Loopback(("127.0.0.1", 0), _Request) + httpd.script = self # type: ignore[attr-defined] + self._httpd = httpd + threading.Thread(target=httpd.serve_forever, daemon=True).start() + host, port = httpd.server_address + self.base_url = f"http://{host}:{port}" + _servers.append(self) + return self.base_url + + def stop(self) -> None: + if self._httpd is not None: + self._httpd.shutdown() + self._httpd.server_close() + self._httpd = None + + def serve_events(self, request: _Request) -> None: + return + + def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: + if path == "/session": + session_id = "ses" if not self.sessions else f"ses-{len(self.sessions) + 1}" + self.sessions.append(session_id) + return 200, {"id": session_id} + if path.endswith("/message"): + self.messages.append(json.loads(body.decode()) if body else {}) + return 200, {} + if path.endswith("/abort"): + self.aborts.append(path) + return 200, {} + return 404, {} + + +_servers: list[_Serve] = [] + + +class Script(_Serve): + """One finite bus. The connection closes after the scripted events.""" - def transport(self) -> httpx.MockTransport: - script = self - - async def handle(request: httpx.Request) -> httpx.Response: - path = request.url.path - if path == "/event" and request.method == "GET": - body = "".join( - f"data: {json.dumps(event)}\n\n" for event in script.events - ) - return httpx.Response(200, text=body) - if path == "/session" and request.method == "POST": - session_id = ( - "ses" if not script.sessions else f"ses-{len(script.sessions) + 1}" - ) - script.sessions.append(session_id) - return httpx.Response(200, json={"id": session_id}) - if path.endswith("/message") and request.method == "POST": - script.messages.append(json.loads(request.content.decode())) - return httpx.Response(200, json={}) - if path.endswith("/abort") and request.method == "POST": - script.aborts.append(path) - return httpx.Response(200, json={}) - return httpx.Response(404) + def __init__(self, events: list[dict[str, Any]]) -> None: + super().__init__() + self.events = events - return httpx.MockTransport(handle) + def serve_events(self, request: _Request) -> None: + for event in self.events: + request._chunk(f"data: {json.dumps(event)}\n\n".encode()) def playing(events: list[dict[str, Any]], **kwargs: Any) -> tuple[Any, Script]: script = Script(events) - client = httpx.AsyncClient(transport=script.transport(), base_url="http://opencode") - handler = create_opencode_handler( - base_url="http://opencode", client=client, **kwargs - ) + handler = create_opencode_handler(base_url=script.start(), **kwargs) return handler, script @@ -177,6 +243,12 @@ def _reset_sdk() -> Iterator[None]: lifecycle_module._reset_for_testing() yield lifecycle_module._reset_for_testing() + for server in _servers: + release = getattr(server, "release", None) + if release is not None: + release.set() + server.stop() + _servers.clear() @pytest.fixture @@ -764,6 +836,11 @@ def test_base_url_is_required(self) -> None: create_opencode_handler(base_url="") with pytest.raises(TypeError, match="base_url"): opencode_agents("flag", "hi", CONTEXT) + with pytest.raises(TypeError): + create_opencode_handler( + base_url="http://127.0.0.1:9", + client=object(), # type: ignore[call-arg] + ) async def test_the_first_session_on_the_stream_is_not_adopted(self) -> None: events = await collect( @@ -1098,15 +1175,11 @@ async def test_the_wrapper_posts_history_on_a_fresh_session( assistant_completed("m1"), ] ) - client = httpx.AsyncClient( - transport=script.transport(), base_url="http://opencode" - ) result = await opencode_agents( "flag", "Hello", CONTEXT, - base_url="http://opencode", - client=client, + base_url=script.start(), history=[{"role": "user", "content": "earlier"}], ) assert result.response == "Hi" @@ -1275,10 +1348,7 @@ def set_span_in_context(self, span: _Span) -> Any: async def test_a_failed_message_post_does_not_wait_for_the_bus(self) -> None: script = _OpenBus(message_status=400) - client = httpx.AsyncClient( - transport=script.transport(), base_url="http://opencode" - ) - handler = create_opencode_handler(base_url="http://opencode", client=client) + handler = create_opencode_handler(base_url=script.start()) async def _run() -> None: async for _event in await handler.stream( @@ -1303,10 +1373,7 @@ async def test_a_cancelled_sse_read_is_awaited_before_the_stream_closes( self, ) -> None: bus = _CloseAfterRead() - client = httpx.AsyncClient( - transport=bus.transport(), base_url="http://opencode" - ) - handler = create_opencode_handler(base_url="http://opencode", client=client) + handler = create_opencode_handler(base_url=bus.start()) async def _run() -> None: async for _event in await handler.stream( @@ -1327,14 +1394,10 @@ async def _run() -> None: await asyncio.wait_for(_run(), timeout=2) assert bus.read_finished.is_set() assert bus.closed - assert not bus.closed_before_read_finished async def test_breaking_aborts_without_waiting_for_gc(self) -> None: script = _OpenBus() - client = httpx.AsyncClient( - transport=script.transport(), base_url="http://opencode" - ) - handler = create_opencode_handler(base_url="http://opencode", client=client) + handler = create_opencode_handler(base_url=script.start()) async def _run() -> None: async with aclosing( @@ -1358,12 +1421,7 @@ async def _run() -> None: async def test_a_late_completion_survives_the_post_returning(self) -> None: script = _OpenBus(delay_tail=0.2) - client = httpx.AsyncClient( - transport=script.transport(), base_url="http://opencode" - ) - handler = create_opencode_handler( - base_url="http://opencode", client=client, capture_content=True - ) + handler = create_opencode_handler(base_url=script.start(), capture_content=True) seen = [ event async for event in await handler.stream( @@ -1393,122 +1451,89 @@ async def test_an_idle_without_a_session_id_does_not_end_the_run(self) -> None: assert events[-1]["output"] == "Hi!" -class _CloseAfterRead(httpx.AsyncByteStream): +class _CloseAfterRead(_Serve): """``/event`` stays open. The message POST fails while the next read is blocked.""" def __init__(self) -> None: - self.second_read = asyncio.Event() - self.read_finished = asyncio.Event() + super().__init__() + self.second_read = threading.Event() + self.read_finished = threading.Event() self.closed = False - self.closed_before_read_finished = False - - def transport(self) -> httpx.MockTransport: - stream = self - - async def handle(request: httpx.Request) -> httpx.Response: - path = request.url.path - if path == "/event" and request.method == "GET": - return httpx.Response(200, stream=stream) - if path == "/session" and request.method == "POST": - return httpx.Response(200, json={"id": "ses"}) - if path.endswith("/message") and request.method == "POST": - await stream.second_read.wait() - return httpx.Response(400, json={}) - if path.endswith("/abort") and request.method == "POST": - return httpx.Response(200, json={}) - return httpx.Response(404) - - return httpx.MockTransport(handle) - - async def __aiter__(self) -> Any: - yield b":\n\n" + + def serve_events(self, request: _Request) -> None: + request._chunk(b":\n\n") self.second_read.set() - try: - await asyncio.Event().wait() - except asyncio.CancelledError: + sock = request.connection + while True: + readable, _, _ = select.select([sock], [], [], 0.05) + if not readable: + continue + try: + peeked = sock.recv(1, socket.MSG_PEEK) + except OSError: + peeked = b"" + if peeked: + continue self.read_finished.set() - raise + self.closed = True + return - async def aclose(self) -> None: - if not self.read_finished.is_set(): - self.closed_before_read_finished = True - self.closed = True + def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: + if path.endswith("/message"): + self.second_read.wait(timeout=2) + self.messages.append(json.loads(body.decode()) if body else {}) + return 400, {} + return super().serve_post(path, body) -class _ByteStream(httpx.AsyncByteStream): - def __init__(self, chunks: Any) -> None: - self._chunks = chunks - - async def __aiter__(self) -> Any: - async for chunk in self._chunks: - yield chunk - - async def aclose(self) -> None: - return None - - -class _OpenBus: +class _OpenBus(_Serve): """An `/event` stream that stays open, unlike a finite ``Script`` body.""" def __init__(self, *, message_status: int = 200, delay_tail: float = 0) -> None: + super().__init__() self.message_status = message_status self.delay_tail = delay_tail - self.sessions: list[str] = [] - self.messages: list[dict[str, Any]] = [] - self.aborts: list[str] = [] - self.release = asyncio.Event() - - def transport(self) -> httpx.MockTransport: - script = self - - async def handle(request: httpx.Request) -> httpx.Response: - path = request.url.path - if path == "/event" and request.method == "GET": - - async def chunks() -> Any: - # A rejected message sends nothing. Events here would let - # the loop notice the 400 without racing the POST. - head = ( - [] - if script.message_status >= 400 - else [ - assistant_open("m1"), - text_delta("m1", "p1", "Hi"), - ] - ) - for event in head: - yield f"data: {json.dumps(event)}\n\n".encode() - if script.delay_tail: - await asyncio.sleep(script.delay_tail) - tail = [ - text_delta("m1", "p1", "!", session_id="ses"), - assistant_completed("m1", tokens={"input": 1, "output": 4}), - bus("session.idle", sessionID="ses"), - ] - for event in tail: - yield f"data: {json.dumps(event)}\n\n".encode() - return - await script.release.wait() - - return httpx.Response(200, stream=_ByteStream(chunks())) - if path == "/session" and request.method == "POST": - script.sessions.append("ses") - return httpx.Response(200, json={"id": "ses"}) - if path.endswith("/message") and request.method == "POST": - script.messages.append(json.loads(request.content.decode())) - if script.message_status >= 400: - return httpx.Response(script.message_status, json={}) - if script.delay_tail: - return httpx.Response(200, json={}) - await script.release.wait() - return httpx.Response(200, json={}) - if path.endswith("/abort") and request.method == "POST": - script.aborts.append(path) - script.release.set() - return httpx.Response(200, json={}) - return httpx.Response(404) - - return httpx.MockTransport(handle) + self.release = threading.Event() + + def serve_events(self, request: _Request) -> None: + # A rejected message sends nothing. Events here would let the loop + # notice the 400 without racing the POST. + head = ( + [] + if self.message_status >= 400 + else [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + ] + ) + for event in head: + request._chunk(f"data: {json.dumps(event)}\n\n".encode()) + if self.delay_tail: + time.sleep(self.delay_tail) + tail = [ + text_delta("m1", "p1", "!", session_id="ses"), + assistant_completed("m1", tokens={"input": 1, "output": 4}), + bus("session.idle", sessionID="ses"), + ] + for event in tail: + request._chunk(f"data: {json.dumps(event)}\n\n".encode()) + return + self.release.wait() + + def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: + if path.endswith("/message"): + self.messages.append(json.loads(body.decode()) if body else {}) + if self.message_status >= 400: + return self.message_status, {} + if self.delay_tail: + return 200, {} + self.release.wait() + return 200, {} + if path.endswith("/abort"): + self.aborts.append(path) + self.release.set() + return 200, {} + return super().serve_post(path, body) async def _drain(stream: Any) -> None: diff --git a/packages/opencode-agents/tests/test_sdk_usage.py b/packages/opencode-agents/tests/test_sdk_usage.py index 28b36e5c..f2d64d9f 100644 --- a/packages/opencode-agents/tests/test_sdk_usage.py +++ b/packages/opencode-agents/tests/test_sdk_usage.py @@ -3,7 +3,6 @@ from typing import Any from unittest.mock import AsyncMock, MagicMock, patch -import httpx import pytest import launchdarkly_ai_server.lifecycle as lifecycle_module @@ -24,19 +23,6 @@ def _fake() -> MagicMock: return client -def _client() -> httpx.AsyncClient: - async def handle(request: httpx.Request) -> httpx.Response: - if request.url.path == "/event": - return httpx.Response(200, text="") - if request.url.path == "/session": - return httpx.Response(200, json={"id": "ses"}) - return httpx.Response(200, json={}) - - return httpx.AsyncClient( - transport=httpx.MockTransport(handle), base_url="http://opencode" - ) - - async def _settle(value: Any) -> None: if hasattr(value, "__await__"): try: @@ -65,7 +51,7 @@ def _usage(client: MagicMock) -> list[Any]: async def test_create_handler_sends_one_usage_event(installed: MagicMock) -> None: - create_opencode_handler(base_url="http://opencode", client=_client()) + create_opencode_handler(base_url="http://127.0.0.1:9") calls = _usage(installed) assert [entry.args[2].get("helper") for entry in calls] == [ "opencode-agents.createOpenCodeHandler" @@ -83,9 +69,7 @@ async def test_create_handler_sends_one_usage_event(installed: MagicMock) -> Non async def test_wrapper_sends_one_usage_event(installed: MagicMock) -> None: - await _settle( - opencode_agents("k", "q", CONTEXT, base_url="http://opencode", client=_client()) - ) + await _settle(opencode_agents("k", "q", CONTEXT, base_url="http://127.0.0.1:9")) calls = _usage(installed) assert [entry.args[2].get("helper") for entry in calls] == [ "opencode-agents.opencodeAgents" From d9bf7f92337fa397b52ee0640bd718f12640de9a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 17:42:08 +0000 Subject: [PATCH 2/2] test(opencode-agents): assert cancelled read finishes before close The loopback fake cannot see that ordering. Record when _next_line finishes and require that before GET /event closes. Poll the loopback server every 0.01s. Co-authored-by: alexis --- .../opencode-agents/tests/test_handler.py | 55 ++++++++++++++----- 1 file changed, 41 insertions(+), 14 deletions(-) diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 069f410b..74f90c48 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -169,7 +169,9 @@ def start(self) -> str: httpd = _Loopback(("127.0.0.1", 0), _Request) httpd.script = self # type: ignore[attr-defined] self._httpd = httpd - threading.Thread(target=httpd.serve_forever, daemon=True).start() + threading.Thread( + target=httpd.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True + ).start() host, port = httpd.server_address self.base_url = f"http://{host}:{port}" _servers.append(self) @@ -1370,10 +1372,37 @@ async def _run() -> None: assert script.aborts async def test_a_cancelled_sse_read_is_awaited_before_the_stream_closes( - self, + self, monkeypatch: pytest.MonkeyPatch ) -> None: + # The loopback socket cannot tell a cancelled read from the close that + # unblocks it. Record the client: ``_next_line`` must finish before + # GET /event is closed. + import httpx + + from launchdarkly_ai_opencode_agents import server as server_mod + from launchdarkly_ai_opencode_agents.server import OpenCodeHTTPError + bus = _CloseAfterRead() handler = create_opencode_handler(base_url=bus.start()) + order: list[str] = [] + real_next_line = server_mod._next_line + + async def _recording_next_line(lines: Any) -> str | None: + try: + return await real_next_line(lines) + finally: + order.append("read_finished") + + monkeypatch.setattr(server_mod, "_next_line", _recording_next_line) + real_aclose = httpx.Response.aclose + + async def _recording_aclose(response: httpx.Response) -> None: + request = response.request + if request.method == "GET" and request.url.path == "/event": + order.append("response_closed") + await real_aclose(response) + + monkeypatch.setattr(httpx.Response, "aclose", _recording_aclose) async def _run() -> None: async for _event in await handler.stream( @@ -1388,12 +1417,15 @@ async def _run() -> None: ): return - from launchdarkly_ai_opencode_agents.server import OpenCodeHTTPError - with pytest.raises(OpenCodeHTTPError, match="400"): await asyncio.wait_for(_run(), timeout=2) - assert bus.read_finished.is_set() - assert bus.closed + close_at = order.index("response_closed") + assert any(name == "read_finished" for name in order) + assert all( + index < close_at + for index, name in enumerate(order) + if name == "read_finished" + ) async def test_breaking_aborts_without_waiting_for_gc(self) -> None: script = _OpenBus() @@ -1457,8 +1489,6 @@ class _CloseAfterRead(_Serve): def __init__(self) -> None: super().__init__() self.second_read = threading.Event() - self.read_finished = threading.Event() - self.closed = False def serve_events(self, request: _Request) -> None: request._chunk(b":\n\n") @@ -1471,12 +1501,9 @@ def serve_events(self, request: _Request) -> None: try: peeked = sock.recv(1, socket.MSG_PEEK) except OSError: - peeked = b"" - if peeked: - continue - self.read_finished.set() - self.closed = True - return + return + if not peeked: + return def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: if path.endswith("/message"):