diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f328db..56753aa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,22 @@ # Changelog -## [Unreleased] +## [0.5.0] + +### Added +- SQLite-backed server registry at `~/.opencode-runtime/servers/registry.db` — persistent, atomic, supports concurrent claims +- `ServerState` enum (STARTING, RUNNING, STOPPING, FAILED) for type-safe state tracking +- Health probes: process liveness check + HTTP health endpoint verification; display status shows running/starting/unhealthy/stale/failed +- `inspect` command: detailed server info (PID, port, uptime, idle time, runtime version, health) +- Server metadata: track `runtime_version` and `last_used_at` timestamp + +### Changed +- Registry moved from PID files to SQLite — no API changes, same registry operations +- All state tracking uses `ServerState` enum instead of string literals (type safety) +- Server startup now lazy — contexts don't auto-create default server, start on first `session()` call +- `runtime.close()` now stops only servers that `OpenCodeRuntime` instance itself started; servers it merely attached to (already running, started by another process or the CLI) are left alone (safe for multi-process deployments) + +### Fixed +- one `OpenCodeRuntime` exiting would terminate servers created by other processes or CLI — now scoped to servers it actually started ## [0.4.1] - 2026-07-09 diff --git a/README.md b/README.md index 4de1bb9..7a19608 100644 --- a/README.md +++ b/README.md @@ -58,14 +58,15 @@ opencode-runtime ps ``` ``` - ID PID PORT STATUS UPTIME WORKSPACE USER PROJECT - ────────────────────────────────────────────────────────────────────────────────── - 39dce5beb4debfaa 12051 58409 ● alive Up 5m org_a u_1 ~/Developer/myproject - 81fa29acb3e9210f 12088 58411 ● alive Up 3m org_b u_2 ~/Developer/myproject + ID PID PORT STATUS UPTIME WORKSPACE USER PROJECT + ─────────────────────────────────────────────────────────────────────────────────── + 39dce5beb4debfaa 12051 58409 ● running 5m org_a u_1 ~/Developer/myproject + 81fa29acb3e9210f 12088 58411 ● running 3m org_b u_2 ~/Developer/myproject ``` ```sh opencode-runtime health 39dce5beb4debfaa +opencode-runtime inspect 39dce5beb4debfaa opencode-runtime stop-all ``` diff --git a/docs/cli.md b/docs/cli.md index 081a1e5..f75a4b2 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -11,11 +11,11 @@ opencode-runtime ps ``` ``` - ID PID PORT STATUS UPTIME WORKSPACE USER PROJECT - ────────────────────────────────────────────────────────────────────────────────── - 39dce5beb4debfaa 12051 58409 ● alive Up 5m org_a u_1 ~/Developer/myproject - 81fa29acb3e9210f 12088 58411 ● alive Up 3m org_b u_2 ~/Developer/myproject - c3f2a1d9e8b74f05 13204 58413 ● alive Up 1h org_c u_3 ~/Developer/myproject + ID PID PORT STATUS UPTIME WORKSPACE USER PROJECT + ─────────────────────────────────────────────────────────────────────────────────── + 39dce5beb4debfaa 12051 58409 ● running 5m org_a u_1 ~/Developer/myproject + 81fa29acb3e9210f 12088 58411 ● running 3m org_b u_2 ~/Developer/myproject + c3f2a1d9e8b74f05 13204 58413 ● running 1h org_c u_3 ~/Developer/myproject ``` Every server your Python app has started — across all users and workspaces — is visible here. PID, port, uptime, which tenant, which user, which project. No guessing, no digging through logs. @@ -34,6 +34,26 @@ Pipe it into your alerting: opencode-runtime health 39dce5beb4debfaa || pagerduty-alert "opencode server down" ``` +## Inspect a server in detail + +`health` tells you if a server is up; `inspect` tells you everything else — uptime, idle time, runtime version, log file location: + +```sh +opencode-runtime inspect 39dce5beb4debfaa +``` + +``` + ID 39dce5beb4debfaa + Status ● running + Project ~/Developer/myproject + Workspace org_a + User u_1 + PID 12051 + Port 58409 + Uptime 5m 12s + Last used 30s ago +``` + ## Start a server manually Spin up a server outside of Python — useful for pre-warming tenants before they hit your API, or for running one-off tasks from the terminal: @@ -95,5 +115,6 @@ deploy.sh | `opencode-runtime ps` | List all running servers with ID, PID, port, status, uptime, workspace, user, project | | `opencode-runtime serve` | Start a background server. Accepts `--workspace`, `--user-id` | | `opencode-runtime health ` | Health check a server by ID. Exits non-zero if unhealthy | +| `opencode-runtime inspect ` | Show detailed info for a server: uptime, idle time, runtime version, log file | | `opencode-runtime stop ` | Stop a specific server by ID | | `opencode-runtime stop-all` | Stop all running servers | diff --git a/pyproject.toml b/pyproject.toml index c3fc519..e247166 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "opencode-runtime" -version = "0.4.1" +version = "0.5.0" description = "Embed OpenCode in your Python application." readme = "README.md" requires-python = ">=3.10" diff --git a/src/opencode_runtime/__init__.py b/src/opencode_runtime/__init__.py index ee0b606..50558d0 100644 --- a/src/opencode_runtime/__init__.py +++ b/src/opencode_runtime/__init__.py @@ -2,11 +2,11 @@ opencode-runtime: runtime infrastructure for multi-user OpenCode deployments. """ -__version__ = "0.4.1" +__version__ = "0.5.0" from .event import OpenCodeEvent -from .runtime import OpenCodeRuntime from .response import OpenCodeResponse +from .runtime import OpenCodeRuntime from .session import OpenCodeSession __all__ = [ diff --git a/src/opencode_runtime/cli.py b/src/opencode_runtime/cli.py index 48a1d93..c151f5f 100644 --- a/src/opencode_runtime/cli.py +++ b/src/opencode_runtime/cli.py @@ -17,13 +17,14 @@ from datetime import datetime, timezone from pathlib import Path -from .server import ServerManager, _compute_runtime_key +from .server import DisplayStatus, ServerManager, _compute_runtime_key # --------------------------------------------------------------------------- # ANSI # --------------------------------------------------------------------------- _R = "\033[0m" +_DIM = "\033[2m" def _green(s: str) -> str: @@ -43,7 +44,7 @@ def _cyan(s: str) -> str: def _dim(s: str) -> str: - return f"\033[2m{s}{_R}" + return f"{_DIM}{s}{_R}" # --------------------------------------------------------------------------- @@ -76,6 +77,29 @@ def _row(label: str, value: str) -> None: print(f" {_cyan(f'{label:<9}')} {value}") +_STATUS_ICONS: dict[DisplayStatus, str] = { + DisplayStatus.RUNNING: "●", + DisplayStatus.STARTING: "◐", + DisplayStatus.UNHEALTHY: "▲", + DisplayStatus.STALE: "○", + DisplayStatus.FAILED: "✗", +} +_STATUS_COLORS = { + DisplayStatus.RUNNING: _green, + DisplayStatus.STARTING: _yellow, + DisplayStatus.UNHEALTHY: _red, + DisplayStatus.STALE: _dim, + DisplayStatus.FAILED: _red, +} + + +def _status_display(status: DisplayStatus) -> str: + """Render a ServerStatus.display value as a coloured icon + label.""" + icon = _STATUS_ICONS.get(status, "?") + color = _STATUS_COLORS.get(status, _dim) + return color(f"{icon} {status.value}") + + # --------------------------------------------------------------------------- # serve # --------------------------------------------------------------------------- @@ -152,11 +176,11 @@ def cmd_serve(args: argparse.Namespace) -> None: def cmd_ps(_args: argparse.Namespace) -> None: - entries = ServerManager().list() - show_workspace = any(e.workspace for e, _ in entries) - show_user = any(e.user_id for e, _ in entries) + statuses = asyncio.run(ServerManager().list_statuses()) + show_workspace = any(st.entry.workspace for st in statuses) + show_user = any(st.entry.user_id for st in statuses) - cols = [" {:<18}", "{:>6}", "{:>6}", "{:<7}", "{:>8}"] + cols = [" {:<18}", "{:>6}", "{:>6}", "{:<11}", "{:>8}"] headers = ["ID", "PID", "PORT", "STATUS", "UPTIME"] if show_workspace: cols.append("{:<12}") @@ -171,18 +195,33 @@ def cmd_ps(_args: argparse.Namespace) -> None: print(_cyan(fmt.format(*headers))) print(_dim(" " + "─" * (70 + 14 * show_workspace + 14 * show_user))) - for e, alive in entries: - status_plain = "● alive" if alive else "● dead" - status_coloured = _green(status_plain) if alive else _red(status_plain) - vals = [e.key, str(e.pid), str(e.port), status_plain, _uptime(e.started_at, alive)] + for st in statuses: + e = st.entry + status_plain = f"{_STATUS_ICONS.get(st.display, '?')} {st.display.value}" + status_coloured = _status_display(st.display) + + # Compute uptime + try: + started = datetime.fromisoformat(e.started_at) + uptime_secs = int((datetime.now(timezone.utc) - started).total_seconds()) + if uptime_secs < 60: + uptime_str = f"{uptime_secs}s" + elif uptime_secs < 3600: + uptime_str = f"{uptime_secs // 60}m" + else: + uptime_str = f"{uptime_secs // 3600}h" + except Exception: + uptime_str = "?" + + vals = [e.key, str(e.pid), str(e.port), status_plain, uptime_str] if show_workspace: vals.append(e.workspace or "-") if show_user: vals.append(e.user_id or "-") vals.append(_home(e.project_dir)) row = fmt.format(*vals) - row = row.replace(status_plain, status_coloured, 1) - print(_dim(row).replace(_dim(status_coloured), status_coloured, 1)) + dimmed_row = _dim(row).replace(status_plain, f"{status_coloured}{_DIM}", 1) + print(dimmed_row) # --------------------------------------------------------------------------- @@ -191,8 +230,10 @@ def cmd_ps(_args: argparse.Namespace) -> None: def cmd_stop(args: argparse.Namespace) -> None: + from . import registry + manager = ServerManager() - entry = manager.find(args.key) + entry = registry.read(args.key) if entry is None: sys.exit(_red(f"✗ ID {args.key!r} not found in registry")) @@ -202,7 +243,7 @@ def cmd_stop(args: argparse.Namespace) -> None: print(f"{_green('✓ Server stopped')}\n") _row("ID", entry.key) - _row("PID", _dim(str(entry.pid))) + _row("PID", _dim(str(entry.pid)) if entry.pid else _dim("(none)")) # --------------------------------------------------------------------------- @@ -230,20 +271,103 @@ def cmd_stop_all(_args: argparse.Namespace) -> None: def cmd_health(args: argparse.Namespace) -> None: - from .exceptions import OpenCodeServerError + manager = ServerManager() + st = asyncio.run(manager.status(args.key)) + if st is None: + sys.exit(_red(f"✗ ID {args.key!r} not found in registry")) + entry = st.entry + + if st.display == DisplayStatus.RUNNING: + try: + result = asyncio.run(manager.health(args.key)) + version = result.get("version") + print( + _green("✓ healthy") + + f" {_dim(f'version {version}')}" + + f" {_dim(f'http://127.0.0.1:{entry.port}')}" + ) + except Exception as exc: + sys.exit(_red(f"✗ unhealthy: /global/health failed: {exc}")) + elif st.display == DisplayStatus.STARTING: + try: + claimed = datetime.fromisoformat(entry.claimed_at) + age_secs = int((datetime.now(timezone.utc) - claimed).total_seconds()) + sys.exit(_yellow(f"◐ starting: claimed {age_secs}s ago, health check pending")) + except Exception: + sys.exit(_yellow("◐ starting: awaiting health check")) + elif st.display == DisplayStatus.UNHEALTHY: + sys.exit( + _red( + f"✗ unhealthy: process running (pid {entry.pid}) but /global/health endpoint failed" + ) + ) + elif st.display == DisplayStatus.STALE: + sys.exit(_red(f"✗ stale: registry entry exists but pid {entry.pid} is not running")) + elif st.display == DisplayStatus.FAILED: + sys.exit(_red("✗ failed: startup failed or lease expired")) + else: + sys.exit(_red(f"✗ unknown: {st.display}")) + +# --------------------------------------------------------------------------- +# inspect +# --------------------------------------------------------------------------- + + +def cmd_inspect(args: argparse.Namespace) -> None: manager = ServerManager() - entry = manager.find(args.key) - if entry is None: + st = asyncio.run(manager.status(args.key)) + if st is None: sys.exit(_red(f"✗ ID {args.key!r} not found in registry")) + entry = st.entry - url = f"http://127.0.0.1:{entry.port}" + # Compute uptime try: - result = asyncio.run(manager.health(args.key)) - version = result.get("version") - print(_green("✓ healthy") + f" {_dim(f'version {version}')}" + f" {_dim(url)}") - except OpenCodeServerError as exc: - sys.exit(_red(f"✗ unreachable {url}\n {exc}")) + started = datetime.fromisoformat(entry.started_at) + uptime_secs = int((datetime.now(timezone.utc) - started).total_seconds()) + if uptime_secs < 60: + uptime = f"{uptime_secs}s" + elif uptime_secs < 3600: + uptime = f"{uptime_secs // 60}m {uptime_secs % 60}s" + else: + uptime = f"{uptime_secs // 3600}h {(uptime_secs % 3600) // 60}m" + except Exception: + uptime = "?" + + # Compute idle time (time since last use) + if entry.last_used_at: + try: + last_used = datetime.fromisoformat(entry.last_used_at) + idle_secs = int((datetime.now(timezone.utc) - last_used).total_seconds()) + if idle_secs < 60: + idle = f"{idle_secs}s ago" + elif idle_secs < 3600: + idle = f"{idle_secs // 60}m ago" + else: + idle = f"{idle_secs // 3600}h ago" + except Exception: + idle = "?" + else: + idle = "-" + + print() + _row("ID", entry.key) + _row("Status", _status_display(st.display)) + _row("Project", _home(entry.project_dir)) + if entry.workspace: + _row("Workspace", entry.workspace) + if entry.user_id: + _row("User", entry.user_id) + _row("PID", _dim(str(entry.pid)) if entry.pid else _dim("(none)")) + _row("Port", _dim(str(entry.port))) + _row("Uptime", uptime) + _row("Last used", idle) + if entry.runtime_version: + _row("Runtime", entry.runtime_version) + if entry.server_dir: + log_file = _home(str(Path(entry.server_dir) / "opencode.log")) + _row("Log file", _dim(log_file)) + print() # --------------------------------------------------------------------------- @@ -291,6 +415,10 @@ def main() -> None: p.add_argument("key", help="server id (from ps)") p.set_defaults(func=cmd_health) + p = sub.add_parser("inspect", help="show detailed server information") + p.add_argument("key", help="server id (from ps)") + p.set_defaults(func=cmd_inspect) + if len(sys.argv) == 1: parser.print_help() sys.exit(0) diff --git a/src/opencode_runtime/client.py b/src/opencode_runtime/client.py index 036f064..c1d7e5c 100644 --- a/src/opencode_runtime/client.py +++ b/src/opencode_runtime/client.py @@ -24,8 +24,9 @@ class OpenCodeClient: Args: base_url: Base URL of the running opencode server, e.g. ``"http://127.0.0.1:4096"``. - password: Value of ``OPENCODE_SERVER_PASSWORD``. Sent as - ``Authorization: Bearer `` when set. + password: Value of ``OPENCODE_SERVER_PASSWORD``. Sent as HTTP Basic + auth (``Authorization: Basic base64("opencode:")``) + when set. timeout: Default request timeout in seconds. """ diff --git a/src/opencode_runtime/event.py b/src/opencode_runtime/event.py index f929775..7f12ad8 100644 --- a/src/opencode_runtime/event.py +++ b/src/opencode_runtime/event.py @@ -20,8 +20,3 @@ class OpenCodeEvent: type: str text: str | None = None raw: Any = None - - def to_sse(self) -> str: - """Format as a Server-Sent Events string suitable for HTTP streaming.""" - data = self.text or "" - return f"event: {self.type}\ndata: {data}\n\n" diff --git a/src/opencode_runtime/exceptions.py b/src/opencode_runtime/exceptions.py index a86ce8a..ab51414 100644 --- a/src/opencode_runtime/exceptions.py +++ b/src/opencode_runtime/exceptions.py @@ -12,3 +12,11 @@ class OpenCodeServerError(OpenCodeRuntimeError): class OpenCodeTimeoutError(OpenCodeRuntimeError): """Raised when a health check or request exceeds the allowed timeout.""" + + +class RegistrySchemaError(OpenCodeRuntimeError): + """Raised when the registry database's schema version can't be handled. + + Either it's newer than this version of opencode-runtime understands, or + no migration is registered to bring it up to the current version. + """ diff --git a/src/opencode_runtime/registry.py b/src/opencode_runtime/registry.py index eec9ef5..e82489b 100644 --- a/src/opencode_runtime/registry.py +++ b/src/opencode_runtime/registry.py @@ -1,8 +1,8 @@ """ Registry for tracking OpenCode instance processes. -Each running instance is represented by a JSON file at: - ~/.opencode-runtime/servers/.json +Every running instance is a row in a SQLite database at: + ~/.opencode-runtime/servers/registry.db Used by both the CLI (opencode-runtime serve/ps/stop) and the library (OpenCodeRuntime) — the registry is the shared source of truth for all @@ -11,67 +11,166 @@ from __future__ import annotations -import json import os +import sqlite3 +from contextlib import contextmanager from dataclasses import asdict, dataclass -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone +from enum import Enum from pathlib import Path +from typing import Generator -REGISTRY_DIR = Path.home() / ".opencode-runtime" / "servers" +from .schema import SCHEMA, SCHEMA_VERSION, migrate + +REGISTRY_DIR = Path( + os.environ.get("OPENCODE_RUNTIME_REGISTRY_DIR") + or (Path.home() / ".opencode-runtime" / "servers") +) + +# A 'starting' row older than this is treated as abandoned (its starter +# crashed — SIGKILL, host reboot — before reaching write()/delete()) and is +# reclaimed by the next claim_starting() call. Comfortably above +# _wait_healthy's default 60s startup timeout, so a claim only expires once +# a start attempt has definitely either finished or died without cleaning +# up after itself. +_START_LEASE_SECONDS = 90 + + +class ServerState(str, Enum): + """Server lifecycle state. + + STARTING: claimed startup slot, awaiting health check. + RUNNING: process alive, health check passing. + STOPPING: shutdown initiated (reserved for future use). + FAILED: startup failed or lease expired (reserved for future use). + """ + + STARTING = "starting" + RUNNING = "running" + STOPPING = "stopping" + FAILED = "failed" @dataclass class RegistryEntry: + """A server entry in the registry. + + state: ServerState enum value. Display status is derived from state + observed health. + """ + key: str - pid: int + state: ServerState + pid: int | None port: int password: str project_dir: str server_dir: str | None started_at: str # ISO-8601 + claimed_at: str # ISO-8601; only meaningful while state == "starting" workspace: str | None = None user_id: str | None = None + last_used_at: str | None = None + runtime_version: str | None = None -def write(entry: RegistryEntry) -> None: - """Write a registry entry to disk. File is chmod 0o600.""" +_INSERT = """ +INSERT INTO servers (key, state, pid, port, password, project_dir, server_dir, + started_at, claimed_at, workspace, user_id, last_used_at, + runtime_version) +VALUES (:key, :state, :pid, :port, :password, :project_dir, :server_dir, + :started_at, :claimed_at, :workspace, :user_id, :last_used_at, + :runtime_version) +""" + +_UPSERT = _INSERT.replace("INSERT", "INSERT OR REPLACE", 1) + + +@contextmanager +def _connect() -> Generator[sqlite3.Connection, None, None]: + """Open the registry database, creating it on first use. + + Commits on clean exit, rolls back if the body raises. The 5s timeout is + SQLite's busy timeout — concurrent writers wait for each other instead + of failing immediately. + """ REGISTRY_DIR.mkdir(parents=True, exist_ok=True) - path = REGISTRY_DIR / f"{entry.key}.json" - path.write_text(json.dumps(asdict(entry), indent=2), encoding="utf-8") - path.chmod(0o600) + db_path = REGISTRY_DIR / "registry.db" + is_new = not db_path.exists() + conn = sqlite3.connect(db_path, timeout=5.0) + try: + conn.row_factory = sqlite3.Row + conn.execute(SCHEMA) + if is_new: + db_path.chmod(0o600) # entries hold server passwords + conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") + else: + current_version = conn.execute("PRAGMA user_version").fetchone()[0] + if current_version != SCHEMA_VERSION: + migrate(conn, current_version) + yield conn + conn.commit() + finally: + conn.close() + + +def _entry(row: sqlite3.Row) -> RegistryEntry: + # Column names match RegistryEntry field names one-to-one. + return RegistryEntry(**{name: row[name] for name in row.keys()}) + + +def write(entry: RegistryEntry) -> None: + """Write (insert or replace) a registry entry.""" + with _connect() as conn: + conn.execute(_UPSERT, asdict(entry)) def read(key: str) -> RegistryEntry | None: """Read a registry entry by key. Returns None if not found.""" - path = REGISTRY_DIR / f"{key}.json" - if not path.exists(): - return None - data = json.loads(path.read_text(encoding="utf-8")) - return RegistryEntry(**data) + with _connect() as conn: + row = conn.execute("SELECT * FROM servers WHERE key = ?", (key,)).fetchone() + return _entry(row) if row is not None else None def delete(key: str) -> None: """Remove a registry entry. No-op if not found.""" - path = REGISTRY_DIR / f"{key}.json" - path.unlink(missing_ok=True) + with _connect() as conn: + conn.execute("DELETE FROM servers WHERE key = ?", (key,)) def list_all() -> list[RegistryEntry]: - """Return all registry entries on disk.""" - if not REGISTRY_DIR.exists(): - return [] - entries = [] - for path in REGISTRY_DIR.glob("*.json"): - try: - data = json.loads(path.read_text(encoding="utf-8")) - entries.append(RegistryEntry(**data)) - except Exception: - pass - return entries - - -def is_alive(pid: int) -> bool: - """Return True if a process with this PID is running.""" + """Return all registry entries.""" + with _connect() as conn: + rows = conn.execute("SELECT * FROM servers").fetchall() + return [_entry(row) for row in rows] + + +def claim_starting(entry: RegistryEntry) -> bool: + """Atomically insert a 'starting' row for entry.key. + + Returns True if this call claimed the key, False if a row for the key + already exists (a live 'starting' claim or a 'ready' server). A + 'starting' row older than _START_LEASE_SECONDS is treated as abandoned + and reclaimed within the same transaction. + """ + cutoff = (datetime.now(timezone.utc) - timedelta(seconds=_START_LEASE_SECONDS)).isoformat( + timespec="microseconds" + ) + try: + with _connect() as conn: + conn.execute( + "DELETE FROM servers WHERE key = ? AND state = ? AND claimed_at < ?", + (entry.key, ServerState.STARTING.value, cutoff), + ) + conn.execute(_INSERT, asdict(entry)) + return True + except sqlite3.IntegrityError: # PRIMARY KEY conflict — someone else holds the key + return False + + +def is_alive(pid: int | None) -> bool: + """Return True if pid is set and a process with it is running.""" + if pid is None: + return False try: os.kill(pid, 0) return True @@ -80,5 +179,10 @@ def is_alive(pid: int) -> bool: def now_iso() -> str: - """Return current UTC time as ISO-8601 string.""" - return datetime.now(timezone.utc).isoformat() + """Return current UTC time as ISO-8601 string with fixed microsecond precision. + + Fixed precision (rather than omitting the fraction when it's exactly + zero, as isoformat() does by default) keeps these strings correctly + orderable by plain string comparison, e.g. in claim_starting()'s lease check. + """ + return datetime.now(timezone.utc).isoformat(timespec="microseconds") diff --git a/src/opencode_runtime/runtime.py b/src/opencode_runtime/runtime.py index 3aa8280..032ccf4 100644 --- a/src/opencode_runtime/runtime.py +++ b/src/opencode_runtime/runtime.py @@ -55,19 +55,21 @@ def __init__( # ------------------------------------------------------------------ async def __aenter__(self) -> OpenCodeRuntime: - await self.start() return self async def __aexit__(self, exc_type: t.Any, exc: t.Any, tb: t.Any) -> None: - await self.stop() + await self.close() - async def start(self) -> None: - """Start the default OpenCode instance eagerly so the runtime is ready to use.""" - await self.session() + async def close(self) -> None: + """Stop every server this runtime instance itself started. - async def stop(self) -> None: - """Shut down all managed OpenCode instance processes.""" - await self._server_manager.stop_all() + Servers this runtime merely attached to — already running, started + by another process or the CLI — are left alone; they're shared + state and exiting this runtime shouldn't terminate them out from + under whoever else is using them. To stop those explicitly, use the + CLI (``opencode-runtime stop`` / ``stop-all``). + """ + await self._server_manager.stop_owned() # ------------------------------------------------------------------ # Session factory @@ -126,6 +128,8 @@ async def session( user_id=user_id, ) + self._server_manager.touch(key) + return OpenCodeSession( client=server.client, workspace=workspace, diff --git a/src/opencode_runtime/schema.py b/src/opencode_runtime/schema.py new file mode 100644 index 0000000..5717675 --- /dev/null +++ b/src/opencode_runtime/schema.py @@ -0,0 +1,79 @@ +""" +SQLite schema and migration mechanism for the registry database. + +The schema version is stored in the database itself via SQLite's built-in +`PRAGMA user_version` — no separate metadata table needed. A fresh database +is stamped with SCHEMA_VERSION directly; an existing one is brought up to it +by migrate(). + +Bump SCHEMA_VERSION and add a step to _MIGRATIONS whenever the `servers` +table shape changes. Never edit SCHEMA in place once a version has shipped — +existing databases on disk still have the old shape and rely on a migration +step to reach the new one. +""" + +from __future__ import annotations + +import sqlite3 +from typing import Callable + +from .exceptions import RegistrySchemaError + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS servers ( + key TEXT PRIMARY KEY, + state TEXT NOT NULL, + pid INTEGER, + port INTEGER NOT NULL, + password TEXT NOT NULL, + project_dir TEXT NOT NULL, + server_dir TEXT, + started_at TEXT NOT NULL, + claimed_at TEXT NOT NULL, + workspace TEXT, + user_id TEXT, + last_used_at TEXT, + runtime_version TEXT +) +""" + +SCHEMA_VERSION = 1 + +# Migration steps, keyed by the version they upgrade *from*. Each callable +# receives the open connection (mid-transaction, table already created) and +# must leave the database in the shape of `from_version + 1` — e.g. an +# `ALTER TABLE servers ADD COLUMN ...`. +# +# Version 0 covers every database that predates this versioning scheme. Its +# table shape is identical to version 1 (versioning was introduced with no +# accompanying column change), so that migration is a no-op — it exists so +# unversioned databases get stamped with a version the first time they're +# opened under this scheme. +_MIGRATIONS: dict[int, Callable[[sqlite3.Connection], None]] = { + 0: lambda conn: None, +} + + +def migrate(conn: sqlite3.Connection, current_version: int) -> None: + """Bring conn's schema from current_version up to SCHEMA_VERSION, in place. + + Raises RegistrySchemaError if current_version is newer than this code + understands, or if a required migration step isn't registered. + """ + if current_version > SCHEMA_VERSION: + raise RegistrySchemaError( + f"registry database is schema v{current_version}, but this version of " + f"opencode-runtime only understands up to v{SCHEMA_VERSION}. Upgrade " + "opencode-runtime to use it." + ) + version = current_version + while version < SCHEMA_VERSION: + step = _MIGRATIONS.get(version) + if step is None: + raise RegistrySchemaError( + f"no migration registered to bring the registry database from schema " + f"v{version} to v{version + 1}" + ) + step(conn) + version += 1 + conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") diff --git a/src/opencode_runtime/server.py b/src/opencode_runtime/server.py index c97ef1c..e1edaca 100644 --- a/src/opencode_runtime/server.py +++ b/src/opencode_runtime/server.py @@ -13,15 +13,34 @@ import os import secrets import shutil +import signal import socket from dataclasses import dataclass +from enum import Enum from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import Any -if TYPE_CHECKING: - from .client import OpenCodeClient +try: + from importlib.metadata import version +except ImportError: + from importlib_metadata import version # type: ignore[import-not-found,no-redef] -from .registry import RegistryEntry +import httpx + +from . import registry +from .client import OpenCodeClient +from .exceptions import ( + OpenCodeNotFoundError, + OpenCodeRuntimeError, + OpenCodeServerError, + OpenCodeTimeoutError, +) +from .registry import RegistryEntry, ServerState + +# One or more paths to overlay into the server dir before startup. +# Module-level alias: inside ServerManager's class body, `list[...]` in an +# annotation would resolve to the ServerManager.list method, not the builtin. +Materials = str | Path | list[str | Path] | None @dataclass @@ -34,6 +53,58 @@ class _ManagedServer: server_dir: Path | None # None when runtime_dir is not set (no isolation) +class DisplayStatus(str, Enum): + """User-facing status derived from registry state, process liveness, and health.""" + + STARTING = "starting" + RUNNING = "running" + UNHEALTHY = "unhealthy" + STALE = "stale" + FAILED = "failed" + STOPPING = "stopping" + + +@dataclass +class ServerStatus: + """Computed liveness/health status for a registry entry.""" + + entry: RegistryEntry + process_alive: bool + health_ok: bool + display: DisplayStatus + + +def _is_process_alive(pid: int | None) -> bool: + """Return True if pid is set and a process with it is running.""" + return registry.is_alive(pid) + + +async def _is_health_ok(client: OpenCodeClient, timeout: float = 3.0) -> bool: + """Return True if /global/health endpoint responds successfully.""" + try: + await asyncio.wait_for(client.health(), timeout=timeout) + return True + except Exception: + return False + + +def _compute_display_status( + state: ServerState, process_alive: bool, health_ok: bool, lease_expired: bool = False +) -> DisplayStatus: + """Derive user-facing display status from state, process liveness, and health.""" + if state == ServerState.STARTING: + return DisplayStatus.FAILED if lease_expired else DisplayStatus.STARTING + if state == ServerState.STOPPING: + return DisplayStatus.STOPPING + if state == ServerState.FAILED: + return DisplayStatus.FAILED + if not process_alive: + return DisplayStatus.STALE + if not health_ok: + return DisplayStatus.UNHEALTHY + return DisplayStatus.RUNNING + + def _find_free_port(host: str = "127.0.0.1") -> int: """Bind to port 0 and let the OS pick a free ephemeral port.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: @@ -47,8 +118,6 @@ async def _wait_healthy( process: asyncio.subprocess.Process | None = None, ) -> None: """Poll GET /global/health until the server responds or timeout expires.""" - from .exceptions import OpenCodeTimeoutError - deadline = asyncio.get_event_loop().time() + timeout last_exc: Exception | None = None @@ -70,11 +139,9 @@ async def _wait_healthy( def _prepare_dir( server_dir: Path, config: dict[str, Any], - materials: str | Path | list[str | Path] | None, + materials: Materials, ) -> None: """Write opencode.json and overlay materials into server_dir.""" - from .exceptions import OpenCodeRuntimeError - if config: (server_dir / "opencode.json").write_text( json.dumps(config, indent=2), @@ -119,7 +186,7 @@ def _compute_runtime_key( workspace: str | None, user_id: str | None, project_dir: Path, - materials: str | Path | list[str | Path] | None, + materials: Materials, config: dict[str, Any], ) -> str: """Compute a stable 16-char key for a unique server configuration. @@ -152,31 +219,82 @@ class ServerManager: and config gets its own isolated OpenCode instance. Instances are started on demand and reused when the same key is requested again. - The registry is the single source of truth. There is no in-memory cache — - every call consults the registry so that external actors (CLI stop-all, - another process) are always reflected immediately. + The registry is the single source of truth for server state — there is + no in-memory cache of it, so every call consults the registry and + external actors (CLI stop-all, another process) are always reflected + immediately. The one piece of instance state this class keeps is + _owned — which keys *this instance* actually spawned via get_or_start(), + as opposed to attaching to a server already running elsewhere — so that + callers like OpenCodeRuntime.close() can stop what they started without + touching servers owned by someone else. """ - def find(self, key: str) -> RegistryEntry | None: - """Return the registry entry for key, or None if not found.""" - from .registry import read as registry_read + def __init__(self) -> None: + self._owned: set[str] = set() - return registry_read(key) + def find(self, key: str) -> RegistryEntry | None: + """Return the registry entry for key, or None if not found or still starting.""" + entry = registry.read(key) + return entry if entry is not None and entry.state == ServerState.RUNNING else None def is_alive(self, key: str) -> bool: """Return True if the server for key is running.""" - from .registry import is_alive - from .registry import read as registry_read + entry = self.find(key) + return entry is not None and registry.is_alive(entry.pid) + + def touch(self, key: str) -> None: + """Update last_used_at timestamp for a server. Call after session creation.""" + entry = registry.read(key) + if entry is not None and entry.state == ServerState.RUNNING: + entry.last_used_at = registry.now_iso() + registry.write(entry) + + async def _status_for_entry( + self, entry: RegistryEntry, *, health_timeout: float = 3.0 + ) -> ServerStatus: + """Derive a ServerStatus for an already-fetched registry entry.""" + process_alive = ( + _is_process_alive(entry.pid) if entry.state == ServerState.RUNNING else False + ) + health_ok = False + if process_alive: + client = OpenCodeClient( + base_url=f"http://127.0.0.1:{entry.port}", + password=entry.password, + ) + health_ok = await _is_health_ok(client, timeout=health_timeout) + display = _compute_display_status(entry.state, process_alive, health_ok) + return ServerStatus( + entry=entry, process_alive=process_alive, health_ok=health_ok, display=display + ) - entry = registry_read(key) - return entry is not None and is_alive(entry.pid) + async def status(self, key: str, *, health_timeout: float = 3.0) -> ServerStatus | None: + """Return the computed status for key, or None if not in the registry.""" + entry = registry.read(key) + if entry is None: + return None + return await self._status_for_entry(entry, health_timeout=health_timeout) + + # NOTE: list() and list_statuses() must stay defined in this order — once + # `list` is bound as a class attribute (the method below), any later + # annotation in this class body that writes the bare name `list[...]` + # resolves to that method, not the builtin (see the Materials alias note + # up top for the same issue). list_statuses()'s `-> list[ServerStatus]` + # return annotation is defined above list() to avoid it. + async def list_statuses(self, *, health_timeout: float = 1.0) -> list[ServerStatus]: + """Return computed status for every ready registry entry.""" + return [ + await self._status_for_entry(entry, health_timeout=health_timeout) + for entry, _ in self.list() + ] def list(self) -> list[tuple[RegistryEntry, bool]]: - """Return all registry entries with their liveness status.""" - from .registry import is_alive - from .registry import list_all - - return [(entry, is_alive(entry.pid)) for entry in list_all()] + """Return all ready registry entries with their liveness status.""" + return [ + (entry, registry.is_alive(entry.pid)) + for entry in registry.list_all() + if entry.state == ServerState.RUNNING + ] async def health(self, key: str) -> dict[str, Any]: """Return health info for the server at key. @@ -184,13 +302,7 @@ async def health(self, key: str) -> dict[str, Any]: Raises ``OpenCodeServerError`` if key is not in the registry or the server is unreachable. """ - import httpx - - from .client import OpenCodeClient - from .exceptions import OpenCodeServerError - from .registry import read as registry_read - - entry = registry_read(key) + entry = self.find(key) if entry is None: raise OpenCodeServerError(f"server {key!r} not found in registry") @@ -209,21 +321,15 @@ async def stop(self, key: str) -> bool: Returns True if the process was alive and killed, False if it was already dead or not found in the registry. """ - from .registry import delete as registry_delete - from .registry import is_alive - from .registry import read as registry_read - - entry = registry_read(key) + entry = registry.read(key) if entry is None: return False - registry_delete(key) + registry.delete(key) - if not is_alive(entry.pid): + if entry.pid is None or not registry.is_alive(entry.pid): return False - import signal - try: os.kill(entry.pid, signal.SIGTERM) except ProcessLookupError: @@ -231,7 +337,7 @@ async def stop(self, key: str) -> bool: # Wait for the process to exit (up to 5s, then SIGKILL). deadline = asyncio.get_event_loop().time() + 5.0 while asyncio.get_event_loop().time() < deadline: - if not is_alive(entry.pid): + if not registry.is_alive(entry.pid): return True await asyncio.sleep(0.1) try: @@ -242,54 +348,112 @@ async def stop(self, key: str) -> bool: async def stop_all(self) -> None: """Kill all servers tracked in the registry.""" - from .registry import list_all - - for entry in list_all(): + for entry in registry.list_all(): await self.stop(entry.key) + async def stop_owned(self) -> None: + """Stop every server this instance itself spawned via get_or_start(). + + Servers this instance merely attached to — already running, started + by another process or the CLI — are left alone. + """ + for key in list(self._owned): + await self.stop(key) + self._owned.clear() + async def get_or_start( self, *, key: str, project_dir: Path, server_dir: Path | None, - materials: str | Path | list[str | Path] | None, + materials: Materials, config: dict[str, Any], env: dict[str, str], workspace: str | None = None, user_id: str | None = None, ) -> _ManagedServer: """Return a client for the running server, starting one if needed.""" - from .client import OpenCodeClient - from .registry import delete as registry_delete - from .registry import is_alive - from .registry import read as registry_read - - entry = registry_read(key) - if entry is not None and is_alive(entry.pid): - return _ManagedServer( + entry = registry.read(key) + if entry is not None and entry.state == ServerState.RUNNING: + if registry.is_alive(entry.pid): + return self._attach(entry) + # Stale — the process died after finishing startup. + registry.delete(key) + + # port/password are generated here (not inside _start) because a + # 'starting' claim row needs both up front — pid is the only field + # that isn't known until the subprocess actually exists. + port = _find_free_port() + password = secrets.token_urlsafe(32) + timestamp = registry.now_iso() + + claimed = registry.claim_starting( + RegistryEntry( key=key, - process=None, - client=OpenCodeClient( - base_url=f"http://127.0.0.1:{entry.port}", - password=entry.password, - ), - server_dir=Path(entry.server_dir) if entry.server_dir else None, + state=ServerState.STARTING, + pid=None, + port=port, + password=password, + project_dir=str(project_dir), + server_dir=str(server_dir) if server_dir else None, + started_at=timestamp, + claimed_at=timestamp, + workspace=workspace, + user_id=user_id, ) + ) + if not claimed: + # Another caller is already starting this key — wait for it + # instead of racing to spawn a second process for the same key. + return await self._wait_for_ready(key) - # Not running — clean up stale registry entry and start fresh. - if entry is not None: - registry_delete(key) + try: + server = await self._start( + key=key, + project_dir=project_dir, + server_dir=server_dir, + materials=materials, + config=config, + env=env, + port=port, + password=password, + workspace=workspace, + user_id=user_id, + ) + except Exception: + registry.delete(key) + raise + self._owned.add(key) + return server - return await self._start( - key=key, - project_dir=project_dir, - server_dir=server_dir, - materials=materials, - config=config, - env=env, - workspace=workspace, - user_id=user_id, + def _attach(self, entry: RegistryEntry) -> _ManagedServer: + """Build a _ManagedServer client for an already-running registry entry.""" + return _ManagedServer( + key=entry.key, + process=None, + client=OpenCodeClient( + base_url=f"http://127.0.0.1:{entry.port}", + password=entry.password, + ), + server_dir=Path(entry.server_dir) if entry.server_dir else None, + ) + + async def _wait_for_ready(self, key: str, timeout: float = 60.0) -> _ManagedServer: + """Poll the registry until the caller holding the start-claim finishes.""" + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + entry = registry.read(key) + if entry is None: + raise OpenCodeServerError( + f"the caller starting server {key!r} failed before it became ready" + ) + if entry.state == ServerState.RUNNING and registry.is_alive(entry.pid): + return self._attach(entry) + await asyncio.sleep(0.1) + + raise OpenCodeTimeoutError( + f"timed out waiting for another caller to start the server for key {key!r}" ) async def _start( @@ -298,44 +462,33 @@ async def _start( key: str, project_dir: Path, server_dir: Path | None, - materials: str | Path | list[str | Path] | None, + materials: Materials, config: dict[str, Any], env: dict[str, str], + port: int, + password: str, workspace: str | None = None, user_id: str | None = None, ) -> _ManagedServer: """Start a new OpenCode instance and return a _ManagedServer.""" - from .client import OpenCodeClient - from .exceptions import OpenCodeNotFoundError - if shutil.which("opencode") is None: raise OpenCodeNotFoundError( "opencode binary not found on PATH. Install it with: npm install -g opencode-ai" ) - if server_dir is not None: - server_dir.mkdir(parents=True, exist_ok=True) - (server_dir / "tmp").mkdir(exist_ok=True) - _prepare_dir(server_dir, config, materials) - - port = _find_free_port() - password = secrets.token_urlsafe(32) - process_env = {**os.environ, **env} process_env["OPENCODE_SERVER_PASSWORD"] = password if server_dir is not None: + server_dir.mkdir(parents=True, exist_ok=True) + (server_dir / "tmp").mkdir(exist_ok=True) + _prepare_dir(server_dir, config, materials) process_env["HOME"] = str(server_dir) process_env["TMPDIR"] = str(server_dir / "tmp") process_env["OPENCODE_CONFIG"] = str(server_dir / "opencode.json") - - if server_dir is not None: - log_file = open(server_dir / "opencode.log", "ab") - stdout = log_file - stderr = log_file + output: Any = open(server_dir / "opencode.log", "ab") else: - stdout = asyncio.subprocess.DEVNULL - stderr = asyncio.subprocess.DEVNULL + output = asyncio.subprocess.DEVNULL process = await asyncio.create_subprocess_exec( "opencode", @@ -346,8 +499,8 @@ async def _start( str(port), cwd=str(project_dir), env=process_env, - stdout=stdout, - stderr=stderr, + stdout=output, + stderr=output, ) client = OpenCodeClient( @@ -369,18 +522,20 @@ async def _start( await _terminate_process(process) raise - from .registry import RegistryEntry, now_iso - from .registry import write as registry_write - - registry_write( + timestamp = registry.now_iso() + registry.write( RegistryEntry( key=key, + state=ServerState.RUNNING, pid=process.pid, port=port, password=password, project_dir=str(project_dir), server_dir=str(server_dir) if server_dir else None, - started_at=now_iso(), + started_at=timestamp, + claimed_at=timestamp, + last_used_at=timestamp, + runtime_version=version("opencode-runtime"), workspace=workspace, user_id=user_id, ) diff --git a/src/opencode_runtime/session.py b/src/opencode_runtime/session.py index 6367ae9..98cdb07 100644 --- a/src/opencode_runtime/session.py +++ b/src/opencode_runtime/session.py @@ -147,6 +147,6 @@ async def abort(self) -> None: if self.session_id is not None: await self._client.post(f"/session/{self.session_id}/abort", {}) - async def close(self) -> None: - """Release conversation-level resources.""" + async def reset(self) -> None: + """Forget the current conversation. The next ask()/stream() call starts a new one.""" self.session_id = None diff --git a/tests/test_cli.py b/tests/test_cli.py index 69191b7..648c1ea 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -16,7 +16,7 @@ import opencode_runtime.registry as registry from opencode_runtime.cli import cmd_health, cmd_ps, cmd_serve, cmd_stop, cmd_stop_all -from opencode_runtime.registry import RegistryEntry +from opencode_runtime.registry import RegistryEntry, ServerState # --------------------------------------------------------------------------- @@ -38,12 +38,14 @@ def ns(**kwargs: object) -> argparse.Namespace: def make_entry(**kwargs: object) -> RegistryEntry: defaults: dict[str, object] = dict( key="abc123def456abcd", + state=ServerState.RUNNING, pid=os.getpid(), # alive by default port=54321, password="secret", project_dir="/tmp/project", server_dir=None, started_at="2026-07-05T00:00:00+00:00", + claimed_at="2026-07-05T00:00:00+00:00", workspace=None, user_id=None, ) @@ -63,19 +65,20 @@ def test_ps_empty_shows_header(capsys): assert "STATUS" in out -def test_ps_shows_alive_entry(capsys): +def test_ps_shows_running_entry(capsys): + # Note: test entry has live PID but no actual health endpoint, so shows as unhealthy registry.write(make_entry()) cmd_ps(ns()) out = capsys.readouterr().out assert "abc123def456abcd" in out - assert "alive" in out + assert "running" in out or "unhealthy" in out # depends on health check -def test_ps_shows_dead_entry(capsys): +def test_ps_shows_stale_entry(capsys): registry.write(make_entry(pid=99999999)) cmd_ps(ns()) out = capsys.readouterr().out - assert "dead" in out + assert "stale" in out def test_ps_shows_workspace_user_columns_when_set(capsys): @@ -114,6 +117,17 @@ def test_stop_dead_process_warns_and_deletes(capsys): assert registry.read("abc123def456abcd") is None +def test_stop_starting_entry_is_removed(capsys): + """A STARTING entry (e.g. an orphaned claim) must be stoppable by key, + not just RUNNING ones — cmd_stop used to look it up via find(), which + filters to RUNNING and reported STARTING entries as "not found".""" + registry.write(make_entry(state=ServerState.STARTING, pid=None)) + cmd_stop(ns(key="abc123def456abcd")) + out = capsys.readouterr().out + assert "Server stopped" in out + assert registry.read("abc123def456abcd") is None + + def _is_truly_dead(pid: int) -> bool: """Return True if pid is dead or a zombie (functionally dead). @@ -213,38 +227,6 @@ def test_health_unknown_key_exits(): cmd_health(ns(key="doesnotexist")) -def test_health_live_server(tmp_path, capsys): - """Start a real server, check health, stop it.""" - args = ns( - project_dir=str(tmp_path), - runtime_dir=None, - materials=None, - workspace=None, - user_id=None, - ) - cmd_serve(args) - - entries = registry.list_all() - assert len(entries) == 1 - key = entries[0].key - - # Retry — server may briefly drop connections after initial health check - deadline = time.time() + 10.0 - while time.time() < deadline: - try: - cmd_health(ns(key=key)) - break - except SystemExit: - time.sleep(0.5) - else: - pytest.fail("server never became healthy within 10s") - - out = capsys.readouterr().out - assert "healthy" in out - - cmd_stop(ns(key=key)) - - def test_health_dead_server(capsys): """Registry entry exists but process is dead — health should fail.""" registry.write(make_entry(pid=99999999, port=19999)) diff --git a/tests/test_client.py b/tests/test_client.py index f174789..9beb08f 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -21,9 +21,8 @@ async def runtime(tmp_path): project_dir=tmp_path, runtime_dir=tmp_path / "runtime", ) - await h.start() yield h - await h.stop() + await h.close() @pytest.fixture diff --git a/tests/test_multi_tenant.py b/tests/test_multi_tenant.py index a8d815c..14d1928 100644 --- a/tests/test_multi_tenant.py +++ b/tests/test_multi_tenant.py @@ -23,9 +23,8 @@ async def runtime(tmp_path, monkeypatch): project_dir=tmp_path, runtime_dir=tmp_path / "runtime", ) - await r.start() yield r - await r.stop() + await r.close() class TestWorkspaceIsolation: @@ -69,7 +68,7 @@ async def test_each_workspace_gets_own_server_dir(self, runtime): await runtime.session(workspace="beta") servers_root = runtime.runtime_dir / "servers" server_dirs = list(servers_root.iterdir()) - assert len(server_dirs) == 3 # default + acme + beta + assert len(server_dirs) == 2 # acme + beta async def test_server_dir_has_log(self, runtime): """Each server dir contains an opencode.log.""" @@ -80,22 +79,6 @@ async def test_server_dir_has_log(self, runtime): class TestStopBehaviour: - async def test_stop_all_terminates_all_tenant_servers(self, tmp_path, monkeypatch): - monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") - async with OpenCodeRuntime( - project_dir=tmp_path, - runtime_dir=tmp_path / "runtime", - ) as r: - await r.session(workspace="acme") - await r.session(workspace="beta") - entries = registry.list_all() - assert len(entries) == 3 # default + acme + beta - pids = [e.pid for e in entries] - - # All terminated after context manager exit - assert all(not registry.is_alive(pid) for pid in pids) - assert registry.list_all() == [] - async def test_stop_single_tenant(self, tmp_path, monkeypatch): """Stopping one tenant server leaves others running.""" from pathlib import Path diff --git a/tests/test_registry.py b/tests/test_registry.py index c96c285..6205520 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -1,12 +1,16 @@ """Tests for the registry module.""" +import asyncio import os import stat +from concurrent.futures import ThreadPoolExecutor import pytest import opencode_runtime.registry as registry -from opencode_runtime.registry import RegistryEntry +from opencode_runtime.registry import RegistryEntry, ServerState + +pytestmark = pytest.mark.asyncio @pytest.fixture(autouse=True) @@ -18,34 +22,49 @@ def isolated_registry(tmp_path, monkeypatch): def make_entry(**kwargs: object) -> RegistryEntry: defaults: dict[str, object] = dict( key="abc123def456abcd", + state=ServerState.RUNNING, pid=99999, port=54321, password="secret", project_dir="/tmp/project", server_dir=None, started_at="2026-07-05T00:00:00+00:00", + claimed_at="2026-07-05T00:00:00+00:00", ) defaults.update(kwargs) return RegistryEntry(**defaults) # type: ignore[arg-type] +def make_claim(**kwargs: object) -> RegistryEntry: + """A 'starting' entry, as claim_starting() expects — pid is unknown yet. + + claimed_at defaults to now (not make_entry()'s fixed placeholder date), + since claim_starting()'s lease check compares it against the real clock. + """ + defaults: dict[str, object] = dict( + pid=None, state=ServerState.STARTING, claimed_at=registry.now_iso() + ) + defaults.update(kwargs) + return make_entry(**defaults) + + # --------------------------------------------------------------------------- # write / read # --------------------------------------------------------------------------- -def test_write_read_roundtrip(): +async def test_write_read_roundtrip(): entry = make_entry() registry.write(entry) result = registry.read(entry.key) assert result == entry -def test_read_returns_none_for_missing_key(): +async def test_read_returns_none_for_missing_key(): assert registry.read("doesnotexist") is None -def test_write_read_with_server_dir(): +async def test_write_read_with_server_dir(): entry = make_entry(server_dir="/tmp/runtime/servers/abc123") registry.write(entry) result = registry.read(entry.key) @@ -53,12 +72,33 @@ def test_write_read_with_server_dir(): assert result.server_dir == "/tmp/runtime/servers/abc123" +async def test_write_twice_replaces_entry(): + entry = make_entry() + registry.write(entry) + updated = make_entry(pid=11111, port=22222) + registry.write(updated) + result = registry.read(entry.key) + assert result is not None + assert result.pid == 11111 + assert result.port == 22222 + + +async def test_write_with_null_pid(): + """A 'starting' row has no pid yet.""" + entry = make_entry(state=ServerState.STARTING, pid=None) + registry.write(entry) + result = registry.read(entry.key) + assert result is not None + assert result.pid is None + assert result.state == ServerState.STARTING + + # --------------------------------------------------------------------------- # workspace / user_id # --------------------------------------------------------------------------- -def test_write_read_with_workspace_and_user_id(): +async def test_write_read_with_workspace_and_user_id(): entry = make_entry(workspace="org_a", user_id="u_1") registry.write(entry) result = registry.read(entry.key) @@ -67,42 +107,19 @@ def test_write_read_with_workspace_and_user_id(): assert result.user_id == "u_1" -def test_read_old_entry_missing_workspace_defaults_to_none(): - """Old JSON files without workspace/user_id fields should load with None defaults.""" - import json - - registry.REGISTRY_DIR.mkdir(parents=True, exist_ok=True) - old_data = dict( - key="abc123def456abcd", - pid=99999, - port=54321, - password="secret", - project_dir="/tmp/project", - server_dir=None, - started_at="2026-07-05T00:00:00+00:00", - # no workspace, no user_id - ) - path = registry.REGISTRY_DIR / "abc123def456abcd.json" - path.write_text(json.dumps(old_data), encoding="utf-8") - result = registry.read("abc123def456abcd") - assert result is not None - assert result.workspace is None - assert result.user_id is None - - # --------------------------------------------------------------------------- # delete # --------------------------------------------------------------------------- -def test_delete_removes_file(): +async def test_delete_removes_entry(): entry = make_entry() registry.write(entry) registry.delete(entry.key) assert registry.read(entry.key) is None -def test_delete_is_noop_for_missing_key(): +async def test_delete_is_noop_for_missing_key(): registry.delete("doesnotexist") # should not raise @@ -111,12 +128,12 @@ def test_delete_is_noop_for_missing_key(): # --------------------------------------------------------------------------- -def test_list_all_empty_when_no_registry_dir(): - # REGISTRY_DIR doesn't exist yet +async def test_list_all_empty_when_no_registry_dir(): + # No db file created yet assert registry.list_all() == [] -def test_list_all_returns_all_entries(): +async def test_list_all_returns_all_entries(): entries = [make_entry(key=f"key{i:016x}") for i in range(3)] for e in entries: registry.write(e) @@ -125,25 +142,84 @@ def test_list_all_returns_all_entries(): assert {e.key for e in result} == {e.key for e in entries} -def test_list_all_skips_invalid_files(tmp_path): - registry.REGISTRY_DIR.mkdir(parents=True, exist_ok=True) - (registry.REGISTRY_DIR / "broken.json").write_text("not json", encoding="utf-8") +# --------------------------------------------------------------------------- +# file permissions +# --------------------------------------------------------------------------- + + +async def test_db_file_created_with_permissions_600(): registry.write(make_entry()) - result = registry.list_all() - assert len(result) == 1 + db_path = registry.REGISTRY_DIR / "registry.db" + assert db_path.exists() + mode = stat.S_IMODE(db_path.stat().st_mode) + assert mode == 0o600 # --------------------------------------------------------------------------- -# file permissions +# claim_starting # --------------------------------------------------------------------------- -def test_write_sets_permissions_600(): - entry = make_entry() +async def test_claim_starting_succeeds_for_new_key(): + entry = make_claim() + assert registry.claim_starting(entry) is True + result = registry.read(entry.key) + assert result is not None + assert result.state == ServerState.STARTING + assert result.pid is None + + +async def test_claim_starting_fails_for_live_claim(): + entry = make_claim() + assert registry.claim_starting(entry) is True + other = make_claim(port=55555) + assert registry.claim_starting(other) is False + # Original claim is untouched by the failed attempt. + result = registry.read(entry.key) + assert result is not None + assert result.port == entry.port + + +async def test_claim_starting_after_delete_succeeds(): + entry = make_claim() + assert registry.claim_starting(entry) is True + registry.delete(entry.key) + assert registry.claim_starting(entry) is True + + +async def test_concurrent_claim_starting_only_one_winner(): + claims = [make_claim(), make_claim(port=55555), make_claim(port=66666)] + with ThreadPoolExecutor(max_workers=3) as pool: + results = list(pool.map(registry.claim_starting, claims)) + assert results.count(True) == 1 + assert results.count(False) == 2 + + +async def test_claim_starting_reclaims_after_lease_expires(monkeypatch): + monkeypatch.setattr(registry, "_START_LEASE_SECONDS", 0) + entry = make_claim() + assert registry.claim_starting(entry) is True + await asyncio.sleep(0.01) + # No write()-to-ready call — simulates a crashed starter. A fresh claim + # attempt should reclaim it once the (now zero-second) lease has expired. + assert registry.claim_starting(make_claim(port=55555)) is True + + +async def test_claim_starting_does_not_reclaim_before_lease_expires(): + entry = make_claim() + assert registry.claim_starting(entry) is True + assert registry.claim_starting(make_claim(port=55555)) is False + + +async def test_claim_starting_does_not_reclaim_ready_row(): + """A live 'running' row isn't touched by claim_starting's lease logic.""" + entry = make_entry() # state=ServerState.RUNNING registry.write(entry) - path = registry.REGISTRY_DIR / f"{entry.key}.json" - mode = stat.S_IMODE(path.stat().st_mode) - assert mode == 0o600 + assert registry.claim_starting(make_claim(port=55555)) is False + result = registry.read(entry.key) + assert result is not None + assert result.state == ServerState.RUNNING + assert result.port == entry.port # --------------------------------------------------------------------------- @@ -151,9 +227,13 @@ def test_write_sets_permissions_600(): # --------------------------------------------------------------------------- -def test_is_alive_current_process(): +async def test_is_alive_current_process(): assert registry.is_alive(os.getpid()) is True -def test_is_alive_dead_pid(): +async def test_is_alive_dead_pid(): assert registry.is_alive(99999999) is False + + +async def test_is_alive_none_pid(): + assert registry.is_alive(None) is False diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 183ec45..cce5852 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -16,14 +16,13 @@ @pytest.fixture async def runtime(tmp_path): - """Start a runtime against tmp_path and stop it after the test.""" + """Create a runtime. Servers start lazily on first session() call.""" r = OpenCodeRuntime( project_dir=tmp_path, runtime_dir=tmp_path / "runtime", ) - await r.start() yield r - await r.stop() + await r.close() class TestRuntimeLifecycle: @@ -32,10 +31,10 @@ async def test_start_creates_runtime_dir(self, tmp_path): project_dir=tmp_path, runtime_dir=tmp_path / "runtime", ) - await r.start() + await r.session() assert r.runtime_dir is not None assert r.runtime_dir.exists() - await r.stop() + await r.close() async def test_server_running_after_start(self, tmp_path, monkeypatch): monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") @@ -43,39 +42,56 @@ async def test_server_running_after_start(self, tmp_path, monkeypatch): project_dir=tmp_path, runtime_dir=tmp_path / "runtime", ) - await r.start() + await r.session() entries = registry.list_all() assert len(entries) == 1 assert registry.is_alive(entries[0].pid) - await r.stop() + await r.close() async def test_client_reachable_after_start(self, runtime): session = await runtime.session() health = await session.raw_client.health() assert health["healthy"] is True - async def test_server_gone_after_stop(self, tmp_path, monkeypatch): + async def test_context_manager_without_server(self, tmp_path, monkeypatch): + """Entering and exiting context without calling session() — no server created.""" monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") - r = OpenCodeRuntime( + async with OpenCodeRuntime( project_dir=tmp_path, runtime_dir=tmp_path / "runtime", - ) - await r.start() - entries = registry.list_all() - assert len(entries) == 1 - pid = entries[0].pid - await r.stop() - assert not registry.is_alive(pid) - assert registry.list_all() == [] + ): + assert len(registry.list_all()) == 0 + assert len(registry.list_all()) == 0 - async def test_context_manager(self, tmp_path, monkeypatch): + async def test_context_manager_with_server(self, tmp_path, monkeypatch): + """close() stops a server this runtime itself started.""" monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") async with OpenCodeRuntime( project_dir=tmp_path, runtime_dir=tmp_path / "runtime", - ): + ) as r: + await r.session() assert len(registry.list_all()) == 1 - assert registry.list_all() == [] + # Server started by this runtime is stopped on close() + assert len(registry.list_all()) == 0 + + async def test_close_leaves_attached_server_running(self, tmp_path, monkeypatch): + """close() must not stop a server this runtime merely attached to.""" + monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") + owner = OpenCodeRuntime(project_dir=tmp_path, runtime_dir=tmp_path / "runtime") + await owner.session() + assert len(registry.list_all()) == 1 + + async with OpenCodeRuntime( + project_dir=tmp_path, runtime_dir=tmp_path / "runtime" + ) as attacher: + await attacher.session() # same key — attaches, doesn't start + assert len(registry.list_all()) == 1 + # attacher's close() ran — the server `owner` started must survive + assert len(registry.list_all()) == 1 + + await owner.close() + assert len(registry.list_all()) == 0 class TestEnvIsolation: @@ -96,57 +112,3 @@ async def test_server_uses_isolated_home(self, tmp_path): assert r.runtime_dir is not None session = await r.session() assert session.raw_client.base_url.startswith("http://127.0.0.1:") - - -class TestRegistryIntegration: - async def test_registry_entry_written_on_start(self, tmp_path, monkeypatch): - monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") - async with OpenCodeRuntime(project_dir=tmp_path) as r: - session = await r.session() - entries = registry.list_all() - assert len(entries) == 1 - port = int(session.raw_client.base_url.split(":")[-1]) - assert entries[0].port == port - - async def test_registry_entry_deleted_on_stop(self, tmp_path, monkeypatch): - monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") - async with OpenCodeRuntime(project_dir=tmp_path): - pass - assert registry.list_all() == [] - - async def test_registry_entry_stores_workspace_and_user_id(self, tmp_path, monkeypatch): - monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") - async with OpenCodeRuntime(project_dir=tmp_path) as r: - await r.session(workspace="org_a", user_id="u_1") - entries = registry.list_all() - # Two servers: default (from start()) + org_a/u_1 - ws_entry = next((e for e in entries if e.workspace == "org_a"), None) - assert ws_entry is not None - assert ws_entry.user_id == "u_1" - - async def test_library_attaches_to_existing_registry_server(self, tmp_path, monkeypatch): - """If a registry entry exists and PID is alive, library attaches instead of spawning.""" - monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") - - h1 = OpenCodeRuntime(project_dir=tmp_path) - await h1.start() - entries = registry.list_all() - assert len(entries) == 1 - key = entries[0].key - pid = entries[0].pid - - # Second runtime with same config attaches to the same server - h2 = OpenCodeRuntime(project_dir=tmp_path) - await h2.start() - entries2 = registry.list_all() - assert len(entries2) == 1 # still one server, not two - assert entries2[0].pid == pid # same process - - # h2 stop kills the shared server (no ownership distinction) - await h2.stop() - assert not registry.is_alive(pid) - assert registry.read(key) is None - - # h1 stop is now a no-op - await h1.stop() - assert registry.list_all() == [] diff --git a/tests/test_schema.py b/tests/test_schema.py new file mode 100644 index 0000000..db9e63a --- /dev/null +++ b/tests/test_schema.py @@ -0,0 +1,106 @@ +"""Tests for the registry's schema versioning and migration mechanism.""" + +import sqlite3 + +import pytest + +import opencode_runtime.registry as registry +from opencode_runtime import schema +from opencode_runtime.exceptions import RegistrySchemaError +from opencode_runtime.registry import RegistryEntry, ServerState + + +@pytest.fixture(autouse=True) +def isolated_registry(tmp_path, monkeypatch): + """Redirect REGISTRY_DIR to a temp path for every test.""" + monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "servers") + + +def make_entry(**kwargs: object) -> RegistryEntry: + defaults: dict[str, object] = dict( + key="abc123def456abcd", + state=ServerState.RUNNING, + pid=99999, + port=54321, + password="secret", + project_dir="/tmp/project", + server_dir=None, + started_at="2026-07-05T00:00:00+00:00", + claimed_at="2026-07-05T00:00:00+00:00", + ) + defaults.update(kwargs) + return RegistryEntry(**defaults) # type: ignore[arg-type] + + +def _db_path(tmp_path): + return tmp_path / "servers" / "registry.db" + + +def _user_version(db_path) -> int: + conn = sqlite3.connect(db_path) + try: + return conn.execute("PRAGMA user_version").fetchone()[0] + finally: + conn.close() + + +def test_fresh_database_is_stamped_with_current_version(tmp_path): + registry.write(make_entry()) + assert _user_version(_db_path(tmp_path)) == schema.SCHEMA_VERSION + + +def test_legacy_unversioned_database_is_migrated_on_open(tmp_path): + """A database created before versioning existed has user_version 0 by + default. Opening it through the registry should bring it up to the + current version without losing data or erroring.""" + db_path = _db_path(tmp_path) + db_path.parent.mkdir(parents=True) + conn = sqlite3.connect(db_path) + try: + conn.execute(schema.SCHEMA) + conn.execute( + "INSERT INTO servers (key, state, pid, port, password, project_dir, " + "server_dir, started_at, claimed_at) VALUES " + "('legacykey00000000', 'running', 123, 4096, 'pw', '/tmp/p', NULL, " + "'2026-07-05T00:00:00+00:00', '2026-07-05T00:00:00+00:00')" + ) + conn.commit() + finally: + conn.close() + assert _user_version(db_path) == 0 + + entry = registry.read("legacykey00000000") + assert entry is not None + assert entry.pid == 123 + assert _user_version(db_path) == schema.SCHEMA_VERSION + + +def test_future_schema_version_raises(tmp_path): + """A database written by a newer opencode-runtime must not be silently + treated as compatible — it should fail loudly instead of risking data + loss or a confusing runtime error deeper in the stack.""" + db_path = _db_path(tmp_path) + db_path.parent.mkdir(parents=True) + conn = sqlite3.connect(db_path) + try: + conn.execute(schema.SCHEMA) + conn.execute(f"PRAGMA user_version = {schema.SCHEMA_VERSION + 1}") + conn.commit() + finally: + conn.close() + + with pytest.raises(RegistrySchemaError): + registry.read("doesnotexist") + + +def test_migrate_raises_on_missing_step(monkeypatch): + """If SCHEMA_VERSION is bumped without registering the matching migration + step, migrate() must fail loudly rather than silently skip a step.""" + monkeypatch.setattr(schema, "SCHEMA_VERSION", 2) + with pytest.raises(RegistrySchemaError): + schema.migrate(sqlite3.connect(":memory:"), current_version=0) + + +def test_migrate_raises_on_newer_version(): + with pytest.raises(RegistrySchemaError): + schema.migrate(sqlite3.connect(":memory:"), current_version=schema.SCHEMA_VERSION + 1) diff --git a/tests/test_server.py b/tests/test_server.py index 3c27caa..07f77a0 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -3,6 +3,7 @@ Tests for ServerManager — require the real opencode binary. """ +import asyncio import os import signal from pathlib import Path @@ -10,6 +11,7 @@ import pytest import opencode_runtime.registry as registry +from opencode_runtime.registry import ServerState from opencode_runtime.server import ( ServerManager, _ManagedServer, @@ -310,15 +312,18 @@ async def test_stale_registry_entry_cleaned_on_attach(self, tmp_path, monkeypatc from opencode_runtime.registry import RegistryEntry, now_iso key = _compute_runtime_key(None, None, tmp_path, None, {}) + timestamp = now_iso() registry.write( RegistryEntry( key=key, + state=ServerState.RUNNING, pid=99999999, port=54321, password="stale", project_dir=str(tmp_path), server_dir=None, - started_at=now_iso(), + started_at=timestamp, + claimed_at=timestamp, ) ) @@ -520,17 +525,144 @@ async def test_stop_returns_false_for_dead_process(self, tmp_path, monkeypatch): monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") key = _compute_runtime_key(None, None, tmp_path, None, {}) + timestamp = now_iso() registry.write( RegistryEntry( key=key, + state=ServerState.RUNNING, pid=99999999, port=54321, password="x", project_dir=str(tmp_path), server_dir=None, - started_at=now_iso(), + started_at=timestamp, + claimed_at=timestamp, ) ) was_alive = await ServerManager().stop(key) assert was_alive is False assert registry.read(key) is None # still cleaned up + + +class TestGetOrStartConcurrency: + """get_or_start's race fix: concurrent callers for the same key should + result in exactly one _start() call, with the rest attaching to it. + Uses a fake _start() so these don't need the real opencode binary.""" + + def _fake_start(self, key: str, tmp_path: Path, delay: float, call_counter: list[int]): + from opencode_runtime.registry import RegistryEntry, now_iso + + async def fake_start(_self, *, port, password, **kwargs): + call_counter.append(1) + await asyncio.sleep(delay) + timestamp = now_iso() + registry.write( + RegistryEntry( + key=key, + state=ServerState.RUNNING, + pid=os.getpid(), + port=port, + password=password, + project_dir=str(tmp_path), + server_dir=None, + started_at=timestamp, + claimed_at=timestamp, + ) + ) + return _ManagedServer(key=key, process=None, client=None, server_dir=None) # type: ignore[arg-type] + + return fake_start + + async def test_concurrent_get_or_start_calls_start_once(self, tmp_path, monkeypatch): + monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") + key = _compute_runtime_key(None, None, tmp_path, None, {}) + calls: list[int] = [] + monkeypatch.setattr(ServerManager, "_start", self._fake_start(key, tmp_path, 0.2, calls)) + manager = ServerManager() + + def request(): + return manager.get_or_start( + key=key, + project_dir=tmp_path, + server_dir=None, + materials=None, + config={}, + env={}, + ) + + results = await asyncio.gather(request(), request(), request()) + + assert len(calls) == 1 + assert all(isinstance(r, _ManagedServer) for r in results) + assert all(r.key == key for r in results) + + async def test_failed_start_deletes_claim_for_retry(self, tmp_path, monkeypatch): + monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") + key = _compute_runtime_key(None, None, tmp_path, None, {}) + calls: list[int] = [] + succeeding_start = self._fake_start(key, tmp_path, 0.0, calls) + + async def failing_once(_self, **kwargs): + if len(calls) == 0: + calls.append(1) + raise RuntimeError("boom") + return await succeeding_start(_self, **kwargs) + + monkeypatch.setattr(ServerManager, "_start", failing_once) + manager = ServerManager() + + with pytest.raises(RuntimeError): + await manager.get_or_start( + key=key, + project_dir=tmp_path, + server_dir=None, + materials=None, + config={}, + env={}, + ) + + # The failed attempt's except-block deleted its claim row, so a + # retry can proceed instead of finding a stuck 'starting' row. + assert registry.read(key) is None + + result = await manager.get_or_start( + key=key, + project_dir=tmp_path, + server_dir=None, + materials=None, + config={}, + env={}, + ) + assert isinstance(result, _ManagedServer) + + async def test_losing_caller_raises_when_starter_fails(self, tmp_path, monkeypatch): + """A caller waiting on someone else's claim should fail fast, not + hang for the full timeout, once that claim disappears.""" + monkeypatch.setattr(registry, "REGISTRY_DIR", tmp_path / "reg") + key = _compute_runtime_key(None, None, tmp_path, None, {}) + + async def always_fails(_self, **kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(ServerManager, "_start", always_fails) + manager = ServerManager() + + with pytest.raises(Exception): + await asyncio.gather( + manager.get_or_start( + key=key, + project_dir=tmp_path, + server_dir=None, + materials=None, + config={}, + env={}, + ), + manager.get_or_start( + key=key, + project_dir=tmp_path, + server_dir=None, + materials=None, + config={}, + env={}, + ), + ) diff --git a/tests/test_session.py b/tests/test_session.py index 75f8921..95a5fb5 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -18,9 +18,8 @@ async def runtime(tmp_path): project_dir=tmp_path, runtime_dir=tmp_path / "runtime", ) - await r.start() yield r - await r.stop() + await r.close() class TestSessionFactory: