diff --git a/.changeset/calm-services-start.md b/.changeset/calm-services-start.md index ccdd17b6b0..94b57738d8 100644 --- a/.changeset/calm-services-start.md +++ b/.changeset/calm-services-start.md @@ -1,5 +1,7 @@ --- "@opencode-ai/client": patch +"@opencode-ai/protocol": patch +"@opencode-ai/cli": patch --- -Reuse a same-version background service when a repeated health probe succeeds instead of replacing an endpoint another client may already be using. +Expose background-service lifecycle status, preserve one process-held owner through startup and failure, reconnect TUIs without activating replacement, and stop exact service instances gracefully. diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 534a8d78a7..9b2e87fc7a 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -70,6 +70,14 @@ jobs: env: OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: ${{ runner.os == 'Windows' && 'true' || 'false' }} + - name: Verify compiled service lifecycle + if: always() + timeout-minutes: 10 + working-directory: packages/cli + run: | + bun run script/build.ts --single --skip-install + bun run script/service-smoke.ts + - name: Check generated client if: runner.os == 'Linux' working-directory: packages/client diff --git a/bun.lock b/bun.lock index fd03c58e7d..58927508e4 100644 --- a/bun.lock +++ b/bun.lock @@ -126,6 +126,7 @@ "uqr": "0.1.3", }, "devDependencies": { + "@opencode-ai/protocol": "workspace:*", "@opencode-ai/script": "workspace:*", "@tsconfig/bun": "catalog:", "@types/bun": "catalog:", diff --git a/docs/design/service-lifecycle.md b/docs/design/service-lifecycle.md new file mode 100644 index 0000000000..d4506521a9 --- /dev/null +++ b/docs/design/service-lifecycle.md @@ -0,0 +1,623 @@ +# Service Lifecycle: Election, Restart, and Reconnect + +Status: in progress + +Incident: [#36688](https://github.com/anomalyco/opencode/issues/36688) + +## Summary + +The managed V2 service keeps its current update policy: the background updater +may install a new package, but only a freshly launched TUI activates that update +after finding an older running service. Existing TUIs never replace a service; +they only reconnect. + +The restart path changes in three places: + +1. A process-held OS lock, not the HTTP port or registration file, elects + exactly one server owner for its lifetime. +2. The elected process binds and registers a minimal lifecycle surface before + it initializes the application, so clients can distinguish a slow winner + from an absent server. +3. TUIs rediscover and reconnect indefinitely. Transport loss is never a + terminal error by itself. + +Several clients may spawn small contenders during a restart. This is safe and +intentional: one contender acquires the lock and initializes, while every loser +exits before expensive server boot. The design does not require clients to +agree on a single initiator. + +This proposal does not introduce a supervisor process, warm candidate server, +protocol negotiation, idle background restart, or general execution-recovery +framework. + +## Implementation Status + +| Area | State | +| ------------------------- | --------------------------------------------------------------------- | +| Lifetime ownership | Implemented on this branch with a scoped OS lock | +| Contender behavior | Implemented; losers exit before the server module is imported | +| Registration repair | Implemented; the owner reasserts deleted or corrupt discovery | +| Channel isolation | Implemented with no-clobber migration for legacy preview discovery | +| Client startup waiting | Implemented; slow winners are not killed and waiting is indefinite | +| Lifecycle shell | Implemented; the owner binds and registers before application boot | +| Failed-state latching | Implemented; deterministic boot failure stays bound and actionable | +| Recovery diagnostics | Implemented; the TUI shows status instead of transport internals | +| Cross-platform validation | macOS runtime verified; Linux and Windows run in the unit-test matrix | + +## Context + +The V2 CLI runs a shared managed service that owns Sessions, location graphs, +plugins, permissions, and tool execution. The service updater can replace the +installed package while the current process continues running the old image. +A later TUI launch then detects the version mismatch and replaces the service. + +Incident #36688 showed four failures in that replacement path: + +- Multiple TUIs spawned heavyweight server contenders. +- A winner remained unobservable while it cold-booted, so another wave treated + it as absent and displaced it. +- A fresh TUI exhausted its reconnect budget and crashed with an unhandled + transport defect. +- A losing contender remained alive and consumed about 1 GB of RSS. + +The `origin/v2` baseline serializes service startup with `EffectFlock`. A +contender acquires a three-second heartbeat lease, checks whether another +service became discoverable, and only the winner crosses the application-boot +boundary. This already prevents simultaneous heavy boots and makes startup +losers exit. + +The lease is released immediately after registration, however, so it is not +lifetime ownership. Registration then reverts to last-writer-wins authority: a +deleted or corrupt registration can admit a second boot, a displaced server +terminates itself through its 10-second registration self-check, and a stalled +lease holder can be displaced after the three-second service staleness timeout. + +`Flock` and `EffectFlock` live in `packages/core/src/util` and are also used for +config writes, MCP auth, npm installs, and repository caching. Despite the +name, the primitive is an atomic-mkdir lease with heartbeat and staleness +takeover, not an OS-held lock. It remains appropriate for bounded critical +sections, including today's startup fence, but is not lifetime service +ownership. + +The current implementation also mixes three different concepts: + +- **Ownership:** which process is allowed to be the managed server. +- **Discovery:** where clients can reach that process. +- **Lifecycle:** whether that process is starting, ready, stopping, or failed. + +This design gives each concept one authority. + +```definitions +[ + { + "term": "Owner", + "definition": "The one process holding the process-held OS service lock." + }, + { + "term": "Contender", + "definition": "A small serve process attempting to acquire the service lock. It must not initialize the application before winning." + }, + { + "term": "Registration", + "definition": "An atomic discovery record containing the elected owner's identity and endpoint. Registration never grants ownership." + }, + { + "term": "Lifecycle shell", + "definition": "The minimal HTTP surface bound by the elected process before application initialization. It serves health and retryable startup responses." + }, + { + "term": "Application", + "definition": "The full server routes and global or location-scoped modules used for normal OpenCode work." + } +] +``` + +## Goals + +- At most one process initializes and serves the managed application. +- Losing contenders exit before database, route, plugin, MCP, or location boot. +- A slow winner becomes observable before expensive initialization. +- Existing and freshly launched TUIs survive retryable service unavailability. +- Reconnect follows service state instead of displaying retry counts or raw + transport failures. +- Version-mismatch replacement remains triggered by a fresh TUI launch. +- A stale or malformed registration cannot create a second owner. +- An unresponsive owner is never killed automatically by an arbitrary TUI. +- Every spawned contender has a bounded path to ownership or exit. + +## Non-goals + +- Restarting automatically when a background update finds an idle window. +- Running old and candidate application servers concurrently. +- Adding a permanent steward, proxy, or supervisor process. +- Zero-downtime worker handoff or automatic rollback. +- Application protocol negotiation or automatic TUI self-restart. +- General hard-crash recovery for active Sessions. +- Defining recovery semantics for provider attempts, tools, shells, sub-agents, + permissions, questions, or background jobs. +- Automatically killing a frozen owner. +- Bounding concurrent location cold boots after clients reconnect. +- Multi-machine or clustered service placement. + +## Invariants + +1. **The service lock is ownership.** Exactly one process may hold the OS lock + for one installation channel and service profile. +2. **Ownership precedes boot.** A contender performs no expensive application + initialization before it acquires the lock. +3. **Ownership lasts for the process lifetime.** The owner holds an open lock + handle until the managed server exits. The OS releases it on process death + without a cleanup callback. +4. **The port is transport, not election.** The owner may select a dynamic port + after acquiring the lock. +5. **Registration is discovery, not election.** Deleting, corrupting, or + replacing registration does not invalidate a live owner's lock. +6. **Only a fresh launch enforces package version.** Existing TUIs reconnect to + the current owner without initiating version replacement. +7. **Transport loss is retryable.** It never terminates a TUI without a separate + diagnosed, non-retryable cause. +8. **Clients do not kill an unresponsive owner automatically.** Destructive + recovery requires the explicit `service restart` command. +9. **Lifecycle does not promise execution semantics.** Graceful replacement + invokes Session suspension and resumption hooks, but tool-level continuity + belongs to a separate design. + +## System Model + +```mermaid +flowchart LR + TUI[Fresh or existing TUI] + REG[Registration file] + LOCK[Process-held OS service lock] + SHELL[Lifecycle shell] + APP[OpenCode application] + + TUI -->|discover| REG + TUI -->|observe| SHELL + TUI -->|normal requests| APP + SHELL --> APP + LOCK -->|authorizes one owner| SHELL +``` + +The lifecycle shell and application run in the same process. The distinction is +initialization order and responsibility, not process topology. + +## Service Status + +The server reports one small status value: + +```typescript +type ServiceStatus = + | { + type: "starting" + } + | { + type: "ready" + } + | { + type: "stopping" + targetVersion?: string + } + | { + type: "failed" + message: string + action: string + } +``` + +The client adds only the discovery states needed by callers: + +```typescript +type Status = { type: "missing" } | { type: "unreachable" } | { type: "unresponsive" } | ServiceStatus +``` + +The health response retains the existing fields for old clients and adds the +status discriminant: + +```typescript +type ServiceHealth = { + healthy: true + version: string + pid: number + instanceID: string + status: ServiceStatus +} +``` + +`healthy: true` means the registered lifecycle shell is responding and its +identity matches registration. New clients use `status.type === "ready"` as +the application-readiness signal. + +During `starting` or `stopping`, application requests are not held in memory. +They receive an immediate retryable response: + +```http +HTTP/1.1 503 Service Unavailable +Retry-After: 1 +Content-Type: application/json + +{"code":"service_starting"} +``` + +`stopping` uses `service_stopping`. A failed application boot uses +`service_failed` and includes a safe diagnostic message. + +A failed owner remains bound and keeps holding the service lock. Exiting on +failure would let every waiting client's `ensureRunning` loop elect a new +contender that repeats the same heavy failing boot, so staying bound turns a +deterministic boot failure into one observable `failed` state instead of a +client-driven respawn loop. Recovery still works: a fresh launch observes the +failed instance through the stop path, and explicit `service restart` replaces +it. + +## Registration Contract + +Registration contains only discovery identity: + +```typescript +type ServiceRegistration = { + schema: 1 + instanceID: string + version: string + url: string + pid: number +} +``` + +Authentication continues to use the existing private service credential +storage. The registration schema does not change that policy. + +The owner writes registration only after the lifecycle shell has bound: + +1. Bind the lifecycle shell. +2. Write a temporary registration file with mode `0600`. +3. Atomically rename it over the old registration. +4. Serve lifecycle health as `starting`. + +On shutdown, the owner removes registration only if the current file still has +its `instanceID`. An old finalizer can never remove a successor's registration. + +While running, the owner periodically asserts its registration. Because the +lock guarantees exactly one live owner, any registration that does not name the +owner is stale or corrupt, and the owner rewrites it. A deleted or clobbered +registration therefore heals within one assertion interval instead of leaving +clients waiting on absent discovery. This inverts today's self-check loop, +which terminates the displaced process instead of repairing discovery. + +Legacy registration shapes are decoded by a compatibility adapter. The new +domain type does not make fields optional to represent old formats. + +## Election + +This design promotes today's startup fence into lifetime ownership. +Last-writer-wins registration is replaced by a process-held OS lock that is +acquired before any expensive boot work and held for the entire service +lifetime. + +A heartbeat-and-staleness lease, including the existing `Flock` utility, is not +sufficient for service ownership: the service configures a three-second stale +timeout, after which its lock can be broken and recreated. An event-loop stall, +a suspended machine, or a debugger pause can therefore make a live owner appear +stale and allow a contender to displace it. Service ownership requires a +process-held OS lock: `flock` on Unix and an exclusively bound named pipe on +Windows. It cannot be broken because a heartbeat exceeded a timeout. Process +death releases the lock through the OS. + +Neither Bun nor Node exposes `flock` directly, the existing `Flock` utility is +an mkdir-plus-heartbeat lease rather than an OS-held lock, and the common +lockfile packages are staleness-based leases as well. The platform layer uses +`bun:ffi` to call `flock` on POSIX and Node's named-pipe server support on +Windows, where Bun FFI is not available on every shipped architecture. It lives +alongside the existing utility in `packages/core/src/util`. This primitive is +the foundation of the design, so the delivery sequence spikes it first. + +```mermaid +flowchart TD + A[Contender process starts] + B{Acquire service lock?} + C[Exit successfully] + D[Bind lifecycle shell] + E[Write registration] + F[Report starting] + G[Initialize application] + H[Report ready] + I[Serve until shutdown] + + A --> B + B -->|No| C + B -->|Yes| D + D --> E --> F --> G + G -->|Success| H --> I + G -->|Failure| J[Report failed and stay bound] +``` + +Lock acquisition by a contender is nonblocking or tightly bounded. A loser +must exit before constructing application routes or importing startup-heavy +modules. + +Several clients may spawn contenders concurrently. The design guarantees one +heavy winner, not one process spawn. If the winner crashes during startup, the +OS releases the lock and a later client retry starts another election. + +The lock is scoped by installation channel and service profile. Local, preview, +and stable installations cannot displace one another. + +## Update Activation + +Background update behavior remains unchanged: + +1. The running service checks for an update. +2. The updater installs the package in the background. +3. The running process continues using its existing process image. +4. No idle check or automatic restart occurs. + +A fresh TUI launch activates the installed update: + +1. Read registration and authenticate the responding service. +2. If its package version matches the fresh client, attach normally. +3. If the version differs, request graceful stop of that exact registered + instance using the existing authenticated stop path. +4. Re-check instance identity before every signal or escalation in that path. +5. Wait for the old process to exit and release the service lock. +6. Call `ensureRunning` until a compatible service becomes ready. + +Concurrent fresh launchers may all observe the same old instance. Stopping that +exact instance must be idempotent. Once registration names a different instance, +a stale launcher stops signaling and returns to discovery. + +No durable restart-transition record is introduced. The initiating fresh TUI +already knows the source and target versions and can display its update +preflight. Existing TUIs may display `Updating...` if they observed `stopping`; +otherwise `Waiting for background service...` is the honest fallback. + +## Fresh Launch Versus Reconnect + +Fresh launch and reconnect deliberately have different version policies: + +```typescript +type ManagedConnection = + | { + type: "launch" + requiredVersion: string + } + | { + type: "reconnect" + } +``` + +- `launch` requires the installed package version and may activate replacement. +- `reconnect` accepts the current owner and never activates replacement. + +This preserves today's permissive reconnect behavior. Explicit application +protocol negotiation and automatic TUI re-exec remain follow-ups. + +## Client Reconnect + +Fresh and existing TUIs use the same status loop after startup: + +1. Read registration on every attempt. Do not retry a stale URL indefinitely. +2. If registration is absent, call `ensureRunning` and continue waiting. +3. If registration is unreachable, call `ensureRunning`. A live owner prevents + contenders from acquiring the lock; a dead owner does not. +4. If status is `starting` or `stopping`, wait. +5. If status is `failed`, show its actionable message. +6. If status is `ready`, rebuild HTTP and event-stream clients for the new + endpoint and perform authoritative state reconciliation. + +Retry cadence is internal policy. Retry counts are telemetry, not user-facing +state. The TUI waits until the service is ready or the user exits. + +Transport failures are handled at the TUI run boundary. A raw client transport +error or Effect defect must not escape to the terminal. Hard exit is reserved +for diagnosed causes such as invalid local configuration, failed authentication, +or a foreign process occupying an explicitly configured port. + +The UI derives text from status: + +| Status | User-facing state | +| ------------------------ | ----------------------------------- | +| No registration | `Starting background service...` | +| Registration unreachable | `Waiting for background service...` | +| `starting` | `Starting OpenCode vX...` | +| `stopping` | `Updating to vX...` | +| `failed` | Actionable failure message | +| `ready` | Normal TUI | + +## Graceful Session Continuity + +Version-mismatch replacement uses the existing graceful Session suspension and +resumption hooks: + +1. The old server snapshots active Session IDs during graceful teardown. +2. The successor schedules those Sessions for continuation. +3. The runner reloads durable Session history before continuing. + +This lifecycle design does not define what an interrupted physical provider +attempt or tool invocation means. It does not promise that external side effects +did not occur, replay the exact interrupted tool, preserve an in-memory form, or +recover process-local background work. + +Those concerns require a separate execution-continuity design covering tools, +shells, sub-agents, permissions, questions, provider attempts, and hard-crash +recovery. + +## Unresponsive Owner + +An unreachable registration does not prove that the owner is dead. A contender +attempts the service lock: + +- If the lock is free, the contender starts a replacement. +- If the lock is held, the contender exits and the client keeps waiting. + +After a bounded diagnostic threshold, the client may show: + +```text +The background service owns the service lock but is not responding. +Run `opencode service restart` to recover it. +``` + +Only explicit `service restart` may perform destructive recovery. It verifies +the complete registration and process instance before signaling, waits for +graceful exit, re-checks identity before escalation, and refuses to kill a +process it cannot positively identify. + +Automatic frozen-owner recovery is deferred. + +## Failure Walkthroughs + +### Update with open TUIs + +1. The old service installs vNext but keeps running. +2. A fresh vNext TUI finds the healthy vOld service and requests graceful stop. +3. The old service reports `stopping`, suspends active Sessions, and exits. +4. Open TUIs enter their indefinite status loops. +5. One or more clients spawn contenders. +6. One contender acquires the service lock. Losers exit before heavy boot. +7. The winner binds and registers the lifecycle shell as `starting`. +8. Clients stop spawning and wait on the observable winner. +9. The winner initializes the application and reports `ready`. +10. TUIs rebuild clients, reconcile state, and resume. + +### Server crashes while ready + +1. The endpoint becomes unreachable and registration may remain stale. +2. Clients call `ensureRunning`. +3. Process death has released the service lock. +4. One contender wins, replaces registration, and starts normally. +5. Detailed active-execution recovery is outside this design. + +### Winner crashes during startup + +1. Clients observed `starting` and remain alive. +2. Process death releases the service lock. +3. A later reconnect attempt starts another election. +4. One new contender wins; all other contenders exit. + +### Registration is deleted while the owner is healthy + +1. Clients may call `ensureRunning` because discovery is absent. +2. Every contender fails to acquire the owner's lock and exits. +3. No second application initializes. +4. The owner's next registration assertion republishes discovery. + +### Owner is alive but unresponsive + +1. Health fails, but the process still holds the service lock. +2. Contenders fail lock acquisition and exit. +3. Clients wait and eventually show explicit recovery guidance. +4. No TUI kills the owner automatically. + +## TDD Verification + +Implementation should proceed test-first with real subprocesses and real locks. +Mocks cannot establish process death, lock release, loser cleanup, or port +behavior. + +### Election tests + +| Scenario | Required result | +| ----------------------------------------------------- | ------------------------------------------------------- | +| Ten contenders start simultaneously | Exactly one crosses the application-boot boundary | +| Winner pauses after lock acquisition | No loser initializes or remains alive | +| Winner event loop pauses beyond the old stale timeout | Ownership is not displaced | +| Winner crashes before bind | Lock releases; a later attempt wins | +| Winner crashes after bind but before registration | Lock releases; a later attempt replaces stale discovery | +| Registration is deleted while owner runs | No second owner initializes | +| Registration is malformed | Lock still prevents a second owner | +| Registration names a dead PID | New contender can acquire the released lock | +| Two installation channels start | Each elects an independent owner | +| Explicit configured port is foreign-owned | Fail diagnostically; do not kill the foreign process | + +The fixture records a marker immediately before application initialization. The +tests assert that only one process writes that marker and that every loser exits +within a bounded interval. The harness should also assert that a loser's peak +RSS stays an order of magnitude below an application boot, since import weight +was the observed incident cost. + +### Lifecycle tests + +| Scenario | Required result | +| ----------------------------------------------- | ---------------------------------------------------------------- | +| Winner owns lock but application boot is paused | Health reports `starting` | +| Application request arrives during startup | Immediate retryable `503` | +| Application becomes ready | Status changes once from `starting` to `ready` | +| Graceful replacement begins | Status reports `stopping` before disconnect | +| Application initialization fails | Actionable `failed` status; owner stays bound and holds the lock | +| Registration is deleted while owner runs | Owner republishes it within one assertion interval | +| Owner exits | Registration is removed only if it still names that owner | + +### Update tests + +| Scenario | Required result | +| -------------------------------------- | -------------------------------------------------------- | +| Background update installs vNext | Running vOld service does not restart | +| Fresh vNext launch finds vOld | Exact old instance stops; vNext eventually becomes ready | +| Two fresh vNext launches race | One heavy successor; both clients attach | +| Existing vOld TUI reconnects to vNext | It never requests replacement | +| Stale launcher observes a new instance | It does not signal the new instance | + +### Reconnect tests + +| Scenario | Required result | +| --------------------------------------------------- | -------------------------------------------------- | +| Endpoint disappears and changes port | TUI rediscovers and rebuilds clients | +| Service remains unavailable beyond old retry budget | TUI remains alive | +| Event stream reconnects | Client performs authoritative state reconciliation | +| Transport returns an unexpected defect | TUI formats it; no raw stack escapes | +| Owner remains unresponsive | TUI waits and shows explicit restart guidance | + +## Delivery Sequence + +1. **Spike the lock primitive.** Prove a nonblocking, process-held OS lock + under Bun on macOS, Linux, and Windows (`bun:ffi` to `flock` on POSIX and a + named pipe on Windows), including release on hard kill and behavior across + containers and network filesystems used in CI. +2. **Expand the subprocess test harness.** Begin from the baseline + two-contender test and cover ten contenders, lock release on crash, a paused + winner, deleted or corrupt registration, and bounded loser exit before + changing ownership. +3. **Contain client failure.** Make transport loss nonterminal, rediscover on + every cycle, and format unexpected failures at the TUI boundary. +4. **Promote the startup fence to process-held ownership.** Preserve the + existing pre-boot acquisition seam, replace its lease with the OS lock, hold + it until process exit, and invert the registration self-check from + self-termination to reassertion. +5. **Bind the lifecycle shell first.** Publish registration and `starting`, + return retryable `503` for application requests, then initialize the app. + The health contract change is public API: regenerate clients from + `packages/client` with `bun run generate`. +6. **Codify launch versus reconnect.** Fresh launch enforces installed version; + reconnect never activates replacement. +7. **Integrate graceful replacement.** Preserve current background-install and + fresh-launch activation behavior while invoking Session continuity hooks. +8. **Harden explicit recovery.** Verify exact process identity during explicit + `service restart`; never automatically kill an unresponsive owner. +9. **Run the full multi-process suite.** Include repeated restart cycles and + assert that no contender or child process remains afterward. + +## Acceptance Criteria + +- Ten concurrent restart observers produce one application initialization. +- No losing contender survives or builds a location graph. +- A 30-second application boot remains continuously observable as `starting`. +- A TUI remains alive through a service outage longer than the previous retry + budget. +- A service endpoint change does not require restarting an existing TUI. +- Background installation alone does not restart the service. +- A fresh mismatched TUI eventually attaches to the installed service version. +- Existing reconnecting TUIs never replace the current owner. +- Registration corruption cannot produce two owners. +- A deleted registration heals without restarting the owner or any client. +- An unresponsive owner is not killed without an explicit recovery command. +- Raw transport defects never escape to the terminal. + +## Follow-ups + +- Idle background update activation with an admission fence. +- Application protocol compatibility and automatic local TUI re-exec. +- Durable execution recovery for provider attempts and tools. +- Shell, sub-agent, permission, question, and background-job continuity. +- Automatic recovery for a positively identified frozen owner. +- Cold-boot concurrency limits and interaction-prioritized location loading. +- A steward or socket-handoff architecture if zero-downtime replacement becomes + a real requirement. diff --git a/packages/cli/package.json b/packages/cli/package.json index 217ac4f86c..76eb75f015 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -55,6 +55,7 @@ }, "devDependencies": { "@opencode-ai/script": "workspace:*", + "@opencode-ai/protocol": "workspace:*", "@tsconfig/bun": "catalog:", "@types/bun": "catalog:", "@types/semver": "catalog:", diff --git a/packages/cli/script/service-smoke.ts b/packages/cli/script/service-smoke.ts new file mode 100644 index 0000000000..ff6b252708 --- /dev/null +++ b/packages/cli/script/service-smoke.ts @@ -0,0 +1,143 @@ +#!/usr/bin/env bun + +import { Service } from "@opencode-ai/client/effect" +import { ServiceStatus } from "@opencode-ai/protocol/groups/health" +import { Schema } from "effect" +import fs from "node:fs/promises" +import os from "node:os" +import path from "node:path" + +const target = `cli-${process.platform === "win32" ? "windows" : process.platform}-${process.arch}` +const directory = path.join(import.meta.dir, "..", "dist", target, "bin") +const binary = path.join(directory, `opencode2${process.platform === "win32" ? ".exe" : ""}`) +if (!(await Bun.file(binary).exists())) throw new Error(`Missing compiled CLI in ${directory}`) + +const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-smoke-")) +const env = { + ...process.env, + HOME: root, + USERPROFILE: root, + OPENCODE_DB: path.join(root, "opencode.db"), + OPENCODE_TEST_HOME: root, + XDG_CACHE_HOME: path.join(root, "cache"), + XDG_CONFIG_HOME: path.join(root, "config"), + XDG_DATA_HOME: path.join(root, "data"), + XDG_STATE_HOME: path.join(root, "state"), +} +const processes: Array> = [] +const errors: Array> = [] +let failure: unknown +try { + spawnService() + spawnService() + const registration = await waitForRegistration() + const info = await Schema.decodeUnknownPromise(Service.Info)(await Bun.file(registration).json()) + if (info.id === undefined || info.password === undefined) throw new Error("Registration is missing service identity") + const credential = btoa(`opencode:${info.password}`) + const headers = { authorization: "Basic " + credential } + const token = encodeURIComponent(credential) + const health = await waitForReady(info.url, headers) + if (health.pid !== info.pid || health.instanceID !== info.id) + throw new Error("Health identity does not match registration") + const tokenHealth = await fetch( + new URL(`/api/health?auth_token=${token}`, info.url), + { signal: AbortSignal.timeout(5_000) }, + ) + if (tokenHealth.status !== 200) throw new Error("Compiled service rejected query authentication") + const tokenOpenApi = await fetch( + new URL(`/openapi.json?auth_token=${token}`, info.url), + { signal: AbortSignal.timeout(5_000) }, + ) + if (tokenOpenApi.status !== 200) throw new Error("Compiled application rejected query authentication") + + const unauthorizedHealth = await fetch(new URL("/api/health", info.url), { + signal: AbortSignal.timeout(5_000), + }) + if (unauthorizedHealth.status !== 401) throw new Error("Compiled service exposed health without authentication") + const unauthorizedOpenApi = await fetch(new URL("/openapi.json", info.url), { + signal: AbortSignal.timeout(5_000), + }) + if (unauthorizedOpenApi.status !== 401) throw new Error("Compiled service exposed application routes without authentication") + const unauthorizedStop = await fetch(new URL("/api/service/stop", info.url), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ instanceID: info.id }), + signal: AbortSignal.timeout(5_000), + }) + if (unauthorizedStop.status !== 401) throw new Error("Compiled service accepted unauthenticated stop") + + const winner = processes.find((process) => process.pid === info.pid) + const loser = processes.find((process) => process.pid !== info.pid) + if (!winner || !loser) throw new Error("Compiled contenders did not elect one registered owner") + if (!(await exitsWithin(loser, 10_000))) throw new Error("Losing compiled contender did not exit") + + const stopped = await Schema.decodeUnknownPromise(ServiceStatus.StopResponse)( + await fetch(new URL("/api/service/stop", info.url), { + method: "POST", + headers: { ...headers, "content-type": "application/json" }, + body: JSON.stringify({ instanceID: info.id, targetVersion: "smoke-next" }), + signal: AbortSignal.timeout(5_000), + }).then((response) => response.json()), + ) + if (!stopped.accepted) throw new Error("Compiled service rejected exact-instance stop") + if (!(await exitsWithin(winner, 10_000))) throw new Error("Compiled service did not stop") + for (let attempt = 0; attempt < 200 && (await Bun.file(registration).exists()); attempt++) await Bun.sleep(25) + if (await Bun.file(registration).exists()) throw new Error("Compiled service registration was not removed") +} catch (cause) { + failure = cause +} finally { + processes.forEach((process) => process.kill()) + await Promise.all(processes.map((process) => process.exited)) +} + +const output = await Promise.all(errors) +await fs.rm(root, { recursive: true, force: true }) +if (failure) + throw new Error(output.filter(Boolean).join("\n") || "Compiled service lifecycle smoke test failed", { + cause: failure, + }) + +function spawnService() { + const process = Bun.spawn([binary, "serve", "--service"], { env, stdout: "ignore", stderr: "pipe" }) + processes.push(process) + errors.push(new Response(process.stderr).text()) + return process +} + +async function waitForRegistration() { + const directory = path.join(root, "state", "opencode") + for (let attempt = 0; attempt < 400; attempt++) { + const files = await fs.readdir(directory).catch(() => []) + const file = files.find( + (file) => file === "service.json" || (file.startsWith("service-") && file.endsWith(".json")), + ) + if (file) return path.join(directory, file) + await Bun.sleep(25) + } + throw new Error("Compiled service did not publish registration") +} + +async function waitForReady(url: string, headers: HeadersInit) { + const deadline = Date.now() + 20_000 + while (Date.now() < deadline) { + const health = await fetch(new URL("/api/health", url), { + headers, + signal: AbortSignal.timeout(1_000), + }) + .then((response) => response.json()) + .then(Schema.decodeUnknownPromise(ServiceStatus.Health)) + .catch(() => undefined) + if (health === undefined) { + await Bun.sleep(25) + continue + } + if (health.status.type === "ready") return health + if (health.status.type === "failed") throw new Error(health.status.message) + await Bun.sleep(25) + } + throw new Error("Compiled service did not become ready") +} + +function exitsWithin(process: Bun.Subprocess, milliseconds: number) { + return Promise.race([process.exited.then(() => true), Bun.sleep(milliseconds).then(() => false)]) +} diff --git a/packages/cli/src/commands/handlers/service/restart.ts b/packages/cli/src/commands/handlers/service/restart.ts index 93b8836acd..7dc3e02c8f 100644 --- a/packages/cli/src/commands/handlers/service/restart.ts +++ b/packages/cli/src/commands/handlers/service/restart.ts @@ -9,7 +9,7 @@ export default Runtime.handler( Commands.commands.service.commands.restart, Effect.fn("cli.service.restart")(function* () { const options = yield* ServiceConfig.options() - yield* Service.stop(options) + yield* Service.stop(options, { targetVersion: options.version }) const transport = yield* Service.start(options) process.stdout.write(transport.url + EOL) }), diff --git a/packages/cli/src/commands/handlers/service/status.ts b/packages/cli/src/commands/handlers/service/status.ts index bf58968eef..22a581f658 100644 --- a/packages/cli/src/commands/handlers/service/status.ts +++ b/packages/cli/src/commands/handlers/service/status.ts @@ -8,7 +8,13 @@ import { ServiceConfig } from "../../../services/service-config" export default Runtime.handler( Commands.commands.service.commands.status, Effect.fn("cli.service.status")(function* () { - const found = yield* Service.discover(yield* ServiceConfig.options()) - process.stdout.write((found ? found.url : "stopped") + EOL) + const options = yield* ServiceConfig.options() + const status = yield* Service.status(options) + if (status.type !== "ready") { + process.stdout.write(status.type + EOL) + return + } + const found = yield* Service.discover({ ...options, version: undefined }) + process.stdout.write((found?.url ?? status.type) + EOL) }), ) diff --git a/packages/cli/src/server-process.ts b/packages/cli/src/server-process.ts index 9f688fb16e..b22e39d73c 100644 --- a/packages/cli/src/server-process.ts +++ b/packages/cli/src/server-process.ts @@ -7,11 +7,10 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Global } from "@opencode-ai/core/global" import { InstallationVersion } from "@opencode-ai/core/installation/version" import { AppProcess } from "@opencode-ai/core/process" -import { EffectFlock } from "@opencode-ai/core/util/effect-flock" -import { start } from "@opencode-ai/server/process" +import { ProcessLock } from "@opencode-ai/core/util/process-lock" import { randomBytes, randomUUID } from "node:crypto" import path from "node:path" -import { Effect, Exit, FileSystem, Logger, Option, Redacted, Schedule, Schema, Scope } from "effect" +import { Effect, FileSystem, Logger, Option, Redacted, Schedule, Schema } from "effect" import { HttpServer } from "effect/unstable/http" import { Env } from "./env" import { ServiceConfig } from "./services/service-config" @@ -28,7 +27,7 @@ export type Options = { export const run = Effect.fn("cli.server-process.run")((options: Options) => processEffect(options).pipe( Effect.provide(Updater.layer), - Effect.provide(AppNodeBuilder.build(LayerNode.group([Global.node, AppProcess.node, EffectFlock.node]))), + Effect.provide(AppNodeBuilder.build(LayerNode.group([Global.node, AppProcess.node]))), Effect.provide(NodeServices.layer), ), ) @@ -38,15 +37,15 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { return yield* Effect.scoped( Effect.gen(function* () { const serviceOptions = options.mode === "service" ? yield* ServiceConfig.options() : undefined - const lockScope = serviceOptions === undefined ? undefined : yield* acquireServiceLock(serviceOptions.file) - if ( - serviceOptions !== undefined && - lockScope !== undefined && - (yield* Service.discover(serviceOptions)) !== undefined - ) { - yield* Scope.close(lockScope, Exit.void) - return + if (serviceOptions !== undefined) { + const acquired = yield* ProcessLock.acquire(serviceOptions.file + ".lock").pipe( + Effect.as(true), + Effect.catchTag("ProcessLockHeldError", () => Effect.succeed(false)), + ) + if (!acquired) return yield* Effect.void + if ((yield* Service.discover(serviceOptions)) !== undefined) return yield* Effect.void } + const { start } = yield* Effect.promise(() => import("@opencode-ai/server/process")) const environmentPassword = yield* Env.password // Keep the lease credential out of the environment inherited by tools. if (options.mode === "stdio") { @@ -61,77 +60,82 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { ? Redacted.value(environmentPassword) : randomBytes(32).toString("base64url") if (!password) return yield* Effect.fail(new Error("Missing server password")) - const address = yield* start({ + const instanceID = randomUUID() + const server = yield* start({ hostname: options.hostname ?? config.hostname ?? "127.0.0.1", port: Option.fromNullishOr(options.port ?? config.port), password, - restartContinuity: options.mode === "service", + instanceID, + service: + serviceOptions === undefined + ? undefined + : { onListen: (address) => register(address, password, instanceID, serviceOptions.file) }, }).pipe(Effect.provide(Logger.layer([], { mergeWithExisting: false }))) - if (lockScope !== undefined) { - yield* register(address, password) - yield* Scope.close(lockScope, Exit.void) - } - const url = HttpServer.formatAddress(address) + const url = HttpServer.formatAddress(server.address) console.log(options.mode === "stdio" ? JSON.stringify({ url }) : `server listening on ${url}`) if (options.mode === "default" && !environmentPassword) console.log(`server password ${password}`) const updater = yield* Updater.Service yield* updater.check().pipe(Effect.schedule(Schedule.spaced("10 minutes")), Effect.forkScoped) - return yield* options.mode === "stdio" ? waitForStdinClose() : Effect.never + return yield* options.mode === "service" + ? server.shutdown + : options.mode === "stdio" + ? waitForStdinClose() + : Effect.never }).pipe(Effect.annotateLogs({ role: "server" })), ) }) -const acquireServiceLock = Effect.fnUntraced(function* (file: string) { - const flock = yield* EffectFlock.Service - const scope = yield* Scope.make() - yield* Effect.addFinalizer((exit) => Scope.close(scope, exit)) - yield* flock - .acquire(`service:${file}`, undefined, { staleMs: 3_000, timeoutMs: 3_000 }) - .pipe(Effect.provideService(Scope.Scope, scope)) - return scope -}) - -// The latest atomic registration wins. A displaced process notices the new id, -// exits, and cannot remove its successor's registration from its finalizer. const infoJson = Schema.fromJsonString(Service.Info) const encodeInfo = Schema.encodeEffect(infoJson) const decodeInfo = Schema.decodeUnknownEffect(infoJson) -const register = Effect.fnUntraced(function* (address: HttpServer.Address, password: string) { +const register = Effect.fnUntraced(function* ( + address: HttpServer.Address, + password: string, + id: string, + file: string, +) { const fs = yield* FileSystem.FileSystem - const options = yield* ServiceConfig.options() - const id = randomUUID() - const temp = options.file + "." + id + ".tmp" - yield* fs.makeDirectory(path.dirname(options.file), { recursive: true }) - const encoded = yield* encodeInfo({ + const temp = file + "." + id + ".tmp" + yield* fs.makeDirectory(path.dirname(file), { recursive: true }) + const info = { id, version: InstallationVersion, url: HttpServer.formatAddress(address), pid: process.pid, password, - }) - yield* fs.writeFileString(temp, encoded, { mode: 0o600 }) - yield* fs.rename(temp, options.file) - const currentID = fs.readFileString(options.file).pipe( + } + const encoded = yield* encodeInfo(info) + const publish = fs.writeFileString(temp, encoded, { mode: 0o600 }).pipe(Effect.andThen(fs.rename(temp, file))) + yield* publish + const current = fs.readFileString(file).pipe( Effect.flatMap(decodeInfo), - Effect.map((info) => info.id), Effect.orElseSucceed(() => undefined), ) - yield* currentID.pipe( - Effect.flatMap((current) => - current === id - ? Effect.void - : Effect.try({ try: () => process.kill(process.pid, "SIGTERM"), catch: (cause) => cause }).pipe(Effect.ignore), - ), - Effect.repeat(Schedule.spaced("10 seconds")), - Effect.forkScoped, - ) + const assertRegistration = Effect.gen(function* () { + const found = yield* current + if ( + found !== undefined && + found.id === info.id && + found.version === info.version && + found.url === info.url && + found.pid === info.pid && + found.password === info.password + ) + return + yield* publish + }) yield* Effect.addFinalizer(() => - currentID.pipe( - Effect.flatMap((current) => (current === id ? fs.remove(options.file) : Effect.void)), + current.pipe( + Effect.flatMap((current) => (current?.id === id ? fs.remove(file) : Effect.void)), Effect.ignore, ), ) + yield* assertRegistration.pipe( + Effect.catchCause((cause) => Effect.logWarning("failed to reassert service registration", { cause })), + Effect.repeat(Schedule.spaced("5 seconds")), + Effect.forkScoped, + ) }) function waitForStdinClose() { diff --git a/packages/cli/src/services/server.ts b/packages/cli/src/services/server.ts index 34ddcf965d..b1cdce4cbb 100644 --- a/packages/cli/src/services/server.ts +++ b/packages/cli/src/services/server.ts @@ -16,7 +16,7 @@ export type Args = { export type Resolved = { readonly endpoint: Service.Endpoint - readonly reconnect?: (attempt: number) => Promise + readonly reconnect?: (onStatus: (status: Service.Status) => void, signal: AbortSignal) => Promise readonly reload?: () => Promise } @@ -27,9 +27,7 @@ export const resolve = Effect.fn("cli.server.resolve")(function* (args: Args) { const password = yield* Env.password const endpoint = { url: args.server, - auth: password - ? { type: "basic" as const, username: "opencode", password: Redacted.value(password) } - : undefined, + auth: password ? { type: "basic" as const, username: "opencode", password: Redacted.value(password) } : undefined, } satisfies Service.Endpoint const client = OpenCode.make({ baseUrl: endpoint.url, headers: Service.headers(endpoint) }) const health = yield* Effect.tryPromise({ @@ -51,19 +49,14 @@ export const resolve = Effect.fn("cli.server.resolve")(function* (args: Args) { const reconnectOptions = { ...options, version: undefined } return { endpoint, - reconnect: (attempt) => - Effect.runPromise( - Effect.gen(function* () { - if (attempt > 3) return yield* Service.start(reconnectOptions) - const endpoint = yield* Service.discover(reconnectOptions) - if (endpoint !== undefined) return endpoint - return yield* Effect.fail(new Error("Background server is unavailable")) - }).pipe(Effect.provide(NodeFileSystem.layer)), - ), + reconnect: (onStatus, signal) => + Effect.runPromise(Service.start({ ...reconnectOptions, onStatus }).pipe(Effect.provide(NodeFileSystem.layer)), { + signal, + }), reload: () => Effect.runPromise( Effect.gen(function* () { - yield* Service.stop(options) + yield* Service.stop(options, { targetVersion: options.version }) yield* Service.start(options) }).pipe(Effect.provide(NodeFileSystem.layer)), ), @@ -80,7 +73,8 @@ const resolveManaged = Effect.fnUntraced(function* ( const compatible = yield* Service.discover(options) if (compatible !== undefined) return compatible const existing = yield* Service.discover({ ...options, version: undefined }) - if (existing !== undefined) return yield* Effect.fail(new Error("Background server version does not match this client")) + if (existing !== undefined) + return yield* Effect.fail(new Error("Background server version does not match this client")) return yield* Service.start(options) }) diff --git a/packages/cli/src/services/service-config.ts b/packages/cli/src/services/service-config.ts index ecdd5bf19a..51df30d3b1 100644 --- a/packages/cli/src/services/service-config.ts +++ b/packages/cli/src/services/service-config.ts @@ -1,7 +1,8 @@ import { Global } from "@opencode-ai/core/global" import { InstallationChannel, InstallationVersion } from "@opencode-ai/core/installation/version" +import { Hash } from "@opencode-ai/core/util/hash" import { Service } from "@opencode-ai/client/effect" -import { Effect, FileSystem, Schema } from "effect" +import { Effect, FileSystem, Option, Schema } from "effect" import { randomBytes } from "crypto" import path from "path" @@ -20,25 +21,63 @@ const keys = ["hostname", "port", "password"] as const type Key = (typeof keys)[number] const decodeInfo = Schema.decodeUnknownEffect(Schema.fromJsonString(Info)) +const decodeRegistration = Schema.decodeUnknownEffect(Schema.fromJsonString(Service.Info)) + +export function filename(channel = InstallationChannel) { + if (channel === "latest") return "service.json" + if (channel === "local") return "service-local.json" + return `service-${Hash.fast(channel)}.json` +} + +export function versionBelongsToChannel( + version: string | undefined, + channel = InstallationChannel, + installedVersion = InstallationVersion, +) { + if (version === undefined) return false + if (version === installedVersion) return true + const prefix = `0.0.0-${channel}-` + if (!version.startsWith(prefix)) return false + return /^\d+(?:\.\d+)?$/.test(version.slice(prefix.length)) +} + +export const migrateRegistration = Effect.fnUntraced(function* ( + legacy: string, + file: string, + channel = InstallationChannel, + installedVersion = InstallationVersion, +) { + if (channel === "latest" || channel === "local") return + const fs = yield* FileSystem.FileSystem + const text = yield* fs.readFileString(legacy).pipe(Effect.option) + if (Option.isNone(text)) return + const registration = yield* decodeRegistration(text.value).pipe(Effect.option) + if (Option.isNone(registration)) return + if (!versionBelongsToChannel(registration.value.version, channel, installedVersion)) return + yield* fs.writeFileString(file, text.value, { flag: "wx", mode: 0o600 }).pipe(Effect.ignore) +}) function configKey(key: string): Key { - if (keys.includes(key as Key)) return key as Key + if (key === "hostname" || key === "port" || key === "password") return key throw new Error(`Unknown service config key: ${key}`) } -const env = Effect.gen(function* () { +const paths = Effect.gen(function* () { const fs = yield* FileSystem.FileSystem const global = yield* Global.Service - const filename = InstallationChannel === "local" ? "service-local.json" : "service.json" + const name = filename() + const file = path.join(global.state, name) return { fs, - file: path.join(global.state, filename), - configFile: path.join(global.config, filename), + file, + legacyFile: path.join(global.state, "service.json"), + configFile: path.join(global.config, name), } }) export const options = Effect.fnUntraced(function* () { - const { file } = yield* env + const { file, legacyFile } = yield* paths + yield* migrateRegistration(legacyFile, file) const compiled = path.basename(process.execPath).replace(/\.exe$/, "") !== "bun" const entrypoint = compiled ? undefined : process.argv[1] if (!compiled && entrypoint === undefined) return yield* Effect.fail(new Error("Failed to resolve CLI entrypoint")) @@ -50,7 +89,7 @@ export const options = Effect.fnUntraced(function* () { }) export const read = Effect.fn("cli.service-config.read")(function* () { - const { fs, configFile } = yield* env + const { fs, configFile } = yield* paths return yield* fs.readFileString(configFile).pipe( Effect.flatMap(decodeInfo), Effect.catch(() => Effect.succeed({} as Info)), @@ -58,7 +97,7 @@ export const read = Effect.fn("cli.service-config.read")(function* () { }) const write = Effect.fn("cli.service-config.write")(function* (value: Info) { - const { fs, configFile } = yield* env + const { fs, configFile } = yield* paths const temp = configFile + ".tmp" yield* fs.makeDirectory(path.dirname(configFile), { recursive: true }) yield* fs.writeFileString(temp, JSON.stringify(value, null, 2) + "\n", { mode: 0o600 }) @@ -93,6 +132,7 @@ export const get = Effect.fn("cli.service-config.get")(function* (key?: string) return yield* password() } } + throw new Error(`Unknown service config key: ${key}`) }) export const set = Effect.fn("cli.service-config.set")(function* (key: string, value: string) { diff --git a/packages/cli/test/service.test.ts b/packages/cli/test/service.test.ts index a793ce0257..b2bb525c76 100644 --- a/packages/cli/test/service.test.ts +++ b/packages/cli/test/service.test.ts @@ -35,6 +35,61 @@ test("local channel stores service config with the local service filename", asyn } }) +test("service filenames isolate installation channels", () => { + expect(ServiceConfig.filename("latest")).toBe("service.json") + expect(ServiceConfig.filename("local")).toBe("service-local.json") + expect(ServiceConfig.filename("preview-a")).not.toBe(ServiceConfig.filename("preview-b")) + expect(ServiceConfig.filename("preview-a")).not.toBe(ServiceConfig.filename("latest")) + expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-1234", "preview-a")).toBe(true) + expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-1234.2", "preview-a")).toBe(true) + expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-other-1234", "preview-a")).toBe(false) + expect(ServiceConfig.versionBelongsToChannel("1.2.3", "preview-a")).toBe(false) +}) + +test("preview registration migration never moves stable discovery", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-migration-")) + const legacy = path.join(root, "service.json") + const target = path.join(root, ServiceConfig.filename("preview-a")) + try { + await fs.writeFile( + legacy, + JSON.stringify({ id: "old-preview", version: "0.0.0-preview-a-1234", url: "http://localhost:4096", pid: 1 }), + ) + await Effect.runPromise( + ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe( + Effect.provide(NodeFileSystem.layer), + ), + ) + expect(await Bun.file(legacy).exists()).toBe(true) + expect(await Bun.file(target).json()).toMatchObject({ id: "old-preview" }) + + await fs.rm(target) + await fs.writeFile(legacy, JSON.stringify({ id: "stable", version: "1.2.3", url: "http://localhost:4096", pid: 1 })) + await Effect.runPromise( + ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe( + Effect.provide(NodeFileSystem.layer), + ), + ) + expect(await Bun.file(legacy).exists()).toBe(true) + expect(await Bun.file(target).exists()).toBe(false) + + await fs.writeFile( + legacy, + JSON.stringify({ id: "old-preview", version: "0.0.0-preview-a-1234", url: "http://localhost:4096", pid: 1 }), + ) + await fs.writeFile(target, JSON.stringify({ id: "current-preview" })) + await Effect.runPromise( + ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe( + Effect.provide(NodeFileSystem.layer), + ), + ) + expect(await Bun.file(legacy).exists()).toBe(true) + expect(await Bun.file(target).json()).toMatchObject({ id: "current-preview" }) + } finally { + await fs.rm(root, { recursive: true, force: true }) + } +}) + test("concurrent service processes elect one server", async () => { const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-election-")) const database = path.join(root, "opencode.db") @@ -74,18 +129,55 @@ test("concurrent service processes elect one server", async () => { }), ) const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"] - const first = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) - const second = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) + const registration = path.join(root, "state", "opencode", "service-local.json") + const processes = Array.from({ length: 10 }, () => Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })) try { - const registration = path.join(root, "state", "opencode", "service-local.json") const info = await waitForInfo(registration) - const winner = info.pid === first.pid ? first : second - const loser = info.pid === first.pid ? second : first - const exited = await Promise.race([loser.exited.then(() => true), Bun.sleep(10_000).then(() => false)]) + const winner = processes.find((process) => process.pid === info.pid) + const losers = processes.filter((process) => process.pid !== info.pid) + const exited = await Promise.all( + losers.map((process) => Promise.race([process.exited.then(() => true), Bun.sleep(10_000).then(() => false)])), + ) - expect(exited).toBe(true) - expect(winner.exitCode).toBe(null) + expect(exited).toEqual(losers.map(() => true)) + expect(winner?.exitCode).toBe(null) + expect( + await fetch(new URL("/api/health", info.url), { + headers: { authorization: "Basic " + btoa(`opencode:${info.password}`) }, + }).then((response) => response.json()), + ).toMatchObject({ + healthy: true, + pid: info.pid, + instanceID: info.id, + status: { type: "ready" }, + }) + const blockedTemp = registration + "." + info.id + ".tmp" + await fs.mkdir(blockedTemp) + await fs.rm(registration) + await Bun.sleep(6_000) + expect(await Bun.file(registration).exists()).toBe(false) + await fs.rm(blockedTemp, { recursive: true }) + const restored = await waitForInfo(registration) + expect(restored.id).toBe(info.id) + expect(restored.pid).toBe(info.pid) + await fs.writeFile(registration, "not-json") + const repaired = await waitForInfo(registration) + expect(repaired.id).toBe(info.id) + expect(repaired.pid).toBe(info.pid) + + const contender = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) + try { + const contenderExited = await Promise.race([ + contender.exited.then(() => true), + Bun.sleep(10_000).then(() => false), + ]) + expect(contenderExited).toBe(true) + expect((await waitForInfo(registration)).id).toBe(info.id) + } finally { + contender.kill("SIGTERM") + await contender.exited + } expect( await withDatabase( database, @@ -100,13 +192,70 @@ test("concurrent service processes elect one server", async () => { ), ).toEqual({ timeSuspended: null }) expect(await waitForExecutionStart(database, sessionID)).toBe(1) + await Effect.runPromise( + Service.stop({ file: registration }, { targetVersion: "next" }).pipe(Effect.provide(NodeFileSystem.layer)), + ) + await winner?.exited } finally { - first.kill("SIGTERM") - second.kill("SIGTERM") - await Promise.all([first.exited, second.exited]) + processes.forEach((process) => process.kill("SIGTERM")) + await Promise.all(processes.map((process) => process.exited)) + try { + expect(await Bun.file(registration).exists()).toBe(false) + } finally { + await fs.rm(root, { recursive: true, force: true }) + } + } +}, 60_000) + +test("a failed service stays registered and owns the lock until stopped", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-failed-")) + const database = path.join(root, "database") + await fs.mkdir(database) + const env = { + ...process.env, + HOME: root, + OPENCODE_DB: database, + OPENCODE_TEST_HOME: root, + XDG_CACHE_HOME: path.join(root, "cache"), + XDG_CONFIG_HOME: path.join(root, "config"), + XDG_DATA_HOME: path.join(root, "data"), + XDG_STATE_HOME: path.join(root, "state"), + } + const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"] + const registration = path.join(root, "state", "opencode", "service-local.json") + const owner = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) + + try { + const info = await waitForInfo(registration) + const status = await Effect.runPromise( + Service.status({ file: registration }).pipe( + Effect.filterOrFail((status) => status.type === "failed"), + Effect.retry(Schedule.spaced("50 millis").pipe(Schedule.both(Schedule.recurs(200)))), + Effect.provide(NodeFileSystem.layer), + ), + ) + expect(status).toEqual({ + type: "failed", + version: info.version, + message: "The background service could not start.", + action: "Run `opencode service restart` after checking the service logs.", + }) + expect(owner.exitCode).toBe(null) + + const contender = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) + expect(await Promise.race([contender.exited.then(() => true), Bun.sleep(10_000).then(() => false)])).toBe(true) + expect((await waitForInfo(registration)).id).toBe(info.id) + expect(owner.exitCode).toBe(null) + + await Effect.runPromise(Service.stop({ file: registration }).pipe(Effect.provide(NodeFileSystem.layer))) + await owner.exited + expect(await Bun.file(registration).exists()).toBe(false) + } finally { + owner.kill("SIGTERM") + await owner.exited await fs.rm(root, { recursive: true, force: true }) } -}) +}, 30_000) function withDatabase(file: string, effect: Effect.Effect) { return Effect.runPromise(effect.pipe(Effect.provide(Database.layerFromPath(file)), Effect.scoped)) @@ -143,7 +292,7 @@ function waitForExecutionStart(file: string, sessionID: SessionV2.ID) { } async function waitForInfo(file: string) { - for (let attempt = 0; attempt < 200; attempt++) { + for (let attempt = 0; attempt < 400; attempt++) { const value = await Bun.file(file) .json() .catch(() => undefined) diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 28c06cbf04..190079e574 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -10,8 +10,17 @@ type StreamValue = A extends Stream.Stream ? Success export type Endpoint0_0Output = EffectValue> export type HealthGetOperation = () => Effect.Effect +type Endpoint0_1Request = Parameters[0] +export type Endpoint0_1Input = { + readonly instanceID: Endpoint0_1Request["payload"]["instanceID"] + readonly targetVersion?: Endpoint0_1Request["payload"]["targetVersion"] +} +export type Endpoint0_1Output = EffectValue> +export type HealthStopOperation = (input: Endpoint0_1Input) => Effect.Effect + export interface HealthApi { readonly get: HealthGetOperation + readonly stop: HealthStopOperation } export type Endpoint1_0Output = EffectValue> diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index 1f58717113..85414eebb8 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -16,7 +16,17 @@ const mapClientError = (error: E) => const Endpoint0_0 = (raw: RawClient["server.health"]) => () => raw["health.get"]({}).pipe(Effect.mapError(mapClientError)) -const adaptGroup0 = (raw: RawClient["server.health"]) => ({ get: Endpoint0_0(raw) }) +type Endpoint0_1Request = Parameters[0] +type Endpoint0_1Input = { + readonly instanceID: Endpoint0_1Request["payload"]["instanceID"] + readonly targetVersion?: Endpoint0_1Request["payload"]["targetVersion"] +} +const Endpoint0_1 = (raw: RawClient["server.health"]) => (input: Endpoint0_1Input) => + raw["health.stop"]({ payload: { instanceID: input["instanceID"], targetVersion: input["targetVersion"] } }).pipe( + Effect.mapError(mapClientError), + ) + +const adaptGroup0 = (raw: RawClient["server.health"]) => ({ get: Endpoint0_0(raw), stop: Endpoint0_1(raw) }) const Endpoint1_0 = (raw: RawClient["server.server"]) => () => raw["server.get"]({}).pipe(Effect.mapError(mapClientError)) diff --git a/packages/client/src/effect/service.ts b/packages/client/src/effect/service.ts index 3c42f0de56..86cbae6403 100644 --- a/packages/client/src/effect/service.ts +++ b/packages/client/src/effect/service.ts @@ -1,5 +1,6 @@ +import { ServiceStatus } from "@opencode-ai/protocol/groups/health" import { Effect, FileSystem, Option, Schedule, Schema } from "effect" -import { spawn } from "node:child_process" +import { spawn, type ChildProcess } from "node:child_process" import { homedir } from "node:os" import { join } from "node:path" @@ -39,6 +40,23 @@ export type StartOptions = Options & { // found, or a healthy service with a different version is being replaced. // `existing` carries the registration of the service being replaced. readonly onStart?: (reason: StartReason, existing?: Info) => void + readonly onStatus?: (status: Status) => void +} + +export type Status = + | { readonly type: "missing" } + | { readonly type: "unreachable" } + | { readonly type: "unresponsive" } + | (ServiceStatus.State & { readonly version?: string }) + +export class FailedError extends Schema.TaggedErrorClass()("ServiceFailedError", { + message: Schema.String, + action: Schema.String, +}) {} + +type Contender = { + readonly child: ChildProcess + readonly error: () => Error | undefined } // Read-only lookup: registration file plus health check and version gate. @@ -47,58 +65,138 @@ export const discover = Effect.fn("service.discover")(function* (options: Option return (yield* discoverLocal(options))?.endpoint }) +export const status = Effect.fn("service.status")(function* (options: Options = {}) { + const result = yield* registered(options.file, true) + if (result.info === undefined) return { type: "missing" } satisfies Status + if (result.service === undefined) return { type: "unreachable" } satisfies Status + return publicStatus(result.service) +}) + +function publicStatus(service: LocalService): Status { + return { ...service.status, version: service.version } +} + const discoverLocal = Effect.fnUntraced(function* (options: Options) { - const info = yield* read(options.file) - if (info === undefined) return undefined - if (options.version !== undefined && info.version !== options.version) return undefined - return yield* probe(info, options.version) + const found = (yield* registered(options.file)).service + if (found?.status.type !== "ready") return undefined + if (options.version !== undefined && found.version !== options.version) return undefined + return found }) // Idempotent ensure-running: reuses a healthy compatible server, replaces a -// version-mismatched one, and otherwise spawns the service command detached. +// version-mismatched one, and otherwise spawns small contenders until a server +// becomes discoverable. A contender is never killed merely for slow startup. export const start = Effect.fn("service.start")(function* (options: StartOptions = {}) { - const compatible = yield* discover(options) - if (compatible !== undefined) return compatible - const existing = yield* find(options) - if (existing?.version !== undefined && (options.version === undefined || existing.version === options.version)) - return existing.endpoint - yield* Effect.sync(() => options.onStart?.(existing === undefined ? "missing" : "version-mismatch", existing?.info)) - if (existing !== undefined) yield* kill(existing.info, options).pipe(Effect.ignore) - - const [command, ...args] = options.command ?? ["opencode", "serve", "--service"] - if (command === undefined) return yield* Effect.fail(new Error("Missing service command")) - const child = yield* Effect.try({ - try: () => { - const child = spawn(command, args, { detached: true, stdio: "ignore" }) - child.unref() - return child - }, - catch: (cause) => new Error("Failed to start server", { cause }), + const contenders = new Set() + let announced = false + let reported: Status | undefined + let lastSpawn = 0 + let spawnDelay = 5_000 + let ownerHeld = false + const announce = (reason: StartReason, existing?: Info) => + Effect.sync(() => { + if (announced) return + announced = true + options.onStart?.(reason, existing) + }) + const spawnContender = Effect.gen(function* () { + const [command, ...args] = options.command ?? ["opencode", "serve", "--service"] + if (command === undefined) return yield* Effect.fail(new Error("Missing service command")) + return yield* Effect.try({ + try: () => { + const child = spawn(command, args, { detached: true, stdio: "ignore" }) + let error: Error | undefined + child.once("error", (cause) => { + error = new Error("Failed to start server", { cause }) + }) + child.unref() + return { child, error: () => error } + }, + catch: (cause) => new Error("Failed to start server", { cause }), + }) }) + const found = yield* Effect.gen(function* () { + const registration = yield* registered(options.file, true) + const info = registration.info + const service = registration.service + const current: Status = + service === undefined ? { type: info === undefined ? "missing" : "unreachable" } : publicStatus(service) + const next = ownerHeld && service === undefined ? ({ type: "unresponsive" } satisfies Status) : current + yield* Effect.sync(() => { + if (sameStatus(reported, next)) return + reported = next + options.onStatus?.(next) + }) + if (service !== undefined) { + ownerHeld = false + spawnDelay = 5_000 + const compatible = !service.legacy && (options.version === undefined || service.version === options.version) + if (compatible && service.status.type === "ready") return Option.some(service) + if (compatible && service.status.type === "failed") + return yield* new FailedError({ message: service.status.message, action: service.status.action }) + if (compatible || service.status.type === "stopping") return Option.none() + yield* announce("version-mismatch", service.info) + yield* kill(service, options, options.version).pipe(Effect.ignore) + lastSpawn = 0 + return Option.none() + } else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now() - return yield* discoverLocal(options).pipe( - Effect.flatMap((found) => - found === undefined ? Effect.fail(new Error("Server is not ready")) : Effect.succeed(found), - ), - Effect.retry(poll), - Effect.tap((found) => - found.info.pid === child.pid - ? Effect.void - : Effect.sync(() => { - child.kill("SIGTERM") - }), - ), - Effect.map((found) => found.endpoint), - Effect.tapError(() => Effect.try({ try: () => child.kill("SIGTERM"), catch: () => undefined }).pipe(Effect.ignore)), - Effect.mapError(() => new Error("Failed to start server")), - ) + const failure = [...contenders].map(contenderFailure).find((error): error is Error => error !== undefined) + if (failure !== undefined) return yield* Effect.fail(failure) + const finished = [...contenders].filter(contenderFinished) + if (finished.some((item) => item.child.exitCode === 0)) { + ownerHeld = true + spawnDelay = Math.min(spawnDelay * 2, 30_000) + } + finished.forEach((item) => contenders.delete(item)) + // Keep one candidate plus one lock probe so a pre-lock stall cannot block recovery. + if (contenders.size < 2 && Date.now() - lastSpawn >= spawnDelay) { + yield* announce("missing") + contenders.add(yield* spawnContender) + lastSpawn = Date.now() + } + return Option.none() + }).pipe(Effect.repeat({ until: Option.isSome, schedule: Schedule.spaced("1 second") })) + return Option.getOrThrow(found).endpoint }) -export const stop = Effect.fn("service.stop")(function* (options: Options = {}) { - const fs = yield* FileSystem.FileSystem +function sameStatus(left: Status | undefined, right: Status) { + if (left?.type !== right.type) return false + if (right.type === "failed") + return ( + left.type === "failed" && + left.version === right.version && + left.message === right.message && + left.action === right.action + ) + if (right.type === "stopping") + return left.type === "stopping" && left.version === right.version && left.targetVersion === right.targetVersion + if (right.type === "starting" || right.type === "ready") + return left.type === right.type && left.version === right.version + return true +} + +function contenderFailure(contender: Contender) { + const error = contender.error() + if (error !== undefined) return error + if (contender.child.exitCode !== null && contender.child.exitCode !== 0) + return new Error(`Server process exited with code ${contender.child.exitCode}`) + if (contender.child.signalCode !== null) + return new Error(`Server process terminated by ${contender.child.signalCode}`) + return undefined +} + +function contenderFinished(contender: Contender) { + return contender.error() !== undefined || contender.child.exitCode !== null || contender.child.signalCode !== null +} + +export type StopMetadata = { + readonly targetVersion?: string +} + +export const stop = Effect.fn("service.stop")(function* (options: Options = {}, metadata: StopMetadata = {}) { const existing = yield* find(options) - if (existing !== undefined) yield* kill(existing.info, options) - yield* fs.remove(options.file ?? fallback()).pipe(Effect.ignore) + if (existing !== undefined) yield* kill(existing, options, metadata.targetVersion) }) function fallback() { @@ -106,7 +204,7 @@ function fallback() { return join(state, "opencode", "service.json") } -export function headers(endpoint: Endpoint): RequestInit["headers"] { +export function headers(endpoint: Endpoint) { if (endpoint.auth === undefined) return undefined return { authorization: "Basic " + btoa(endpoint.auth.username + ":" + endpoint.auth.password) } } @@ -121,9 +219,7 @@ export const Info = Schema.Struct({ export type Info = typeof Info.Type const decode = Schema.decodeUnknownEffect(Schema.fromJsonString(Info)) -const decodeHealth = Schema.decodeUnknownOption( - Schema.Struct({ healthy: Schema.Literal(true), version: Schema.String, pid: Schema.Int }), -) +const decodeHealth = Schema.decodeUnknownOption(ServiceStatus.Health) const decodeLegacyHealth = Schema.decodeUnknownOption(Schema.Struct({ healthy: Schema.Literal(true) })) // A missing or corrupt file means no valid info; callers treat both @@ -139,9 +235,11 @@ type LocalService = { readonly info: Info readonly endpoint: Endpoint readonly version?: string + readonly status: ServiceStatus.State + readonly legacy: boolean } -const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLegacy = false) { +const probe = Effect.fnUntraced(function* (info: Info, allowLegacy = false) { const endpoint = { url: info.url, auth: @@ -155,14 +253,21 @@ const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLe signal: AbortSignal.timeout(2_000), }), ).pipe(Effect.option, Effect.map(Option.getOrUndefined)) - if (response === undefined || !response.ok) return undefined + if (response === undefined) return undefined const body = yield* Effect.tryPromise(() => response.json()).pipe(Effect.option, Effect.map(Option.getOrUndefined)) const health = decodeHealth(body) if (Option.isSome(health)) { if (health.value.pid !== info.pid) return undefined if (info.version !== undefined && health.value.version !== info.version) return undefined - if (version !== undefined && health.value.version !== version) return undefined - return { info, endpoint, version: health.value.version } satisfies LocalService + if (info.id !== undefined && health.value.instanceID !== undefined && health.value.instanceID !== info.id) + return undefined + return { + info, + endpoint, + version: health.value.version, + status: health.value.status, + legacy: false, + } satisfies LocalService } if ( !allowLegacy || @@ -170,18 +275,23 @@ const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLe (typeof body === "object" && body !== null && ("version" in body || "pid" in body)) ) return undefined - return { info, endpoint } satisfies LocalService + return { info, endpoint, status: { type: "ready" }, legacy: true } satisfies LocalService }) -// Health-checked lookup without the version gate: lifecycle operations must be +const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = false) { + const info = yield* read(file) + if (info === undefined) return { info: undefined, service: undefined } + return { info, service: yield* probe(info, allowLegacy) } +}) + +// Health-checked lookup without the version gate: status operations must be // able to see (and replace or stop) a server from a different version. const find = Effect.fnUntraced(function* (options: Options) { - const info = yield* read(options.file) - if (info === undefined) return undefined - return yield* probe(info, undefined, true) + return (yield* registered(options.file, true)).service }) -// 50ms cadence bounded at ~5s, shared by stop escalation and start readiness. +// 50ms cadence bounded at ~5s, shared by stop escalation and each start +// discovery window. const poll = Schedule.spaced("50 millis").pipe(Schedule.both(Schedule.recurs(100))) const signal = (pid: number, name: NodeJS.Signals) => @@ -199,20 +309,42 @@ function same(left: Info, right: Info) { return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid } -const kill = Effect.fnUntraced(function* (info: Info, options: Options) { - // A stale registration may point at a PID that has since been reused by - // another process. Only signal the PID after authenticating the server. - const current = yield* find(options) - if (current === undefined || !same(current.info, info)) return - - yield* signal(info.pid, "SIGTERM") - const done = yield* stopped(info.pid).pipe(Effect.retry(poll), Effect.option) +const kill = Effect.fnUntraced(function* (service: LocalService, options: Options, targetVersion?: string) { + const requested = yield* requestStop(service, targetVersion) + if (requested === "rejected") return + if (requested === "unsupported") { + // A stale registration may point at a reused PID. Authenticate again + // immediately before the legacy signal fallback. + const current = yield* find(options) + if (current === undefined || !same(current.info, service.info)) return + yield* signal(service.info.pid, "SIGTERM") + } + const done = yield* stopped(service.info.pid).pipe(Effect.retry(poll), Effect.option) if (Option.isSome(done)) return const latest = yield* find(options) - if (latest === undefined || !same(latest.info, info)) return - yield* signal(info.pid, "SIGKILL") - yield* stopped(info.pid).pipe(Effect.retry(poll)) + if (latest === undefined || !same(latest.info, service.info)) return + yield* signal(service.info.pid, "SIGKILL") + yield* stopped(service.info.pid).pipe(Effect.retry(poll)) +}) + +const decodeStopResponse = Schema.decodeUnknownOption(ServiceStatus.StopResponse) + +const requestStop = Effect.fnUntraced(function* (service: LocalService, targetVersion?: string) { + if (service.info.id === undefined || service.legacy) return "unsupported" as const + const response = yield* Effect.tryPromise(() => + fetch(new URL("/api/service/stop", service.info.url), { + method: "POST", + headers: { ...headers(service.endpoint), "content-type": "application/json" }, + body: JSON.stringify({ instanceID: service.info.id, targetVersion }), + signal: AbortSignal.timeout(2_000), + }), + ).pipe(Effect.option, Effect.map(Option.getOrUndefined)) + if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const + const body = yield* Effect.tryPromise(() => response.json()).pipe(Effect.option, Effect.map(Option.getOrUndefined)) + const decoded = decodeStopResponse(body) + if (!response.ok || Option.isNone(decoded) || !decoded.value.accepted) return "rejected" as const + return "accepted" as const }) export * as Service from "./service.js" diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index bed325d976..97c4c0a350 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -1,5 +1,7 @@ import type { HealthGetOutput, + HealthStopInput, + HealthStopOutput, ServerGetOutput, LocationGetInput, LocationGetOutput, @@ -332,6 +334,18 @@ export function make(options: ClientOptions) { { method: "GET", path: `/api/health`, successStatus: 200, declaredStatuses: [401, 400], empty: false }, requestOptions, ), + stop: (input: HealthStopInput, requestOptions?: RequestOptions) => + request( + { + method: "POST", + path: `/api/service/stop`, + body: { instanceID: input["instanceID"], targetVersion: input["targetVersion"] }, + successStatus: 200, + declaredStatuses: [401, 400], + empty: false, + }, + requestOptions, + ), }, server: { get: (requestOptions?: RequestOptions) => diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 255beaa1bb..2718a0640b 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -1,5 +1,13 @@ export type JsonValue = null | boolean | number | string | Array | { [key: string]: JsonValue } +export type ServiceStatus = + | { type: "starting" } + | { type: "ready" } + | { type: "stopping"; targetVersion?: string | null } + | { type: "failed"; message: string; action: string } + +export type ServiceStopResponse = { accepted: boolean } + export type ModelRef = { id: string; providerID: string; variant?: string } export type ProviderSettings = { [x: string]: JsonValue } @@ -498,6 +506,14 @@ export type VcsFileStatus = { status: "added" | "deleted" | "modified" } +export type ServiceHealth = { + healthy: true + version: string + pid: number + instanceID?: string | null + status?: ServiceStatus +} + export type SessionMessageModelSelected = { id: string metadata?: { [x: string]: JsonValue } @@ -2483,7 +2499,14 @@ export type ProjectCopyError = { export const isProjectCopyError = (value: unknown): value is ProjectCopyError => typeof value === "object" && value !== null && "name" in value && value["name"] === "ProjectCopyError" -export type HealthGetOutput = { healthy: true; version: string; pid: number } +export type HealthGetOutput = ServiceHealth + +export type HealthStopInput = { + readonly instanceID: { readonly instanceID: string; readonly targetVersion?: string | undefined }["instanceID"] + readonly targetVersion?: { readonly instanceID: string; readonly targetVersion?: string | undefined }["targetVersion"] +} + +export type HealthStopOutput = ServiceStopResponse export type ServerGetOutput = { urls: Array } diff --git a/packages/client/test/effect.test.ts b/packages/client/test/effect.test.ts index c1254ab7d3..623907ea9e 100644 --- a/packages/client/test/effect.test.ts +++ b/packages/client/test/effect.test.ts @@ -15,6 +15,18 @@ import { const synced = { type: "log.synced" as const, aggregateID: "ses_test", seq: Event.Seq.make(1) } +test("health.get treats an old server as ready", async () => { + const httpClient = HttpClient.make((request) => + Effect.succeed(HttpClientResponse.fromWeb(request, Response.json({ healthy: true, version: "old", pid: 123 }))), + ) + const result = await Effect.gen(function* () { + const client = yield* OpenCode.make({ baseUrl: "http://localhost:3000" }) + return yield* client.health.get() + }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient), Effect.runPromise) + + expect(result.status).toEqual({ type: "ready" }) +}) + test("session.get returns the decoded Effect projection", async () => { const httpClient = HttpClient.make((request) => Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(session))), diff --git a/packages/client/test/fixture/service.ts b/packages/client/test/fixture/service.ts index 3f77edb419..7b78ac190f 100644 --- a/packages/client/test/fixture/service.ts +++ b/packages/client/test/fixture/service.ts @@ -1,13 +1,45 @@ -import { rename, writeFile } from "node:fs/promises" +import { appendFile, rename, writeFile } from "node:fs/promises" -const [registration, mode] = process.argv.slice(2) +const [registration, mode, delay] = process.argv.slice(2) if (registration === undefined || mode === undefined) throw new Error("Missing service fixture arguments") +if (mode === "failed") process.exit(1) +if (mode === "record-start") { + await writeFile(registration + ".started", "") + process.exit(1) +} +if (mode === "signal") process.kill(process.pid, process.platform === "win32" ? "SIGTERM" : "SIGKILL") + +if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated") { + await appendFile(registration + ".starts", process.pid + "\n") + const owner = await writeFile(registration + ".owner", String(process.pid), { flag: "wx" }) + .then(() => true) + .catch(() => false) + if (!owner) process.exit() + if (mode === "coordinated") { + while ((await Bun.file(registration + ".starts").text()).trim().split("\n").length < 2) await Bun.sleep(10) + } else await Bun.sleep(Number(delay)) + if (mode === "delayed-failed") process.exit(1) +} let requests = 0 +const version = mode === "old" || mode === "reject-stop" ? "old" : "test" +const id = crypto.randomUUID() const server = Bun.serve({ port: 0, async fetch(request) { - if (new URL(request.url).pathname !== "/api/health") return new Response(null, { status: 404 }) + const pathname = new URL(request.url).pathname + if (pathname === "/api/service/stop" && mode === "reject-stop") { + await writeFile(registration + ".stop-attempt", "") + return Response.json({ accepted: false }) + } + if (pathname === "/api/service/stop" && mode === "graceful") { + const body = await request.json() + if (typeof body !== "object" || body === null || body.instanceID !== id) return Response.json({ accepted: false }) + await writeFile(registration + ".stop", JSON.stringify(body)) + setTimeout(shutdown, 25) + return Response.json({ accepted: true }) + } + if (pathname !== "/api/health") return new Response(null, { status: 404 }) requests += 1 if (mode === "modern" && requests === 1) { await writeFile(registration + ".first-request", "") @@ -15,15 +47,33 @@ const server = Bun.serve({ return new Response(null, { status: 503 }) } if (mode === "legacy") return Response.json({ healthy: true }) - return Response.json({ healthy: true, version: "test", pid: process.pid }) + if (mode === "starting" && !(await Bun.file(registration + ".release").exists())) + return Response.json( + { healthy: true, version, pid: process.pid, instanceID: id, status: { type: "starting" } }, + { status: 503 }, + ) + if (mode === "failed-owner") + return Response.json( + { + healthy: true, + version, + pid: process.pid, + instanceID: id, + status: { type: "failed", message: "Could not open the database.", action: "Check the service logs." }, + }, + { status: 503 }, + ) + if (mode === "starting" || mode === "graceful" || mode === "reject-stop") + return Response.json({ healthy: true, version, pid: process.pid, instanceID: id, status: { type: "ready" } }) + return Response.json({ healthy: true, version, pid: process.pid }) }, }) await writeFile( registration + ".tmp", JSON.stringify({ - id: crypto.randomUUID(), - version: mode === "legacy" ? undefined : "test", + id, + version: mode === "legacy" ? undefined : version, url: server.url.toString(), pid: process.pid, }), @@ -31,7 +81,7 @@ await writeFile( ) await rename(registration + ".tmp", registration) -const shutdown = () => { +function shutdown() { server.stop(true) process.exit() } diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index ba822f28e5..dea3225297 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -16,7 +16,7 @@ test("exposes every standard HTTP API group", () => { "generate", "provider", "integration", - "server.mcp", + "mcp", "credential", "project", "form", @@ -61,6 +61,22 @@ test("server.get uses the public HTTP contract", async () => { expect(request?.url).toBe("http://localhost:3000/api/server") }) +test("health.stop sends exact replacement identity", async () => { + let request: Request | undefined + const client = OpenCode.make({ + baseUrl: "http://localhost:3000", + fetch: async (input, init) => { + request = input instanceof Request ? input : new Request(input, init) + return Response.json({ accepted: true }) + }, + }) + + expect(await client.health.stop({ instanceID: "instance", targetVersion: "next" })).toEqual({ accepted: true }) + expect(request?.method).toBe("POST") + expect(request?.url).toBe("http://localhost:3000/api/service/stop") + expect(await request?.json()).toEqual({ instanceID: "instance", targetVersion: "next" }) +}) + test("MCP resource catalog uses the public HTTP contract", async () => { let request: Request | undefined const client = OpenCode.make({ @@ -77,7 +93,7 @@ test("MCP resource catalog uses the public HTTP contract", async () => { }, }) - const result = await client["server.mcp"].resource.catalog({ location: { directory: "/tmp/project" } }) + const result = await client.mcp.resource.catalog({ location: { directory: "/tmp/project" } }) expect(result.data.resources[0]?.uri).toBe("docs://readme") expect(request?.method).toBe("GET") diff --git a/packages/client/test/service.test.ts b/packages/client/test/service.test.ts index 21428c9166..484baf7922 100644 --- a/packages/client/test/service.test.ts +++ b/packages/client/test/service.test.ts @@ -43,6 +43,75 @@ test("a concurrent same-version start cannot invalidate a resolved endpoint", as expect(starts).toEqual([]) expect(await Bun.file(registration).json()).toEqual(original) expect(await health(resolved.url)).toEqual({ healthy: true, version: "test", pid: original.pid }) + expect(await run(Service.status({ file: registration }))).toEqual({ type: "ready", version: "test" }) +}) + +test("waits for a registered service to finish starting", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + const process = spawn(registration, "starting") + await waitForFile(registration) + const statuses: Service.Status[] = [] + const result = run( + Service.start({ file: registration, version: "test", command: [], onStatus: (status) => statuses.push(status) }), + ) + + await Bun.sleep(500) + expect(process.exitCode).toBe(null) + expect(statuses).toContainEqual({ type: "starting", version: "test" }) + expect(statuses.filter((status) => status.type === "starting")).toHaveLength(1) + await writeFile(registration + ".release", "") + expect((await result).url).toBe((await Bun.file(registration).json()).url) +}) + +test("reports a failed registered service without spawning", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + const process = spawn(registration, "failed-owner") + await waitForFile(registration) + + await expect(run(Service.start({ file: registration, version: "test", command: [] }))).rejects.toMatchObject({ + message: "Could not open the database.", + action: "Check the service logs.", + }) + expect(process.exitCode).toBe(null) +}) + +test("requests graceful replacement of the exact service instance", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + const process = spawn(registration, "graceful") + await waitForFile(registration) + const info = await Bun.file(registration).json() + + await run(Service.stop({ file: registration }, { targetVersion: "next" })) + await process.exited + expect(await Bun.file(registration + ".stop").json()).toEqual({ instanceID: info.id, targetVersion: "next" }) +}) + +test("does not spawn contenders while an incompatible service rejects replacement", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + const contender = join(directory, "contender.json") + const existing = spawn(registration, "reject-stop") + await waitForFile(registration) + const controller = new AbortController() + const starting = Effect.runPromise( + Service.start({ + file: registration, + version: "test", + command: [process.execPath, fixture, contender, "record-start"], + }).pipe(Effect.provide(NodeFileSystem.layer)), + { signal: controller.signal }, + ) + + await waitForFile(registration + ".stop-attempt") + await Bun.sleep(500) + controller.abort() + await starting.catch(() => undefined) + + expect(await Bun.file(contender + ".started").exists()).toBe(false) + expect(existing.exitCode).toBe(null) }) test("a legacy health response is still replaced", async () => { @@ -57,9 +126,95 @@ test("a legacy health response is still replaced", async () => { await expect(result).rejects.toThrow("Missing service command") expect(starts).toEqual(["version-mismatch"]) await existing.exited -}) +}, 10_000) -function run(effect: Effect.Effect) { +test("waits for a slow winner while bounding lock probes", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + const endpoint = await run( + Service.start({ + file: registration, + version: "test", + command: [process.execPath, fixture, registration, "coordinated"], + }), + ) + const info = await Bun.file(registration).json() + try { + expect(endpoint.url).toBe(info.url) + expect(await health(endpoint.url)).toEqual({ healthy: true, version: "test", pid: info.pid }) + expect((await Bun.file(registration + ".starts").text()).trim().split("\n")).toHaveLength(2) + } finally { + process.kill(info.pid, "SIGTERM") + } +}, 15_000) + +test("reports a contender that fails to start", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + await expect( + run( + Service.start({ + file: registration, + version: "test", + command: [process.execPath, fixture, registration, "failed"], + }), + ), + ).rejects.toThrow("Server process exited with code 1") +}, 10_000) + +test("reports a contender terminated by a signal", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + await expect( + run( + Service.start({ + file: registration, + version: "test", + command: [process.execPath, fixture, registration, "signal"], + }), + ), + ).rejects.toThrow(/Server process (terminated by|exited with code)/) +}, 10_000) + +test("reports a slow contender that eventually fails", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + await expect( + run( + Service.start({ + file: registration, + version: "test", + command: [process.execPath, fixture, registration, "delayed-failed", "8000"], + }), + ), + ).rejects.toThrow("Server process exited with code 1") +}, 15_000) + +test("replaces an incompatible owner that appears during startup", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + const starting = run( + Service.start({ + file: registration, + version: "test", + command: [process.execPath, fixture, registration, "delayed", "8000"], + }), + ) + await Bun.sleep(1_000) + const old = spawn(registration, "old") + await waitForFile(registration) + const endpoint = await starting + const info = await Bun.file(registration).json() + try { + expect(endpoint.url).toBe(info.url) + expect(info.version).toBe("test") + await old.exited + } finally { + process.kill(info.pid, "SIGTERM") + } +}, 20_000) + +function run(effect: Effect.Effect) { return Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer))) } diff --git a/packages/core/src/util/process-lock.ts b/packages/core/src/util/process-lock.ts new file mode 100644 index 0000000000..045ac9fa05 --- /dev/null +++ b/packages/core/src/util/process-lock.ts @@ -0,0 +1,181 @@ +import { dlopen, read, type Pointer } from "bun:ffi" +import { closeSync, existsSync, mkdirSync, openSync } from "node:fs" +import { connect, createServer, type Server, type Socket } from "node:net" +import path from "node:path" +import { Effect, Schema } from "effect" +import { Hash } from "./hash" + +export namespace ProcessLock { + export class HeldError extends Schema.TaggedErrorClass()("ProcessLockHeldError", { + file: Schema.String, + }) { + override get message() { + return `Process lock is already held: ${this.file}` + } + } + + export class SystemError extends Schema.TaggedErrorClass()("ProcessLockSystemError", { + file: Schema.String, + operation: Schema.Literals(["open", "acquire"]), + code: Schema.String, + }) { + override get message() { + return `Process lock ${this.operation} failed for ${this.file}: ${this.code}` + } + } + + export type LockError = HeldError | SystemError + + const acquirePosix = Effect.fnUntraced(function* (file: string) { + const fd = yield* Effect.try({ + try: () => { + mkdirSync(path.dirname(file), { recursive: true }) + return openSync(file, "a+", 0o600) + }, + catch: (cause) => + new SystemError({ + file, + operation: "open", + code: cause instanceof Error ? cause.message : String(cause), + }), + }) + const result = yield* Effect.try({ + try: () => lock(fd), + catch: (cause) => + new SystemError({ + file, + operation: "acquire", + code: cause instanceof Error ? cause.message : String(cause), + }), + }).pipe( + Effect.tapError(() => + Effect.sync(() => { + closeSync(fd) + }), + ), + ) + if (result.acquired) { + return fd + } + closeSync(fd) + return yield* result.held + ? new HeldError({ file }) + : new SystemError({ file, operation: "acquire", code: String(result.code) }) + }) + + export const acquire = Effect.fn("ProcessLock.acquire")(function* (file: string) { + if (process.platform === "win32") { + yield* Effect.acquireRelease(acquireWindows(file), closeWindows) + return + } + yield* Effect.acquireRelease(acquirePosix(file), (fd) => + Effect.sync(() => { + closeSync(fd) + }), + ) + }) +} + +type Result = + | { readonly acquired: true } + | { readonly acquired: false; readonly held: true } + | { readonly acquired: false; readonly held: false; readonly code: number } + +const LOCK_EX = 2 +const LOCK_NB = 4 +const DARWIN_EWOULDBLOCK = 35 +const LINUX_EWOULDBLOCK = 11 + +function lock(fd: number): Result { + if (process.platform === "darwin") return lockDarwin(fd) + if (process.platform === "linux") return lockLinux(fd) + throw new Error(`Unsupported process lock platform: ${process.platform}`) +} + +function lockDarwin(fd: number): Result { + const library = dlopen("/usr/lib/libSystem.B.dylib", { + flock: { args: ["i32", "i32"], returns: "i32" }, + __error: { args: [], returns: "ptr" }, + }) + try { + const result = library.symbols.flock(fd, LOCK_EX | LOCK_NB) + const code = result === 0 ? 0 : errorCode(library.symbols.__error()) + if (result === 0) return { acquired: true } + if (code === DARWIN_EWOULDBLOCK) return { acquired: false, held: true } + return { acquired: false, held: false, code } + } finally { + library.close() + } +} + +function lockLinux(fd: number): Result { + const musl = `/lib/libc.musl-${process.arch === "arm64" ? "aarch64" : "x86_64"}.so.1` + const library = dlopen(existsSync(musl) ? musl : "libc.so.6", { + flock: { args: ["i32", "i32"], returns: "i32" }, + __errno_location: { args: [], returns: "ptr" }, + }) + try { + const result = library.symbols.flock(fd, LOCK_EX | LOCK_NB) + const code = result === 0 ? 0 : errorCode(library.symbols.__errno_location()) + if (result === 0) return { acquired: true } + if (code === LINUX_EWOULDBLOCK) return { acquired: false, held: true } + return { acquired: false, held: false, code } + } finally { + library.close() + } +} + +function errorCode(pointer: Pointer | null) { + if (pointer === null) throw new Error("Failed to read process lock error code") + return read.i32(pointer, 0) +} + +function acquireWindows(file: string) { + return Effect.callback((resume) => { + const server = createServer() + let probe: Socket | undefined + const pipe = `\\\\.\\pipe\\opencode-process-lock-${Hash.sha256(path.resolve(file).toLowerCase())}` + const onError = (cause: NodeJS.ErrnoException) => { + server.off("listening", onListening) + probe = connect(pipe) + const onProbeError = () => { + probe?.off("connect", onConnect) + resume( + Effect.fail( + new ProcessLock.SystemError({ + file, + operation: "acquire", + code: cause.code ?? cause.message, + }), + ), + ) + } + const onConnect = () => { + probe?.off("error", onProbeError) + probe?.destroy() + resume(Effect.fail(new ProcessLock.HeldError({ file }))) + } + probe.once("connect", onConnect) + probe.once("error", onProbeError) + } + const onListening = () => { + server.off("error", onError) + resume(Effect.succeed(server)) + } + server.once("error", onError) + server.once("listening", onListening) + server.on("connection", (socket) => socket.destroy()) + server.listen(pipe) + return Effect.sync(() => { + probe?.destroy() + server.close() + }) + }) +} + +function closeWindows(server: Server) { + return Effect.callback((resume) => { + if (!server.listening) return resume(Effect.void) + server.close((error) => resume(error ? Effect.die(error) : Effect.void)) + }) +} diff --git a/packages/core/test/fixture/process-lock-worker.ts b/packages/core/test/fixture/process-lock-worker.ts new file mode 100644 index 0000000000..8a283c68fe --- /dev/null +++ b/packages/core/test/fixture/process-lock-worker.ts @@ -0,0 +1,17 @@ +import { ProcessLock } from "@opencode-ai/core/util/process-lock" +import { Effect, Schema } from "effect" +import fs from "node:fs/promises" + +const input = Schema.decodeUnknownSync( + Schema.fromJsonString(Schema.Struct({ file: Schema.String, ready: Schema.String })), +)(process.argv[2]) + +await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + yield* ProcessLock.acquire(input.file) + yield* Effect.promise(() => fs.writeFile(input.ready, String(process.pid))) + return yield* Effect.never + }), + ), +) diff --git a/packages/core/test/util/process-lock.test.ts b/packages/core/test/util/process-lock.test.ts new file mode 100644 index 0000000000..9457b9c08e --- /dev/null +++ b/packages/core/test/util/process-lock.test.ts @@ -0,0 +1,71 @@ +import { expect } from "bun:test" +import { ProcessLock } from "@opencode-ai/core/util/process-lock" +import { Effect } from "effect" +import fs from "node:fs/promises" +import os from "node:os" +import path from "node:path" +import { it } from "../lib/effect" + +const worker = path.join(import.meta.dir, "../fixture/process-lock-worker.ts") + +it.live( + "releases ownership when the scope closes", + Effect.gen(function* () { + const root = yield* temp("opencode-process-lock-") + const file = path.join(root, "service.lock") + yield* Effect.scoped(ProcessLock.acquire(file)) + yield* Effect.scoped(ProcessLock.acquire(file)) + }), +) + +it.live( + "releases ownership when the process dies", + Effect.gen(function* () { + const root = yield* temp("opencode-process-lock-death-") + const file = path.join(root, "service.lock") + const ready = path.join(root, "ready") + const child = yield* Effect.acquireRelease( + Effect.sync(() => + Bun.spawn([process.execPath, worker, JSON.stringify({ file, ready })], { + stdout: "ignore", + stderr: "pipe", + }), + ), + (child) => + Effect.promise(async () => { + kill(child) + await child.exited + }), + ) + yield* Effect.promise(async () => { + for (let attempt = 0; attempt < 100 && !(await Bun.file(ready).exists()); attempt++) await Bun.sleep(20) + }) + expect(yield* Effect.promise(() => Bun.file(ready).exists())).toBe(true) + + const error = yield* Effect.scoped(ProcessLock.acquire(file)).pipe(Effect.flip) + expect(error._tag).toBe("ProcessLockHeldError") + + if (process.platform !== "win32") { + process.kill(child.pid, "SIGSTOP") + const paused = yield* Effect.scoped(ProcessLock.acquire(file)).pipe(Effect.flip) + expect(paused._tag).toBe("ProcessLockHeldError") + process.kill(child.pid, "SIGCONT") + } + + kill(child) + yield* Effect.promise(() => child.exited) + yield* Effect.scoped(ProcessLock.acquire(file)) + }), +) + +function temp(prefix: string) { + return Effect.acquireRelease( + Effect.promise(() => fs.mkdtemp(path.join(os.tmpdir(), prefix))), + (root) => Effect.promise(() => fs.rm(root, { recursive: true, force: true })), + ) +} + +function kill(child: Bun.Subprocess) { + if (process.platform === "win32") return child.kill() + return child.kill("SIGKILL") +} diff --git a/packages/protocol/src/groups/health.ts b/packages/protocol/src/groups/health.ts index 625cce3919..0bddc52c08 100644 --- a/packages/protocol/src/groups/health.ts +++ b/packages/protocol/src/groups/health.ts @@ -1,19 +1,64 @@ -import { Schema } from "effect" +import { Effect, Schema } from "effect" import { HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi" +export namespace ServiceStatus { + export const State = Schema.Union([ + Schema.Struct({ type: Schema.Literal("starting") }), + Schema.Struct({ type: Schema.Literal("ready") }), + Schema.Struct({ + type: Schema.Literal("stopping"), + targetVersion: Schema.String.pipe(Schema.optional), + }), + Schema.Struct({ + type: Schema.Literal("failed"), + message: Schema.String, + action: Schema.String, + }), + ]).annotate({ identifier: "ServiceStatus" }) + export type State = typeof State.Type + + export const Health = Schema.Struct({ + healthy: Schema.Literal(true), + version: Schema.String, + pid: Schema.Int.check(Schema.isGreaterThan(0)), + instanceID: Schema.String.pipe(Schema.optional), + status: State.pipe(Schema.withDecodingDefaultKey(Effect.succeed({ type: "ready" as const }))), + }).annotate({ identifier: "ServiceHealth" }) + export type Health = typeof Health.Type + + export const StopRequest = Schema.Struct({ + instanceID: Schema.String, + targetVersion: Schema.String.pipe(Schema.optional), + }).annotate({ identifier: "ServiceStopRequest" }) + export type StopRequest = typeof StopRequest.Type + + export const StopResponse = Schema.Struct({ + accepted: Schema.Boolean, + }).annotate({ identifier: "ServiceStopResponse" }) + export type StopResponse = typeof StopResponse.Type +} + export const HealthGroup = HttpApiGroup.make("server.health") .add( HttpApiEndpoint.get("health.get", "/api/health", { - success: Schema.Struct({ - healthy: Schema.Literal(true), - version: Schema.String, - pid: Schema.Int.check(Schema.isGreaterThan(0)), - }), + success: ServiceStatus.Health, }).annotateMerge( OpenApi.annotations({ identifier: "v2.health.get", summary: "Check server health", - description: "Check whether the API server is ready to accept requests.", + description: "Report the owning server process and its application status.", + }), + ), + ) + .add( + HttpApiEndpoint.post("health.stop", "/api/service/stop", { + payload: ServiceStatus.StopRequest, + success: ServiceStatus.StopResponse, + }).annotateMerge( + OpenApi.annotations({ + identifier: "v2.health.stop", + summary: "Stop the managed server", + description: "Request graceful shutdown of one exact managed server instance.", }), ), ) diff --git a/packages/server/src/handlers/health.ts b/packages/server/src/handlers/health.ts index a73b050aa7..f3e07688f1 100644 --- a/packages/server/src/handlers/health.ts +++ b/packages/server/src/handlers/health.ts @@ -4,7 +4,14 @@ import { HttpApiBuilder } from "effect/unstable/httpapi" import { Api } from "../api" export const HealthHandler = HttpApiBuilder.group(Api, "server.health", (handlers) => - handlers.handle("health.get", () => - Effect.succeed({ healthy: true as const, version: InstallationVersion, pid: process.pid }), - ), + handlers + .handle("health.get", () => + Effect.succeed({ + healthy: true as const, + version: InstallationVersion, + pid: process.pid, + status: { type: "ready" }, + }), + ) + .handle("health.stop", () => Effect.succeed({ accepted: false })), ) diff --git a/packages/server/src/middleware/authorization.ts b/packages/server/src/middleware/authorization.ts index cc785ee247..a004fa973d 100644 --- a/packages/server/src/middleware/authorization.ts +++ b/packages/server/src/middleware/authorization.ts @@ -35,6 +35,10 @@ function credentialFromRequest(request: HttpServerRequest.HttpServerRequest) { return Effect.succeed(emptyCredential()) } +export function authorizedRequest(request: HttpServerRequest.HttpServerRequest, config: ServerAuth.Info) { + return credentialFromRequest(request).pipe(Effect.map((credential) => ServerAuth.authorized(credential, config))) +} + export const authorizationLayer = Layer.effect( Authorization, Effect.gen(function* () { @@ -46,8 +50,7 @@ export const authorizationLayer = Layer.effect( // Browsers cannot set headers on WebSocket upgrades, so a ticketed PTY connect skips // credential checks here; the connect handler consumes and validates the ticket. if (hasPtyConnectTicketURL(new URL(request.url, "http://localhost"))) return yield* effect - const credential = yield* credentialFromRequest(request) - if (ServerAuth.authorized(credential, config)) return yield* effect + if (yield* authorizedRequest(request, config)) return yield* effect yield* HttpEffect.appendPreResponseHandler((_request, response) => Effect.succeed(HttpServerResponse.setHeader(response, "www-authenticate", WWW_AUTHENTICATE)), ) diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index d9bef69e10..6eb8937d00 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -1,74 +1,211 @@ export * as ServerProcess from "./process" -import { NodeHttpClient, NodeHttpServer } from "@effect/platform-node" -import { HealthGroup } from "@opencode-ai/protocol/groups/health" -import { Context, Effect, Layer, Option } from "effect" -import { HttpClient, HttpClientRequest, HttpMiddleware, HttpRouter, HttpServer } from "effect/unstable/http" -import { HttpApi, HttpApiClient } from "effect/unstable/httpapi" -import { createServer } from "node:http" +import { NodeHttpServer, NodeHttpServerRequest } from "@effect/platform-node" import { SessionRestart } from "@opencode-ai/core/session/execution/restart" +import { ServiceStatus } from "@opencode-ai/protocol/groups/health" +import { hasPtyConnectTicketURL } from "@opencode-ai/protocol/groups/pty" +import { Cause, Context, Deferred, Effect, Exit, Layer, Option, Ref, Schema, Scope } from "effect" +import { HttpMiddleware, HttpRouter, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http" +import { createServer } from "node:http" import { ServerAuth } from "./auth" +import { authorizedRequest } from "./middleware/authorization" import { createRoutes } from "./routes" import { ServerInfo } from "./server-info" +import { Status } from "./service-status" -export type Options = { +export type Options = { readonly hostname: string readonly port: Option.Option readonly password: string - readonly restartContinuity?: boolean + readonly instanceID: string + readonly service?: { + readonly onListen: (address: HttpServer.Address) => Effect.Effect + } } -const ReadinessApi = HttpApi.make("readiness").add(HealthGroup) +type App = Effect.Effect< + HttpServerResponse.HttpServerResponse, + unknown, + HttpServerRequest.HttpServerRequest | Scope.Scope +> -export const start = Effect.fn("ServerProcess.start")(function* (options: Options) { +export const start = Effect.fn("ServerProcess.start")(function* (options: Options) { if (!options.password) return yield* Effect.fail(new Error("Missing server password")) - const address = yield* listen(options) - yield* Effect.gen(function* () { - const client = yield* HttpApiClient.make(ReadinessApi, { - baseUrl: HttpServer.formatAddress(address), - transformClient: HttpClient.mapRequest((request) => - HttpClientRequest.setHeader( - request, - "authorization", - ServerAuth.header({ username: "opencode", password: options.password }) ?? "", - ), + const shutdown = yield* Deferred.make() + const status = yield* Status.make({ + instanceID: options.instanceID, + managed: options.service !== undefined, + }) + const bound = yield* listen(options) + const application = yield* Ref.make(Option.none()) + yield* bound.http.serve(dispatch(options.password, status, application, shutdown), HttpMiddleware.logger) + if (options.service) yield* options.service.onListen(bound.http.address) + + const parentScope = yield* Scope.Scope + const applicationScope = yield* Scope.fork(parentScope) + yield* Effect.addFinalizer(() => + status + .beginStopping() + .pipe( + Effect.andThen(Ref.set(application, Option.none())), + Effect.andThen(Effect.sync(() => bound.server.closeAllConnections())), ), - }) - yield* client["server.health"]["health.get"]({}) - }).pipe(Effect.provide(NodeHttpClient.layerNodeHttp)) - return address + ) + + const boot = Effect.gen(function* () { + const context = yield* Layer.buildWithScope( + createRoutes(options.password, () => { + const address = bound.server.address() + if (address === null || typeof address === "string") return [] + const host = address.family === "IPv6" ? `[${address.address}]` : address.address + return ServerInfo.connectionURLs(`http://${host}:${address.port}`, options.hostname) + }).pipe(Layer.provide(NodeHttpServer.layerHttpServices)), + applicationScope, + ) + if (options.service) { + yield* installRestartContinuity(Context.get(context, SessionRestart.Service)).pipe( + Effect.provideService(Scope.Scope, applicationScope), + ) + } + yield* Ref.set(application, Option.some(Context.get(context, HttpRouter.HttpRouter).asHttpEffect())) + yield* status.ready + return { address: bound.http.address, shutdown: Deferred.await(shutdown) } + }).pipe( + Effect.catchCause((cause) => { + if (!options.service || Cause.hasInterruptsOnly(cause)) return Effect.failCause(cause) + return status + .fail({ + message: "The background service could not start.", + action: "Run `opencode service restart` after checking the service logs.", + }) + .pipe( + Effect.andThen( + Scope.close(applicationScope, Exit.failCause(cause)).pipe( + Effect.catchCause((cleanupCause) => + Effect.logError("failed to clean up background service boot", { cause: cleanupCause }), + ), + ), + ), + Effect.andThen(Effect.logError("background service boot failed", { cause })), + Effect.andThen(Effect.never), + ) + }), + ) + if (!options.service) return yield* boot + return yield* Effect.raceFirst(boot, Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt))) }) -function listen(options: Options) { - if (Option.isSome(options.port)) return bind(options, options.port.value) +function listen(options: { readonly hostname: string; readonly port: Option.Option }) { + if (Option.isSome(options.port)) return bind(options.hostname, options.port.value) const next = (port: number): ReturnType => - bind(options, port).pipe(Effect.catch((error) => (port === 65_535 ? Effect.fail(error) : next(port + 1)))) + bind(options.hostname, port).pipe( + Effect.catch((error) => (port < 65_535 && addressInUse(error) ? next(port + 1) : Effect.fail(error))), + ) return next(4096) } -function bind(options: Options, port: number) { - const server = createServer() - return Layer.build( - createRoutes(options.password, () => { - const address = server.address() - if (address === null || typeof address === "string") return [] - const host = address.family === "IPv6" ? `[${address.address}]` : address.address - return ServerInfo.connectionURLs(`http://${host}:${address.port}`, options.hostname) +function bind(hostname: string, port: number) { + return Effect.gen(function* () { + const parentScope = yield* Scope.Scope + const serverScope = yield* Scope.fork(parentScope) + const server = createServer() + return yield* Effect.gen(function* () { + const http = yield* NodeHttpServer.make(() => server, { port, host: hostname }) + yield* Effect.addFinalizer(() => Effect.sync(() => server.closeAllConnections())) + return { http, server } }).pipe( - Layer.flatMap((context) => { - const serve = HttpServer.serve( - Context.get(context, HttpRouter.HttpRouter).asHttpEffect(), - HttpMiddleware.logger, - ) - if (!options.restartContinuity) return serve - const restart = Context.get(context, SessionRestart.Service) - return Layer.merge(serve, restartContinuity(restart)) - }), - Layer.provideMerge(NodeHttpServer.layer(() => server, { port, host: options.hostname })), - ), - ).pipe( - Effect.tap(() => Effect.addFinalizer(() => Effect.sync(() => server.closeAllConnections()))), - Effect.map((context) => Context.get(context, HttpServer.HttpServer).address), + Effect.provideService(Scope.Scope, serverScope), + Effect.onError((cause) => Scope.close(serverScope, Exit.failCause(cause))), + ) + }) +} + +function addressInUse(error: unknown) { + if (typeof error !== "object" || error === null || !("cause" in error)) return false + const cause = error.cause + return typeof cause === "object" && cause !== null && "code" in cause && cause.code === "EADDRINUSE" +} + +function dispatch( + password: string, + status: Status.Interface, + application: Ref.Ref>, + shutdown: Deferred.Deferred, +): App { + const auth = ServerAuth.Config.of({ username: "opencode", password: Option.some(password) }) + return Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest + const url = new URL(request.url, "http://localhost") + const lifecycle = + request.method === "GET" && url.pathname === "/api/health" + ? "health" + : request.method === "POST" && url.pathname === "/api/service/stop" + ? "stop" + : undefined + if (lifecycle !== undefined) { + if (!(yield* authorizedRequest(request, auth))) return unauthorized() + return yield* control(request, lifecycle, status, () => Deferred.doneUnsafe(shutdown, Effect.void)) + } + const state = yield* status.current + const app = yield* Ref.get(application) + const ready = state.type === "ready" && Option.isSome(app) + if ((!ready || !hasPtyConnectTicketURL(url)) && !(yield* authorizedRequest(request, auth))) return unauthorized() + if (ready) return yield* app.value + return unavailable(state) + }) +} + +function unauthorized() { + return HttpServerResponse.empty({ + status: 401, + headers: { "www-authenticate": 'Basic realm="Secure Area"' }, + }) +} + +const control = Effect.fnUntraced(function* ( + request: HttpServerRequest.HttpServerRequest, + route: "health" | "stop", + status: Status.Interface, + stop: () => void, +) { + if (route === "health") return yield* healthResponse(status) + const body = yield* request.json.pipe(Effect.option) + const input = Option.isSome(body) ? Schema.decodeUnknownOption(ServiceStatus.StopRequest)(body.value) : Option.none() + if (Option.isNone(input)) return HttpServerResponse.jsonUnsafe({ code: "invalid_request" }, { status: 400 }) + const accepted = yield* status.requestStop(input.value) + if (accepted) { + const response = NodeHttpServerRequest.toServerResponse(request) + yield* Effect.sync(() => { + const complete = () => { + response.off("finish", complete) + response.off("close", complete) + stop() + } + response.once("finish", complete) + response.once("close", complete) + }) + } + return HttpServerResponse.jsonUnsafe({ accepted }) +}) + +const healthResponse = Effect.fnUntraced(function* (status: Status.Interface) { + const health = yield* status.health + return HttpServerResponse.jsonUnsafe(health, { + status: health.status.type === "ready" ? 200 : 503, + headers: + health.status.type === "starting" || health.status.type === "stopping" ? { "retry-after": "1" } : undefined, + }) +}) + +function unavailable(status: ServiceStatus.State) { + if (status.type === "failed") + return HttpServerResponse.jsonUnsafe( + { code: "service_failed", message: status.message, action: status.action }, + { status: 503 }, + ) + return HttpServerResponse.jsonUnsafe( + { code: status.type === "stopping" ? "service_stopping" : "service_starting" }, + { status: 503, headers: { "retry-after": "1" } }, ) } @@ -77,12 +214,8 @@ function bind(options: Options, port: number) { * suspends its own active Sessions on graceful shutdown. Suspension runs while the drains are still * alive: connections close first, this finalizer runs next, and Session execution teardown follows. */ -function restartContinuity(restart: SessionRestart.Interface) { - return Layer.effectDiscard( - Effect.gen(function* () { - yield* Effect.forkScoped(restart.resumeSuspendedSessions) - // Registered after the fork so suspension observes still-running resumed drains during teardown. - yield* Effect.addFinalizer(() => restart.suspendActiveSessions) - }), - ) -} +const installRestartContinuity = Effect.fnUntraced(function* (restart: SessionRestart.Interface) { + yield* Effect.forkScoped(restart.resumeSuspendedSessions) + // Registered after the fork so suspension observes still-running resumed drains during teardown. + yield* Effect.addFinalizer(() => restart.suspendActiveSessions) +}) diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index 3f44ec3bf1..32647df3ff 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -120,6 +120,5 @@ function driveEnabled() { return !!process.env.OPENCODE_DRIVE } -export const routes = createRoutes() - -export const webHandler = () => HttpRouter.toWebHandler(routes.pipe(Layer.provide(HttpServer.layerServices))) +export const webHandler = () => + HttpRouter.toWebHandler(createRoutes().pipe(Layer.provide(HttpServer.layerServices))) diff --git a/packages/server/src/service-status.ts b/packages/server/src/service-status.ts new file mode 100644 index 0000000000..95eaf2f8c6 --- /dev/null +++ b/packages/server/src/service-status.ts @@ -0,0 +1,57 @@ +export * as Status from "./service-status" + +import { InstallationVersion } from "@opencode-ai/core/installation/version" +import { ServiceStatus } from "@opencode-ai/protocol/groups/health" +import { Effect, Ref } from "effect" + +export interface Interface { + readonly health: Effect.Effect + readonly current: Effect.Effect + readonly ready: Effect.Effect + readonly fail: (failure: { readonly message: string; readonly action: string }) => Effect.Effect + readonly beginStopping: (targetVersion?: string) => Effect.Effect + readonly requestStop: (request: ServiceStatus.StopRequest) => Effect.Effect +} + +export const make = Effect.fnUntraced(function* (options: { + readonly instanceID: string + readonly managed: boolean + readonly initial?: ServiceStatus.State +}) { + const current = yield* Ref.make(options.initial ?? ({ type: "starting" } satisfies ServiceStatus.State)) + const transitionToStopping = (targetVersion?: string) => + Ref.update(current, (status) => { + if (status.type === "stopping") return status + return ( + targetVersion === undefined || targetVersion === InstallationVersion + ? { type: "stopping" } + : { type: "stopping", targetVersion } + ) satisfies ServiceStatus.State + }) + + return { + current: Ref.get(current), + health: Effect.gen(function* () { + return { + healthy: true as const, + version: InstallationVersion, + pid: process.pid, + instanceID: options.instanceID, + status: yield* Ref.get(current), + } + }), + ready: Ref.update(current, (status) => + status.type === "starting" ? ({ type: "ready" } satisfies ServiceStatus.State) : status, + ), + fail: (failure) => + Ref.update(current, (status) => + status.type === "starting" ? ({ type: "failed", ...failure } satisfies ServiceStatus.State) : status, + ), + beginStopping: transitionToStopping, + requestStop: (request) => { + if (!options.managed || request.instanceID !== options.instanceID) + return Effect.succeed(false) + return transitionToStopping(request.targetVersion).pipe(Effect.as(true)) + }, + } satisfies Interface +}) diff --git a/packages/server/test/service-status.test.ts b/packages/server/test/service-status.test.ts new file mode 100644 index 0000000000..c8afc42c54 --- /dev/null +++ b/packages/server/test/service-status.test.ts @@ -0,0 +1,51 @@ +import { InstallationVersion } from "@opencode-ai/core/installation/version" +import { expect } from "bun:test" +import { Effect } from "effect" +import { it } from "../../core/test/lib/effect" +import { Status } from "../src/service-status" + +it.effect("moves from starting to ready", () => + Effect.gen(function* () { + const status = yield* Status.make({ instanceID: "one", managed: false }) + expect(yield* status.current).toEqual({ type: "starting" }) + yield* status.ready + expect(yield* status.current).toEqual({ type: "ready" }) + }), +) + +it.effect("keeps a startup failure until shutdown", () => + Effect.gen(function* () { + const status = yield* Status.make({ instanceID: "one", managed: true }) + yield* status.fail({ message: "Could not open the database.", action: "Check the database path." }) + yield* status.ready + yield* status.fail({ message: "Different failure.", action: "Different action." }) + expect(yield* status.current).toEqual({ + type: "failed", + message: "Could not open the database.", + action: "Check the database path.", + }) + }), +) + +it.effect("stops only the addressed managed instance", () => + Effect.gen(function* () { + const status = yield* Status.make({ instanceID: "one", managed: true }) + + expect(yield* status.requestStop({ instanceID: "other", targetVersion: "next" })).toBe(false) + expect(yield* status.current).toEqual({ type: "starting" }) + expect(yield* status.requestStop({ instanceID: "one", targetVersion: "next" })).toBe(true) + expect(yield* status.requestStop({ instanceID: "one", targetVersion: InstallationVersion })).toBe(true) + expect(yield* status.current).toEqual({ type: "stopping", targetVersion: "next" }) + }), +) + +it.effect("preserves the original stopping target after shutdown begins", () => + Effect.gen(function* () { + const status = yield* Status.make({ instanceID: "one", managed: true }) + + yield* status.beginStopping("next") + expect(yield* status.current).toEqual({ type: "stopping", targetVersion: "next" }) + expect(yield* status.requestStop({ instanceID: "one" })).toBe(true) + expect(yield* status.current).toEqual({ type: "stopping", targetVersion: "next" }) + }), +) diff --git a/packages/tui/src/app.tsx b/packages/tui/src/app.tsx index db970e3ef7..0611941a9e 100644 --- a/packages/tui/src/app.tsx +++ b/packages/tui/src/app.tsx @@ -139,7 +139,7 @@ const appBindingCommands = [ export type TuiInput = { server: { endpoint: Service.Endpoint - reconnect?: (attempt: number) => Promise + reconnect?: (onStatus: (status: Service.Status) => void, signal: AbortSignal) => Promise reload?: () => Promise } args: Args @@ -186,8 +186,8 @@ export const run = Effect.fn("Tui.run")(function* (input: TuiInput) { const handoff = input.terminalHandoff ? yield* Effect.promise(input.terminalHandoff) : undefined const reconnectEndpoint = input.server.reconnect const reconnect = reconnectEndpoint - ? async (attempt: number) => { - const endpoint = await reconnectEndpoint(attempt) + ? async (onStatus: (status: Service.Status) => void, signal: AbortSignal) => { + const endpoint = await reconnectEndpoint(onStatus, signal) const next = { baseUrl: endpoint.url, headers: Service.headers(endpoint) } return { api: OpenCode.make(next), @@ -1121,7 +1121,7 @@ function App(props: { pair?: DialogPairCredentials }) { - + ) diff --git a/packages/tui/src/component/reconnecting.tsx b/packages/tui/src/component/reconnecting.tsx index 2ebbda9909..101e5d1775 100644 --- a/packages/tui/src/component/reconnecting.tsx +++ b/packages/tui/src/component/reconnecting.tsx @@ -1,9 +1,11 @@ +import type { Service } from "@opencode-ai/client/effect" import { Show } from "solid-js" import { useTheme } from "../context/theme" import { Spinner } from "./spinner" -export function Reconnecting(props: { attempt: number; error?: string }) { +export function Reconnecting(props: { status?: Service.Status }) { const theme = useTheme().theme + const copy = () => reconnectingCopy(props.status) return ( - - Connection lost - Reconnecting to server... - Attempt {props.attempt} - - - {props.error} - + + {copy().message}}> + {copy().message} + + {(detail) => ( + + {detail()} + + )} + + + {(action) => ( + + {action()} + + )} + ) } + +export function reconnectingCopy(status?: Service.Status) { + if (status?.type === "starting") + return { + loading: true, + message: status.version ? `Starting OpenCode ${status.version}...` : "Starting background service...", + } + if (status?.type === "stopping") + return { + loading: true, + message: status.targetVersion ? `Updating to ${status.targetVersion}...` : "Restarting background service...", + } + if (status?.type === "failed") + return { loading: false, message: "Background service failed", detail: status.message, action: status.action } + if (status?.type === "unresponsive") + return { + loading: false, + message: "Background service is not responding", + action: "Run `opencode service restart` to recover it.", + } + return { loading: true, message: "Waiting for background service..." } +} diff --git a/packages/tui/src/context/client.tsx b/packages/tui/src/context/client.tsx index 04e347900a..6994f6a6f7 100644 --- a/packages/tui/src/context/client.tsx +++ b/packages/tui/src/context/client.tsx @@ -1,7 +1,9 @@ import type { OpenCodeClient, OpenCodeEvent } from "@opencode-ai/client" +import type { Service } from "@opencode-ai/client/effect" import { createGlobalEmitter } from "@solid-primitives/event-bus" -import { onCleanup, onMount } from "solid-js" +import { createSignal, onCleanup, onMount } from "solid-js" import { createStore } from "solid-js/store" +import { errorMessage } from "../util/error" import { createSimpleContext } from "./helper" import { useLog } from "./log" @@ -24,7 +26,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( name: "Client", init: (props: { api: OpenCodeClient - reconnect?: (attempt: number) => Promise<{ api: OpenCodeClient }> + reconnect?: (onStatus: (status: Service.Status) => void, signal: AbortSignal) => Promise<{ api: OpenCodeClient }> // Stops and starts the managed service; present only in service mode. reload?: () => Promise }) => { @@ -41,6 +43,7 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( status: "connecting", attempt: 0, }) + const [service, setService] = createSignal() let stream: AbortController | undefined function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) { @@ -51,43 +54,37 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( function start() { stream?.abort() const controller = new AbortController() - let connected!: () => void - const ready = new Promise((resolve) => { - connected = resolve - }) stream = controller void (async () => { let attempt = 0 while (!abort.signal.aborted && !controller.signal.aborted) { - const connection = new AbortController() - const cancel = () => connection.abort(controller.signal.reason) - const timeout = setTimeout( - () => connection.abort(new Error("Timed out connecting to server")), - connectTimeout, - ) + let connectedAt: number | undefined + const request = new AbortController() + const cancel = () => request.abort(controller.signal.reason) + const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout) controller.signal.addEventListener("abort", cancel, { once: true }) const error = await (async () => { record(attempt === 0 ? "connecting" : "reconnecting", attempt) log.info("event stream connecting", { attempt }) - const iterator = api.event.subscribe({ signal: connection.signal })[Symbol.asyncIterator]() + const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]() const first = await iterator.next() - if (abort.signal.aborted || controller.signal.aborted) return + if (abort.signal.aborted || controller.signal.aborted) return undefined if (first.done) - return connection.signal.reason instanceof Error - ? connection.signal.reason + return request.signal.reason instanceof Error + ? request.signal.reason : new Error("Event stream disconnected") if (first.value.type !== "server.connected") return new Error("Event stream did not start with server.connected") clearTimeout(timeout) record("connected", attempt) - attempt = 0 + connectedAt = Date.now() log.info("event stream connected") events.emit(first.value.type, first.value) setConnection({ status: "connected", attempt: 0, error: undefined }) - connected() + setService(undefined) while (!abort.signal.aborted && !controller.signal.aborted) { const event = await iterator.next() - if (abort.signal.aborted || controller.signal.aborted) return + if (abort.signal.aborted || controller.signal.aborted) return undefined if (event.done) return new Error("Event stream disconnected") if ("durable" in event.value) log.debug("event", { @@ -97,42 +94,47 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( }) events.emit(event.value.type, event.value) } + return undefined })() .catch((error) => error) .finally(() => { + request.abort() clearTimeout(timeout) controller.signal.removeEventListener("abort", cancel) }) if (abort.signal.aborted || controller.signal.aborted) return + if (connectedAt !== undefined && Date.now() - connectedAt >= 1_000) attempt = 0 attempt += 1 - const message = error instanceof Error ? error.message : String(error) + const message = errorMessage(error) record("disconnected", attempt, message) log.info("event stream disconnected", { attempt, error: message, }) + setConnection({ status: "reconnecting", attempt, error: message }) // Re-resolve the transport before retrying: the server may have // moved (service restarted on a new port) or need starting. Static // transports (--server, standalone) resolve to the same address. if (props.reconnect) { - const next = await props.reconnect(attempt).catch(() => undefined) + const next = await props.reconnect(setService, controller.signal).catch((error) => { + if (!controller.signal.aborted) + log.info("server resolution failed", { + attempt, + error: errorMessage(error), + }) + }) if (abort.signal.aborted || controller.signal.aborted) return if (next) { api = next.api + if (attempt === 1) continue } } - setConnection({ - status: "reconnecting", - attempt, - error: message, - }) await wait(1_000, controller.signal) } })() - return ready } - onMount(() => void start()) + onMount(start) onCleanup(() => { abort.abort() stream?.abort() @@ -157,6 +159,9 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext( error() { return connection.error }, + service() { + return service() + }, internal: { history() { return history.slice() diff --git a/packages/tui/test/cli/tui/reconnecting.test.ts b/packages/tui/test/cli/tui/reconnecting.test.ts new file mode 100644 index 0000000000..b8cca1ccc9 --- /dev/null +++ b/packages/tui/test/cli/tui/reconnecting.test.ts @@ -0,0 +1,31 @@ +import { expect, test } from "bun:test" +import { reconnectingCopy } from "../../../src/component/reconnecting" + +test("describes service status without transport diagnostics", () => { + expect(reconnectingCopy({ type: "starting", version: "2.0.0" })).toEqual({ + loading: true, + message: "Starting OpenCode 2.0.0...", + }) + expect(reconnectingCopy({ type: "stopping", targetVersion: "2.0.0" })).toEqual({ + loading: true, + message: "Updating to 2.0.0...", + }) + expect( + reconnectingCopy({ + type: "failed", + message: "Could not open the database.", + action: "Check the service logs.", + }), + ).toEqual({ + loading: false, + message: "Background service failed", + detail: "Could not open the database.", + action: "Check the service logs.", + }) + expect(reconnectingCopy({ type: "unresponsive" })).toEqual({ + loading: false, + message: "Background service is not responding", + action: "Run `opencode service restart` to recover it.", + }) + expect(JSON.stringify(reconnectingCopy())).not.toMatch(/Attempt|ECONNREFUSED|Event stream disconnected/) +}) diff --git a/packages/tui/test/cli/tui/use-event.test.tsx b/packages/tui/test/cli/tui/use-event.test.tsx index 68b5f275c4..542f33b3d1 100644 --- a/packages/tui/test/cli/tui/use-event.test.tsx +++ b/packages/tui/test/cli/tui/use-event.test.tsx @@ -1,6 +1,7 @@ /** @jsxImportSource @opentui/solid */ import { describe, expect, test } from "bun:test" import type { OpenCodeClient, OpenCodeEvent } from "@opencode-ai/client" +import type { Service } from "@opencode-ai/client/effect" import { testRender } from "@opentui/solid" import { onMount } from "solid-js" import { ProjectProvider, useProject } from "../../../src/context/project" @@ -53,7 +54,7 @@ function update(version: string): OpenCodeEvent { } async function mount( - reconnect?: (attempt: number) => Promise<{ api: OpenCodeClient }>, + reconnect?: (onStatus: (status: Service.Status) => void, signal: AbortSignal) => Promise<{ api: OpenCodeClient }>, log?: LogSink, ) { const events = createEventStream() @@ -194,8 +195,8 @@ describe("useEvent", () => { const replacementEvents = createEventStream() const replacementCalls = createFetch(undefined, replacementEvents) const replacement = { api: createApi(replacementCalls.fetch) } - const { app, events, client, seen } = await mount(async (attempt) => { - attempts.push(attempt) + const { app, events, client, seen } = await mount(async () => { + attempts.push(attempts.length + 1) return replacement }) @@ -246,4 +247,106 @@ describe("useEvent", () => { app.renderer.destroy() } }) + + test("backs off when a resolved event stream keeps failing", async () => { + let calls = 0 + const encoder = new TextEncoder() + const replacementCalls = createFetch((url) => { + if (url.pathname !== "/api/event") return undefined + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + encoder.encode('data: {"id":"evt_connected","type":"server.connected","data":{}}\n\n'), + ) + controller.close() + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ) + }) + const replacement = { + api: createApi(replacementCalls.fetch), + } + const { app, events, client } = await mount(async () => { + calls += 1 + return replacement + }) + + try { + await wait(() => client.connection.status() === "connected") + events.disconnect() + await Promise.race([ + wait(() => calls === 2), + Bun.sleep(500).then(() => { + throw new Error("resolved event stream did not retry immediately") + }), + ]) + await Bun.sleep(200) + expect(calls).toBe(2) + } finally { + app.renderer.destroy() + } + }) + + test("reports service status while endpoint resolution is pending", async () => { + const replacementEvents = createEventStream() + const replacement = { api: createApi(createFetch(undefined, replacementEvents).fetch) } + let report!: (status: Service.Status) => void + let resolve!: (value: typeof replacement) => void + const endpoint = new Promise((done) => { + resolve = done + }) + const { app, events, client } = await mount(async (onStatus) => { + report = onStatus + onStatus({ type: "starting", version: "2.0.0" }) + return endpoint + }) + + try { + await wait(() => client.connection.status() === "connected") + events.disconnect() + await wait( + () => client.connection.status() === "reconnecting" && client.connection.service()?.type === "starting", + ) + expect(client.connection.service()).toEqual({ type: "starting", version: "2.0.0" }) + + report({ type: "failed", message: "Could not open the database.", action: "Check the service logs." }) + await wait(() => client.connection.service()?.type === "failed") + expect(client.connection.service()).toEqual({ + type: "failed", + message: "Could not open the database.", + action: "Check the service logs.", + }) + + resolve(replacement) + await wait(() => client.connection.status() === "connected") + expect(client.connection.service()).toBeUndefined() + } finally { + app.renderer.destroy() + } + }) + + test("cancels pending endpoint resolution on cleanup", async () => { + let aborted = false + const { app, events, client } = await mount( + (_onStatus, signal) => + new Promise((_, reject) => { + signal.addEventListener( + "abort", + () => { + aborted = true + reject(signal.reason) + }, + { once: true }, + ) + }), + ) + + await wait(() => client.connection.status() === "connected") + events.disconnect() + await wait(() => client.connection.status() === "reconnecting") + app.renderer.destroy() + await wait(() => aborted) + }) })