diff --git a/bun.lock b/bun.lock index 58927508e4..9f3d31e66c 100644 --- a/bun.lock +++ b/bun.lock @@ -880,7 +880,7 @@ "name": "@opencode-ai/simulation", "version": "1.17.13", "dependencies": { - "@fontsource/adwaita-mono": "5.2.1", + "@fontsource/commit-mono": "5.2.5", "@napi-rs/canvas": "1.0.2", "@opencode-ai/core": "workspace:*", "@opencode-ai/llm": "workspace:*", @@ -1765,7 +1765,7 @@ "@floating-ui/utils": ["@floating-ui/utils@0.2.11", "", {}, "sha512-RiB/yIh78pcIxl6lLMG0CgBXAZ2Y0eVHqMPYugu+9U0AeT6YBeiJpf7lbdJNIugFP5SIjwNRgo4DhR1Qxi26Gg=="], - "@fontsource/adwaita-mono": ["@fontsource/adwaita-mono@5.2.1", "", {}, "sha512-6+Q1UIvklJ9REijs6kv7YlRNt6yktRj0iW8H69YIugdD9P2h3eIX1AB8/9ICMfpVyVeywlsrCXg82y/LfRrjyg=="], + "@fontsource/commit-mono": ["@fontsource/commit-mono@5.2.5", "", {}, "sha512-htX8yQWtiPt5L1Hzh4sirvfUJT2+KYiquDB/Q2sY2tWQYplpBUOD5zHnIM3k36Hnm4V+JIIqA/wmwupSQ68WjA=="], "@fontsource/ibm-plex-mono": ["@fontsource/ibm-plex-mono@5.2.5", "", {}, "sha512-G09N3GfuT9qj3Ax2FDZvKqZttzM3v+cco2l8uXamhKyXLdmlaUDH5o88/C3vtTHj2oT7yRKsvxz9F+BXbWKMYA=="], diff --git a/packages/opencode/specs/simulation/simulated-network-llm.md b/packages/opencode/specs/simulation/simulated-network-llm.md index f09685b26a..29ccc5d4a8 100644 --- a/packages/opencode/specs/simulation/simulated-network-llm.md +++ b/packages/opencode/specs/simulation/simulated-network-llm.md @@ -4,7 +4,7 @@ Status: design for the Phase 2 network and LLM items in `simulation-phases.md`. ## Summary -Simulation replaces the `HttpClient.HttpClient` platform node with a simulated network. The LLM is not a separate fake: it is one registered route in that network (`api.openai.com`), answered by the **external driver** over the existing control WebSocket. When the app issues a provider request, the backend forwards it to the driver and the driver streams response chunks back. There is no enqueueing and no scripted-response store; the driver is the model. +Simulation replaces the `HttpClient.HttpClient` platform node with a simulated network. The LLM is not replaced: an OpenAI route intercepts the real provider request and delegates its response to a **simulated model provider** controlled by the external driver. There is no server-side response script or replay adapter; the driver decides what the provider returns. Everything above the HTTP boundary runs real: catalog and auth resolution, `LLMClient`, request body construction, SSE framing, the OpenAI protocol event schema, the `step` state machine, `Lifecycle` grammar, tool-argument accumulation, the session runner, tools, and permissions. @@ -29,12 +29,12 @@ Replacing `httpClient` (already a `LayerNode` in `app-node-platform.ts`, already ### 1. Simulated network (`packages/simulation/src/backend/network.ts`) -Replaces `httpClient` in `simulationReplacements`. An in-memory route table: +Replaces `httpClient` in `simulationReplacements`. Each acquired network run owns its route table and bounded request log: -- `register(matcher, responder)` where matcher is method + URL pattern and responder is `(HttpClientRequest) => Effect`. +- `make(routes)` constructs one isolated client and log; routes are ordinary matchers supplied at acquisition. - Unknown requests fail loudly with a typed simulation error (spec: deny unknown external network by default). - Optional loopback allowance for the app's own server is not required server-side (the server does not call itself over HTTP); revisit if a consumer needs it. -- Every request/response summary is traced. +- Every request summary is timestamped through Effect `Clock` and retained only for that run. ### 2. OpenAI endpoint route (`packages/simulation/src/backend/openai.ts`) @@ -42,36 +42,44 @@ Registered in the network at startup for `POST {DEFAULT_BASE_URL}{PATH}` from `p On request: -1. Allocate an exchange id. Parse the real OpenAI request body (available to the driver for assertions). -2. Publish a `request` record to the LLM exchange service (below) and create a chunk `Queue`. -3. Return `HttpClientResponse` with `content-type: text/event-stream` whose body stream reads from the queue, encoding each item as an SSE `data:` frame, terminated by `[DONE]`. +1. Parse the real OpenAI request body, which remains available to the driver for assertions. +2. Call `SimulatedProvider.Service.stream({ url, body })`. +3. Encode the returned provider response events as SSE `data:` frames and terminate a finished response with `[DONE]`. Chunks are constructed through the `OpenAIChatEvent` schema so drift in the protocol schema breaks the build, not the runtime. -The response stream is interruptible like a real HTTP response: if the runner cancels (user interrupt), the exchange closes and the driver is notified. +The response stream is interruptible like a real HTTP response. If the runner cancels, the provider invocation is removed and later driver commands for its id fail. -### 3. LLM exchange service (`packages/simulation/src/backend/llm-exchange.ts`) +### 3. Simulated provider (`packages/simulation/src/backend/simulated-provider.ts`) -Process-global simulation service owning pending exchanges: +The OpenAI route sees one Effect service: -``` -Exchange = { id, body, queue: Queue, deferred lifecycle } +```ts +interface SimulatedProvider { + stream(request: ProviderRequest): Stream +} ``` -- `requests()` — stream of newly opened exchanges (consumed by the control route). -- `push(id, item)` — append one response item to an open exchange. -- `finish(id, reason)` / `fail(id, failure)` — terminate the exchange. -- Exchanges that receive no driver within a configurable timeout fail the provider request with a simulation error (surfaces in the real provider-error path). +`SimulatedProvider.layerDrive({ endpoint })` owns the Drive adapter in one Effect scope: + +- Pending provider invocations and response queues. +- Late controller attachment and pending-invocation replay. +- The backend control WebSocket and its request fibers. +- Stream interruption, explicit disconnect, finish, and scope cleanup. + +Invocation ids, queues, controller attachment, and WebSocket commands remain private to `layerDrive`. The OpenAI route only sees a provider request producing a response stream. ### 4. Backend control WebSocket (simulation-gated) Started when `OPENCODE_DRIVE` names a registry manifest: a loopback JSON-RPC 2.0 WebSocket at that manifest's exact backend endpoint, hosted by the backend process. Drivers connect to it directly — the standalone topology has exactly one backend per TUI, so there is no proxying through the frontend. This socket is also the headless-simulation interface: it works with no TUI at all. -Server -> driver notification (after `llm.attach`; pending exchanges are replayed on attach so late-attaching drivers miss nothing): +The backend and frontend control sockets share one scoped Effect adapter. It owns the Bun server, a bounded sequential message queue, its worker fiber, schema-based JSON decoding, and shutdown ordering. + +Server -> driver notification (after `llm.attach`; pending invocations are replayed on attach so late-attaching drivers miss nothing): ``` { "jsonrpc": "2.0", "method": "llm.request", - "params": { "id": "ex_1", "url": "...", "body": { ...openai request body... } } } + "params": { "id": "inv_1", "url": "...", "body": { ...openai request body... } } } ``` Driver -> server methods: @@ -79,8 +87,9 @@ Driver -> server methods: ``` llm.attach subscribe to llm.request notifications llm.chunk { id, items: Item[] } append response items -llm.finish { id, reason?: "stop" | ... } finish the exchange -llm.pending list open exchanges +llm.finish { id, reason?: "stop" | ... } finish the invocation +llm.disconnect { id } fail the provider response stream +llm.pending list pending invocations network.log simulated network request log ``` @@ -102,13 +111,13 @@ Failure injection (`llm.fail`: HTTP status instead of SSE) is specced but not ye A driver manages two loopback WebSocket connections: - TUI control server (manifest `endpoints.ui`) — UI state, actions, render, trace. -- Backend control server (manifest `endpoints.backend`) — LLM exchanges, network log. +- Backend control server (manifest `endpoints.backend`) — simulated provider invocations. The network request log remains run-local diagnostic state. Both speak the same JSON-RPC shape. Headless drivers use only the backend socket plus the normal HTTP API. Multiple drivers are out of scope; last attach wins. ### 6. Pacing and the clock -No server-side pacing by default: the driver controls timing by when it sends chunks, which is the point of driver-in-the-loop. A convenience `llm.chunk` option `{ delayMs }` may sleep via `Effect.sleep` between items server-side; because that uses the fiber `Clock`, scoping a controllable clock to the exchange stream (`Stream.provideService(Clock.Clock, simClock)`) remains available for deterministic replay without touching app time. Defer until replay work needs it. +No server-side pacing exists. The driver controls timing by deciding when to send chunks. ### 7. Catalog and auth seeding @@ -123,14 +132,14 @@ driver TUI drive server backend + drive WS | |-- (normal app HTTP) ---->| session runner starts | | | llm.stream -> HttpClient | | | simulated network matches openai route - |<================== llm.request {ex_1} ===============| exchange ex_1 opened - |-- llm.chunk {ex_1,[...]} ============================>| SSE frames flow into the real - |-- llm.chunk {ex_1,[...]} ============================>| decode -> step -> LLMEvents -> - |-- llm.finish {ex_1} =================================>| runner publishes, TUI renders + |<================= llm.request {inv_1} ================| provider invocation inv_1 opened + |-- llm.chunk {inv_1,[...]} ===========================>| SSE frames flow into the real + |-- llm.chunk {inv_1,[...]} ===========================>| decode -> step -> LLMEvents -> + |-- llm.finish {inv_1} ================================>| runner publishes, TUI renders | | | | (if toolCall was sent: runner executes the real tool against the - | fake filesystem, then issues the next provider turn -> new exchange - | ex_2 -> driver decides the next response) + | fake filesystem, then starts the next model invocation -> inv_2 + | -> driver decides the next provider response) ``` The driver observes the TUI through `ui.state` while chunks stream, so mid-stream UI assertions need no clock control at all: the driver simply has not sent the rest yet. @@ -138,13 +147,13 @@ The driver observes the TUI through `ui.state` while chunks stream, so mid-strea ## Implementation order 1. `network.ts`: simulated `HttpClient` + route table + deny-unknown + trace. Replace `httpClient` in `simulationReplacements`. -2. `llm-exchange.ts` + `openai.ts`: exchange service and the OpenAI SSE route (schema-constructed chunks, `[DONE]`, interruption). -3. `control.ts`: backend-hosted control WebSocket (`llm.attach|chunk|finish|pending`, `network.log`), started when the simulation module loads. +2. `simulated-provider.ts` + `openai.ts`: scoped Drive-controlled provider and the OpenAI SSE route (schema-constructed chunks, `[DONE]`, interruption). +3. `SimulatedProvider.layerDrive`: backend-hosted control WebSocket (`llm.attach|chunk|finish|disconnect|pending`), acquired only when `OPENCODE_DRIVE` is set. 4. Config seeding for the sim provider; end-to-end verification via `packages/server/script/e2e-sim.ts` (headless) and `packages/tui/script/sim-llm-driver.ts` (TUI + backend sockets). -5. Trace records for network and LLM exchange activity. +5. Trace records for network and simulated provider activity. ## Consequences - No enqueue/script store to keep consistent; the driver is the single source of model behavior. -- Deterministic tests write drivers (respond to `llm.request` programmatically) instead of pre-baked scripts; replay (Phase 4) records exchanges and replays them as an automatic driver. +- Deterministic tests write drivers that respond to `llm.request` programmatically instead of adding a second provider implementation. - Provider-coupling is confined to `openai.ts` (one wire encoder against a schema that lives in the repo); a second simulated provider (e.g. Anthropic) is another route file if ever needed. diff --git a/packages/opencode/specs/simulation/simulation-phases.md b/packages/opencode/specs/simulation/simulation-phases.md index c8a470a449..74fd92889b 100644 --- a/packages/opencode/specs/simulation/simulation-phases.md +++ b/packages/opencode/specs/simulation/simulation-phases.md @@ -64,14 +64,16 @@ Implementation checklist: - [x] Verify end to end: `opencode serve` boots with `OPENCODE_SIMULATE=1` + `OPENCODE_SIMULATE_STATE` + path/DB env seams (`OPENCODE_CONFIG_DIR`, `OPENCODE_TEST_HOME`, `OPENCODE_DB=:memory:`); `fs.list`/`fs.read` observe only seeded in-memory files; the anchor directory on the host remains empty after the run. - [ ] Create the anchor directory + `chdir` + env seam setup automatically in CLI startup when simulation mode is enabled (currently set manually by the runner; a full run needs `OPENCODE_SIMULATE_STATE`, `OPENCODE_CONFIG_DIR`, `OPENCODE_TEST_HOME`, `OPENCODE_DB=:memory:`, and `XDG_*_HOME` pointed into the anchor, plus Bun's `--preload=@opentui/solid/preload` when launched outside `packages/cli`). - [ ] Assert the anchor directory is still empty at the end of the run (KV/log/flock still write through real XDG paths; they are contained in the anchor by the env seams but not yet in-memory). -- [x] Add simulated network registry (`packages/simulation/src/backend/network.ts`): replaces the `httpClient` platform node, resolves all outbound HTTP against an in-memory route table, denies unknown destinations loudly, and keeps a bounded request log (design: `simulated-network-llm.md`). -- [x] Add driver-answered LLM as an OpenAI route in the simulated network (`openai.ts` + `llm-exchange.ts`): provider requests open exchanges; the driver streams chunks back which are encoded as real OpenAI Chat SSE (schema-checked against `OpenAIChatEvent`) and consumed by the real protocol pipeline. No enqueue store — the driver is the model. -- [x] Add backend-hosted drive control WebSocket (`control.ts`): JSON-RPC at the named manifest's backend endpoint, started when `OPENCODE_DRIVE` is set. Drivers connect directly (standalone topology — no frontend proxy): `llm.attach` (replays pending exchanges), `llm.chunk`, `llm.finish`, `llm.pending`, `network.log`; `llm.request` notifications push opened exchanges. This is also the headless-simulation interface. Drivers manage the manifest's UI endpoint for UI control and backend endpoint for LLM/network control. +- [x] Add run-local simulated network (`packages/simulation/src/backend/network.ts`): replaces the `httpClient` platform node, resolves outbound HTTP against routes supplied at acquisition, denies unknown destinations loudly, and keeps an isolated bounded request log timestamped through Effect `Clock` (design: `simulated-network-llm.md`). +- [x] Add a simulated model provider behind the OpenAI route (`simulated-provider.ts` + `openai.ts`): real provider requests call `SimulatedProvider.Service.stream`; the Drive adapter streams response events back as schema-checked OpenAI Chat SSE consumed by the real protocol pipeline. +- [x] Scope the backend Drive control WebSocket, pending provider invocations, queues, and request fibers to `SimulatedProvider.layerDrive`. JSON-RPC remains at the named manifest's backend endpoint: `llm.attach` replays pending invocations; `llm.chunk`, `llm.finish`, `llm.disconnect`, and `llm.pending` control them; `llm.request` reports provider-native requests. +- [x] Scope the frontend Drive control WebSocket, request queue, renderer, and optional recording timeline to the TUI Effect scope. Server shutdown and request interruption precede renderer destruction; timeline finalization runs last and remains explicitly finishable through `ui.recording.finish`. +- [x] Decode Drive manifests through Effect `Config`, `FileSystem`, and `Schema`, with typed config, not-found, read, and decode failures. - [x] Answer `https://models.dev/api.json` with an empty catalog in the simulated network; providers come from seeded config (`opencode.json` in the snapshot defines an openai-compatible provider with a dummy `apiKey`, which passes the catalog availability gate and resolves onto the real openai-chat route). - [x] Fix `buildLocationServiceMap` to apply replacements when compiling hoisted global nodes; platform-node replacements (filesystem, httpClient) were silently ignored inside hoisted globals. - [x] Verify end to end headless (real route stack in-process + backend control WS: prompt -> `llm.request` -> driver chunks -> assistant message contains driver text; script: `packages/server/script/e2e-sim.ts`) and through the TUI (fake renderer, both sockets: type + submit via TUI WS, answer `llm.request` via backend WS, assistant reply rendered on screen; script: `packages/tui/script/sim-llm-driver.ts`). - [ ] Add simulated process registry (shell via `just-bash`, minimal fake `git`, deny unsupported spawns). -- [ ] Trace filesystem, process, and LLM exchange activity (network requests are traced in the backend network log ring buffer; LLM exchange trace records moved out with the frontend proxy and need re-adding on the backend control server). +- [ ] Trace filesystem, process, and simulated provider activity (network requests are traced in the backend network log ring buffer; provider trace records still need adding on the backend control server). Scope: diff --git a/packages/plugin/src/v2/tui/context.ts b/packages/plugin/src/v2/tui/context.ts index 5e33b03079..f4a2fb9a39 100644 --- a/packages/plugin/src/v2/tui/context.ts +++ b/packages/plugin/src/v2/tui/context.ts @@ -24,7 +24,8 @@ import type { JSX } from "@opentui/solid" interface LocationCollection { list(location?: LocationRef): Value[] | undefined - refresh(location?: LocationRef): Promise + sync(location?: LocationRef): Promise + invalidate(location?: LocationRef): void } export interface Data { @@ -42,37 +43,45 @@ export interface Data { status(sessionID: string): "idle" | "running" readonly pending: { list(sessionID: string): SessionPendingInfo[] - refresh(sessionID: string): Promise + sync(sessionID: string): Promise + invalidate(sessionID: string): void } - refresh(sessionID: string): Promise + sync(sessionID: string): Promise + invalidate(sessionID: string): void readonly message: { list(sessionID: string): SessionMessageInfo[] get(sessionID: string, messageID: string): SessionMessageInfo | undefined - refresh(sessionID: string): Promise + sync(sessionID: string): Promise + invalidate(sessionID: string): void } readonly permission: { list(sessionID: string): PermissionV2Request[] | undefined - refresh(sessionID: string): Promise + sync(sessionID: string): Promise + invalidate(sessionID: string): void } readonly form: { list(sessionID: string, location?: LocationRef): Array | undefined - refresh(sessionID: string, location?: LocationRef): Promise + sync(sessionID: string, location?: LocationRef): Promise + invalidate(sessionID: string, location?: LocationRef): void } } readonly project: { readonly permission: { list(projectID: string): PermissionSavedInfo[] | undefined - refresh(projectID: string): Promise + sync(projectID: string): Promise + invalidate(projectID: string): void } } readonly shell: { list(location?: LocationRef): ShellInfo[] get(id: string): ShellInfo | undefined - refresh(location?: LocationRef): Promise + sync(location?: LocationRef): Promise + invalidate(location?: LocationRef): void } readonly location: { default(): LocationRef - refresh(location?: LocationRef): Promise + sync(location?: LocationRef): Promise + invalidate(location?: LocationRef): void readonly agent: LocationCollection readonly command: LocationCollection readonly integration: LocationCollection diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index 32647df3ff..f9c44d7356 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -76,14 +76,9 @@ function makeRoutes( const serviceLayer = simulateEnabled() ? Layer.unwrap( Effect.gen(function* () { - const { simulationReplacements, startDriveServer } = yield* Effect.promise( - () => import("@opencode-ai/simulation/backend"), - ) - if (driveEnabled()) startDriveServer() - return AppNodeBuilder.build(applicationServices, [ - ...replacements, - ...(simulateEnabled() ? simulationReplacements : []), - ]) + const { simulationReplacements } = yield* Effect.promise(() => import("@opencode-ai/simulation/backend")) + const simulation = yield* simulationReplacements() + return AppNodeBuilder.build(applicationServices, [...replacements, ...simulation]) }), ) : AppNodeBuilder.build(applicationServices, replacements) @@ -116,9 +111,5 @@ function simulateEnabled() { return !!process.env.OPENCODE_SIMULATE } -function driveEnabled() { - return !!process.env.OPENCODE_DRIVE -} - export const webHandler = () => HttpRouter.toWebHandler(createRoutes().pipe(Layer.provide(HttpServer.layerServices))) diff --git a/packages/simulation/package.json b/packages/simulation/package.json index 624cf32ef7..1902a7a14a 100644 --- a/packages/simulation/package.json +++ b/packages/simulation/package.json @@ -17,7 +17,7 @@ "typecheck": "tsgo --noEmit" }, "dependencies": { - "@fontsource/adwaita-mono": "5.2.1", + "@fontsource/commit-mono": "5.2.5", "@napi-rs/canvas": "1.0.2", "@opencode-ai/core": "workspace:*", "@opencode-ai/llm": "workspace:*", diff --git a/packages/simulation/src/assets.d.ts b/packages/simulation/src/assets.d.ts new file mode 100644 index 0000000000..f80c06f5d1 --- /dev/null +++ b/packages/simulation/src/assets.d.ts @@ -0,0 +1,4 @@ +declare module "*.woff2" { + const path: string + export default path +} diff --git a/packages/simulation/src/backend/control.ts b/packages/simulation/src/backend/control.ts deleted file mode 100644 index eaacc162c1..0000000000 --- a/packages/simulation/src/backend/control.ts +++ /dev/null @@ -1,97 +0,0 @@ -import { Effect } from "effect" -import { SimulationProtocol } from "../protocol" -import { SimulationLLMExchange } from "./llm-exchange" - -/** - * Backend-hosted simulation control WebSocket. - * - * JSON-RPC 2.0 over a loopback WebSocket, mirroring the protocol of the TUI - * simulation server. Drivers connect directly (standalone topology; no - * frontend proxy) to answer LLM exchanges and inspect the simulated network. - * This is also the headless-simulation interface: it works with no TUI at - * all. - * - * Methods: - * - `llm.attach` -> subscribe; pending and future exchanges arrive - * as `llm.request` notifications - * - `llm.chunk` { id, items } append response items to an exchange - * - `llm.finish` { id, reason? } finish an exchange - * - `llm.disconnect` { id } abruptly terminate an exchange without a finish - * - `llm.pending` list open exchanges - */ - -type ControlSocket = Bun.ServerWebSocket<{ unsubscribe?: () => void }> - -function parseRequest(input: string | Buffer) { - return SimulationProtocol.Backend.decodeRequest(JSON.parse(typeof input === "string" ? input : input.toString())) -} - -async function handle(socket: ControlSocket, request: SimulationProtocol.Backend.Request): Promise { - switch (request.method) { - case "llm.attach": { - socket.data.unsubscribe?.() - socket.data.unsubscribe = SimulationLLMExchange.subscribe((exchange) => { - socket.send(JSON.stringify({ jsonrpc: "2.0", method: "llm.request", params: exchange })) - }) - return { attached: true } - } - case "llm.chunk": { - await Effect.runPromise( - SimulationLLMExchange.push( - request.params.id, - request.params.items.map((item) => ({ type: "item", item }) as const), - ), - ) - return { ok: true } - } - case "llm.finish": { - await Effect.runPromise( - SimulationLLMExchange.push(request.params.id, [{ type: "finish", reason: request.params.reason }]), - ) - return { ok: true } - } - case "llm.disconnect": { - await Effect.runPromise(SimulationLLMExchange.disconnect(request.params.id)) - return { ok: true } - } - case "llm.pending": - return { exchanges: SimulationLLMExchange.pending() } - } -} - -export function start(endpoint: string) { - const url = new URL(endpoint) - const server = Bun.serve<{ unsubscribe?: () => void }>({ - hostname: url.hostname, - port: Number(url.port), - fetch(request, server) { - if (server.upgrade(request, { data: {} })) return undefined - return new Response("opencode drive backend websocket", { status: 426 }) - }, - websocket: { - close(socket) { - socket.data.unsubscribe?.() - }, - async message(socket, message) { - let request: SimulationProtocol.Backend.Request | undefined - try { - request = parseRequest(message) - const result = await handle(socket, request) - const response = SimulationProtocol.JsonRpc.success(request.id, result) - if (response) socket.send(JSON.stringify(response)) - } catch (error) { - socket.send(JSON.stringify(SimulationProtocol.JsonRpc.failure(request?.id, error))) - } - }, - }, - }) - process.stderr.write(`opencode drive backend websocket: ${endpoint}\n`) - return { - url: endpoint, - stop: () => { - server.stop(true) - }, - } -} - -export * as SimulationControl from "./control" diff --git a/packages/simulation/src/backend/index.ts b/packages/simulation/src/backend/index.ts index 2b8bb876e6..56c2ed87c2 100644 --- a/packages/simulation/src/backend/index.ts +++ b/packages/simulation/src/backend/index.ts @@ -1,9 +1,11 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { httpClient } from "@opencode-ai/core/effect/app-node-platform" +import { Config, Effect, Layer } from "effect" +import { HttpClient } from "effect/unstable/http" import { DriveManifest } from "../manifest" -import { SimulationControl } from "./control" import { SimulationNetwork } from "./network" import { SimulationOpenAI } from "./openai" +import { SimulatedProvider } from "./simulated-provider" /** * Layer replacements applied when the server is built in simulation mode. @@ -17,17 +19,29 @@ import { SimulationOpenAI } from "./openai" * */ -SimulationNetwork.register(SimulationOpenAI.route) -// ModelsDev dies when its catalog fetch fails, so simulation answers it with -// an empty catalog; providers come from seeded config instead. -SimulationNetwork.register(SimulationNetwork.json("GET", "https://models.dev/api.json", {})) +export const simulationReplacements = Effect.fn("Simulation.replacements")(function* () { + // ModelsDev dies when its catalog fetch fails, so simulation answers it with + // an empty catalog; providers come from seeded config instead. + const models = SimulationNetwork.json("GET", "https://models.dev/api.json", {}) + const drive = yield* Config.string("OPENCODE_DRIVE").pipe(Config.withDefault(undefined)) + if (!drive) return [[httpClient, SimulationNetwork.layer([models])]] satisfies LayerNode.Replacements -export function startDriveServer() { - return SimulationControl.start(DriveManifest.resolve().endpoints.backend) -} - -export const simulationReplacements: LayerNode.Replacements = [ - [httpClient, SimulationNetwork.layer], -] + const manifest = yield* DriveManifest.resolve() + const networkLayer = Layer.effect( + HttpClient.HttpClient, + Effect.gen(function* () { + const provider = yield* SimulatedProvider.Service + const network = yield* SimulationNetwork.make([SimulationOpenAI.route(provider), models]) + return network.client + }), + ).pipe( + Layer.provide( + SimulatedProvider.layerDrive({ + endpoint: manifest.endpoints.backend, + }), + ), + ) + return [[httpClient, networkLayer]] satisfies LayerNode.Replacements +}) export * as Simulation from "./index" diff --git a/packages/simulation/src/backend/llm-exchange.ts b/packages/simulation/src/backend/llm-exchange.ts deleted file mode 100644 index 638f80c1d0..0000000000 --- a/packages/simulation/src/backend/llm-exchange.ts +++ /dev/null @@ -1,119 +0,0 @@ -import { Effect, Queue } from "effect" - -/** - * Pending driver-answered LLM exchanges. - * - * When the simulated network receives a provider request it opens an - * exchange: the parsed request body plus a queue of response chunks. The - * simulation control WebSocket notifies the external driver, and the driver - * pushes chunks back until it finishes the exchange. The driver is the - * model; nothing is scripted or enqueued server-side. - * - * Process-global by design (plain module state, like the network route - * table): the simulated network and the control server must observe the same - * exchanges regardless of which layer instance touched them. - */ - -/** One response item the driver sends back. Compiled to provider wire chunks by the endpoint. */ -export type Item = - | { readonly type: "textDelta"; readonly text: string } - | { readonly type: "reasoningDelta"; readonly text: string } - | { - readonly type: "toolCall" - readonly index: number - readonly id: string - readonly name: string - readonly input: unknown - } - | { readonly type: "raw"; readonly chunk: unknown } - -export type FinishReason = "stop" | "tool-calls" | "length" | "content-filter" - -export type Chunk = - | { readonly type: "item"; readonly item: Item } - | { readonly type: "finish"; readonly reason: FinishReason } - -export interface Exchange { - readonly id: string - readonly url: string - readonly body: unknown - readonly queue: Queue.Queue -} - -export interface OpenedExchange { - readonly id: string - readonly url: string - readonly body: unknown -} - -const state = { - counter: 0, - exchanges: new Map(), - listeners: new Set<(exchange: OpenedExchange) => void>(), -} - -export class ExchangeNotFoundError extends Error { - constructor(id: string) { - super(`Simulation LLM exchange not found or already finished: ${id}`) - } -} - -/** Opens an exchange and notifies listeners. Called by the simulated provider endpoint. */ -export const open = (input: { readonly url: string; readonly body: unknown }) => - Effect.gen(function* () { - const id = `ex_${++state.counter}` - const queue = yield* Queue.unbounded() - const exchange: Exchange = { id, url: input.url, body: input.body, queue } - state.exchanges.set(id, exchange) - for (const listener of state.listeners) listener({ id, url: input.url, body: input.body }) - return exchange - }) - -/** Closes an exchange without consuming remaining chunks (response interrupted or finished). */ -export const close = (id: string) => - Effect.suspend(() => { - const exchange = state.exchanges.get(id) - state.exchanges.delete(id) - if (!exchange) return Effect.void - return Queue.shutdown(exchange.queue).pipe(Effect.asVoid) - }) - -/** Appends response chunks to an open exchange. Driver-facing. */ -export const push = (id: string, chunks: readonly Chunk[]) => - Effect.gen(function* () { - const exchange = state.exchanges.get(id) - if (!exchange) return yield* Effect.fail(new ExchangeNotFoundError(id)) - yield* Queue.offerAll(exchange.queue, chunks) - }) - -/** Abruptly ends the provider body without a finish chunk or SSE sentinel. */ -export const disconnect = (id: string) => - Effect.gen(function* () { - const exchange = state.exchanges.get(id) - if (!exchange) return yield* Effect.fail(new ExchangeNotFoundError(id)) - yield* Queue.shutdown(exchange.queue) - }) - -/** - * Registers a listener for newly opened exchanges and immediately replays - * currently-pending ones, so a late-attaching driver observes requests that - * arrived before it connected. Returns an unsubscribe function. - */ -export function subscribe(listener: (exchange: OpenedExchange) => void) { - state.listeners.add(listener) - for (const exchange of pending()) listener(exchange) - return () => { - state.listeners.delete(listener) - } -} - -/** Snapshot of currently open exchanges, for control-surface inspection. */ -export function pending(): OpenedExchange[] { - return [...state.exchanges.values()].map((exchange) => ({ - id: exchange.id, - url: exchange.url, - body: exchange.body, - })) -} - -export * as SimulationLLMExchange from "./llm-exchange" diff --git a/packages/simulation/src/backend/network.ts b/packages/simulation/src/backend/network.ts index 7b002fcb4e..f91ec8246c 100644 --- a/packages/simulation/src/backend/network.ts +++ b/packages/simulation/src/backend/network.ts @@ -1,7 +1,8 @@ -import { Effect, Layer } from "effect" -import { HttpClient, HttpClientResponse } from "effect/unstable/http" +import { Clock, Effect, Layer, Ref } from "effect" +import { HttpClient, HttpClientResponse, type HttpMethod } from "effect/unstable/http" import { HttpClientError, TransportError } from "effect/unstable/http/HttpClientError" import type { HttpClientRequest } from "effect/unstable/http" +import { SimulationProtocol } from "../protocol" /** * Simulated network. @@ -12,8 +13,7 @@ import type { HttpClientRequest } from "effect/unstable/http" * silently reach the real network. The scripted LLM is one registered route, * not a separate mechanism. * - * The route table is process-global module state so the control surface and - * the client layer observe the same registrations. + * Each acquired run owns its routes and request log. */ export interface Route { @@ -21,33 +21,15 @@ export interface Route { readonly match: ( request: HttpClientRequest.HttpClientRequest, url: URL, - ) => Effect.Effect | undefined + ) => Effect.Effect | undefined } -interface LogEntry { - readonly time: number - readonly method: string - readonly url: string - readonly matched: boolean -} - -const state = { - routes: [] as Route[], - log: [] as LogEntry[], -} +export type LogEntry = SimulationProtocol.Backend.NetworkLogEntry const LOG_LIMIT = 1000 -export function register(route: Route) { - state.routes.push(route) - return () => { - const index = state.routes.indexOf(route) - if (index >= 0) state.routes.splice(index, 1) - } -} - /** Static JSON route: exact method + origin/path match answered with a fixed body. */ -export function json(method: string, url: string, body: unknown): Route { +export function json(method: HttpMethod.HttpMethod, url: string, body: unknown): Route { return { match: (request, requestUrl) => { if (request.method !== method) return undefined @@ -62,24 +44,29 @@ export function json(method: string, url: string, body: unknown): Route { } } -export function log(): readonly LogEntry[] { - return state.log +export interface Run { + readonly client: HttpClient.HttpClient + readonly log: () => Effect.Effect } -function record(entry: LogEntry) { - state.log.push(entry) - if (state.log.length > LOG_LIMIT) state.log.splice(0, state.log.length - LOG_LIMIT) -} - -export const layer = Layer.sync(HttpClient.HttpClient)(() => - HttpClient.make((request, url) => - Effect.suspend(() => { - const matched = state.routes - .map((route) => route.match(request, url)) - .find((response) => response !== undefined) - record({ time: Date.now(), method: request.method, url: url.toString(), matched: matched !== undefined }) - if (matched) return matched - return Effect.fail( +export const make = Effect.fn("SimulationNetwork.make")(function* (routes: readonly Route[] = []) { + const log = yield* Ref.make([]) + const client = HttpClient.make((request, url) => + Effect.gen(function* () { + let matched: Effect.Effect | undefined + for (const route of routes) { + matched = route.match(request, url) + if (matched) break + } + const entry = { + time: yield* Clock.currentTimeMillis, + method: request.method, + url: url.toString(), + matched: matched !== undefined, + } + yield* Ref.update(log, (entries) => [...entries, entry].slice(-LOG_LIMIT)) + if (matched) return yield* matched + return yield* Effect.fail( new HttpClientError({ reason: new TransportError({ request, @@ -88,7 +75,11 @@ export const layer = Layer.sync(HttpClient.HttpClient)(() => }), ) }), - ), -) + ) + return { client, log: () => Ref.get(log) } satisfies Run +}) + +export const layer = (routes: readonly Route[] = []) => + Layer.effect(HttpClient.HttpClient, make(routes).pipe(Effect.map((run) => run.client))) export * as SimulationNetwork from "./network" diff --git a/packages/simulation/src/backend/openai.ts b/packages/simulation/src/backend/openai.ts index fb58a44724..b1a524547c 100644 --- a/packages/simulation/src/backend/openai.ts +++ b/packages/simulation/src/backend/openai.ts @@ -1,14 +1,15 @@ import { Effect, Schema, Stream } from "effect" import { HttpClientResponse } from "effect/unstable/http" +import { HttpClientError, TransportError } from "effect/unstable/http/HttpClientError" import { OpenAIChatEvent, DEFAULT_BASE_URL, PATH } from "@opencode-ai/llm/protocols/openai-chat" -import { SimulationLLMExchange } from "./llm-exchange" import { SimulationNetwork } from "./network" +import { SimulatedProvider } from "./simulated-provider" /** * Driver-answered OpenAI endpoint for the simulated network. * * Claims `POST {DEFAULT_BASE_URL}{PATH}` (the real openai-chat route - * endpoint), opens an LLM exchange, and streams the driver's chunks back as + * endpoint), invokes the simulated provider, and streams the driver's events back as * an OpenAI Chat SSE response terminated by `[DONE]`. Everything downstream * of the response bytes is the real pipeline: SSE framing, the OpenAIChat * event schema, the protocol state machine, and Lifecycle grammar. @@ -17,11 +18,15 @@ import { SimulationNetwork } from "./network" const encodeChunk = Schema.encodeUnknownSync(OpenAIChatEvent) const encoder = new TextEncoder() +const decodeBody = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Json)) // The simulated model id is echoed back only in non-schema fields; the // protocol event schema ignores unknown fields, so id/object/model are // decorative wire realism. -function chunkOf(item: SimulationLLMExchange.Item): OpenAIChatEvent | unknown { +type ProviderItem = Exclude +type FinishReason = Extract["reason"] + +function chunkOf(item: ProviderItem): OpenAIChatEvent | unknown { if (item.type === "textDelta") return { choices: [{ delta: { content: item.text } }] } if (item.type === "reasoningDelta") return { choices: [{ delta: { reasoning_content: item.text } }] } if (item.type === "toolCall") @@ -43,7 +48,7 @@ function chunkOf(item: SimulationLLMExchange.Item): OpenAIChatEvent | unknown { return item.chunk } -const finishReasonWire: Record = { +const finishReasonWire: Record = { stop: "stop", "tool-calls": "tool_calls", length: "length", @@ -54,40 +59,49 @@ function frame(payload: unknown): Uint8Array { return encoder.encode(`data: ${JSON.stringify(payload)}\n\n`) } -function sseBody(exchange: SimulationLLMExchange.Exchange): Stream.Stream { - const chunks = Stream.fromQueue(exchange.queue).pipe( - Stream.takeUntil((chunk) => chunk.type === "finish"), - Stream.map((chunk) => { - if (chunk.type === "finish") - return frame(encodeChunk({ choices: [{ delta: {}, finish_reason: finishReasonWire[chunk.reason] }] })) - if (chunk.item.type === "raw") return frame(chunk.item.chunk) - return frame(encodeChunk(chunkOf(chunk.item))) +function sseBody( + events: Stream.Stream, +): Stream.Stream { + return events.pipe( + Stream.map((event) => { + if (event.type === "finish") + return frame(encodeChunk({ choices: [{ delta: {}, finish_reason: finishReasonWire[event.reason] }] })) + if (event.type === "raw") return frame(event.chunk) + return frame(encodeChunk(chunkOf(event))) }), - ) - return chunks.pipe( Stream.concat(Stream.make(encoder.encode("data: [DONE]\n\n"))), - // Close the exchange when the response body ends or is interrupted, so - // late driver pushes fail with ExchangeNotFoundError instead of leaking. - Stream.ensuring(SimulationLLMExchange.close(exchange.id)), ) } -export const route: SimulationNetwork.Route = { +export const route = (provider: SimulatedProvider.Interface): SimulationNetwork.Route => ({ match: (request, url) => { if (request.method !== "POST") return undefined if (url.origin + url.pathname !== DEFAULT_BASE_URL + PATH) return undefined return Effect.gen(function* () { - const body = request.body._tag === "Uint8Array" ? JSON.parse(new TextDecoder().decode(request.body.body)) : {} - const exchange = yield* SimulationLLMExchange.open({ url: url.toString(), body }) + const body = + request.body._tag === "Uint8Array" + ? yield* decodeBody(new TextDecoder().decode(request.body.body)).pipe( + Effect.mapError( + (cause) => + new HttpClientError({ + reason: new TransportError({ + request, + cause, + description: "Simulation received an invalid OpenAI request body", + }), + }), + ), + ) + : {} return HttpClientResponse.fromWeb( request, - new Response(Stream.toReadableStream(sseBody(exchange)), { + new Response(Stream.toReadableStream(sseBody(provider.stream({ url: url.toString(), body }))), { status: 200, headers: { "content-type": "text/event-stream" }, }), ) }) }, -} +}) export * as SimulationOpenAI from "./openai" diff --git a/packages/simulation/src/backend/simulated-provider.ts b/packages/simulation/src/backend/simulated-provider.ts new file mode 100644 index 0000000000..69f01184aa --- /dev/null +++ b/packages/simulation/src/backend/simulated-provider.ts @@ -0,0 +1,270 @@ +import { Cause, Context, Effect, Fiber, FiberSet, Layer, PubSub, Queue, Ref, Schema, Semaphore, Stream } from "effect" +import { SimulationControlServer } from "../control-server" +import { SimulationProtocol } from "../protocol" + +export interface ProviderRequest { + readonly url: string + readonly body: unknown +} + +export type ProviderResponseEvent = + | SimulationProtocol.Backend.Item + | { readonly type: "finish"; readonly reason: SimulationProtocol.Backend.FinishReason } + +export class ProviderDisconnectedError extends Schema.TaggedErrorClass()( + "SimulatedProvider.ProviderDisconnectedError", + { message: Schema.String }, +) {} + +export interface Interface { + readonly stream: (request: ProviderRequest) => Stream.Stream +} + +export class Service extends Context.Service()("@opencode/simulation/SimulatedProvider") {} + +interface ProviderInvocation extends ProviderRequest { + readonly id: string +} + +interface PendingInvocation extends ProviderInvocation { + readonly responses: Queue.Queue +} + +interface State { + readonly counter: number + readonly pending: ReadonlyMap +} + +interface Driver { + readonly requests: Stream.Stream + readonly push: ( + id: string, + items: readonly SimulationProtocol.Backend.Item[], + ) => Effect.Effect + readonly finish: ( + id: string, + reason: SimulationProtocol.Backend.FinishReason, + ) => Effect.Effect + readonly disconnect: (id: string) => Effect.Effect + readonly pending: () => Effect.Effect +} + +class InvocationNotFoundError extends Schema.TaggedErrorClass()( + "SimulatedProvider.InvocationNotFoundError", + { id: Schema.String, message: Schema.String }, +) {} + +class ControllerDisconnectedError extends Schema.TaggedErrorClass()( + "SimulatedProvider.ControllerDisconnectedError", + { message: Schema.String }, +) {} + +type ControlSocket = SimulationControlServer.Socket + +export const layerDrive = (options: { readonly endpoint: string }) => + Layer.effect( + Service, + Effect.gen(function* () { + const state = yield* Ref.make({ counter: 0, pending: new Map() }) + const opened = yield* PubSub.unbounded() + const lock = yield* Semaphore.make(1) + + const close = (invocation: PendingInvocation) => + Effect.gen(function* () { + yield* Queue.shutdown(invocation.responses) + yield* lock.withPermit( + Ref.update(state, (current) => + current.pending.get(invocation.id) === invocation ? remove(current, invocation.id) : current, + ), + ) + }) + + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + const current = yield* Ref.get(state) + yield* Effect.forEach(current.pending.values(), (invocation) => Queue.shutdown(invocation.responses), { + discard: true, + }) + yield* PubSub.shutdown(opened) + }), + ) + + const open = (request: ProviderRequest) => + lock.withPermit( + Effect.gen(function* () { + const current = yield* Ref.get(state) + const id = `inv_${current.counter + 1}` + const responses = yield* Queue.bounded(256) + const invocation: PendingInvocation = { id, ...request, responses } + yield* Ref.set(state, { + counter: current.counter + 1, + pending: new Map(current.pending).set(id, invocation), + }) + yield* PubSub.publish(opened, { id, ...request }) + return invocation + }), + ) + + const requireInvocation = (id: string) => + Effect.gen(function* () { + const current = yield* Ref.get(state) + const invocation = current.pending.get(id) + if (invocation) return invocation + return yield* Effect.fail( + new InvocationNotFoundError({ + id, + message: `Simulated provider invocation not found or already finished: ${id}`, + }), + ) + }) + + const remove = (current: State, id: string) => { + const pending = new Map(current.pending) + pending.delete(id) + return { ...current, pending } + } + + const driver: Driver = { + requests: Stream.unwrap( + lock.withPermit( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(opened) + const current = yield* Ref.get(state) + const pending = Array.from(current.pending.values(), ({ id, url, body }) => ({ id, url, body })) + return Stream.concat(Stream.fromIterable(pending), Stream.fromEffectRepeat(PubSub.take(subscription))) + }), + ), + ), + push: (id, items) => + Effect.gen(function* () { + const invocation = yield* lock.withPermit(requireInvocation(id)) + yield* Queue.offerAll(invocation.responses, items) + }), + finish: (id, reason) => + Effect.gen(function* () { + const invocation = yield* lock.withPermit( + Effect.gen(function* () { + const invocation = yield* requireInvocation(id) + const current = yield* Ref.get(state) + yield* Ref.set(state, remove(current, id)) + return invocation + }), + ) + yield* Queue.offer(invocation.responses, { type: "finish", reason }) + yield* Queue.end(invocation.responses) + }), + disconnect: (id) => + Effect.gen(function* () { + const invocation = yield* lock.withPermit( + Effect.gen(function* () { + const invocation = yield* requireInvocation(id) + const current = yield* Ref.get(state) + yield* Ref.set(state, remove(current, id)) + return invocation + }), + ) + yield* Queue.fail( + invocation.responses, + new ProviderDisconnectedError({ message: "Simulated model provider disconnected" }), + ) + }), + pending: () => + lock.withPermit( + Ref.get(state).pipe( + Effect.map((current) => Array.from(current.pending.values(), ({ id, url, body }) => ({ id, url, body }))), + ), + ), + } + + const fibers = yield* FiberSet.make() + const activeController = yield* Ref.make | undefined>(undefined) + const controllerLock = yield* Semaphore.make(1) + yield* SimulationControlServer.start({ + endpoint: options.endpoint, + label: "opencode drive backend websocket", + data: () => ({}), + decode: SimulationProtocol.Backend.decodeRequestEffect, + handle: (socket, request) => handle(driver, fibers, activeController, controllerLock, socket, request), + close: (socket) => releaseController(activeController, controllerLock, socket), + }) + yield* Effect.sync(() => process.stderr.write(`opencode drive backend websocket: ${options.endpoint}\n`)) + + return Service.of({ + stream: (request) => + Stream.unwrap( + Effect.acquireRelease(open(request), close).pipe( + Effect.map((invocation) => + Stream.fromQueue(invocation.responses).pipe(Stream.takeUntil((event) => event.type === "finish")), + ), + ), + ), + }) + }), + ) + +function handle( + driver: Driver, + fibers: FiberSet.FiberSet, + activeController: Ref.Ref | undefined>, + controllerLock: Semaphore.Semaphore, + socket: ControlSocket, + request: SimulationProtocol.Backend.Request, +) { + switch (request.method) { + case "llm.attach": + return controllerLock.withPermit( + Effect.gen(function* () { + if (socket.data.closed) + return yield* Effect.fail( + new ControllerDisconnectedError({ message: "Drive controller disconnected before attachment" }), + ) + const previous = yield* Ref.get(activeController) + if (previous) yield* Fiber.interrupt(previous) + const attachment = yield* FiberSet.run( + fibers, + driver.requests.pipe( + Stream.runForEach((invocation) => + Effect.sync(() => { + socket.send(JSON.stringify({ jsonrpc: "2.0", method: "llm.request", params: invocation })) + }), + ), + ), + ) + if (socket.data.closed) { + yield* Fiber.interrupt(attachment) + return yield* Effect.fail( + new ControllerDisconnectedError({ message: "Drive controller disconnected during attachment" }), + ) + } + socket.data.attachment = attachment + yield* Ref.set(activeController, attachment) + return { attached: true } + }), + ) + case "llm.chunk": + return driver.push(request.params.id, request.params.items).pipe(Effect.as({ ok: true })) + case "llm.finish": + return driver.finish(request.params.id, request.params.reason).pipe(Effect.as({ ok: true })) + case "llm.disconnect": + return driver.disconnect(request.params.id).pipe(Effect.as({ ok: true })) + case "llm.pending": + return driver.pending().pipe(Effect.map((invocations) => ({ invocations }))) + } +} + +function releaseController( + activeController: Ref.Ref | undefined>, + controllerLock: Semaphore.Semaphore, + socket: ControlSocket, +) { + return controllerLock.withPermit( + Effect.gen(function* () { + const attachment = socket.data.attachment + if (!attachment) return + yield* Fiber.interrupt(attachment) + yield* Ref.update(activeController, (active) => (active === attachment ? undefined : active)) + }), + ) +} + +export * as SimulatedProvider from "./simulated-provider" diff --git a/packages/simulation/src/control-server.ts b/packages/simulation/src/control-server.ts new file mode 100644 index 0000000000..fd4211669e --- /dev/null +++ b/packages/simulation/src/control-server.ts @@ -0,0 +1,91 @@ +import { Effect, Fiber, Queue, Stream } from "effect" +import { SimulationProtocol } from "./protocol" + +export interface Server { + readonly url: string +} + +interface Request { + readonly id?: string | number | null +} + +export interface SocketData { + readonly drive?: true + attachment?: Fiber.Fiber + closed?: true +} + +export type Socket = Bun.ServerWebSocket + +export function start(options: { + readonly endpoint: string + readonly label: string + readonly data: () => SocketData + readonly decode: (input: string) => Effect.Effect + readonly handle: (socket: Socket, request: RequestType) => Effect.Effect + readonly close?: (socket: Socket) => Effect.Effect +}) { + return Effect.gen(function* () { + const messages = yield* Queue.bounded<{ readonly socket: Socket; readonly input: string }>(256) + const closures = yield* Queue.unbounded() + yield* Stream.fromQueue(messages).pipe( + Stream.runForEach((message) => + options.decode(message.input).pipe( + Effect.flatMap((request) => + options.handle(message.socket, request).pipe( + Effect.matchEffect({ + onFailure: (error) => send(message.socket, SimulationProtocol.JsonRpc.failure(request.id, error)), + onSuccess: (result) => send(message.socket, SimulationProtocol.JsonRpc.success(request.id, result)), + }), + ), + ), + Effect.catch((error) => send(message.socket, SimulationProtocol.JsonRpc.failure(undefined, error))), + ), + ), + Effect.forkScoped, + ) + yield* Stream.fromQueue(closures).pipe( + Stream.runForEach((socket) => options.close?.(socket) ?? Effect.void), + Effect.forkScoped, + ) + const url = yield* Effect.try({ try: () => new URL(options.endpoint), catch: (cause) => cause }) + yield* Effect.acquireRelease( + Effect.sync(() => + Bun.serve({ + hostname: url.hostname, + port: Number(url.port), + fetch(request, server) { + if (server.upgrade(request, { data: options.data() })) return undefined + return new Response(options.label, { status: 426 }) + }, + websocket: { + close(socket) { + socket.data.closed = true + Queue.offerUnsafe(closures, socket) + }, + message(socket, message) { + const input = typeof message === "string" ? message : message.toString() + if (Queue.offerUnsafe(messages, { socket, input })) return + socket.send( + JSON.stringify( + SimulationProtocol.JsonRpc.failure(undefined, new Error("Simulation control queue is full")), + ), + ) + }, + }, + }), + ), + (server) => Effect.promise(() => server.stop(true)), + ) + return { url: options.endpoint } satisfies Server + }) +} + +function send(socket: Socket, response: SimulationProtocol.JsonRpc.Response | undefined) { + if (!response) return Effect.void + return Effect.sync(() => { + socket.send(JSON.stringify(response)) + }) +} + +export * as SimulationControlServer from "./control-server" diff --git a/packages/simulation/src/frontend/actions.ts b/packages/simulation/src/frontend/actions.ts index dbd54eb23b..19baf6a732 100644 --- a/packages/simulation/src/frontend/actions.ts +++ b/packages/simulation/src/frontend/actions.ts @@ -1,11 +1,10 @@ -import { mkdir } from "node:fs/promises" import { tmpdir } from "node:os" import { extname, join, resolve } from "node:path" import type { CliRenderer, Renderable } from "@opentui/core" import { createMockKeys, createMockMouse, type MockInput, type MockMouse } from "@opentui/core/testing" +import { Config, Effect, FileSystem } from "effect" import type { SimulationProtocol } from "../protocol" import { SimulationRenderer } from "./renderer" -import { SimulationPng } from "./png" export type Action = SimulationProtocol.Frontend.Action export type Element = SimulationProtocol.Frontend.Element @@ -72,10 +71,7 @@ export function createHarness(renderer: CliRenderer): Harness { // captureCharFrame follows the test renderer's output sink. Recording // redirects that sink to the timeline, so read the live render buffer // instead; it is also the source used by screenshots. - screen: () => - decoder.decode( - (Reflect.get(renderer, "currentRenderBuffer") as RenderBuffer).getRealCharBytes(), - ), + screen: () => decoder.decode((Reflect.get(renderer, "currentRenderBuffer") as RenderBuffer).getRealCharBytes()), } } @@ -114,31 +110,29 @@ export function matches(harness: Pick, text: string) { return harness.screen().includes(text) } -export async function screenshot(harness: Harness, name?: string) { - await harness.renderOnce() - const image = SimulationPng.screenshot(harness.renderer) +export const screenshot = Effect.fn("SimulationActions.screenshot")(function* (harness: Harness, name?: string) { const filename = name ?? `screenshot-${crypto.randomUUID()}` - if ( - !filename || - filename.includes("/") || - filename.includes("\\") || - extname(filename) - ) - throw new Error("screenshot name must not contain a path or extension") + if (!filename || filename.includes("/") || filename.includes("\\") || extname(filename)) + return yield* Effect.fail(new Error("screenshot name must not contain a path or extension")) + yield* Effect.tryPromise(() => harness.renderOnce()) + const { SimulationPng } = yield* Effect.promise(() => import("./png")) + const image = SimulationPng.screenshot(harness.renderer) const directory = resolve( - process.env.OPENCODE_DRIVE_MEDIA_DIR ?? - join(tmpdir(), "opencode-drive", "output"), + yield* Config.string("OPENCODE_DRIVE_MEDIA_DIR").pipe( + Config.withDefault(join(tmpdir(), "opencode-drive", "output")), + ), ) - await mkdir(directory, { recursive: true }) + const fs = yield* FileSystem.FileSystem + yield* fs.makeDirectory(directory, { recursive: true }) const path = join(directory, `${filename}.png`) - await Bun.write(path, image.data) + yield* fs.writeFile(path, image.data) return path -} +}) -export async function execute(harness: Harness, action: Action) { +export const execute = Effect.fn("SimulationActions.execute")(function* (harness: Harness, action: Action) { switch (action.type) { case "ui.type": - await harness.mockInput.typeText(action.text) + yield* Effect.tryPromise(() => harness.mockInput.typeText(action.text)) break case "ui.press": harness.mockInput.pressKey(action.key, action.modifiers) @@ -155,18 +149,23 @@ export async function execute(harness: Harness, action: Action) { ?.focus() break case "ui.click": - await harness.mockMouse.click(action.x, action.y) + yield* Effect.tryPromise(() => harness.mockMouse.click(action.x, action.y)) break case "ui.resize": - if (!Number.isSafeInteger(action.cols) || action.cols <= 0 || !Number.isSafeInteger(action.rows) || action.rows <= 0) { - throw new Error("resize cols and rows must be positive integers") + if ( + !Number.isSafeInteger(action.cols) || + action.cols <= 0 || + !Number.isSafeInteger(action.rows) || + action.rows <= 0 + ) { + return yield* Effect.fail(new Error("resize cols and rows must be positive integers")) } harness.resize(action.cols, action.rows) SimulationRenderer.recordResize(harness.renderer, action.cols, action.rows) break } - await harness.renderOnce() + yield* Effect.tryPromise(() => harness.renderOnce()) return state(harness) -} +}) export * as SimulationActions from "./actions" diff --git a/packages/simulation/src/frontend/png.ts b/packages/simulation/src/frontend/png.ts index 4f7b11c1a0..cecbe4cb77 100644 --- a/packages/simulation/src/frontend/png.ts +++ b/packages/simulation/src/frontend/png.ts @@ -1,22 +1,20 @@ -import { fileURLToPath } from "node:url" -import { GlobalFonts, createCanvas } from "@napi-rs/canvas" +/// +import { GlobalFonts, createCanvas, type SKRSContext2D } from "@napi-rs/canvas" import { TextAttributes, type CapturedFrame, type CliRenderer, type RGBA } from "@opentui/core" +import regularFont from "@fontsource/commit-mono/files/commit-mono-latin-400-normal.woff2" with { type: "file" } +import boldFont from "@fontsource/commit-mono/files/commit-mono-latin-700-normal.woff2" with { type: "file" } +import italicFont from "@fontsource/commit-mono/files/commit-mono-latin-400-italic.woff2" with { type: "file" } +import boldItalicFont from "@fontsource/commit-mono/files/commit-mono-latin-700-italic.woff2" with { type: "file" } const CellWidth = 10 const CellHeight = 20 const FontSize = 16 const FontFamily = "OpenCode Mono" -for (const file of [ - "adwaita-mono-latin-400-normal.woff2", - "adwaita-mono-latin-700-normal.woff2", - "adwaita-mono-latin-400-italic.woff2", - "adwaita-mono-latin-700-italic.woff2", -]) { - GlobalFonts.registerFromPath( - fileURLToPath(import.meta.resolve(`@fontsource/adwaita-mono/files/${file}`)), - FontFamily, - ) +for (const file of [regularFont, boldFont, italicFont, boldItalicFont]) { + const font = Buffer.from(await Bun.file(file).arrayBuffer()) + if (!GlobalFonts.register(font, FontFamily)) + throw new Error(`Failed to register screenshot font: ${file}`) } export function screenshot(renderer: CliRenderer) { @@ -54,13 +52,17 @@ export function screenshotFrame(frame: CapturedFrame) { } if (!hidden && char.codePointAt(0) !== 0x0a00) { context.fillStyle = color(foreground, attributes & TextAttributes.DIM ? 0.55 : 1) - context.font = `${attributes & TextAttributes.ITALIC ? "italic " : ""}${attributes & TextAttributes.BOLD ? "bold " : ""}${FontSize}px "${FontFamily}"` - context.fillText(char, column * CellWidth, row * CellHeight + 1) + const x = column * CellWidth + const y = row * CellHeight + if (!drawBlockElement(context, char, x, y, cells)) { + context.font = `${attributes & TextAttributes.ITALIC ? "italic " : ""}${attributes & TextAttributes.BOLD ? "bold " : ""}${FontSize}px "${FontFamily}"` + context.fillText(char, x, y + 1) + } if (attributes & TextAttributes.UNDERLINE) { - context.fillRect(column * CellWidth, row * CellHeight + 17, cells * CellWidth, 1) + context.fillRect(x, y + 17, cells * CellWidth, 1) } if (attributes & TextAttributes.STRIKETHROUGH) { - context.fillRect(column * CellWidth, row * CellHeight + 10, cells * CellWidth, 1) + context.fillRect(x, y + 10, cells * CellWidth, 1) } } column += cells @@ -83,6 +85,15 @@ export function screenshotFrame(frame: CapturedFrame) { } } +function drawBlockElement(context: SKRSContext2D, char: string, x: number, y: number, cells: number) { + const width = cells * CellWidth + if (char === "█") context.fillRect(x, y, width, CellHeight) + else if (char === "▀") context.fillRect(x, y, width, CellHeight / 2) + else if (char === "▄") context.fillRect(x, y + CellHeight / 2, width, CellHeight / 2) + else return false + return true +} + function color(value: RGBA, opacity = 1) { const [red, green, blue, alpha] = value.toInts() return `rgba(${red}, ${green}, ${blue}, ${(alpha / 255) * opacity})` diff --git a/packages/simulation/src/frontend/renderer.ts b/packages/simulation/src/frontend/renderer.ts index 8224a082c3..f10f0ec076 100644 --- a/packages/simulation/src/frontend/renderer.ts +++ b/packages/simulation/src/frontend/renderer.ts @@ -1,5 +1,6 @@ import type { CliRenderer, CliRendererConfig } from "@opentui/core" import { createTestRenderer, type TestRendererSetup } from "@opentui/core/testing" +import { Effect } from "effect" import { Timeline } from "../recording" const setups = new WeakMap() @@ -16,37 +17,47 @@ export interface Viewport { readonly rows: number } -export async function create(options: CliRendererConfig, path?: string, viewport?: Viewport): Promise { +export const create = Effect.fn("SimulationRenderer.create")(function* ( + options: CliRendererConfig, + path?: string, + viewport?: Viewport, +) { const cols = viewport?.cols ?? 100 const rows = viewport?.rows ?? 40 - if (!path) { - const setup = await createTestRenderer({ - ...options, - width: cols, - height: rows, - }) - setups.set(setup.renderer, setup) - return setup.renderer - } - const recording = await Timeline.create(path, cols, rows) - const setup = await createTestRenderer({ - ...options, - width: cols, - height: rows, - stdout: recording as unknown as NodeJS.WriteStream, - bufferedOutput: "stdout", - onDestroy: () => { - void recording.finish().catch((error) => process.stderr.write(`Failed to finish UI recording: ${error}\n`)) - options.onDestroy?.() - }, - }).catch(async (error) => { - await recording.finish().catch(() => undefined) - throw error - }) + const recording = path + ? yield* Effect.acquireRelease( + Effect.tryPromise(() => Timeline.create(path, cols, rows)), + (recording) => + Effect.tryPromise(() => recording.finish()).pipe( + Effect.catch((error) => + Effect.sync(() => process.stderr.write(`Failed to finish UI recording: ${error}\n`)), + ), + ), + ) + : undefined + const setup = yield* Effect.acquireRelease( + Effect.tryPromise(() => + createTestRenderer({ + ...options, + width: cols, + height: rows, + ...(recording + ? { + stdout: recording as unknown as NodeJS.WriteStream, + bufferedOutput: "stdout" as const, + } + : {}), + }), + ), + (setup) => + Effect.sync(() => { + if (!setup.renderer.isDestroyed) setup.renderer.destroy() + }), + ) setups.set(setup.renderer, setup) - recordings.set(setup.renderer, recording) + if (recording) recordings.set(setup.renderer, recording) return setup.renderer -} +}) export function recordResize(renderer: CliRenderer, cols: number, rows: number) { recordings.get(renderer)?.resize(cols, rows) @@ -58,8 +69,8 @@ export function setupFor(renderer: CliRenderer): TestRendererSetup | undefined { export function finish(renderer: CliRenderer) { const recording = recordings.get(renderer) - if (!recording) throw new Error("UI recording is not available") - return recording.finish() + if (!recording) return Effect.fail(new Error("UI recording is not available")) + return Effect.tryPromise(() => recording.finish()) } export * as SimulationRenderer from "./renderer" diff --git a/packages/simulation/src/frontend/server.ts b/packages/simulation/src/frontend/server.ts index ccb1bd626e..991234b5b3 100644 --- a/packages/simulation/src/frontend/server.ts +++ b/packages/simulation/src/frontend/server.ts @@ -1,31 +1,19 @@ +import { Effect } from "effect" +import { SimulationControlServer } from "../control-server" import { SimulationProtocol } from "../protocol" import { SimulationActions, type Harness } from "./actions" +import { SimulationRenderer } from "./renderer" -export interface Server { - readonly url: string - readonly stop: () => void -} - -function parseRequest(input: string | Buffer) { - return SimulationProtocol.Frontend.decodeRequest(JSON.parse(typeof input === "string" ? input : input.toString())) -} - -async function handle( - harness: Harness, - request: SimulationProtocol.Frontend.Request, - finishRecording?: () => Promise, -) { +function handle(harness: Harness, request: SimulationProtocol.Frontend.Request) { switch (request.method) { case "ui.screenshot": return SimulationActions.screenshot(harness, request.params?.name) - case "ui.state": { - return SimulationActions.state(harness) - } + case "ui.state": + return Effect.sync(() => SimulationActions.state(harness)) case "ui.matches": - return SimulationActions.matches(harness, request.params.text) + return Effect.sync(() => SimulationActions.matches(harness, request.params.text)) case "ui.recording.finish": - if (!finishRecording) throw new Error("UI recording is not available") - return finishRecording() + return SimulationRenderer.finish(harness.renderer) case "ui.type": return SimulationActions.execute(harness, { type: "ui.type", text: request.params.text }) case "ui.enter": @@ -48,39 +36,22 @@ async function handle( y: request.params.y, }) case "ui.resize": - return SimulationActions.execute(harness, { type: "ui.resize", cols: request.params.cols, rows: request.params.rows }) + return SimulationActions.execute(harness, { + type: "ui.resize", + cols: request.params.cols, + rows: request.params.rows, + }) } } -export function start(harness: Harness, endpoint: string, finishRecording?: () => Promise): Server { - const url = new URL(endpoint) - const server = Bun.serve<{ readonly drive: true }>({ - hostname: url.hostname, - port: Number(url.port), - fetch(request, server) { - if (server.upgrade(request, { data: { drive: true } })) return undefined - return new Response("opencode drive ui websocket", { status: 426 }) - }, - websocket: { - async message(socket, message) { - let request: SimulationProtocol.Frontend.Request | undefined - try { - request = parseRequest(message) - const result = await handle(harness, request, finishRecording) - const next = SimulationProtocol.JsonRpc.success(request.id, result) - if (next) socket.send(JSON.stringify(next)) - } catch (error) { - socket.send(JSON.stringify(SimulationProtocol.JsonRpc.failure(request?.id, error))) - } - }, - }, +export const start = Effect.fn("SimulationServer.start")(function* (harness: Harness, endpoint: string) { + return yield* SimulationControlServer.start({ + endpoint, + label: "opencode drive ui websocket", + data: () => ({ drive: true as const }), + decode: SimulationProtocol.Frontend.decodeRequestEffect, + handle: (_socket, request) => handle(harness, request), }) - return { - url: endpoint, - stop: () => { - server.stop(true) - }, - } -} +}) export * as SimulationServer from "./server" diff --git a/packages/simulation/src/frontend/simulation.ts b/packages/simulation/src/frontend/simulation.ts index 591ee4e0ba..0843754ff1 100644 --- a/packages/simulation/src/frontend/simulation.ts +++ b/packages/simulation/src/frontend/simulation.ts @@ -1,32 +1,27 @@ -import { createCliRenderer, type CliRenderer, type CliRendererConfig } from "@opentui/core" +import { createCliRenderer, type CliRendererConfig } from "@opentui/core" +import { Config, Effect } from "effect" import { DriveManifest } from "../manifest" import { SimulationActions } from "./actions" import { SimulationRenderer } from "./renderer" import { SimulationServer } from "./server" -/** - * Drive-mode renderer entry point. - * - * Creates the renderer (headless when OPENCODE_DRIVE_RENDERER=headless, the normal - * visible renderer otherwise) and starts the UI control - * server against it. The server stops when the renderer is destroyed, so the - * caller only manages the renderer lifecycle. - */ -export async function create(options: CliRendererConfig): Promise { - const headless = process.env.OPENCODE_DRIVE_RENDERER === "headless" - const manifest = DriveManifest.resolve() +/** Drive-mode renderer and control-server acquisition. */ +export const create = Effect.fn("Drive.create")(function* (options: CliRendererConfig) { + const headless = (yield* Config.string("OPENCODE_DRIVE_RENDERER").pipe(Config.withDefault("visible"))) === "headless" + const manifest = yield* DriveManifest.resolve() const renderer = headless - ? await SimulationRenderer.create(options, manifest.recording?.timeline, manifest.viewport) - : await createCliRenderer(options) + ? yield* SimulationRenderer.create(options, manifest.recording?.timeline, manifest.viewport) + : yield* Effect.acquireRelease( + Effect.tryPromise(() => createCliRenderer(options)), + (renderer) => + Effect.sync(() => { + if (!renderer.isDestroyed) renderer.destroy() + }), + ) if (!headless && manifest.viewport) renderer.resize(manifest.viewport.cols, manifest.viewport.rows) - const server = SimulationServer.start( - SimulationActions.createHarness(renderer), - manifest.endpoints.ui, - headless && manifest.recording ? () => SimulationRenderer.finish(renderer) : undefined, - ) - process.stderr.write(`opencode drive ui websocket: ${server.url}\n`) - renderer.once("destroy", () => server.stop()) + const server = yield* SimulationServer.start(SimulationActions.createHarness(renderer), manifest.endpoints.ui) + yield* Effect.sync(() => process.stderr.write(`opencode drive ui websocket: ${server.url}\n`)) return renderer -} +}) export * as Drive from "./simulation" diff --git a/packages/simulation/src/manifest.ts b/packages/simulation/src/manifest.ts index 4896c2d740..fe276ee688 100644 --- a/packages/simulation/src/manifest.ts +++ b/packages/simulation/src/manifest.ts @@ -1,20 +1,54 @@ -import { existsSync, readFileSync } from "node:fs" import { homedir } from "node:os" import { isAbsolute, join } from "node:path" +import { Config, Effect, FileSystem, Schema } from "effect" +import { PositiveInt } from "@opencode-ai/core/schema" -export interface Manifest { - readonly endpoints: { - readonly ui: string - readonly backend: string - } - readonly viewport?: { - readonly cols: number - readonly rows: number - } - readonly recording?: { - readonly timeline: string - } -} +const InstanceName = Schema.String.check( + Schema.makeFilter((value) => + /^[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}$/.test(value) ? undefined : "a valid Drive instance name", + ), +) + +const Endpoint = Schema.String.check( + Schema.makeFilter((value) => { + if (!URL.canParse(value)) return "a loopback WebSocket endpoint with an explicit port" + const endpoint = new URL(value) + const port = Number(endpoint.port) + return endpoint.protocol === "ws:" && endpoint.hostname === "127.0.0.1" && Number.isInteger(port) && port >= 1 + ? undefined + : "a loopback WebSocket endpoint with an explicit port" + }), +) + +const AbsolutePath = Schema.String.check( + Schema.makeFilter((value) => (isAbsolute(value) ? undefined : "an absolute path")), +) + +export const Manifest = Schema.Struct({ + endpoints: Schema.Struct({ + ui: Endpoint, + backend: Endpoint, + }), + viewport: Schema.optionalKey( + Schema.Struct({ + cols: PositiveInt, + rows: PositiveInt, + }), + ), + recording: Schema.optionalKey( + Schema.Struct({ + timeline: AbsolutePath, + }), + ), +}) +export interface Manifest extends Schema.Schema.Type {} + +export class ResolveError extends Schema.TaggedErrorClass()("DriveManifest.ResolveError", { + reason: Schema.Literals(["config", "not-found", "read", "decode"]), + path: Schema.optionalKey(Schema.String), + message: Schema.String, + cause: Schema.Defect(), +}) {} export const defaults: Manifest = { endpoints: { @@ -23,48 +57,54 @@ export const defaults: Manifest = { }, } -export function resolve() { - const name = process.env.OPENCODE_DRIVE - if (!name) throw new Error("OPENCODE_DRIVE must contain a drive instance name") +const decode = Schema.decodeUnknownEffect(Schema.fromJsonString(Manifest)) + +const configError = (cause: unknown) => + new ResolveError({ + reason: "config", + message: `Invalid Drive configuration: ${String(cause)}`, + cause, + }) + +export const resolve = Effect.fn("DriveManifest.resolve")(function* () { + const name = yield* Config.schema(InstanceName, "OPENCODE_DRIVE").pipe(Effect.mapError(configError)) if (name === "1") return defaults - if (!/^[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}$/.test(name)) throw new Error(`Invalid drive instance name: ${name}`) - const directory = - process.env.DRIVE_REGISTRY_DIR ?? - join(process.env.XDG_STATE_HOME ?? join(homedir(), ".local", "state"), "opencode-drive", "instances") + const state = yield* Config.string("XDG_STATE_HOME").pipe( + Config.withDefault(join(homedir(), ".local", "state")), + Effect.mapError(configError), + ) + const directory = yield* Config.string("DRIVE_REGISTRY_DIR").pipe( + Config.withDefault(join(state, "opencode-drive", "instances")), + Effect.mapError(configError), + ) const file = join(directory, `${name}.json`) - if (!existsSync(file)) throw new Error(`Drive manifest not found: ${file}`) - - const manifest: unknown = JSON.parse(readFileSync(file, "utf8")) - if (!isManifest(manifest)) throw new Error(`Invalid drive manifest: ${file}`) - validateEndpoint(manifest.endpoints.ui, "ui") - validateEndpoint(manifest.endpoints.backend, "backend") - if (manifest.viewport) validateViewport(manifest.viewport) - if (manifest.recording && !isAbsolute(manifest.recording.timeline)) { - throw new Error(`Invalid drive recording timeline path: ${manifest.recording.timeline}`) - } - return manifest -} - -function isManifest(value: unknown): value is Manifest { - if (typeof value !== "object" || value === null || !("endpoints" in value)) return false - if (typeof value.endpoints !== "object" || value.endpoints === null) return false - return "ui" in value.endpoints && "backend" in value.endpoints -} - -function validateEndpoint(value: string, name: string) { - const endpoint = new URL(value) - const port = Number(endpoint.port) - if (endpoint.protocol !== "ws:" || endpoint.hostname !== "127.0.0.1" || !Number.isInteger(port) || port < 1) { - throw new Error(`Invalid drive ${name} endpoint: ${value}`) - } -} - -function validateViewport(value: Manifest["viewport"]) { - if (!value) return - if (!Number.isSafeInteger(value.cols) || value.cols <= 0 || !Number.isSafeInteger(value.rows) || value.rows <= 0) { - throw new Error(`Invalid drive viewport: ${JSON.stringify(value)}`) - } -} + const fs = yield* FileSystem.FileSystem + const contents = yield* fs.readFileString(file).pipe( + Effect.mapError( + (cause) => + new ResolveError({ + reason: cause.reason._tag === "NotFound" ? "not-found" : "read", + path: file, + message: + cause.reason._tag === "NotFound" + ? `Drive manifest not found: ${file}` + : `Failed to read Drive manifest: ${file}: ${cause.message}`, + cause, + }), + ), + ) + return yield* decode(contents).pipe( + Effect.mapError( + (cause) => + new ResolveError({ + reason: "decode", + path: file, + message: `Invalid Drive manifest: ${file}: ${cause.message}`, + cause, + }), + ), + ) +}) export * as DriveManifest from "./manifest" diff --git a/packages/simulation/src/protocol/index.ts b/packages/simulation/src/protocol/index.ts index 556e644a02..7ea6263b08 100644 --- a/packages/simulation/src/protocol/index.ts +++ b/packages/simulation/src/protocol/index.ts @@ -145,6 +145,7 @@ export namespace Frontend { ]) export type Request = Schema.Schema.Type export const decodeRequest = Schema.decodeUnknownSync(Request) + export const decodeRequestEffect = Schema.decodeUnknownEffect(Schema.fromJsonString(Request)) } export namespace Backend { @@ -188,9 +189,10 @@ export namespace Backend { ]) export type Request = Schema.Schema.Type export const decodeRequest = Schema.decodeUnknownSync(Request) + export const decodeRequestEffect = Schema.decodeUnknownEffect(Schema.fromJsonString(Request)) - export const OpenedExchange = Schema.Struct({ id: Schema.String, url: Schema.String, body: Schema.Json }) - export interface OpenedExchange extends Schema.Schema.Type {} + export const ProviderInvocation = Schema.Struct({ id: Schema.String, url: Schema.String, body: Schema.Json }) + export interface ProviderInvocation extends Schema.Schema.Type {} export const NetworkLogEntry = Schema.Struct({ time: Schema.Number, diff --git a/packages/simulation/test/fixture/websocket.ts b/packages/simulation/test/fixture/websocket.ts new file mode 100644 index 0000000000..092b81b4fb --- /dev/null +++ b/packages/simulation/test/fixture/websocket.ts @@ -0,0 +1,26 @@ +import { Effect } from "effect" + +export function availableEndpoint() { + const server = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: () => new Response() }) + const endpoint = `ws://127.0.0.1:${server.port}` + server.stop(true) + return endpoint +} + +export function connect(endpoint: string) { + return Effect.acquireRelease( + Effect.callback((resume) => { + const socket = new WebSocket(endpoint) + const open = () => resume(Effect.succeed(socket)) + const error = () => resume(Effect.fail(new Error(`Failed to connect to ${endpoint}`))) + socket.addEventListener("open", open, { once: true }) + socket.addEventListener("error", error, { once: true }) + return Effect.sync(() => { + socket.removeEventListener("open", open) + socket.removeEventListener("error", error) + socket.close() + }) + }), + (socket) => Effect.sync(() => socket.close()), + ) +} diff --git a/packages/simulation/test/frontend-server.test.ts b/packages/simulation/test/frontend-server.test.ts new file mode 100644 index 0000000000..80cdde4962 --- /dev/null +++ b/packages/simulation/test/frontend-server.test.ts @@ -0,0 +1,40 @@ +import { expect, test } from "bun:test" +import { Effect, FileSystem, Queue } from "effect" +import { SimulationActions } from "../src/frontend/actions" +import { SimulationRenderer } from "../src/frontend/renderer" +import { SimulationServer } from "../src/frontend/server" +import { availableEndpoint, connect } from "./fixture/websocket" + +test("scopes the frontend control server and reports malformed JSON", async () => { + const endpoint = availableEndpoint() + + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const renderer = yield* SimulationRenderer.create({}) + yield* SimulationServer.start(SimulationActions.createHarness(renderer), endpoint) + const socket = yield* connect(endpoint) + const messages = yield* Queue.unbounded() + socket.addEventListener("message", (event) => { + Queue.offerUnsafe(messages, JSON.parse(String(event.data))) + }) + + socket.send(JSON.stringify({ jsonrpc: "2.0", id: 1, method: "ui.state" })) + expect(yield* Queue.take(messages)).toMatchObject({ + id: 1, + result: { focused: { editor: false }, elements: [] }, + }) + + socket.send("{") + expect(yield* Queue.take(messages)).toMatchObject({ + id: null, + error: { code: -32000 }, + }) + }), + ).pipe(Effect.provide(FileSystem.layerNoop({}))), + ) + + const url = new URL(endpoint) + const rebound = Bun.serve({ hostname: url.hostname, port: Number(url.port), fetch: () => new Response() }) + await rebound.stop(true) +}) diff --git a/packages/simulation/test/manifest.test.ts b/packages/simulation/test/manifest.test.ts new file mode 100644 index 0000000000..7755421d6f --- /dev/null +++ b/packages/simulation/test/manifest.test.ts @@ -0,0 +1,74 @@ +import { expect, test } from "bun:test" +import { ConfigProvider, Effect, FileSystem, Layer } from "effect" +import { DriveManifest } from "../src/manifest" + +test("loads and validates a Drive manifest through Effect services", async () => { + const manifest = await Effect.runPromise( + DriveManifest.resolve().pipe( + Effect.provide( + Layer.merge( + FileSystem.layerNoop({ + readFileString: () => + Effect.succeed( + JSON.stringify({ + endpoints: { + ui: "ws://127.0.0.1:41000", + backend: "ws://127.0.0.1:41050", + }, + viewport: { cols: 120, rows: 50 }, + recording: { timeline: "/tmp/drive/timeline.jsonl" }, + }), + ), + }), + ConfigProvider.layer( + ConfigProvider.fromUnknown({ + OPENCODE_DRIVE: "test-instance", + DRIVE_REGISTRY_DIR: "/tmp/drive", + }), + ), + ), + ), + ), + ) + + expect(manifest).toEqual({ + endpoints: { + ui: "ws://127.0.0.1:41000", + backend: "ws://127.0.0.1:41050", + }, + viewport: { cols: 120, rows: 50 }, + recording: { timeline: "/tmp/drive/timeline.jsonl" }, + }) +}) + +test("reports schema-invalid manifests as typed decode failures", async () => { + const error = await Effect.runPromise( + DriveManifest.resolve().pipe( + Effect.flip, + Effect.provide( + Layer.merge( + FileSystem.layerNoop({ + readFileString: () => + Effect.succeed( + JSON.stringify({ + endpoints: { + ui: "https://example.com", + backend: "ws://127.0.0.1:41050", + }, + }), + ), + }), + ConfigProvider.layer( + ConfigProvider.fromUnknown({ + OPENCODE_DRIVE: "test-instance", + DRIVE_REGISTRY_DIR: "/tmp/drive", + }), + ), + ), + ), + ), + ) + + expect(error).toBeInstanceOf(DriveManifest.ResolveError) + expect(error.reason).toBe("decode") +}) diff --git a/packages/simulation/test/network.test.ts b/packages/simulation/test/network.test.ts new file mode 100644 index 0000000000..9049b5700c --- /dev/null +++ b/packages/simulation/test/network.test.ts @@ -0,0 +1,31 @@ +import { expect, test } from "bun:test" +import { Effect, Exit } from "effect" +import { TestClock } from "effect/testing" +import { HttpClientRequest } from "effect/unstable/http" +import { SimulationNetwork } from "../src/backend/network" + +test("keeps routes and request logs local to each network", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(1_234) + const first = yield* SimulationNetwork.make([ + SimulationNetwork.json("GET", "https://example.test/value", { source: "first" }), + ]) + const second = yield* SimulationNetwork.make() + const request = HttpClientRequest.get("https://example.test/value") + + const response = yield* first.client.execute(request) + expect(yield* response.text).toBe('{"source":"first"}') + expect(Exit.isFailure(yield* second.client.execute(request).pipe(Effect.exit))).toBe(true) + + expect(yield* first.log()).toEqual([ + { time: 1_234, method: "GET", url: "https://example.test/value", matched: true }, + ]) + expect(yield* second.log()).toEqual([ + { time: 1_234, method: "GET", url: "https://example.test/value", matched: false }, + ]) + }).pipe(Effect.provide(TestClock.layer())), + ), + ) +}) diff --git a/packages/simulation/test/openai.test.ts b/packages/simulation/test/openai.test.ts new file mode 100644 index 0000000000..3632bdb70b --- /dev/null +++ b/packages/simulation/test/openai.test.ts @@ -0,0 +1,47 @@ +import { expect, test } from "bun:test" +import { DEFAULT_BASE_URL, PATH } from "@opencode-ai/llm/protocols/openai-chat" +import { Effect, Stream } from "effect" +import { HttpClientRequest } from "effect/unstable/http" +import { HttpClientError } from "effect/unstable/http/HttpClientError" +import { SimulationOpenAI } from "../src/backend/openai" +import { SimulatedProvider } from "../src/backend/simulated-provider" + +test("encodes every simulated provider event as OpenAI SSE", async () => { + const provider: SimulatedProvider.Interface = { + stream: () => + Stream.make( + { type: "textDelta", text: "Hello " }, + { type: "textDelta", text: "from Drive" }, + { type: "finish", reason: "stop" }, + ), + } + const url = new URL(DEFAULT_BASE_URL + PATH) + const request = HttpClientRequest.post(url).pipe(HttpClientRequest.bodyJsonUnsafe({ model: "gpt-5" })) + const matched = SimulationOpenAI.route(provider).match(request, url) + if (!matched) throw new Error("The simulated OpenAI route did not match") + + const body = await Effect.runPromise(matched.pipe(Effect.flatMap((response) => response.text))) + + expect(body).toBe( + [ + 'data: {"choices":[{"delta":{"content":"Hello "}}]}', + 'data: {"choices":[{"delta":{"content":"from Drive"}}]}', + 'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}', + "data: [DONE]", + "", + ].join("\n\n"), + ) +}) + +test("rejects malformed intercepted OpenAI JSON as an HTTP client error", async () => { + const provider: SimulatedProvider.Interface = { stream: () => Stream.empty } + const url = new URL(DEFAULT_BASE_URL + PATH) + const request = HttpClientRequest.post(url).pipe(HttpClientRequest.bodyText("{")) + const matched = SimulationOpenAI.route(provider).match(request, url) + if (!matched) throw new Error("The simulated OpenAI route did not match") + + const error = await Effect.runPromise(matched.pipe(Effect.flip)) + + expect(error).toBeInstanceOf(HttpClientError) + expect(error.reason._tag).toBe("TransportError") +}) diff --git a/packages/simulation/test/png.test.ts b/packages/simulation/test/png.test.ts new file mode 100644 index 0000000000..4e420ebf16 --- /dev/null +++ b/packages/simulation/test/png.test.ts @@ -0,0 +1,61 @@ +import { expect, test } from "bun:test" +import { createCanvas, loadImage } from "@napi-rs/canvas" +import { RGBA, TextAttributes, type CapturedFrame } from "@opentui/core" +import { SimulationPng } from "../src/frontend/png" + +test("renders captured frames with bundled fonts", () => { + const frame: CapturedFrame = { + cols: 4, + rows: 1, + cursor: [0, 0], + lines: [ + { + spans: [ + { + text: "Test", + width: 4, + fg: RGBA.fromInts(255, 255, 255), + bg: RGBA.fromInts(0, 0, 0), + attributes: TextAttributes.BOLD | TextAttributes.ITALIC, + }, + ], + }, + ], + } + + const image = SimulationPng.screenshotFrame(frame) + expect(image.width).toBe(40) + expect(image.height).toBe(20) + expect(image.data.subarray(1, 4).toString()).toBe("PNG") +}) + +test("fills adjacent block elements without glyph gaps", async () => { + const image = SimulationPng.screenshotFrame({ + cols: 2, + rows: 1, + cursor: [0, 0], + lines: [ + { + spans: [ + { + text: "▀▀", + width: 2, + fg: RGBA.fromInts(255, 255, 255), + bg: RGBA.fromInts(0, 0, 0), + attributes: 0, + }, + ], + }, + ], + }) + const canvas = createCanvas(image.width, image.height) + const context = canvas.getContext("2d") + context.drawImage(await loadImage(image.data), 0, 0) + + expect([...context.getImageData(0, 5, image.width, 1).data]).toEqual( + Array.from({ length: image.width }, () => [255, 255, 255, 255]).flat(), + ) + expect([...context.getImageData(0, 15, image.width, 1).data]).toEqual( + Array.from({ length: image.width }, () => [0, 0, 0, 255]).flat(), + ) +}) diff --git a/packages/simulation/test/recording.test.ts b/packages/simulation/test/recording.test.ts index cbff7ac74c..00ef08b69d 100644 --- a/packages/simulation/test/recording.test.ts +++ b/packages/simulation/test/recording.test.ts @@ -5,6 +5,7 @@ import { join } from "node:path" import { TextRenderable } from "@opentui/core" import { createHarness, matches } from "../src/frontend/actions" import { SimulationRenderer } from "../src/frontend/renderer" +import { Effect } from "effect" import { Timeline, type Event } from "../src/recording" test("streams ANSI chunks into a versioned JSONL timeline", async () => { @@ -39,21 +40,25 @@ test("streams ANSI chunks into a versioned JSONL timeline", async () => { test("captures native renderer output and finishes on destroy", async () => { const directory = await mkdtemp(join(tmpdir(), "simulation-renderer-recording-")) const path = join(directory, "timeline.jsonl") - const renderer = await SimulationRenderer.create({}, path) try { - await SimulationRenderer.setupFor(renderer)?.renderOnce() - renderer.destroy() - expect(await SimulationRenderer.finish(renderer)).toBe(path) + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const renderer = yield* SimulationRenderer.create({}, path) + yield* Effect.promise(() => SimulationRenderer.setupFor(renderer)?.renderOnce() ?? Promise.resolve()) + renderer.destroy() + expect(yield* SimulationRenderer.finish(renderer)).toBe(path) - const events = (await Bun.file(path).text()) - .trim() - .split("\n") - .map((line) => JSON.parse(line) as Event) - expect(events.some((event) => event.type === "output")).toBe(true) + const events = (yield* Effect.promise(() => Bun.file(path).text())) + .trim() + .split("\n") + .map((line) => JSON.parse(line) as Event) + expect(events.some((event) => event.type === "output")).toBe(true) + }), + ), + ) } finally { - if (!renderer.isDestroyed) renderer.destroy() - await SimulationRenderer.finish(renderer) await rm(directory, { recursive: true, force: true }) } }) @@ -61,16 +66,20 @@ test("captures native renderer output and finishes on destroy", async () => { test("matches live screen text while recording", async () => { const directory = await mkdtemp(join(tmpdir(), "simulation-recording-matches-")) const path = join(directory, "timeline.jsonl") - const renderer = await SimulationRenderer.create({}, path) try { - renderer.root.add(new TextRenderable(renderer, { content: "recorded screen text" })) - await SimulationRenderer.setupFor(renderer)?.renderOnce() + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const renderer = yield* SimulationRenderer.create({}, path) + renderer.root.add(new TextRenderable(renderer, { content: "recorded screen text" })) + yield* Effect.promise(() => SimulationRenderer.setupFor(renderer)?.renderOnce() ?? Promise.resolve()) - expect(matches(createHarness(renderer), "recorded screen text")).toBe(true) + expect(matches(createHarness(renderer), "recorded screen text")).toBe(true) + }), + ), + ) } finally { - renderer.destroy() - await SimulationRenderer.finish(renderer) await rm(directory, { recursive: true, force: true }) } }) diff --git a/packages/simulation/test/simulated-provider.test.ts b/packages/simulation/test/simulated-provider.test.ts new file mode 100644 index 0000000000..23da2d9d0d --- /dev/null +++ b/packages/simulation/test/simulated-provider.test.ts @@ -0,0 +1,230 @@ +import { expect, test } from "bun:test" +import { Deferred, Effect, Fiber, Queue, Stream } from "effect" +import type { Scope } from "effect/Scope" +import { SimulatedProvider } from "../src/backend/simulated-provider" +import { availableEndpoint, connect } from "./fixture/websocket" + +test("streams a Drive-controlled provider response and removes the finished invocation", async () => { + await runProvider((provider, socket, messages) => + Effect.gen(function* () { + socket.send("{") + expect(yield* Queue.take(messages)).toMatchObject({ id: null, error: { code: -32000 } }) + yield* attach(socket, messages) + + const response = yield* provider.stream(request).pipe(Stream.runCollect, Effect.forkScoped) + + const opened = yield* takeInvocation(messages) + expect(opened).toMatchObject({ + method: "llm.request", + params: { + url: "https://api.openai.com/v1/chat/completions", + body: { model: "gpt-5" }, + }, + }) + const params = requireRecord(opened.params) + if (typeof params.id !== "string") throw new Error("llm.request did not contain an invocation id") + expect(response.pollUnsafe()).toBeUndefined() + + socket.send( + JSON.stringify({ + jsonrpc: "2.0", + id: 2, + method: "llm.chunk", + params: { id: params.id, items: [{ type: "textDelta", text: "Hello from Drive" }] }, + }), + ) + expect(yield* Queue.take(messages)).toMatchObject({ id: 2, result: { ok: true } }) + + socket.send( + JSON.stringify({ + jsonrpc: "2.0", + id: 3, + method: "llm.finish", + params: { id: params.id, reason: "stop" }, + }), + ) + expect(yield* Queue.take(messages)).toMatchObject({ id: 3, result: { ok: true } }) + + expect(Array.from(yield* Fiber.join(response))).toEqual([ + { type: "textDelta", text: "Hello from Drive" }, + { type: "finish", reason: "stop" }, + ]) + + socket.send(JSON.stringify({ jsonrpc: "2.0", id: 4, method: "llm.pending" })) + expect(yield* Queue.take(messages)).toMatchObject({ id: 4, result: { invocations: [] } }) + }), + ) +}) + +test("replays an invocation to a controller that attaches after it opens", async () => { + await runProvider((provider, socket, messages) => + Effect.gen(function* () { + const response = yield* provider.stream(request).pipe(Stream.runCollect, Effect.forkScoped) + + socket.send(JSON.stringify({ jsonrpc: "2.0", id: 1, method: "llm.attach" })) + const received = [requireRecord(yield* Queue.take(messages)), requireRecord(yield* Queue.take(messages))] + expect(received).toContainEqual(expect.objectContaining({ id: 1, result: { attached: true } })) + const opened = received.find((message) => message.method === "llm.request") + if (!opened) throw new Error("The pending invocation was not replayed") + const params = requireRecord(opened.params) + if (typeof params.id !== "string") throw new Error("llm.request did not contain an invocation id") + + socket.send( + JSON.stringify({ jsonrpc: "2.0", id: 2, method: "llm.finish", params: { id: params.id, reason: "stop" } }), + ) + expect(yield* Queue.take(messages)).toMatchObject({ id: 2, result: { ok: true } }) + expect(Array.from(yield* Fiber.join(response))).toEqual([{ type: "finish", reason: "stop" }]) + }), + ) +}) + +test("replaces the previous attached controller", async () => { + const endpoint = availableEndpoint() + await Effect.runPromise( + Effect.gen(function* () { + const provider = yield* SimulatedProvider.Service + const first = yield* connect(endpoint) + const second = yield* connect(endpoint) + const firstMessages = yield* messagesFrom(first) + const secondMessages = yield* messagesFrom(second) + + yield* attach(first, firstMessages) + yield* attach(second, secondMessages) + const response = yield* provider.stream(request).pipe(Stream.runCollect, Effect.forkScoped) + const opened = yield* takeInvocation(secondMessages) + expect(yield* Queue.size(firstMessages)).toBe(0) + const params = requireRecord(opened.params) + if (typeof params.id !== "string") throw new Error("llm.request did not contain an invocation id") + + second.send( + JSON.stringify({ jsonrpc: "2.0", id: 2, method: "llm.finish", params: { id: params.id, reason: "stop" } }), + ) + expect(yield* Queue.take(secondMessages)).toMatchObject({ id: 2, result: { ok: true } }) + expect(Array.from(yield* Fiber.join(response))).toEqual([{ type: "finish", reason: "stop" }]) + }).pipe(Effect.provide(SimulatedProvider.layerDrive({ endpoint })), Effect.scoped), + ) +}) + +test("removes an invocation when its response stream is interrupted", async () => { + await runProvider((provider, socket, messages) => + Effect.gen(function* () { + yield* attach(socket, messages) + const response = yield* provider.stream(request).pipe(Stream.runDrain, Effect.forkScoped) + yield* takeInvocation(messages) + + yield* Fiber.interrupt(response) + + socket.send(JSON.stringify({ jsonrpc: "2.0", id: 2, method: "llm.pending" })) + expect(yield* Queue.take(messages)).toMatchObject({ id: 2, result: { invocations: [] } }) + }), + ) +}) + +test("releases a backpressured response when its consumer is interrupted", async () => { + await runProvider((provider, socket, messages) => + Effect.gen(function* () { + yield* attach(socket, messages) + const started = yield* Deferred.make() + const response = yield* provider.stream(request).pipe( + Stream.runForEach(() => Deferred.succeed(started, void 0).pipe(Effect.andThen(Effect.never))), + Effect.forkScoped, + ) + const opened = yield* takeInvocation(messages) + const params = requireRecord(opened.params) + if (typeof params.id !== "string") throw new Error("llm.request did not contain an invocation id") + + socket.send( + JSON.stringify({ + jsonrpc: "2.0", + id: 2, + method: "llm.chunk", + params: { + id: params.id, + items: Array.from({ length: 300 }, (_, index) => ({ type: "textDelta", text: String(index) })), + }, + }), + ) + const result = yield* Queue.take(messages).pipe(Effect.forkScoped) + yield* Deferred.await(started) + expect(result.pollUnsafe()).toBeUndefined() + + yield* Fiber.interrupt(response) + expect(yield* Fiber.join(result)).toMatchObject({ id: 2 }) + }), + ) +}) + +test("fails the provider stream when Drive disconnects the invocation", async () => { + await runProvider((provider, socket, messages) => + Effect.gen(function* () { + yield* attach(socket, messages) + const response = yield* provider.stream(request).pipe(Stream.runCollect, Effect.flip, Effect.forkScoped) + const opened = yield* takeInvocation(messages) + const params = requireRecord(opened.params) + if (typeof params.id !== "string") throw new Error("llm.request did not contain an invocation id") + + socket.send(JSON.stringify({ jsonrpc: "2.0", id: 2, method: "llm.disconnect", params: { id: params.id } })) + expect(yield* Queue.take(messages)).toMatchObject({ id: 2, result: { ok: true } }) + expect(yield* Fiber.join(response)).toBeInstanceOf(SimulatedProvider.ProviderDisconnectedError) + }), + ) +}) + +const request: SimulatedProvider.ProviderRequest = { + url: "https://api.openai.com/v1/chat/completions", + body: { model: "gpt-5", messages: [{ role: "user", content: "Hello" }] }, +} + +function runProvider( + body: ( + provider: SimulatedProvider.Interface, + socket: WebSocket, + messages: Queue.Queue, + ) => Effect.Effect, +) { + const endpoint = availableEndpoint() + return Effect.runPromise( + Effect.gen(function* () { + const provider = yield* SimulatedProvider.Service + const socket = yield* connect(endpoint) + const messages = yield* messagesFrom(socket) + yield* body(provider, socket, messages) + }).pipe(Effect.provide(SimulatedProvider.layerDrive({ endpoint })), Effect.scoped), + ) +} + +function messagesFrom(socket: WebSocket) { + return Effect.gen(function* () { + const messages = yield* Queue.unbounded() + socket.addEventListener("message", (event) => { + Queue.offerUnsafe(messages, JSON.parse(String(event.data))) + }) + return messages + }) +} + +function attach(socket: WebSocket, messages: Queue.Queue) { + return Effect.gen(function* () { + socket.send(JSON.stringify({ jsonrpc: "2.0", id: 1, method: "llm.attach" })) + expect(yield* Queue.take(messages)).toMatchObject({ id: 1, result: { attached: true } }) + }) +} + +function takeInvocation(messages: Queue.Queue) { + return Queue.take(messages).pipe( + Effect.map((message) => { + const opened = requireRecord(message) + if (opened.method !== "llm.request") throw new Error("Expected an llm.request notification") + return opened + }), + ) +} + +function requireRecord(value: unknown): Record { + if (!isRecord(value)) throw new Error("Expected an object") + return value +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null +} diff --git a/packages/tui/src/app.tsx b/packages/tui/src/app.tsx index ab32bace8f..4f1587e2c3 100644 --- a/packages/tui/src/app.tsx +++ b/packages/tui/src/app.tsx @@ -197,42 +197,38 @@ export const run = Effect.fn("Tui.run")(function* (input: TuiInput) { const exit = { epilogue: undefined as string | undefined, reason: undefined as unknown } const result = yield* Effect.scoped( Effect.gen(function* () { - const renderer = yield* Effect.acquireRelease( - Effect.tryPromise({ - try: async () => { - const options = { - externalOutputMode: "passthrough", - targetFps: 60, - gatherStats: false, - exitOnCtrlC: false, - useKittyKeyboard: {}, - autoFocus: false, - openConsoleOnError: false, - useMouse: !Flag.OPENCODE_DISABLE_MOUSE && config.mouse, - consoleOptions: { - keyBindings: [{ name: "y", ctrl: true, action: "copy-selection" }], - }, - } satisfies CliRendererConfig - - if (handoff) { - handoff.renderer.useMouse = options.useMouse - return handoff.renderer - } - - if (process.env.OPENCODE_DRIVE) { - const { Drive } = await import("@opencode-ai/simulation/frontend") - return Drive.create(options) - } - - return createCliRenderer(options) - }, - catch: (error) => (error instanceof Error ? error : new Error(String(error))), - }), - (renderer) => - Effect.sync(() => { - destroyRenderer(renderer) + const options = { + externalOutputMode: "passthrough", + targetFps: 60, + gatherStats: false, + exitOnCtrlC: false, + useKittyKeyboard: {}, + autoFocus: false, + openConsoleOnError: false, + useMouse: !Flag.OPENCODE_DISABLE_MOUSE && config.mouse, + consoleOptions: { + keyBindings: [{ name: "y", ctrl: true, action: "copy-selection" }], + }, + } satisfies CliRendererConfig + const renderer = yield* Effect.gen(function* () { + if (handoff) { + handoff.renderer.useMouse = options.useMouse + return yield* Effect.acquireRelease(Effect.succeed(handoff.renderer), (renderer) => + Effect.sync(() => destroyRenderer(renderer)), + ) + } + if (process.env.OPENCODE_DRIVE) { + const { Drive } = yield* Effect.promise(() => import("@opencode-ai/simulation/frontend")) + return yield* Drive.create(options) + } + return yield* Effect.acquireRelease( + Effect.tryPromise({ + try: () => createCliRenderer(options), + catch: (error) => (error instanceof Error ? error : new Error(String(error))), }), - ) + (renderer) => Effect.sync(() => destroyRenderer(renderer)), + ) + }) win32DisableProcessedInput() const finalizers = new Set<() => Promise>() yield* Effect.addFinalizer(() => diff --git a/packages/tui/src/component/dialog-integration.tsx b/packages/tui/src/component/dialog-integration.tsx index 58b18e16f4..218e2cd220 100644 --- a/packages/tui/src/component/dialog-integration.tsx +++ b/packages/tui/src/component/dialog-integration.tsx @@ -467,11 +467,10 @@ async function connected( toast: ReturnType, onConnected?: OnIntegrationConnected, ) { - await Promise.all([ - data.location.integration.refresh(), - data.location.model.refresh(), - data.location.provider.refresh(), - ]) + data.location.integration.invalidate() + data.location.model.invalidate() + data.location.provider.invalidate() + await Promise.all([data.location.integration.sync(), data.location.model.sync(), data.location.provider.sync()]) toast.show({ variant: "success", message: `Connected ${integration.name}` }) if (onConnected) { onConnected(providerID(data, integration.id)) @@ -498,11 +497,10 @@ async function disconnected( dialog: ReturnType, toast: ReturnType, ) { - await Promise.all([ - data.location.integration.refresh(), - data.location.model.refresh(), - data.location.provider.refresh(), - ]) + data.location.integration.invalidate() + data.location.model.invalidate() + data.location.provider.invalidate() + await Promise.all([data.location.integration.sync(), data.location.model.sync(), data.location.provider.sync()]) toast.show({ variant: "success", message: `Disconnected ${name}` }) dialog.clear() } diff --git a/packages/tui/src/component/dialog-skill.tsx b/packages/tui/src/component/dialog-skill.tsx index 3826a5a3c5..7387012dee 100644 --- a/packages/tui/src/component/dialog-skill.tsx +++ b/packages/tui/src/component/dialog-skill.tsx @@ -25,7 +25,7 @@ export function DialogSkill(props: DialogSkillProps) { .then(async () => { const current = data.location.skill.list(props.location) if (current) return current - await data.location.skill.refresh(props.location) + await data.location.skill.sync(props.location) return data.location.skill.list(props.location) ?? [] }) // Catch so the rejected resource never reaches the memo below: reading diff --git a/packages/tui/src/component/prompt/index.tsx b/packages/tui/src/component/prompt/index.tsx index 7d8dec1379..4a66a03ef1 100644 --- a/packages/tui/src/component/prompt/index.tsx +++ b/packages/tui/src/component/prompt/index.tsx @@ -1055,7 +1055,7 @@ export function Prompt(props: PromptProps) { } else { move.startSubmit() if (!session) { - await data.session.refresh(sessionID) + await data.session.sync(sessionID) session = data.session.get(sessionID) } if (session?.agent !== agent.id) { diff --git a/packages/tui/src/component/prompt/move.tsx b/packages/tui/src/component/prompt/move.tsx index cc63f5ed5f..a1c0471e94 100644 --- a/packages/tui/src/component/prompt/move.tsx +++ b/packages/tui/src/component/prompt/move.tsx @@ -42,7 +42,7 @@ export function usePromptMove(input: { projectID: () => string | undefined; sess const directory = result.directory if (!directory) throw new Error("No project copy directory returned") - // Call a location-based route to make sure it's bootstrapped before moving on. + // Call a location-based route to initialize it before moving on. await client.api.location.get({ location: { directory } }) setProgress("Creating session") @@ -139,7 +139,7 @@ export function usePromptMove(input: { projectID: () => string | undefined; sess async function resolveSession(sessionID: string) { const session = data.session.get(sessionID) if (session) return session - await data.session.refresh(sessionID).catch(() => undefined) + await data.session.sync(sessionID).catch(() => undefined) return data.session.get(sessionID) } diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 2506023fbe..04ec20f38c 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -1,7 +1,7 @@ // Client data layer: apply server events and cache API reads into a Solid store. // Prefer straightforward projection. Do not add generation counters, stale-response // merges, live/history overlays, or other race machinery here—last write wins. -// Reconnect may re-bootstrap; that is enough. UI and the server own ordering concerns. +// Reconnect invalidates cached reads; active UI owners decide what to sync again. import type { AgentInfo, @@ -31,7 +31,7 @@ import type { Plugin } from "@opencode-ai/plugin/v2/tui" import { createStore, produce, reconcile } from "solid-js/store" import { createSimpleContext } from "./helper" import { useClient } from "./client" -import { createSignal, onCleanup } from "solid-js" +import { createEffect, createSignal, onCleanup } from "solid-js" export type DataSessionStatus = "idle" | "running" @@ -66,11 +66,10 @@ type Store = { // true root is not yet loaded). The value is a flat deduplicated list of every // session ID in that family, including the key itself once its info arrives. family: Record - status: Record + active: Record message: Record pending: Record input: Record - compaction: Record permission: Record // Pending forms keyed by owner: a session ID or the temporary "global" elicitation sentinel. form: Record @@ -89,6 +88,37 @@ function locationQuery(ref?: LocationRef) { return ref ? { directory: ref.directory, workspace: ref.workspaceID } : undefined } +function createSync() { + const state = new Map>() + return { + run(key: string, load: () => Promise) { + const active = state.get(key) + if (active === true) return Promise.resolve() + if (active) return active + const pending = load() + .then(() => { + if (state.get(key) === pending) state.set(key, true) + }) + .finally(() => { + if (state.get(key) === pending) state.delete(key) + }) + state.set(key, pending) + return pending + }, + complete(key: string) { + if (state.has(key)) return + state.set(key, true) + }, + invalidate(key?: string) { + if (key) { + state.delete(key) + return + } + state.clear() + }, + } +} + export const { use: useData, provider: DataProvider } = createSimpleContext({ name: "Data", init: () => { @@ -96,11 +126,10 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ session: { info: {}, family: {}, - status: {}, + active: {}, message: {}, pending: {}, input: {}, - compaction: {}, permission: {}, form: {}, }, @@ -115,16 +144,10 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ directory: process.cwd(), }) const messageIndex = new Map>() - let bootstrapping: Promise | undefined - let connected = false + const sync = createSync() - function setSessionStatus(sessionID: string, status: DataSessionStatus) { - setStore("session", "status", sessionID, status) - } - - function addCompaction(sessionID: string, inputID: string) { - if (store.session.compaction[sessionID]?.includes(inputID)) return - setStore("session", "compaction", sessionID, [...(store.session.compaction[sessionID] ?? []), inputID]) + function setSessionActive(sessionID: string, status: DataSessionStatus) { + setStore("session", "active", sessionID, status) } function addPending(item: SessionPendingInfo) { @@ -142,16 +165,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ ) } - function removeCompaction(sessionID: string, inputID?: string) { - if (!inputID || !store.session.compaction[sessionID]?.includes(inputID)) return - setStore( - "session", - "compaction", - sessionID, - store.session.compaction[sessionID].filter((id) => id !== inputID), - ) - } - const message = { update(sessionID: string, fn: (messages: SessionMessageInfo[], index: Map) => void) { setStore( @@ -226,7 +239,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ return current } - // Register one session into the family index. Idempotent: refreshing an + // Register one session into the family index. Idempotent: syncing an // existing session never duplicates its ID. When a tentative family keyed by // sessionID exists (descendants arrived while sessionID's own info was // absent) but sessionID turns out to have a parent, fold the orphan subtree @@ -254,15 +267,19 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ function removeSession(sessionID: string) { messageIndex.delete(sessionID) + sync.invalidate(`session:${sessionID}`) + sync.invalidate(`session.pending:${sessionID}`) + sync.invalidate(`session.message:${sessionID}`) + sync.invalidate(`session.permission:${sessionID}`) + sync.invalidate(`session.form:${sessionID}:`) setStore( "session", produce((draft) => { delete draft.info[sessionID] - delete draft.status[sessionID] + delete draft.active[sessionID] delete draft.message[sessionID] delete draft.pending[sessionID] delete draft.input[sessionID] - delete draft.compaction[sessionID] delete draft.permission[sessionID] delete draft.form[sessionID] for (const [rootID, family] of Object.entries(draft.family)) { @@ -277,7 +294,8 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ function handleEvent(event: OpenCodeEvent) { switch (event.type) { case "session.created": - void result.session.refresh(event.data.sessionID) + result.session.invalidate(event.data.sessionID) + void result.session.sync(event.data.sessionID) break case "session.deleted": removeSession(event.data.sessionID) @@ -290,19 +308,21 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) break case "catalog.updated": - void Promise.all([ - result.location.model.refresh(event.location), - result.location.provider.refresh(event.location), - ]) + result.location.model.invalidate(event.location) + result.location.provider.invalidate(event.location) + void Promise.all([result.location.model.sync(event.location), result.location.provider.sync(event.location)]) break case "agent.updated": - void result.location.agent.refresh(event.location) + result.location.agent.invalidate(event.location) + void result.location.agent.sync(event.location) break case "command.updated": - void result.location.command.refresh(event.location) + result.location.command.invalidate(event.location) + void result.location.command.sync(event.location) break case "skill.updated": - void result.location.skill.refresh(event.location) + result.location.skill.invalidate(event.location) + void result.location.skill.sync(event.location) break case "session.agent.selected": if (store.session.info[event.data.sessionID]) @@ -666,7 +686,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }) break case "session.execution.started": - setSessionStatus(event.data.sessionID, "running") + setSessionActive(event.data.sessionID, "running") break case "session.compaction.admitted": addPending({ @@ -676,11 +696,9 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ timeCreated: event.created, type: "compaction", }) - addCompaction(event.data.sessionID, event.data.inputID) break case "session.compaction.started": removePending(event.data.sessionID, event.data.inputID) - removeCompaction(event.data.sessionID, event.data.inputID) message.update(event.data.sessionID, (draft, index) => { message.append(draft, index, { id: event.data.inputID ?? messageIDFromEvent(event.id), @@ -696,7 +714,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ case "session.execution.succeeded": case "session.execution.failed": case "session.execution.interrupted": - setSessionStatus(event.data.sessionID, "idle") + setSessionActive(event.data.sessionID, "idle") message.update(event.data.sessionID, (draft) => { const currentAssistant = message.activeAssistant(draft) if (currentAssistant) currentAssistant.retry = undefined @@ -758,7 +776,6 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ break case "session.compaction.failed": removePending(event.data.sessionID, event.data.inputID) - removeCompaction(event.data.sessionID, event.data.inputID) message.update(event.data.sessionID, (draft, index) => { const position = draft.findLastIndex((item) => item.type === "compaction" && item.status === "running") const current = draft[position] @@ -837,22 +854,28 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ ) break case "reference.updated": - void result.location.reference.refresh() + result.location.reference.invalidate() + void result.location.reference.sync() break case "integration.updated": + result.location.integration.invalidate(event.location) + result.location.model.invalidate(event.location) + result.location.provider.invalidate(event.location) void Promise.all([ - result.location.integration.refresh(event.location), - result.location.model.refresh(event.location), - result.location.provider.refresh(event.location), + result.location.integration.sync(event.location), + result.location.model.sync(event.location), + result.location.provider.sync(event.location), ]) break // Authenticating an MCP integration reconnects its server, which emits mcp.status.changed, - // so the mcp list refreshes here rather than off integration.updated. + // so the mcp list syncs here rather than off integration.updated. case "mcp.status.changed": - void result.location.mcp.server.refresh(event.location) + result.location.mcp.server.invalidate(event.location) + void result.location.mcp.server.sync(event.location) break case "mcp.resources.changed": - void result.location.mcp.resource.refresh(event.location) + result.location.mcp.resource.invalidate(event.location) + void result.location.mcp.resource.sync(event.location) break } } @@ -883,7 +906,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ ) }, status(sessionID: string) { - return store.session.status[sessionID] ?? "idle" + return store.session.active[sessionID] ?? "idle" }, input: { list(sessionID: string) { @@ -893,38 +916,34 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ return store.session.input[sessionID]?.includes(inputID) ?? false }, }, - compaction: { - list(sessionID: string) { - return store.session.compaction[sessionID] ?? [] - }, - async refresh(sessionID: string) { - await result.session.pending.refresh(sessionID) - }, - }, pending: { list(sessionID: string) { return store.session.pending[sessionID] ?? [] }, - async refresh(sessionID: string) { - const pending = await client.api.session.pending.list({ sessionID }) - setStore("session", "pending", sessionID, reconcile(pending)) - setStore( - "session", - "input", - sessionID, - reconcile(pending.filter((item) => item.type !== "compaction").map((item) => item.id)), - ) - setStore( - "session", - "compaction", - sessionID, - reconcile(pending.filter((item) => item.type === "compaction").map((item) => item.id)), - ) + sync(sessionID: string) { + return sync.run(`session.pending:${sessionID}`, async () => { + const pending = await client.api.session.pending.list({ sessionID }) + setStore("session", "pending", sessionID, reconcile(pending)) + setStore( + "session", + "input", + sessionID, + reconcile(pending.filter((item) => item.type !== "compaction").map((item) => item.id)), + ) + }) + }, + invalidate(sessionID: string) { + sync.invalidate(`session.pending:${sessionID}`) }, }, - async refresh(sessionID: string) { - setStore("session", "info", sessionID, await client.api.session.get({ sessionID })) - registerSession(sessionID) + sync(sessionID: string) { + return sync.run(`session:${sessionID}`, async () => { + setStore("session", "info", sessionID, await client.api.session.get({ sessionID })) + registerSession(sessionID) + }) + }, + invalidate(sessionID: string) { + sync.invalidate(`session:${sessionID}`) }, message: { list(sessionID: string) { @@ -935,18 +954,30 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ const position = messageIndex.get(sessionID)?.get(messageID) return position === undefined ? undefined : messages?.[position] }, - async refresh(sessionID: string) { - const messages = (await client.api.message.list({ sessionID, limit: 200, order: "desc" })).data.toReversed() - messageIndex.set(sessionID, new Map(messages.map((message, index) => [message.id, index]))) - setStore("session", "message", sessionID, reconcile(messages)) + sync(sessionID: string) { + return sync.run(`session.message:${sessionID}`, async () => { + const messages = ( + await client.api.message.list({ sessionID, limit: 200, order: "desc" }) + ).data.toReversed() + messageIndex.set(sessionID, new Map(messages.map((message, index) => [message.id, index]))) + setStore("session", "message", sessionID, reconcile(messages)) + }) + }, + invalidate(sessionID: string) { + sync.invalidate(`session.message:${sessionID}`) }, }, permission: { list(sessionID: string) { return store.session.permission[sessionID] }, - async refresh(sessionID: string) { - setStore("session", "permission", sessionID, await client.api.permission.list({ sessionID })) + sync(sessionID: string) { + return sync.run(`session.permission:${sessionID}`, async () => { + setStore("session", "permission", sessionID, await client.api.permission.list({ sessionID })) + }) + }, + invalidate(sessionID: string) { + sync.invalidate(`session.permission:${sessionID}`) }, }, form: { @@ -957,23 +988,33 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ const key = locationKey(ref) return forms?.filter((form) => form.location && locationKey(form.location) === key) }, - async refresh(sessionID: string, ref?: LocationRef) { - if (sessionID === "global") { - const response = await client.api.form.request.list({ location: locationQuery(ref ?? defaultLocation()) }) - const location = { - directory: response.location.directory, - workspaceID: response.location.workspaceID, + sync(sessionID: string, ref?: LocationRef) { + const key = `session.form:${sessionID}:${sessionID === "global" ? locationKey(ref ?? defaultLocation()) : ""}` + return sync.run(key, async () => { + if (sessionID === "global") { + const response = await client.api.form.request.list({ + location: locationQuery(ref ?? defaultLocation()), + }) + const location = { + directory: response.location.directory, + workspaceID: response.location.workspaceID, + } + const locationID = locationKey(location) + setStore("session", "form", sessionID, [ + ...(store.session.form[sessionID] ?? []).filter( + (form) => form.location && locationKey(form.location) !== locationID, + ), + ...response.data.filter((form) => form.sessionID === "global").map((form) => ({ ...form, location })), + ]) + return } - const key = locationKey(location) - setStore("session", "form", sessionID, [ - ...(store.session.form[sessionID] ?? []).filter( - (form) => form.location && locationKey(form.location) !== key, - ), - ...response.data.filter((form) => form.sessionID === "global").map((form) => ({ ...form, location })), - ]) - return - } - setStore("session", "form", sessionID, await client.api.form.list({ sessionID })) + setStore("session", "form", sessionID, await client.api.form.list({ sessionID })) + }) + }, + invalidate(sessionID: string, ref?: LocationRef) { + sync.invalidate( + `session.form:${sessionID}:${sessionID === "global" ? locationKey(ref ?? defaultLocation()) : ""}`, + ) }, }, }, @@ -982,8 +1023,13 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ list(projectID: string) { return store.project.permission[projectID] }, - async refresh(projectID: string) { - setStore("project", "permission", projectID, await client.api.permission.saved.list({ projectID })) + sync(projectID: string) { + return sync.run(`project.permission:${projectID}`, async () => { + setStore("project", "permission", projectID, await client.api.permission.saved.list({ projectID })) + }) + }, + invalidate(projectID: string) { + sync.invalidate(`project.permission:${projectID}`) }, }, }, @@ -996,53 +1042,109 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ .map((data) => data.shell?.[id]) .find((shell) => shell !== undefined) }, - async refresh(ref?: LocationRef) { - const result = await client.api.shell.list({ location: locationQuery(ref) }) - const key = locationKey(result.location) - setStore("location", key, { - ...store.location[key], - shell: Object.fromEntries(result.data.map((info) => [info.id, info])), + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.shell:${id}`, async () => { + const response = await client.api.shell.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { + ...store.location[key], + shell: Object.fromEntries(response.data.map((info) => [info.id, info])), + }) }) }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.shell:${locationKey(ref ?? defaultLocation())}`) + }, }, location: { default() { return defaultLocation() }, - async refresh(ref?: LocationRef) { - const location = await client.api.location.get({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(location) - if (!store.location[key]) setStore("location", key, {}) - if (!ref) setDefaultLocation({ directory: location.directory, workspaceID: location.workspaceID }) + async sync(ref?: LocationRef) { + const current = ref ?? defaultLocation() + await sync.run(`location:${locationKey(current)}`, async () => { + const location = await client.api.location.get({ location: locationQuery(current) }) + const key = locationKey(location) + if (!store.location[key]) setStore("location", key, {}) + if (!ref) setDefaultLocation({ directory: location.directory, workspaceID: location.workspaceID }) + }) + const location = ref ?? defaultLocation() + await Promise.all([ + result.location.agent.sync(location), + result.location.command.sync(location), + result.location.integration.sync(location), + result.location.mcp.server.sync(location), + result.location.mcp.resource.sync(location), + result.location.model.sync(location), + result.location.provider.sync(location), + result.location.reference.sync(location), + result.location.skill.sync(location), + result.shell.sync(location), + result.session.form.sync("global", location), + ]) + }, + invalidate(ref?: LocationRef) { + const location = ref ?? defaultLocation() + sync.invalidate(`location:${locationKey(location)}`) + result.location.agent.invalidate(location) + result.location.command.invalidate(location) + result.location.integration.invalidate(location) + result.location.mcp.server.invalidate(location) + result.location.mcp.resource.invalidate(location) + result.location.model.invalidate(location) + result.location.provider.invalidate(location) + result.location.reference.invalidate(location) + result.location.skill.invalidate(location) + result.shell.invalidate(location) + result.session.form.invalidate("global", location) }, agent: { list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.agent }, - async refresh(ref?: LocationRef) { - const result = await client.api.agent.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { ...store.location[key], agent: result.data }) + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.agent:${id}`, async () => { + const response = await client.api.agent.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], agent: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.agent:${locationKey(ref ?? defaultLocation())}`) }, }, command: { list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.command }, - async refresh(ref?: LocationRef) { - const result = await client.api.command.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { ...store.location[key], command: result.data }) + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.command:${id}`, async () => { + const response = await client.api.command.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], command: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.command:${locationKey(ref ?? defaultLocation())}`) }, }, integration: { list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.integration }, - async refresh(ref?: LocationRef) { - const result = await client.api.integration.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { ...store.location[key], integration: result.data }) + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.integration:${id}`, async () => { + const response = await client.api.integration.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], integration: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.integration:${locationKey(ref ?? defaultLocation())}`) }, }, mcp: { @@ -1050,180 +1152,150 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.mcp?.server }, - async refresh(ref?: LocationRef) { - const result = await client.api.mcp.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { - ...store.location[key], - mcp: { ...store.location[key]?.mcp, server: result.data }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.mcp.server:${id}`, async () => { + const response = await client.api.mcp.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { + ...store.location[key], + mcp: { ...store.location[key]?.mcp, server: response.data }, + }) }) }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.mcp.server:${locationKey(ref ?? defaultLocation())}`) + }, }, resource: { list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.mcp?.resource }, - async refresh(ref?: LocationRef) { - const result = await client.api.mcp.resource.catalog({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { - ...store.location[key], - mcp: { ...store.location[key]?.mcp, resource: result.data.resources }, + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.mcp.resource:${id}`, async () => { + const response = await client.api.mcp.resource.catalog({ + location: locationQuery(ref ?? defaultLocation()), + }) + const key = locationKey(response.location) + setStore("location", key, { + ...store.location[key], + mcp: { ...store.location[key]?.mcp, resource: response.data.resources }, + }) }) }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.mcp.resource:${locationKey(ref ?? defaultLocation())}`) + }, }, }, model: { list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.model }, - async refresh(ref?: LocationRef) { - const result = await client.api.model.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { ...store.location[key], model: result.data }) + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.model:${id}`, async () => { + const response = await client.api.model.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], model: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.model:${locationKey(ref ?? defaultLocation())}`) }, }, provider: { list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.provider }, - async refresh(ref?: LocationRef) { - const result = await client.api.provider.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { ...store.location[key], provider: result.data }) + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.provider:${id}`, async () => { + const response = await client.api.provider.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], provider: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.provider:${locationKey(ref ?? defaultLocation())}`) }, }, reference: { list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.reference }, - async refresh(ref?: LocationRef) { - const result = await client.api.reference.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { ...store.location[key], reference: result.data }) + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.reference:${id}`, async () => { + const response = await client.api.reference.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], reference: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.reference:${locationKey(ref ?? defaultLocation())}`) }, }, skill: { list(location?: LocationRef) { return store.location[locationKey(location ?? defaultLocation())]?.skill }, - async refresh(ref?: LocationRef) { - const result = await client.api.skill.list({ location: locationQuery(ref ?? defaultLocation()) }) - const key = locationKey(result.location) - setStore("location", key, { ...store.location[key], skill: result.data }) + sync(ref?: LocationRef) { + const id = locationKey(ref ?? defaultLocation()) + return sync.run(`location.skill:${id}`, async () => { + const response = await client.api.skill.list({ location: locationQuery(ref ?? defaultLocation()) }) + const key = locationKey(response.location) + setStore("location", key, { ...store.location[key], skill: response.data }) + }) + }, + invalidate(ref?: LocationRef) { + sync.invalidate(`location.skill:${locationKey(ref ?? defaultLocation())}`) }, }, }, } result satisfies Plugin.Context["data"] - async function bootstrap() { - if (bootstrapping) return bootstrapping - bootstrapping = Promise.allSettled([ - client.api.session - .list({ - limit: 50, - order: "desc", - directory: defaultLocation().directory, - workspace: defaultLocation().workspaceID, - }) - .then((response) => { - setStore( - "session", - "info", - produce((draft) => { - for (const session of response.data) draft[session.id] = session - }), - ) - for (const session of response.data) registerSession(session.id) - }), - client.api.permission.request.list({ location: locationQuery(defaultLocation()) }).then((response) => { - const permissions = response.data.reduce>( - (result, request) => ({ - ...result, - [request.sessionID]: [...(result[request.sessionID] ?? []), request], - }), - {}, - ) - setStore("session", "permission", reconcile(permissions)) - }), - client.api.form.request.list({ location: locationQuery(defaultLocation()) }).then((response) => { - const location = { - directory: response.location.directory, - workspaceID: response.location.workspaceID, - } - const forms = response.data.reduce>( - (result, form) => ({ - ...result, - [form.sessionID]: [ - ...(result[form.sessionID] ?? []), - form.sessionID === "global" ? { ...form, location } : form, - ], - }), - {}, - ) - setStore("session", "form", reconcile(forms)) - }), - result.location.refresh(), - result.location.agent.refresh(), - result.location.integration.refresh(), - result.location.mcp.server.refresh(), - result.location.mcp.resource.refresh(), - result.location.model.refresh(), - result.location.provider.refresh(), - result.location.reference.refresh(), - result.location.command.refresh(), - result.location.skill.refresh(), - result.shell.refresh(), - ]) - .then(async (settled) => { - for (const failure of settled.filter((item) => item.status === "rejected")) - console.error("Failed to refresh default location data", failure.reason) - const key = locationKey(defaultLocation()) - const locations = new Map( - Object.values(store.session.info).map( - (session) => [locationKey(session.location), session.location] as const, - ), - ) - const refreshed = await Promise.allSettled( - Array.from(locations) - .filter(([location]) => location !== key) - .map(([, location]) => result.session.form.refresh("global", location)), - ) - for (const failure of refreshed.filter((item) => item.status === "rejected")) - console.error("Failed to refresh global forms", failure.reason) - }) - .finally(() => { - bootstrapping = undefined - }) - return bootstrapping - } - - function refreshActive() { - void client.api.session - .active() - .then((active) => { - setStore( - "session", - "status", - reconcile(Object.fromEntries(Object.keys(active).map((sessionID) => [sessionID, "running" as const]))), - ) - }) - .catch(() => undefined) - } + createEffect(() => { + if (client.connection.status() === "connected") return + sync.invalidate() + }) onCleanup( client.event.listen(({ details }) => { if (details.type === "server.connected") { - const messages = connected ? Object.keys(store.session.message) : [] - const compactions = connected ? Object.keys(store.session.compaction) : [] - connected = true - refreshActive() - void Promise.allSettled([ - bootstrap(), - ...messages.map(result.session.message.refresh), - ...compactions.map(result.session.compaction.refresh), - ]) + void client.api.session + .active() + .then((active) => { + setStore( + "session", + "active", + reconcile(Object.fromEntries(Object.keys(active).map((sessionID) => [sessionID, "running" as const]))), + ) + }) + .catch(() => undefined) + void client.api.session + .list({ + limit: 50, + order: "desc", + directory: defaultLocation().directory, + workspace: defaultLocation().workspaceID, + }) + .then((response) => { + setStore( + "session", + "info", + produce((draft) => { + for (const session of response.data) draft[session.id] = session + }), + ) + for (const session of response.data) { + sync.complete(`session:${session.id}`) + registerSession(session.id) + } + }) + .catch((error) => console.error("Failed to preload sessions", error)) return } handleEvent(details) diff --git a/packages/tui/src/context/location.tsx b/packages/tui/src/context/location.tsx index 0714ffdb7e..0bde553381 100644 --- a/packages/tui/src/context/location.tsx +++ b/packages/tui/src/context/location.tsx @@ -1,13 +1,35 @@ import type { LocationRef } from "@opencode-ai/client" -import { createContext, createSignal, useContext, type Accessor, type ParentProps, type Setter } from "solid-js" +import { createContext, createSignal, onCleanup, useContext, type Accessor, type ParentProps } from "solid-js" +import { useClient } from "./client" +import { useData } from "./data" const context = createContext<{ current: Accessor - set: Setter + set: (location?: LocationRef) => void }>() export function LocationProvider(props: ParentProps) { - const [current, set] = createSignal() + const client = useClient() + const data = useData() + const [current, setCurrent] = createSignal() + + function sync(location?: LocationRef) { + if (!location) return + const defaultLocation = data.location.default() + const target = + location.directory === defaultLocation.directory && location.workspaceID === defaultLocation.workspaceID + ? undefined + : location + void data.location.sync(target).catch(() => undefined) + } + + function set(location?: LocationRef) { + setCurrent(location) + if (client.connection.status() === "connected") sync(location) + } + + onCleanup(client.event.on("server.connected", () => sync(current()))) + return {props.children} } diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index f8d3302461..6e865269ca 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/packages/tui/src/routes/session/index.tsx @@ -184,23 +184,22 @@ export function Session() { const rows = createSessionRows(() => route.sessionID) createEffect( - on(descendantSessionIDs, (sessionIDs) => { + on([descendantSessionIDs, () => client.connection.status()], ([sessionIDs, status]) => { + if (status !== "connected") return void Promise.all( - sessionIDs.flatMap((sessionID) => [ - data.session.permission.refresh(sessionID), - data.session.form.refresh(sessionID), - ]), + sessionIDs.flatMap((sessionID) => [data.session.permission.sync(sessionID), data.session.form.sync(sessionID)]), ) }), ) createEffect(() => { + if (client.connection.status() !== "connected") return const sessionID = route.sessionID void (async () => { await Promise.all([ - data.session.refresh(sessionID), - data.session.permission.refresh(sessionID), - data.session.form.refresh(sessionID), + data.session.sync(sessionID), + data.session.permission.sync(sessionID), + data.session.form.sync(sessionID), ]) const info = data.session.get(sessionID) if (!info) { @@ -212,13 +211,6 @@ export function Session() { navigate({ type: "home" }) return } - void data.session.form.refresh("global", info.location).catch((error) => - toast.show({ - message: `Failed to refresh global forms: ${errorMessage(error)}`, - variant: "error", - duration: 5000, - }), - ) project.workspace.set(info.location.workspaceID) editor.reconnect(info.location.directory) if (route.sessionID === sessionID && scroll) scroll.scrollBy(100_000) diff --git a/packages/tui/src/routes/session/rows.ts b/packages/tui/src/routes/session/rows.ts index 3fb9323715..471b282097 100644 --- a/packages/tui/src/routes/session/rows.ts +++ b/packages/tui/src/routes/session/rows.ts @@ -2,6 +2,7 @@ import type { SessionMessageAssistant, SessionMessageInfo } from "@opencode-ai/c import { createEffect, on, onCleanup, type Accessor } from "solid-js" import { createStore, produce, reconcile } from "solid-js/store" import { useData } from "../../context/data" +import { useClient } from "../../context/client" export type PartRef = { messageID: string @@ -29,6 +30,7 @@ export type SessionRow = export function createSessionRows(sessionID: Accessor) { const data = useData() + const client = useClient() const [rows, setRows] = createStore([]) const revertBoundary = () => data.session.get(sessionID())?.revert?.messageID @@ -42,9 +44,10 @@ export function createSessionRows(sessionID: Accessor) { rows.splice( position === -1 ? rows.length : position, 0, - ...data.session.compaction + ...data.session.pending .list(sessionID()) - .map((inputID): SessionRow => ({ type: "compaction-queued", inputID })), + .filter((item) => item.type === "compaction") + .map((item): SessionRow => ({ type: "compaction-queued", inputID: item.id })), ) return rows } @@ -67,10 +70,11 @@ export function createSessionRows(sessionID: Accessor) { }) createEffect( - on(sessionID, (id) => { + on([sessionID, () => client.connection.status()], ([id, status]) => { + if (status !== "connected") return setRows(reconcile(reduce())) - void data.session.compaction.refresh(id).catch(() => undefined) - void data.session.message.refresh(id).then( + void data.session.pending.sync(id).catch(() => undefined) + void data.session.message.sync(id).then( () => { if (sessionID() !== id) return setRows(reconcile(reduce())) @@ -89,7 +93,11 @@ export function createSessionRows(sessionID: Accessor) { createEffect( on( - () => data.session.compaction.list(sessionID()).map((inputID) => inputID), + () => + data.session.pending + .list(sessionID()) + .filter((item) => item.type === "compaction") + .map((item) => item.id), () => setRows(reconcile(reduce())), ), ) diff --git a/packages/tui/test/app-lifecycle.test.tsx b/packages/tui/test/app-lifecycle.test.tsx index bad8245fcd..a6cd421ea1 100644 --- a/packages/tui/test/app-lifecycle.test.tsx +++ b/packages/tui/test/app-lifecycle.test.tsx @@ -1,6 +1,6 @@ import { expect, mock, test } from "bun:test" import { createTestRenderer } from "@opentui/core/testing" -import { Effect } from "effect" +import { Effect, FileSystem } from "effect" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { Global } from "@opencode-ai/core/global" import { createEventStream, createFetch, directory, json } from "./fixture/tui-client" @@ -33,7 +33,7 @@ test("SIGHUP clears title and disposes scoped resources once", async () => { packages: { resolve: async () => undefined }, args: {}, log: () => {}, - }).pipe(Effect.provide(AppNodeBuilder.build(Global.node))), + }).pipe(Effect.provide(AppNodeBuilder.build(Global.node)), Effect.provide(FileSystem.layerNoop({}))), ) await ready process.emit("SIGHUP") @@ -105,7 +105,7 @@ test("session lifecycle updates the terminal title and prints the epilogue after packages: { resolve: async () => undefined }, args: { sessionID: "dummy" }, log: () => {}, - }).pipe(Effect.provide(AppNodeBuilder.build(Global.node))), + }).pipe(Effect.provide(AppNodeBuilder.build(Global.node)), Effect.provide(FileSystem.layerNoop({}))), ) await initialTitleSet diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index ae1eabf683..444d0bb2ff 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -4,10 +4,11 @@ import { testRender } from "@opentui/solid" import type { OpenCodeEvent } from "@opencode-ai/client" import { SessionMessage } from "@opencode-ai/core/session/message" import { EventV2 } from "@opencode-ai/core/event" -import { onMount } from "solid-js" +import { createEffect, onMount, type ParentProps } from "solid-js" import { ProjectProvider } from "../../../src/context/project" import { ClientProvider, useClient } from "../../../src/context/client" -import { DataProvider, useData } from "../../../src/context/data" +import { DataProvider as DataProviderBase, useData } from "../../../src/context/data" +import { LocationProvider, useSetLocation } from "../../../src/context/location" import { createSessionRows, type SessionRow } from "../../../src/routes/session/rows" import { createApi, createEventStream, createFetch, directory, json } from "../../fixture/tui-client" import { TestTuiContexts } from "../../fixture/tui-environment" @@ -32,6 +33,24 @@ function emitEvent(events: ReturnType, event: OpenCode events.emit({ ...event, location: { directory } }) } +function DataProvider(props: ParentProps) { + return ( + + + + {props.children} + + + ) +} + +function SyncLocation() { + const data = useData() + const setLocation = useSetLocation() + createEffect(() => setLocation(data.location.default())) + return null +} + function durable(sessionID: string, seq?: number): { aggregateID: string; seq: number; version: 1 } function durable( sessionID: string, @@ -64,16 +83,13 @@ test("bootstraps MCP data for the TUI location", async () => { try { await wait(() => requests.length === 2) - expect(requests.map((url) => url.searchParams.get("location[directory]"))).toEqual([ - process.cwd(), - process.cwd(), - ]) + expect(requests.map((url) => url.searchParams.get("location[directory]"))).toEqual([directory, directory]) } finally { app.renderer.destroy() } }) -test("refreshes MCP status when a connection settles during bootstrap", async () => { +test("syncs MCP status when a connection settles during bootstrap", async () => { const events = createEventStream() let mcpRequests = 0 let resolveModels!: (response: Response) => void @@ -190,9 +206,9 @@ test("refreshes resources into reactive getters", async () => { expect(data.session.get("ses_test")).toBeUndefined() expect(data.location.agent.list(location)).toBeUndefined() - await data.session.refresh("ses_test") - await data.session.message.refresh("ses_test") - await data.location.agent.refresh() + await data.session.sync("ses_test") + await data.session.message.sync("ses_test") + await data.location.agent.sync() expect(data.session.get("ses_test")?.title).toBe("Test session") expect(data.session.message.list("ses_test").map((message) => message.id)).toEqual(["msg_first", "msg_second"]) @@ -243,7 +259,7 @@ test("applies absolute usage events to session info", async () => { )) try { - await data.session.refresh(sessionID) + await data.session.sync(sessionID) emitEvent(events, { id: "evt_usage_2", created: 2, @@ -328,7 +344,7 @@ test("truncates committed revert messages without changing lifetime usage", asyn )) try { - await data.session.refresh(sessionID) + await data.session.sync(sessionID) emitEvent(events, { id: "evt_revert_boundary_started", created: 1, @@ -467,7 +483,7 @@ test("updates session location when moved", async () => { try { await mounted - await data.session.refresh("ses_test") + await data.session.sync("ses_test") emitEvent(events, { id: "evt_moved_1", created: 1, @@ -525,7 +541,7 @@ test("restores running manual compaction before applying live deltas", async () )) try { - await data.session.message.refresh("session-compaction") + await data.session.message.sync("session-compaction") expect(data.session.message.get("session-compaction", "message-compaction")).toMatchObject({ type: "compaction", status: "running", @@ -548,7 +564,7 @@ test("restores running manual compaction before applying live deltas", async () } }) -test("reconnects the event stream and bootstraps fresh data", async () => { +test("reconnects the event stream and resyncs active data", async () => { const events = createEventStream() const requests = { active: 0, event: 0, message: 0, model: 0 } let resolveActive!: (response: Response) => void @@ -621,7 +637,7 @@ test("reconnects the event stream and bootstraps fresh data", async () => { try { await wait(() => data.location.model.list()?.[0]?.id === "model-1") await wait(() => data.session.status("session-stale") === "running") - await data.session.message.refresh("session-stale") + await data.session.message.sync("session-stale") expect(data.session.message.get("session-stale", "message-stale")?.id).toBe("message-stale") expect(client.connection.status()).toBe("connected") expect(client.connection.attempt()).toBe(0) @@ -633,6 +649,7 @@ test("reconnects the event stream and bootstraps fresh data", async () => { await wait(() => requests.active === 2 && client.connection.status() === "connected", 4000) resolveActive(json({ data: { "session-new": { type: "running" } } })) + void data.session.message.sync("session-stale") await wait(() => data.location.model.list()?.[0]?.id === "model-2", 4000) await wait(() => data.session.status("session-stale") === "idle") @@ -664,8 +681,10 @@ test("completes exploration when a queued prompt is promoted", async () => { if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) }, events) let rows!: ReturnType + let client!: ReturnType function Probe() { + client = useClient() rows = createSessionRows(() => sessionID) return } @@ -683,6 +702,7 @@ test("completes exploration when a queued prompt is promoted", async () => { )) try { + await wait(() => client.connection.status() === "connected") emitEvent(events, { id: "evt_step_started", created: 1, @@ -954,7 +974,7 @@ test("tracks session status from active sessions and execution events", async () try { await wait(() => data.session.status("session-active") === "running") expect(data.session.status("session-idle")).toBe("idle") - await data.session.refresh("session-live") + await data.session.sync("session-live") settled = true emitEvent(events, { @@ -1021,7 +1041,7 @@ test("tracks session status from active sessions and execution events", async () }) await wait(() => data.session.status("session-live") === "idle") - await data.session.refresh("session-failed") + await data.session.sync("session-failed") emitEvent(events, { id: "evt_failed_execution_started", created: 0, @@ -1184,7 +1204,7 @@ test("tracks session status from active sessions and execution events", async () durable: durable("session-manual", 1), data: { sessionID: "session-manual", inputID: "message-compaction" }, }) - await wait(() => data.session.compaction.list("session-manual").includes("message-compaction")) + await wait(() => data.session.pending.list("session-manual").some((item) => item.id === "message-compaction")) emitEvent(events, { id: "evt_manual_compaction_started", created: 1, @@ -1202,10 +1222,8 @@ test("tracks session status from active sessions and execution events", async () const message = data.session.message.get("session-manual", "message-compaction") return message?.type === "compaction" && message.status === "running" && message.summary === "Streamed summary" }) - expect(data.session.compaction.list("session-manual")).toEqual([]) - const compactionRow = manualRows.find( - (row) => row.type === "message" && row.messageID === "message-compaction", - ) + expect(data.session.pending.list("session-manual")).toEqual([]) + const compactionRow = manualRows.find((row) => row.type === "message" && row.messageID === "message-compaction") emitEvent(events, { id: "evt_manual_compaction_ended", created: 3, @@ -1247,9 +1265,7 @@ test("tracks session status from active sessions and execution events", async () const message = data.session.message.get("session-live", "msg_compaction_started") return message?.type === "compaction" && message.status === "running" && message.summary === "Live summary" }) - const autoCompactionRow = rows.find( - (row) => row.type === "message" && row.messageID === "msg_compaction_started", - ) + const autoCompactionRow = rows.find((row) => row.type === "message" && row.messageID === "msg_compaction_started") emitEvent(events, { id: "evt_compaction_ended", @@ -1301,9 +1317,11 @@ test("restores queued compaction from durable pending input", async () => { }, events) let data!: ReturnType let rows!: ReturnType + let client!: ReturnType function Probe() { data = useData() + client = useClient() rows = createSessionRows(() => sessionID) return } @@ -1321,8 +1339,9 @@ test("restores queued compaction from durable pending input", async () => { )) try { - await wait(() => data.session.compaction.list(sessionID).length === 2) - expect(data.session.compaction.list(sessionID)).toEqual([ + await wait(() => client.connection.status() === "connected") + await wait(() => data.session.pending.list(sessionID).length === 2) + expect(data.session.pending.list(sessionID).map((item) => item.id)).toEqual([ "message-compaction-queued", "message-compaction-later", ]) @@ -1359,8 +1378,8 @@ test("restores queued compaction from durable pending input", async () => { inputID: "message-compaction-queued", }, }) - await wait(() => data.session.compaction.list(sessionID).length === 1) - expect(data.session.compaction.list(sessionID)).toEqual(["message-compaction-later"]) + await wait(() => data.session.pending.list(sessionID).length === 1) + expect(data.session.pending.list(sessionID).map((item) => item.id)).toEqual(["message-compaction-later"]) emitEvent(events, { id: "evt_compaction_ended", @@ -1369,15 +1388,12 @@ test("restores queued compaction from durable pending input", async () => { durable: durable(sessionID, 5), data: { sessionID, reason: "manual", text: "Summary", recent: "" }, }) - expect(data.session.compaction.list(sessionID)).toEqual(["message-compaction-later"]) + expect(data.session.pending.list(sessionID).map((item) => item.id)).toEqual(["message-compaction-later"]) pending = [] - emitEvent(events, { - id: "evt_reconnected", - type: "server.connected", - data: {}, - }) - await wait(() => data.session.compaction.list(sessionID).length === 0) + data.session.pending.invalidate(sessionID) + await data.session.pending.sync(sessionID) + await wait(() => data.session.pending.list(sessionID).length === 0) } finally { app.renderer.destroy() } @@ -1689,7 +1705,7 @@ test("keeps shell state scoped to location", async () => { try { await wait(() => data.shell.list().some((shell) => shell.id === "sh_default")) - await data.shell.refresh({ directory: other }) + await data.shell.sync({ directory: other }) expect(data.shell.list().map((shell) => shell.id)).toEqual(["sh_default"]) expect(data.shell.list({ directory: other }).map((shell) => shell.id)).toEqual(["sh_other"]) @@ -1790,22 +1806,27 @@ test("adds and dismisses permission requests from live events", async () => { } }) -test("reconciles all pending permission requests when the event stream reconnects", async () => { +test("reconciles active session permissions when the event stream reconnects", async () => { const events = createEventStream() let requests = [ - { id: "per_old", sessionID: "ses_old", action: "read", resources: ["old.txt"] }, - { id: "per_keep", sessionID: "ses_keep", action: "shell", resources: ["bun test"] }, + { id: "per_old", sessionID: "ses_active", action: "read", resources: ["old.txt"] }, + { id: "per_keep", sessionID: "ses_active", action: "shell", resources: ["bun test"] }, ] let calls = 0 const fetch = createFetch((url) => { - if (url.pathname !== "/api/permission/request") return + if (url.pathname !== "/api/session/ses_active/permission") return calls++ - return json({ location: { directory, project: { id: "proj_test", directory } }, data: requests }) + return json({ data: requests }) }, events) let data!: ReturnType function Probe() { data = useData() + const client = useClient() + createEffect(() => { + if (client.connection.status() !== "connected") return + void data.session.permission.sync("ses_active") + }) return } @@ -1822,15 +1843,12 @@ test("reconciles all pending permission requests when the event stream reconnect )) try { - await wait(() => data.session.permission.list("ses_old")?.[0]?.id === "per_old") - expect(data.session.permission.list("ses_keep")?.[0]?.id).toBe("per_keep") + await wait(() => data.session.permission.list("ses_active")?.length === 2) - requests = [{ id: "per_new", sessionID: "ses_new", action: "edit", resources: ["new.txt"] }] + requests = [{ id: "per_new", sessionID: "ses_active", action: "edit", resources: ["new.txt"] }] events.disconnect() - await wait(() => calls === 2 && data.session.permission.list("ses_new")?.[0]?.id === "per_new") - expect(data.session.permission.list("ses_old")).toBeUndefined() - expect(data.session.permission.list("ses_keep")).toBeUndefined() + await wait(() => calls === 2 && data.session.permission.list("ses_active")?.[0]?.id === "per_new") } finally { app.renderer.destroy() } @@ -1903,7 +1921,7 @@ test("adds, dismisses, and refreshes form requests", async () => { }) await wait(() => data.session.form.list("ses_1")?.length === 0) - await data.session.form.refresh("ses_1") + await data.session.form.sync("ses_1") expect(data.session.form.list("ses_1")?.map((form) => form.id)).toEqual(["frm_remote"]) } finally { app.renderer.destroy() @@ -1975,7 +1993,7 @@ test("tracks global forms by location", async () => { } }) -test("refreshes global forms for the requested location", async () => { +test("syncs global forms once for each requested location", async () => { const events = createEventStream() const requests: URL[] = [] const other = { directory: "/tmp/opencode-other", workspaceID: "wrk_other" } @@ -2025,20 +2043,24 @@ test("refreshes global forms for the requested location", async () => { await wait(() => client.connection.status() === "connected" && requests.length > 0) requests.length = 0 - await data.session.form.refresh("global", { directory }) - await data.session.form.refresh("global", other) + await data.session.form.sync("global", { directory }) + await data.session.form.sync("global", other) - expect(requests).toHaveLength(2) - expect(requests[1]?.searchParams.get("location[directory]")).toBe(other.directory) - expect(requests[1]?.searchParams.get("location[workspace]")).toBe(other.workspaceID) + expect(requests).toHaveLength(1) + expect(requests[0]?.searchParams.get("location[directory]")).toBe(other.directory) + expect(requests[0]?.searchParams.get("location[workspace]")).toBe(other.workspaceID) expect(data.session.form.list("global", other)?.map((form) => form.id)).toEqual(["frm_other"]) expect(data.session.form.list("global", { directory })?.map((form) => form.id)).toEqual(["frm_default"]) + + data.session.form.invalidate("global", other) + await data.session.form.sync("global", other) + expect(requests).toHaveLength(2) } finally { app.renderer.destroy() } }) -test("refreshes global forms once per loaded location after reconnect", async () => { +test("resyncs global forms only for the active location after reconnect", async () => { const events = createEventStream() const requests: URL[] = [] const counts = new Map() @@ -2098,58 +2120,54 @@ test("refreshes global forms once per loaded location after reconnect", async () )) try { - await wait( - () => - data.session.form.list("global", home)?.[0]?.id === "frm_default_1" && - data.session.form.list("global", other)?.[0]?.id === "frm_other_1", - ) + await wait(() => data.session.form.list("global", home)?.[0]?.id === "frm_default_1") + await data.session.form.sync("global", other) + expect(data.session.form.list("global", other)?.[0]?.id).toBe("frm_other_1") expect(requests).toHaveLength(2) requests.length = 0 events.disconnect() - await wait( - () => - data.session.form.list("global", home)?.[0]?.id === "frm_default_2" && - data.session.form.list("global", other)?.[0]?.id === "frm_other_2", - 4000, - ) - expect(requests).toHaveLength(2) + await wait(() => data.session.form.list("global", home)?.[0]?.id === "frm_default_2", 4000) + expect(data.session.form.list("global", other)?.[0]?.id).toBe("frm_other_1") + expect(requests).toHaveLength(1) expect( requests.map((url) => [ url.searchParams.get("location[directory]") ?? directory, url.searchParams.get("location[workspace]") ?? undefined, ]), - ).toEqual([ - [home.directory, undefined], - [other.directory, other.workspaceID], - ]) + ).toEqual([[home.directory, undefined]]) } finally { app.renderer.destroy() } }) -test("reconciles all pending form requests when the event stream reconnects", async () => { +test("reconciles active session forms when the event stream reconnects", async () => { const events = createEventStream() let requests = [ - { id: "frm_old", sessionID: "ses_old", title: "Input requested", fields: formFields }, + { id: "frm_old", sessionID: "ses_active", title: "Input requested", fields: formFields }, { id: "frm_keep", - sessionID: "ses_keep", + sessionID: "ses_active", title: "Input requested", fields: [{ key: "authorization", type: "external" as const, url: "https://example.com" }], }, ] let calls = 0 const fetch = createFetch((url) => { - if (url.pathname !== "/api/form/request") return + if (url.pathname !== "/api/session/ses_active/form") return calls++ - return json({ location: { directory, project: { id: "proj_test", directory } }, data: requests }) + return json({ data: requests }) }, events) let data!: ReturnType function Probe() { data = useData() + const client = useClient() + createEffect(() => { + if (client.connection.status() !== "connected") return + void data.session.form.sync("ses_active") + }) return } @@ -2166,15 +2184,12 @@ test("reconciles all pending form requests when the event stream reconnects", as )) try { - await wait(() => data.session.form.list("ses_old")?.[0]?.id === "frm_old") - expect(data.session.form.list("ses_keep")?.[0]?.id).toBe("frm_keep") + await wait(() => data.session.form.list("ses_active")?.length === 2) - requests = [{ id: "frm_new", sessionID: "ses_new", title: "Input requested", fields: formFields }] + requests = [{ id: "frm_new", sessionID: "ses_active", title: "Input requested", fields: formFields }] events.disconnect() - await wait(() => calls === 2 && data.session.form.list("ses_new")?.[0]?.id === "frm_new") - expect(data.session.form.list("ses_old")).toBeUndefined() - expect(data.session.form.list("ses_keep")).toBeUndefined() + await wait(() => calls === 2 && data.session.form.list("ses_active")?.[0]?.id === "frm_new") } finally { app.renderer.destroy() } @@ -2421,7 +2436,7 @@ test("renders admitted prompts immediately and tracks them until promoted", asyn ]) expect(sync.session.input.list(sessionID)).toEqual([messageID]) - await sync.session.message.refresh(sessionID) + await sync.session.message.sync(sessionID) expect(sync.session.message.list(sessionID)?.[0]?.metadata).toBeUndefined() emitEvent(events, { @@ -2538,8 +2553,7 @@ function sessionInfo(id: string, parentID: string | undefined, cost = 0) { async function mountData(parents: Record, costs: Record = {}) { const calls = createFetch((url) => { const match = url.pathname.match(/^\/api\/session\/([^/]+)$/) - if (match && match[1] !== "active") - return json({ data: sessionInfo(match[1], parents[match[1]], costs[match[1]]) }) + if (match && match[1] !== "active") return json({ data: sessionInfo(match[1], parents[match[1]], costs[match[1]]) }) }) let data!: ReturnType let ready!: () => void @@ -2569,13 +2583,13 @@ async function mountData(parents: Record, costs: Record { const { data, app } = await mountData({ child: "root" }) try { - await data.session.refresh("child") + await data.session.sync("child") // Parent info is absent, so the missing parent is the furthest-known ancestor. expect(data.session.root("child")).toBe("root") expect(data.session.family("child")).toEqual(["child"]) expect(data.session.family("root")).toEqual(["child"]) - await data.session.refresh("root") + await data.session.sync("root") expect(data.session.root("root")).toBe("root") // The tentative root entry folds into the now-known root's family. expect(data.session.family("child")).toEqual(["child", "root"]) @@ -2588,17 +2602,17 @@ test("groups an orphan child under its missing parent until the root arrives", a test("indexes arbitrarily deep nesting under a single root", async () => { const { data, app } = await mountData({ grandchild: "child", child: "root" }) try { - await data.session.refresh("grandchild") + await data.session.sync("grandchild") expect(data.session.root("grandchild")).toBe("child") expect(data.session.family("grandchild")).toEqual(["grandchild"]) - await data.session.refresh("child") + await data.session.sync("child") // grandchild's tentative family (keyed by the missing "child") merges up // toward the still-missing "root". expect(data.session.root("child")).toBe("root") expect(data.session.family("grandchild")).toEqual(["grandchild", "child"]) - await data.session.refresh("root") + await data.session.sync("root") expect(data.session.root("grandchild")).toBe("root") expect(data.session.root("child")).toBe("root") expect(data.session.family("root")).toEqual(["grandchild", "child", "root"]) @@ -2608,14 +2622,11 @@ test("indexes arbitrarily deep nesting under a single root", async () => { }) test("totals family cost for roots and keeps subagent cost scoped", async () => { - const { data, app } = await mountData( - { grandchild: "child", child: "root" }, - { root: 1, child: 2, grandchild: 3 }, - ) + const { data, app } = await mountData({ grandchild: "child", child: "root" }, { root: 1, child: 2, grandchild: 3 }) try { - await data.session.refresh("grandchild") - await data.session.refresh("child") - await data.session.refresh("root") + await data.session.sync("grandchild") + await data.session.sync("child") + await data.session.sync("root") expect(data.session.cost("root")).toBe(6) expect(data.session.cost("child")).toBe(2) @@ -2628,15 +2639,15 @@ test("totals family cost for roots and keeps subagent cost scoped", async () => test("re-registering an existing session is idempotent", async () => { const { data, app } = await mountData({ grandchild: "child", child: "root" }) try { - await data.session.refresh("grandchild") - await data.session.refresh("child") - await data.session.refresh("root") + await data.session.sync("grandchild") + await data.session.sync("child") + await data.session.sync("root") const before = data.session.family("root") expect(before).toEqual(["grandchild", "child", "root"]) - await data.session.refresh("child") - await data.session.refresh("root") - await data.session.refresh("grandchild") + await data.session.sync("child") + await data.session.sync("root") + await data.session.sync("grandchild") expect(data.session.family("root")).toEqual(before) expect(data.session.family("root")).toHaveLength(3) } finally { @@ -2647,8 +2658,8 @@ test("re-registering an existing session is idempotent", async () => { test("stops at the last non-repeating ancestor on a parent cycle", async () => { const { data, app } = await mountData({ x: "y", y: "x" }) try { - await data.session.refresh("x") - await data.session.refresh("y") + await data.session.sync("x") + await data.session.sync("y") // Does not hang; walking up from "y" stops before re-entering "x". expect(data.session.root("y")).toBe("x") expect(data.session.family("y")).toEqual(["x", "y"])