From c5f463400dc81de853e9fce2c8db180c2665ca17 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 22:45:55 +0000 Subject: [PATCH 1/6] feat(opencode)!: start and stop opencode serve in the handler Each call starts opencode serve from the working directory, config file, and environment, speaks the OpenCode HTTP API itself, and stops the process when the stream finishes or the caller closes it. Callers no longer pass base_url. A test can pass a fake command instead of a URL. config().stream() is unchanged. Co-authored-by: alexis --- AGENTS.md | 2 +- packages/opencode-agents/README.md | 13 +- packages/opencode-agents/agents.md | 16 +- .../handler.py | 64 +-- .../process.py | 194 +++++++++ .../launchdarkly_ai_opencode_agents/server.py | 7 +- .../opencode-agents/tests/fake_opencode.py | 262 ++++++++++++ .../opencode-agents/tests/test_handler.py | 398 +++++++++--------- .../opencode-agents/tests/test_sdk_usage.py | 13 +- 9 files changed, 730 insertions(+), 239 deletions(-) create mode 100644 packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py create mode 100644 packages/opencode-agents/tests/fake_opencode.py 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..73c9be8d 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`. Each call starts `opencode serve`, speaks the OpenCode HTTP API, and stops the process when the stream finishes or the caller closes it. 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,7 +18,11 @@ 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") + handler = create_opencode_handler( + cwd="/work", + config="/home/app/.vega/opencode.json", + env=os.environ.copy(), + ) async for event in config(key="my-agent", handler=handler).stream( "Hello", {"kind": "user", "key": "user-1"} ): @@ -28,4 +33,6 @@ async def main() -> None: 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` and binds `opencode serve` to loopback. 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. + +Each call, including a judge routed to this handler, starts its own process and session. A judge 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..1bfd8294 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`. `command` defaults to `opencode`. A test passes a fake executable instead of a URL. There is no `base_url`, client, or transport argument. + +The process starts on the call and stops when the stream finishes or the caller closes it. Closing the stream still aborts the session before the process is stopped. 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..3cea4861 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,14 @@ """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``. +Each call starts ``opencode serve``, opens one session, and stops the process +when the stream finishes or the caller closes it. 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 os from collections.abc import AsyncGenerator, AsyncIterable, Mapping from contextlib import aclosing from typing import Any @@ -33,6 +35,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 +490,32 @@ 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*. *command* is the + executable, ``opencode`` when omitted. A test passes a fake executable + here. There is no URL, client, or transport argument. + + The process starts when the call starts and stops when the stream finishes + or the caller closes it. A judge routed here gets its own process and + session and 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, + serve=bound, capture_content=capture_content, model=model or default_model, agent=agent, @@ -525,15 +533,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, + serve=serve_from(cwd=cwd, config=config_path, env=env, command=command), capture_content=capture_content, model=model or default_model, agent=agent, @@ -544,13 +557,13 @@ def opencode_agents( def _create_opencode_handler( *, - base_url: str, + serve: OpenCodeServe, capture_content: bool | None, model: ModelResolver, agent: str | None, ) -> ProviderHandler: content_override = capture_content - bound_url = base_url + bound_serve = serve resolve_model = model bound_agent = agent @@ -570,6 +583,7 @@ async def _stream_impl( ) system = _system_text(config, variables) prompt = _user_prompt(config, user_input, history, variables) + owned = bound_serve.spawn() try: set_input_content_attributes( span, @@ -577,11 +591,12 @@ async def _stream_impl( system_instructions=system, messages=_input_messages(user_input, history), ) + base_url = await owned.start() created = _Created() async with aclosing( _reduce_session( created, - base_url=bound_url, + base_url=base_url, prompt=prompt, system=system, provider_id=provider_id, @@ -612,6 +627,7 @@ async def _stream_impl( fail_span(span, exc, ended) raise finally: + await owned.stop() end_span_once(span, ended) async def _call_impl( 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..22c7a878 --- /dev/null +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py @@ -0,0 +1,194 @@ +"""Start and stop one ``opencode serve`` for a single handler call. + +The process is loopback-only. 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 socket +import time +from collections.abc import Mapping, Sequence + +import httpx + +_STARTUP_TIMEOUT_SECONDS = 45.0 +_SHUTDOWN_GRACE_SECONDS = 5.0 +_PROBE_INTERVAL_SECONDS = 0.05 + +_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 + + 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.") + port = _free_port() + argv = [ + *self._command, + "serve", + "--port", + str(port), + "--hostname", + "127.0.0.1", + "--print-logs", + "--log-level", + "INFO", + ] + try: + self._proc = await asyncio.create_subprocess_exec( + *argv, + cwd=cwd, + env=_child_env(self._env, config), + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + except OSError as exc: + raise RuntimeError("opencode serve failed to start.") from exc + base = f"http://127.0.0.1:{port}" + try: + await self._wait_until_listening(base) + except BaseException: + await self.stop() + raise + return base + + async def stop(self) -> None: + proc = self._proc + if proc is None: + return + self._proc = None + try: + 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 + + async def _wait_until_listening(self, base: 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() as http: + while time.monotonic() < deadline: + if proc.returncode is not None: + raise RuntimeError( + "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 response.status_code < 500: + return + await asyncio.sleep(_PROBE_INTERVAL_SECONDS) + raise RuntimeError("opencode serve did not become healthy in time.") + + +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) -> dict[str, str]: + child = dict(env) + child["OPENCODE_CONFIG"] = config + return child + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + allocated = sock.getsockname()[1] + if not isinstance(allocated, int): + raise RuntimeError("opencode serve failed to start.") + return allocated 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..443b0362 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 @@ -63,8 +64,8 @@ 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 + The handler owns the HTTP client. ``base_url`` is the ``opencode serve`` + this call started. 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 diff --git a/packages/opencode-agents/tests/fake_opencode.py b/packages/opencode-agents/tests/fake_opencode.py new file mode 100644 index 00000000..01e6b22a --- /dev/null +++ b/packages/opencode-agents/tests/fake_opencode.py @@ -0,0 +1,262 @@ +"""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 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] + path = self.path.split("?", 1)[0] + script.note("GET", path) + if path == "/global/health": + body = b"{}" + 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] + 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 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", ""), + "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() + 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..c872a4da 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -4,13 +4,13 @@ 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 @@ -25,6 +25,8 @@ 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 +114,96 @@ 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, + ) -> 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, + } + ), + 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 +231,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 +819,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 +1165,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 +1187,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 +1355,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 +1388,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 +1434,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 +1458,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 +1489,76 @@ 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 __init__(self) -> None: - super().__init__() - self.second_read = threading.Event() +def _alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except OSError: + return False + return True - 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 - 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 [ +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"), ] ) - for event in head: - request._chunk(f"data: {json.dumps(event)}\n\n".encode()) - if self.delay_tail: - time.sleep(self.delay_tail) - tail = [ - text_delta("m1", "p1", "!", session_id="ses"), - assistant_completed("m1", tokens={"input": 1, "output": 4}), - bus("session.idle", sessionID="ses"), - ] - for event in tail: - request._chunk(f"data: {json.dumps(event)}\n\n".encode()) - return - self.release.wait() - - def serve_post(self, path: str, body: bytes) -> tuple[int, dict[str, Any]]: - if path.endswith("/message"): - self.messages.append(json.loads(body.decode()) if body else {}) - if self.message_status >= 400: - return self.message_status, {} - if self.delay_tail: - return 200, {} - self.release.wait() - return 200, {} - if path.endswith("/abort"): - self.aborts.append(path) - self.release.set() - return 200, {} - return super().serve_post(path, body) + 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 records[0]["cwd"] == str(script.cwd) + 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()) + + 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 _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" From 008bfa0617429b9c155b9b53563c11307a2717e7 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 23:19:49 +0000 Subject: [PATCH 2/6] fix(opencode): stop opencode serve after the caller returns async with handler starts the process 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. A wall-clock budget stays around config().stream(), so boot is not part of it. Co-authored-by: alexis --- packages/opencode-agents/README.md | 13 +- packages/opencode-agents/agents.md | 2 +- .../handler.py | 120 ++++++++++++------ .../opencode-agents/tests/fake_opencode.py | 3 + .../opencode-agents/tests/test_handler.py | 53 ++++++++ 5 files changed, 145 insertions(+), 46 deletions(-) diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index 73c9be8d..74a2add1 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`. Each call starts `opencode serve`, speaks the OpenCode HTTP API, and stops the process when the stream finishes or the caller closes it. 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. @@ -23,10 +23,11 @@ async def main() -> None: config="/home/app/.vega/opencode.json", env=os.environ.copy(), ) - async for event in config(key="my-agent", handler=handler).stream( - "Hello", {"kind": "user", "key": "user-1"} - ): - print(event) + async with handler: + async for event in config(key="my-agent", handler=handler).stream( + "Hello", {"kind": "user", "key": "user-1"} + ): + print(event) await shutdown() @@ -35,4 +36,4 @@ asyncio.run(main()) `cwd` is the working directory, `config` is the OpenCode config file, and `env` is the child environment. The handler sets `OPENCODE_CONFIG` from `config` and binds `opencode serve` to loopback. 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. -Each call, including a judge routed to this handler, starts its own process and session. A judge 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`. +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 1bfd8294..41bb5457 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -34,7 +34,7 @@ A run opens one `invoke_agent` span and sets `launchdarkly.operation.type`. The `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`. `command` defaults to `opencode`. A test passes a fake executable instead of a URL. There is no `base_url`, client, or transport argument. -The process starts on the call and stops when the stream finishes or the caller closes it. Closing the stream still aborts the session before the process is stopped. +`async with handler` starts the process and returns once it is healthy. The process stops when that 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. 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 3cea4861..57ea59ea 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -1,9 +1,10 @@ """Reduce OpenCode session events into handler stream events. -Each call starts ``opencode serve``, opens one session, and stops the process -when the stream finishes or the caller closes it. 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 @@ -19,7 +20,6 @@ SpanMessage, SpanUsage, _config, - _create_handler, add_cached_tokens_to_input, compose_history, content_to_text, @@ -506,15 +506,18 @@ def create_opencode_handler( executable, ``opencode`` when omitted. A test passes a fake executable here. There is no URL, client, or transport argument. - The process starts when the call starts and stops when the stream finishes - or the caller closes it. A judge routed here gets its own process and - session and 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. + 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. """ bound = serve_from(cwd=cwd, config=config, env=env, command=command) report_usage("opencode-agents.createOpenCodeHandler", PACKAGE_NAME, __version__) - return _create_opencode_handler( + return OpenCodeHandler( serve=bound, capture_content=capture_content, model=model or default_model, @@ -545,7 +548,7 @@ def opencode_agents( agent = kwargs.pop("agent", None) return _config( key=config_key, - handler=_create_opencode_handler( + handler=OpenCodeHandler( serve=serve_from(cwd=cwd, config=config_path, env=env, command=command), capture_content=capture_content, model=model or default_model, @@ -555,19 +558,57 @@ def opencode_agents( ).invoke(user_input, context, variables, history) -def _create_opencode_handler( - *, - serve: OpenCodeServe, - capture_content: bool | None, - model: ModelResolver, - agent: str | None, -) -> ProviderHandler: - content_override = capture_content - bound_serve = serve - 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._depth = 0 + 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.""" + if self._active is None: + active = self._serve.spawn() + self._base_url = await active.start() + 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.""" + self._depth -= 1 + if self._depth > 0: + return + active = self._active + self._active = None + self._base_url = 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, @@ -575,15 +616,20 @@ 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) - owned = bound_serve.spawn() + # 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: + base_url = self._base_url + else: + owned = self._serve.spawn() + base_url = "" try: set_input_content_attributes( span, @@ -591,7 +637,8 @@ async def _stream_impl( system_instructions=system, messages=_input_messages(user_input, history), ) - base_url = await owned.start() + if owned is not None: + base_url = await owned.start() created = _Created() async with aclosing( _reduce_session( @@ -601,7 +648,7 @@ async def _stream_impl( system=system, provider_id=provider_id, model_id=model_id, - agent=bound_agent, + agent=self._agent, capture=capture, ) ) as reduced: @@ -627,10 +674,12 @@ async def _stream_impl( fail_span(span, exc, ended) raise finally: - await owned.stop() + 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, @@ -640,7 +689,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": @@ -651,13 +700,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] diff --git a/packages/opencode-agents/tests/fake_opencode.py b/packages/opencode-agents/tests/fake_opencode.py index 01e6b22a..bc763319 100644 --- a/packages/opencode-agents/tests/fake_opencode.py +++ b/packages/opencode-agents/tests/fake_opencode.py @@ -253,6 +253,9 @@ def main() -> None: raise SystemExit(2) server = _Server(spec, os.environ.get("LD_OPENCODE_FAKE_DIR", "")) server.write() + 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() diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index c872a4da..3b449b36 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -124,6 +124,7 @@ def __init__( mode: str = "script", message_status: int = 200, delay_tail: float = 0, + delay_ready: float = 0, ) -> None: self.root = Path(tempfile.mkdtemp(prefix="oc-fake-")) self.cwd = self.root / "work" @@ -140,6 +141,7 @@ def __init__( "events": events, "message_status": message_status, "delay_tail": delay_tail, + "delay_ready": delay_ready, } ), encoding="utf-8", @@ -1560,6 +1562,57 @@ async def _run() -> None: 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 _drain(stream: Any) -> None: async for _event in stream: From 920a75035e9fa50d8cc2e9c8ad06793fa9f879f7 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 23:26:54 +0000 Subject: [PATCH 3/6] fix(opencode): pass only serve flags to opencode --print-logs and --log-level are not serve options. A binary that rejects them exits before it listens, and the handler already discards the child's output. Co-authored-by: alexis --- .../src/launchdarkly_ai_opencode_agents/process.py | 3 --- packages/opencode-agents/tests/test_handler.py | 2 ++ 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py index 22c7a878..28e68b17 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py @@ -63,9 +63,6 @@ async def start(self) -> str: str(port), "--hostname", "127.0.0.1", - "--print-logs", - "--log-level", - "INFO", ] try: self._proc = await asyncio.create_subprocess_exec( diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 3b449b36..28a3b802 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -1527,6 +1527,8 @@ async def test_a_fake_opencode_is_started_spoken_to_and_stopped(self) -> None: 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) assert records[0]["config"] == str(script.config.resolve()) paths = [item["path"] for item in script.requests] From 021187ee8484ef8b9b4d076aaaa1e0fe5ee84597 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 22 Oct 2026 14:58:27 +0000 Subject: [PATCH 4/6] fix(opencode): lock serve lifetime and require a server password Concurrent entries share one process, and the last exit stops it. Each start sets OPENCODE_SERVER_PASSWORD and sends that basic auth. A failed start includes the stderr tail, and health requires a healthy response. Co-authored-by: alexis --- packages/opencode-agents/README.md | 2 +- packages/opencode-agents/agents.md | 4 +- .../handler.py | 62 ++++++++---- .../process.py | 95 +++++++++++++++---- .../launchdarkly_ai_opencode_agents/server.py | 11 ++- .../opencode-agents/tests/fake_opencode.py | 28 +++++- .../opencode-agents/tests/test_handler.py | 70 ++++++++++++++ 7 files changed, 228 insertions(+), 44 deletions(-) diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index 74a2add1..6c8ab3ea 100644 --- a/packages/opencode-agents/README.md +++ b/packages/opencode-agents/README.md @@ -34,6 +34,6 @@ async def main() -> None: asyncio.run(main()) ``` -`cwd` is the working directory, `config` is the OpenCode config file, and `env` is the child environment. The handler sets `OPENCODE_CONFIG` from `config` and binds `opencode serve` to loopback. 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. +`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_PASSWORD` in `env` with a generated secret, and binds `opencode serve` to loopback. Requests to that server use that 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 41bb5457..bed82e86 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -32,9 +32,9 @@ A run opens one `invoke_agent` span and sets `launchdarkly.operation.type`. The ## Call shape -`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`. `command` defaults to `opencode`. A test passes a fake executable instead of a URL. There is no `base_url`, client, or transport argument. +`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_PASSWORD` to a generated secret, replacing one inherited from `env`, and sends it 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 it is healthy. The process stops when that 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. +`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. 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 57ea59ea..23e51b2d 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -9,6 +9,7 @@ from __future__ import annotations +import asyncio import os from collections.abc import AsyncGenerator, AsyncIterable, Mapping from contextlib import aclosing @@ -502,9 +503,11 @@ def create_opencode_handler( ``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*. *command* is the - executable, ``opencode`` when omitted. A test passes a fake executable - here. There is no URL, client, or transport argument. + The handler sets ``OPENCODE_CONFIG`` from *config* and + ``OPENCODE_SERVER_PASSWORD`` to a generated secret, replacing any password + 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 @@ -574,7 +577,9 @@ def __init__( 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"), @@ -583,13 +588,20 @@ def __init__( ) async def __aenter__(self) -> OpenCodeHandler: - """Start ``opencode serve`` and return once it answers on loopback.""" - if self._active is None: - active = self._serve.spawn() - self._base_url = await active.start() - self._active = active - self._depth += 1 - return self + """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, @@ -598,14 +610,16 @@ async def __aexit__( traceback: Any, ) -> None: """Stop the process and wait until it has exited.""" - self._depth -= 1 - if self._depth > 0: - return - active = self._active - self._active = None - self._base_url = None - if active is not None: - await active.stop() + 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, @@ -625,11 +639,17 @@ async def _stream_impl( # 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: + 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, @@ -639,11 +659,13 @@ async def _stream_impl( ) if owned is not None: base_url = await owned.start() + password = owned.password created = _Created() async with aclosing( _reduce_session( created, base_url=base_url, + password=password, prompt=prompt, system=system, provider_id=provider_id, @@ -714,6 +736,7 @@ async def _reduce_session( created: _Created, *, base_url: str, + password: str, prompt: str, system: str | None, provider_id: str, @@ -727,6 +750,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 index 28e68b17..9beba73a 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py @@ -1,22 +1,28 @@ """Start and stop one ``opencode serve`` for a single handler call. -The process is loopback-only. Callers pass a working directory, a config file, -and an environment. They do not pass a URL, an HTTP client, or a transport. +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 _CommandPart = str | os.PathLike[str] Command = _CommandPart | Sequence[_CommandPart] | None @@ -38,6 +44,15 @@ def __init__( 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.""" @@ -55,6 +70,8 @@ async def start(self) -> str: 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, @@ -64,19 +81,23 @@ async def start(self) -> str: "--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), + env=_child_env(self._env, config, password), stdout=asyncio.subprocess.DEVNULL, - stderr=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) + await self._wait_until_listening(base, password) except BaseException: await self.stop() raise @@ -84,10 +105,10 @@ async def start(self) -> str: async def stop(self) -> None: proc = self._proc - if proc is None: - return self._proc = None try: + if proc is None: + return if proc.returncode is None: proc.terminate() try: @@ -99,16 +120,18 @@ async def stop(self) -> None: await proc.wait() except (ProcessLookupError, OSError): return + finally: + self._close_stderr() - async def _wait_until_listening(self, base: str) -> None: + 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() as http: + async with httpx.AsyncClient(auth=("opencode", password)) as http: while time.monotonic() < deadline: if proc.returncode is not None: - raise RuntimeError( + raise self._startup_error( "opencode serve exited during startup " f"(code {proc.returncode})." ) @@ -116,10 +139,39 @@ async def _wait_until_listening(self, base: str) -> None: response = await http.get(f"{base}/global/health", timeout=2.0) except httpx.HTTPError: response = None - if response is not None and response.status_code < 500: + if response is not None and _is_healthy(response): return await asyncio.sleep(_PROBE_INTERVAL_SECONDS) - raise RuntimeError("opencode serve did not become healthy in time.") + 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( @@ -176,16 +228,25 @@ def _one_command(part: str) -> str: return part -def _child_env(env: Mapping[str, str], config: str) -> dict[str, str]: +def _child_env(env: Mapping[str, str], config: str, password: str) -> dict[str, str]: child = dict(env) child["OPENCODE_CONFIG"] = config + 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)) - allocated = sock.getsockname()[1] - if not isinstance(allocated, int): - raise RuntimeError("opencode serve failed to start.") - return allocated + 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 443b0362..10bee240 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py @@ -55,6 +55,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, @@ -65,14 +66,16 @@ async def session_events( """Open one session and yield its bus events. The handler owns the HTTP client. ``base_url`` is the ``opencode serve`` - this call started. 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 + 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=("opencode", 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 index bc763319..60a55631 100644 --- a/packages/opencode-agents/tests/fake_opencode.py +++ b/packages/opencode-agents/tests/fake_opencode.py @@ -6,6 +6,7 @@ from __future__ import annotations +import base64 import json import os import select @@ -30,10 +31,13 @@ def log_message(self, fmt: str, *args: Any) -> None: 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"{}" + 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))) @@ -55,6 +59,9 @@ def do_GET(self) -> None: 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") @@ -84,6 +91,19 @@ def __init__(self, spec: dict[str, Any], state_dir: str) -> None: 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}) @@ -97,6 +117,7 @@ def _write(self) -> None: "argv": sys.argv, "cwd": os.getcwd(), "config": os.environ.get("OPENCODE_CONFIG", ""), + "password": os.environ.get("OPENCODE_SERVER_PASSWORD", ""), "sessions": self.sessions, "messages": self.messages, "aborts": self.aborts, @@ -253,6 +274,11 @@ def main() -> None: 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) diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 28a3b802..dfc27af3 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -14,6 +14,7 @@ from typing import Any from unittest.mock import AsyncMock, MagicMock +import httpx import pytest import launchdarkly_ai_server.lifecycle as lifecycle_module @@ -22,6 +23,7 @@ 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 @@ -125,6 +127,7 @@ def __init__( 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" @@ -142,6 +145,7 @@ def __init__( "message_status": message_status, "delay_tail": delay_tail, "delay_ready": delay_ready, + "exit_stderr": exit_stderr, } ), encoding="utf-8", @@ -1615,6 +1619,72 @@ async def test_boot_is_outside_the_caller_timeout(self) -> 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"), + ] + ) + 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_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: async for _event in stream: From 38533a9f42ef173b293e421bf63a6ac15680ba41 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 22 Oct 2026 15:08:58 +0000 Subject: [PATCH 5/6] fix(opencode): force the serve username to opencode A copied environment can set OPENCODE_SERVER_USERNAME. The clients always authenticate as opencode, so that inherited name made every request 401. Co-authored-by: alexis --- packages/opencode-agents/README.md | 2 +- packages/opencode-agents/agents.md | 2 +- .../handler.py | 5 ++-- .../process.py | 4 ++- .../launchdarkly_ai_opencode_agents/server.py | 4 ++- .../opencode-agents/tests/fake_opencode.py | 1 + .../opencode-agents/tests/test_handler.py | 29 +++++++++++++++++++ 7 files changed, 41 insertions(+), 6 deletions(-) diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index 6c8ab3ea..9037e5d5 100644 --- a/packages/opencode-agents/README.md +++ b/packages/opencode-agents/README.md @@ -34,6 +34,6 @@ async def main() -> None: asyncio.run(main()) ``` -`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_PASSWORD` in `env` with a generated secret, and binds `opencode serve` to loopback. Requests to that server use that 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. +`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 bed82e86..f6f22412 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -32,7 +32,7 @@ A run opens one `invoke_agent` span and sets `launchdarkly.operation.type`. The ## Call shape -`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_PASSWORD` to a generated secret, replacing one inherited from `env`, and sends it 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. +`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. 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 23e51b2d..6335ee1d 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -503,8 +503,9 @@ def create_opencode_handler( ``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* and - ``OPENCODE_SERVER_PASSWORD`` to a generated secret, replacing any password + 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. diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py index 9beba73a..119da01e 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/process.py @@ -23,6 +23,7 @@ _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 @@ -128,7 +129,7 @@ async def _wait_until_listening(self, base: str, password: str) -> None: if proc is None: raise RuntimeError("opencode serve failed to start.") deadline = time.monotonic() + _STARTUP_TIMEOUT_SECONDS - async with httpx.AsyncClient(auth=("opencode", password)) as http: + async with httpx.AsyncClient(auth=(SERVER_USERNAME, password)) as http: while time.monotonic() < deadline: if proc.returncode is not None: raise self._startup_error( @@ -231,6 +232,7 @@ def _one_command(part: str) -> str: 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 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 10bee240..013f2726 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/server.py @@ -15,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. @@ -74,7 +76,7 @@ async def session_events( frame. Leaving early cancels the post and aborts the session. """ http = httpx.AsyncClient( - auth=("opencode", password), + auth=(SERVER_USERNAME, password), timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS, read=MESSAGE_TIMEOUT_SECONDS), ) base = base_url.rstrip("/") diff --git a/packages/opencode-agents/tests/fake_opencode.py b/packages/opencode-agents/tests/fake_opencode.py index 60a55631..202ecaf3 100644 --- a/packages/opencode-agents/tests/fake_opencode.py +++ b/packages/opencode-agents/tests/fake_opencode.py @@ -117,6 +117,7 @@ def _write(self) -> None: "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, diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index dfc27af3..510edf92 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -1665,6 +1665,35 @@ async def test_a_generated_password_replaces_one_from_the_environment( 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"), + ] + ) + 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" From ab91afd44400b5bf0a4c20dea715a55af9b9b4dc Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 22 Oct 2026 15:31:09 +0000 Subject: [PATCH 6/6] test(opencode): resolve the fake cwd before comparing macOS reports /var through /private/var. The child cwd is already resolved, so the assertion has to compare the resolved path. Co-authored-by: alexis --- packages/opencode-agents/tests/test_handler.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 510edf92..8df949c3 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -1533,7 +1533,7 @@ async def test_a_fake_opencode_is_started_spoken_to_and_stopped(self) -> None: 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) + 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