From fdc77388e04e589876acc661862eed5d678c13eb Mon Sep 17 00:00:00 2001 From: ashish16052 Date: Fri, 24 Jul 2026 18:12:50 +0530 Subject: [PATCH 1/8] refactor: add OS process primitives module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Centralises all OS-process concerns — spawning, identity, signalling, and liveness — in a single new module so that registry.py (persistence) and server.py (orchestration) don't each carry their own copies. Public surface: spawn() asyncio subprocess as its own process group leader is_alive(pid) signal-0 liveness check start_time(pid) ps-reported start time (PID-reuse guard) is_same(pid, t) liveness + start-time identity check kill_group(pid) send signal to full process group terminate(proc) SIGTERM → wait 5s → SIGKILL wait_until_dead() poll until pid is gone or timeout --- src/opencode_runtime/process.py | 105 ++++++++++++++++++++++++++++++++ tests/test_process.py | 53 ++++++++++++++++ 2 files changed, 158 insertions(+) create mode 100644 src/opencode_runtime/process.py create mode 100644 tests/test_process.py diff --git a/src/opencode_runtime/process.py b/src/opencode_runtime/process.py new file mode 100644 index 0000000..55e5eaa --- /dev/null +++ b/src/opencode_runtime/process.py @@ -0,0 +1,105 @@ +""" +OS process primitives: spawning, identity, signaling, and liveness. + +Kept separate from registry.py (which only persists JSON state) and +server.py (which orchestrates OpenCode-specific startup and health) so +that everything OS-process-specific lives in one place — and so a future +non-local backend (Docker, remote) could replace just this module. +""" + +from __future__ import annotations + +import asyncio +import os +import signal +import subprocess +from typing import Any + + +async def spawn( + *args: str, cwd: str, env: dict[str, str], output: Any +) -> asyncio.subprocess.Process: + """Start a subprocess as its own process group leader. + + start_new_session=True means kill_group() reaches children the process + spawns too, not just the pid we hold. + """ + return await asyncio.create_subprocess_exec( + *args, + cwd=cwd, + env=env, + stdout=output, + stderr=output, + start_new_session=True, + ) + + +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 + except (ProcessLookupError, PermissionError): + return False + + +def start_time(pid: int | None) -> str | None: + """Return pid's process start time as reported by `ps`, or None if it's not running.""" + if pid is None: + return None + try: + result = subprocess.run( + ["ps", "-o", "lstart=", "-p", str(pid)], + capture_output=True, + text=True, + timeout=3.0, + ) + except (OSError, subprocess.SubprocessError): + return None + return result.stdout.strip() or None + + +def is_same(pid: int | None, started_at: str | None) -> bool: + """Return True if pid is alive and, when started_at is known, still matches it. + + started_at is None for entries written before this field existed — + those fall back to a plain liveness check. + """ + if not is_alive(pid): + return False + if started_at is None: + return True + return start_time(pid) == started_at + + +def kill_group(pid: int, sig: int) -> bool: + """Send sig to pid's process group. Returns False if it's already gone.""" + try: + os.killpg(os.getpgid(pid), sig) + return True + except (ProcessLookupError, PermissionError): + return False + + +async def terminate(process: asyncio.subprocess.Process) -> None: + """Terminate a process group gracefully, kill if it doesn't exit within 5s.""" + if process.returncode is not None: + return # already exited + if not kill_group(process.pid, signal.SIGTERM): + return # already dead + try: + await asyncio.wait_for(process.wait(), timeout=5.0) + except asyncio.TimeoutError: + kill_group(process.pid, signal.SIGKILL) + + +async def wait_until_dead(pid: int, started_at: str | None, timeout: float) -> bool: + """Poll until pid is confirmed gone, or timeout elapses.""" + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + if not await asyncio.to_thread(is_same, pid, started_at): + return True + await asyncio.sleep(0.1) + return not is_same(pid, started_at) diff --git a/tests/test_process.py b/tests/test_process.py new file mode 100644 index 0000000..596cc65 --- /dev/null +++ b/tests/test_process.py @@ -0,0 +1,53 @@ +"""Tests for the process module (OS process identity and liveness).""" + +import os + +import pytest + +from opencode_runtime import process + +pytestmark = pytest.mark.asyncio + + +async def test_is_alive_current_process(): + assert process.is_alive(os.getpid()) is True + + +async def test_is_alive_dead_pid(): + assert process.is_alive(99999999) is False + + +async def test_is_alive_none_pid(): + assert process.is_alive(None) is False + + +async def test_start_time_for_current_process(): + assert process.start_time(os.getpid()) + + +async def test_start_time_for_dead_pid(): + assert process.start_time(99999999) is None + + +async def test_start_time_for_none_pid(): + assert process.start_time(None) is None + + +async def test_is_same_true_for_matching_pid_and_start_time(): + started_at = process.start_time(os.getpid()) + assert process.is_same(os.getpid(), started_at) is True + + +async def test_is_same_false_for_mismatched_start_time(): + """A pid that's alive but whose start time doesn't match is a different + process generation — e.g. the original died and the pid was reused.""" + assert process.is_same(os.getpid(), "Mon Jan 1 00:00:00 1990") is False + + +async def test_is_same_false_for_dead_pid(): + assert process.is_same(99999999, "whatever") is False + + +async def test_is_same_falls_back_to_liveness_when_start_time_unknown(): + """Entries written before pid_start_time existed have it as None.""" + assert process.is_same(os.getpid(), None) is True From ff5a9451953646e84aea281f05878532159ef86d Mon Sep 17 00:00:00 2001 From: ashish16052 Date: Fri, 24 Jul 2026 18:14:57 +0530 Subject: [PATCH 2/8] refactor: use process module for spawn, signaling, liveness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire server.py to the new process module instead of carrying its own OS-process helpers: - process.spawn() replaces asyncio.create_subprocess_exec directly; start_new_session=True means kill_group() reaches every child the opencode binary spawns, not just the pid we hold - process.terminate() replaces the local _terminate_process() helper - process.is_alive() replaces registry.is_alive() and _is_process_alive() at every call site - process.kill_group() + process.wait_until_dead() replace the inline os.kill + poll loop in stop(); stop() now kills before deleting the registry entry, so the entry stays discoverable if the process can't be confirmed dead SQLite registry, ServerState, claimed_at, and last_used_at are unchanged — storage format is a separate concern from process handling. --- src/opencode_runtime/server.py | 88 +++++++++++++--------------------- 1 file changed, 32 insertions(+), 56 deletions(-) diff --git a/src/opencode_runtime/server.py b/src/opencode_runtime/server.py index e1edaca..9e9daf0 100644 --- a/src/opencode_runtime/server.py +++ b/src/opencode_runtime/server.py @@ -27,7 +27,7 @@ import httpx -from . import registry +from . import process, registry from .client import OpenCodeClient from .exceptions import ( OpenCodeNotFoundError, @@ -74,11 +74,6 @@ class ServerStatus: 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: @@ -165,23 +160,6 @@ def _prepare_dir( shutil.copy2(src, server_dir / src.name) -async def _terminate_process(process: asyncio.subprocess.Process) -> None: - """Terminate a process gracefully, kill if it doesn't exit within 5s.""" - if process.returncode is not None: - return # already exited - try: - process.terminate() - except ProcessLookupError: - return # already dead - try: - await asyncio.wait_for(process.wait(), timeout=5.0) - except asyncio.TimeoutError: - try: - process.kill() - except ProcessLookupError: - pass - - def _compute_runtime_key( workspace: str | None, user_id: str | None, @@ -240,7 +218,7 @@ def find(self, key: str) -> RegistryEntry | None: def is_alive(self, key: str) -> bool: """Return True if the server for key is running.""" entry = self.find(key) - return entry is not None and registry.is_alive(entry.pid) + return entry is not None and process.is_alive(entry.pid) def touch(self, key: str) -> None: """Update last_used_at timestamp for a server. Call after session creation.""" @@ -253,9 +231,7 @@ 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 - ) + process_alive = process.is_alive(entry.pid) if entry.state == ServerState.RUNNING else False health_ok = False if process_alive: client = OpenCodeClient( @@ -291,7 +267,7 @@ async def list_statuses(self, *, health_timeout: float = 1.0) -> list[ServerStat def list(self) -> list[tuple[RegistryEntry, bool]]: """Return all ready registry entries with their liveness status.""" return [ - (entry, registry.is_alive(entry.pid)) + (entry, process.is_alive(entry.pid)) for entry in registry.list_all() if entry.state == ServerState.RUNNING ] @@ -318,33 +294,34 @@ async def health(self, key: str) -> dict[str, Any]: async def stop(self, key: str) -> bool: """Kill the server for key and remove its registry entry. - Returns True if the process was alive and killed, False if it was - already dead or not found in the registry. + Returns True if the process was alive and was confirmed killed, + False if it was already dead, not found, or couldn't be confirmed + dead (in which case the entry is left in place, still discoverable, + rather than forgotten while the process may still be running). """ entry = registry.read(key) if entry is None: return False - registry.delete(key) - - if entry.pid is None or not registry.is_alive(entry.pid): + if not process.is_alive(entry.pid): + registry.delete(key) return False - try: - os.kill(entry.pid, signal.SIGTERM) - except ProcessLookupError: + assert entry.pid is not None + if not process.kill_group(entry.pid, signal.SIGTERM): + registry.delete(key) return False - # 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 registry.is_alive(entry.pid): - return True - await asyncio.sleep(0.1) - try: - os.kill(entry.pid, signal.SIGKILL) - except ProcessLookupError: - pass - return True + + if await process.wait_until_dead(entry.pid, None, timeout=5.0): + registry.delete(key) + return True + + process.kill_group(entry.pid, signal.SIGKILL) + if await process.wait_until_dead(entry.pid, None, timeout=2.0): + registry.delete(key) + return True + + return False async def stop_all(self) -> None: """Kill all servers tracked in the registry.""" @@ -376,7 +353,7 @@ async def get_or_start( """Return a client for the running server, starting one if needed.""" entry = registry.read(key) if entry is not None and entry.state == ServerState.RUNNING: - if registry.is_alive(entry.pid): + if process.is_alive(entry.pid): return self._attach(entry) # Stale — the process died after finishing startup. registry.delete(key) @@ -448,7 +425,7 @@ async def _wait_for_ready(self, key: str, timeout: float = 60.0) -> _ManagedServ 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): + if entry.state == ServerState.RUNNING and process.is_alive(entry.pid): return self._attach(entry) await asyncio.sleep(0.1) @@ -490,7 +467,7 @@ async def _start( else: output = asyncio.subprocess.DEVNULL - process = await asyncio.create_subprocess_exec( + proc = await process.spawn( "opencode", "serve", "--hostname", @@ -499,8 +476,7 @@ async def _start( str(port), cwd=str(project_dir), env=process_env, - stdout=output, - stderr=output, + output=output, ) client = OpenCodeClient( @@ -517,9 +493,9 @@ async def _start( ) try: - await _wait_healthy(poll_client, process=process) + await _wait_healthy(poll_client, process=proc) except Exception: - await _terminate_process(process) + await process.terminate(proc) raise timestamp = registry.now_iso() @@ -527,7 +503,7 @@ async def _start( RegistryEntry( key=key, state=ServerState.RUNNING, - pid=process.pid, + pid=proc.pid, port=port, password=password, project_dir=str(project_dir), @@ -543,7 +519,7 @@ async def _start( return _ManagedServer( key=key, - process=process, + process=proc, client=client, server_dir=server_dir, ) From 583a4474379697315b4ccaca88fb6471116d9cf3 Mon Sep 17 00:00:00 2001 From: ashish16052 Date: Fri, 24 Jul 2026 18:19:53 +0530 Subject: [PATCH 3/8] refactor: replace SQLite backend with atomic JSON files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Storage: - One JSON file per server at .json; atomic write via temp-then-rename so readers never see a partial write - Per-key lock file (O_CREAT|O_EXCL) used only for the two operations that read-then-conditionally-write: claim_starting (abandoned-claim reclaim) and delete_if_instance (generation- scoped delete); plain reads and writes are lock-free - schema.py and test_schema.py deleted — no schema, no migrations; missing fields default to None on read (_FIELD_NAMES filter) Generation fencing: - RegistryEntry gains instance_id and pid_start_time fields - delete_if_instance(key, instance_id) replaces bare delete() in stop() and the get_or_start() failure path, so a slow stop() can't delete a replacement server that started in the gap - pid_start_time recorded at spawn; process.is_same() uses it to guard against PID reuse (stop/wait_until_dead now pass it through) Exceptions: - RegistryBusyError added (lock wait timeout) - RegistrySchemaError removed (belonged to the SQLite schema system) ServerState / claimed_at / last_used_at are still present in RegistryEntry — they are removed in the following commits. --- src/opencode_runtime/exceptions.py | 8 +- src/opencode_runtime/registry.py | 199 ++++++++++++++++------------- src/opencode_runtime/schema.py | 79 ------------ src/opencode_runtime/server.py | 72 +++++++---- tests/test_cli.py | 16 ++- tests/test_multi_tenant.py | 6 +- tests/test_registry.py | 120 +++++++++++++---- tests/test_runtime.py | 4 +- tests/test_schema.py | 106 --------------- tests/test_server.py | 22 ++-- 10 files changed, 278 insertions(+), 354 deletions(-) delete mode 100644 src/opencode_runtime/schema.py delete mode 100644 tests/test_schema.py diff --git a/src/opencode_runtime/exceptions.py b/src/opencode_runtime/exceptions.py index ab51414..af5e2a7 100644 --- a/src/opencode_runtime/exceptions.py +++ b/src/opencode_runtime/exceptions.py @@ -14,9 +14,5 @@ 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. - """ +class RegistryBusyError(OpenCodeRuntimeError): + """Raised when a registry entry's lock can't be acquired.""" diff --git a/src/opencode_runtime/registry.py b/src/opencode_runtime/registry.py index e82489b..08b4043 100644 --- a/src/opencode_runtime/registry.py +++ b/src/opencode_runtime/registry.py @@ -1,49 +1,51 @@ """ Registry for tracking OpenCode instance processes. -Every running instance is a row in a SQLite database at: - ~/.opencode-runtime/servers/registry.db +Every running instance is one file at: + ~/.opencode-runtime/servers/.json Used by both the CLI (opencode-runtime serve/ps/stop) and the library (OpenCodeRuntime) — the registry is the shared source of truth for all running instances regardless of how they were started. + +This module only persists and locks JSON state. It has no opinion on +whether a pid is actually alive (see process.py) or whether OpenCode is +healthy (see server.py). """ from __future__ import annotations +import json import os -import sqlite3 +import time +import uuid from contextlib import contextmanager -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, fields from datetime import datetime, timedelta, timezone from enum import Enum from pathlib import Path from typing import Generator -from .schema import SCHEMA, SCHEMA_VERSION, migrate +from .exceptions import RegistryBusyError 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. +# A claim (pid still None) older than this is treated as abandoned — its +# starter crashed (SIGKILL, host reboot) before finishing _start(). Comfortably +# above _wait_healthy's default 60s startup timeout. _START_LEASE_SECONDS = 90 +# A lock file older than this is treated as abandoned — its holder crashed +# mid read-check-write. Locks here are only ever held across a handful of +# filesystem syscalls, so a few seconds of staleness tolerance is generous. +_LOCK_STALE_SECONDS = 5 -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). - """ +class ServerState(str, Enum): + """Server lifecycle state (kept for server.py compatibility).""" STARTING = "starting" RUNNING = "running" @@ -55,7 +57,11 @@ class ServerState(str, Enum): class RegistryEntry: """A server entry in the registry. - state: ServerState enum value. Display status is derived from state + observed health. + pid is None while the key is claimed but the process hasn't been + spawned yet. instance_id identifies this process generation, so a + delete can be scoped to "this generation only" via delete_if_instance(). + pid_start_time guards against pid having been reused by an unrelated + process — see process.is_same(). """ key: str @@ -71,111 +77,128 @@ class RegistryEntry: user_id: str | None = None last_used_at: str | None = None runtime_version: str | None = None + instance_id: str | None = None + pid_start_time: str | None = None -_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) -""" +_FIELD_NAMES = {f.name for f in fields(RegistryEntry)} -_UPSERT = _INSERT.replace("INSERT", "INSERT OR REPLACE", 1) + +def _path(key: str) -> Path: + return REGISTRY_DIR / f"{key}.json" + + +def _deserialize(text: str) -> RegistryEntry: + data = json.loads(text) + return RegistryEntry(**{k: v for k, v in data.items() if k in _FIELD_NAMES}) @contextmanager -def _connect() -> Generator[sqlite3.Connection, None, None]: - """Open the registry database, creating it on first use. +def _locked(key: str, wait_seconds: float = 1.0) -> Generator[None, None, None]: + """Hold a short-lived exclusive lock on key for a read-check-write step. - 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. + Legitimate holders never keep this past a few syscalls, so genuine + contention clears in milliseconds — wait_seconds is generous headroom, + not a real operation budget. """ REGISTRY_DIR.mkdir(parents=True, exist_ok=True) - db_path = REGISTRY_DIR / "registry.db" - is_new = not db_path.exists() - conn = sqlite3.connect(db_path, timeout=5.0) + lock_path = REGISTRY_DIR / f"{key}.lock" + deadline = time.monotonic() + wait_seconds + while True: + try: + os.close(os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)) + break + except FileExistsError: + pass + try: + stale = time.time() - lock_path.stat().st_mtime > _LOCK_STALE_SECONDS + except FileNotFoundError: + stale = False + if stale: + lock_path.unlink(missing_ok=True) # holder crashed; reclaim + continue + if time.monotonic() >= deadline: + raise RegistryBusyError(f"registry entry {key!r} is locked by another operation") + time.sleep(0.01) 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() + yield 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()}) + lock_path.unlink(missing_ok=True) def write(entry: RegistryEntry) -> None: """Write (insert or replace) a registry entry.""" - with _connect() as conn: - conn.execute(_UPSERT, asdict(entry)) + REGISTRY_DIR.mkdir(parents=True, exist_ok=True) + tmp = REGISTRY_DIR / f".{entry.key}.{uuid.uuid4().hex}.tmp" + tmp.write_text(json.dumps(asdict(entry)), encoding="utf-8") + tmp.chmod(0o600) + os.replace(tmp, _path(entry.key)) # atomic — readers never see a partial write def read(key: str) -> RegistryEntry | None: """Read a registry entry by key. Returns None if not found.""" - with _connect() as conn: - row = conn.execute("SELECT * FROM servers WHERE key = ?", (key,)).fetchone() - return _entry(row) if row is not None else None + try: + return _deserialize(_path(key).read_text(encoding="utf-8")) + except FileNotFoundError: + return None def delete(key: str) -> None: """Remove a registry entry. No-op if not found.""" - with _connect() as conn: - conn.execute("DELETE FROM servers WHERE key = ?", (key,)) + _path(key).unlink(missing_ok=True) def list_all() -> list[RegistryEntry]: """Return all registry entries.""" - with _connect() as conn: - rows = conn.execute("SELECT * FROM servers").fetchall() - return [_entry(row) for row in rows] + if not REGISTRY_DIR.exists(): + return [] + entries = [] + for path in REGISTRY_DIR.glob("*.json"): + try: + entries.append(_deserialize(path.read_text(encoding="utf-8"))) + except (FileNotFoundError, json.JSONDecodeError): + continue # deleted mid-scan, or a crash left a partial write + return entries def claim_starting(entry: RegistryEntry) -> bool: - """Atomically insert a 'starting' row for entry.key. + """Claim entry.key for a new start attempt. - 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. + Returns True if this call claimed the key, False if a live claim or a + running server already occupies it. A claim (pid still None) older + than _START_LEASE_SECONDS is treated as abandoned and reclaimed. """ - 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)) + with _locked(entry.key): + existing = read(entry.key) + if existing is not None: + if existing.pid is not None: + return False + claimed_at = datetime.fromisoformat(existing.claimed_at) + if datetime.now(timezone.utc) - claimed_at < timedelta(seconds=_START_LEASE_SECONDS): + return False + write(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) +def write_if_instance(entry: RegistryEntry) -> bool: + """Write entry only if its generation still owns the registry key.""" + with _locked(entry.key): + current = read(entry.key) + if current is None or current.instance_id != entry.instance_id: + return False + + write(entry) + return True + + +def delete_if_instance(key: str, instance_id: str | None) -> bool: + """Delete the entry for key iff its instance_id matches. Returns whether deleted.""" + with _locked(key): + entry = read(key) + if entry is None or entry.instance_id != instance_id: + return False + delete(key) return True - except (ProcessLookupError, PermissionError): - return False def now_iso() -> str: diff --git a/src/opencode_runtime/schema.py b/src/opencode_runtime/schema.py deleted file mode 100644 index 5717675..0000000 --- a/src/opencode_runtime/schema.py +++ /dev/null @@ -1,79 +0,0 @@ -""" -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 9e9daf0..c7a7c03 100644 --- a/src/opencode_runtime/server.py +++ b/src/opencode_runtime/server.py @@ -15,6 +15,7 @@ import shutil import signal import socket +import uuid from dataclasses import dataclass from enum import Enum from pathlib import Path @@ -303,22 +304,25 @@ async def stop(self, key: str) -> bool: if entry is None: return False + async def forget() -> None: + await asyncio.to_thread(registry.delete_if_instance, key, entry.instance_id) + if not process.is_alive(entry.pid): - registry.delete(key) + await forget() return False assert entry.pid is not None if not process.kill_group(entry.pid, signal.SIGTERM): - registry.delete(key) + await forget() return False - if await process.wait_until_dead(entry.pid, None, timeout=5.0): - registry.delete(key) + if await process.wait_until_dead(entry.pid, entry.pid_start_time, timeout=5.0): + await forget() return True process.kill_group(entry.pid, signal.SIGKILL) - if await process.wait_until_dead(entry.pid, None, timeout=2.0): - registry.delete(key) + if await process.wait_until_dead(entry.pid, entry.pid_start_time, timeout=2.0): + await forget() return True return False @@ -355,14 +359,16 @@ async def get_or_start( if entry is not None and entry.state == ServerState.RUNNING: if process.is_alive(entry.pid): return self._attach(entry) - # Stale — the process died after finishing startup. - registry.delete(key) + # Stale — the process died after finishing startup. Scoped to + # this generation so a concurrent fresh claim isn't clobbered. + registry.delete_if_instance(key, entry.instance_id) - # port/password are generated here (not inside _start) because a - # 'starting' claim row needs both up front — pid is the only field + # port/password/instance_id are generated here (not inside _start) + # because a claim needs them all 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) + instance_id = uuid.uuid4().hex timestamp = registry.now_iso() claimed = registry.claim_starting( @@ -378,6 +384,7 @@ async def get_or_start( claimed_at=timestamp, workspace=workspace, user_id=user_id, + instance_id=instance_id, ) ) if not claimed: @@ -395,11 +402,13 @@ async def get_or_start( env=env, port=port, password=password, + instance_id=instance_id, + started_at=timestamp, workspace=workspace, user_id=user_id, ) except Exception: - registry.delete(key) + registry.delete_if_instance(key, instance_id) raise self._owned.add(key) return server @@ -444,6 +453,8 @@ async def _start( env: dict[str, str], port: int, password: str, + instance_id: str, + started_at: str, workspace: str | None = None, user_id: str | None = None, ) -> _ManagedServer: @@ -498,25 +509,30 @@ async def _start( await process.terminate(proc) raise - timestamp = registry.now_iso() - registry.write( - RegistryEntry( - key=key, - state=ServerState.RUNNING, - pid=proc.pid, - 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, - last_used_at=timestamp, - runtime_version=version("opencode-runtime"), - workspace=workspace, - user_id=user_id, - ) + ready_entry = RegistryEntry( + key=key, + state=ServerState.RUNNING, + pid=proc.pid, + port=port, + password=password, + project_dir=str(project_dir), + server_dir=str(server_dir) if server_dir else None, + started_at=started_at, + claimed_at=started_at, + last_used_at=started_at, + runtime_version=version("opencode-runtime"), + workspace=workspace, + user_id=user_id, + instance_id=instance_id, + pid_start_time=process.start_time(proc.pid), ) + if not registry.write_if_instance(ready_entry): + await process.terminate(proc) + raise OpenCodeServerError( + f"startup claim for server {key!r} was replaced before startup completed" + ) + return _ManagedServer( key=key, process=proc, diff --git a/tests/test_cli.py b/tests/test_cli.py index 648c1ea..4cf631f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -15,6 +15,7 @@ import pytest import opencode_runtime.registry as registry +from opencode_runtime import process from opencode_runtime.cli import cmd_health, cmd_ps, cmd_serve, cmd_stop, cmd_stop_all from opencode_runtime.registry import RegistryEntry, ServerState @@ -118,9 +119,10 @@ def test_stop_dead_process_warns_and_deletes(capsys): 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".""" + """A claim entry (pid still None, e.g. an orphaned claim) must be + stoppable by key, not just entries with a pid — cmd_stop used to look + it up via find(), which filters to entries with a pid and reported + claims as "not found".""" registry.write(make_entry(state=ServerState.STARTING, pid=None)) cmd_stop(ns(key="abc123def456abcd")) out = capsys.readouterr().out @@ -135,7 +137,7 @@ def _is_truly_dead(pid: int) -> bool: exists in the process table but the process has already exited. We treat zombies as dead since we are not the parent and cannot reap them. """ - if not registry.is_alive(pid): + if not process.is_alive(pid): return True try: with open(f"/proc/{pid}/status") as f: @@ -161,7 +163,7 @@ def test_stop_live_server(tmp_path): entries = registry.list_all() assert len(entries) == 1 entry = entries[0] - assert registry.is_alive(entry.pid) + assert process.is_alive(entry.pid) cmd_stop(ns(key=entry.key)) @@ -251,7 +253,7 @@ def test_serve_starts_server(tmp_path): entries = registry.list_all() assert len(entries) == 1 - assert registry.is_alive(entries[0].pid) + assert process.is_alive(entries[0].pid) # cleanup cmd_stop(ns(key=entries[0].key)) @@ -298,7 +300,7 @@ def test_serve_stale_entry_cleaned_and_restarted(tmp_path): entries = registry.list_all() assert len(entries) == 1 - assert registry.is_alive(entries[0].pid) + assert process.is_alive(entries[0].pid) # cleanup cmd_stop(ns(key=entries[0].key)) diff --git a/tests/test_multi_tenant.py b/tests/test_multi_tenant.py index 14d1928..89eecb4 100644 --- a/tests/test_multi_tenant.py +++ b/tests/test_multi_tenant.py @@ -11,7 +11,7 @@ import pytest import opencode_runtime.registry as registry -from opencode_runtime import OpenCodeRuntime +from opencode_runtime import OpenCodeRuntime, process pytestmark = pytest.mark.asyncio @@ -104,6 +104,6 @@ async def test_stop_single_tenant(self, tmp_path, monkeypatch): await r._server_manager.stop(acme_key) assert registry.read(acme_key) is None - assert not registry.is_alive(acme_entry.pid) + assert not process.is_alive(acme_entry.pid) assert registry.read(beta_key) is not None - assert registry.is_alive(beta_entry.pid) + assert process.is_alive(beta_entry.pid) diff --git a/tests/test_registry.py b/tests/test_registry.py index 6205520..6e6023f 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -1,13 +1,16 @@ """Tests for the registry module.""" import asyncio +import json import os import stat +import time from concurrent.futures import ThreadPoolExecutor import pytest import opencode_runtime.registry as registry +from opencode_runtime.exceptions import RegistryBusyError from opencode_runtime.registry import RegistryEntry, ServerState pytestmark = pytest.mark.asyncio @@ -36,13 +39,17 @@ def make_entry(**kwargs: object) -> RegistryEntry: def make_claim(**kwargs: object) -> RegistryEntry: - """A 'starting' entry, as claim_starting() expects — pid is unknown yet. + """A claim entry, as claim_starting() expects — pid is unknown yet. - claimed_at defaults to now (not make_entry()'s fixed placeholder date), + started_at/claimed_at default to now (not make_entry()'s fixed placeholder date), since claim_starting()'s lease check compares it against the real clock. """ + now = registry.now_iso() defaults: dict[str, object] = dict( - pid=None, state=ServerState.STARTING, claimed_at=registry.now_iso() + state=ServerState.STARTING, + pid=None, + started_at=now, + claimed_at=now, ) defaults.update(kwargs) return make_entry(**defaults) @@ -84,13 +91,12 @@ async def test_write_twice_replaces_entry(): async def test_write_with_null_pid(): - """A 'starting' row has no pid yet.""" - entry = make_entry(state=ServerState.STARTING, pid=None) + """A claim entry has no pid yet.""" + entry = make_entry(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 # --------------------------------------------------------------------------- @@ -129,7 +135,6 @@ async def test_delete_is_noop_for_missing_key(): async def test_list_all_empty_when_no_registry_dir(): - # No db file created yet assert registry.list_all() == [] @@ -147,11 +152,12 @@ async def test_list_all_returns_all_entries(): # --------------------------------------------------------------------------- -async def test_db_file_created_with_permissions_600(): - registry.write(make_entry()) - db_path = registry.REGISTRY_DIR / "registry.db" - assert db_path.exists() - mode = stat.S_IMODE(db_path.stat().st_mode) +async def test_entry_file_created_with_permissions_600(): + entry = make_entry() + registry.write(entry) + path = registry.REGISTRY_DIR / f"{entry.key}.json" + assert path.exists() + mode = stat.S_IMODE(path.stat().st_mode) assert mode == 0o600 @@ -165,7 +171,6 @@ async def test_claim_starting_succeeds_for_new_key(): 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 @@ -211,29 +216,96 @@ async def test_claim_starting_does_not_reclaim_before_lease_expires(): 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 +async def test_claim_starting_does_not_reclaim_entry_with_pid(): + """An entry that already has a pid isn't touched by the lease logic, + no matter how old its started_at is.""" + entry = make_entry() # has a pid registry.write(entry) 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.pid == entry.pid assert result.port == entry.port # --------------------------------------------------------------------------- -# is_alive +# delete_if_instance +# --------------------------------------------------------------------------- + + +async def test_delete_if_instance_deletes_on_match(): + entry = make_entry(instance_id="gen-1") + registry.write(entry) + assert registry.delete_if_instance(entry.key, "gen-1") is True + assert registry.read(entry.key) is None + + +async def test_delete_if_instance_leaves_mismatched_generation(): + entry = make_entry(instance_id="gen-2") + registry.write(entry) + assert registry.delete_if_instance(entry.key, "gen-1") is False + assert registry.read(entry.key) is not None + + +async def test_delete_if_instance_false_for_missing_key(): + assert registry.delete_if_instance("doesnotexist", "gen-1") is False + + +async def test_delete_if_instance_reclaims_stale_lock(): + """A lock file left behind by a crashed holder doesn't wedge the registry.""" + entry = make_entry(instance_id="gen-1") + registry.write(entry) + lock_path = registry.REGISTRY_DIR / f"{entry.key}.lock" + lock_path.write_text("", encoding="utf-8") + stale = time.time() - registry._LOCK_STALE_SECONDS - 1 + os.utime(lock_path, (stale, stale)) + + assert registry.delete_if_instance(entry.key, "gen-1") is True + assert registry.read(entry.key) is None + + +async def test_delete_if_instance_raises_when_genuinely_locked(): + entry = make_entry(instance_id="gen-1") + registry.write(entry) + lock_path = registry.REGISTRY_DIR / f"{entry.key}.lock" + lock_path.write_text("", encoding="utf-8") + + with pytest.raises(RegistryBusyError): + registry.delete_if_instance(entry.key, "gen-1") + + +# --------------------------------------------------------------------------- +# forward/backward compatibility # --------------------------------------------------------------------------- -async def test_is_alive_current_process(): - assert registry.is_alive(os.getpid()) is True +async def test_read_tolerates_unknown_fields(): + """A file written by a newer version can carry fields this version + doesn't know about — they should be ignored, not raise.""" + entry = make_entry() + registry.write(entry) + path = registry.REGISTRY_DIR / f"{entry.key}.json" + data = json.loads(path.read_text(encoding="utf-8")) + data["from_a_future_version"] = "whatever" + path.write_text(json.dumps(data), encoding="utf-8") + result = registry.read(entry.key) + assert result is not None + assert result.key == entry.key -async def test_is_alive_dead_pid(): - assert registry.is_alive(99999999) is False +async def test_read_defaults_fields_missing_from_older_writer(): + """A file written before instance_id/pid_start_time existed should load + fine, with those fields defaulting to None.""" + entry = make_entry() + registry.write(entry) + path = registry.REGISTRY_DIR / f"{entry.key}.json" + data = json.loads(path.read_text(encoding="utf-8")) + del data["instance_id"] + del data["pid_start_time"] + path.write_text(json.dumps(data), encoding="utf-8") -async def test_is_alive_none_pid(): - assert registry.is_alive(None) is False + result = registry.read(entry.key) + assert result is not None + assert result.instance_id is None + assert result.pid_start_time is None diff --git a/tests/test_runtime.py b/tests/test_runtime.py index cce5852..f0c1a6f 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -9,7 +9,7 @@ import pytest import opencode_runtime.registry as registry -from opencode_runtime import OpenCodeRuntime +from opencode_runtime import OpenCodeRuntime, process pytestmark = pytest.mark.asyncio @@ -45,7 +45,7 @@ async def test_server_running_after_start(self, tmp_path, monkeypatch): await r.session() entries = registry.list_all() assert len(entries) == 1 - assert registry.is_alive(entries[0].pid) + assert process.is_alive(entries[0].pid) await r.close() async def test_client_reachable_after_start(self, runtime): diff --git a/tests/test_schema.py b/tests/test_schema.py deleted file mode 100644 index db9e63a..0000000 --- a/tests/test_schema.py +++ /dev/null @@ -1,106 +0,0 @@ -"""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 07f77a0..7c79a13 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -11,6 +11,7 @@ import pytest import opencode_runtime.registry as registry +from opencode_runtime import process from opencode_runtime.registry import ServerState from opencode_runtime.server import ( ServerManager, @@ -97,7 +98,7 @@ async def test_get_or_start_starts_server(self, tmp_path): ) entry = registry.read(key) assert entry is not None - assert registry.is_alive(entry.pid) + assert process.is_alive(entry.pid) assert server.client is not None await manager.stop_all() @@ -176,8 +177,8 @@ async def test_stop_all_terminates_all(self, tmp_path): await manager.stop_all() - assert not registry.is_alive(e1_before.pid) - assert not registry.is_alive(e2_before.pid) + assert not process.is_alive(e1_before.pid) + assert not process.is_alive(e2_before.pid) assert registry.read(k1) is None assert registry.read(k2) is None @@ -226,9 +227,9 @@ async def test_stop_single_server(self, tmp_path): await manager.stop(k1) assert registry.read(k1) is None - assert not registry.is_alive(e1.pid) + assert not process.is_alive(e1.pid) assert registry.read(k2) is not None # still running - assert registry.is_alive(e2.pid) + assert process.is_alive(e2.pid) await manager.stop_all() async def test_stop_nonexistent_key_is_noop(self, tmp_path): @@ -252,7 +253,7 @@ async def test_start_writes_registry_entry(self, tmp_path, monkeypatch): ) entry = registry.read(key) assert entry is not None - assert registry.is_alive(entry.pid) + assert process.is_alive(entry.pid) assert entry.port == int(server.client.base_url.split(":")[-1]) await manager.stop_all() @@ -301,7 +302,7 @@ async def test_attaches_to_alive_registry_entry(self, tmp_path, monkeypatch): # manager2 stop kills the process and deletes registry await manager2.stop(key) assert registry.read(key) is None - assert not registry.is_alive(e1.pid) + assert not process.is_alive(e1.pid) # manager1 stop is now a no-op (already gone) await manager1.stop_all() @@ -339,7 +340,7 @@ async def test_stale_registry_entry_cleaned_on_attach(self, tmp_path, monkeypatc # Fresh server was spawned — registry has a new alive entry entry = registry.read(key) assert entry is not None - assert registry.is_alive(entry.pid) + assert process.is_alive(entry.pid) assert entry.pid != 99999999 assert entry.port == int(server.client.base_url.split(":")[-1]) await manager.stop_all() @@ -398,7 +399,7 @@ async def test_server_restarted_after_external_kill(self, tmp_path, monkeypatch) new_entry = registry.read(key) assert new_entry is not None assert new_entry.pid != old_pid - assert registry.is_alive(new_entry.pid) + assert process.is_alive(new_entry.pid) await manager.stop_all() @@ -468,8 +469,7 @@ async def test_list_returns_entries_with_liveness(self, tmp_path, monkeypatch): ) entries = manager.list() assert len(entries) == 2 - assert all(alive is True for _, alive in entries) - keys = {e.key for e, _ in entries} + keys = {e.key for e in entries} assert k1 in keys and k2 in keys await manager.stop_all() From 6e81bf59d370077572d163f5c4f3a4a8b7ea8488 Mon Sep 17 00:00:00 2001 From: ashish16052 Date: Fri, 24 Jul 2026 18:21:24 +0530 Subject: [PATCH 4/8] refactor: remove idle-time tracking (touch/last_used_at) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit touch() / last_used_at exist to support idle-based cleanup, but that cleanup logic doesn't exist yet — the field was written and read but never acted on. Removing it now keeps the data model honest and shrinks the write surface. - RegistryEntry.last_used_at field removed - ServerManager.touch() removed - runtime.py no longer calls touch() after get_or_start() - cli.py cmd_inspect() no longer shows a "Last used" row --- src/opencode_runtime/cli.py | 31 ------------------------------- src/opencode_runtime/registry.py | 1 - src/opencode_runtime/runtime.py | 2 -- src/opencode_runtime/server.py | 8 -------- 4 files changed, 42 deletions(-) diff --git a/src/opencode_runtime/cli.py b/src/opencode_runtime/cli.py index c151f5f..a256e69 100644 --- a/src/opencode_runtime/cli.py +++ b/src/opencode_runtime/cli.py @@ -59,20 +59,6 @@ def _home(path: str) -> str: return path -def _uptime(started_at: str, alive: bool) -> str: - try: - mins = max( - 0, - int( - (datetime.now(timezone.utc) - datetime.fromisoformat(started_at)).total_seconds() - // 60 - ), - ) - except Exception: - return "?" - return f"Up {mins}m" if alive else f"Dead {mins}m" - - def _row(label: str, value: str) -> None: print(f" {_cyan(f'{label:<9}')} {value}") @@ -334,22 +320,6 @@ def cmd_inspect(args: argparse.Namespace) -> None: 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)) @@ -361,7 +331,6 @@ def cmd_inspect(args: argparse.Namespace) -> None: _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: diff --git a/src/opencode_runtime/registry.py b/src/opencode_runtime/registry.py index 08b4043..2d54fc5 100644 --- a/src/opencode_runtime/registry.py +++ b/src/opencode_runtime/registry.py @@ -75,7 +75,6 @@ class RegistryEntry: 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 instance_id: str | None = None pid_start_time: str | None = None diff --git a/src/opencode_runtime/runtime.py b/src/opencode_runtime/runtime.py index 032ccf4..0b4d3af 100644 --- a/src/opencode_runtime/runtime.py +++ b/src/opencode_runtime/runtime.py @@ -128,8 +128,6 @@ 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/server.py b/src/opencode_runtime/server.py index c7a7c03..6163b83 100644 --- a/src/opencode_runtime/server.py +++ b/src/opencode_runtime/server.py @@ -221,13 +221,6 @@ def is_alive(self, key: str) -> bool: entry = self.find(key) return entry is not None and process.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: @@ -519,7 +512,6 @@ async def _start( server_dir=str(server_dir) if server_dir else None, started_at=started_at, claimed_at=started_at, - last_used_at=started_at, runtime_version=version("opencode-runtime"), workspace=workspace, user_id=user_id, From b9b679f6ece9e5995a8024a1daecda109aa40b8f Mon Sep 17 00:00:00 2001 From: ashish16052 Date: Fri, 24 Jul 2026 18:22:55 +0530 Subject: [PATCH 5/8] refactor: derive server status from OS + health MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously server state was a persisted field (ServerState: STARTING / RUNNING / STOPPING / FAILED) that drove a DisplayStatus (6 values) computed from state + liveness + health. Now status is fully derived at read time from two OS observations: - process.is_same(pid, pid_start_time) — is the pid alive and the same process generation we recorded? - /global/health HTTP check — is opencode responding? RuntimeStatus (running / unhealthy / stale) replaces ServerState + DisplayStatus. A pid-less entry (still starting) has status=None. Concrete changes: - RegistryEntry drops the state field entirely - ServerState and DisplayStatus enums removed from server.py - _compute_display_status() removed - ServerStatus.display → ServerStatus.status (RuntimeStatus | None) - cli.py status icons/colors keyed on RuntimeStatus | None - cmd_health() branches on RuntimeStatus instead of DisplayStatus - ServerManager.find() now checks pid is not None instead of state == RUNNING; get_or_start() / _wait_for_ready() likewise --- src/opencode_runtime/cli.py | 65 +++++++++++----------- src/opencode_runtime/registry.py | 16 +----- src/opencode_runtime/server.py | 92 ++++++++++++-------------------- tests/test_cli.py | 6 +-- tests/test_registry.py | 14 ++--- tests/test_server.py | 7 --- 6 files changed, 70 insertions(+), 130 deletions(-) diff --git a/src/opencode_runtime/cli.py b/src/opencode_runtime/cli.py index a256e69..3a959e0 100644 --- a/src/opencode_runtime/cli.py +++ b/src/opencode_runtime/cli.py @@ -17,7 +17,7 @@ from datetime import datetime, timezone from pathlib import Path -from .server import DisplayStatus, ServerManager, _compute_runtime_key +from .server import RuntimeStatus, ServerManager, _compute_runtime_key # --------------------------------------------------------------------------- # ANSI @@ -63,27 +63,27 @@ 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: "✗", +# None means "claimed but not yet spawned" (ServerStatus.status while pid is None). +_STATUS_ICONS: dict[RuntimeStatus | None, str] = { + None: "◐", + RuntimeStatus.RUNNING: "●", + RuntimeStatus.UNHEALTHY: "▲", + RuntimeStatus.STALE: "○", } _STATUS_COLORS = { - DisplayStatus.RUNNING: _green, - DisplayStatus.STARTING: _yellow, - DisplayStatus.UNHEALTHY: _red, - DisplayStatus.STALE: _dim, - DisplayStatus.FAILED: _red, + None: _yellow, + RuntimeStatus.RUNNING: _green, + RuntimeStatus.UNHEALTHY: _red, + RuntimeStatus.STALE: _dim, } -def _status_display(status: DisplayStatus) -> str: - """Render a ServerStatus.display value as a coloured icon + label.""" +def _status_display(status: RuntimeStatus | None) -> str: + """Render a ServerStatus.status value as a coloured icon + label.""" icon = _STATUS_ICONS.get(status, "?") color = _STATUS_COLORS.get(status, _dim) - return color(f"{icon} {status.value}") + label = status.value if status is not None else "starting" + return color(f"{icon} {label}") # --------------------------------------------------------------------------- @@ -183,10 +183,10 @@ def cmd_ps(_args: argparse.Namespace) -> None: for st in statuses: e = st.entry - status_plain = f"{_STATUS_ICONS.get(st.display, '?')} {st.display.value}" - status_coloured = _status_display(st.display) + label = st.status.value if st.status is not None else "starting" + status_plain = f"{_STATUS_ICONS.get(st.status, '?')} {label}" + status_coloured = _status_display(st.status) - # Compute uptime try: started = datetime.fromisoformat(e.started_at) uptime_secs = int((datetime.now(timezone.utc) - started).total_seconds()) @@ -247,7 +247,7 @@ def cmd_stop_all(_args: argparse.Namespace) -> None: asyncio.run(manager.stop_all()) print(f"{_green(f'✓ Stopped {len(entries)} server(s)')}\n") - for e, _ in entries: + for e in entries: print(f" {_dim(e.key)} {_dim(f'pid {e.pid}')}") @@ -263,7 +263,14 @@ def cmd_health(args: argparse.Namespace) -> None: sys.exit(_red(f"✗ ID {args.key!r} not found in registry")) entry = st.entry - if st.display == DisplayStatus.RUNNING: + if st.status is None: + try: + claimed = datetime.fromisoformat(entry.started_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.status == RuntimeStatus.RUNNING: try: result = asyncio.run(manager.health(args.key)) version = result.get("version") @@ -274,25 +281,14 @@ def cmd_health(args: argparse.Namespace) -> None: ) 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: + elif st.status == RuntimeStatus.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}")) + sys.exit(_red(f"✗ stale: registry entry exists but pid {entry.pid} is not running")) # --------------------------------------------------------------------------- @@ -307,7 +303,6 @@ def cmd_inspect(args: argparse.Namespace) -> None: sys.exit(_red(f"✗ ID {args.key!r} not found in registry")) entry = st.entry - # Compute uptime try: started = datetime.fromisoformat(entry.started_at) uptime_secs = int((datetime.now(timezone.utc) - started).total_seconds()) @@ -322,7 +317,7 @@ def cmd_inspect(args: argparse.Namespace) -> None: print() _row("ID", entry.key) - _row("Status", _status_display(st.display)) + _row("Status", _status_display(st.status)) _row("Project", _home(entry.project_dir)) if entry.workspace: _row("Workspace", entry.workspace) diff --git a/src/opencode_runtime/registry.py b/src/opencode_runtime/registry.py index 2d54fc5..f345c7c 100644 --- a/src/opencode_runtime/registry.py +++ b/src/opencode_runtime/registry.py @@ -22,7 +22,6 @@ from contextlib import contextmanager from dataclasses import asdict, dataclass, fields from datetime import datetime, timedelta, timezone -from enum import Enum from pathlib import Path from typing import Generator @@ -44,15 +43,6 @@ _LOCK_STALE_SECONDS = 5 -class ServerState(str, Enum): - """Server lifecycle state (kept for server.py compatibility).""" - - STARTING = "starting" - RUNNING = "running" - STOPPING = "stopping" - FAILED = "failed" - - @dataclass class RegistryEntry: """A server entry in the registry. @@ -65,14 +55,12 @@ class RegistryEntry: """ key: str - 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" + started_at: str # ISO-8601; when this generation was claimed workspace: str | None = None user_id: str | None = None runtime_version: str | None = None @@ -172,7 +160,7 @@ def claim_starting(entry: RegistryEntry) -> bool: if existing is not None: if existing.pid is not None: return False - claimed_at = datetime.fromisoformat(existing.claimed_at) + claimed_at = datetime.fromisoformat(existing.started_at) if datetime.now(timezone.utc) - claimed_at < timedelta(seconds=_START_LEASE_SECONDS): return False write(entry) diff --git a/src/opencode_runtime/server.py b/src/opencode_runtime/server.py index 6163b83..6ccac06 100644 --- a/src/opencode_runtime/server.py +++ b/src/opencode_runtime/server.py @@ -36,7 +36,7 @@ OpenCodeServerError, OpenCodeTimeoutError, ) -from .registry import RegistryEntry, ServerState +from .registry import RegistryEntry # One or more paths to overlay into the server dir before startup. # Module-level alias: inside ServerManager's class body, `list[...]` in an @@ -54,25 +54,27 @@ 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.""" +class RuntimeStatus(str, Enum): + """Health of a server that has a pid, derived from OS liveness + OpenCode health. + + A registry entry whose pid is still None (claimed, not yet spawned) + has no RuntimeStatus — see ServerStatus.status. + """ - STARTING = "starting" RUNNING = "running" UNHEALTHY = "unhealthy" STALE = "stale" - FAILED = "failed" - STOPPING = "stopping" @dataclass class ServerStatus: - """Computed liveness/health status for a registry entry.""" + """Computed status for a registry entry. + + status is None while entry.pid hasn't been assigned yet (still starting). + """ entry: RegistryEntry - process_alive: bool - health_ok: bool - display: DisplayStatus + status: RuntimeStatus | None async def _is_health_ok(client: OpenCodeClient, timeout: float = 3.0) -> bool: @@ -84,23 +86,6 @@ async def _is_health_ok(client: OpenCodeClient, timeout: float = 3.0) -> bool: 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: @@ -214,29 +199,28 @@ def __init__(self) -> None: 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 + return entry if entry is not None and entry.pid is not None else None def is_alive(self, key: str) -> bool: """Return True if the server for key is running.""" entry = self.find(key) - return entry is not None and process.is_alive(entry.pid) + return entry is not None and process.is_same(entry.pid, entry.pid_start_time) 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 = process.is_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 + if entry.pid is None: + return ServerStatus(entry=entry, status=None) + if not await asyncio.to_thread(process.is_same, entry.pid, entry.pid_start_time): + return ServerStatus(entry=entry, status=RuntimeStatus.STALE) + client = OpenCodeClient( + base_url=f"http://127.0.0.1:{entry.port}", + password=entry.password, ) + if not await _is_health_ok(client, timeout=health_timeout): + return ServerStatus(entry=entry, status=RuntimeStatus.UNHEALTHY) + return ServerStatus(entry=entry, status=RuntimeStatus.RUNNING) 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.""" @@ -252,19 +236,15 @@ async def status(self, key: str, *, health_timeout: float = 3.0) -> ServerStatus # 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 computed status for every entry that has a pid.""" return [ await self._status_for_entry(entry, health_timeout=health_timeout) - for entry, _ in self.list() + for entry in self.list() ] - def list(self) -> list[tuple[RegistryEntry, bool]]: - """Return all ready registry entries with their liveness status.""" - return [ - (entry, process.is_alive(entry.pid)) - for entry in registry.list_all() - if entry.state == ServerState.RUNNING - ] + def list(self) -> list[RegistryEntry]: + """Return every registry entry that has progressed past a bare claim.""" + return [entry for entry in registry.list_all() if entry.pid is not None] async def health(self, key: str) -> dict[str, Any]: """Return health info for the server at key. @@ -300,7 +280,7 @@ async def stop(self, key: str) -> bool: async def forget() -> None: await asyncio.to_thread(registry.delete_if_instance, key, entry.instance_id) - if not process.is_alive(entry.pid): + if not process.is_same(entry.pid, entry.pid_start_time): await forget() return False @@ -349,8 +329,8 @@ async def get_or_start( ) -> _ManagedServer: """Return a client for the running server, starting one if needed.""" entry = registry.read(key) - if entry is not None and entry.state == ServerState.RUNNING: - if process.is_alive(entry.pid): + if entry is not None and entry.pid is not None: + if process.is_same(entry.pid, entry.pid_start_time): return self._attach(entry) # Stale — the process died after finishing startup. Scoped to # this generation so a concurrent fresh claim isn't clobbered. @@ -367,22 +347,18 @@ async def get_or_start( claimed = registry.claim_starting( RegistryEntry( key=key, - 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, instance_id=instance_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) try: @@ -427,7 +403,7 @@ async def _wait_for_ready(self, key: str, timeout: float = 60.0) -> _ManagedServ raise OpenCodeServerError( f"the caller starting server {key!r} failed before it became ready" ) - if entry.state == ServerState.RUNNING and process.is_alive(entry.pid): + if entry.pid is not None and process.is_same(entry.pid, entry.pid_start_time): return self._attach(entry) await asyncio.sleep(0.1) @@ -504,14 +480,12 @@ async def _start( ready_entry = RegistryEntry( key=key, - state=ServerState.RUNNING, pid=proc.pid, port=port, password=password, project_dir=str(project_dir), server_dir=str(server_dir) if server_dir else None, started_at=started_at, - claimed_at=started_at, runtime_version=version("opencode-runtime"), workspace=workspace, user_id=user_id, diff --git a/tests/test_cli.py b/tests/test_cli.py index 4cf631f..52c4974 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -17,7 +17,7 @@ import opencode_runtime.registry as registry from opencode_runtime import process from opencode_runtime.cli import cmd_health, cmd_ps, cmd_serve, cmd_stop, cmd_stop_all -from opencode_runtime.registry import RegistryEntry, ServerState +from opencode_runtime.registry import RegistryEntry # --------------------------------------------------------------------------- @@ -39,14 +39,12 @@ 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, ) @@ -123,7 +121,7 @@ def test_stop_starting_entry_is_removed(capsys): stoppable by key, not just entries with a pid — cmd_stop used to look it up via find(), which filters to entries with a pid and reported claims as "not found".""" - registry.write(make_entry(state=ServerState.STARTING, pid=None)) + registry.write(make_entry(pid=None)) cmd_stop(ns(key="abc123def456abcd")) out = capsys.readouterr().out assert "Server stopped" in out diff --git a/tests/test_registry.py b/tests/test_registry.py index 6e6023f..56f8391 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -11,7 +11,7 @@ import opencode_runtime.registry as registry from opencode_runtime.exceptions import RegistryBusyError -from opencode_runtime.registry import RegistryEntry, ServerState +from opencode_runtime.registry import RegistryEntry pytestmark = pytest.mark.asyncio @@ -25,14 +25,12 @@ 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] @@ -41,16 +39,10 @@ def make_entry(**kwargs: object) -> RegistryEntry: def make_claim(**kwargs: object) -> RegistryEntry: """A claim entry, as claim_starting() expects — pid is unknown yet. - started_at/claimed_at default to now (not make_entry()'s fixed placeholder date), + started_at defaults to now (not make_entry()'s fixed placeholder date), since claim_starting()'s lease check compares it against the real clock. """ - now = registry.now_iso() - defaults: dict[str, object] = dict( - state=ServerState.STARTING, - pid=None, - started_at=now, - claimed_at=now, - ) + defaults: dict[str, object] = dict(pid=None, started_at=registry.now_iso()) defaults.update(kwargs) return make_entry(**defaults) diff --git a/tests/test_server.py b/tests/test_server.py index 7c79a13..b17836e 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -12,7 +12,6 @@ import opencode_runtime.registry as registry from opencode_runtime import process -from opencode_runtime.registry import ServerState from opencode_runtime.server import ( ServerManager, _ManagedServer, @@ -317,14 +316,12 @@ async def test_stale_registry_entry_cleaned_on_attach(self, tmp_path, monkeypatc registry.write( RegistryEntry( key=key, - state=ServerState.RUNNING, pid=99999999, port=54321, password="stale", project_dir=str(tmp_path), server_dir=None, started_at=timestamp, - claimed_at=timestamp, ) ) @@ -529,14 +526,12 @@ async def test_stop_returns_false_for_dead_process(self, tmp_path, monkeypatch): registry.write( RegistryEntry( key=key, - state=ServerState.RUNNING, pid=99999999, port=54321, password="x", project_dir=str(tmp_path), server_dir=None, started_at=timestamp, - claimed_at=timestamp, ) ) was_alive = await ServerManager().stop(key) @@ -559,14 +554,12 @@ async def fake_start(_self, *, port, password, **kwargs): 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] From 90544b5cffd8c91d171da0bf513129d2a3e72b68 Mon Sep 17 00:00:00 2001 From: ashish16052 Date: Fri, 24 Jul 2026 19:01:14 +0530 Subject: [PATCH 6/8] fix: ruff 0.16.0 compliance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BLE001: add noqa suppression with reason on intentional broad-except sites (health probes, uptime formatting, timestamp parsing — caller cannot know which specific exception network/parse code raises). PLW1510: add check=False to subprocess.run() in process.start_time(). PYI036: tighten __aexit__ signature to type[BaseException] | None, BaseException | None, types.TracebackType | None. ASYNC230 / SIM115: suppress open() in _start() — the file handle is intentionally opened before spawn() and kept for the subprocess lifetime; wrapping it in a context manager would close it immediately. --- src/opencode_runtime/__init__.py | 4 +- src/opencode_runtime/cli.py | 19 +++-- src/opencode_runtime/client.py | 119 +++++++++++++++---------------- src/opencode_runtime/process.py | 1 + src/opencode_runtime/registry.py | 2 +- src/opencode_runtime/runtime.py | 12 +++- src/opencode_runtime/server.py | 6 +- tests/test_cli.py | 29 ++++---- tests/test_multi_tenant.py | 3 +- tests/test_registry.py | 22 +++--- tests/test_runtime.py | 3 +- tests/test_server.py | 7 +- 12 files changed, 115 insertions(+), 112 deletions(-) diff --git a/src/opencode_runtime/__init__.py b/src/opencode_runtime/__init__.py index 50558d0..d526673 100644 --- a/src/opencode_runtime/__init__.py +++ b/src/opencode_runtime/__init__.py @@ -10,8 +10,8 @@ from .session import OpenCodeSession __all__ = [ - "OpenCodeRuntime", - "OpenCodeSession", "OpenCodeEvent", "OpenCodeResponse", + "OpenCodeRuntime", + "OpenCodeSession", ] diff --git a/src/opencode_runtime/cli.py b/src/opencode_runtime/cli.py index 3a959e0..55cc1e9 100644 --- a/src/opencode_runtime/cli.py +++ b/src/opencode_runtime/cli.py @@ -107,12 +107,11 @@ async def _serve(args: argparse.Namespace) -> None: manager = ServerManager() existing = manager.find(key) - if existing is not None: - if manager.is_alive(key): - sys.exit( - _yellow(f"● Server already running id={existing.key} pid={existing.pid}\n") - + _dim(f" use: opencode-runtime stop {existing.key}") - ) + if existing is not None and manager.is_alive(key): + sys.exit( + _yellow(f"● Server already running id={existing.key} pid={existing.pid}\n") + + _dim(f" use: opencode-runtime stop {existing.key}") + ) server_dir: Path | None = None if runtime_dir is not None: @@ -196,7 +195,7 @@ def cmd_ps(_args: argparse.Namespace) -> None: uptime_str = f"{uptime_secs // 60}m" else: uptime_str = f"{uptime_secs // 3600}h" - except Exception: + except Exception: # noqa: BLE001 — malformed timestamp, unknown format uptime_str = "?" vals = [e.key, str(e.pid), str(e.port), status_plain, uptime_str] @@ -268,7 +267,7 @@ def cmd_health(args: argparse.Namespace) -> None: claimed = datetime.fromisoformat(entry.started_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: + except Exception: # noqa: BLE001 — malformed timestamp, unknown format sys.exit(_yellow("◐ starting: awaiting health check")) elif st.status == RuntimeStatus.RUNNING: try: @@ -279,7 +278,7 @@ def cmd_health(args: argparse.Namespace) -> None: + f" {_dim(f'version {version}')}" + f" {_dim(f'http://127.0.0.1:{entry.port}')}" ) - except Exception as exc: + except Exception as exc: # noqa: BLE001 — network error, timeout, etc. sys.exit(_red(f"✗ unhealthy: /global/health failed: {exc}")) elif st.status == RuntimeStatus.UNHEALTHY: sys.exit( @@ -312,7 +311,7 @@ def cmd_inspect(args: argparse.Namespace) -> None: uptime = f"{uptime_secs // 60}m {uptime_secs % 60}s" else: uptime = f"{uptime_secs // 3600}h {(uptime_secs % 3600) // 60}m" - except Exception: + except Exception: # noqa: BLE001 — malformed timestamp, unknown format uptime = "?" print() diff --git a/src/opencode_runtime/client.py b/src/opencode_runtime/client.py index c1d7e5c..2d124ef 100644 --- a/src/opencode_runtime/client.py +++ b/src/opencode_runtime/client.py @@ -162,64 +162,63 @@ async def events(self, session_id: str) -> t.AsyncIterator[OpenCodeEvent]: Only ``sync`` bus-noise events are suppressed. """ - async with self._http() as http: - async with http.stream("GET", "/global/event") as r: + async with self._http() as http, http.stream("GET", "/global/event") as r: + try: + r.raise_for_status() + except httpx.HTTPStatusError as exc: + raise OpenCodeServerError( + f"SSE /global/event returned {exc.response.status_code}" + ) from exc + + async for line in r.aiter_lines(): + line = line.strip() + + if not line.startswith("data:"): + continue + + raw_data = line[len("data:") :].strip() + if not raw_data: + continue + try: - r.raise_for_status() - except httpx.HTTPStatusError as exc: - raise OpenCodeServerError( - f"SSE /global/event returned {exc.response.status_code}" - ) from exc - - async for line in r.aiter_lines(): - line = line.strip() - - if not line.startswith("data:"): - continue - - raw_data = line[len("data:") :].strip() - if not raw_data: - continue - - try: - envelope = json.loads(raw_data) - except json.JSONDecodeError: - continue - - # /global/event wraps each event: {"payload": {"type": ..., "properties": ...}} - payload = envelope.get("payload", {}) - if not isinstance(payload, dict): - continue - - event_type = payload.get("type", "") - - # Suppress internal bus noise — never useful to callers. - if event_type == "sync": - continue - - props = payload.get("properties", {}) - if not isinstance(props, dict): - props = {} - - # Filter to this session (events without sessionID are global, pass through) - sid = props.get("sessionID") - if sid is not None and sid != session_id: - continue - - # Derive a convenience text field where it naturally exists, - # so callers don't have to dig into raw for the common case. - text: str | None = None - if event_type == "message.part.delta" and props.get("field") == "text": - text = props.get("delta") or None - elif event_type == "message.part.updated": - part = props.get("part") or {} - if isinstance(part, dict) and part.get("type") == "text": - text = part.get("text") or None - - yield OpenCodeEvent(type=event_type, text=text, raw=payload) - - # Terminal events — stop iterating after yielding. - if event_type == "session.idle": - return - if event_type == "session.error": - return + envelope = json.loads(raw_data) + except json.JSONDecodeError: + continue + + # /global/event wraps each event: {"payload": {"type": ..., "properties": ...}} + payload = envelope.get("payload", {}) + if not isinstance(payload, dict): + continue + + event_type = payload.get("type", "") + + # Suppress internal bus noise — never useful to callers. + if event_type == "sync": + continue + + props = payload.get("properties", {}) + if not isinstance(props, dict): + props = {} + + # Filter to this session (events without sessionID are global, pass through) + sid = props.get("sessionID") + if sid is not None and sid != session_id: + continue + + # Derive a convenience text field where it naturally exists, + # so callers don't have to dig into raw for the common case. + text: str | None = None + if event_type == "message.part.delta" and props.get("field") == "text": + text = props.get("delta") or None + elif event_type == "message.part.updated": + part = props.get("part") or {} + if isinstance(part, dict) and part.get("type") == "text": + text = part.get("text") or None + + yield OpenCodeEvent(type=event_type, text=text, raw=payload) + + # Terminal events — stop iterating after yielding. + if event_type == "session.idle": + return + if event_type == "session.error": + return diff --git a/src/opencode_runtime/process.py b/src/opencode_runtime/process.py index 55e5eaa..0488de2 100644 --- a/src/opencode_runtime/process.py +++ b/src/opencode_runtime/process.py @@ -55,6 +55,7 @@ def start_time(pid: int | None) -> str | None: capture_output=True, text=True, timeout=3.0, + check=False, ) except (OSError, subprocess.SubprocessError): return None diff --git a/src/opencode_runtime/registry.py b/src/opencode_runtime/registry.py index f345c7c..9cf3c3e 100644 --- a/src/opencode_runtime/registry.py +++ b/src/opencode_runtime/registry.py @@ -19,11 +19,11 @@ import os import time import uuid +from collections.abc import Generator from contextlib import contextmanager from dataclasses import asdict, dataclass, fields from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Generator from .exceptions import RegistryBusyError diff --git a/src/opencode_runtime/runtime.py b/src/opencode_runtime/runtime.py index 0b4d3af..1e42b37 100644 --- a/src/opencode_runtime/runtime.py +++ b/src/opencode_runtime/runtime.py @@ -1,8 +1,11 @@ from __future__ import annotations +import types import typing as t from pathlib import Path +from typing_extensions import Self + from .server import ServerManager, _compute_runtime_key if t.TYPE_CHECKING: @@ -54,10 +57,15 @@ def __init__( # Lifecycle # ------------------------------------------------------------------ - async def __aenter__(self) -> OpenCodeRuntime: + async def __aenter__(self) -> Self: return self - async def __aexit__(self, exc_type: t.Any, exc: t.Any, tb: t.Any) -> None: + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: types.TracebackType | None, + ) -> None: await self.close() async def close(self) -> None: diff --git a/src/opencode_runtime/server.py b/src/opencode_runtime/server.py index 6ccac06..082ab00 100644 --- a/src/opencode_runtime/server.py +++ b/src/opencode_runtime/server.py @@ -82,7 +82,7 @@ async def _is_health_ok(client: OpenCodeClient, timeout: float = 3.0) -> bool: try: await asyncio.wait_for(client.health(), timeout=timeout) return True - except Exception: + except Exception: # noqa: BLE001 — any network/timeout/parse error means unhealthy return False @@ -108,7 +108,7 @@ async def _wait_healthy( try: await client.health() return - except Exception as exc: + except Exception as exc: # noqa: BLE001 — any network/timeout/parse error while polling last_exc = exc await asyncio.sleep(1.0) @@ -443,7 +443,7 @@ async def _start( process_env["HOME"] = str(server_dir) process_env["TMPDIR"] = str(server_dir / "tmp") process_env["OPENCODE_CONFIG"] = str(server_dir / "opencode.json") - output: Any = open(server_dir / "opencode.log", "ab") + output: Any = open(server_dir / "opencode.log", "ab") # noqa: ASYNC230, SIM115 else: output = asyncio.subprocess.DEVNULL diff --git a/tests/test_cli.py b/tests/test_cli.py index 52c4974..ee900fb 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -14,12 +14,10 @@ import pytest -import opencode_runtime.registry as registry -from opencode_runtime import process +from opencode_runtime import process, registry from opencode_runtime.cli import cmd_health, cmd_ps, cmd_serve, cmd_stop, cmd_stop_all from opencode_runtime.registry import RegistryEntry - # --------------------------------------------------------------------------- # fixtures # --------------------------------------------------------------------------- @@ -37,17 +35,17 @@ def ns(**kwargs: object) -> argparse.Namespace: def make_entry(**kwargs: object) -> RegistryEntry: - defaults: dict[str, object] = dict( - key="abc123def456abcd", - 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", - workspace=None, - user_id=None, - ) + defaults: dict[str, object] = { + "key": "abc123def456abcd", + "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", + "workspace": None, + "user_id": None, + } defaults.update(kwargs) return RegistryEntry(**defaults) # type: ignore[arg-type] @@ -278,9 +276,10 @@ def test_serve_duplicate_key_exits(tmp_path): def test_serve_stale_entry_cleaned_and_restarted(tmp_path): """Dead PID in registry — serve should clean it up and start fresh.""" - from opencode_runtime.server import _compute_runtime_key from pathlib import Path + from opencode_runtime.server import _compute_runtime_key + # Compute the same key serve will use key = _compute_runtime_key( workspace=None, user_id=None, project_dir=Path(tmp_path), materials=None, config={} diff --git a/tests/test_multi_tenant.py b/tests/test_multi_tenant.py index 89eecb4..409ecde 100644 --- a/tests/test_multi_tenant.py +++ b/tests/test_multi_tenant.py @@ -10,8 +10,7 @@ import pytest -import opencode_runtime.registry as registry -from opencode_runtime import OpenCodeRuntime, process +from opencode_runtime import OpenCodeRuntime, process, registry pytestmark = pytest.mark.asyncio diff --git a/tests/test_registry.py b/tests/test_registry.py index 56f8391..bd409a9 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -9,7 +9,7 @@ import pytest -import opencode_runtime.registry as registry +from opencode_runtime import registry from opencode_runtime.exceptions import RegistryBusyError from opencode_runtime.registry import RegistryEntry @@ -23,15 +23,15 @@ def isolated_registry(tmp_path, monkeypatch): def make_entry(**kwargs: object) -> RegistryEntry: - defaults: dict[str, object] = 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", - ) + defaults: dict[str, object] = { + "key": "abc123def456abcd", + "pid": 99999, + "port": 54321, + "password": "secret", + "project_dir": "/tmp/project", + "server_dir": None, + "started_at": "2026-07-05T00:00:00+00:00", + } defaults.update(kwargs) return RegistryEntry(**defaults) # type: ignore[arg-type] @@ -42,7 +42,7 @@ def make_claim(**kwargs: object) -> RegistryEntry: started_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, started_at=registry.now_iso()) + defaults: dict[str, object] = {"pid": None, "started_at": registry.now_iso()} defaults.update(kwargs) return make_entry(**defaults) diff --git a/tests/test_runtime.py b/tests/test_runtime.py index f0c1a6f..e3c8cc4 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -8,8 +8,7 @@ import pytest -import opencode_runtime.registry as registry -from opencode_runtime import OpenCodeRuntime, process +from opencode_runtime import OpenCodeRuntime, process, registry pytestmark = pytest.mark.asyncio diff --git a/tests/test_server.py b/tests/test_server.py index b17836e..d1f2c22 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -10,12 +10,11 @@ import pytest -import opencode_runtime.registry as registry -from opencode_runtime import process +from opencode_runtime import process, registry from opencode_runtime.server import ( ServerManager, - _ManagedServer, _compute_runtime_key, + _ManagedServer, ) @@ -640,7 +639,7 @@ async def always_fails(_self, **kwargs): monkeypatch.setattr(ServerManager, "_start", always_fails) manager = ServerManager() - with pytest.raises(Exception): + with pytest.raises(Exception): # noqa: B017 await asyncio.gather( manager.get_or_start( key=key, From 79d605abd230e0226bf85206f39ef148b68e0b46 Mon Sep 17 00:00:00 2001 From: ashish16052 Date: Fri, 24 Jul 2026 19:20:27 +0530 Subject: [PATCH 7/8] fix: use psutil for cross-platform process liveness and identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The manual os.kill(pid,0) + /proc zombie check and the ps-based start_time() were both fragile: - os.kill(pid,0) returns True for zombie processes on Linux, causing stop() to never confirm death and leave registry entries behind - ps -o lstart= is macOS/BSD-only; Linux ps ignores the = suppressor and may produce empty output or behave differently across distros Replace both with psutil: - is_alive() → psutil.Process.is_running() (zombie-aware, all platforms) - start_time() → psutil.Process.create_time() (float, all platforms) subprocess import removed. psutil added to runtime dependencies. types-psutil added to dev dependencies for mypy. pid_start_time field type updated str→float in RegistryEntry. --- pyproject.toml | 2 ++ src/opencode_runtime/process.py | 39 ++++++++++++++++---------------- src/opencode_runtime/registry.py | 2 +- tests/test_process.py | 4 ++-- 4 files changed, 25 insertions(+), 22 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e247166..9b21b97 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,7 @@ classifiers = [ dependencies = [ "httpx", + "psutil", ] [build-system] @@ -43,6 +44,7 @@ dev = [ "pytest-timeout", "ruff", "mypy", + "types-psutil", "build", "twine", "hatchling", diff --git a/src/opencode_runtime/process.py b/src/opencode_runtime/process.py index 0488de2..d99ab95 100644 --- a/src/opencode_runtime/process.py +++ b/src/opencode_runtime/process.py @@ -12,9 +12,10 @@ import asyncio import os import signal -import subprocess from typing import Any +import psutil + async def spawn( *args: str, cwd: str, env: dict[str, str], output: Any @@ -35,35 +36,35 @@ async def spawn( def is_alive(pid: int | None) -> bool: - """Return True if pid is set and a process with it is running.""" + """Return True if pid is set and a non-zombie process with it is running. + + psutil.Process.is_running() returns False for zombie processes, unlike + os.kill(pid, 0) which returns True for zombies on Linux. + """ if pid is None: return False try: - os.kill(pid, 0) - return True - except (ProcessLookupError, PermissionError): + proc = psutil.Process(pid) + return proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE + except psutil.NoSuchProcess: return False -def start_time(pid: int | None) -> str | None: - """Return pid's process start time as reported by `ps`, or None if it's not running.""" +def start_time(pid: int | None) -> float | None: + """Return pid's process creation time as a Unix timestamp, or None. + + Uses psutil for a portable, zombie-aware result on all platforms. + """ if pid is None: return None try: - result = subprocess.run( - ["ps", "-o", "lstart=", "-p", str(pid)], - capture_output=True, - text=True, - timeout=3.0, - check=False, - ) - except (OSError, subprocess.SubprocessError): + return psutil.Process(pid).create_time() + except psutil.NoSuchProcess: return None - return result.stdout.strip() or None -def is_same(pid: int | None, started_at: str | None) -> bool: - """Return True if pid is alive and, when started_at is known, still matches it. +def is_same(pid: int | None, started_at: float | None) -> bool: + """Return True if pid is alive and its creation time matches started_at. started_at is None for entries written before this field existed — those fall back to a plain liveness check. @@ -96,7 +97,7 @@ async def terminate(process: asyncio.subprocess.Process) -> None: kill_group(process.pid, signal.SIGKILL) -async def wait_until_dead(pid: int, started_at: str | None, timeout: float) -> bool: +async def wait_until_dead(pid: int, started_at: float | None, timeout: float) -> bool: """Poll until pid is confirmed gone, or timeout elapses.""" deadline = asyncio.get_event_loop().time() + timeout while asyncio.get_event_loop().time() < deadline: diff --git a/src/opencode_runtime/registry.py b/src/opencode_runtime/registry.py index 9cf3c3e..53fc9ac 100644 --- a/src/opencode_runtime/registry.py +++ b/src/opencode_runtime/registry.py @@ -65,7 +65,7 @@ class RegistryEntry: user_id: str | None = None runtime_version: str | None = None instance_id: str | None = None - pid_start_time: str | None = None + pid_start_time: float | None = None _FIELD_NAMES = {f.name for f in fields(RegistryEntry)} diff --git a/tests/test_process.py b/tests/test_process.py index 596cc65..027faad 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -41,11 +41,11 @@ async def test_is_same_true_for_matching_pid_and_start_time(): async def test_is_same_false_for_mismatched_start_time(): """A pid that's alive but whose start time doesn't match is a different process generation — e.g. the original died and the pid was reused.""" - assert process.is_same(os.getpid(), "Mon Jan 1 00:00:00 1990") is False + assert process.is_same(os.getpid(), 0.0) is False async def test_is_same_false_for_dead_pid(): - assert process.is_same(99999999, "whatever") is False + assert process.is_same(99999999, 0.0) is False async def test_is_same_falls_back_to_liveness_when_start_time_unknown(): From 1a45de0a6fbdd35d47ae445fe6ad0a276cd0aedc Mon Sep 17 00:00:00 2001 From: ashish16052 Date: Fri, 24 Jul 2026 19:22:56 +0530 Subject: [PATCH 8/8] refactor: replace polling loop with psutil.wait_procs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit wait_until_dead() previously polled is_same() every 100ms in an asyncio sleep loop. psutil.wait_procs() blocks at the OS level until the process exits or the timeout elapses — no busy-polling, and it correctly handles the generation check (create_time mismatch) before waiting. --- src/opencode_runtime/process.py | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/src/opencode_runtime/process.py b/src/opencode_runtime/process.py index d99ab95..420e6b4 100644 --- a/src/opencode_runtime/process.py +++ b/src/opencode_runtime/process.py @@ -98,10 +98,18 @@ async def terminate(process: asyncio.subprocess.Process) -> None: async def wait_until_dead(pid: int, started_at: float | None, timeout: float) -> bool: - """Poll until pid is confirmed gone, or timeout elapses.""" - deadline = asyncio.get_event_loop().time() + timeout - while asyncio.get_event_loop().time() < deadline: - if not await asyncio.to_thread(is_same, pid, started_at): - return True - await asyncio.sleep(0.1) - return not is_same(pid, started_at) + """Wait until pid is confirmed gone, or timeout elapses. + + Uses psutil.wait_procs() for efficient OS-level waiting instead of a + polling loop. Returns True if the process is gone within timeout. + """ + try: + p = psutil.Process(pid) + # Verify we have the right generation before waiting. + if started_at is not None and p.create_time() != started_at: + return True # already a different process — original is gone + except psutil.NoSuchProcess: + return True # already gone + + gone, _ = await asyncio.to_thread(psutil.wait_procs, [p], timeout=timeout) + return len(gone) > 0