From a69ffc7fe25c4dc22ee17cdd64dd55904f125647 Mon Sep 17 00:00:00 2001 From: shrijayan <81805145+shrijayan@users.noreply.github.com> Date: Thu, 22 Oct 2026 15:57:24 +0530 Subject: [PATCH 1/3] Add local background service helpers ported from the TypeScript client opencode.service gets discover, ensure, stop and headers, with async twins, matching packages/client/src/promise/service.ts at OpenCode v2.1.2: the registration-file contract, health classification, version replacement, contender race handling and timing, and persistent terminal handoff. Opencode.from_service and AsyncOpencode.from_service connect to it. Includes unit tests with a fake service process, a live test in temporary XDG dirs, a README section and docs pages. --- README.md | 21 + docs/getting-started/index.md | 1 + docs/getting-started/service.md | 39 ++ docs/reference/index.md | 1 + docs/reference/service.md | 15 + mkdocs.yml | 2 + src/opencode/__init__.py | 5 + src/opencode/_client.py | 108 +++- src/opencode/_service.py | 1000 +++++++++++++++++++++++++++++++ src/opencode/service.py | 50 ++ tests/fake_service.py | 77 +++ tests/test_service.py | 273 +++++++++ tests/test_service_live.py | 67 +++ 13 files changed, 1658 insertions(+), 1 deletion(-) create mode 100644 docs/getting-started/service.md create mode 100644 docs/reference/service.md create mode 100644 src/opencode/_service.py create mode 100644 src/opencode/service.py create mode 100644 tests/fake_service.py create mode 100644 tests/test_service.py create mode 100644 tests/test_service_live.py diff --git a/README.md b/README.md index 60d7068..05ba03d 100644 --- a/README.md +++ b/README.md @@ -382,6 +382,27 @@ with server: The server runs in `--stdio` mode, so it also exits when the Python process dies. `AsyncOpencodeServer` is the async equivalent. +### The background service + +OpenCode can also run one shared background server per user (`opencode serve --service`), which registers itself in +`$XDG_STATE_HOME/opencode/service.json`. `Opencode.from_service()` connects to it, starting it first if it is not +running: + +```python +from opencode import Opencode, service + +with Opencode.from_service(version="2.1.2") as client: # replaces a service running another version + print(client.server.info().version) + +endpoint = service.discover() # an Endpoint, or None; never starts anything +service.stop(pty="handoff") # hand persistent terminals to the next service instead of closing them +``` + +`service.ensure()` returns the `Endpoint` (URL and credentials) without creating a client, and `service.headers()` +gives its `Authorization` header for other HTTP clients. Concurrent callers are safe: they race to start the service +and all get the one that won. `await AsyncOpencode.from_service()`, `service.async_discover()`, +`service.async_ensure()` and `service.async_stop()` are the async equivalents. + ## Advanced ```python diff --git a/docs/getting-started/index.md b/docs/getting-started/index.md index f938b54..4aa4613 100644 --- a/docs/getting-started/index.md +++ b/docs/getting-started/index.md @@ -33,3 +33,4 @@ as well, so that `opencode` is on `PATH`. - [Connect to a server](connect.md) that is already running. - [Run a server from Python](server.md) for scripts, tests and embedded use. +- [Use the background service](service.md) to share one `opencode serve --service` between programs. diff --git a/docs/getting-started/service.md b/docs/getting-started/service.md new file mode 100644 index 0000000..da2177e --- /dev/null +++ b/docs/getting-started/service.md @@ -0,0 +1,39 @@ +# Use the background service + +OpenCode can run one shared background server per user, `opencode serve --service`. It registers itself in +`$XDG_STATE_HOME/opencode/service.json` (default `~/.local/state/opencode/service.json`) with its URL, PID, version +and password. `Opencode.from_service()` connects to it, starting it first if it is not running: + +```python +from opencode import Opencode + +with Opencode.from_service(version="2.1.2") as client: # (1)! + print(client.server.info().version) +``` + +1. Optional. A service running another version is stopped and replaced. `version` can also be a predicate, + such as `lambda v: v.startswith("2.")`. + +Pass `start=False` to only connect to a running service; it raises `ServiceError` when there is none. + +## Without a client + +```python +from opencode import service + +endpoint = service.discover() # an Endpoint, or None; never starts anything +endpoint = service.ensure() # starts the service when needed +service.headers(endpoint) # {"authorization": "Basic ..."} for other HTTP clients +service.stop(pty="handoff") # hand persistent terminals to the next service +``` + +Several processes may call `ensure()` at once: each starts at most two attempts, the service's own lock picks a +winner, and every caller gets the same service. It gives up after 120 seconds with a `ServiceError` that includes +the end of the failed process's stderr. + +## Async + +`await AsyncOpencode.from_service()`, `service.async_discover()`, `service.async_ensure()` and +`service.async_stop()` take the same arguments. + +These helpers follow the TypeScript client's `Service` from `@opencode-ai/client`. diff --git a/docs/reference/index.md b/docs/reference/index.md index 75627eb..83ca3b0 100644 --- a/docs/reference/index.md +++ b/docs/reference/index.md @@ -6,6 +6,7 @@ Generated from the docstrings in `src/opencode`. | --- | --- | | [Clients](clients.md) | `Opencode`, `AsyncOpencode` | | [Server launcher](server.md) | `OpencodeServer`, `AsyncOpencodeServer` | +| [Background service](service.md) | `opencode.service`, `Endpoint`, `ServiceError` | | [Resources](resources.md) | Every `client.` and its methods | | [Pagination and streams](pagination.md) | `SyncCursorPage`, `AsyncPaginator`, `Stream`, `AsyncStream` | | [Shared events and RPC](events.md) | `EventHub`, `Subscription`, `RpcDefinition`, `RpcClient` | diff --git a/docs/reference/service.md b/docs/reference/service.md new file mode 100644 index 0000000..f95d632 --- /dev/null +++ b/docs/reference/service.md @@ -0,0 +1,15 @@ +# Background service + +The `opencode.service` module, and the `from_service` constructors on +[`Opencode`](clients.md) and [`AsyncOpencode`](clients.md). + +::: opencode.service + options: + show_root_heading: false + members: [discover, ensure, stop, headers, default_file, async_discover, async_ensure, async_stop] + +::: opencode.Endpoint + +::: opencode.service.BasicAuth + +::: opencode.ServiceError diff --git a/mkdocs.yml b/mkdocs.yml index 1e47f8d..957293e 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -132,6 +132,7 @@ nav: - getting-started/index.md - Connect to a server: getting-started/connect.md - Run a server from Python: getting-started/server.md + - Use the background service: getting-started/service.md - Guides: - guides/index.md - Sessions and prompts: guides/sessions.md @@ -152,6 +153,7 @@ nav: - reference/index.md - Clients: reference/clients.md - Server launcher: reference/server.md + - Background service: reference/service.md - Resources: reference/resources.md - Pagination and streams: reference/pagination.md - Shared events and RPC: reference/events.md diff --git a/src/opencode/__init__.py b/src/opencode/__init__.py index c122b20..ef6ef34 100644 --- a/src/opencode/__init__.py +++ b/src/opencode/__init__.py @@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any +from . import service from ._client import AsyncOpencode, Opencode # @@ -78,6 +79,7 @@ from ._pagination import AsyncCursorPage, AsyncPaginator, SyncCursorPage from ._pty import AsyncPtyConnection, PtyConnection, PtyConnectionError from ._server import AsyncOpencodeServer, OpencodeServer, ServerStartError +from ._service import Endpoint, ServiceError from ._streaming import AsyncStream, Stream from ._types import NOT_GIVEN, NotGiven from ._version import __version__ @@ -136,6 +138,7 @@ "BaseModel", "ConflictError", "ContentTooLargeError", + "Endpoint", "InternalServerError", "NotFoundError", "NotGiven", @@ -148,11 +151,13 @@ "PtyConnectionError", "RateLimitError", "ServerStartError", + "ServiceError", "ServiceUnavailableError", "Stream", "SyncCursorPage", "UnprocessableEntityError", "__version__", + "service", "types", # Shared event stream and typed plugin RPC "SUBSCRIBER_CAPACITY", diff --git a/src/opencode/_client.py b/src/opencode/_client.py index 0e0d46c..a419dd7 100644 --- a/src/opencode/_client.py +++ b/src/opencode/_client.py @@ -9,12 +9,21 @@ from ._constants import DEFAULT_MAX_RETRIES, DEFAULT_TIMEOUT, DEFAULT_USERNAME from ._events import AsyncEventHub, EventHub from ._rpc import AsyncTypedRpcResource, TypedRpcResource +from ._service import ( + EnsureReason, + ServiceError, + Version, + async_discover, + async_ensure, + discover as discover_service, + ensure as ensure_service, +) from ._types import NOT_GIVEN, NotGiven, Timeout from .resources import AsyncResources, SyncResources if TYPE_CHECKING: import os - from collections.abc import Iterable, Mapping + from collections.abc import Callable, Iterable, Mapping, Sequence import httpx @@ -82,6 +91,67 @@ def __init__( transport=transport, ) + @classmethod + def from_service( + cls, + *, + start: bool = True, + file: str | os.PathLike[str] | None = None, + version: Version | None = None, + command: Sequence[str] | None = None, + env: Mapping[str, str] | None = None, + on_start: Callable[[EnsureReason, str | None], None] | None = None, + directory: str | os.PathLike[str] | None = None, + timeout: Timeout = DEFAULT_TIMEOUT, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + ) -> Self: + """Connect to the local background service, starting ``opencode serve --service`` if needed. + + Uses :func:`opencode.service.ensure`, or :func:`opencode.service.discover` when ``start`` is false, and + sends the service's password with every request. + + Args: + start: Start or replace the service when no matching one is running. When false, raise instead. + file: Registration file. Defaults to ``$XDG_STATE_HOME/opencode/service.json``. + version: Required exact version, or a predicate on the service version. + command: Service command and arguments, used when starting it. + env: Variables added to the environment of a started service. + on_start: Called once before a service process is started. + directory: Default project directory, as in the constructor. + timeout: Default request timeout, as in the constructor. + max_retries: Retries for failed requests, as in the constructor. + default_headers: Headers sent with every request. + default_query: Query parameters sent with every request. + + Raises: + ServiceError: No matching service is running and ``start`` is false, or it could not be started. + + Example:: + + with Opencode.from_service() as client: + print(client.server.info().version) + """ + if start: + endpoint = ensure_service(file, version, command=command, env=env, on_start=on_start) + else: + found = discover_service(file, version) + if found is None: + raise ServiceError("No matching OpenCode background service is running") + endpoint = found + return cls( + endpoint.url, + # An empty password sends no credentials instead of falling back to OPENCODE_PASSWORD. + password="" if endpoint.auth is None else endpoint.auth.password, + username=DEFAULT_USERNAME if endpoint.auth is None else endpoint.auth.username, + directory=directory, + timeout=timeout, + max_retries=max_retries, + default_headers=default_headers, + default_query=default_query, + ) + @cached_property def events(self) -> EventHub: """The shared event stream: one connection for every subscriber. See :class:`opencode.EventHub`.""" @@ -189,6 +259,42 @@ def __init__( transport=transport, ) + @classmethod + async def from_service( + cls, + *, + start: bool = True, + file: str | os.PathLike[str] | None = None, + version: Version | None = None, + command: Sequence[str] | None = None, + env: Mapping[str, str] | None = None, + on_start: Callable[[EnsureReason, str | None], None] | None = None, + directory: str | os.PathLike[str] | None = None, + timeout: Timeout = DEFAULT_TIMEOUT, + max_retries: int = DEFAULT_MAX_RETRIES, + default_headers: Mapping[str, str] | None = None, + default_query: Mapping[str, object] | None = None, + ) -> Self: + """Async version of :meth:`Opencode.from_service`. Await it: ``client = await AsyncOpencode.from_service()``.""" + if start: + endpoint = await async_ensure(file, version, command=command, env=env, on_start=on_start) + else: + found = await async_discover(file, version) + if found is None: + raise ServiceError("No matching OpenCode background service is running") + endpoint = found + return cls( + endpoint.url, + # An empty password sends no credentials instead of falling back to OPENCODE_PASSWORD. + password="" if endpoint.auth is None else endpoint.auth.password, + username=DEFAULT_USERNAME if endpoint.auth is None else endpoint.auth.username, + directory=directory, + timeout=timeout, + max_retries=max_retries, + default_headers=default_headers, + default_query=default_query, + ) + @cached_property def events(self) -> AsyncEventHub: """The shared event stream: one connection for every subscriber. See :class:`opencode.AsyncEventHub`.""" diff --git a/src/opencode/_service.py b/src/opencode/_service.py new file mode 100644 index 0000000..10a9c21 --- /dev/null +++ b/src/opencode/_service.py @@ -0,0 +1,1000 @@ +"""Find, start and stop the local OpenCode background service (``opencode serve --service``). + +A port of ``packages/client/src/promise/service.ts`` and its helpers from the OpenCode TypeScript client. The +registration file written by the service is the whole discovery contract: it names the URL, PID, version and +password of the one service that owns it. + +The API is a set of module functions (``discover``, ``ensure``, ``stop``, ``headers``) with ``async_`` twins, +because the operations share no state between calls: a class would only be a namespace, and the module already is +one. ``from opencode import service`` gives the same ``service.ensure()`` spelling as the TypeScript ``Service`` +object. +""" + +from __future__ import annotations + +import base64 +import contextlib +import json +import logging +import os +import shutil +import signal +import subprocess +import sys +import threading +import time +import uuid +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Literal, TypeAlias, cast + +import httpx +from anyio import sleep as _async_sleep + +from ._constants import DEFAULT_USERNAME +from ._exceptions import OpencodeError + +__all__ = [ + "BasicAuth", + "Endpoint", + "EnsureReason", + "PtyMode", + "ServiceError", + "Version", + "async_discover", + "async_ensure", + "async_stop", + "default_file", + "discover", + "ensure", + "headers", + "stop", +] + +log = logging.getLogger("opencode") +_sleep = time.sleep + +EnsureReason: TypeAlias = Literal["missing", "version-mismatch"] +"""Why :func:`ensure` starts a new service process.""" + +Version: TypeAlias = str | Callable[[str], bool] +"""A required exact service version, or a predicate that accepts compatible versions.""" + +PtyMode: TypeAlias = Literal["clear", "handoff"] +"""How :func:`stop` treats persistent terminals.""" + +_STDERR_LIMIT = 8 * 1024 +_HANDOFF_TTL_MS = 30_000 + + +class ServiceError(OpencodeError): + """The local service could not be found, started or stopped.""" + + +@dataclass(frozen=True) +class BasicAuth: + """HTTP basic credentials of a local service.""" + + username: str + """Basic auth user name, always ``opencode``.""" + password: str + """The service's private password.""" + + +@dataclass(frozen=True) +class Endpoint: + """Connection details of a local OpenCode service. + + Pass it to :meth:`opencode.Opencode.from_service`, or use ``url`` and :func:`headers` with any HTTP client. + """ + + url: str + """Base URL of the service.""" + auth: BasicAuth | None = None + """Credentials the service requires, or ``None`` when it runs without a password.""" + + +@dataclass(frozen=True) +class _Info: + url: str + pid: int + id: str | None = None + version: str | None = None + password: str | None = None + + +@dataclass(frozen=True) +class _LocalService: + info: _Info + endpoint: Endpoint + version: str | None + state: Literal["ready", "waiting", "failed"] + compatible: bool + + +@dataclass(frozen=True) +class _Probe: + service: _LocalService | None = None + timed_out: bool = False + + +@dataclass(frozen=True) +class _Registration: + info: _Info | None = None + service: _LocalService | None = None + timed_out: bool = False + + +@dataclass(frozen=True) +class _Timing: + # Seconds. A freshly spawned service registers in ~250 ms, so the poll cadence is a large share of the time a + # client waits for it. + poll_interval: float = 0.025 + request_timeout: float = 2.0 + spawn_delay: float = 5.0 + max_spawn_delay: float = 30.0 + timeout: float = 120.0 + stop_poll_interval: float = 0.05 + stop_poll_attempts: int = 100 + + +# Tests replace this to keep the lifecycle fast; it is not part of the public signature, as in TypeScript. +_TIMING = _Timing() + + +def default_file() -> Path: + """The default registration file, ``$XDG_STATE_HOME/opencode/service.json``. + + ``XDG_STATE_HOME`` defaults to ``~/.local/state`` on every platform, as in OpenCode itself. + """ + state = os.environ.get("XDG_STATE_HOME") + return Path(state if state is not None else Path.home() / ".local" / "state") / "opencode" / "service.json" + + +def headers(endpoint: Endpoint) -> dict[str, str] | None: + """The ``Authorization`` header for ``endpoint``, or ``None`` when it needs no credentials.""" + if endpoint.auth is None: + return None + return {"authorization": _basic(endpoint.auth.username, endpoint.auth.password)} + + +def _basic(username: str, password: str) -> str: + return "Basic " + base64.b64encode(f"{username}:{password}".encode()).decode() + + +def _file(file: str | os.PathLike[str] | None) -> Path: + return Path(file) if file is not None else default_file() + + +def _now_ms() -> float: + return time.time() * 1000 + + +# --- versions and probing (service-version.ts, service-probe.ts) --- + + +def _matches_version(version: str | None, required: Version | None) -> bool: + if required is None: + return True + if version is None: + return False + if callable(required): + return required(version) + return version == required + + +@dataclass(frozen=True) +class _Decision: + tag: Literal["reuse", "wait", "replace", "fail"] + pty: PtyMode = "clear" + error: str = "" + + +def _decide(service: _LocalService, required: Version | None) -> _Decision: + if not _matches_version(service.version, required): + return _Decision("replace", pty="handoff" if service.state == "ready" else "clear") + if not service.compatible: + return _Decision( + "fail", + error="Background service uses an incompatible health protocol. " + "Update this client or explicitly restart the service.", + ) + if service.state == "ready": + return _Decision("reuse") + if service.state == "failed": + return _Decision("fail", error="Background service failed to start") + return _Decision("wait") + + +def _same(left: _Info, right: _Info) -> bool: + return left.id == right.id and left.version == right.version and left.url == right.url and left.pid == right.pid + + +def _endpoint(info: _Info) -> Endpoint: + return Endpoint(info.url, None if info.password is None else BasicAuth(DEFAULT_USERNAME, info.password)) + + +def _decode_server_info(body: object) -> tuple[str, int] | None: + if not isinstance(body, dict): + return None + data = cast("dict[str, object]", body) + version, pid = data.get("version"), data.get("pid") + if not isinstance(version, str): + return None + if isinstance(pid, bool) or not isinstance(pid, (int, float)) or pid != int(pid) or pid < 0: + return None + return version, int(pid) + + +def _classify(info: _Info, status: int, body: object) -> _Probe: + endpoint = _endpoint(info) + # A missing health endpoint identifies protocol incompatibility, not an older version. Only an unmet version + # requirement lets ensure replace this owner. + if status == 404: + return _Probe(_LocalService(info, endpoint, info.version, "ready", compatible=False)) + decoded = _decode_server_info(body) + if decoded is None: + return _Probe() + version, pid = decoded + if pid != info.pid: + return _Probe() + if info.version is not None and version != info.version: + return _Probe() + state: Literal["ready", "waiting", "failed"] = ( + "ready" if 200 <= status < 300 else "failed" if status == 500 else "waiting" + ) + return _Probe(_LocalService(info, endpoint, version, state, compatible=True)) + + +def _info_url(info: _Info) -> httpx.URL: + return httpx.URL(info.url).join("/api/info") + + +def _probe(info: _Info, timeout: float) -> _Probe: + # The service is local: ignore proxy variables, as Node's fetch in the TypeScript client does. + try: + with httpx.Client(timeout=timeout, trust_env=False) as http: + response = http.get(_info_url(info), headers=headers(_endpoint(info))) + body: object = None if response.status_code == 404 else response.json() + except httpx.TimeoutException: + return _Probe(timed_out=True) + except (httpx.HTTPError, ValueError): + return _Probe() + return _classify(info, response.status_code, body) + + +async def _async_probe(info: _Info, timeout: float) -> _Probe: + try: + async with httpx.AsyncClient(timeout=timeout, trust_env=False) as http: + response = await http.get(_info_url(info), headers=headers(_endpoint(info))) + body: object = None if response.status_code == 404 else response.json() + except httpx.TimeoutException: + return _Probe(timed_out=True) + except (httpx.HTTPError, ValueError): + return _Probe() + return _classify(info, response.status_code, body) + + +# --- registration file --- + + +def _read_json(path: Path) -> object: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + +def _parse_info(value: object) -> _Info | None: + if not isinstance(value, dict): + return None + data = cast("dict[str, object]", value) + url, pid = data.get("url"), data.get("pid") + if not isinstance(url, str) or isinstance(pid, bool) or not isinstance(pid, int): + return None + + def text(key: str) -> str | None: + item = data.get(key) + return item if isinstance(item, str) else None + + return _Info(url=url, pid=pid, id=text("id"), version=text("version"), password=text("password")) + + +def _read(file: Path) -> _Info | None: + return _parse_info(_read_json(file)) + + +def _registered(file: Path, timeout: float) -> _Registration: + info = _read(file) + if info is None: + return _Registration() + probe = _probe(info, timeout) + return _Registration(info, probe.service, probe.timed_out) + + +async def _async_registered(file: Path, timeout: float) -> _Registration: + info = _read(file) + if info is None: + return _Registration() + probe = await _async_probe(info, timeout) + return _Registration(info, probe.service, probe.timed_out) + + +def _discovered(registration: _Registration, version: Version | None) -> Endpoint | None: + found = registration.service + if found is None or found.state != "ready" or not found.compatible: + return None + if not _matches_version(found.version, version): + return None + return found.endpoint + + +# --- persistent terminal handoff (pty-handoff.ts) --- + + +@dataclass(frozen=True) +class _Source: + pid: int + url: str + id: str | None = None + + +@dataclass(frozen=True) +class _Sidecar: + source: _Source + handoff: dict[str, Any] | None + expires_at: float + + +def _sidecar_path(file: Path) -> Path: + return file.with_name(file.name + ".pty-handoff") + + +def _is_handoff(value: object) -> bool: + if not isinstance(value, dict): + return False + data = cast("dict[str, object]", value) + expires = data.get("expiresAt") + return ( + isinstance(data.get("directory"), str) + and isinstance(data.get("instanceID"), str) + and isinstance(data.get("ticket"), str) + and _is_number(expires) + ) + + +def _is_number(value: object) -> bool: + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and value == value + and abs(value) != float("inf") + ) + + +def _read_sidecar(file: Path) -> _Sidecar | None: + value = _read_json(_sidecar_path(file)) + if not isinstance(value, dict): + return None + data = cast("dict[str, object]", value) + if "source" not in data or "handoff" not in data: + return None + source = data["source"] + if not isinstance(source, dict): + return None + src = cast("dict[str, object]", source) + pid, url, id_ = src.get("pid"), src.get("url"), src.get("id") + if not _is_number(pid) or not isinstance(url, str): + return None + if "id" in src and not isinstance(id_, str): + return None + handoff = data["handoff"] + if handoff is not None and not _is_handoff(handoff): + return None + expires = data.get("expiresAt") + if not _is_number(expires): + return None + return _Sidecar( + _Source(pid=cast("int", pid), url=url, id=id_ if isinstance(id_, str) else None), + cast("dict[str, Any] | None", handoff), + cast("float", expires), + ) + + +def _same_source(source: _Source, info: _Info) -> bool: + return source.id == info.id and source.pid == info.pid and source.url == info.url + + +def _fresh_for(record: _Sidecar | None, info: _Info) -> bool: + return record is not None and record.expires_at > _now_ms() and _same_source(record.source, info) + + +def _parse_handoff(body: object) -> dict[str, Any] | None: + if not isinstance(body, dict) or "handoff" not in body: + raise ValueError("Invalid persistent terminal handoff response") + handoff = cast("dict[str, object]", body)["handoff"] + if handoff is None: + return None + if not _is_handoff(handoff) or cast("dict[str, Any]", handoff)["expiresAt"] <= _now_ms(): + raise ValueError("Invalid or expired persistent terminal handoff") + return cast("dict[str, Any]", handoff) + + +def _publish(file: Path, info: _Info, handoff: dict[str, Any] | None) -> None: + temporary = file.with_name(f"{file.name}.pty-handoff.{uuid.uuid4()}.tmp") + record = { + "source": { + key: value for key, value in (("id", info.id), ("pid", info.pid), ("url", info.url)) if value is not None + }, + "handoff": handoff, + "expiresAt": handoff["expiresAt"] if handoff is not None else _now_ms() + _HANDOFF_TTL_MS, + } + try: + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + json.dump(record, stream, separators=(",", ":")) + os.replace(temporary, _sidecar_path(file)) + finally: + with contextlib.suppress(FileNotFoundError): + temporary.unlink() + + +def _pty_url(info: _Info, action: str) -> httpx.URL: + return httpx.URL(info.url).join(f"/api/experimental/persistent-pty/{action}") + + +def _pty_headers(info: _Info) -> dict[str, str] | None: + return None if info.password is None else {"authorization": _basic(DEFAULT_USERNAME, info.password)} + + +def _handoff_failed(file: Path, info: _Info, cause: Exception) -> bool: + """Whether the old service's terminals must be shut down, after a failed handoff request.""" + # Another caller may already have prepared and stopped this server. + if _fresh_for(_read_sidecar(file), info): + return False + log.warning( + "Background service cannot hand off persistent terminals; shutting them down before replacement: %s", cause + ) + return True + + +def _publish_logged(file: Path, info: _Info, handoff: dict[str, Any] | None) -> None: + try: + _publish(file, info, handoff) + except OSError as cause: + log.warning("Failed to publish persistent terminal handoff: %s", cause) + + +def _pty_prepare(file: Path, info: _Info, timeout: float) -> None: + # Publish the ticket before stopping its owner so every replacement contender can adopt it. + if _fresh_for(_read_sidecar(file), info): + return + with httpx.Client(timeout=timeout, trust_env=False, headers=_pty_headers(info)) as http: + try: + response = http.post(_pty_url(info, "handoff")) + response.raise_for_status() + handoff = _parse_handoff(response.json()) + except (httpx.HTTPError, ValueError) as cause: + if not _handoff_failed(file, info, cause): + return + with contextlib.suppress(httpx.HTTPError): + http.post(_pty_url(info, "shutdown")) + handoff = None + _publish_logged(file, info, handoff) + + +async def _async_pty_prepare(file: Path, info: _Info, timeout: float) -> None: + if _fresh_for(_read_sidecar(file), info): + return + async with httpx.AsyncClient(timeout=timeout, trust_env=False, headers=_pty_headers(info)) as http: + try: + response = await http.post(_pty_url(info, "handoff")) + response.raise_for_status() + handoff = _parse_handoff(response.json()) + except (httpx.HTTPError, ValueError) as cause: + if not _handoff_failed(file, info, cause): + return + with contextlib.suppress(httpx.HTTPError): + await http.post(_pty_url(info, "shutdown")) + handoff = None + _publish_logged(file, info, handoff) + + +def _pty_environment(file: Path, env: Mapping[str, str] | None) -> dict[str, str | None]: + record = _read_sidecar(file) + current = _read_json(file) + usable = record is not None and record.expires_at > _now_ms() + if usable and record is not None and isinstance(current, dict): + data = cast("dict[str, object]", current) + usable = record.source.id == data.get("id") and record.source.pid == data.get("pid") + usable = usable and record.source.url == data.get("url") + handoff = record.handoff if usable and record is not None else None + return { + **(env or {}), + "OPENCODE_PTY_HANDOFF": None if handoff is None else json.dumps(handoff, separators=(",", ":")), + } + + +def _pty_complete(file: Path, info: _Info) -> None: + record = _read_sidecar(file) + if record is not None and not _same_source(record.source, info): + _pty_clear(file) + + +def _pty_clear(file: Path) -> None: + with contextlib.suppress(FileNotFoundError): + _sidecar_path(file).unlink() + + +# --- startup attempts (service-contender.ts) --- + +# Detached service processes this interpreter started. Keeping them referenced until they are reaped stops +# ``subprocess`` from warning about a still-running child when the ``Popen`` object is collected. +_children: set[subprocess.Popen[bytes]] = set() + + +class _Contender: + """One ``opencode serve --service`` attempt. Losers of the startup race exit with code 0.""" + + def __init__(self, command: Sequence[str], env: Mapping[str, str | None]) -> None: + merged = {**os.environ, **env} + environment = {key: value for key, value in merged.items() if value is not None} + executable = shutil.which(command[0]) or command[0] + if sys.platform == "win32": # pragma: no cover - Windows only + self.process = subprocess.Popen( + [executable, *command[1:]], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + env=environment, + creationflags=subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW, + ) + else: + self.process = subprocess.Popen( + [executable, *command[1:]], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + env=environment, + start_new_session=True, + ) + _children.add(self.process) + self._stderr: bytes = b"" + self._capture = True + self._closed = threading.Event() + threading.Thread(target=self._drain, name="opencode-service-stderr", daemon=True).start() + + def _drain(self) -> None: + stream = self.process.stderr + assert stream is not None + with stream: + for chunk in iter(lambda: os.read(stream.fileno(), 4096), b""): + if self._capture: + self._stderr = (self._stderr + chunk)[-_STDERR_LIMIT:] + self.process.wait() + _children.discard(self.process) + self._closed.set() + + @property + def pid(self) -> int: + return self.process.pid + + def closed(self) -> bool: + return self._closed.is_set() + + def stderr(self) -> str: + return self._stderr.decode("utf-8", "replace").strip() + + def release(self) -> None: + # Keep draining so a chatty service never blocks on a full pipe, but stop keeping its output. + self._capture = False + self._stderr = b"" + + def failure(self) -> ServiceError | None: + code = self.process.returncode + if code is None or code == 0: + return None + if code > 0 or sys.platform == "win32": + return _startup_error(f"Server process exited with code {code}", self.stderr()) + try: + name = signal.Signals(-code).name + except ValueError: + name = str(-code) + return _startup_error(f"Server process terminated by {name}", self.stderr()) + + +def _startup_error(message: str, stderr: str) -> ServiceError: + return ServiceError(f"{message}\n{stderr}" if stderr else message) + + +@dataclass +class _Pool: + """The startup attempts of one ``ensure`` call: at most two alive, the first failure remembered.""" + + timing: _Timing + contenders: list[_Contender] = field(default_factory=cast("Callable[[], list[_Contender]]", list)) + failure: ServiceError | None = None + spawn_delay: float = 0.0 + last_spawn: float | None = None + + def __post_init__(self) -> None: + self.spawn_delay = self.timing.spawn_delay + + def service_answered(self) -> None: + self.spawn_delay = self.timing.spawn_delay + + def evict(self, pid: int) -> None: + for item in list(self.contenders): + if item.pid == pid or item.closed(): + item.release() + self.contenders.remove(item) + self.failure = None + self.last_spawn = None + + def recruit_now(self) -> None: + self.last_spawn = time.monotonic() - self.spawn_delay + + def reap(self) -> ServiceError | None: + finished = [item for item in self.contenders if item.closed()] + if self.failure is None: + self.failure = next((error for item in finished if (error := item.failure()) is not None), None) + if any(item.process.returncode == 0 for item in finished): + self.spawn_delay = min(self.spawn_delay * 2, self.timing.max_spawn_delay) + for item in finished: + self.contenders.remove(item) + return self.failure if not self.contenders else None + + def should_recruit(self, registered: bool) -> bool: + if self.last_spawn is None and registered: + self.last_spawn = time.monotonic() + # Keep one candidate plus one lock probe for pre-lock stalls. After a failure, let the survivors finish + # without recruiting replacements that could hide the error indefinitely. + since = float("inf") if self.last_spawn is None else time.monotonic() - self.last_spawn + return self.failure is None and len(self.contenders) < 2 and since >= self.spawn_delay + + def add(self, contender: _Contender) -> None: + self.contenders.append(contender) + self.last_spawn = time.monotonic() + + def release_all(self) -> None: + for contender in self.contenders: + contender.release() + + +# --- stopping processes --- + + +def _signal(pid: int, sig: int) -> None: + with contextlib.suppress(OSError): + os.kill(pid, sig) + + +def _stopped(pid: int) -> bool: + if sys.platform == "win32": + return not _windows_alive(pid) + try: + os.kill(pid, 0) + except OSError: + return True + return False + + +def _windows_alive(pid: int) -> bool: # pragma: no cover - Windows only + import ctypes + + kernel32: Any = getattr(ctypes, "windll").kernel32 # noqa: B009 - only defined on Windows + handle = kernel32.OpenProcess(0x1000, False, pid) # PROCESS_QUERY_LIMITED_INFORMATION + if not handle: + return False + try: + code = ctypes.c_ulong() + kernel32.GetExitCodeProcess(handle, ctypes.byref(code)) + return bool(code.value == 259) # STILL_ACTIVE + finally: + kernel32.CloseHandle(handle) + + +_KILL = getattr(signal, "SIGKILL", signal.SIGTERM) + + +def _terminate_steps(file: Path, info: _Info) -> bool: + current = _read(file) + if current is None or not _same(current, info): + return False + _signal(info.pid, signal.SIGTERM) + return True + + +def _remove_if_same(file: Path, info: _Info) -> None: + latest = _read(file) + if latest is None or not _same(latest, info): + return + with contextlib.suppress(FileNotFoundError): + file.unlink() + + +def _wait_until_stopped(pid: int, timing: _Timing) -> bool: + for attempt in range(timing.stop_poll_attempts + 1): + if _stopped(pid): + return True + if attempt < timing.stop_poll_attempts: + _sleep(timing.stop_poll_interval) + return False + + +async def _async_wait_until_stopped(pid: int, timing: _Timing) -> bool: + for attempt in range(timing.stop_poll_attempts + 1): + if _stopped(pid): + return True + if attempt < timing.stop_poll_attempts: + await _async_sleep(timing.stop_poll_interval) + return False + + +def _terminate(file: Path, info: _Info, timing: _Timing) -> None: + if not _terminate_steps(file, info): + return + # The registration can disappear or change hands before this process exits. Only the PID we signalled can + # tell us whether it has stopped, so escalate based on that process. + if not _wait_until_stopped(info.pid, timing): + _signal(info.pid, _KILL) + if not _wait_until_stopped(info.pid, timing): + raise ServiceError(f"Server process {info.pid} is still running") + _remove_if_same(file, info) + + +async def _async_terminate(file: Path, info: _Info, timing: _Timing) -> None: + if not _terminate_steps(file, info): + return + if not await _async_wait_until_stopped(info.pid, timing): + _signal(info.pid, _KILL) + if not await _async_wait_until_stopped(info.pid, timing): + raise ServiceError(f"Server process {info.pid} is still running") + _remove_if_same(file, info) + + +# --- public operations --- + + +def discover(file: str | os.PathLike[str] | None = None, version: Version | None = None) -> Endpoint | None: + """Find a healthy, compatible local service without starting one. + + Args: + file: Registration file. Defaults to :func:`default_file`. + version: Required exact version, or a predicate on the service version. + + Returns: + The service endpoint, or ``None`` when no ready service with a matching version is registered. + """ + return _discovered(_registered(_file(file), _TIMING.request_timeout), version) + + +async def async_discover(file: str | os.PathLike[str] | None = None, version: Version | None = None) -> Endpoint | None: + """Async version of :func:`discover`.""" + return _discovered(await _async_registered(_file(file), _TIMING.request_timeout), version) + + +@dataclass +class _Ensure: + """State and decisions shared by the sync and async ``ensure`` loops.""" + + file: Path + version: Version | None + command: Sequence[str] + env: Mapping[str, str] | None + on_start: Callable[[EnsureReason, str | None], None] | None + timing: _Timing + pool: _Pool + deadline: float + timeouts: tuple[_Info, int] | None = None + announced: bool = False + + def announce(self, reason: EnsureReason, previous_version: str | None = None) -> None: + if self.announced: + return + self.announced = True + if self.on_start is not None: + self.on_start(reason, previous_version) + + def check_deadline(self) -> None: + if time.monotonic() >= self.deadline: + raise self.pool.failure or ServiceError("Timed out waiting for the background service to start") + + def unresponsive(self, registration: _Registration) -> _Info | None: + """The registered owner to replace after three consecutive timed-out probes.""" + info = registration.info + if not registration.timed_out or info is None: + self.timeouts = None + return None + count = self.timeouts[1] + 1 if self.timeouts is not None and _same(self.timeouts[0], info) else 1 + self.timeouts = (info, count) + if count < 3: + return None + self.announce("missing") + log.warning("Background service is unresponsive; recovery cannot preserve persistent terminals") + _pty_clear(self.file) + return info + + def recovered(self, info: _Info) -> None: + self.pool.evict(info.pid) + self.pool.recruit_now() + self.timeouts = None + + def spawn(self) -> None: + if not self.command: + raise ServiceError("Missing service command") + try: + contender = _Contender(self.command, _pty_environment(self.file, self.env)) + except OSError as cause: + raise ServiceError("Failed to start server") from cause + self.pool.add(contender) + + def no_answer(self, registration: _Registration) -> None: + failed = self.pool.reap() + if failed is not None: + raise failed + if self.pool.should_recruit(registration.info is not None): + self.announce("missing") + self.spawn() + + +def _ensure_state( + file: str | os.PathLike[str] | None, + version: Version | None, + command: Sequence[str] | None, + env: Mapping[str, str] | None, + on_start: Callable[[EnsureReason, str | None], None] | None, +) -> _Ensure: + timing = _TIMING + return _Ensure( + file=_file(file), + version=version, + command=command if command is not None else ("opencode", "serve", "--service"), + env=env, + on_start=on_start, + timing=timing, + pool=_Pool(timing), + deadline=time.monotonic() + timing.timeout, + ) + + +def ensure( + file: str | os.PathLike[str] | None = None, + version: Version | None = None, + *, + command: Sequence[str] | None = None, + env: Mapping[str, str] | None = None, + on_start: Callable[[EnsureReason, str | None], None] | None = None, +) -> Endpoint: + """Return a healthy local service, starting ``opencode serve --service`` when none is running. + + A registered service whose version does not satisfy ``version`` is stopped and replaced; its persistent + terminals are handed off to the replacement when it is ready. Several processes may call this at once: each + starts at most two attempts, the service's own lock picks one winner, and the losers exit cleanly. Gives up + after 120 seconds. + + Args: + file: Registration file. Defaults to :func:`default_file`. + version: Required exact version, or a predicate on the service version. + command: Service command and arguments. Defaults to ``opencode serve --service``. + env: Variables added to the inherited environment of a started service. + on_start: Called once, before the first process is started, with the reason and, for + ``"version-mismatch"``, the version being replaced. + + Raises: + ServiceError: The service failed to start, uses an incompatible protocol, or did not answer in time. + """ + state = _ensure_state(file, version, command, env, on_start) + try: + while True: + state.check_deadline() + registration = _registered(state.file, state.timing.request_timeout) + stale = state.unresponsive(registration) + if stale is not None: + _terminate(state.file, stale, state.timing) + state.recovered(stale) + + service = registration.service + if service is not None: + state.pool.service_answered() + decision = _decide(service, state.version) + if decision.tag == "fail": + raise ServiceError(decision.error) + if decision.tag == "reuse": + _pty_complete(state.file, service.info) + return service.endpoint + if decision.tag == "replace": + state.announce("version-mismatch", service.version) + if service.state != "ready": + log.warning("Background service is not ready; replacement cannot preserve persistent terminals") + with contextlib.suppress(Exception): + stop(state.file, pty=decision.pty) + state.pool.evict(service.info.pid) + else: + state.no_answer(registration) + _sleep(state.timing.poll_interval) + finally: + state.pool.release_all() + + +async def async_ensure( + file: str | os.PathLike[str] | None = None, + version: Version | None = None, + *, + command: Sequence[str] | None = None, + env: Mapping[str, str] | None = None, + on_start: Callable[[EnsureReason, str | None], None] | None = None, +) -> Endpoint: + """Async version of :func:`ensure`.""" + state = _ensure_state(file, version, command, env, on_start) + try: + while True: + state.check_deadline() + registration = await _async_registered(state.file, state.timing.request_timeout) + stale = state.unresponsive(registration) + if stale is not None: + await _async_terminate(state.file, stale, state.timing) + state.recovered(stale) + + service = registration.service + if service is not None: + state.pool.service_answered() + decision = _decide(service, state.version) + if decision.tag == "fail": + raise ServiceError(decision.error) + if decision.tag == "reuse": + _pty_complete(state.file, service.info) + return service.endpoint + if decision.tag == "replace": + state.announce("version-mismatch", service.version) + if service.state != "ready": + log.warning("Background service is not ready; replacement cannot preserve persistent terminals") + with contextlib.suppress(Exception): + await async_stop(state.file, pty=decision.pty) + state.pool.evict(service.info.pid) + else: + state.no_answer(registration) + await _async_sleep(state.timing.poll_interval) + finally: + state.pool.release_all() + + +def stop(file: str | os.PathLike[str] | None = None, *, pty: PtyMode = "clear") -> None: + """Stop the registered local service and remove its registration. + + Args: + file: Registration file. Defaults to :func:`default_file`. + pty: ``"clear"`` shuts persistent terminals down with the service. ``"handoff"`` first asks the service + for a handoff ticket that the next service started by :func:`ensure` adopts. The handoff is best + effort and never keeps the old service running. + + Raises: + ServiceError: The process was still running after ``SIGTERM`` and ``SIGKILL``. + """ + path = _file(file) + info = _read(path) + try: + if pty == "handoff" and info is not None: + _pty_prepare(path, info, _TIMING.request_timeout) + else: + _pty_clear(path) + except Exception as cause: + log.warning("Failed to prepare persistent terminals for replacement: %s", cause) + if info is not None: + _terminate(path, info, _TIMING) + + +async def async_stop(file: str | os.PathLike[str] | None = None, *, pty: PtyMode = "clear") -> None: + """Async version of :func:`stop`.""" + path = _file(file) + info = _read(path) + try: + if pty == "handoff" and info is not None: + await _async_pty_prepare(path, info, _TIMING.request_timeout) + else: + _pty_clear(path) + except Exception as cause: + log.warning("Failed to prepare persistent terminals for replacement: %s", cause) + if info is not None: + await _async_terminate(path, info, _TIMING) diff --git a/src/opencode/service.py b/src/opencode/service.py new file mode 100644 index 0000000..3c6c109 --- /dev/null +++ b/src/opencode/service.py @@ -0,0 +1,50 @@ +"""Find, start and stop the local OpenCode background service. + +Example:: + + from opencode import Opencode, service + + endpoint = service.ensure() # or service.discover(), which never starts one + print(endpoint.url) + + with Opencode.from_service() as client: + print(client.server.info().version) + +See :mod:`opencode._service` for how this follows the TypeScript client. +""" + +from __future__ import annotations + +from ._service import ( + BasicAuth, + Endpoint, + EnsureReason, + PtyMode, + ServiceError, + Version, + async_discover, + async_ensure, + async_stop, + default_file, + discover, + ensure, + headers, + stop, +) + +__all__ = [ + "BasicAuth", + "Endpoint", + "EnsureReason", + "PtyMode", + "ServiceError", + "Version", + "async_discover", + "async_ensure", + "async_stop", + "default_file", + "discover", + "ensure", + "headers", + "stop", +] diff --git a/tests/fake_service.py b/tests/fake_service.py new file mode 100644 index 0000000..b667e94 --- /dev/null +++ b/tests/fake_service.py @@ -0,0 +1,77 @@ +"""A stand-in for `opencode serve --service`, used by test_service.py. + +Usage: python fake_service.py REGISTRATION_FILE VERSION [MODE] + +MODE is ``serve`` (default), ``fail`` (exit 1 with an error on stderr) or ``lose`` (exit 0 without registering, like +a contender that lost the startup race). A serving instance writes the registration file, answers ``GET /api/info`` +and the persistent-terminal endpoints, and exits on SIGTERM. It runs without a password: auth is covered by the +mocked probes in test_service.py and by the live test. +""" + +from __future__ import annotations + +import json +import os +import socketserver +import sys +import uuid +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from typing_extensions import override + + +def main() -> None: + file, version = Path(sys.argv[1]), sys.argv[2] + mode = sys.argv[3] if len(sys.argv) > 3 else "serve" + if mode == "fail": + sys.stderr.write("fake service: boom\n") + sys.exit(1) + if mode == "lose": + sys.exit(0) + handoff_env = os.environ.get("OPENCODE_PTY_HANDOFF") + + class Handler(BaseHTTPRequestHandler): + def _send(self, status: int, body: object) -> None: + data = json.dumps(body).encode() + self.send_response(status) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self) -> None: + if self.path == "/api/info": + self._send(200, {"version": version, "pid": os.getpid(), "handoff": handoff_env}) + else: + self._send(404, {}) + + def do_POST(self) -> None: + if self.path == "/api/experimental/persistent-pty/handoff": + self._send(200, {"handoff": None}) + else: + self._send(204 if self.path.endswith("/shutdown") else 404, {}) + + @override + def log_message(self, format: str, *args: object) -> None: + pass + + class Server(ThreadingHTTPServer): + @override + def server_bind(self) -> None: + # HTTPServer.server_bind resolves the host name with getfqdn, which can stall for many seconds on CI. + socketserver.TCPServer.server_bind(self) + self.server_name, self.server_port = "127.0.0.1", self.server_address[1] + + server = Server(("127.0.0.1", 0), Handler) + url = f"http://127.0.0.1:{server.server_address[1]}" + file.parent.mkdir(parents=True, exist_ok=True) + record = {"id": str(uuid.uuid4()), "version": version, "url": url, "pid": os.getpid()} + temporary = file.with_name(file.name + f".{os.getpid()}.tmp") + temporary.write_text(json.dumps(record)) + os.replace(temporary, file) + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..ec24b5a --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,273 @@ +from __future__ import annotations + +import json +import sys +import time +from collections.abc import Iterator +from dataclasses import replace +from pathlib import Path + +import httpx +import pytest +import respx + +from opencode import AsyncOpencode, Endpoint, Opencode, ServiceError, _service as impl, service + +FAKE = str(Path(__file__).with_name("fake_service.py")) +URL = "http://127.0.0.1:4999" + + +@pytest.fixture +def file(tmp_path: Path) -> Path: + return tmp_path / "state" / "opencode" / "service.json" + + +@pytest.fixture(autouse=True) +def _fast_timing(monkeypatch: pytest.MonkeyPatch) -> None: + timing = replace(impl._TIMING, spawn_delay=0.5, max_spawn_delay=1.0, timeout=20.0, stop_poll_interval=0.02) + monkeypatch.setattr(impl, "_TIMING", timing) + + +def register(file: Path, **fields: object) -> None: + file.parent.mkdir(parents=True, exist_ok=True) + file.write_text(json.dumps({"url": URL, "pid": 1234, "version": "2.1.2", "password": "pw", **fields})) + + +def fake(file: Path, version: str, mode: str = "serve") -> list[str]: + return [sys.executable, FAKE, str(file), version, mode] + + +@pytest.fixture +def cleanup(file: Path) -> Iterator[None]: + yield + service.stop(file) + + +def test_headers_and_default_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + assert service.headers(Endpoint(URL)) is None + assert service.headers(Endpoint(URL, service.BasicAuth("opencode", "pw"))) == { + "authorization": "Basic b3BlbmNvZGU6cHc=" + } + monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path)) + assert service.default_file() == tmp_path / "opencode" / "service.json" + monkeypatch.delenv("XDG_STATE_HOME") + assert service.default_file() == Path.home() / ".local" / "state" / "opencode" / "service.json" + + +def test_discover_without_registration(file: Path) -> None: + assert service.discover(file) is None + file.parent.mkdir(parents=True) + file.write_text("{not json") + assert service.discover(file) is None + + +@pytest.mark.parametrize( + ("status", "body", "version", "found"), + [ + (200, {"version": "2.1.2", "pid": 1234}, None, True), + (200, {"version": "2.1.2", "pid": 1234}, "2.1.2", True), + (200, {"version": "2.1.2", "pid": 1234}, "2.0.0", False), + (200, {"version": "2.1.2", "pid": 1234}, lambda v: v.startswith("2."), True), + (200, {"version": "2.1.2", "pid": 99}, None, False), # another process answered on that port + (200, {"version": "2.0.0", "pid": 1234}, None, False), # registration and server disagree + (200, {"pid": 1234}, None, False), + (503, {"version": "2.1.2", "pid": 1234}, None, False), # still starting + (404, None, None, False), # incompatible health protocol + ], +) +def test_discover_classifies_health( + file: Path, status: int, body: object, version: service.Version | None, found: bool +) -> None: + register(file) + with respx.mock() as router: + route = router.get(f"{URL}/api/info").respond(status, json=body) + endpoint = service.discover(file, version) + assert route.calls.last.request.headers["authorization"] == "Basic b3BlbmNvZGU6cHc=" + assert endpoint == (Endpoint(URL, service.BasicAuth("opencode", "pw")) if found else None) + + +def test_discover_without_password(file: Path) -> None: + register(file, password=None) + with respx.mock() as router: + route = router.get(f"{URL}/api/info").respond(200, json={"version": "2.1.2", "pid": 1234}) + assert service.discover(file) == Endpoint(URL) + assert "authorization" not in route.calls.last.request.headers + + +def test_ensure_reuses_and_clears_foreign_handoff(file: Path) -> None: + register(file, id="new") + sidecar = file.with_name("service.json.pty-handoff") + sidecar.write_text(json.dumps({"source": {"id": "old", "pid": 1, "url": URL}, "handoff": None, "expiresAt": 1e15})) + started: list[str] = [] + with respx.mock() as router: + router.get(f"{URL}/api/info").respond(200, json={"version": "2.1.2", "pid": 1234}) + endpoint = service.ensure(file, "2.1.2", on_start=lambda reason, _: started.append(reason)) + assert endpoint.url == URL + assert started == [] + assert not sidecar.exists() + + +@pytest.mark.parametrize( + ("status", "body", "message"), + [(404, None, "incompatible health protocol"), (500, {"version": "2.1.2", "pid": 1234}, "failed to start")], +) +def test_ensure_fails_on_unusable_service(file: Path, status: int, body: object, message: str) -> None: + register(file) + with respx.mock() as router: + router.get(f"{URL}/api/info").respond(status, json=body) + with pytest.raises(ServiceError, match=message): + service.ensure(file, command=["false-command-never-run"]) + + +def test_ensure_starts_missing_service_and_stop_removes_it(file: Path) -> None: + started: list[tuple[str, str | None]] = [] + endpoint = service.ensure(file, command=fake(file, "1.0.0"), on_start=lambda *args: started.append(args)) + assert started == [("missing", None)] + assert endpoint.auth is None # the fake runs without a password + assert service.discover(file, "1.0.0") == endpoint + # A second call reuses the running service. + assert service.ensure(file, command=fake(file, "1.0.0")) == endpoint + pid = json.loads(file.read_text())["pid"] + + service.stop(file) + assert not file.exists() + assert impl._stopped(pid) + assert service.discover(file) is None + service.stop(file) # nothing registered: a no-op + + +@pytest.mark.usefixtures("cleanup") +def test_ensure_replaces_version_mismatch_with_handoff(file: Path) -> None: + first = service.ensure(file, command=fake(file, "1.0.0")) + started: list[tuple[str, str | None]] = [] + second = service.ensure(file, "2.0.0", command=fake(file, "2.0.0"), on_start=lambda *args: started.append(args)) + assert started == [("version-mismatch", "1.0.0")] + assert second.url != first.url + # The handoff the old service published was adopted, then cleared once the new owner was ready. + assert not file.with_name("service.json.pty-handoff").exists() + + +def test_ensure_reports_startup_failure(file: Path) -> None: + with pytest.raises(ServiceError, match=r"exited with code 1\nfake service: boom"): + service.ensure(file, command=fake(file, "1.0.0", "fail")) + + +def test_ensure_times_out_when_attempts_keep_losing(file: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(impl, "_TIMING", replace(impl._TIMING, timeout=1.0, spawn_delay=0.1)) + with pytest.raises(ServiceError, match="Timed out"): + service.ensure(file, command=fake(file, "1.0.0", "lose")) + + +def test_ensure_rejects_bad_commands(file: Path) -> None: + with pytest.raises(ServiceError, match="Missing service command"): + service.ensure(file, command=[]) + with pytest.raises(ServiceError, match="Failed to start server"): + service.ensure(file, command=[str(file.parent / "does-not-exist")]) + + +def test_unresponsive_service_is_replaced(file: Path, monkeypatch: pytest.MonkeyPatch) -> None: + register(file, pid=2**22 + 12345) # a PID that is not running + started: list[str] = [] + calls = 0 + + def probe(info: impl._Info, timeout: float) -> impl._Probe: + nonlocal calls + calls += 1 + return impl._Probe(timed_out=True) + + monkeypatch.setattr(impl, "_probe", probe) + monkeypatch.setattr(impl, "_TIMING", replace(impl._TIMING, timeout=1.0)) + with pytest.raises(ServiceError, match="Timed out"): + service.ensure(file, command=fake(file, "1.0.0", "lose"), on_start=lambda reason, _: started.append(reason)) + assert started == ["missing"] + assert calls >= 3 + assert not file.exists() # the unresponsive registration was removed + + +def test_pty_environment_passes_fresh_matching_handoff(file: Path) -> None: + register(file, id="a") + handoff = {"directory": "/x", "instanceID": "i", "ticket": "t", "expiresAt": time.time() * 1000 + 60_000} + sidecar = file.with_name("service.json.pty-handoff") + sidecar.write_text( + json.dumps({"source": {"id": "a", "pid": 1234, "url": URL}, "handoff": handoff, "expiresAt": 1e15}) + ) + env = impl._pty_environment(file, {"EXTRA": "1"}) + assert env["EXTRA"] == "1" + assert json.loads(env["OPENCODE_PTY_HANDOFF"] or "") == handoff + register(file, id="b") # a different owner registered since + assert impl._pty_environment(file, None)["OPENCODE_PTY_HANDOFF"] is None + + +def test_stop_handoff_publishes_ticket(file: Path, monkeypatch: pytest.MonkeyPatch) -> None: + register(file, id="a", pid=2**22 + 12345) + handoff = {"directory": "/x", "instanceID": "i", "ticket": "t", "expiresAt": time.time() * 1000 + 60_000} + with respx.mock() as router: + router.post(f"{URL}/api/experimental/persistent-pty/handoff").respond(200, json={"handoff": handoff}) + service.stop(file, pty="handoff") + record = json.loads(file.with_name("service.json.pty-handoff").read_text()) + assert record == { + "source": {"id": "a", "pid": 2**22 + 12345, "url": URL}, + "handoff": handoff, + "expiresAt": handoff["expiresAt"], + } + assert not file.exists() + + +def test_stop_handoff_failure_shuts_terminals_down(file: Path) -> None: + register(file, pid=2**22 + 12345) + with respx.mock() as router: + router.post(f"{URL}/api/experimental/persistent-pty/handoff").respond(503) + shutdown = router.post(f"{URL}/api/experimental/persistent-pty/shutdown").respond(204) + service.stop(file, pty="handoff") + assert shutdown.called + record = json.loads(file.with_name("service.json.pty-handoff").read_text()) + assert record["handoff"] is None + + +def test_from_service(file: Path) -> None: + with pytest.raises(ServiceError, match="No matching"): + Opencode.from_service(start=False, file=file) + register(file) + with respx.mock() as router: + router.get(f"{URL}/api/info").respond(200, json={"version": "2.1.2", "pid": 1234}) + with Opencode.from_service(file=file) as client: + assert str(client.base_url).rstrip("/") == URL + assert client.password == "pw" + register(file, password=None) + with respx.mock() as router: + router.get(f"{URL}/api/info").respond(200, json={"version": "2.1.2", "pid": 1234}) + with Opencode.from_service(start=False, file=file) as client: + assert not client.password + + +@pytest.mark.anyio +async def test_async_lifecycle(file: Path) -> None: + assert await service.async_discover(file) is None + endpoint = await service.async_ensure(file, command=fake(file, "1.0.0")) + assert await service.async_discover(file) == endpoint + async with await AsyncOpencode.from_service(start=False, file=file) as client: + assert not client.password + async with httpx.AsyncClient(trust_env=False) as http: # CI runners may set proxy variables + response = await http.get(f"{endpoint.url}/api/info", headers=service.headers(endpoint)) + assert response.json()["version"] == "1.0.0" + second = await service.async_ensure(file, "2.0.0", command=fake(file, "2.0.0")) + assert second != endpoint + await service.async_stop(file, pty="handoff") + assert not file.exists() + + +@pytest.mark.anyio +async def test_async_from_service_raises_when_missing(file: Path) -> None: + with pytest.raises(ServiceError): + await AsyncOpencode.from_service(start=False, file=file) + with pytest.raises(ServiceError, match="exited with code 1"): + await AsyncOpencode.from_service(file=file, command=fake(file, "1.0.0", "fail")) + + +def test_probe_timeout_is_reported(file: Path) -> None: + register(file) + with respx.mock() as router: + router.get(f"{URL}/api/info").mock(side_effect=httpx.ConnectTimeout("slow")) + registration = impl._registered(file, 0.1) + assert registration.timed_out + assert registration.service is None diff --git a/tests/test_service_live.py b/tests/test_service_live.py new file mode 100644 index 0000000..1f5dfb2 --- /dev/null +++ b/tests/test_service_live.py @@ -0,0 +1,67 @@ +"""The service lifecycle against a real `opencode serve --service`, in temporary XDG directories. + +Run with `uv run pytest -m live`. Skipped when the opencode CLI is not installed. +""" + +from __future__ import annotations + +import json +import shutil +import socket +from pathlib import Path + +import pytest + +from opencode import API_VERSION, AsyncOpencode, Opencode, service + +pytestmark = [ + pytest.mark.live, + pytest.mark.skipif(shutil.which("opencode") is None, reason="opencode CLI is not installed"), +] + + +def _sandbox(root: Path) -> tuple[Path, dict[str, str]]: + # The service must never see, reuse or stop the developer's real background service. + env = { + name: str(root / name.lower()) + for name in ("XDG_DATA_HOME", "XDG_CONFIG_HOME", "XDG_STATE_HOME", "XDG_CACHE_HOME") + } + env["OPENCODE_CONFIG_PROJECT_DISABLE"] = "1" + env["OPENCODE_DISABLE_CLAUDE_CODE"] = "1" + # The service binds a fixed managed port, which a real service on this machine may hold. Give it a free one. + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + config = Path(env["XDG_CONFIG_HOME"]) / "opencode" / "service.json" + config.parent.mkdir(parents=True) + config.write_text(json.dumps({"port": port})) + return Path(env["XDG_STATE_HOME"]) / "opencode" / "service.json", env + + +def test_service_lifecycle(tmp_path: Path) -> None: + file, env = _sandbox(tmp_path) + assert service.discover(file) is None + reasons: list[str] = [] + try: + endpoint = service.ensure(file, env=env, on_start=lambda reason, _: reasons.append(reason)) + assert reasons == ["missing"] + assert file.exists() + assert service.discover(file, API_VERSION) == endpoint + assert service.ensure(file, env=env) == endpoint + with Opencode.from_service(start=False, file=file) as client: + assert client.server.info().version == API_VERSION + finally: + service.stop(file, pty="handoff") + assert not file.exists() + assert service.discover(file) is None + + +@pytest.mark.anyio +async def test_async_service_lifecycle(tmp_path: Path) -> None: + file, env = _sandbox(tmp_path) + try: + async with await AsyncOpencode.from_service(file=file, env=env) as client: + assert (await client.server.info()).version == API_VERSION + finally: + await service.async_stop(file) + assert not file.exists() From 744cc3ff83908edfecfe24b89d3fa7a52dfeaa81 Mon Sep 17 00:00:00 2001 From: shrijayan <81805145+shrijayan@users.noreply.github.com> Date: Thu, 22 Oct 2026 16:00:08 +0530 Subject: [PATCH 2/3] Keep service passwords out of repr and search only PATH for the command BasicAuth and the registration record hide their password from repr, so endpoints can be logged safely. The service command is resolved from PATH entries only: shutil.which on Windows also tries the current directory, which Node's spawn does not. --- src/opencode/_service.py | 22 ++++++++++++++++++---- tests/test_service.py | 16 ++++++++++++++++ 2 files changed, 34 insertions(+), 4 deletions(-) diff --git a/src/opencode/_service.py b/src/opencode/_service.py index 10a9c21..753e48e 100644 --- a/src/opencode/_service.py +++ b/src/opencode/_service.py @@ -78,8 +78,8 @@ class BasicAuth: username: str """Basic auth user name, always ``opencode``.""" - password: str - """The service's private password.""" + password: str = field(repr=False) + """The service's private password. Hidden from ``repr`` so it stays out of logs.""" @dataclass(frozen=True) @@ -101,7 +101,7 @@ class _Info: pid: int id: str | None = None version: str | None = None - password: str | None = None + password: str | None = field(default=None, repr=False) @dataclass(frozen=True) @@ -540,7 +540,7 @@ class _Contender: def __init__(self, command: Sequence[str], env: Mapping[str, str | None]) -> None: merged = {**os.environ, **env} environment = {key: value for key, value in merged.items() if value is not None} - executable = shutil.which(command[0]) or command[0] + executable = _resolve(command[0]) if sys.platform == "win32": # pragma: no cover - Windows only self.process = subprocess.Popen( [executable, *command[1:]], @@ -604,6 +604,20 @@ def failure(self) -> ServiceError | None: return _startup_error(f"Server process terminated by {name}", self.stderr()) +def _resolve(command: str) -> str: + """Find ``command`` on ``PATH`` only. + + ``shutil.which`` on Windows also searches the current directory first, which Node's spawn in the TypeScript + client does not, so a planted ``opencode.exe`` there could run. A command with a directory part is used as is. + """ + if os.path.dirname(command): + return command + for directory in os.environ.get("PATH", os.defpath).split(os.pathsep): + if directory and (found := shutil.which(os.path.join(directory, command))) is not None: + return found + return command + + def _startup_error(message: str, stderr: str) -> ServiceError: return ServiceError(f"{message}\n{stderr}" if stderr else message) diff --git a/tests/test_service.py b/tests/test_service.py index ec24b5a..e52d0de 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -271,3 +271,19 @@ def test_probe_timeout_is_reported(file: Path) -> None: registration = impl._registered(file, 0.1) assert registration.timed_out assert registration.service is None + + +def test_resolve_searches_path_only(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + bin_dir, cwd = tmp_path / "bin", tmp_path / "cwd" + bin_dir.mkdir() + cwd.mkdir() + name = "tool.exe" if sys.platform == "win32" else "tool" + for folder in (bin_dir, cwd): + (folder / name).write_text("") + (folder / name).chmod(0o755) + monkeypatch.chdir(cwd) + monkeypatch.setenv("PATH", str(bin_dir)) + assert Path(impl._resolve("tool")).parent == bin_dir + monkeypatch.setenv("PATH", "") + assert impl._resolve("tool") == "tool" + assert impl._resolve(str(cwd / name)) == str(cwd / name) From c0f7fc5499ebe885842a18cc29d42c3c2e9593f9 Mon Sep 17 00:00:00 2001 From: shrijayan <81805145+shrijayan@users.noreply.github.com> Date: Thu, 22 Oct 2026 16:01:38 +0530 Subject: [PATCH 3/3] Refuse to start a service command that is not on PATH Falling back to the bare name let Windows CreateProcess search the current directory. Look the command up on the child's PATH, as Node's spawn does, and raise (surfaced as "Failed to start server") when it is missing. --- src/opencode/_service.py | 14 +++++++++----- tests/test_service.py | 9 ++++----- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/src/opencode/_service.py b/src/opencode/_service.py index 753e48e..763d4bb 100644 --- a/src/opencode/_service.py +++ b/src/opencode/_service.py @@ -540,7 +540,7 @@ class _Contender: def __init__(self, command: Sequence[str], env: Mapping[str, str | None]) -> None: merged = {**os.environ, **env} environment = {key: value for key, value in merged.items() if value is not None} - executable = _resolve(command[0]) + executable = _resolve(command[0], environment.get("PATH", os.defpath)) if sys.platform == "win32": # pragma: no cover - Windows only self.process = subprocess.Popen( [executable, *command[1:]], @@ -604,18 +604,22 @@ def failure(self) -> ServiceError | None: return _startup_error(f"Server process terminated by {name}", self.stderr()) -def _resolve(command: str) -> str: - """Find ``command`` on ``PATH`` only. +def _resolve(command: str, search_path: str) -> str: + """Find ``command`` on ``search_path``, the child's ``PATH``, as Node's spawn does. ``shutil.which`` on Windows also searches the current directory first, which Node's spawn in the TypeScript client does not, so a planted ``opencode.exe`` there could run. A command with a directory part is used as is. + + Raises: + FileNotFoundError: ``command`` is not on ``PATH``. """ if os.path.dirname(command): return command - for directory in os.environ.get("PATH", os.defpath).split(os.pathsep): + for directory in search_path.split(os.pathsep): if directory and (found := shutil.which(os.path.join(directory, command))) is not None: return found - return command + # Never hand Popen a bare name: Windows CreateProcess would search the current directory for it. + raise FileNotFoundError(f"{command!r} was not found on PATH") def _startup_error(message: str, stderr: str) -> ServiceError: diff --git a/tests/test_service.py b/tests/test_service.py index e52d0de..38aae70 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -282,8 +282,7 @@ def test_resolve_searches_path_only(tmp_path: Path, monkeypatch: pytest.MonkeyPa (folder / name).write_text("") (folder / name).chmod(0o755) monkeypatch.chdir(cwd) - monkeypatch.setenv("PATH", str(bin_dir)) - assert Path(impl._resolve("tool")).parent == bin_dir - monkeypatch.setenv("PATH", "") - assert impl._resolve("tool") == "tool" - assert impl._resolve(str(cwd / name)) == str(cwd / name) + assert Path(impl._resolve("tool", str(bin_dir))).parent == bin_dir + with pytest.raises(FileNotFoundError, match="not found on PATH"): + impl._resolve("tool", "") # the copy in the current directory is never used + assert impl._resolve(str(cwd / name), "") == str(cwd / name)