diff --git a/AGENTS.md b/AGENTS.md index 10d181d6..31e72bb4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -128,7 +128,7 @@ graph TD - **Tier 0 — Core** (`launchdarkly-ai-server`): The foundation. Owns all LaunchDarkly integration, telemetry orchestration, shared data types, and the primary entry points (`config()`, `graph()`, `resolve_graph()`). Has no dependency on any other `launchdarkly-ai-*` package. - **Tier 0 — Convenience barrel** (`launchdarkly-ai-python`): A pure re-export package that makes all of `launchdarkly-ai-server` available under a shorter install name. No new logic — intended as the default install for most Python applications. -- **Tier 1 — Handler packages** (`launchdarkly-ai-claude-agents`, `launchdarkly-ai-claude-messages`, `launchdarkly-ai-openai-agents`, `launchdarkly-ai-openai-messages`, `launchdarkly-ai-langchain-agents`, `launchdarkly-ai-langchain-messages`, `launchdarkly-ai-litellm-agents`, `launchdarkly-ai-litellm-messages`, `launchdarkly-ai-opencode-agents`, `launchdarkly-ai-vercel-agents`, `launchdarkly-ai-vercel-messages`, …): Each wraps a specific AI provider SDK. Depends on `launchdarkly-ai-server` for shared types and utilities. Must not depend on other Tier 1 packages. The OpenCode handler reduces caller-supplied session events and does not depend on an OpenCode SDK. +- **Tier 1 — Handler packages** (`launchdarkly-ai-claude-agents`, `launchdarkly-ai-claude-messages`, `launchdarkly-ai-openai-agents`, `launchdarkly-ai-openai-messages`, `launchdarkly-ai-langchain-agents`, `launchdarkly-ai-langchain-messages`, `launchdarkly-ai-litellm-agents`, `launchdarkly-ai-litellm-messages`, `launchdarkly-ai-opencode-agents`, `launchdarkly-ai-vercel-agents`, `launchdarkly-ai-vercel-messages`, …): Each wraps a specific AI provider SDK. Depends on `launchdarkly-ai-server` for shared types and utilities. Must not depend on other Tier 1 packages. The OpenCode handler starts `opencode serve` per call and speaks the OpenCode HTTP API itself. It does not depend on an OpenCode SDK. - **Tier 2 — Consumer applications** (e.g. `main.py`, downstream projects): Imports from one or more handler packages and either `launchdarkly-ai-python` or `launchdarkly-ai-server`. Owns tool implementations and orchestration logic. No `launchdarkly-ai-*` package should ever depend on Tier 2 code. ### Rules diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index ec941cf1..9037e5d5 100644 --- a/packages/opencode-agents/README.md +++ b/packages/opencode-agents/README.md @@ -1,6 +1,6 @@ # `launchdarkly-ai-opencode-agents` -OpenCode handler for `launchdarkly-ai-server`. It reduces caller-supplied OpenCode session events into assistant text. It does not start OpenCode and it does not depend on an OpenCode SDK. +OpenCode handler for `launchdarkly-ai-server`. `async with handler` starts `opencode serve` and waits until it is healthy. The process stops when that context exits. It does not depend on an OpenCode SDK. **`provides_for`:** `['*', 'agent']` — matches an agent-mode variation for any provider. @@ -10,6 +10,7 @@ pip install launchdarkly-ai-server launchdarkly-server-sdk launchdarkly-ai-openc ```python import asyncio +import os from launchdarkly_ai_opencode_agents import create_opencode_handler from launchdarkly_ai_server import config, init_client, shutdown @@ -17,15 +18,22 @@ from launchdarkly_ai_server import config, init_client, shutdown async def main() -> None: await init_client() - 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) + handler = create_opencode_handler( + cwd="/work", + config="/home/app/.vega/opencode.json", + env=os.environ.copy(), + ) + async with handler: + async for event in config(key="my-agent", handler=handler).stream( + "Hello", {"kind": "user", "key": "user-1"} + ): + print(event) await shutdown() asyncio.run(main()) ``` -`base_url` is a running `opencode serve`. The handler owns every HTTP call to that server. A caller does not pass an HTTP client or transport. Each call opens its own session, including a judge routed to this handler. A judge runs as a full OpenCode agent: the server's tools, in the server's working directory, apply to the judge prompt. There is no tool-less judge mode. The handler yields AG-UI text and tool events. `config().stream()` yields `chunk` then `done`. +`cwd` is the working directory, `config` is the OpenCode config file, and `env` is the child environment. The handler sets `OPENCODE_CONFIG` from `config`, replaces any `OPENCODE_SERVER_USERNAME` in `env` with `opencode`, replaces any `OPENCODE_SERVER_PASSWORD` with a generated secret, and binds `opencode serve` to loopback. Requests to that server use that username and secret as basic auth. A caller does not pass a URL, an HTTP client, or a transport. `command` overrides the `opencode` executable; a test passes a fake executable there. + +Entering the handler starts `opencode serve` and returns once it is healthy. The process stops when the context exits, after `config().stream()` returns. Put a wall-clock limit around `config().stream()`, not around the context. A judge routed here runs as a full OpenCode agent: the server's tools, in `cwd`, 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 38721fee..f6f22412 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -4,7 +4,7 @@ provides_for = ('*', 'agent') ``` -The handler reduces OpenCode session events the caller supplies. It does not import or construct `anomalyco/opencode-sdk-python`, and it does not start an OpenCode process. +Each call starts `opencode serve`, speaks the stable HTTP API (`/session`, `/session/{id}/message`, `/event`, `/session/{id}/abort`), and stops the process when the stream finishes or the caller closes it. The package does not import or construct `anomalyco/opencode-sdk-python`. This package is Python-only. There is no JS counterpart in this slice. The spec section, API surface list, and `usage-ids.txt` row live in `launchdarkly/ai-sdks-monorepo`, which is not this repository. The usage helper ids are `opencode-agents.createOpenCodeHandler` and `opencode-agents.opencodeAgents`. The package is a normal pre-v1 package, same as the other handlers. @@ -13,12 +13,14 @@ This package is Python-only. There is no JS counterpart in this slice. The spec | File | Role | | --- | --- | | `handler.py` | Reducer, `create_opencode_handler`, `opencode_agents` | +| `process.py` | Starts and stops `opencode serve` for one call | +| `server.py` | The HTTP session this process just started | | `spans.py` | `invoke_agent` root, and the `chat` / `execute_tool` constructors the parity checks call | | `_version.py` | Package name and version sent with usage events | ## Dependencies -`launchdarkly-ai-server` and `opentelemetry-api`. `launchdarkly-server-sdk` is what `init_client()` loads for a normal process. No OpenCode SDK. +`launchdarkly-ai-server`, `httpx`, and `opentelemetry-api`. `launchdarkly-server-sdk` is what `init_client()` loads for a normal process. No OpenCode SDK. The `opencode` executable is the default command. ## init_client @@ -30,14 +32,16 @@ A run opens one `invoke_agent` span and sets `launchdarkly.operation.type`. The ## Call shape -`create_opencode_handler(*, base_url, capture_content=None, model=None, agent=None)` returns a `ProviderHandler`. `base_url` is a running `opencode serve`. The handler owns the HTTP client. Each call opens its own session. +`create_opencode_handler(*, cwd, config, env, capture_content=None, model=None, agent=None, command=None)` returns a `ProviderHandler`. `cwd` is the working directory. `config` is the path of the OpenCode config file. `env` replaces the child environment, and the handler sets `OPENCODE_CONFIG` from `config`. Each start also sets `OPENCODE_SERVER_USERNAME` to `opencode` and `OPENCODE_SERVER_PASSWORD` to a generated secret, replacing either value inherited from `env`, and sends that pair as basic auth. `command` defaults to `opencode`. A test passes a fake executable instead of a URL. There is no `base_url`, client, or transport argument. + +`async with handler` starts the process and returns once `/global/health` returns 200 and `healthy: true`. Concurrent entries share that process. The process stops when the last context exits, after `config().stream()` returns, including when the caller times out or closes the stream. A wall-clock budget goes around `config().stream()`, not around the context. Closing the stream still aborts the session. Without the context, one call starts and stops its own process. 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 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`. 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. +`provides_for` is `('*', 'agent')`, so a judge is routed here too. Each call starts its own process, 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 `cwd`. 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. + +`env` is not merged with the parent environment. A bare `opencode` command needs `PATH` in `env`. 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 80ac2701..6335ee1d 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -1,12 +1,16 @@ """Reduce OpenCode session events into handler stream events. -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``. +``async with handler`` starts ``opencode serve`` and waits until it is healthy. +The process stops when that context exits, after ``config().stream()`` returns, +including when the caller times out or closes the stream. The handler yields +the AG-UI text and tool subset, then one ``done``. ``config().stream()`` still +yields ``chunk`` then ``done``. """ from __future__ import annotations +import asyncio +import os from collections.abc import AsyncGenerator, AsyncIterable, Mapping from contextlib import aclosing from typing import Any @@ -17,7 +21,6 @@ SpanMessage, SpanUsage, _config, - _create_handler, add_cached_tokens_to_input, compose_history, content_to_text, @@ -33,6 +36,7 @@ from launchdarkly_ai_server.utils import number_or_zero from ._version import PACKAGE_NAME, __version__ +from .process import Command, OpenCodeServe, serve_from from .server import ModelResolver, default_model, session_events from .spans import fail_span, finish_root_span, mark_ok, start_root_span @@ -487,27 +491,38 @@ def _input_messages( def create_opencode_handler( *, - base_url: str, + cwd: str | os.PathLike[str], + config: str | os.PathLike[str], + env: Mapping[str, str], capture_content: bool | None = None, model: ModelResolver | None = None, agent: str | None = None, + command: Command = None, ) -> ProviderHandler: - """Return a handler that opens one OpenCode session per call. - - ``provides_for`` is ``('*', 'agent')``. *base_url* is a running - ``opencode serve``. The handler opens its own HTTP client. A caller does - not pass a client or a transport. A judge routed here gets its own session - and runs as a full OpenCode agent: the server's tools, in the server's - working directory, apply to the judge prompt too. This package does not - offer a tool-less judge agent. A Vega-local handler was rejected: the - session boundary belongs to this package. *model* overrides provider and - model selection. + """Return a handler that starts ``opencode serve`` for each call. + + ``provides_for`` is ``('*', 'agent')``. *cwd* is the working directory, + *config* is the OpenCode config file, and *env* is the child environment. + The handler sets ``OPENCODE_CONFIG`` from *config*, + ``OPENCODE_SERVER_USERNAME`` to ``opencode``, and + ``OPENCODE_SERVER_PASSWORD`` to a generated secret, replacing either value + in *env*. *command* is the executable, ``opencode`` when omitted. A test + passes a fake executable here. There is no URL, client, or transport + argument. + + Entering the handler as an async context starts the process and returns + only once it is healthy. Exiting waits until the process has stopped, so a + caller can remove *cwd* after the context returns. A wall-clock budget + belongs around ``config().stream()``, not around enter or exit. Without + the context, one call starts and stops its own process. A judge routed + here runs as a full OpenCode agent: the server's tools, in *cwd*, apply to + the judge prompt too. This package does not offer a tool-less judge agent. + *model* overrides provider and model selection. """ - if not isinstance(base_url, str) or not base_url: - raise TypeError("The OpenCode handler requires a base_url.") + bound = serve_from(cwd=cwd, config=config, env=env, command=command) report_usage("opencode-agents.createOpenCodeHandler", PACKAGE_NAME, __version__) - return _create_opencode_handler( - base_url=base_url, + return OpenCodeHandler( + serve=bound, capture_content=capture_content, model=model or default_model, agent=agent, @@ -525,15 +540,20 @@ 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", None) - if not isinstance(base_url, str) or not base_url: - raise TypeError("The OpenCode handler requires a base_url.") + cwd = kwargs.pop("cwd", None) + config_path = kwargs.pop("config", None) + env = kwargs.pop("env", None) + command = kwargs.pop("command", None) + if cwd is None or config_path is None or env is None: + raise TypeError( + "The OpenCode handler requires a working directory, config file, and env." + ) model = kwargs.pop("model", None) agent = kwargs.pop("agent", None) return _config( key=config_key, - handler=_create_opencode_handler( - base_url=base_url, + handler=OpenCodeHandler( + serve=serve_from(cwd=cwd, config=config_path, env=env, command=command), capture_content=capture_content, model=model or default_model, agent=agent, @@ -542,19 +562,68 @@ def opencode_agents( ).invoke(user_input, context, variables, history) -def _create_opencode_handler( - *, - base_url: str, - capture_content: bool | None, - model: ModelResolver, - agent: str | None, -) -> ProviderHandler: - content_override = capture_content - bound_url = base_url - resolve_model = model - bound_agent = agent +class OpenCodeHandler(ProviderHandler): + """Handler whose async context owns one ``opencode serve`` process.""" + + def __init__( + self, + *, + serve: OpenCodeServe, + capture_content: bool | None, + model: ModelResolver, + agent: str | None, + ) -> None: + self._serve = serve + self._model = model + self._agent = agent + self._active: OpenCodeServe | None = None + self._base_url: str | None = None + self._password: str | None = None + self._depth = 0 + self._lock = asyncio.Lock() + super().__init__( + fn=self._call_impl, + provides_for=("*", "agent"), + stream_fn=self._stream_impl, + capture_content=capture_content, + ) + + async def __aenter__(self) -> OpenCodeHandler: + """Start ``opencode serve`` and return once it answers on loopback. + + Concurrent entries share that process. The lock covers startup so a + second entry cannot replace ``_active`` and leave the first process + running. + """ + async with self._lock: + if self._active is None: + active = self._serve.spawn() + self._base_url = await active.start() + self._password = active.password + self._active = active + self._depth += 1 + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + traceback: Any, + ) -> None: + """Stop the process and wait until it has exited.""" + async with self._lock: + self._depth -= 1 + if self._depth > 0: + return + active = self._active + self._active = None + self._base_url = None + self._password = None + if active is not None: + await active.stop() async def _stream_impl( + self, config: AiConfigRep, user_input: str | None, tool_handlers: dict[str, Any] | None, @@ -562,14 +631,26 @@ async def _stream_impl( history: list[dict[str, Any]] | None = None, ) -> AsyncGenerator[dict[str, Any], None]: del tool_handlers - capture = effective_capture_content(content_override) + capture = effective_capture_content(self._capture_content_override) span = start_root_span(config, variables or {}) ended: set[int] = set() - provider_id, model_id = resolve_model( - config if isinstance(config, dict) else {} - ) + provider_id, model_id = self._model(config if isinstance(config, dict) else {}) system = _system_text(config, variables) prompt = _user_prompt(config, user_input, history, variables) + # A context-owned process outlives this stream, so a caller timeout + # that cancels the stream cannot cancel its shutdown. + owned: OpenCodeServe | None = None + if ( + self._active is not None + and self._base_url is not None + and self._password is not None + ): + base_url = self._base_url + password = self._password + else: + owned = self._serve.spawn() + base_url = "" + password = "" try: set_input_content_attributes( span, @@ -577,16 +658,20 @@ async def _stream_impl( system_instructions=system, messages=_input_messages(user_input, history), ) + if owned is not None: + base_url = await owned.start() + password = owned.password created = _Created() async with aclosing( _reduce_session( created, - base_url=bound_url, + base_url=base_url, + password=password, prompt=prompt, system=system, provider_id=provider_id, model_id=model_id, - agent=bound_agent, + agent=self._agent, capture=capture, ) ) as reduced: @@ -612,9 +697,12 @@ async def _stream_impl( fail_span(span, exc, ended) raise finally: + if owned is not None: + await owned.stop() end_span_once(span, ended) async def _call_impl( + self, config: AiConfigRep, user_input: str | None, tool_handlers: dict[str, Any] | None, @@ -624,7 +712,7 @@ async def _call_impl( output = "" usage: dict[str, Any] = {} async with aclosing( - _stream_impl(config, user_input, tool_handlers, variables, history) + self._stream_impl(config, user_input, tool_handlers, variables, history) ) as stream: async for event in stream: if event.get("type") != "done": @@ -635,13 +723,6 @@ async def _call_impl( usage = raw_usage if isinstance(raw_usage, dict) else {} return {"output": output, "usage": usage} - return _create_handler( - ("*", "agent"), - _call_impl, - _stream_impl, - capture_content=content_override, - ) - class _Created: deltas: list[str] @@ -656,6 +737,7 @@ async def _reduce_session( created: _Created, *, base_url: str, + password: str, prompt: str, system: str | None, provider_id: str, @@ -669,6 +751,7 @@ async def _reduce_session( async with aclosing( session_events( base_url=base_url, + password=password, prompt=prompt, system=system, provider_id=provider_id, diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py new file mode 100644 index 00000000..119da01e --- /dev/null +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py @@ -0,0 +1,254 @@ +"""Start and stop one ``opencode serve`` for a single handler call. + +The process is loopback-only. Each start sets a generated basic-auth password +on the child, and the health probe sends it. Callers pass a working directory, +a config file, and an environment. They do not pass a URL, an HTTP client, or +a transport. +""" + +from __future__ import annotations + +import asyncio +import os +import secrets +import socket +import tempfile +import time +from collections.abc import Mapping, Sequence +from typing import BinaryIO, cast + +import httpx + +_STARTUP_TIMEOUT_SECONDS = 45.0 +_SHUTDOWN_GRACE_SECONDS = 5.0 +_PROBE_INTERVAL_SECONDS = 0.05 +_STDERR_TAIL_BYTES = 2048 +SERVER_USERNAME = "opencode" + +_CommandPart = str | os.PathLike[str] +Command = _CommandPart | Sequence[_CommandPart] | None + + +class OpenCodeServe: + """One ``opencode serve`` child. ``start`` returns its loopback origin.""" + + def __init__( + self, + *, + command: list[str], + cwd: str, + config: str, + env: Mapping[str, str], + ) -> None: + self._command = command + self._cwd = cwd + self._config = config + self._env = env + self._proc: asyncio.subprocess.Process | None = None + self._password: str | None = None + self._stderr: BinaryIO | None = None + + @property + def password(self) -> str: + """The basic-auth secret ``start`` gave this child.""" + if self._password is None: + raise RuntimeError("opencode serve failed to start.") + return self._password + + def spawn(self) -> OpenCodeServe: + """A new child. Calls do not share a process.""" + return OpenCodeServe( + command=list(self._command), + cwd=self._cwd, + config=self._config, + env=dict(self._env), + ) + + async def start(self) -> str: + cwd = os.path.abspath(self._cwd) + config = os.path.abspath(self._config) + if not os.path.isdir(cwd): + raise FileNotFoundError("OpenCode working directory does not exist.") + if not os.path.isfile(config): + raise FileNotFoundError("OpenCode config file does not exist.") + password = secrets.token_urlsafe(32) + self._password = password + port = _free_port() + argv = [ + *self._command, + "serve", + "--port", + str(port), + "--hostname", + "127.0.0.1", + ] + # A pipe fills and blocks when the server writes and nothing reads it. + stderr = tempfile.TemporaryFile() + self._stderr = stderr + try: + self._proc = await asyncio.create_subprocess_exec( + *argv, + cwd=cwd, + env=_child_env(self._env, config, password), + stdout=asyncio.subprocess.DEVNULL, + stderr=stderr, + ) + except OSError as exc: + self._close_stderr() + raise RuntimeError("opencode serve failed to start.") from exc + base = f"http://127.0.0.1:{port}" + try: + await self._wait_until_listening(base, password) + except BaseException: + await self.stop() + raise + return base + + async def stop(self) -> None: + proc = self._proc + self._proc = None + try: + if proc is None: + return + if proc.returncode is None: + proc.terminate() + try: + await asyncio.wait_for(proc.wait(), _SHUTDOWN_GRACE_SECONDS) + except TimeoutError: + proc.kill() + await proc.wait() + else: + await proc.wait() + except (ProcessLookupError, OSError): + return + finally: + self._close_stderr() + + async def _wait_until_listening(self, base: str, password: str) -> None: + proc = self._proc + if proc is None: + raise RuntimeError("opencode serve failed to start.") + deadline = time.monotonic() + _STARTUP_TIMEOUT_SECONDS + async with httpx.AsyncClient(auth=(SERVER_USERNAME, password)) as http: + while time.monotonic() < deadline: + if proc.returncode is not None: + raise self._startup_error( + "opencode serve exited during startup " + f"(code {proc.returncode})." + ) + try: + response = await http.get(f"{base}/global/health", timeout=2.0) + except httpx.HTTPError: + response = None + if response is not None and _is_healthy(response): + return + await asyncio.sleep(_PROBE_INTERVAL_SECONDS) + raise self._startup_error("opencode serve did not become healthy in time.") + + def _startup_error(self, message: str) -> RuntimeError: + tail = self._stderr_tail() + if not tail: + return RuntimeError(message) + return RuntimeError(f"{message}\n{tail}") + + def _stderr_tail(self) -> str: + handle = self._stderr + if handle is None: + return "" + try: + handle.seek(0, os.SEEK_END) + size = handle.tell() + handle.seek(max(0, size - _STDERR_TAIL_BYTES), os.SEEK_SET) + raw = handle.read() + except OSError: + return "" + return raw.decode("utf-8", errors="replace").strip() + + def _close_stderr(self) -> None: + handle = self._stderr + self._stderr = None + if handle is None: + return + try: + handle.close() + except OSError: + return + + +def serve_from( + *, + cwd: str | os.PathLike[str], + config: str | os.PathLike[str], + env: Mapping[str, str], + command: Command, +) -> OpenCodeServe: + """Validate the public inputs and return a server that has not started.""" + return OpenCodeServe( + command=_command_argv(command), + cwd=_require_text(cwd, "working directory"), + config=_require_text(config, "config file"), + env=_require_env(env), + ) + + +def _require_text(value: str | os.PathLike[str], name: str) -> str: + if isinstance(value, os.PathLike): + value = os.fspath(value) + if not isinstance(value, str) or not value.strip(): + raise TypeError(f"The OpenCode handler requires a {name}.") + return value + + +def _require_env(env: Mapping[str, str]) -> dict[str, str]: + if not isinstance(env, Mapping): + raise TypeError("The OpenCode handler requires an env mapping.") + copied: dict[str, str] = {} + for key, item in env.items(): + if not isinstance(key, str) or not isinstance(item, str): + raise TypeError("The OpenCode handler env must be strings.") + copied[key] = item + return copied + + +def _command_argv(command: Command) -> list[str]: + if command is None: + return ["opencode"] + if isinstance(command, (str, os.PathLike)): + return [_one_command(os.fspath(command))] + if isinstance(command, Sequence): + argv = [_one_command(os.fspath(part)) for part in command] + if not argv: + raise TypeError("The OpenCode handler takes a command, not a URL.") + return argv + raise TypeError("The OpenCode handler takes a command, not a URL.") + + +def _one_command(part: str) -> str: + if not part or "://" in part: + raise TypeError("The OpenCode handler takes a command, not a URL.") + return part + + +def _child_env(env: Mapping[str, str], config: str, password: str) -> dict[str, str]: + child = dict(env) + child["OPENCODE_CONFIG"] = config + child["OPENCODE_SERVER_USERNAME"] = SERVER_USERNAME + child["OPENCODE_SERVER_PASSWORD"] = password + return child + + +def _is_healthy(response: httpx.Response) -> bool: + """``/global/health`` is ready only with 200 and ``healthy: true``.""" + if response.status_code != 200: + return False + try: + body = response.json() + except (ValueError, UnicodeDecodeError): + return False + return isinstance(body, dict) and body.get("healthy") is True + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return cast(int, sock.getsockname()[1]) 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 09b02834..013f2726 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py @@ -2,7 +2,8 @@ ``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. +route rejects the amazon-bedrock provider. The base URL is the loopback server +this process just started. It is not a caller argument. """ from __future__ import annotations @@ -14,6 +15,8 @@ import httpx +from .process import SERVER_USERNAME + # 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. @@ -54,6 +57,7 @@ def default_model(config: dict[str, Any]) -> tuple[str, str]: async def session_events( *, base_url: str, + password: str, prompt: str, system: str | None, provider_id: str, @@ -63,15 +67,17 @@ async def session_events( ) -> AsyncGenerator[dict[str, Any], None]: """Open one session and yield its bus events. - The handler owns the HTTP client. Callers pass ``base_url`` for a server - that is already running and do not supply a client or a transport. The - event stream is opened before the message post. The post runs - concurrently with the read. This session's ``session.idle`` ends the + The handler owns the HTTP client. ``base_url`` is the ``opencode serve`` + this call started. ``password`` is the secret that start set on the child, + sent as basic auth with username ``opencode``. Callers do not supply a + client or a transport. The event stream is opened before the message post. + The post runs concurrently with the read. This session's ``session.idle`` ends the turn. A failed message POST raises without waiting for another bus frame. Leaving early cancels the post and aborts the session. """ http = httpx.AsyncClient( - timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS, read=MESSAGE_TIMEOUT_SECONDS) + auth=(SERVER_USERNAME, password), + timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS, read=MESSAGE_TIMEOUT_SECONDS), ) base = base_url.rstrip("/") post: asyncio.Task[httpx.Response] | None = None diff --git a/packages/opencode-agents/tests/fake_opencode.py b/packages/opencode-agents/tests/fake_opencode.py new file mode 100644 index 00000000..202ecaf3 --- /dev/null +++ b/packages/opencode-agents/tests/fake_opencode.py @@ -0,0 +1,292 @@ +"""Fake ``opencode serve`` for handler tests. + +The handler starts this executable. It binds the ``--port`` it was given, +speaks the stable HTTP routes, and exits when the handler stops the process. +""" + +from __future__ import annotations + +import base64 +import json +import os +import select +import socket +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + + +class _Loopback(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + +class _Request(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, fmt: str, *args: Any) -> None: + return + + def do_GET(self) -> None: + script: _Server = self.server.script # type: ignore[attr-defined] + if not script.authorized(self.headers): + self.send_error(401) + return + path = self.path.split("?", 1)[0] + script.note("GET", path) + if path == "/global/health": + body = b'{"healthy":true,"version":"fake"}' + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + if path != "/event": + self.send_error(404) + return + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + try: + script.serve_events(self) + self._chunk(b"") + except (BrokenPipeError, ConnectionResetError): + return + + def do_POST(self) -> None: + script: _Server = self.server.script # type: ignore[attr-defined] + if not script.authorized(self.headers): + self.send_error(401) + return + path = self.path.split("?", 1)[0] + script.note("POST", path) + length = int(self.headers.get("Content-Length", "0") or "0") + body = self.rfile.read(length) if length else b"" + status, payload = script.serve_post(path, body) + raw = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def _chunk(self, data: bytes) -> None: + self.wfile.write(f"{len(data):X}\r\n".encode() + data + b"\r\n") + self.wfile.flush() + + +class _Server: + def __init__(self, spec: dict[str, Any], state_dir: str) -> None: + self.spec = spec + self.state_dir = state_dir + self.sessions: list[str] = [] + self.messages: list[dict[str, Any]] = [] + self.aborts: list[str] = [] + self.requests: list[dict[str, str]] = [] + self.second_read = threading.Event() + self.release = threading.Event() + self._lock = threading.Lock() + + def authorized(self, headers: Any) -> bool: + expected = os.environ.get("OPENCODE_SERVER_PASSWORD", "") + if not expected: + return False + header = headers.get("Authorization") + if not isinstance(header, str) or not header.startswith("Basic "): + return False + try: + decoded = base64.b64decode(header[6:].strip(), validate=True).decode() + except (ValueError, UnicodeDecodeError): + return False + return decoded == f"opencode:{expected}" + + def note(self, method: str, path: str) -> None: + with self._lock: + self.requests.append({"method": method, "path": path}) + self._write() + + def _write(self) -> None: + if not self.state_dir: + return + payload = { + "pid": os.getpid(), + "argv": sys.argv, + "cwd": os.getcwd(), + "config": os.environ.get("OPENCODE_CONFIG", ""), + "username": os.environ.get("OPENCODE_SERVER_USERNAME", ""), + "password": os.environ.get("OPENCODE_SERVER_PASSWORD", ""), + "sessions": self.sessions, + "messages": self.messages, + "aborts": self.aborts, + "requests": self.requests, + } + path = os.path.join(self.state_dir, f"{os.getpid()}.json") + temporary = path + ".tmp" + with open(temporary, "w", encoding="utf-8") as handle: + json.dump(payload, handle) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + + def write(self) -> None: + with self._lock: + self._write() + + def serve_events(self, request: _Request) -> None: + mode = self.spec.get("mode") + if mode == "close_after_read": + request._chunk(b":\n\n") + self.second_read.set() + sock = request.connection + while True: + readable, _, _ = select.select([sock], [], [], 0.05) + if not readable: + continue + try: + peeked = sock.recv(1, socket.MSG_PEEK) + except OSError: + return + if not peeked: + return + events = list(self.spec.get("events") or []) + if mode == "open": + status = int(self.spec.get("message_status") or 200) + delay = float(self.spec.get("delay_tail") or 0) + if status < 400: + events = [ + { + "type": "message.updated", + "properties": { + "info": { + "id": "m1", + "role": "assistant", + "time": {"created": 1}, + } + }, + }, + { + "type": "message.part.delta", + "properties": { + "sessionID": "ses", + "messageID": "m1", + "field": "text", + "delta": "Hi", + "partID": "p1", + }, + }, + ] + else: + events = [] + for event in events: + request._chunk(f"data: {json.dumps(event)}\n\n".encode()) + if delay: + time.sleep(delay) + tail = [ + { + "type": "message.part.delta", + "properties": { + "sessionID": "ses", + "messageID": "m1", + "field": "text", + "delta": "!", + "partID": "p1", + }, + }, + { + "type": "message.updated", + "properties": { + "info": { + "id": "m1", + "role": "assistant", + "time": {"created": 1, "completed": 2}, + "tokens": {"input": 1, "output": 4}, + } + }, + }, + { + "type": "session.idle", + "properties": {"sessionID": "ses"}, + }, + ] + for event in tail: + request._chunk(f"data: {json.dumps(event)}\n\n".encode()) + return + self.release.wait() + return + for event in events: + request._chunk(f"data: {json.dumps(event)}\n\n".encode()) + + def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: + action = "" + with self._lock: + if path == "/session": + session_id = ( + "ses" if not self.sessions else f"ses-{len(self.sessions) + 1}" + ) + self.sessions.append(session_id) + self._write() + return 200, {"id": session_id} + if path.endswith("/message"): + self.messages.append(json.loads(body.decode()) if body else {}) + self._write() + mode = self.spec.get("mode") + if mode == "close_after_read": + action = "second_read" + elif mode == "open": + status = int(self.spec.get("message_status") or 200) + if status >= 400: + return status, {} + if self.spec.get("delay_tail"): + return 200, {} + action = "release" + else: + return 200, {} + elif path.endswith("/abort"): + self.aborts.append(path) + self.release.set() + self._write() + return 200, {} + else: + return 404, {} + if action == "second_read": + self.second_read.wait(timeout=2) + return 400, {} + if action == "release": + self.release.wait() + return 200, {} + return 404, {} + + +def _port(argv: list[str]) -> int: + if "--port" not in argv: + raise SystemExit(2) + return int(argv[argv.index("--port") + 1]) + + +def main() -> None: + spec_path = os.environ["LD_OPENCODE_FAKE_SPEC"] + with open(spec_path, encoding="utf-8") as handle: + spec = json.load(handle) + if not isinstance(spec, dict): + raise SystemExit(2) + server = _Server(spec, os.environ.get("LD_OPENCODE_FAKE_DIR", "")) + server.write() + exit_stderr = spec.get("exit_stderr") + if isinstance(exit_stderr, str) and exit_stderr: + sys.stderr.write(exit_stderr) + sys.stderr.flush() + raise SystemExit(1) + delay_ready = spec.get("delay_ready") + if isinstance(delay_ready, int | float) and delay_ready > 0: + time.sleep(delay_ready) + httpd = _Loopback(("127.0.0.1", _port(sys.argv)), _Request) + httpd.script = server # type: ignore[attr-defined] + httpd.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 74f90c48..8df949c3 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -4,16 +4,17 @@ import asyncio import json -import select -import socket -import threading -import time +import os +import shutil +import sys +import tempfile from collections.abc import Iterator from contextlib import aclosing -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path from typing import Any from unittest.mock import AsyncMock, MagicMock +import httpx import pytest import launchdarkly_ai_server.lifecycle as lifecycle_module @@ -22,9 +23,12 @@ create_opencode_handler, opencode_agents, ) +from launchdarkly_ai_opencode_agents.process import _is_healthy from launchdarkly_ai_opencode_agents.server import default_model from launchdarkly_ai_server import SDKOptions, config, configure +_FAKE = Path(__file__).with_name("fake_opencode.py") + CONTEXT = {"kind": "user", "key": "user-1"} VARIATION = { @@ -112,112 +116,100 @@ def assistant_completed( return bus("message.updated", info=info) -class _Loopback(ThreadingHTTPServer): - daemon_threads = True - allow_reuse_address = True - - -class _Request(BaseHTTPRequestHandler): - protocol_version = "HTTP/1.1" - - def log_message(self, fmt: str, *args: Any) -> None: - return - - def do_GET(self) -> None: - script: _Serve = self.server.script # type: ignore[attr-defined] - if self.path.split("?", 1)[0] != "/event": - self.send_error(404) - return - self.send_response(200) - self.send_header("Content-Type", "text/event-stream") - self.send_header("Transfer-Encoding", "chunked") - self.end_headers() - try: - script.serve_events(self) - self._chunk(b"") - except (BrokenPipeError, ConnectionResetError): - return - - def do_POST(self) -> None: - script: _Serve = self.server.script # type: ignore[attr-defined] - length = int(self.headers.get("Content-Length", "0") or "0") - body = self.rfile.read(length) if length else b"" - status, payload = script.serve_post(self.path.split("?", 1)[0], body) - raw = json.dumps(payload).encode() - self.send_response(status) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(raw))) - self.end_headers() - self.wfile.write(raw) - - def _chunk(self, data: bytes) -> None: - self.wfile.write(f"{len(data):X}\r\n".encode() + data + b"\r\n") - self.wfile.flush() - - -class _Serve: - """A loopback ``opencode serve``. The handler uses its own HTTP client.""" - - def __init__(self) -> None: - self.sessions: list[str] = [] - self.messages: list[dict[str, Any]] = [] - self.aborts: list[str] = [] - self.base_url = "" - self._httpd: _Loopback | None = None - - def start(self) -> str: - httpd = _Loopback(("127.0.0.1", 0), _Request) - httpd.script = self # type: ignore[attr-defined] - self._httpd = httpd - threading.Thread( - target=httpd.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True - ).start() - host, port = httpd.server_address - self.base_url = f"http://{host}:{port}" - _servers.append(self) - return self.base_url - - def stop(self) -> None: - if self._httpd is not None: - self._httpd.shutdown() - self._httpd.server_close() - self._httpd = None - - def serve_events(self, request: _Request) -> None: - return - - def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: - if path == "/session": - session_id = "ses" if not self.sessions else f"ses-{len(self.sessions) + 1}" - self.sessions.append(session_id) - return 200, {"id": session_id} - if path.endswith("/message"): - self.messages.append(json.loads(body.decode()) if body else {}) - return 200, {} - if path.endswith("/abort"): - self.aborts.append(path) - return 200, {} - return 404, {} - - -_servers: list[_Serve] = [] - - -class Script(_Serve): - """One finite bus. The connection closes after the scripted events.""" - - def __init__(self, events: list[dict[str, Any]]) -> None: - super().__init__() - self.events = events - - def serve_events(self, request: _Request) -> None: - for event in self.events: - request._chunk(f"data: {json.dumps(event)}\n\n".encode()) +class Script: + """A fake ``opencode`` executable. The handler starts and stops it.""" + + def __init__( + self, + events: list[dict[str, Any]], + *, + mode: str = "script", + message_status: int = 200, + delay_tail: float = 0, + delay_ready: float = 0, + exit_stderr: str | None = None, + ) -> None: + self.root = Path(tempfile.mkdtemp(prefix="oc-fake-")) + self.cwd = self.root / "work" + self.cwd.mkdir() + self.config = self.root / "opencode.json" + self.config.write_text("{}", encoding="utf-8") + self.state = self.root / "state" + self.state.mkdir() + self.spec_path = self.root / "spec.json" + self.spec_path.write_text( + json.dumps( + { + "mode": mode, + "events": events, + "message_status": message_status, + "delay_tail": delay_tail, + "delay_ready": delay_ready, + "exit_stderr": exit_stderr, + } + ), + encoding="utf-8", + ) + _scripts.append(self) + + def handler_args(self) -> dict[str, Any]: + return { + "cwd": self.cwd, + "config": self.config, + "env": { + "LD_OPENCODE_FAKE_DIR": str(self.state), + "LD_OPENCODE_FAKE_SPEC": str(self.spec_path), + }, + "command": [sys.executable, str(_FAKE)], + } + + def _records(self) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for path in sorted( + self.state.glob("*.json"), key=lambda item: item.stat().st_mtime + ): + records.append(json.loads(path.read_text(encoding="utf-8"))) + return records + + @property + def sessions(self) -> list[str]: + found: list[str] = [] + for record in self._records(): + found.extend(record.get("sessions") or []) + return found + + @property + def messages(self) -> list[dict[str, Any]]: + found: list[dict[str, Any]] = [] + for record in self._records(): + found.extend(record.get("messages") or []) + return found + + @property + def aborts(self) -> list[str]: + found: list[str] = [] + for record in self._records(): + found.extend(record.get("aborts") or []) + return found + + @property + def pids(self) -> list[int]: + return [int(record["pid"]) for record in self._records() if "pid" in record] + + @property + def requests(self) -> list[dict[str, str]]: + found: list[dict[str, str]] = [] + for record in self._records(): + found.extend(record.get("requests") or []) + return found + + +_scripts: list[Script] = [] def playing(events: list[dict[str, Any]], **kwargs: Any) -> tuple[Any, Script]: script = Script(events) - handler = create_opencode_handler(base_url=script.start(), **kwargs) + handler = create_opencode_handler(**script.handler_args(), **kwargs) return handler, script @@ -245,12 +237,14 @@ def _reset_sdk() -> Iterator[None]: lifecycle_module._reset_for_testing() yield lifecycle_module._reset_for_testing() - for server in _servers: - release = getattr(server, "release", None) - if release is not None: - release.set() - server.stop() - _servers.clear() + for script in _scripts: + for pid in script.pids: + try: + os.kill(pid, 15) + except OSError: + pass + shutil.rmtree(script.root, ignore_errors=True) + _scripts.clear() @pytest.fixture @@ -831,18 +825,36 @@ async def test_session_error_does_not_yield_stable_done( class TestInvocation: - def test_base_url_is_required(self) -> None: + def test_directory_config_and_env_are_required(self) -> None: with pytest.raises(TypeError): 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"): + with pytest.raises(TypeError, match="working directory"): + create_opencode_handler(cwd="", config="/tmp/opencode.json", env={}) + with pytest.raises(TypeError, match="config file"): + create_opencode_handler(cwd="/tmp", config="", env={}) + with pytest.raises(TypeError, match="working directory"): opencode_agents("flag", "hi", CONTEXT) + with pytest.raises(TypeError, match="base_url"): + create_opencode_handler( + cwd="/tmp", + config="/tmp/opencode.json", + env={}, + base_url="http://127.0.0.1:9", # type: ignore[call-arg] + ) with pytest.raises(TypeError): create_opencode_handler( - base_url="http://127.0.0.1:9", + cwd="/tmp", + config="/tmp/opencode.json", + env={}, client=object(), # type: ignore[call-arg] ) + with pytest.raises(TypeError, match="URL"): + create_opencode_handler( + cwd="/tmp", + config="/tmp/opencode.json", + env={}, + command="http://127.0.0.1:9", + ) async def test_the_first_session_on_the_stream_is_not_adopted(self) -> None: events = await collect( @@ -1159,7 +1171,7 @@ async def variation(key: str, *_args: Any, **_kwargs: Any) -> dict[str, Any]: 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] + assert len(set(script.pids)) == 2 prompts = [message["parts"][0]["text"] for message in script.messages] systems = [message.get("system", "") for message in script.messages] assert prompts[0] == "q" @@ -1181,8 +1193,8 @@ async def test_the_wrapper_posts_history_on_a_fresh_session( "flag", "Hello", CONTEXT, - base_url=script.start(), history=[{"role": "user", "content": "earlier"}], + **script.handler_args(), ) assert result.response == "Hi" assert script.messages[0]["system"] == "Be brief." @@ -1349,8 +1361,8 @@ 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) - handler = create_opencode_handler(base_url=script.start()) + script = Script([], mode="open", message_status=400) + handler = create_opencode_handler(**script.handler_args()) async def _run() -> None: async for _event in await handler.stream( @@ -1382,8 +1394,8 @@ async def test_a_cancelled_sse_read_is_awaited_before_the_stream_closes( from launchdarkly_ai_opencode_agents import server as server_mod from launchdarkly_ai_opencode_agents.server import OpenCodeHTTPError - bus = _CloseAfterRead() - handler = create_opencode_handler(base_url=bus.start()) + bus = Script([], mode="close_after_read") + handler = create_opencode_handler(**bus.handler_args()) order: list[str] = [] real_next_line = server_mod._next_line @@ -1428,8 +1440,8 @@ async def _run() -> None: ) async def test_breaking_aborts_without_waiting_for_gc(self) -> None: - script = _OpenBus() - handler = create_opencode_handler(base_url=script.start()) + script = Script([], mode="open") + handler = create_opencode_handler(**script.handler_args()) async def _run() -> None: async with aclosing( @@ -1452,8 +1464,8 @@ async def _run() -> None: assert script.aborts async def test_a_late_completion_survives_the_post_returning(self) -> None: - script = _OpenBus(delay_tail=0.2) - handler = create_opencode_handler(base_url=script.start(), capture_content=True) + script = Script([], mode="open", delay_tail=0.2) + handler = create_opencode_handler(**script.handler_args(), capture_content=True) seen = [ event async for event in await handler.stream( @@ -1483,84 +1495,224 @@ async def test_an_idle_without_a_session_id_does_not_end_the_run(self) -> None: assert events[-1]["output"] == "Hi!" -class _CloseAfterRead(_Serve): - """``/event`` stays open. The message POST fails while the next read is blocked.""" +def _alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except OSError: + return False + return True - def __init__(self) -> None: - super().__init__() - self.second_read = threading.Event() - def serve_events(self, request: _Request) -> None: - request._chunk(b":\n\n") - self.second_read.set() - sock = request.connection - while True: - readable, _, _ = select.select([sock], [], [], 0.05) - if not readable: - continue - try: - peeked = sock.recv(1, socket.MSG_PEEK) - except OSError: - return - if not peeked: - return +class TestProcess: + async def test_a_fake_opencode_is_started_spoken_to_and_stopped(self) -> None: + handler, script = playing( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ] + ) + 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" + records = script._records() + assert len(records) == 1 + argv = records[0]["argv"] + assert argv[argv.index("serve") + 1] == "--port" + assert argv[argv.index("--hostname") + 1] == "127.0.0.1" + assert "--print-logs" not in argv + assert "--log-level" not in argv + assert records[0]["cwd"] == str(script.cwd.resolve()) + assert records[0]["config"] == str(script.config.resolve()) + paths = [item["path"] for item in script.requests] + assert "/event" in paths + assert "/session" in paths + assert any(path.endswith("/message") for path in paths) + assert script.pids + assert all(not _alive(pid) for pid in script.pids) + + async def test_closing_the_stream_stops_the_process(self) -> None: + script = Script([], mode="open") + handler = create_opencode_handler(**script.handler_args()) - def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: - if path.endswith("/message"): - self.second_read.wait(timeout=2) - self.messages.append(json.loads(body.decode()) if body else {}) - return 400, {} - return super().serve_post(path, body) - - -class _OpenBus(_Serve): - """An `/event` stream that stays open, unlike a finite ``Script`` body.""" - - def __init__(self, *, message_status: int = 200, delay_tail: float = 0) -> None: - super().__init__() - self.message_status = message_status - self.delay_tail = delay_tail - self.release = threading.Event() - - def serve_events(self, request: _Request) -> None: - # A rejected message sends nothing. Events here would let the loop - # notice the 400 without racing the POST. - head = ( - [] - if self.message_status >= 400 - else [ + 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=5) + assert script.aborts + assert script.pids + assert all(not _alive(pid) for pid in script.pids) + + async def test_the_process_is_dead_when_the_context_returns(self) -> None: + script = Script([], mode="open") + handler = create_opencode_handler(**script.handler_args()) + with pytest.raises(TimeoutError): + async with handler: + async with asyncio.timeout(0.25): + 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": + await asyncio.sleep(2) + assert script.pids + assert all(not _alive(pid) for pid in script.pids) + + async def test_boot_is_outside_the_caller_timeout(self) -> None: + script = Script( + [ assistant_open("m1"), text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ], + delay_ready=0.6, + ) + handler = create_opencode_handler(**script.handler_args()) + async with handler: + async with asyncio.timeout(0.2): + 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" + + async def test_two_concurrent_contexts_leave_no_process(self) -> None: + script = Script([]) + handler = create_opencode_handler(**script.handler_args()) + + async def user() -> None: + async with handler: + await asyncio.sleep(0.1) + + await asyncio.gather(user(), user()) + assert script.pids + assert len(script.pids) == 1 + assert all(not _alive(pid) for pid in script.pids) + + async def test_a_generated_password_replaces_one_from_the_environment( + self, + ) -> None: + script = Script( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), ] ) - for event in head: - request._chunk(f"data: {json.dumps(event)}\n\n".encode()) - if self.delay_tail: - time.sleep(self.delay_tail) - tail = [ - text_delta("m1", "p1", "!", session_id="ses"), - assistant_completed("m1", tokens={"input": 1, "output": 4}), - bus("session.idle", sessionID="ses"), + args = script.handler_args() + args["env"] = {**args["env"], "OPENCODE_SERVER_PASSWORD": "inherited-secret"} + handler = create_opencode_handler(**args) + 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" + records = script._records() + assert len(records) == 1 + password = records[0]["password"] + assert isinstance(password, str) and password + assert password != "inherited-secret" + + async def test_an_inherited_username_is_replaced(self) -> None: + script = Script( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), ] - for event in tail: - request._chunk(f"data: {json.dumps(event)}\n\n".encode()) - return - self.release.wait() - - def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: - if path.endswith("/message"): - self.messages.append(json.loads(body.decode()) if body else {}) - if self.message_status >= 400: - return self.message_status, {} - if self.delay_tail: - return 200, {} - self.release.wait() - return 200, {} - if path.endswith("/abort"): - self.aborts.append(path) - self.release.set() - return 200, {} - return super().serve_post(path, body) + ) + args = script.handler_args() + args["env"] = {**args["env"], "OPENCODE_SERVER_USERNAME": "someone-else"} + handler = create_opencode_handler(**args) + 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" + records = script._records() + assert len(records) == 1 + assert records[0]["username"] == "opencode" + + async def test_a_startup_failure_includes_the_stderr_tail(self) -> None: + prefix = "HEAD-MARKER" + tail = "TAIL-MARKER" + script = Script([], exit_stderr=prefix + ("x" * 4000) + tail) + handler = create_opencode_handler(**script.handler_args()) + with pytest.raises(RuntimeError) as caught: + async with handler: + pass + message = str(caught.value) + assert "exited during startup" in message + assert tail in message + assert prefix not in message + + def test_health_requires_a_healthy_body(self) -> None: + assert _is_healthy(httpx.Response(200, json={"healthy": True, "version": "1"})) + assert not _is_healthy(httpx.Response(404, json={"healthy": True})) + assert not _is_healthy(httpx.Response(200, json={})) + assert not _is_healthy(httpx.Response(200, json={"healthy": False})) + assert not _is_healthy(httpx.Response(200, content=b"ok")) async def _drain(stream: Any) -> None: diff --git a/packages/opencode-agents/tests/test_sdk_usage.py b/packages/opencode-agents/tests/test_sdk_usage.py index f2d64d9f..7667eb90 100644 --- a/packages/opencode-agents/tests/test_sdk_usage.py +++ b/packages/opencode-agents/tests/test_sdk_usage.py @@ -51,7 +51,7 @@ def _usage(client: MagicMock) -> list[Any]: async def test_create_handler_sends_one_usage_event(installed: MagicMock) -> None: - create_opencode_handler(base_url="http://127.0.0.1:9") + create_opencode_handler(cwd="/tmp", config="/tmp/opencode.json", env={}) calls = _usage(installed) assert [entry.args[2].get("helper") for entry in calls] == [ "opencode-agents.createOpenCodeHandler" @@ -69,7 +69,16 @@ async def test_create_handler_sends_one_usage_event(installed: MagicMock) -> Non async def test_wrapper_sends_one_usage_event(installed: MagicMock) -> None: - await _settle(opencode_agents("k", "q", CONTEXT, base_url="http://127.0.0.1:9")) + await _settle( + opencode_agents( + "k", + "q", + CONTEXT, + cwd="/tmp", + config="/no/such/opencode.json", + env={}, + ) + ) calls = _usage(installed) assert [entry.args[2].get("helper") for entry in calls] == [ "opencode-agents.opencodeAgents"