From 9671642cca78be5edc4c9976729edcbfb263f531 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 19 Oct 2026 18:47:35 +0000 Subject: [PATCH 1/7] feat(opencode-agents): open one session per call Co-authored-by: alexis --- packages/opencode-agents/pyproject.toml | 1 + .../handler.py | 255 ++++++---- .../launchdarkly_ai_opencode_agents/server.py | 169 ++++++ .../opencode-agents/tests/test_handler.py | 481 +++++++----------- .../opencode-agents/tests/test_sdk_usage.py | 29 +- uv.lock | 2 + 6 files changed, 517 insertions(+), 420 deletions(-) create mode 100644 packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py diff --git a/packages/opencode-agents/pyproject.toml b/packages/opencode-agents/pyproject.toml index cd65e988..ef4323f0 100644 --- a/packages/opencode-agents/pyproject.toml +++ b/packages/opencode-agents/pyproject.toml @@ -4,6 +4,7 @@ version = "0.1.0" requires-python = ">=3.12" dependencies = [ "launchdarkly-ai-server", + "httpx>=0.27", "opentelemetry-api>=1.25", ] description = "OpenCode agent handler for LaunchDarkly AI SDK" 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 f535f766..f83ba41b 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -1,18 +1,16 @@ """Reduce OpenCode session events into handler stream events. -The session is a caller-supplied async iterable of OpenCode bus events. This -module does not start a process and does not construct an OpenCode SDK client. -The handler yields the AG-UI text and tool subset, then one ``done``. -``config().stream()`` still yields ``chunk`` then ``done``. +Each call opens its own session on a running ``opencode serve``. This module +does not start that process. The handler yields the AG-UI text and tool subset, +then one ``done``. ``config().stream()`` still yields ``chunk`` then ``done``. """ from __future__ import annotations -from collections.abc import AsyncGenerator, AsyncIterable, Callable, Mapping +from collections.abc import AsyncGenerator, Mapping from typing import Any -from opentelemetry import context as otel_context - +import httpx from launchdarkly_ai_server import ( AiConfigRep, ProviderHandler, @@ -29,25 +27,14 @@ text_message, ) from launchdarkly_ai_server.agui import TurnEmitter -from launchdarkly_ai_server.conversation import _EVAL_KEY from launchdarkly_ai_server.lifecycle import effective_capture_content from launchdarkly_ai_server.sdk_usage import report_usage from launchdarkly_ai_server.utils import number_or_zero from ._version import PACKAGE_NAME, __version__ +from .server import ModelResolver, default_model, session_events from .spans import fail_span, finish_root_span, mark_ok, start_root_span -Events = Callable[ - [ - AiConfigRep, - str | None, - dict[str, Any] | None, - list[dict[str, Any]] | None, - str | None, - ], - AsyncIterable[Mapping[str, Any]], -] - class OpenCodeSessionError(Exception): """The OpenCode session failed. @@ -465,16 +452,6 @@ def _session_error_message(properties: object) -> str: return f"The OpenCode session failed ({name})." -def _is_judge_evaluation() -> bool: - """Whether this call is a judge evaluation, not the agent's own session. - - ``provides_for`` stays ``('*', 'agent')``, so judge selection still lands - here. The judge prompt is not this OpenCode session. Calling *events* for - it would answer the judge with that session. - """ - return otel_context.get_value(_EVAL_KEY) is not None - - def _input_messages( user_input: str | None, history: list[dict[str, Any]] | None, @@ -505,49 +482,30 @@ def _input_messages( return messages -def _open_events( - events: Events, - config: AiConfigRep, - user_input: str | None, - variables: dict[str, Any] | None, - history: list[dict[str, Any]] | None, - session_id: str | None, -) -> AsyncIterable[Mapping[str, Any]]: - produced = events(config, user_input, variables, history, session_id) - if not hasattr(produced, "__aiter__"): - raise TypeError("OpenCode event source must return an async iterable.") - return produced - - def create_opencode_handler( *, - events: Events, - session_id: str, + base_url: str, capture_content: bool | None = None, + model: ModelResolver | None = None, + agent: str | None = None, + client: httpx.AsyncClient | None = None, ) -> ProviderHandler: - """Return a handler that runs an OpenCode session supplied by *events*. - - ``provides_for`` is ``('*', 'agent')``, so the handler matches an agent-mode - variation for any provider. *events* is called on every invocation as - ``events(config, user_input, variables, history, session_id)`` and must - return a fresh async iterable of that session's bus events. *session_id* - is required. The reducer ignores every other session and does not adopt - the first one it sees. This function does not start OpenCode and does not - construct an OpenCode SDK client. - - The stream yields the AG-UI text and tool subset and one ``done``. - ``config().stream()`` still yields ``chunk`` then ``done``. Pass - ``skip_judges=True`` to ``config()``. Judge selection lands here, because - ``provides_for`` is ``('*', 'agent')``, and the handler does not run the - judge prompt: the call raises without calling *events*, so the judge is - tracked as a failed generation. Do not pass this handler to ``run_judge``. - *capture_content* follows the other handlers. + """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. A Vega-local + handler was rejected: the session boundary belongs to this package. + *model* overrides provider and model selection. *client* is for tests. """ + if not isinstance(base_url, str) or not base_url: + raise TypeError("The OpenCode handler requires a base_url.") report_usage("opencode-agents.createOpenCodeHandler", PACKAGE_NAME, __version__) return _create_opencode_handler( - events=events, - session_id=_require_session_id(session_id), + base_url=base_url, capture_content=capture_content, + model=model or default_model, + agent=agent, + client=client, ) @@ -557,45 +515,41 @@ def opencode_agents( context: dict[str, Any], **kwargs: Any, ) -> Any: - """Convenience wrapper: creates a handler and calls ``config(...).invoke()``. - - *events* is required and is the same callable :func:`create_opencode_handler` - takes. *variables*, *history*, *session_id*, and *capture_content* configure - that call. Pass ``skip_judges=True``. A judge evaluation raises without - calling *events* again, so a judged config tracks the judge as a failed - generation. - """ + """Convenience wrapper: creates a handler and calls ``config(...).invoke()``.""" report_usage("opencode-agents.opencodeAgents", PACKAGE_NAME, __version__) variables = kwargs.pop("variables", None) history = kwargs.pop("history", None) capture_content = kwargs.pop("capture_content", None) - session_id = _require_session_id(kwargs.pop("session_id", None)) - events = kwargs.pop("events") + base_url = kwargs.pop("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( - events=events, - session_id=session_id, + base_url=base_url, capture_content=capture_content, + model=model or default_model, + agent=agent, + client=client, ), **kwargs, ).invoke(user_input, context, variables, history) -def _require_session_id(session_id: object) -> str: - if not isinstance(session_id, str) or not session_id: - raise TypeError("The OpenCode handler requires a session_id.") - return session_id - - def _create_opencode_handler( *, - events: Events, - session_id: str, + base_url: str, capture_content: bool | None, + model: ModelResolver, + agent: str | None, + client: httpx.AsyncClient | None, ) -> ProviderHandler: content_override = capture_content - bound_session = session_id + bound_url = base_url + resolve_model = model + bound_agent = agent + bound_client = client async def _stream_impl( config: AiConfigRep, @@ -604,38 +558,41 @@ async def _stream_impl( variables: dict[str, Any] | None, history: list[dict[str, Any]] | None = None, ) -> AsyncGenerator[dict[str, Any], None]: - if _is_judge_evaluation(): - raise RuntimeError("The OpenCode handler does not run judge prompts.") + del tool_handlers capture = effective_capture_content(content_override) span = start_root_span(config, variables or {}) ended: set[int] = set() - reducer = _Reducer(capture_content=capture, session_id=bound_session) + provider_id, model_id = resolve_model( + config if isinstance(config, dict) else {} + ) + system = _system_text(config) + prompt = _user_prompt(config, user_input, history) try: - instructions = ( - config.get("instructions") if isinstance(config, dict) else None - ) set_input_content_attributes( span, capture, - system_instructions=instructions - if isinstance(instructions, str) - else None, + system_instructions=system, messages=_input_messages(user_input, history), ) - async for event in reducer.reduce( - _open_events( - events, config, user_input, variables, history, bound_session - ) + created = _Created() + async for event in _reduce_session( + created, + base_url=bound_url, + prompt=prompt, + system=system, + provider_id=provider_id, + model_id=model_id, + agent=bound_agent, + client=bound_client, + capture=capture, ): yield event - for event in reducer.close(): - yield event - output = "".join(reducer.deltas) + output = "".join(created.deltas) if output: set_output_content_attributes( span, capture, [text_message("assistant", output)] ) - usage = reducer.usage_payload() + usage = created.usage finish_root_span( span, config, @@ -677,3 +634,97 @@ async def _call_impl( _stream_impl, capture_content=content_override, ) + + +class _Created: + deltas: list[str] + usage: dict[str, int] | None + + def __init__(self) -> None: + self.deltas = [] + self.usage = None + + +async def _reduce_session( + created: _Created, + *, + base_url: str, + prompt: str, + system: str | None, + 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.""" + session_ids: list[str] = [] + reducer: _Reducer | None = None + async for raw in session_events( + base_url=base_url, + prompt=prompt, + system=system, + provider_id=provider_id, + model_id=model_id, + agent=agent, + session_ids=session_ids, + client=client, + ): + if reducer is None: + reducer = _Reducer(capture_content=capture, session_id=session_ids[0]) + async for event in reducer.reduce(_once(raw)): + yield event + if reducer is None: + return + for event in reducer.close(): + yield event + created.deltas = list(reducer.deltas) + created.usage = reducer.usage_payload() + + +async def _once(raw: dict[str, Any]) -> AsyncGenerator[dict[str, Any], None]: + yield raw + + +def _system_text(config: AiConfigRep) -> str | None: + if not isinstance(config, dict): + return None + instructions = config.get("instructions") + if isinstance(instructions, str) and instructions.strip(): + return instructions + return None + + +def _user_prompt( + config: AiConfigRep, + user_input: str | None, + history: list[dict[str, Any]] | None, +) -> str: + messages: list[dict[str, Any]] = [] + if isinstance(config, dict): + raw_messages = config.get("messages") + if isinstance(raw_messages, list): + for message in raw_messages: + if isinstance(message, dict) and message.get("role") != "system": + messages.append(message) + turns = compose_history( + history=[item for item in (history or []) if isinstance(item, dict)], + user_input=user_input, + config_messages=messages, + ) + lines: list[str] = [] + for turn in turns: + role = turn.get("role") + text = content_to_text(turn.get("content") or "") + if not text: + continue + label = "Assistant" if role == "assistant" else "User" + lines.append(f"{label}: {text}") + if ( + len(lines) == 1 + and lines[0].startswith("User: ") + and not messages + and not history + ): + return lines[0].removeprefix("User: ") + return "\n\n".join(lines) diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py new file mode 100644 index 00000000..936331b0 --- /dev/null +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py @@ -0,0 +1,169 @@ +"""One OpenCode session per call, over the stable HTTP API. + +``GET /event`` is opened before ``POST /session/{id}/message``. The message +route is the stable API. ``/api/session/{id}/prompt`` is not used: that preview +route rejects the amazon-bedrock provider. +""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import AsyncIterator, Callable +from typing import Any + +import httpx + +DRAIN_SECONDS = 0.05 + +ModelResolver = Callable[[dict[str, Any]], tuple[str, str]] + + +class OpenCodeHTTPError(RuntimeError): + """The OpenCode server rejected a session or message call.""" + + +def default_model(config: dict[str, Any]) -> tuple[str, str]: + """Map an AI Config model onto an OpenCode provider id and model id. + + A ``provider/model`` name is split. Otherwise Bedrock and Gemini use + OpenCode's provider ids, and any other provider name is lowercased. + """ + model = config.get("model") if isinstance(config.get("model"), dict) else {} + name = model.get("name") if isinstance(model, dict) else "" + name = name if isinstance(name, str) else "" + provider = ( + config.get("provider") if isinstance(config.get("provider"), dict) else {} + ) + provider_name = provider.get("name") if isinstance(provider, dict) else "" + provider_name = provider_name if isinstance(provider_name, str) else "" + if "/" in name: + provider_id, model_id = name.split("/", 1) + return provider_id, model_id + mapped = {"Bedrock": "amazon-bedrock", "Gemini": "google"} + return mapped.get(provider_name, provider_name.lower()), name + + +async def session_events( + *, + base_url: str, + prompt: str, + system: str | None, + provider_id: str, + model_id: str, + agent: str | None, + session_ids: list[str], + client: httpx.AsyncClient | None = None, +) -> AsyncIterator[dict[str, Any]]: + """Open one session and yield its bus events. + + The event stream is opened before the message post. The post runs + concurrently with the read. ``session.idle`` ends the turn. A returned + post is followed by a brief drain. Leaving early cancels the post and + aborts the session. + """ + owns_client = client is None + http = client or httpx.AsyncClient(timeout=None) + base = base_url.rstrip("/") + post: asyncio.Task[httpx.Response] | None = None + session_id: str | None = None + succeeded = False + try: + async with http.stream("GET", f"{base}/event") as sse: + sse.raise_for_status() + created = await http.post(f"{base}/session") + if created.status_code >= 400: + raise OpenCodeHTTPError(f"POST /session -> {created.status_code}") + session_id = _session_id(created.json()) + session_ids.append(session_id) + body: dict[str, Any] = { + "parts": [{"type": "text", "text": prompt}], + "model": {"providerID": provider_id, "modelID": model_id}, + } + if system: + body["system"] = system + if agent: + body["agent"] = agent + post = asyncio.create_task( + http.post(f"{base}/session/{session_id}/message", json=body) + ) + async for event in _read_until_done(sse, post, session_id): + yield event + response = await post + if response.status_code >= 400: + raise OpenCodeHTTPError( + f"POST /session/{session_id}/message -> {response.status_code}" + ) + succeeded = True + finally: + if not succeeded and session_id is not None: + if post is not None and not post.done(): + post.cancel() + try: + await http.post(f"{base}/session/{session_id}/abort") + except Exception: + pass + if owns_client: + await http.aclose() + + +async def _read_until_done( + response: httpx.Response, + post: asyncio.Task[httpx.Response], + session_id: str, +) -> AsyncIterator[dict[str, Any]]: + lines = response.aiter_lines() + while True: + timeout = DRAIN_SECONDS if post.done() else None + try: + if timeout is None: + line = await lines.__anext__() + else: + line = await asyncio.wait_for(lines.__anext__(), timeout) + except StopAsyncIteration: + return + except TimeoutError: + return + event = _parse_frame(line) + if event is None: + continue + yield event + if event.get("type") != "session.idle": + continue + owner = _frame_session(event) + if owner is None or owner == session_id: + return + + +def _session_id(body: object) -> str: + if isinstance(body, dict): + raw = body.get("id") + if isinstance(raw, str) and raw: + return raw + data = body.get("data") + if isinstance(data, dict): + nested = data.get("id") + if isinstance(nested, str) and nested: + return nested + raise OpenCodeHTTPError("session create returned no id") + + +def _parse_frame(line: str) -> dict[str, Any] | None: + text = line.strip() + if text.startswith("data:"): + text = text[5:].lstrip() + if not text or text[0] != "{": + return None + try: + parsed = json.loads(text) + except json.JSONDecodeError: + return None + return parsed if isinstance(parsed, dict) else None + + +def _frame_session(event: dict[str, Any]) -> str | None: + properties = event.get("properties") + if not isinstance(properties, dict): + return None + raw = properties.get("sessionID") + return raw if isinstance(raw, str) and raw else None diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index e7084af6..aed8cdf5 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -3,10 +3,13 @@ from __future__ import annotations import json -from collections.abc import AsyncIterator, Iterator +from collections.abc import Iterator from typing import Any from unittest.mock import AsyncMock, MagicMock +import json + +import httpx import pytest import launchdarkly_ai_server.lifecycle as lifecycle_module @@ -15,6 +18,7 @@ create_opencode_handler, opencode_agents, ) +from launchdarkly_ai_opencode_agents.server import default_model from launchdarkly_ai_server import SDKOptions, config, configure CONTEXT = {"kind": "user", "key": "user-1"} @@ -104,26 +108,63 @@ def assistant_completed( return bus("message.updated", info=info) -def source(events: list[dict[str, Any]]) -> Any: - async def _events( - config: Any, - user_input: str | None, - variables: Any, - history: Any, - session_id: str | None, - ) -> AsyncIterator[dict[str, Any]]: - del config, user_input, variables, history, session_id - for event in events: - yield event +class Script: + """One fake `opencode serve` over ``httpx.MockTransport``.""" + + def __init__(self, events: list[dict[str, Any]]) -> None: + self.events = events + self.sessions: list[str] = [] + self.messages: list[dict[str, Any]] = [] + self.aborts: list[str] = [] - return _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) + + return httpx.MockTransport(handle) + + +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 + ) + return handler, script async def collect(events: list[dict[str, Any]]) -> list[dict[str, Any]]: - built = create_opencode_handler( - capture_content=True, events=source(events), session_id="ses" + built, _script = playing(events, capture_content=True) + gen = await built.stream( + { + "instructions": "Be brief.", + "model": {"name": "m"}, + "provider": {"name": "Acme"}, + }, + "hi", + None, + None, ) - gen = await built.stream({}, "hi", None, None) return [event async for event in gen] @@ -150,18 +191,23 @@ def mock_ld_client() -> Iterator[MagicMock]: lifecycle_module._reset_for_testing() -def judge_errors(client: MagicMock) -> int: - return sum( - 1 - for call in client.track.call_args_list - if call.args[0] == "$ld:ai:generation:error" - and call.args[2].get("configKey") == "judge-flag" - ) - - class TestSurface: + def test_model_names_map_onto_opencode_providers(self) -> None: + assert default_model( + {"provider": {"name": "Bedrock"}, "model": {"name": "claude"}} + ) == ("amazon-bedrock", "claude") + assert default_model( + {"provider": {"name": "Gemini"}, "model": {"name": "gemini-2"}} + ) == ("google", "gemini-2") + assert default_model( + { + "provider": {"name": "Anthropic"}, + "model": {"name": "amazon-bedrock/claude"}, + } + ) == ("amazon-bedrock", "claude") + def test_provides_for_is_any_provider_in_agent_mode(self) -> None: - handler = create_opencode_handler(events=source([]), session_id="ses") + handler, _script = playing([]) assert handler.provides_for == ("*", "agent") def test_package_exports_the_handler(self) -> None: @@ -177,10 +223,8 @@ def test_package_exports_the_handler(self) -> None: def test_capture_content_follows_configure(self) -> None: configure(options=SDKOptions(capture_content=True)) - omitted = create_opencode_handler(events=source([]), session_id="ses") - explicit = create_opencode_handler( - capture_content=False, events=source([]), session_id="ses" - ) + omitted, _script = playing([]) + explicit, _explicit_script = playing([], capture_content=False) assert omitted.capture_content is True assert explicit.capture_content is False @@ -479,17 +523,14 @@ async def test_a_completed_message_is_emitted_when_nothing_is_open(self) -> None assert events[-1]["output"] == "KeptAfter" async def test_capture_off_hides_tool_content(self) -> None: - built = create_opencode_handler( + built, _script = playing( + [ + assistant_open("m1"), + tool_part("c1", "running", raw_input={"command": "ls"}), + tool_part("c1", "completed", output="file.txt"), + assistant_completed("m1"), + ], capture_content=False, - session_id="ses", - events=source( - [ - assistant_open("m1"), - tool_part("c1", "running", raw_input={"command": "ls"}), - tool_part("c1", "completed", output="file.txt"), - assistant_completed("m1"), - ] - ), ) gen = await built.stream({}, "hi", None, None) events = [event async for event in gen] @@ -636,24 +677,21 @@ async def test_user_text_is_not_the_assistant_reply(self) -> None: assert "secret prompt" not in str(events) async def test_session_error_raises_without_the_provider_body(self) -> None: - handler = create_opencode_handler( - capture_content=True, - session_id="ses", - events=source( - [ - assistant_open("m1"), - text_delta("m1", "p1", "Hi"), - { - "type": "session.error", - "properties": { - "error": { - "name": "ProviderAuthError", - "data": {"message": "secret-body"}, - } - }, + handler, _script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + { + "type": "session.error", + "properties": { + "error": { + "name": "ProviderAuthError", + "data": {"message": "secret-body"}, + } }, - ] - ), + }, + ], + capture_content=True, ) seen: list[str] = [] with pytest.raises(OpenCodeSessionError, match="session failed") as raised: @@ -668,25 +706,22 @@ class TestStableStream: async def test_a_tool_does_not_close_the_stream( self, mock_ld_client: MagicMock ) -> None: - handler = create_opencode_handler( - session_id="ses", - events=source( - [ - assistant_open("m1"), - text_delta("m1", "p1", "Looking up"), - tool_part("c1", "running", raw_input={"command": "ls"}), - tool_part( - "c1", - "completed", - raw_input={"command": "ls"}, - output="file.txt", - ), - assistant_completed("m1"), - assistant_open("m2"), - text_delta("m2", "p2", "Done."), - assistant_completed("m2"), - ] - ), + handler, _script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Looking up"), + tool_part("c1", "running", raw_input={"command": "ls"}), + tool_part( + "c1", + "completed", + raw_input={"command": "ls"}, + output="file.txt", + ), + assistant_completed("m1"), + assistant_open("m2"), + text_delta("m2", "p2", "Done."), + assistant_completed("m2"), + ] ) seen = [ event @@ -705,15 +740,12 @@ async def test_a_tool_does_not_close_the_stream( async def test_session_error_does_not_yield_stable_done( self, mock_ld_client: MagicMock ) -> None: - handler = create_opencode_handler( - session_id="ses", - events=source( - [ - assistant_open("m1"), - text_delta("m1", "p1", "Hi"), - {"type": "session.error", "properties": {"error": {"name": "X"}}}, - ] - ), + handler, _script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + {"type": "session.error", "properties": {"error": {"name": "X"}}}, + ] ) seen: list[dict[str, Any]] = [] with pytest.raises(OpenCodeSessionError): @@ -724,17 +756,11 @@ async def test_session_error_does_not_yield_stable_done( class TestInvocation: - def test_events_is_required(self) -> None: + def test_base_url_is_required(self) -> None: with pytest.raises(TypeError): create_opencode_handler() # type: ignore[call-arg] - - def test_session_id_is_required(self) -> None: - with pytest.raises(TypeError, match="session_id"): - create_opencode_handler(events=source([])) # type: ignore[call-arg] - with pytest.raises(TypeError, match="session_id"): - create_opencode_handler(events=source([]), session_id="") - with pytest.raises(TypeError, match="session_id"): - create_opencode_handler(events=source([]), session_id=None) # type: ignore[arg-type] + with pytest.raises(TypeError, match="base_url"): + create_opencode_handler(base_url="") async def test_the_first_session_on_the_stream_is_not_adopted(self) -> None: events = await collect( @@ -767,42 +793,34 @@ async def test_the_first_session_on_the_stream_is_not_adopted(self) -> None: assert "their answer" not in str(events) async def test_events_receives_the_prompt_and_the_session(self) -> None: - seen: dict[str, Any] = {} - - async def events( - config: Any, - user_input: str | None, - variables: Any, - history: Any, - session_id: str | None, - ) -> AsyncIterator[dict[str, Any]]: - seen["instructions"] = config.get("instructions") - seen["user_input"] = user_input - seen["variables"] = variables - seen["history"] = history - seen["session_id"] = session_id - yield assistant_open("m1") - yield text_delta("m1", "p1", "Hi") - yield assistant_completed("m1") - - handler = create_opencode_handler(events=events, session_id="ses") + handler, script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ] + ) events_out = [ event async for event in await handler.stream( - {"instructions": "Be brief."}, + { + "instructions": "Be brief.", + "model": {"name": "claude"}, + "provider": {"name": "Anthropic"}, + }, "Hello", None, {"name": "Ada"}, [{"role": "user", "content": "earlier"}], ) ] - assert seen == { - "instructions": "Be brief.", - "user_input": "Hello", - "variables": {"name": "Ada"}, - "history": [{"role": "user", "content": "earlier"}], - "session_id": "ses", - } + posted = script.messages[0] + assert posted["system"] == "Be brief." + assert posted["parts"] == [ + {"type": "text", "text": "User: earlier\n\nUser: Hello"} + ] + assert posted["model"] == {"providerID": "anthropic", "modelID": "claude"} + assert script.sessions == ["ses"] assert events_out[-1]["output"] == "Hi" async def test_other_sessions_are_ignored_and_idle_ends_the_run(self) -> None: @@ -893,7 +911,7 @@ async def test_a_delta_without_part_id_is_not_repeated_by_its_snapshot( assert contents == ["Hi"] assert events[-1]["output"] == "Hi" - async def test_skip_judges_runs_events_once( + async def test_a_judge_opens_its_own_session( self, mock_ld_client: MagicMock ) -> None: judged = { @@ -903,86 +921,45 @@ async def test_skip_judges_runs_events_once( }, } mock_ld_client.variation = AsyncMock(return_value=judged) - calls: list[str | None] = [] - - async def events( - config: Any, - user_input: str | None, - variables: Any, - history: Any, - session_id: str | None, - ) -> AsyncIterator[dict[str, Any]]: - del config, variables, history, session_id - calls.append(user_input) - yield assistant_open("m1") - yield text_delta("m1", "p1", "Hi") - yield assistant_completed("m1") - - handler = create_opencode_handler(events=events, session_id="ses") - skipped = [ - event - async for event in config( - key="flag", handler=handler, skip_judges=True - ).stream("q", CONTEXT) - ] - assert [event["type"] for event in skipped] == ["chunk", "done"] - assert calls == ["q"] - assert not judge_errors(mock_ld_client) - - calls.clear() - judged_run = [ - event - async for event in config(key="flag", handler=handler).stream("q", CONTEXT) - ] - assert calls == ["q"] - assert judged_run[-1]["type"] == "done" - assert judged_run[-1]["response"] == "Hi" - assert not judged_run[-1].get("judge_results") - # Without skip_judges the raise is the safety net: the judge is - # tracked as a failed generation instead of scoring this session. - assert judge_errors(mock_ld_client) == 1 - - async def test_the_wrapper_forwards_variables_history_and_session( + handler, script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ] + ) + await _drain(config(key="flag", handler=handler).stream("q", CONTEXT)) + assert len(script.sessions) == 2 + assert script.sessions[0] != script.sessions[1] + prompts = [message["parts"][0]["text"] for message in script.messages] + assert prompts[0] == "q" + assert prompts[1] != "q" + + async def test_the_wrapper_posts_history_on_a_fresh_session( self, mock_ld_client: MagicMock ) -> None: - seen: dict[str, Any] = {} - - async def events( - config: Any, - user_input: str | None, - variables: Any, - history: Any, - session_id: str | None, - ) -> AsyncIterator[dict[str, Any]]: - seen["user_input"] = user_input - seen["name"] = ( - variables.get("name") if isinstance(variables, dict) else None - ) - seen["history"] = history - seen["session_id"] = session_id - seen["instructions"] = ( - config.get("instructions") if isinstance(config, dict) else None - ) - yield assistant_open("m1") - yield text_delta("m1", "p1", "Hi") - yield assistant_completed("m1") - + script = Script( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ] + ) + client = httpx.AsyncClient( + transport=script.transport(), base_url="http://opencode" + ) result = await opencode_agents( "flag", "Hello", CONTEXT, - events=events, - session_id="ses", - variables={"name": "Ada"}, + base_url="http://opencode", + client=client, history=[{"role": "user", "content": "earlier"}], - skip_judges=True, ) assert result.response == "Hi" - assert seen["user_input"] == "Hello" - assert seen["name"] == "Ada" - assert seen["history"] == [{"role": "user", "content": "earlier"}] - assert seen["session_id"] == "ses" - assert seen["instructions"] == "Be brief." + assert script.messages[0]["system"] == "Be brief." + assert "User: earlier" in script.messages[0]["parts"][0]["text"] + assert "User: Hello" in script.messages[0]["parts"][0]["text"] async def test_a_run_opens_an_invoke_agent_span( self, monkeypatch: pytest.MonkeyPatch @@ -1030,15 +1007,12 @@ def set_span_in_context(self, span: _Span) -> Any: tracer = _Tracer() monkeypatch.setattr(span_module, "trace", tracer) - handler = create_opencode_handler( - session_id="ses", - events=source( - [ - assistant_open("m1"), - text_delta("m1", "p1", "Hi"), - assistant_completed("m1"), - ] - ), + handler, _script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ] ) seen = [event async for event in await handler.stream({}, "hi", None, None)] assert seen[-1]["output"] == "Hi" @@ -1046,109 +1020,6 @@ def set_span_in_context(self, span: _Span) -> Any: assert tracer.spans[0].attributes["gen_ai.operation.name"] == "invoke_agent" assert tracer.spans[0].attributes["launchdarkly.operation.type"] == "gen_ai" - async def test_block_array_history_reaches_the_span( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - from launchdarkly_ai_opencode_agents import spans as span_module - - class _Span: - def __init__(self, name: str, context: Any = None) -> None: - self.name = name - self.context = context - self.attributes: dict[str, Any] = {} - self.events: list[str] = [] - - def set_attribute(self, key: str, value: Any) -> None: - self.attributes[key] = value - - def add_event( - self, name: str, attributes: dict[str, Any] | None = None - ) -> None: - self.events.append(name) - - def set_status(self, code: Any, description: str | None = None) -> None: - return None - - def record_exception(self, exc: BaseException) -> None: - return None - - def end(self) -> None: - return None - - class _Tracer: - def __init__(self) -> None: - self.spans: list[_Span] = [] - - def get_tracer(self, name: str) -> _Tracer: - return self - - def start_span(self, name: str, context: Any = None) -> _Span: - span = _Span(name, context) - self.spans.append(span) - return span - - def set_span_in_context(self, span: _Span) -> Any: - return span - - tracer = _Tracer() - monkeypatch.setattr(span_module, "trace", tracer) - handler = create_opencode_handler( - capture_content=True, - session_id="ses", - events=source( - [ - assistant_open("m1"), - text_delta("m1", "p1", "Hi"), - assistant_completed("m1"), - ] - ), - ) - history = [ - {"role": "system", "content": "do not record"}, - {"role": "user", "content": "earlier"}, - { - "role": "user", - "content": [ - {"type": "text", "text": "see "}, - { - "type": "image", - "source": { - "type": "url", - "url": "https://example.test/a.png", - }, - }, - {"type": "text", "text": "this"}, - ], - }, - { - "role": "assistant", - "content": [{"type": "text", "text": "noted"}], - }, - ] - seen = [ - event async for event in await handler.stream({}, "", None, None, history) - ] - assert seen[-1]["output"] == "Hi" - recorded = json.loads(tracer.spans[0].attributes["gen_ai.input.messages"]) - assert recorded == [ - { - "role": "user", - "parts": [{"type": "text", "content": "earlier"}], - }, - { - "role": "user", - "parts": [{"type": "text", "content": "see this"}], - }, - { - "role": "assistant", - "parts": [{"type": "text", "content": "noted"}], - }, - ] - assert ( - "do not record" not in tracer.spans[0].attributes["gen_ai.input.messages"] - ) - assert "gen_ai.prompt.3.role" not in tracer.spans[0].attributes - async def _drain(stream: Any) -> None: async for _event in stream: diff --git a/packages/opencode-agents/tests/test_sdk_usage.py b/packages/opencode-agents/tests/test_sdk_usage.py index 64d946e4..28b36e5c 100644 --- a/packages/opencode-agents/tests/test_sdk_usage.py +++ b/packages/opencode-agents/tests/test_sdk_usage.py @@ -1,9 +1,9 @@ """Tests for TESTING.md §3.27 — opencode-agents.""" -from collections.abc import AsyncIterator from typing import Any from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest import launchdarkly_ai_server.lifecycle as lifecycle_module @@ -24,16 +24,17 @@ def _fake() -> MagicMock: return client -async def _events( - config: Any, - user_input: str | None, - variables: Any, - history: Any, - session_id: str | None, -) -> AsyncIterator[dict[str, Any]]: - del config, user_input, variables, history, session_id - if False: - yield {} +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: @@ -64,7 +65,7 @@ def _usage(client: MagicMock) -> list[Any]: async def test_create_handler_sends_one_usage_event(installed: MagicMock) -> None: - create_opencode_handler(events=_events, session_id="ses") + create_opencode_handler(base_url="http://opencode", client=_client()) calls = _usage(installed) assert [entry.args[2].get("helper") for entry in calls] == [ "opencode-agents.createOpenCodeHandler" @@ -82,7 +83,9 @@ 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, events=_events, session_id="ses")) + await _settle( + opencode_agents("k", "q", CONTEXT, base_url="http://opencode", client=_client()) + ) calls = _usage(installed) assert [entry.args[2].get("helper") for entry in calls] == [ "opencode-agents.opencodeAgents" diff --git a/uv.lock b/uv.lock index 8aebcdd1..b7c145b0 100644 --- a/uv.lock +++ b/uv.lock @@ -956,12 +956,14 @@ name = "launchdarkly-ai-opencode-agents" version = "0.1.0" source = { editable = "packages/opencode-agents" } dependencies = [ + { name = "httpx" }, { name = "launchdarkly-ai-server" }, { name = "opentelemetry-api" }, ] [package.metadata] requires-dist = [ + { name = "httpx", specifier = ">=0.27" }, { name = "launchdarkly-ai-server", editable = "packages/client" }, { name = "opentelemetry-api", specifier = ">=1.25" }, ] From 2a7324ee1ca94ca57764b43c0b11514ae2acb342 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 19 Oct 2026 18:55:44 +0000 Subject: [PATCH 2/7] fix(opencode-agents): satisfy ruff on the session handler Import AsyncIterable for the existing reducer and drop the unused test import. Co-authored-by: alexis --- .../src/launchdarkly_ai_opencode_agents/handler.py | 3 ++- packages/opencode-agents/tests/test_handler.py | 2 -- 2 files changed, 2 insertions(+), 3 deletions(-) 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 f83ba41b..b7fc9be5 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -7,10 +7,11 @@ from __future__ import annotations -from collections.abc import AsyncGenerator, Mapping +from collections.abc import AsyncGenerator, AsyncIterable, Mapping from typing import Any import httpx + from launchdarkly_ai_server import ( AiConfigRep, ProviderHandler, diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index aed8cdf5..1d13e322 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -7,8 +7,6 @@ from typing import Any from unittest.mock import AsyncMock, MagicMock -import json - import httpx import pytest From 8bf3237b09b9225a027cbe0b61e41c2e33ba4e6a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 19 Oct 2026 20:57:25 +0000 Subject: [PATCH 3/7] fix(opencode-agents): record block-array history on the span Take #177's _input_messages change from e54111d. Judges already open their own session, so the judge raise and skip_judges docs are not kept. Co-authored-by: alexis --- packages/opencode-agents/README.md | 32 +----- packages/opencode-agents/agents.md | 8 +- .../opencode-agents/tests/test_handler.py | 100 ++++++++++++++++++ 3 files changed, 109 insertions(+), 31 deletions(-) diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index bcbdc2c1..a9f4c0ba 100644 --- a/packages/opencode-agents/README.md +++ b/packages/opencode-agents/README.md @@ -10,35 +10,17 @@ pip install launchdarkly-ai-server launchdarkly-server-sdk launchdarkly-ai-openc ```python import asyncio -from collections.abc import AsyncIterator from launchdarkly_ai_opencode_agents import create_opencode_handler from launchdarkly_ai_server import config, init_client, shutdown -async def session_events(config, user_input, variables, history, session_id): - """Subscribe to session_id, send user_input, and stop at session.idle.""" - del config, variables, history - yield {"type": "message.updated", "properties": {"info": {"id": "m1", "role": "assistant", "time": {"created": 1}}}} - yield { - "type": "message.part.delta", - "properties": { - "sessionID": session_id, - "messageID": "m1", - "partID": "p1", - "field": "text", - "delta": "Hello", - }, - } - yield {"type": "session.idle", "properties": {"sessionID": session_id}} - - async def main() -> None: await init_client() - handler = create_opencode_handler(events=session_events, session_id="ses") - async for event in config( - key="my-agent", handler=handler, skip_judges=True - ).stream("Hello", {"kind": "user", "key": "user-1"}): + handler = create_opencode_handler(base_url="http://127.0.0.1:4096") + async for event in config(key="my-agent", handler=handler).stream( + "Hello", {"kind": "user", "key": "user-1"} + ): print(event) await shutdown() @@ -46,8 +28,4 @@ async def main() -> None: asyncio.run(main()) ``` -`events` is required. Each invocation calls `events(config, user_input, variables, history, session_id)` and the return value must be a fresh async iterable of that session only. The iterable must end: return it after `session.idle`, or close it yourself. The handler yields AG-UI text and tool events. `config().stream()` yields `chunk` then `done`. - -Pass `skip_judges=True`. This handler matches every agent-mode provider, so judge selection can land here. It does not run the judge prompt: that call raises without calling `events`, so the judge is tracked as a failed generation and logged as an error. - -Do not pass this handler to `run_judge`. It would be picked as the agent fallback and `run_judge` would raise. +`base_url` is a running `opencode serve`. Each call opens its own session, including a judge routed to this handler. 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 e47fecdd..bcf840b5 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -30,14 +30,14 @@ A run opens one `invoke_agent` span and sets `launchdarkly.operation.type`. The ## Call shape -`create_opencode_handler(*, events, session_id, capture_content=None)` returns a `ProviderHandler`. `events` and `session_id` are required. Each invocation calls `events(config, user_input, variables, history, session_id)` and expects a fresh async iterable. +`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. 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. ## Common pitfalls -The AI Config, user input, variables, and history are arguments to `events`. A callable that ignores them never sees the prompt. +The handler posts the prompt for the session it just opened. A caller that only watches a pre-existing session never sees that prompt. -`provides_for` is `('*', 'agent')`, so judge selection still lands here. Pass `skip_judges=True`. The handler does not run that prompt. A judge evaluation raises without calling `events`, so the judge is tracked as a failed generation. Do not pass this handler to `run_judge`, which would raise. +`provides_for` is `('*', 'agent')`, so a judge is routed here too. Each call opens its own session, so the judge prompt does not enter the agent's session and does not need `skip_judges=True`. -`session_id` is required. The reducer does not adopt the first session on a shared stream. Other sessions are dropped, including their `session.error`. `session.idle` for this session ends the run. If the iterable is a server-wide stream and you neither send `session.idle` nor close it, `done` never arrives. +The reducer does not adopt the first session on the bus. Other sessions are dropped, including their `session.error`. `session.idle` for this session ends the run. diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 1d13e322..d8f58eca 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -1018,6 +1018,106 @@ def set_span_in_context(self, span: _Span) -> Any: assert tracer.spans[0].attributes["gen_ai.operation.name"] == "invoke_agent" assert tracer.spans[0].attributes["launchdarkly.operation.type"] == "gen_ai" + async def test_block_array_history_reaches_the_span( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + from launchdarkly_ai_opencode_agents import spans as span_module + + class _Span: + def __init__(self, name: str, context: Any = None) -> None: + self.name = name + self.context = context + self.attributes: dict[str, Any] = {} + self.events: list[str] = [] + + def set_attribute(self, key: str, value: Any) -> None: + self.attributes[key] = value + + def add_event( + self, name: str, attributes: dict[str, Any] | None = None + ) -> None: + self.events.append(name) + + def set_status(self, code: Any, description: str | None = None) -> None: + return None + + def record_exception(self, exc: BaseException) -> None: + return None + + def end(self) -> None: + return None + + class _Tracer: + def __init__(self) -> None: + self.spans: list[_Span] = [] + + def get_tracer(self, name: str) -> _Tracer: + return self + + def start_span(self, name: str, context: Any = None) -> _Span: + span = _Span(name, context) + self.spans.append(span) + return span + + def set_span_in_context(self, span: _Span) -> Any: + return span + + tracer = _Tracer() + monkeypatch.setattr(span_module, "trace", tracer) + handler, _script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ], + capture_content=True, + ) + history = [ + {"role": "system", "content": "do not record"}, + {"role": "user", "content": "earlier"}, + { + "role": "user", + "content": [ + {"type": "text", "text": "see "}, + { + "type": "image", + "source": { + "type": "url", + "url": "https://example.test/a.png", + }, + }, + {"type": "text", "text": "this"}, + ], + }, + { + "role": "assistant", + "content": [{"type": "text", "text": "noted"}], + }, + ] + seen = [ + event async for event in await handler.stream({}, "", None, None, history) + ] + assert seen[-1]["output"] == "Hi" + recorded = json.loads(tracer.spans[0].attributes["gen_ai.input.messages"]) + assert recorded == [ + { + "role": "user", + "parts": [{"type": "text", "content": "earlier"}], + }, + { + "role": "user", + "parts": [{"type": "text", "content": "see this"}], + }, + { + "role": "assistant", + "parts": [{"type": "text", "content": "noted"}], + }, + ] + assert ( + "do not record" not in tracer.spans[0].attributes["gen_ai.input.messages"] + ) + assert "gen_ai.prompt.3.role" not in tracer.spans[0].attributes + async def _drain(stream: Any) -> None: async for _event in stream: From 8e85667f94de89e1a7e3f1e3681f33a7a56ac89f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 16:43:07 +0000 Subject: [PATCH 4/7] fix(opencode-agents): abort failed sessions and render judge prompts A failed message POST raises without waiting on an open event bus. Early exit and session.error abort through aclosing. This session's idle ends the turn, with a longer fallback so the final usage is kept. Agent instructions, including a judge prompt, are rendered. Co-authored-by: alexis --- packages/opencode-agents/README.md | 2 +- packages/opencode-agents/agents.md | 2 +- .../handler.py | 99 ++++++---- .../launchdarkly_ai_opencode_agents/server.py | 131 ++++++++++--- .../opencode-agents/tests/test_handler.py | 184 +++++++++++++++++- 5 files changed, 343 insertions(+), 75 deletions(-) diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index a9f4c0ba..b88866b7 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. The handler yields AG-UI text and tool events. `config().stream()` yields `chunk` then `done`. +`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`. diff --git a/packages/opencode-agents/agents.md b/packages/opencode-agents/agents.md index bcf840b5..49457ebd 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -38,6 +38,6 @@ Assistant text is emitted only after `message.updated` marks that message `role The handler posts the prompt for the session it just opened. A caller that only watches a pre-existing session never sees that prompt. -`provides_for` is `('*', 'agent')`, so a judge is routed here too. Each call opens its own session, so the judge prompt does not enter the agent's session and does not need `skip_judges=True`. +`provides_for` is `('*', 'agent')`, so a judge is routed here too. Each call opens its own session, so the judge prompt does not enter the agent's session and does not need `skip_judges=True`. The judge still runs as a full OpenCode agent, with the server's tools, in the server's working directory. This package does not take a tool-less judge agent. The reducer does not adopt the first session on the bus. Other sessions are dropped, including their `session.error`. `session.idle` for this session ends the run. 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 b7fc9be5..08d165cd 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -8,6 +8,7 @@ from __future__ import annotations from collections.abc import AsyncGenerator, AsyncIterable, Mapping +from contextlib import aclosing from typing import Any import httpx @@ -23,6 +24,7 @@ compose_history, content_to_text, end_span_once, + parse_template, set_input_content_attributes, set_output_content_attributes, text_message, @@ -111,7 +113,9 @@ async def reduce( if session_id is not None and session_id != self._session_id: continue if kind == "session.idle": - return + if session_id == self._session_id: + return + continue if kind == "session.error": raise OpenCodeSessionError(_session_error_message(properties)) for event in self._event(raw): @@ -494,9 +498,12 @@ def create_opencode_handler( """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. 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``. 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. """ if not isinstance(base_url, str) or not base_url: raise TypeError("The OpenCode handler requires a base_url.") @@ -521,7 +528,9 @@ def opencode_agents( variables = kwargs.pop("variables", None) history = kwargs.pop("history", None) capture_content = kwargs.pop("capture_content", None) - base_url = kwargs.pop("base_url") + base_url = kwargs.pop("base_url", None) + if not isinstance(base_url, str) or not base_url: + raise TypeError("The OpenCode handler requires a base_url.") model = kwargs.pop("model", None) agent = kwargs.pop("agent", None) client = kwargs.pop("client", None) @@ -566,7 +575,7 @@ async def _stream_impl( provider_id, model_id = resolve_model( config if isinstance(config, dict) else {} ) - system = _system_text(config) + system = _system_text(config, variables) prompt = _user_prompt(config, user_input, history) try: set_input_content_attributes( @@ -576,18 +585,21 @@ async def _stream_impl( messages=_input_messages(user_input, history), ) created = _Created() - async for event in _reduce_session( - created, - base_url=bound_url, - prompt=prompt, - system=system, - provider_id=provider_id, - model_id=model_id, - agent=bound_agent, - client=bound_client, - capture=capture, - ): - yield event + async with aclosing( + _reduce_session( + created, + base_url=bound_url, + prompt=prompt, + system=system, + provider_id=provider_id, + model_id=model_id, + agent=bound_agent, + client=bound_client, + capture=capture, + ) + ) as reduced: + async for event in reduced: + yield event output = "".join(created.deltas) if output: set_output_content_attributes( @@ -619,10 +631,12 @@ async def _call_impl( ) -> dict[str, Any]: output = "" usage: dict[str, Any] = {} - async for event in _stream_impl( - config, user_input, tool_handlers, variables, history - ): - if event.get("type") == "done": + async with aclosing( + _stream_impl(config, user_input, tool_handlers, variables, history) + ) as stream: + async for event in stream: + if event.get("type") != "done": + continue raw_output = event.get("output") output = raw_output if isinstance(raw_output, str) else "" raw_usage = event.get("usage") @@ -661,20 +675,24 @@ async def _reduce_session( """Reduce the session ``POST /session`` created. The reducer is unchanged.""" session_ids: list[str] = [] reducer: _Reducer | None = None - async for raw in session_events( - base_url=base_url, - prompt=prompt, - system=system, - provider_id=provider_id, - model_id=model_id, - agent=agent, - session_ids=session_ids, - client=client, - ): - if reducer is None: - reducer = _Reducer(capture_content=capture, session_id=session_ids[0]) - async for event in reducer.reduce(_once(raw)): - yield event + async with aclosing( + session_events( + base_url=base_url, + prompt=prompt, + system=system, + provider_id=provider_id, + model_id=model_id, + agent=agent, + session_ids=session_ids, + client=client, + ) + ) as raw_events: + async for raw in raw_events: + if reducer is None: + reducer = _Reducer(capture_content=capture, session_id=session_ids[0]) + async with aclosing(reducer.reduce(_once(raw))) as reduced: + async for event in reduced: + yield event if reducer is None: return for event in reducer.close(): @@ -687,12 +705,17 @@ async def _once(raw: dict[str, Any]) -> AsyncGenerator[dict[str, Any], None]: yield raw -def _system_text(config: AiConfigRep) -> str | None: +def _system_text(config: AiConfigRep, variables: dict[str, Any] | None) -> str | None: + """Agent instructions, with ``{{variables}}`` rendered. + + History and the user turn are not rendered. A judge prompt is rendered + the same way, because a judge uses this handler. + """ if not isinstance(config, dict): return None instructions = config.get("instructions") if isinstance(instructions, str) and instructions.strip(): - return instructions + return parse_template(instructions, variables or {}) return None 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 936331b0..732d5b89 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py @@ -9,12 +9,19 @@ import asyncio import json -from collections.abc import AsyncIterator, Callable +from collections.abc import AsyncGenerator, Callable from typing import Any import httpx -DRAIN_SECONDS = 0.05 +# After the message POST returns, keep reading until this session's +# ``session.idle``. A quiet bus ends the turn only after this fallback, so a +# late ``message.updated`` is not dropped the way a 50 ms drain dropped it. +DRAIN_FALLBACK_SECONDS = 5.0 +# Session create and abort. The message POST blocks for the whole turn, so +# its timeout is longer, and still finite. +HTTP_TIMEOUT_SECONDS = 30.0 +MESSAGE_TIMEOUT_SECONDS = 600.0 ModelResolver = Callable[[dict[str, Any]], tuple[str, str]] @@ -54,24 +61,30 @@ async def session_events( agent: str | None, session_ids: list[str], client: httpx.AsyncClient | None = None, -) -> AsyncIterator[dict[str, Any]]: +) -> 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 - concurrently with the read. ``session.idle`` ends the turn. A returned - post is followed by a brief drain. Leaving early cancels the post and - aborts the session. + 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(timeout=None) + http = client or httpx.AsyncClient( + timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS, read=MESSAGE_TIMEOUT_SECONDS) + ) base = base_url.rstrip("/") post: asyncio.Task[httpx.Response] | None = None session_id: str | None = None succeeded = False try: - async with http.stream("GET", f"{base}/event") as sse: + async with http.stream( + "GET", + f"{base}/event", + timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS, read=None), + ) as sse: sse.raise_for_status() - created = await http.post(f"{base}/session") + created = await http.post(f"{base}/session", timeout=HTTP_TIMEOUT_SECONDS) if created.status_code >= 400: raise OpenCodeHTTPError(f"POST /session -> {created.status_code}") session_id = _session_id(created.json()) @@ -85,7 +98,11 @@ async def session_events( if agent: body["agent"] = agent post = asyncio.create_task( - http.post(f"{base}/session/{session_id}/message", json=body) + http.post( + f"{base}/session/{session_id}/message", + json=body, + timeout=MESSAGE_TIMEOUT_SECONDS, + ) ) async for event in _read_until_done(sse, post, session_id): yield event @@ -99,8 +116,15 @@ async def session_events( if not succeeded and session_id is not None: if post is not None and not post.done(): post.cancel() + try: + await post + except BaseException: + pass try: - await http.post(f"{base}/session/{session_id}/abort") + await http.post( + f"{base}/session/{session_id}/abort", + timeout=HTTP_TIMEOUT_SECONDS, + ) except Exception: pass if owns_client: @@ -111,28 +135,71 @@ async def _read_until_done( response: httpx.Response, post: asyncio.Task[httpx.Response], session_id: str, -) -> AsyncIterator[dict[str, Any]]: +) -> AsyncGenerator[dict[str, Any], None]: + """Read the bus until this session is idle, or the message POST fails. + + The next SSE line is raced against the post task. A status of 400 or + higher raises immediately, even when ``/event`` stays open. After the + post returns successfully, reading continues until this session's + ``session.idle`` or ``DRAIN_FALLBACK_SECONDS`` of silence. An idle event + with no session id does not end the turn. + """ lines = response.aiter_lines() - while True: - timeout = DRAIN_SECONDS if post.done() else None - try: - if timeout is None: - line = await lines.__anext__() - else: - line = await asyncio.wait_for(lines.__anext__(), timeout) - except StopAsyncIteration: - return - except TimeoutError: - return - event = _parse_frame(line) - if event is None: - continue - yield event - if event.get("type") != "session.idle": - continue - owner = _frame_session(event) - if owner is None or owner == session_id: - return + line_task: asyncio.Task[str | None] | None = None + try: + while True: + if post.done(): + _raise_if_message_failed(post, session_id) + if line_task is None: + line_task = asyncio.create_task(_next_line(lines)) + waiting: set[asyncio.Task[Any]] = {line_task} + if not post.done(): + waiting.add(post) + finished, _pending = await asyncio.wait( + waiting, + timeout=DRAIN_FALLBACK_SECONDS if post.done() else None, + return_when=asyncio.FIRST_COMPLETED, + ) + if not finished: + return + if post in finished: + _raise_if_message_failed(post, session_id) + if line_task not in finished: + continue + line = line_task.result() + line_task = None + if line is None: + return + event = _parse_frame(line) + if event is None: + continue + yield event + if ( + event.get("type") == "session.idle" + and _frame_session(event) == session_id + ): + return + finally: + if line_task is not None and not line_task.done(): + line_task.cancel() + + +async def _next_line(lines: Any) -> str | None: + try: + line = await lines.__anext__() + except StopAsyncIteration: + return None + return line if isinstance(line, str) else None + + +def _raise_if_message_failed( + post: asyncio.Task[httpx.Response], session_id: str +) -> None: + response = post.result() + if response.status_code >= 400: + raise OpenCodeHTTPError( + f"POST /session/{session_id}/message -> {response.status_code}" + ) def _session_id(body: object) -> str: diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index d8f58eca..3c715ac3 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -2,8 +2,10 @@ from __future__ import annotations +import asyncio import json from collections.abc import Iterator +from contextlib import aclosing from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -738,7 +740,7 @@ async def test_a_tool_does_not_close_the_stream( async def test_session_error_does_not_yield_stable_done( self, mock_ld_client: MagicMock ) -> None: - handler, _script = playing( + handler, script = playing( [ assistant_open("m1"), text_delta("m1", "p1", "Hi"), @@ -751,6 +753,7 @@ async def test_session_error_does_not_yield_stable_done( seen.append(event) assert [event["type"] for event in seen] == ["chunk"] assert seen[0]["text"] == "Hi" + assert script.aborts class TestInvocation: @@ -759,6 +762,8 @@ def test_base_url_is_required(self) -> None: create_opencode_handler() # type: ignore[call-arg] with pytest.raises(TypeError, match="base_url"): create_opencode_handler(base_url="") + with pytest.raises(TypeError, match="base_url"): + opencode_agents("flag", "hi", CONTEXT) async def test_the_first_session_on_the_stream_is_not_adopted(self) -> None: events = await collect( @@ -918,7 +923,6 @@ async def test_a_judge_opens_its_own_session( "judges": [{"key": "judge-flag", "samplingRate": 1}] }, } - mock_ld_client.variation = AsyncMock(return_value=judged) handler, script = playing( [ assistant_open("m1"), @@ -926,12 +930,27 @@ async def test_a_judge_opens_its_own_session( assistant_completed("m1"), ] ) + + async def variation(key: str, *_args: Any, **_kwargs: Any) -> dict[str, Any]: + if key == "judge-flag": + return { + **VARIATION, + "instructions": ( + "Grade {{response_to_evaluate}}. History: {{message_history}}" + ), + } + return judged + + mock_ld_client.variation = variation await _drain(config(key="flag", handler=handler).stream("q", CONTEXT)) assert len(script.sessions) == 2 assert script.sessions[0] != script.sessions[1] prompts = [message["parts"][0]["text"] for message in script.messages] + systems = [message.get("system", "") for message in script.messages] assert prompts[0] == "q" - assert prompts[1] != "q" + assert "{{response_to_evaluate}}" not in systems[1] + assert "{{message_history}}" not in systems[1] + assert "Hi" in systems[1] async def test_the_wrapper_posts_history_on_a_fresh_session( self, mock_ld_client: MagicMock @@ -1118,6 +1137,165 @@ def set_span_in_context(self, span: _Span) -> Any: ) assert "gen_ai.prompt.3.role" not in tracer.spans[0].attributes + 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) + + async def _run() -> None: + async for _event in await handler.stream( + { + "instructions": "Be brief.", + "model": {"name": "m"}, + "provider": {"name": "Acme"}, + }, + "hi", + None, + None, + ): + return + + from launchdarkly_ai_opencode_agents.server import OpenCodeHTTPError + + with pytest.raises(OpenCodeHTTPError, match="400"): + await asyncio.wait_for(_run(), timeout=2) + assert script.aborts + + 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) + + async def _run() -> None: + async with aclosing( + await handler.stream( + { + "instructions": "Be brief.", + "model": {"name": "m"}, + "provider": {"name": "Acme"}, + }, + "hi", + None, + None, + ) + ) as gen: + async for event in gen: + if event["type"] == "TEXT_MESSAGE_CONTENT": + break + + await asyncio.wait_for(_run(), timeout=2) + assert script.aborts + + 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 + ) + seen = [ + event + async for event in await handler.stream( + { + "instructions": "Be brief.", + "model": {"name": "m"}, + "provider": {"name": "Acme"}, + }, + "hi", + None, + None, + ) + ] + assert seen[-1]["output"] == "Hi!" + assert seen[-1]["usage"]["output_tokens"] == 4 + + async def test_an_idle_without_a_session_id_does_not_end_the_run(self) -> None: + events = await collect( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + bus("session.idle"), + text_delta("m1", "p1", "!", session_id="ses"), + bus("session.idle", sessionID="ses"), + ] + ) + assert events[-1]["output"] == "Hi!" + + +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: + """An `/event` stream that stays open, unlike a finite ``Script`` body.""" + + def __init__(self, *, message_status: int = 200, delay_tail: float = 0) -> None: + 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: + head = [ + 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) + async def _drain(stream: Any) -> None: async for _event in stream: From dba0d7b03c508a586433ef448601203e9c2ec692 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 18:41:02 +0000 Subject: [PATCH 5/7] fix(opencode-agents): await the cancelled SSE read before closing the stream A quiet bus and a failed message POST both leave the pending /event line task running. Cancel it and await it, swallowing CancelledError, before GET /event closes. Co-authored-by: alexis --- .../launchdarkly_ai_opencode_agents/server.py | 8 +++ .../opencode-agents/tests/test_handler.py | 72 +++++++++++++++++++ 2 files changed, 80 insertions(+) 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 732d5b89..d86f314e 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py @@ -180,8 +180,16 @@ async def _read_until_done( ): return finally: + # Cancel and await before the caller closes GET /event. A bare cancel + # leaves this read running into the closed stream. The quiet-bus + # return and a failed message POST both leave it pending. if line_task is not None and not line_task.done(): line_task.cancel() + if line_task is not None: + try: + await line_task + except asyncio.CancelledError: + pass async def _next_line(lines: Any) -> str | None: diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 3c715ac3..b86120e1 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -1163,6 +1163,36 @@ async def _run() -> None: await asyncio.wait_for(_run(), timeout=2) assert script.aborts + 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) + + async def _run() -> None: + async for _event in await handler.stream( + { + "instructions": "Be brief.", + "model": {"name": "m"}, + "provider": {"name": "Acme"}, + }, + "hi", + None, + 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 + assert not bus.closed_before_read_finished + async def test_breaking_aborts_without_waiting_for_gc(self) -> None: script = _OpenBus() client = httpx.AsyncClient( @@ -1227,6 +1257,48 @@ 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): + """``/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() + 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" + self.second_read.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + self.read_finished.set() + raise + + async def aclose(self) -> None: + if not self.read_finished.is_set(): + self.closed_before_read_finished = True + self.closed = True + + class _ByteStream(httpx.AsyncByteStream): def __init__(self, chunks: Any) -> None: self._chunks = chunks From 491939d85e425b6ef7c64f715bec2045aeb69017 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 18:49:51 +0000 Subject: [PATCH 6/7] ci: rerun checks The Bugbot fix push did not trigger the pull_request workflow. From 6469e10bdaef963b945b3e989676366868c46c38 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 20 Oct 2026 20:33:29 +0000 Subject: [PATCH 7/7] test(opencode-agents): leave the bus silent when the message POST fails A 400 used to be followed by assistant events, so the test passed even without racing the POST. A rejected message sends nothing. Co-authored-by: alexis --- packages/opencode-agents/tests/test_handler.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index b86120e1..2d7ccf18 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -1330,10 +1330,16 @@ async def handle(request: httpx.Request) -> httpx.Response: if path == "/event" and request.method == "GET": async def chunks() -> Any: - head = [ - assistant_open("m1"), - text_delta("m1", "p1", "Hi"), - ] + # 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: