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..763d4bb --- /dev/null +++ b/src/opencode/_service.py @@ -0,0 +1,1018 @@ +"""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 = field(repr=False) + """The service's private password. Hidden from ``repr`` so it stays out of logs.""" + + +@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 = field(default=None, repr=False) + + +@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 = _resolve(command[0], environment.get("PATH", os.defpath)) + 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 _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 search_path.split(os.pathsep): + if directory and (found := shutil.which(os.path.join(directory, command))) is not None: + return found + # 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: + 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..38aae70 --- /dev/null +++ b/tests/test_service.py @@ -0,0 +1,288 @@ +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 + + +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) + 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) 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()