diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index bcbdc2c1..b88866b7 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. 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 e47fecdd..49457ebd 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`. 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. -`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/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..08d165cd 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -1,17 +1,17 @@ """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, AsyncIterable, Mapping +from contextlib import aclosing from typing import Any -from opentelemetry import context as otel_context +import httpx from launchdarkly_ai_server import ( AiConfigRep, @@ -24,30 +24,20 @@ compose_history, content_to_text, end_span_once, + parse_template, set_input_content_attributes, set_output_content_attributes, 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. @@ -123,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): @@ -465,16 +457,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 +487,33 @@ 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 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.") 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 +523,43 @@ 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", 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) 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 +568,44 @@ 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, variables) + 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 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, ) - ): - yield event - for event in reducer.close(): - yield event - output = "".join(reducer.deltas) + ) as reduced: + async for event in reduced: + yield event + 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, @@ -661,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") @@ -677,3 +649,106 @@ 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 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(): + 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, 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 parse_template(instructions, variables or {}) + 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..d86f314e --- /dev/null +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py @@ -0,0 +1,244 @@ +"""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 AsyncGenerator, Callable +from typing import Any + +import httpx + +# 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]] + + +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, +) -> 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. 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=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", + timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS, read=None), + ) as sse: + sse.raise_for_status() + 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()) + 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, + timeout=MESSAGE_TIMEOUT_SECONDS, + ) + ) + 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 post + except BaseException: + pass + try: + await http.post( + f"{base}/session/{session_id}/abort", + timeout=HTTP_TIMEOUT_SECONDS, + ) + 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, +) -> 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() + 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: + # 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: + 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: + 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..2d7ccf18 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -2,11 +2,14 @@ from __future__ import annotations +import asyncio import json -from collections.abc import AsyncIterator, Iterator +from collections.abc import Iterator +from contextlib import aclosing from typing import Any from unittest.mock import AsyncMock, MagicMock +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``.""" - return _events + 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] = [] + + 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): @@ -721,20 +753,17 @@ 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: - 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="") + 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( @@ -767,42 +796,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 +914,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 = { @@ -902,87 +923,60 @@ async def test_skip_judges_runs_events_once( "judges": [{"key": "judge-flag", "samplingRate": 1}] }, } - 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) + handler, script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ] + ) - 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( + 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 "{{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 ) -> 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 +1024,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" @@ -1092,16 +1083,13 @@ def set_span_in_context(self, span: _Span) -> Any: tracer = _Tracer() monkeypatch.setattr(span_module, "trace", tracer) - handler = create_opencode_handler( + handler, _script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ], 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"}, @@ -1149,6 +1137,243 @@ 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_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( + 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 _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 + + 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: + # 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) + 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" }, ]