From 5a1e1878447d8d0e0bbb7565fdf0c543e72ae245 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 3 Oct 2026 22:58:34 -0700 Subject: [PATCH] chore(app): clear lint warnings in GUI files touched by the Effect upgrade --- packages/app/e2e/utils/mock-api.ts | 20 +- packages/app/e2e/utils/mock-server.ts | 573 ++++++++++++------ .../app/src/providers/models/selection.tsx | 84 ++- .../src/runtime/persistence/schema.test.ts | 12 +- .../app/src/runtime/persistence/schema.ts | 21 +- .../app/src/runtime/server/persistence.ts | 15 +- .../src/main/files/attachment-picker.ts | 9 + .../desktop/src/main/ipc-transport.test.ts | 80 ++- packages/desktop/src/main/ipc-transport.ts | 9 + packages/desktop/src/main/ipc.ts | 20 +- packages/desktop/src/renderer/ipc-client.ts | 59 +- packages/desktop/src/shared/ipc-rpc.ts | 7 + packages/desktop/src/shared/ipc-rpc/app.ts | 15 + packages/desktop/src/shared/ipc-rpc/files.ts | 16 + packages/desktop/src/shared/ipc-rpc/menu.ts | 1 + .../desktop/src/shared/ipc-rpc/storage.ts | 7 + packages/desktop/src/shared/ipc-rpc/window.ts | 10 + .../gui-extensions/src/browser/network.ts | 5 + .../gui-extensions/src/ssh/askpass.test.ts | 5 + packages/gui-extensions/src/ssh/askpass.ts | 12 +- packages/gui-extensions/src/ssh/bootstrap.ts | 25 +- packages/gui-extensions/src/ssh/command.ts | 51 +- packages/gui-extensions/src/ssh/controller.ts | 85 ++- packages/gui-extensions/src/ssh/remote-cli.ts | 10 + 24 files changed, 887 insertions(+), 264 deletions(-) diff --git a/packages/app/e2e/utils/mock-api.ts b/packages/app/e2e/utils/mock-api.ts index 665c3009c016..c5da7c4c839c 100644 --- a/packages/app/e2e/utils/mock-api.ts +++ b/packages/app/e2e/utils/mock-api.ts @@ -1,4 +1,4 @@ -import { Schema, SchemaGetter } from "effect" +import { Predicate, Schema, SchemaGetter } from "effect" import { HttpApi, HttpApiEndpoint, HttpApiGroup, HttpApiSchema } from "effect/unstable/httpapi" import { Pty } from "@opencode/schema/pty" import { Worktree } from "@opencode/schema/worktree" @@ -10,7 +10,9 @@ const Json = Schema.Json.pipe( }), HttpApiSchema.asJson(), ) + const JsonPayload = Schema.Unknown.pipe(HttpApiSchema.asJson()) + const Query = Schema.Struct({ directory: Schema.optional(Schema.String), parentID: Schema.optional(Schema.String), @@ -23,8 +25,11 @@ const Query = Schema.Struct({ type: Schema.optional(Schema.String), mode: Schema.optional(Schema.String), }) + const SessionParams = { sessionID: Schema.String } + const PtyParams = { ptyID: Pty.ID } + const NoContent = HttpApiSchema.NoContent export class MockNotFound extends Schema.TaggedError()("MockNotFound", { @@ -49,6 +54,7 @@ export class MockShellNotFound extends Schema.TaggedError()(" export class MockUnsupported extends Schema.TaggedError()("MockUnsupported", { message: Schema.String, }) {} + const Unsupported = MockUnsupported.pipe(HttpApiSchema.status(501)) const Group = HttpApiGroup.make("mock") @@ -410,11 +416,17 @@ const Group = HttpApiGroup.make("mock") export const MockApi = HttpApi.make("mock").add(Group) +// SAFETY: handlers answer fixture data of any kind; this encoder is the boundary that turns it into JSON. +// oxlint-disable-next-line anti-slop/no-unknown-parameters -- see SAFETY above function jsonValue(value: unknown): Schema.Json { - if (value === null || typeof value === "string" || typeof value === "boolean") return value - if (typeof value === "number") return Number.isFinite(value) ? value : null + if (value === null || Predicate.isString(value) || Predicate.isBoolean(value)) return value + + if (Predicate.isNumber(value)) return Number.isFinite(value) ? value : null + if (Array.isArray(value)) return value.map(jsonValue) - if (!value || typeof value !== "object") return null + + if (!Predicate.isObject(value)) return null + return Object.fromEntries( Object.entries(value).flatMap(([key, item]) => (item === undefined ? [] : [[key, jsonValue(item)]])), ) diff --git a/packages/app/e2e/utils/mock-server.ts b/packages/app/e2e/utils/mock-server.ts index 4777c9d12f0e..58cfbd136e4a 100644 --- a/packages/app/e2e/utils/mock-server.ts +++ b/packages/app/e2e/utils/mock-server.ts @@ -1,20 +1,48 @@ import type { Page } from "@playwright/test" import type { JsonValue, OpenCodeEvent, SessionMessageInfo } from "@opencode/client/promise" -import { Duration, Effect, Layer } from "effect" +import { Duration, Effect, Layer, Option, Predicate, Schema } from "effect" import { HttpRouter, HttpServer, HttpServerResponse } from "effect/unstable/http" import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi" +import type { Worktree } from "@opencode/schema/worktree" import { SERVER } from "./app" import { MockApi, MockBadRequest, MockInternal, MockNotFound, MockShellNotFound, MockUnsupported } from "./mock-api" import { installSseTransport } from "./sse-transport" type Resolvable = T | (() => T) +// Data a test hands the mock, which answers it as JSON; undefined fields are left out. +export type MockFixture = + | string + | number + | boolean + | null + | undefined + | readonly MockFixture[] + | { readonly [key: string]: MockFixture } + +// A fixture or request object, such as a session, a permission request, or a request body. +export type MockFields = { [key: string]: MockFixture } + +export type MockSession = { id: string } & MockFields + +type MockSessionTime = { created: number; updated: number; idle?: number; viewed?: number; archived?: MockFixture } + +// The `project` fixture: the fields the mock reads, echoes, or updates, and whatever else the test adds. +export type MockProject = { + id?: string + worktree?: string + canonical?: string + sandboxes?: string[] + vcs?: string + [key: string]: MockFixture +} + // A hook's replacement response. export type MockAnswer = { status: number; body: unknown } export interface MockServerConfig { server?: string - provider: unknown | (() => unknown) + provider: Resolvable integrations?: unknown[] onConnectKey?: (input: { integrationID: string; body: unknown }) => void // Terminal shells the settings offer (`/api/config/shell`). @@ -30,10 +58,10 @@ export interface MockServerConfig { onFileWrite?: (input: { path: string; directory: string; body: string }) => void configEntries?: unknown[] directory: string - project: unknown + project: MockProject // Replaces the `/api/project` inventory, which defaults to `[project]`. projects?: Resolvable - sessions: ({ id: string } & Record)[] + sessions: MockSession[] pageMessages: ( sessionId: string, limit: number, @@ -58,23 +86,19 @@ export interface MockServerConfig { eventRetry?: number // Idle event streams send a comment every 15 s like the real server. Set false only to test the client's stall watchdog. keepalive?: boolean - permissions?: unknown[] | (() => unknown[]) + permissions?: Resolvable // Requests only listed by `/api/session/:id/permission`, keyed by session ID. - sessionPermissions?: Record + sessionPermissions?: Record // Returning true fails the next `/api/permission/request` listing with a 500. permissionListFailures?: () => boolean // Without it, permission replies answer 501 MockUnsupported. onPermissionReply?: (input: { sessionID: string; permissionID: string; body: unknown }) => void - forms?: unknown[] | (() => unknown[]) + forms?: Resolvable // MCP servers. A list serves every workspace; a function receives the requested directory. mcp?: unknown[] | ((directory: string) => unknown[]) // Connect/disconnect record the server's new status in that workspace (`connected`/`disabled`); the hook may return // another status (for example `{ status: "failed", error }`). Unknown servers answer 404. - onMcpAction?: (input: { - server: string - action: "connect" | "disconnect" - directory: string - }) => void | Record + onMcpAction?: (input: { server: string; action: "connect" | "disconnect"; directory: string }) => void | MockFields // Starts an OAuth attempt (POST .../connect/oauth) and returns its authorization URL; the attempt then stays pending. // Without it, OAuth connects answer 501 MockUnsupported. onIntegrationOAuth?: (input: { integrationID: string; directory: string; body: unknown }) => { url: string } @@ -84,22 +108,22 @@ export interface MockServerConfig { worktrees?: Resolvable // Without them, creating or removing a worktree answers 501 MockUnsupported. `onWorktreeCreate` may hold the request // and return the answer; by default it creates `/`. A created directory joins the project's sandboxes. - onWorktreeCreate?: (input: unknown) => void | MockAnswer | Promise - onWorktreeRemove?: (input: unknown) => void | Promise + onWorktreeCreate?: (input: Schema.Json) => void | MockAnswer | Promise + onWorktreeRemove?: (input: Worktree.RemoveInput) => void | Promise // POST /api/session keeps the client-reserved `id` and `location`. Return an answer to fail the attempt (1-based). - onSessionCreate?: (body: Record, attempt: number) => void | MockAnswer + onSessionCreate?: (body: MockFields, attempt: number) => void | MockAnswer // Title of created sessions (default: the request's title, else "New session"). createdSessionTitle?: string // Slash commands served by `/api/command`. commands?: Resolvable // Records POST /api/session/:id/command (204). Without it, commands answer 501 MockUnsupported. onCommand?: (input: { sessionID: string; body: unknown }) => void - fileList?: (path: string) => unknown | Promise - fileContent?: (path: string) => unknown | Promise - findFiles?: (input: { query: string; dirs?: string; limit?: number }) => unknown - sessionStatus?: Record | (() => Record) + fileList?: (path: string) => MockFixture | Promise + fileContent?: (path: string) => MockFixture | Promise + findFiles?: (input: { query: string; dirs?: string; limit?: number }) => MockFixture + sessionStatus?: Resolvable> inbox?: unknown[] | (() => unknown[]) - onPrompt?: (input: { sessionID: string; body: Record }) => void + onPrompt?: (input: { sessionID: string; body: MockFields }) => void generate?: (input: { sessionID: string; prompt: string }) => { text: string } | Promise<{ text: string }> onInboxChange?: (input: { sessionID: string; inboxID: string; action: "cancel" | "steer" | "queue" }) => void // Serves `/api/pty*` and mock PTY WebSockets. Created IDs are the first unused `${prefix}` (prefix must start with "pty"). @@ -141,8 +165,11 @@ type MockStream = { push: (payloads: unknown[]) => void } type MockStreamWindow = Window & { // Set to any value by benchmarks that bring their own event stream. __testSseTransport?: unknown - // `installSseTransport` registrations; `command` takes its browser command shape. - __testSseTransports?: Record unknown }> + // `installSseTransport` registrations; the mock only sends them events. + __testSseTransports?: Record< + string, + { command: (input: { type: "send"; deliveries: { payload: unknown }[]; burst: boolean }) => void } + > // Per-origin mock event streams; in-page benchmark probes push through them directly. __mockServerStreams?: Record } @@ -154,46 +181,61 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { await page.addInitScript( ({ server, retry, keepalive: idle }) => { + // SAFETY: every field `MockStreamWindow` adds to a window is optional. const host = window as MockStreamWindow + if (host.__testSseTransport || host.__testSseTransports?.[server] || host.__mockServerStreams?.[server]) return const originalFetch = window.fetch.bind(window) const encoder = new TextEncoder() - const state: { - controller?: ReadableStreamDefaultController - buffer: string[] - connections: number - } = { buffer: [], connections: 0 } - const frame = (payload: unknown) => `data: ${JSON.stringify(payload)}\n\n` - const stream = { - push(payloads: unknown[]) { - const frames = payloads.map(frame) - const controller = state.controller + + // Frames pushed while no stream is open, and the open stream that receives new ones. + const buffer: string[] = [] + let active: ReadableStreamDefaultController | undefined + let connections = 0 + + const frame = (json: string) => `data: ${json}\n\n` + + const stream: MockStream = { + push(payloads) { + const frames = payloads.map((payload) => frame(JSON.stringify(payload))) + const controller = active + if (!controller) { - state.buffer.push(...frames) + buffer.push(...frames) + return } + frames.forEach((item) => controller.enqueue(encoder.encode(item))) }, } + host.__mockServerStreams = { ...host.__mockServerStreams, [server]: stream } + const fetch = (input: RequestInfo | URL, init?: RequestInit) => { const request = new Request(input, init) const url = new URL(request.url) + if (url.origin !== server || url.pathname !== "/api/event") return originalFetch(request) - state.connections += 1 - const id = state.connections + connections += 1 + const id = connections let ended = false let own: ReadableStreamDefaultController | undefined let keepalive: ReturnType | undefined + const body = new ReadableStream({ start(controller) { own = controller - state.controller = controller + active = controller + if (retry !== undefined) controller.enqueue(encoder.encode(`retry: ${retry}\n\n`)) controller.enqueue( - encoder.encode(frame({ id: `evt_mock_connected_${id}`, type: "server.connected", data: {} })), + encoder.encode( + frame(JSON.stringify({ id: `evt_mock_connected_${id}`, type: "server.connected", data: {} })), + ), ) - state.buffer.splice(0).forEach((item) => controller.enqueue(encoder.encode(item))) + buffer.splice(0).forEach((item) => controller.enqueue(encoder.encode(item))) + // Match the real server's idle stream so long scenarios do not // trigger the client's 45-second stall watchdog and reload history. if (idle) keepalive = setInterval(() => controller.enqueue(encoder.encode(": keepalive\n\n")), 15_000) @@ -203,7 +245,8 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { if (ended) return ended = true clearInterval(keepalive) - if (state.controller === controller) state.controller = undefined + + if (active === controller) active = undefined controller.error(request.signal.reason ?? new DOMException("The operation was aborted", "AbortError")) }, { once: true }, @@ -213,9 +256,11 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { if (ended) return ended = true clearInterval(keepalive) - if (state.controller === own) state.controller = undefined + + if (active === own) active = undefined }, }) + return Promise.resolve( new Response(body, { status: 200, @@ -223,6 +268,7 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { }), ) } + Object.defineProperty(window, "fetch", { configurable: true, writable: true, value: fetch }) }, { server, retry: config.eventRetry, keepalive: config.keepalive !== false }, @@ -233,28 +279,35 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { const push = (payloads: readonly OpenCodeEvent[]) => page.evaluate( ({ server, payloads }) => { + // SAFETY: every field `MockStreamWindow` adds to a window is optional. const host = window as MockStreamWindow const stream = host.__mockServerStreams?.[server] + if (stream) return stream.push(payloads) const transport = host.__testSseTransports?.[server] + if (!transport) throw new Error(`No mock event stream for ${server}`) transport.command({ type: "send", deliveries: payloads.map((payload) => ({ payload })), burst: true }) }, + // SAFETY: events cross to the page as they are; as `unknown[]`, Playwright's argument type does not recurse into them. { server, payloads: payloads as unknown[] }, ) + // Server-side events the mock publishes itself; delivery failures other than a missing document fail the test. const emit = (events: OpenCodeEvent[]) => - void push(events).catch((error: unknown) => { - if (page.isClosed() || retryableDelivery(error)) return - throw error + void push(events).catch((cause: unknown) => { + if (page.isClosed() || retryableDelivery(cause)) return + throw cause }) if (config.events) { // Batches stay queued until the page accepts them; failures other than a missing document fail the test. - const pump = { busy: false, pending: [] as OpenCodeEvent[] } + const pump = { busy: false, pending: new Array() } + const timer = setInterval(() => { if (pump.busy) return pump.pending.push(...(config.events?.() ?? [])) + if (pump.pending.length === 0) return pump.busy = true const batch = pump.pending.slice() @@ -263,31 +316,39 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { () => { pump.pending.splice(0, batch.length) }, - (error: unknown) => { + (cause: unknown) => { if (page.isClosed()) return clearInterval(timer) - if (retryableDelivery(error)) return + + if (retryableDelivery(cause)) return clearInterval(timer) - throw error + throw cause }, ) .finally(() => { pump.busy = false }) }, 50) + page.on("close", () => clearInterval(timer)) } + const transport = createMockServerHandler(config, emit) page.on("close", () => void transport.dispose()) await page.route("**/api/**", async (route) => { const url = new URL(route.request().url()) + if (!answers(page, server, url)) return route.fallback() + // Production serves the UI and API from one origin; leave app assets to Vite. if (!url.pathname.startsWith("/api/")) return route.fallback() + if (route.request().method() === "OPTIONS") { return route.fulfill({ status: 204, headers: corsHeaders }) } + const password = config.password === undefined ? undefined : resolve(config.password) + if ( password !== undefined && (await route.request().headerValue("authorization")) !== `Basic ${btoa(`opencode:${password}`)}` @@ -295,15 +356,20 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { return route.fulfill({ status: 401, headers: corsHeaders, + // SAFETY: the server's wire body for a rejected password, which the client decodes into its own error. + // oxlint-disable-next-line anti-slop-effect/no-manual-tagged-construction -- see SAFETY above json: { _tag: "UnauthorizedError", message: "Authentication required" }, }) } + const directory = url.searchParams.get("directory") ?? url.searchParams.get("location[directory]") + if (config.strictDirectory && directory && !ownedDirectories(config).has(directory)) { return route.fulfill({ status: 500, headers: corsHeaders, json: { name: "InvalidDirectory" } }) } const body = route.request().postDataBuffer() + const response = await transport.handler( new Request(url, { method: route.request().method(), @@ -311,7 +377,9 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { body: body ? Uint8Array.from(body) : undefined, }), ) + const payload = Buffer.from(await response.arrayBuffer()) + // A handler's 404 carries a tagged error; a route the mock does not define must not reach the app server, whose // SPA fallback would answer HTML 200. Answer 501 and fail the test from the route handler. if (response.status === 404 && !payload.toString().includes('"_tag"')) { @@ -323,6 +391,7 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { }) throw new Error(`Unmocked API request: ${request} (add it to e2e/utils/mock-server.ts)`) } + return route.fulfill({ status: response.status, headers: { ...Object.fromEntries(response.headers), ...corsHeaders }, @@ -338,10 +407,13 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { const url = new URL(ws.url()) const id = decodeURIComponent(url.pathname.split("/")[3]!) const reason = transport.pty.admit(id, url) + if (reason) { transport.pty.rejected.push({ id, url, reason }) + return ws.close({ code: 1008, reason }) } + const socket: MockPtySocket = { id, url, @@ -349,6 +421,7 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) { closed: false, send: (data) => ws.send(data), } + ws.onMessage((message) => socket.input.push(message.toString())) ws.onClose(() => { socket.closed = true @@ -371,7 +444,9 @@ export async function mockServers(page: Page, servers: Record void = () => {}) { const pty = createPty(config) + const web = HttpRouter.toWebHandler( HttpApiBuilder.layer(MockApi).pipe( Layer.provide( @@ -390,7 +466,7 @@ export function createMockServerHandler(config: MockServerConfig, emit: (events: sessionCreates: 0, pty, emit, - mcp: new Map>(), + mcp: new Map(), attempts: new Map(), }), ), @@ -398,6 +474,7 @@ export function createMockServerHandler(config: MockServerConfig, emit: (events: ), { disableLogger: true }, ) + return { ...web, pty } } @@ -411,13 +488,16 @@ const corsHeaders = { const APP_ORIGIN = new URL( process.env.PLAYWRIGHT_BASE_URL ?? `http://127.0.0.1:${process.env.PLAYWRIGHT_PORT ?? "3000"}`, ).origin + const registeredOrigins = new WeakMap>() function mockedOrigins(page: Page) { const found = registeredOrigins.get(page) + if (found) return found const created = new Set() registeredOrigins.set(page, created) + return created } @@ -425,12 +505,14 @@ function mockedOrigins(page: Page) { // answers unless a server was configured for the app origin explicitly. function answers(page: Page, server: string, url: URL) { if (url.origin === server) return true + return server === SERVER && url.origin === APP_ORIGIN && !mockedOrigins(page).has(APP_ORIGIN) } // The document is not loaded yet or is being replaced; the pump retries on its next tick. -function retryableDelivery(error: unknown) { - const message = error instanceof Error ? error.message : String(error) +function retryableDelivery(cause: unknown) { + const message = cause instanceof Error ? cause.message : String(cause) + return [ "No mock event stream", "Execution context was destroyed", @@ -438,6 +520,16 @@ function retryableDelivery(error: unknown) { ].some((text) => message.includes(text)) } +// Prompt bodies are parsed JSON the mock API leaves undecoded. +const decodeJson = Schema.decodeUnknownSync(Schema.Json) + +const decodeFields = Schema.decodeUnknownOption(Schema.Record(Schema.String, Schema.Json)) + +// The fields of a JSON request body; a body that is not an object has none. +function fields(body: Schema.Json): MockFields { + return Option.getOrElse(decodeFields(body), () => ({})) +} + const PTY_TICKET = "e2e-ticket" function createPty(config: MockServerConfig) { @@ -450,12 +542,14 @@ function createPty(config: MockServerConfig) { status: "running", pid: 1, }) + // Core scopes PTYs by Location, which can differ from the process cwd, so ownership is kept apart from `cwd`. const owners = new Map((config.pty?.initial ?? []).map((item) => [item.id, item.directory ?? config.directory])) // Issued, not yet used connect tickets, each bound to one PTY and workspace. const tickets: { id: string; directory: string; ticket: string }[] = [] const issued = { count: 0 } const owns = (item: MockPtyInfo, directory: string) => owners.get(item.id) === directory + const pty: MockPty = { list: (config.pty?.initial ?? []).map((item) => info(item.id, item.title, item.cwd ?? item.directory)), created: [], @@ -466,19 +560,23 @@ function createPty(config: MockServerConfig) { rejected: [], send(data, id) { const socket = pty.sockets.findLast((item) => !item.closed && (id === undefined || item.id === id)) + if (!socket) throw new Error(`No open PTY socket${id ? ` for ${id}` : ""}`) socket.send(data) }, } + return Object.assign(pty, { info, // The first `${prefix}` no initial, created, or removed PTY has used. allocate() { const prefix = config.pty?.prefix ?? "pty_" const used = new Set([...pty.list, ...pty.created].map((item) => item.id).concat(pty.removed)) + const number = Array.from({ length: used.size + 1 }, (_, index) => index + 1).find( (value) => !used.has(`${prefix}${value}`), )! + return { id: `${prefix}${number}`, number } }, add(created: MockPtyInfo, directory: string) { @@ -493,19 +591,24 @@ function createPty(config: MockServerConfig) { issued.count += 1 const ticket = issued.count === 1 ? PTY_TICKET : `${PTY_TICKET}-${issued.count}` tickets.push({ id, directory, ticket }) + return ticket }, // Returns why a connect URL is refused: the PTY must exist, belong to the requested workspace, and carry the exact // unused ticket issued for that PTY and workspace. Admission consumes the ticket. admit(id: string, url: URL) { const found = pty.list.find((item) => item.id === id) + if (!found) return "PTY not found" const directory = url.searchParams.get("location[directory]") || config.directory + if (!owns(found, directory)) return "PTY belongs to another workspace" const ticket = url.searchParams.get("ticket") + const index = tickets.findIndex( (item) => item.id === id && item.directory === directory && item.ticket === ticket, ) + if (index < 0) return "No unused ticket was issued for this PTY" tickets.splice(index, 1) }, @@ -514,7 +617,8 @@ function createPty(config: MockServerConfig) { // Every directory this server's configuration names: its own, project and inventory worktrees, and session locations. function ownedDirectories(config: MockServerConfig) { - const projects = [config.project, ...(config.projects ? resolve(config.projects) : [])].filter(record) + const projects = [config.project, ...(config.projects ? resolve(config.projects) : [])].filter(Predicate.isObject) + return new Set( [ config.directory, @@ -523,16 +627,17 @@ function ownedDirectories(config: MockServerConfig) { item.canonical, ...(Array.isArray(item.sandboxes) ? item.sandboxes : []), ]), - ...(config.worktrees ? resolve(config.worktrees) : []).filter(record).map((item) => item.directory), - ...config.sessions.map((session) => (record(session.location) ? session.location.directory : session.directory)), - ].filter((item): item is string => typeof item === "string"), + ...(config.worktrees ? resolve(config.worktrees) : []).filter(Predicate.isObject).map((item) => item.directory), + ...config.sessions.map((session) => + Predicate.isObject(session.location) ? session.location.directory : session.directory, + ), + ].filter(Predicate.isString), ) } function addSandbox(config: MockServerConfig, directory: string) { - const project = config.project as { sandboxes?: string[] } - if (project.sandboxes?.includes(directory)) return - project.sandboxes = [...(project.sandboxes ?? []), directory] + if (config.project.sandboxes?.includes(directory)) return + config.project.sandboxes = [...(config.project.sandboxes ?? []), directory] } // The requested `location[directory]`; absent or empty (a server-level read) means the server directory. @@ -541,7 +646,8 @@ function requestDirectory(config: MockServerConfig, request: { url: string }) { } function resolve(value: Resolvable) { - return typeof value === "function" ? (value as () => T)() : value + // SAFETY: a fixture is data, never a function, so a function is the fixture's resolver. + return Predicate.isFunction(value) ? (value as () => T)() : value } function mockHandlers( @@ -553,16 +659,18 @@ function mockHandlers( pty: ReturnType emit: (events: OpenCodeEvent[]) => void // MCP status overrides by `\n`, and OAuth attempt creation times by attempt ID. - mcp: Map> + mcp: Map attempts: Map }, ) { const noContent = Effect.succeed(HttpApiSchema.NoContent.make()) const delay = config.messageDelay === undefined ? Effect.void : Effect.sleep(Duration.millis(config.messageDelay)) const configEntries = config.configEntries ?? [] + const ptyEnabled = Effect.suspend(() => config.pty ? Effect.void : Effect.fail(new MockNotFound({ message: "PTY is not enabled for this mock server" })), ) + // PTYs are scoped to the workspace (`location[directory]`) they were created in, like the real server. const findPty = (id: string, request: { url: string }) => ptyEnabled.pipe( @@ -570,46 +678,60 @@ function mockHandlers( Effect.suspend(() => { const directory = requestDirectory(config, request) const found = state.pty.find(id, directory) + return found ? Effect.succeed(found) : Effect.fail(new MockNotFound({ message: "PTY not found" })) }), ), ) + const mcpServers = (directory: string) => - (typeof config.mcp === "function" ? config.mcp(directory) : (config.mcp ?? [])).map((server) => { - const status = record(server) ? state.mcp.get(`${directory}\n${String(server.name)}`) : undefined - return status && record(server) ? { ...server, status } : server + (Predicate.isFunction(config.mcp) ? config.mcp(directory) : (config.mcp ?? [])).map((server) => { + const status = Predicate.isObject(server) ? state.mcp.get(`${directory}\n${String(server.name)}`) : undefined + + return status && Predicate.isObject(server) ? { ...server, status } : server }) + const mcpAction = (server: string, action: "connect" | "disconnect", request: { url: string }) => Effect.suspend(() => { const directory = requestDirectory(config, request) - if (!mcpServers(directory).some((item) => record(item) && item.name === server)) + + if (!mcpServers(directory).some((item) => Predicate.isObject(item) && item.name === server)) return Effect.fail(new MockNotFound({ message: `MCP server ${server} not found` })) + const status = config.onMcpAction?.({ server, action, directory }) ?? { status: action === "connect" ? "connected" : "disabled", } + state.mcp.set(`${directory}\n${server}`, status) + return noContent }) + const unsupported = (operation: string, handler: string) => Effect.fail( new MockUnsupported({ message: `The mock server does not ${operation}; configure ${handler} for this scenario` }), ) + return HttpApiBuilder.group(MockApi, "mock", (handlers) => handlers .handleRaw("event", () => { const events = config.events?.() const retry = config.eventRetry === undefined ? "" : `retry: ${config.eventRetry}\n\n` + const body = [{ id: "evt_mock_connected", type: "server.connected", data: {} }, ...(events ?? [])] .map((event) => `data: ${JSON.stringify(event)}\n\n`) .join("") + return Effect.succeed(HttpServerResponse.text(retry + body, { contentType: "text/event-stream" })) }) .handleRaw("fsRead", (ctx) => Effect.gen(function* () { const path = decodeURIComponent(new URL(ctx.request.url, "http://localhost").pathname.slice(13)) const value = yield* Effect.promise(() => Promise.resolve(config.fileContent?.(path))) + const content = - value && typeof value === "object" && "content" in value ? String(value.content) : String(value ?? "") + Predicate.isObjectOrArray(value) && "content" in value ? String(value.content) : String(value ?? "") + return HttpServerResponse.uint8Array(new TextEncoder().encode(content)) }), ) @@ -617,40 +739,47 @@ function mockHandlers( .handleRaw("worktreeCreate", (ctx) => Effect.gen(function* () { const create = config.onWorktreeCreate + if (!create) return yield* unsupported("create worktrees", "onWorktreeCreate") const input = yield* Effect.orDie(ctx.request.json) - const payload = record(input) ? input : {} + const payload = fields(input) + const answer = (yield* Effect.promise(async () => create(input))) || { status: 200, body: { - directory: `${typeof payload.directory === "string" ? payload.directory : config.directory}/${ - typeof payload.name === "string" ? payload.name : "copy" + directory: `${Predicate.isString(payload.directory) ? payload.directory : config.directory}/${ + Predicate.isString(payload.name) ? payload.name : "copy" }`, }, } - if (answer.status === 200 && record(answer.body) && typeof answer.body.directory === "string") + + if (answer.status === 200 && Predicate.isObject(answer.body) && Predicate.isString(answer.body.directory)) addSandbox(config, answer.body.directory) + return HttpServerResponse.jsonUnsafe(answer.body, { status: answer.status }) }), ) .handleRaw("sessionCreate", (ctx) => Effect.gen(function* () { - const input = yield* Effect.orDie(ctx.request.json) - const payload = record(input) ? input : {} + const payload = fields(yield* Effect.orDie(ctx.request.json)) state.sessionCreates += 1 const answer = config.onSessionCreate?.(payload, state.sessionCreates) + if (answer) return HttpServerResponse.jsonUnsafe(answer.body, { status: answer.status }) + const created = currentSession( { ...payload, - id: typeof payload.id === "string" ? payload.id : "ses_mock_created", - projectID: (config.project as { id?: string }).id, - title: config.createdSessionTitle ?? (typeof payload.title === "string" ? payload.title : "New session"), - parentID: typeof payload.parentID === "string" ? payload.parentID : undefined, + id: Predicate.isString(payload.id) ? payload.id : "ses_mock_created", + projectID: config.project.id, + title: config.createdSessionTitle ?? (Predicate.isString(payload.title) ? payload.title : "New session"), + parentID: Predicate.isString(payload.parentID) ? payload.parentID : undefined, }, config.directory, ) + config.sessions.push(created) + return HttpServerResponse.jsonUnsafe({ data: created }) }), ) @@ -668,7 +797,7 @@ function mockHandlers( location: { directory: config.directory, project: { - id: (config.project as { id?: string }).id, + id: config.project.id, directory: config.directory, canonical: config.directory, }, @@ -698,7 +827,7 @@ function mockHandlers( Effect.succeed({ location: location(config), data: config.integrations - ?.filter(record) + ?.filter(Predicate.isObject) .find((integration) => integration.id === ctx.params.integrationID) ?? { id: ctx.params.integrationID, name: ctx.params.integrationID, @@ -712,13 +841,16 @@ function mockHandlers( ), integrationOAuthConnect: (ctx) => { const start = config.onIntegrationOAuth + if (!start) return unsupported("start OAuth connections", "onIntegrationOAuth") + return Effect.sync(() => { const directory = requestDirectory(config, ctx.request) const created = Date.now() const attemptID = `con_mock_${state.attempts.size + 1}` state.attempts.set(attemptID, created) const started = start({ integrationID: ctx.params.integrationID, directory, body: ctx.payload }) + return { location: location(config, directory), data: { @@ -734,7 +866,9 @@ function mockHandlers( integrationOAuthStatus: (ctx) => Effect.suspend(() => { const created = state.attempts.get(ctx.params.attemptID) + if (created === undefined) return Effect.fail(new MockNotFound({ message: "OAuth attempt not found" })) + return Effect.succeed({ location: location(config, requestDirectory(config, ctx.request)), data: { status: "pending", time: { created, expires: created + 600_000 } }, @@ -751,6 +885,7 @@ function mockHandlers( mcp: (ctx) => Effect.sync(() => { const directory = requestDirectory(config, ctx.request) + return { location: location(config, directory), data: mcpServers(directory) } }), mcpConnect: (ctx) => mcpAction(ctx.params.server, "connect", ctx.request), @@ -763,27 +898,28 @@ function mockHandlers( projectList: () => Effect.sync(() => { if (config.projects) return resolve(config.projects) - const project = config.project as typeof config.project & { canonical?: string; worktree?: string } - return [{ ...project, canonical: project.canonical ?? project.worktree ?? config.directory }] + + return [ + { ...config.project, canonical: config.project.canonical ?? config.project.worktree ?? config.directory }, + ] }), - projectUpdate: (ctx) => { - const project = config.project as { canonical?: string } - return Effect.succeed({ - ...project, + projectUpdate: (ctx) => + Effect.succeed({ + ...config.project, ...ctx.payload, id: ctx.params.projectID, - canonical: project.canonical ?? config.directory, - }) - }, + canonical: config.project.canonical ?? config.directory, + }), configShells: () => Effect.succeed(config.shells ?? []), configUpdate: () => noContent, websearchProviders: () => Effect.succeed({ location: location(config), data: [] }), worktreeList: () => Effect.sync(() => { if (config.worktrees) return resolve(config.worktrees) + return [ { directory: config.directory }, - ...((config.project as { sandboxes?: string[] }).sandboxes ?? []).map((directory) => ({ + ...(config.project.sandboxes ?? []).map((directory) => ({ directory, strategy: "git", })), @@ -791,7 +927,9 @@ function mockHandlers( }), worktreeRemove: (ctx) => { const remove = config.onWorktreeRemove + if (!remove) return unsupported("remove worktrees", "onWorktreeRemove") + return Effect.promise(async () => remove(ctx.payload)).pipe(Effect.andThen(noContent)) }, // Discovery against a static inventory changes nothing, and the app refreshes whenever worktree settings open. @@ -803,16 +941,13 @@ function mockHandlers( ? Effect.fail(new MockInternal({ message: "Permission list failed" })) : Effect.succeed({ location: location(config), - data: (typeof config.permissions === "function" - ? config.permissions() - : (config.permissions ?? []) - ).map(currentPermission), + data: resolve(config.permissions ?? []).map(currentPermission), }), ), formRequests: () => Effect.succeed({ location: location(config), - data: typeof config.forms === "function" ? config.forms() : (config.forms ?? []), + data: resolve(config.forms ?? []), }), vcs: () => Effect.succeed({ @@ -821,11 +956,13 @@ function mockHandlers( }), vcsInit: (ctx) => { if (!config.onVcsInit) return unsupported("initialize VCS", "onVcsInit") + return Effect.sync(() => { const url = new URL(ctx.request.url, "http://localhost") const provider = url.searchParams.get("provider") ?? undefined config.onVcsInit?.({ directory: requestDirectory(config, ctx.request), provider }) - const project = config.project as { id: string; vcs?: string } + // SAFETY: a project that initializes its VCS is a full project fixture, which has an id. + const project = config.project as MockProject & { id: string } project.vcs = provider ?? "git" state.emit([ { @@ -842,8 +979,9 @@ function mockHandlers( vcsDiff: (ctx) => Effect.sync(() => ({ location: location(config), - data: - typeof config.vcsDiff === "function" ? config.vcsDiff({ mode: ctx.query.mode }) : (config.vcsDiff ?? []), + data: Predicate.isFunction(config.vcsDiff) + ? config.vcsDiff({ mode: ctx.query.mode }) + : (config.vcsDiff ?? []), })), fsList: (ctx) => Effect.promise(() => Promise.resolve(config.fileList?.(ctx.query.path ?? ""))).pipe( @@ -859,7 +997,7 @@ function mockHandlers( location: location(config), data: Array.isArray(entries) ? entries.map((entry) => - typeof entry === "string" + Predicate.isString(entry) ? { name: entry.split(/[\\/]/).at(-1) ?? entry, path: entry, @@ -874,11 +1012,14 @@ function mockHandlers( ), fsWrite: (ctx) => { const write = config.onFileWrite + if (!write) return unsupported("write files", "onFileWrite") + return Effect.sync(() => { const directory = requestDirectory(config, ctx.request) const path = new URL(ctx.request.url, "http://localhost").searchParams.get("path") ?? "" write({ path, directory, body: new TextDecoder().decode(ctx.payload) }) + return { location: location(config, directory), data: { path } } }) }, @@ -891,6 +1032,7 @@ function mockHandlers( Effect.suspend(() => { const directory = requestDirectory(config, ctx.request) const output = config.shellOutput?.({ id: ctx.params.id, directory }) + if (output === undefined) return Effect.fail( new MockShellNotFound({ id: ctx.params.id, message: `Shell command not found: ${ctx.params.id}` }), @@ -900,6 +1042,7 @@ function mockHandlers( const cursor = Math.min(Number(query.get("cursor") ?? 0), bytes.length) // The server answers at most one page (`Shell.output` defaults `limit` to 65,536 bytes). const end = Math.min(bytes.length, cursor + Number(query.get("limit") ?? 65_536)) + return Effect.succeed({ location: location(config, directory), data: { @@ -914,6 +1057,7 @@ function mockHandlers( ptyEnabled.pipe( Effect.map(() => { const directory = requestDirectory(config, ctx.request) + return { location: location(config, directory), data: state.pty.owned(directory) } }), ), @@ -922,12 +1066,15 @@ function mockHandlers( Effect.map(() => { const next = state.pty.allocate() const directory = requestDirectory(config, ctx.request) + const created = state.pty.info( next.id, ctx.payload.title ?? `Terminal ${next.number}`, ctx.payload.cwd ?? directory, ) + state.pty.add(created, directory) + return { location: location(config, directory), data: created } }), ), @@ -939,7 +1086,9 @@ function mockHandlers( findPty(ctx.params.ptyID, ctx.request).pipe( Effect.map((found) => { state.pty.updates.push({ id: found.id, body: ctx.payload }) + if (ctx.payload.title) found.title = ctx.payload.title + return { location: location(config, requestDirectory(config, ctx.request)), data: found } }), ), @@ -948,6 +1097,7 @@ function mockHandlers( Effect.map((found) => { state.pty.removed.push(found.id) state.pty.list.splice(state.pty.list.indexOf(found), 1) + return HttpApiSchema.NoContent.make() }), ), @@ -961,22 +1111,23 @@ function mockHandlers( headers: Object.fromEntries(Object.entries(ctx.request.headers)), ticket, }) + return { location: location(config, directory), data: { ticket, expires_in: 60 } } }), ), sessionList: (ctx) => { const sessions = config.sessions - .filter((session) => { - const location = session.location as { directory?: string } | undefined - return ( + .filter( + (session) => !ctx.query.directory || - location?.directory === ctx.query.directory || - session.directory === ctx.query.directory - ) - }) + (Predicate.isObject(session.location) && session.location.directory === ctx.query.directory) || + session.directory === ctx.query.directory, + ) .filter((session) => { if (ctx.query.parentID === undefined) return true + if (ctx.query.parentID === "null") return session.parentID === undefined + return session.parentID === ctx.query.parentID }) .filter((session) => @@ -986,19 +1137,20 @@ function mockHandlers( .toLowerCase() .includes(ctx.query.search.toLowerCase()), ) + const ordered = ctx.query.order === "asc" ? sessions : sessions.toReversed() const offset = Number(ctx.query.cursor ?? 0) const limit = ctx.query.limit ?? 50 const data = ordered.slice(offset, offset + limit) + return Effect.succeed({ data: data.map((session) => currentSession(session, config.directory)), cursor: { next: offset + limit < ordered.length ? String(offset + limit) : undefined }, }) }, sessionActive: () => { - const statuses = ( - typeof config.sessionStatus === "function" ? config.sessionStatus() : (config.sessionStatus ?? {}) - ) as Record + const statuses = config.sessionStatus === undefined ? {} : resolve(config.sessionStatus) + return Effect.succeed({ data: Object.fromEntries( Object.entries(statuses).flatMap(([id, status]) => @@ -1011,6 +1163,7 @@ function mockHandlers( Effect.suspend(() => { config.onSession?.(ctx.params.sessionID) const session = config.sessions.find((item) => item.id === ctx.params.sessionID) + return session ? Effect.succeed({ data: currentSession(session, config.directory) }) : Effect.fail(new MockNotFound({ message: "Session not found" })) @@ -1018,33 +1171,30 @@ function mockHandlers( sessionRemove: () => noContent, sessionShell: () => noContent, sessionForm: (ctx) => { - const forms = typeof config.forms === "function" ? config.forms() : (config.forms ?? []) return Effect.succeed({ - data: forms.filter((form) => (form as { sessionID?: string }).sessionID === ctx.params.sessionID), + data: resolve(config.forms ?? []).filter((form) => form.sessionID === ctx.params.sessionID), }) }, sessionFormReply: () => noContent, sessionFormCancel: () => noContent, sessionBackground: () => noContent, - sessionInbox: () => - Effect.sync(() => ({ data: typeof config.inbox === "function" ? config.inbox() : (config.inbox ?? []) })), + sessionInbox: () => Effect.sync(() => ({ data: resolve(config.inbox ?? []) })), sessionPrompt: (ctx) => Effect.sync(() => { - const body = record(ctx.payload) ? ctx.payload : {} + const body = fields(decodeJson(ctx.payload)) config.onPrompt?.({ sessionID: ctx.params.sessionID, body }) + + const parts = ["files", "agents", "skills", "metadata"].flatMap((key) => + body[key] === undefined ? [] : [[key, body[key]]], + ) + return { data: { - id: typeof body.id === "string" ? body.id : `inb_mock_${Date.now()}`, + id: Predicate.isString(body.id) ? body.id : `inb_mock_${Date.now()}`, sessionID: ctx.params.sessionID, time: { created: Date.now() }, type: "user", - payload: { - text: typeof body.text === "string" ? body.text : "", - ...(body.files === undefined ? {} : { files: body.files }), - ...(body.agents === undefined ? {} : { agents: body.agents }), - ...(body.skills === undefined ? {} : { skills: body.skills }), - ...(body.metadata === undefined ? {} : { metadata: body.metadata }), - }, + payload: { text: Predicate.isString(body.text) ? body.text : "", ...Object.fromEntries(parts) }, delivery: body.delivery === "queue" ? "queue" : "steer", }, } @@ -1079,20 +1229,23 @@ function mockHandlers( // Like the server, a reply publishes `permission.replied`, and later reads no longer list the request. sessionPermissionReply: (ctx) => { const reply = config.onPermissionReply + if (!reply) return unsupported("record permission replies", "onPermissionReply") + return Effect.sync(() => { const sessionID = ctx.params.sessionID const permissionID = ctx.params.permissionID reply({ sessionID, permissionID, body: ctx.payload }) - const pending = [ - config.sessionPermissions?.[sessionID] ?? [], - typeof config.permissions === "function" ? config.permissions() : (config.permissions ?? []), - ] + + const pending = [config.sessionPermissions?.[sessionID] ?? [], resolve(config.permissions ?? [])] + pending.forEach((list) => { - const index = list.findIndex((item) => record(item) && item.id === permissionID) + const index = list.findIndex((item) => item.id === permissionID) + if (index >= 0) list.splice(index, 1) }) state.emit([ + // SAFETY: `reply` echoes the decision the client sent, which is one of the permission reply kinds. { id: `evt_permission_replied_${permissionID}`, created: Date.now(), @@ -1102,7 +1255,9 @@ function mockHandlers( sessionID, requestID: permissionID, reply: - record(ctx.payload) && typeof ctx.payload.decision === "string" ? ctx.payload.decision : "once", + Predicate.isObject(ctx.payload) && Predicate.isString(ctx.payload.decision) + ? ctx.payload.decision + : "once", }, } as OpenCodeEvent, ]) @@ -1110,24 +1265,28 @@ function mockHandlers( }, sessionRename: (ctx) => Effect.sync(() => { - const title = record(ctx.payload) ? ctx.payload.title : undefined + const title = Predicate.isObject(ctx.payload) ? ctx.payload.title : undefined const session = config.sessions.find((item) => item.id === ctx.params.sessionID) - if (session && typeof title === "string") session.title = title + + if (session && Predicate.isString(title)) session.title = title }).pipe(Effect.andThen(noContent)), sessionCommand: (ctx) => { const recordCommand = config.onCommand + if (!recordCommand) return unsupported("run session commands", "onCommand") + return Effect.sync(() => recordCommand({ sessionID: ctx.params.sessionID, body: ctx.payload })).pipe( Effect.andThen(noContent), ) }, sessionInterrupt: () => noContent, sessionRevertStage: (ctx) => { - const payload = record(ctx.payload) ? ctx.payload : {} - const messageID = payload.messageID - if (typeof messageID !== "string") { + const messageID = Predicate.isObject(ctx.payload) ? ctx.payload.messageID : undefined + + if (!Predicate.isString(messageID)) { return Effect.fail(new MockBadRequest({ message: "Invalid revert request" })) } + return Effect.sync(() => config.onRevertStage?.({ sessionID: ctx.params.sessionID, messageID })).pipe( Effect.as({ data: { messageID } }), ) @@ -1138,28 +1297,37 @@ function mockHandlers( Effect.gen(function* () { config.onMessage?.({ sessionID: ctx.params.sessionID, messageID: ctx.params.messageID }) yield* delay + const message = config.message?.(ctx.params.sessionID, ctx.params.messageID) ?? config .pageMessages(ctx.params.sessionID, Number.MAX_SAFE_INTEGER) .items.find((item) => item.id === ctx.params.messageID) + if (!message) return yield* new MockNotFound({ message: "Message not found" }) + return { data: message } }), messageList: (ctx) => { const token = ctx.query.cursor const before = token ? state.cursors.get(token) : undefined + if (token && !before) return Effect.fail(new MockBadRequest({ message: "Invalid cursor" })) + return Effect.gen(function* () { config.onMessages?.({ sessionID: ctx.params.sessionID, before, phase: "start" }) + if (config.beforeMessagesResponse) { yield* Effect.promise(() => config.beforeMessagesResponse!({ sessionID: ctx.params.sessionID, before })) } + yield* delay const pageData = config.pageMessages(ctx.params.sessionID, ctx.query.limit ?? 50, before) config.onMessages?.({ sessionID: ctx.params.sessionID, before, phase: "end" }) const cursor = pageData.cursor ? `cursor_${++state.nextCursor}` : undefined + if (cursor) state.cursors.set(cursor, pageData.cursor!) + return { data: ctx.query.order === "asc" ? pageData.items : pageData.items.toReversed(), cursor: { next: cursor }, @@ -1174,21 +1342,21 @@ function mockHandlers( function location(config: MockServerConfig, directory = config.directory) { return { directory, - project: { id: (config.project as { id?: string }).id, directory: config.directory, canonical: config.directory }, + project: { id: config.project.id, directory: config.directory, canonical: config.directory }, } } function providerConfig(config: MockServerConfig) { - return typeof config.provider === "function" ? config.provider() : config.provider + return resolve(config.provider) } -function currentProviders(value: unknown) { - if (!record(value) || !Array.isArray(value.all)) return Array.isArray(value) ? value : [] - const connected = new Set( - Array.isArray(value.connected) ? value.connected.filter((id) => typeof id === "string") : [], - ) - return value.all.filter(record).flatMap((provider) => - typeof provider.id === "string" && typeof provider.name === "string" +function currentProviders(value: MockFixture) { + if (!Predicate.isObject(value) || !Array.isArray(value.all)) return Array.isArray(value) ? value : [] + + const connected = new Set(Array.isArray(value.connected) ? value.connected.filter(Predicate.isString) : []) + + return value.all.filter(isFields).flatMap((provider) => + Predicate.isString(provider.id) && Predicate.isString(provider.name) ? [ { id: provider.id, @@ -1201,42 +1369,46 @@ function currentProviders(value: unknown) { ) } -function currentModels(value: unknown) { - if (!record(value) || !Array.isArray(value.all)) return [] - return value.all.filter(record).flatMap((provider) => { - if (typeof provider.id !== "string" || !record(provider.models)) return [] +function currentModels(value: MockFixture) { + if (!Predicate.isObject(value) || !Array.isArray(value.all)) return [] + + return value.all.filter(isFields).flatMap((provider) => { + if (!Predicate.isString(provider.id) || !isFields(provider.models)) return [] + return Object.values(provider.models) - .filter(record) + .filter(isFields) .flatMap((model) => { - if (typeof model.id !== "string" || typeof model.name !== "string") return [] - const limit = record(model.limit) ? model.limit : {} - const cost = record(model.cost) ? model.cost : {} + if (!Predicate.isString(model.id) || !Predicate.isString(model.name)) return [] + const limit = Predicate.isObject(model.limit) ? model.limit : {} + const cost = Predicate.isObject(model.cost) ? model.cost : {} + return [ { id: model.id, - modelID: record(model.api) && typeof model.api.id === "string" ? model.api.id : model.id, + modelID: Predicate.isObject(model.api) && Predicate.isString(model.api.id) ? model.api.id : model.id, providerID: provider.id, name: model.name, capabilities: { tools: true, input: ["text"], output: ["text"] }, - variants: record(model.variants) - ? Object.entries(model.variants).map(([id, settings]) => ({ - id, - ...(jsonRecord(settings) ? { settings: jsonRecord(settings) } : {}), - })) + variants: isFields(model.variants) + ? Object.entries(model.variants).map(([id, value]) => { + const settings = jsonRecord(value) + + return settings ? { id, settings } : { id } + }) : [], time: { released: Date.now() }, cost: [ { - input: typeof cost.input === "number" ? cost.input : 0, - output: typeof cost.output === "number" ? cost.output : 0, + input: Predicate.isNumber(cost.input) ? cost.input : 0, + output: Predicate.isNumber(cost.output) ? cost.output : 0, cache: { read: 0, write: 0 }, }, ], status: "active", enabled: true, limit: { - context: typeof limit.context === "number" ? limit.context : 200_000, - output: typeof limit.output === "number" ? limit.output : 32_000, + context: Predicate.isNumber(limit.context) ? limit.context : 200_000, + output: Predicate.isNumber(limit.output) ? limit.output : 32_000, }, }, ] @@ -1244,17 +1416,18 @@ function currentModels(value: unknown) { }) } -function currentDefaultModel(value: unknown) { - if (!record(value) || !record(value.default)) return null +function currentDefaultModel(value: MockFixture) { + if (!Predicate.isObject(value) || !Predicate.isObject(value.default)) return null const selected = value.default const models = currentModels(value) + return models.find((model) => model.providerID === selected.providerID && model.id === selected.modelID) ?? null } -function currentPermission(value: unknown) { - const permission = value as Record +function currentPermission(permission: MockFields) { if (permission.action) return permission - const tool = permission.tool as { messageID?: string; callID?: string; id?: string } | undefined + const tool = Predicate.isObject(permission.tool) ? permission.tool : undefined + return { id: permission.id, sessionID: permission.sessionID, @@ -1269,10 +1442,11 @@ function currentPermission(value: unknown) { } } -export function currentSession(session: { id: string } & Record, fallbackDirectory?: string) { - const time = session.time && typeof session.time === "object" ? session.time : {} - const location = session.location && typeof session.location === "object" ? session.location : {} - return { +export function currentSession(session: MockSession, fallbackDirectory?: string) { + const time: MockFields = Predicate.isObject(session.time) ? session.time : {} + const location: MockFields = Predicate.isObject(session.location) ? session.location : {} + + const current: MockSession = { id: session.id, parentID: session.parentID, projectID: session.projectID ?? "project", @@ -1280,47 +1454,60 @@ export function currentSession(session: { id: string } & Record model: session.model ?? { id: "mock-model", providerID: "mock-provider" }, cost: session.cost ?? 0, tokens: session.tokens ?? { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, - ...(typeof session.outcome === "string" ? { outcome: session.outcome } : {}), - time: { - created: "created" in time && typeof time.created === "number" ? time.created : 0, - updated: "updated" in time && typeof time.updated === "number" ? time.updated : 0, - ...("idle" in time && typeof time.idle === "number" ? { idle: time.idle } : {}), - ...("viewed" in time && typeof time.viewed === "number" ? { viewed: time.viewed } : {}), - ...(session.time && typeof session.time === "object" && "archived" in session.time - ? { archived: session.time.archived } - : {}), - }, - title: session.title ?? session.id, - location: { - directory: - "directory" in location && typeof location.directory === "string" - ? location.directory - : typeof session.directory === "string" - ? session.directory - : fallbackDirectory, - }, - subpath: session.subpath ?? session.path, - revert: session.revert, } + + if (Predicate.isString(session.outcome)) current.outcome = session.outcome + + const times: MockSessionTime = { + created: Predicate.isNumber(time.created) ? time.created : 0, + updated: Predicate.isNumber(time.updated) ? time.updated : 0, + } + + if (Predicate.isNumber(time.idle)) times.idle = time.idle + + if (Predicate.isNumber(time.viewed)) times.viewed = time.viewed + + if ("archived" in time) times.archived = time.archived + current.time = times + current.title = session.title ?? session.id + + current.location = { + directory: Predicate.isString(location.directory) + ? location.directory + : Predicate.isString(session.directory) + ? session.directory + : fallbackDirectory, + } + + current.subpath = session.subpath ?? session.path + current.revert = session.revert + + return current } -function jsonRecord(value: unknown): Record | undefined { - if (!record(value)) return +function jsonRecord(value: MockFixture): Record | undefined { + if (!Predicate.isObject(value)) return + return Object.fromEntries( Object.entries(value).flatMap(([key, item]) => { const next = jsonValue(item) + return next === undefined ? [] : [[key, next]] }), ) } -function jsonValue(value: unknown): JsonValue | undefined { - if (value === null || typeof value === "string" || typeof value === "boolean") return value - if (typeof value === "number") return Number.isFinite(value) ? value : null +function jsonValue(value: MockFixture): JsonValue | undefined { + if (value === null || Predicate.isString(value) || Predicate.isBoolean(value)) return value + + if (Predicate.isNumber(value)) return Number.isFinite(value) ? value : null + if (Array.isArray(value)) return value.map((item) => jsonValue(item) ?? null) + return jsonRecord(value) } -function record(value: unknown): value is Record { - return !!value && typeof value === "object" && !Array.isArray(value) +// A fixture object, as opposed to a list or a primitive value. +function isFields(value: MockFixture): value is MockFields { + return Predicate.isObject(value) } diff --git a/packages/app/src/providers/models/selection.tsx b/packages/app/src/providers/models/selection.tsx index 5ad40f52dd6f..dbc1c342cbc7 100644 --- a/packages/app/src/providers/models/selection.tsx +++ b/packages/app/src/providers/models/selection.tsx @@ -23,17 +23,20 @@ const ModelKeySchema = Schema.Struct({ modelID: Schema.String, variant: Schema.optional(Schema.String), }) + export type ModelKey = typeof ModelKeySchema.Type const ChoiceSchema = Schema.Struct({ model: Persistence.optional(ModelKeySchema), variant: Persistence.optional(Schema.NullOr(Schema.String)), }) + const StateSchema = Schema.Struct({ ...ChoiceSchema.fields, agent: Persistence.optional(Schema.String), choices: Persistence.optional(Schema.Record(Schema.String, ChoiceSchema)), }) + type State = typeof StateSchema.Type const SessionsSchema = Schema.Record( @@ -61,12 +64,14 @@ export const ModelSelectionSchema = Persistence.migrate( ) const WORKSPACE_KEY = "__workspace__" + const handoff = new Map() const handoffKey = (scope: ServerScope, dir: string, id: string) => ScopedKey.from(scope, dir, id) const clone = (value: State | undefined) => { if (!value) return + return { ...value, model: value.model ? { ...value.model } : undefined, @@ -86,11 +91,13 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ const configuredModel = useConfiguredModel() const id = createMemo(() => params.id || undefined) + const list = createMemo(() => normalizeAgentList(data.location.agent.list({ directory: sdk().directory }) ?? []).filter( (item) => item.mode !== "subagent" && !item.hidden, ), ) + const agentsVisible = createMemo(() => settings.visibility.customAgents() || hasCustomAgent(list())) const connected = createMemo(() => new Set(providers.connected().map((item) => item.id))) @@ -111,13 +118,16 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ const validModel = (model: ModelKey) => { const provider = providers.all().get(model.providerID) + return !!provider?.models[model.modelID] && connected().has(model.providerID) } const firstModel = (...items: Array<() => ModelKey | undefined>) => { for (const item of items) { const model = item() + if (!model) continue + if (validModel(model)) return model } } @@ -128,30 +138,39 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ createEffect(() => { const items = list() + if (items.length === 0) { if (store.current !== undefined) setStore("current", undefined) + return } + if (items.some((item) => item.name === store.current)) return setStore("current", items[0]?.name) }) const scope = createMemo(() => { const session = id() + if (!session) return store.draft ?? store.promoting + return saved.session[session] ?? handoff.get(handoffKey(serverSDK.scope, sdk().directory, session)) }) createEffect(() => { const session = id() + if (!session) return const key = handoffKey(serverSDK.scope, sdk().directory, session) const next = handoff.get(key) + if (!next) return + if (saved.session[session] !== undefined) { handoff.delete(key) setStore("promoting", undefined) + return } @@ -169,15 +188,19 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ const defaultModel = () => { for (const provider of providers.connected()) { const first = Object.values(provider.models)[0] + if (!first) continue const model = { providerID: provider.id, modelID: first.id } + if (validModel(model)) return model } } const fallback = createMemo(() => firstModel(configuredModel, recentModel, defaultModel)) + const durable = () => { const session = id() + return session ? data.session.get(session) : undefined } @@ -189,43 +212,55 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ }, set(name: string | undefined) { const item = pickAgent(name) + if (!item) { setStore("current", undefined) + return } batch(() => { const previous = snapshot() + if (previous.agent === item.name) return const prev = scope() - const choices = { - ...prev?.choices, - ...(previous.agent ? { [previous.agent]: { model: previous.model, variant: previous.variant } } : {}), - } + + const choices = { ...prev?.choices } + + if (previous.agent) choices[previous.agent] = { model: previous.model, variant: previous.variant } + setStore("current", item.name) + const next = { agent: item.name, model: choices[item.name]?.model, variant: choices[item.name]?.variant, choices, } satisfies State + write(next) // Pin both choices while the agent and model acknowledgments arrive separately. const selected = current() + if (selected) model.set({ providerID: selected.provider.id, modelID: selected.id }) }) }, move(direction: 1 | -1) { const items = list() + if (items.length === 0) { setStore("current", undefined) + return } let next = items.findIndex((item) => item.name === agent.current()?.name) + direction + if (next < 0) next = items.length - 1 + if (next >= items.length) next = 0 const item = items[next] + if (!item) return agent.set(item.name) }, @@ -233,26 +268,33 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ const current = createMemo(() => { if (!configuredModel.ready()) return + const item = firstModel( () => scope()?.model, () => { const session = durable() + if (session?.agent && session.agent !== agent.current()?.name) return const model = session?.model + return model && { providerID: model.providerID, modelID: model.id } }, () => agent.current()?.model, fallback, ) + if (!item) return + return models.find(item) }) const configured = () => { const item = agent.current() const model = current() + if (!item || !model) return const global = configuredModel() + return ( getConfiguredAgentVariant({ agent: { model: item.model, variant: item.variant }, @@ -267,15 +309,19 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ const selected = () => { const draft = scope() + if (draft?.model && validModel(draft.model)) return draft.variant const session = durable() + if (session?.agent && session.agent !== agent.current()?.name) return const value = session?.model + if (value && validModel({ providerID: value.providerID, modelID: value.id })) return value.variant ?? null } const snapshot = () => { const selected = current() + return { agent: agent.current()?.name, model: selected ? { providerID: selected.provider.id, modelID: selected.id } : undefined, @@ -291,15 +337,19 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ } satisfies State const session = id() + if (session) { setSaved("session", session, state) + return } + setStore("draft", state) } const recent = createMemo(() => models.recent.list().filter(validModel).map(models.find).filter(Boolean)) const pending = new Map() + const sameSelection = (a: State, b: State) => a.agent === b.agent && a.model?.providerID === b.model?.providerID && @@ -309,7 +359,9 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ const reconcile = (sessionID: string) => { const expected = pending.get(sessionID) const session = data.session.get(sessionID) + if (!expected || !session?.model) return + if ( !sameSelection(expected, { agent: session.agent, @@ -320,9 +372,11 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ return pending.delete(sessionID) const draft = saved.session[sessionID] + if (id() !== sessionID || !draft || !sameSelection(draft, expected)) return setSaved("session", sessionID, { agent: undefined, model: undefined, variant: undefined }) } + onCleanup(serverSDK.event.on("session.model.selected", (event) => reconcile(event.data.sessionID))) onCleanup(serverSDK.event.on("session.agent.selected", (event) => reconcile(event.data.sessionID))) onCleanup( @@ -340,6 +394,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ trackSessionCommit(sessionID: string, selection: { agent: string; model: ModelKey; variant?: string }) { pending.set(sessionID, selection) reconcile(sessionID) + return () => { if (pending.get(sessionID) === selection) pending.delete(sessionID) } @@ -347,14 +402,18 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ cycle(direction: 1 | -1) { const items = recent() const item = current() + if (!item) return const index = items.findIndex((entry) => entry?.provider.id === item.provider.id && entry?.id === item.id) let next = index === -1 ? (direction === 1 ? 0 : items.length - 1) : index + direction + if (next < 0) next = items.length - 1 + if (next >= items.length) next = 0 const entry = items[next] + if (!entry) return model.set({ providerID: entry.provider.id, modelID: entry.id }) }, @@ -364,10 +423,13 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ const previous = current() const same = item && previous?.provider.id === item.providerID && previous.id === item.modelID write({ model: item, variant: same ? (model.variant.current() ?? null) : undefined }) + if (!item) return + // A session draft owns its variant even when preferences change in another session. if (id() && !same) write({ variant: model.variant.current() ?? null }) models.setVisibility(item, true) + if (!options?.recent) return models.recent.push(item) }) @@ -383,6 +445,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ selected, current() { const model = current() + return resolveModelVariant({ variants: this.list(), selected: this.selected(), @@ -392,12 +455,15 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ }, list() { const item = current() + if (!item?.variants) return [] + return Object.keys(item.variants) }, set(value: string | undefined) { batch(() => { const model = current() + if (!model) return write({ model: { providerID: model.provider.id, modelID: model.id }, variant: value ?? null }) models.variant.set({ providerID: model.provider.id, modelID: model.id }, value) @@ -405,6 +471,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ }, cycle() { const items = this.list() + if (items.length === 0) return this.set( cycleModelVariant({ @@ -428,11 +495,13 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ }, promote(dir: string, session: string, state?: State) { const next = clone(state ?? snapshot()) + if (!next) return // Creation already owns the active selection; keep only agent memory once it is in the read model. // Otherwise a first-message command's configured overrides would stay hidden behind this handoff. const created = data.session.get(session) const selection = created?.model + const committed = selection && sameSelection(next, { @@ -442,6 +511,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ }) ? { choices: next.choices } : next + const key = handoffKey(serverSDK.scope, dir, session) handoff.set(key, committed) @@ -454,10 +524,15 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ }, restore(msg: { sessionID: string; agent: string; model: ModelKey }) { const session = id() + if (!session) return + if (msg.sessionID !== session) return + if (durable()?.model) return + if (saved.session[session] !== undefined) return + if (handoff.has(handoffKey(serverSDK.scope, sdk().directory, session))) return setSaved("session", session, { @@ -468,6 +543,7 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({ }, }, } + return result }, }) diff --git a/packages/app/src/runtime/persistence/schema.test.ts b/packages/app/src/runtime/persistence/schema.test.ts index 99fa06786a56..f492bd763e96 100644 --- a/packages/app/src/runtime/persistence/schema.test.ts +++ b/packages/app/src/runtime/persistence/schema.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test" -import { Effect, Schema, SchemaGetter } from "effect" +import { Effect, Predicate, Schema, SchemaGetter } from "effect" import { Persistence } from "./schema" describe("persistence schemas", () => { @@ -12,6 +12,7 @@ describe("persistence schemas", () => { }), { enabled: true, appearance: { width: 240, font: "default" }, variant: "high" }, ) + const decode = Schema.decodeUnknownSync(schema) expect(decode({ appearance: { width: 300 } })).toEqual({ enabled: true, @@ -32,6 +33,7 @@ describe("persistence schemas", () => { test("legacy migration observes missing fields before initial defaults are applied", () => { const current = Persistence.struct({ mode: Schema.Literals(["compact", "full"]), enabled: Schema.Boolean }) + const stored = Schema.Struct({ mode: Schema.optional(Schema.Unknown), expanded: Schema.optional(Schema.Boolean), @@ -45,6 +47,7 @@ describe("persistence schemas", () => { encode: SchemaGetter.passthrough(), }), ) + const schema = Persistence.withInitial(Persistence.migrate(current, stored), { mode: "compact", enabled: true }) const decode = Schema.decodeUnknownSync(schema) expect(decode({ expanded: true, enabled: false })).toEqual({ mode: "full", enabled: false }) @@ -58,6 +61,7 @@ describe("persistence schemas", () => { amount: Schema.NumberFromString.check(Schema.isFinite()), items: Schema.mutable(Schema.Array(Schema.String)), }) + const schema = Persistence.withInitial(current, { amount: 7, items: ["initial"] }) const decode = Schema.decodeUnknownSync(schema) expect(decode({ amount: "12", items: [] })).toEqual({ amount: 12, items: [] }) @@ -69,6 +73,7 @@ describe("persistence schemas", () => { const schema = Persistence.struct({ enabled: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), }) + const decode = Schema.decodeUnknownSync(schema) expect(decode({})).toEqual({ enabled: true }) expect(decode({ enabled: undefined })).toEqual({ enabled: true }) @@ -84,6 +89,7 @@ describe("persistence schemas", () => { label: Persistence.fallback(Schema.String, () => "default"), amount: Persistence.fallback(Schema.NumberFromString.check(Schema.isFinite()), () => 7), }) + const decode = Schema.decodeUnknownSync(schema) expect(decode({})).toEqual({ enabled: true, label: "default", amount: 7 }) expect(decode({ enabled: "false", label: "saved", amount: "invalid" })).toEqual({ @@ -142,14 +148,16 @@ describe("persistence schemas", () => { test("recovers and migrates individual array entries", () => { const current = Schema.Struct({ name: Schema.String }) + const schema = Persistence.array( Schema.Union([current, Schema.String]).pipe( Schema.decodeTo(current, { - decode: SchemaGetter.transform((value) => (typeof value === "string" ? { name: value } : value)), + decode: SchemaGetter.transform((value) => (Predicate.isString(value) ? { name: value } : value)), encode: SchemaGetter.passthrough(), }), ), ) + const decode = Schema.decodeUnknownSync(schema) const value = decode(["old", { name: "new" }, null, { name: false }]) expect(value).toEqual([{ name: "old" }, { name: "new" }]) diff --git a/packages/app/src/runtime/persistence/schema.ts b/packages/app/src/runtime/persistence/schema.ts index bab04ef3940b..0818b8609b2c 100644 --- a/packages/app/src/runtime/persistence/schema.ts +++ b/packages/app/src/runtime/persistence/schema.ts @@ -23,10 +23,13 @@ export function withInitial>( initial: NoInfer, ) { const schema = isMigrated(definition) ? definition.current : definition + const read = isMigrated(definition) ? SchemaParser.decodeUnknownResult(definition.read, { onExcessProperty: "preserve" }) : Result.succeed + const encode = Schema.encodeUnknownSync(schema) + return Schema.Unknown.pipe( Schema.decode({ decode: SchemaGetter.transformOrFail((value) => @@ -39,25 +42,37 @@ export function withInitial>( } // Object-level codecs own their recovery. Plain structs can recover fields independently. +// SAFETY: stored values and their defaults are untyped trees walked against `ast`; `withInitial` decodes the +// recovered and merged result with the current schema before anyone reads it. +// oxlint-disable-next-line anti-slop/no-unknown-parameters, anti-slop/no-unknown-returns -- see SAFETY above function recover(ast: SchemaAST.AST, value: unknown, initial: unknown): unknown { if (value === undefined) return initial - if (ast._tag === "Objects" && !ast.encoding && ast.indexSignatures.length === 0 && Predicate.isObject(value)) { + + if (SchemaAST.isObjects(ast) && !ast.encoding && ast.indexSignatures.length === 0 && Predicate.isObject(value)) { return Object.fromEntries( ast.propertySignatures.flatMap((field) => { const defaults = Predicate.isObject(initial) ? initial[field.name] : undefined const next = recover(field.type, value[field.name], defaults) + if (next === undefined && !Object.hasOwn(value, field.name) && defaults === undefined) return [] + return [[field.name, next]] }), ) } + const decoded = Schema.decodeUnknownOption(Schema.make>(ast))(value) + return Option.isSome(decoded) ? decoded.value : initial } +// SAFETY: like `recover`, merges untyped trees that `withInitial` decodes with the current schema afterwards. +// oxlint-disable-next-line anti-slop/no-unknown-parameters, anti-slop/no-unknown-returns -- see SAFETY above function merge(initial: unknown, value: unknown): unknown { if (value === undefined) return initial + if (!Predicate.isObject(initial) || !Predicate.isObject(value)) return value + return Object.fromEntries( [...new Set([...Object.keys(initial), ...Object.keys(value)])].map((key) => [key, merge(initial[key], value[key])]), ) @@ -66,11 +81,13 @@ function merge(initial: unknown, value: unknown): unknown { // Unlike a decoding default, a fallback also replaces invalid persisted values. export function fallback>(schema: S, value: () => S["Type"]) { const defaulted = Schema.withDecodingDefaultType(Effect.sync(value))(schema) + return Schema.catchDecoding(() => Effect.sync(() => Option.some(value())))(defaulted) } export function optional>(schema: S) { const field = Schema.optional(schema) + return Schema.catchDecoding(() => Effect.succeed(Option.none()))(field) } @@ -80,6 +97,7 @@ export function struct(fields: Fields export function record>(schema: S) { const entries = Schema.Record(Schema.String, Schema.mutableKey(schema)) + return fallback(entries, () => Schema.decodeUnknownSync(entries)({})) } @@ -87,6 +105,7 @@ export function record>(schem export function array>(schema: S) { const decode = Schema.decodeUnknownOption(schema) const encode = Schema.encodeSync(schema) + return fallback( Schema.Array(Schema.Unknown).pipe( Schema.decodeTo(Schema.mutable(Schema.Array(Schema.toType(schema))), { diff --git a/packages/app/src/runtime/server/persistence.ts b/packages/app/src/runtime/server/persistence.ts index fefc54aa0e78..caffeebdd96d 100644 --- a/packages/app/src/runtime/server/persistence.ts +++ b/packages/app/src/runtime/server/persistence.ts @@ -1,4 +1,4 @@ -import { Effect, Option, Schema, SchemaGetter } from "effect" +import { Effect, Option, Predicate, Schema, SchemaGetter } from "effect" import { Persistence } from "@/runtime/persistence/schema" export const ServerKey = Schema.String.pipe(Schema.brand("ServerConnection.Key")) @@ -19,8 +19,10 @@ export const ServerHttp = Persistence.struct({ const StoredServer = Schema.Union([ServerHttp, ServerHttpBase, Schema.String]).pipe( Schema.decodeTo(ServerHttp, { decode: SchemaGetter.transform((value) => { - if (typeof value === "string") return { type: "http", http: { url: value } } + if (Predicate.isString(value)) return { type: "http", http: { url: value } } + if ("http" in value) return value + return { type: "http", http: value } }), encode: SchemaGetter.transform((value) => value), @@ -33,7 +35,9 @@ const ProjectList = Persistence.array( expanded: Persistence.fallback(Schema.Boolean, () => true), }), ) + const Projects = Persistence.record(ProjectList) + const LastProject = Persistence.record(Schema.String.pipe(Schema.catchDecoding(() => Effect.succeed(Option.none())))) const State = Persistence.struct({ @@ -57,12 +61,15 @@ export function serverState(canonicalLocalServer: () => string | undefined = () Schema.decode({ decode: SchemaGetter.transform((value) => { const canonical = canonicalLocalServer() + if (!canonical || canonical === "local") return value const previous = value.projects[canonical] const last = value.lastProject[canonical] + if (!previous && last === undefined) return value const projects = { ...value.projects } + if (previous) { const local = projects.local ?? [] const worktrees = new Set(local.map((project) => project.worktree)) @@ -71,16 +78,20 @@ export function serverState(canonicalLocalServer: () => string | undefined = () ...previous.filter((project) => { if (worktrees.has(project.worktree)) return false worktrees.add(project.worktree) + return true }), ] delete projects[canonical] } + const lastProject = { ...value.lastProject } + if (last !== undefined) { lastProject.local ??= last delete lastProject[canonical] } + return { ...value, projects, lastProject } }), encode: SchemaGetter.transform((value) => value), diff --git a/packages/desktop/src/main/files/attachment-picker.ts b/packages/desktop/src/main/files/attachment-picker.ts index 11a460dce8dd..1419a55d4d03 100644 --- a/packages/desktop/src/main/files/attachment-picker.ts +++ b/packages/desktop/src/main/files/attachment-picker.ts @@ -13,15 +13,19 @@ export function createPickedFileAuthorizations( add(sender: number, paths: string[]) { const token = randomUUID() selections.set(token, { sender, paths: new Set(paths), remaining: MAX_ATTACHMENT_BYTES }) + return token }, read: Effect.fn("DesktopFiles.readPickedFile")(function* (sender: number, token: string, path: string) { const selection = selections.get(token) + if (selection?.sender !== sender || !selection.paths.delete(path)) throw new Error(nativeT("desktop.picker.error.notSelected")) const bytes = yield* read(path, selection.remaining) selection.remaining -= bytes.byteLength + if (selection.paths.size === 0) selections.delete(token) + return bytes }), release(sender: number, token: string) { @@ -32,6 +36,7 @@ export function createPickedFileAuthorizations( export function assertAttachmentBudget(files: { size: number }[]) { const total = files.reduce((sum, file) => sum + file.size, 0) + if (total <= MAX_ATTACHMENT_BYTES) return throw new Error(nativeT("desktop.picker.error.sizeLimit", { limit: MAX_ATTACHMENT_BYTES / 1024 / 1024 })) } @@ -42,16 +47,20 @@ export function readAttachment(filePath: string, maxBytes = MAX_ATTACHMENT_BYTES const fs = yield* FileSystem.FileSystem const file = yield* fs.open(filePath, { flag: "r" }) const info = yield* file.stat + if (info.size > FileSystem.Size(maxBytes)) throw new Error(nativeT("desktop.picker.error.sizeLimit", { limit: MAX_ATTACHMENT_BYTES / 1024 / 1024 })) const bytes = new Uint8Array(Number(info.size)) let offset = 0 + while (offset < bytes.byteLength) { const read = Number(yield* file.read(bytes.subarray(offset))) + if (read === 0) break offset += read } + return bytes.buffer.slice(0, offset) }), ) diff --git a/packages/desktop/src/main/ipc-transport.test.ts b/packages/desktop/src/main/ipc-transport.test.ts index f48eca43b70d..754690389986 100644 --- a/packages/desktop/src/main/ipc-transport.test.ts +++ b/packages/desktop/src/main/ipc-transport.test.ts @@ -2,7 +2,7 @@ import { describe, expect, test } from "bun:test" import { EventEmitter } from "node:events" import { MessageChannel } from "node:worker_threads" import type { MessagePortMain, WebContents } from "electron" -import { Effect, Layer, ManagedRuntime, Schema, Stream } from "effect" +import { Effect, Layer, ManagedRuntime, Predicate, Schema, Stream } from "effect" import { Rpc, RpcGroup, RpcMessage, RpcServer } from "effect/unstable/rpc" import { Transferable } from "effect/unstable/workers" import { omitUndefined } from "../shared/ipc-transport" @@ -13,13 +13,16 @@ describe("desktop RPC transport", () => { test("decodes renderer payloads whose optional fields are undefined", async () => { let received: unknown const rpcs = RpcGroup.make(FilesOpenFilePicker) + const handlers = rpcs.toLayer({ FilesOpenFilePicker: ({ options }) => Effect.sync(() => { received = options + return null }), }) + const live = RpcServer.layer(rpcs).pipe(Layer.provide(handlers), Layer.provideMerge(IpcServerProtocolLive)) const runtime = ManagedRuntime.make(live) const handoff = await runtime.runPromise(IpcPortHandoff) @@ -30,13 +33,10 @@ describe("desktop RPC transport", () => { // Structured clone keeps a present-but-undefined key, so it reaches the JSON codec. const rejected = await call(channel.port2, 0, "FilesOpenFilePicker", payload) - expect(rejected.exit).toMatchObject({ - _tag: "Failure", - cause: [{ _tag: "Die", defect: expect.stringContaining('["options"]["title"]') }], - }) + expect(rejected.exit).toMatchObject(died(expect.stringContaining('["options"]["title"]'))) const accepted = await call(channel.port2, 1, "FilesOpenFilePicker", omitUndefined(payload)) - expect(accepted.exit).toEqual({ _tag: "Success", value: null }) + expect(accepted.exit).toEqual(success(null)) expect(received).toEqual({ multiple: true, defaultPath: "C:\\project" }) channel.port2.close() @@ -56,13 +56,16 @@ describe("desktop RPC transport", () => { test("keeps multiple renderer ports independent", async () => { let received: unknown + const handlers = TestRpcs.toLayer( Effect.gen(function* () { const handoff = yield* IpcPortHandoff + return TestRpcs.of({ "test.focused": (_request, context) => Effect.succeed(handoff.sender(context.client.id)?.id === 1), "test.blob.put": ({ data }) => { received = data + return Effect.succeed([...data].join(",")) }, "test.blob.get": () => Effect.succeed(new Uint8Array([3, 1, 4])), @@ -70,6 +73,7 @@ describe("desktop RPC transport", () => { }) }), ) + const live = RpcServer.layer(TestRpcs).pipe(Layer.provide(handlers), Layer.provideMerge(IpcServerProtocolLive)) const runtime = ManagedRuntime.make(live) const handoff = await runtime.runPromise(IpcPortHandoff) @@ -82,34 +86,36 @@ describe("desktop RPC transport", () => { call(first.port2, 0, "test.focused", null), call(second.port2, 0, "test.focused", null), ]) - expect(focused.exit).toEqual({ _tag: "Success", value: true }) - expect(unfocused.exit).toEqual({ _tag: "Success", value: false }) + + expect(focused.exit).toEqual(success(true)) + expect(unfocused.exit).toEqual(success(false)) const put = await call(first.port2, 1, "test.blob.put", omitUndefined({ data: new Uint8Array([2, 7, 1]) })) - expect(put.exit).toEqual({ _tag: "Success", value: "2,7,1" }) + expect(put.exit).toEqual(success("2,7,1")) // Binary payloads arrive as bytes, not as base64 text or a plain object. expect(received).toBeInstanceOf(Uint8Array) - expect((await call(first.port2, 2, "test.blob.get", null)).exit).toEqual({ - _tag: "Success", - value: new Uint8Array([3, 1, 4]), - }) + expect((await call(first.port2, 2, "test.blob.get", null)).exit).toEqual(success(new Uint8Array([3, 1, 4]))) expect((await call(first.port2, 3, "test.events", null)).chunks).toEqual([ - { _tag: "TestEvent", value: "session.new" }, + Schema.encodeSync(TestEvent)(new TestEvent({ value: "session.new" })), ]) const reloaded = new MessageChannel() handoff.bind(sender(1), serverPort(reloaded.port1)) + const [reloadedFocused, stillUnfocused] = await Promise.all([ call(reloaded.port2, 0, "test.focused", null), call(second.port2, 1, "test.focused", null), ]) - expect(reloadedFocused.exit).toEqual({ _tag: "Success", value: true }) - expect(stillUnfocused.exit).toEqual({ _tag: "Success", value: false }) + + expect(reloadedFocused.exit).toEqual(success(true)) + expect(stillUnfocused.exit).toEqual(success(false)) + for (const port of [first.port2, second.port2, reloaded.port2]) port.close() await runtime.dispose() }) }) class TestEvent extends Schema.TaggedClass()("TestEvent", { value: Schema.String }) {} + const TestRpcs = RpcGroup.make( Rpc.make("test.focused", { success: Schema.Boolean }), Rpc.make("test.blob.put", { payload: { data: Transferable.Uint8Array }, success: Schema.String }), @@ -117,28 +123,47 @@ const TestRpcs = RpcGroup.make( Rpc.make("test.events", { success: TestEvent, stream: true }), ) +// SAFETY: these are the RPC wire format's plain encoded messages and exits, which the tests post and expect as is. +/* oxlint-disable anti-slop-effect/no-manual-tagged-construction -- see SAFETY above */ +function success(value: A) { + return { _tag: "Success", value } as const +} + +function died(defect: A) { + return { _tag: "Failure", cause: [{ _tag: "Die", defect }] } as const +} + // Speaks the wire format the way src/renderer/ipc-client.ts does: post a request, ack each chunk, // and settle on the exit. The payload is posted as given so a test can send what omitUndefined drops. -function call(port: MessageChannel["port2"], id: number, tag: string, payload: unknown) { +function call(port: MessageChannel["port2"], id: number, tag: string, payload: RpcMessage.RequestEncoded["payload"]) { const chunks: unknown[] = [] + return new Promise<{ chunks: unknown[]; exit: RpcMessage.ResponseExitEncoded["exit"] }>((resolve) => { const onMessage = (message: RpcMessage.FromServerEncoded) => { if (!("requestId" in message) || Number(message.requestId) !== id) return - if (message._tag === "Chunk") { + + if (Predicate.isTagged(message, "Chunk")) { chunks.push(...message.values) port.postMessage({ _tag: "Ack", requestId: message.requestId } satisfies RpcMessage.AckEncoded) + return } + port.off("message", onMessage) resolve({ chunks, exit: message.exit }) } + port.on("message", onMessage) port.postMessage({ _tag: "Request", id, tag, payload, headers: [] }) }) } +/* oxlint-enable anti-slop-effect/no-manual-tagged-construction */ function sender(id: number) { const events = new EventEmitter() + + // SAFETY: the transport reads only a sender's `id` and `isDestroyed`, and its `destroyed` event. + // oxlint-disable-next-line anti-slop/no-chained-type-assertions -- see SAFETY above return { id, isDestroyed: () => false, @@ -148,27 +173,38 @@ function sender(id: number) { } function serverPort(port: MessageChannel["port1"]) { - const listeners = new Map<(event: Electron.MessageEvent) => void, (data: unknown) => void>() - return { + const listeners = new Map<(event: Electron.MessageEvent) => void, (data: Electron.MessageEvent["data"]) => void>() + + const fake = { on(event: string, listener: (event: Electron.MessageEvent) => void) { if (event !== "message") { port.on(event, listener) + return } - const wrapped = (data: unknown) => listener({ data } as Electron.MessageEvent) + + // SAFETY: the transport reads only the `data` of a message event. + const wrapped = (data: Electron.MessageEvent["data"]) => listener({ data } as Electron.MessageEvent) listeners.set(listener, wrapped) port.on("message", wrapped) }, off(event: string, listener: (event: Electron.MessageEvent) => void) { if (event !== "message") { port.off(event, listener) + return } + const wrapped = listeners.get(listener) + if (wrapped) port.off("message", wrapped) }, postMessage: port.postMessage.bind(port), start: port.start.bind(port), close: port.close.bind(port), - } as unknown as MessagePortMain + } + + // SAFETY: the transport uses only these methods of a port. + // oxlint-disable-next-line anti-slop/no-chained-type-assertions -- see SAFETY above + return fake as unknown as MessagePortMain } diff --git a/packages/desktop/src/main/ipc-transport.ts b/packages/desktop/src/main/ipc-transport.ts index 91e6e84acce8..b9e3bf3864c1 100644 --- a/packages/desktop/src/main/ipc-transport.ts +++ b/packages/desktop/src/main/ipc-transport.ts @@ -39,8 +39,10 @@ export const IpcServerProtocolLive = Layer.unwrap( const disconnect = Effect.fnUntraced(function* (id: number) { const binding = bindings.get(id) + if (!binding) return bindings.delete(id) + if (senderBindings.get(binding.sender.id) === id) senderBindings.delete(binding.sender.id) binding.port.off("message", binding.onMessage) binding.port.off("close", binding.onClose) @@ -52,16 +54,23 @@ export const IpcServerProtocolLive = Layer.unwrap( const bind = Effect.fnUntraced(function* (sender: WebContents, port: MessagePortMain) { const previous = senderBindings.get(sender.id) + if (previous !== undefined) yield* disconnect(previous) + if (sender.isDestroyed()) { port.close() + return } const id = nextClientId++ + const onMessage = (event: Electron.MessageEvent) => { + // SAFETY: the other end of this port is the renderer's ipc-client, which posts only RPC client + // messages; the RPC server decodes each request's payload with that RPC's schema. Queue.offerUnsafe(inbound, [id, event.data as RpcMessage.FromClientEncoded] as const) } + const onClose = () => runFork(disconnect(id)) const unbindEvents = yield* bindIpcEvents(sender.id) const binding = { id, sender, port, onMessage, onClose, unbindEvents } diff --git a/packages/desktop/src/main/ipc.ts b/packages/desktop/src/main/ipc.ts index 01308f3ceaf3..e09f6cdf0d15 100644 --- a/packages/desktop/src/main/ipc.ts +++ b/packages/desktop/src/main/ipc.ts @@ -22,6 +22,7 @@ import { DesktopCli } from "./service/desktop-cli" import { getLastFocusedWindow } from "./windows" const services = Layer.mergeAll(DesktopFiles.layer, Extensions.layer) + const handlers = Layer.mergeAll( appHandlers, storageHandlers, @@ -31,6 +32,7 @@ const handlers = Layer.mergeAll( eventHandlers, extensionHandlers, ) + export const layer = RpcServer.layer(DesktopRpcs, { disableFatalDefects: true }).pipe( Layer.provide(handlers), Layer.provideMerge(IpcServerProtocolLive), @@ -42,9 +44,11 @@ export const registerIpcHandlers = Effect.gen(function* () { const lifecycle = yield* ApplicationLifecycle.Service const desktopCli = yield* DesktopCli.Service const runFork = Effect.runForkWith(yield* Effect.context()) + const menu = { trigger: (id: string) => { const win = getLastFocusedWindow() + if (win) sendMenuCommand(win, id) }, installCli: () => runFork(showCliInstaller(desktopCli)), @@ -52,26 +56,34 @@ export const registerIpcHandlers = Effect.gen(function* () { openExternal: (url: string) => runFork(openExternalURL(url)), relaunch: lifecycle.relaunch, } - const wire = (_event: Electron.Event, win: BrowserWindow) => { + + const wire = (win: BrowserWindow) => { win.webContents.on("before-input-event", (_event, input) => { if (input.type !== "keyDown" || input.key !== "Escape") return win.webContents.send(DragCancelEvent) }) + const post = () => { if (win.isDestroyed() || win.webContents.isDestroyed()) return const channel = new MessageChannelMain() handoff.bind(win.webContents, channel.port1) win.webContents.postMessage(IpcTransportPort, null, [channel.port2]) } + win.webContents.on("did-finish-load", post) + // The first window starts loading before the layers exist and may already be done. if (!win.webContents.isLoading() && win.webContents.getURL()) post() } + + const onWindowCreated = (_event: Electron.Event, win: BrowserWindow) => wire(win) + yield* Effect.sync(() => { - app.on("browser-window-created", wire) - BrowserWindow.getAllWindows().forEach((win) => wire({} as Electron.Event, win)) + app.on("browser-window-created", onWindowCreated) + BrowserWindow.getAllWindows().forEach((win) => wire(win)) }) - yield* Effect.addFinalizer(() => Effect.sync(() => app.off("browser-window-created", wire))) + yield* Effect.addFinalizer(() => Effect.sync(() => app.off("browser-window-created", onWindowCreated))) + return { installMenu: () => createMenu(menu), } diff --git a/packages/desktop/src/renderer/ipc-client.ts b/packages/desktop/src/renderer/ipc-client.ts index 904b0592bc17..d6cb754b0963 100644 --- a/packages/desktop/src/renderer/ipc-client.ts +++ b/packages/desktop/src/renderer/ipc-client.ts @@ -11,33 +11,47 @@ import { IpcTransportPort, omitUndefined } from "../shared/ipc-transport" // out of the renderer's initial module graph is worth about a third of its startup script. type EventTag = DesktopEvent["_tag"] + type InvokeTag = Exclude + type InvokeArgs = Parameters + type InvokeResult = ReturnType extends Effect.Effect ? Value : never + type EventValue = Extract type Pending = { + // SAFETY: the reply is the RPC's success value as the trusted main process encoded it; see the note above. + // oxlint-disable-next-line anti-slop/no-unknown-parameters -- see SAFETY above readonly resolve: (value: unknown) => void - readonly reject: (error: unknown) => void + readonly reject: (cause: unknown) => void readonly chunk?: (values: ReadonlyArray) => void } const pending = new Map() -const listeners = new Map void>>() + +const listeners = new Map void>>() + +// SAFETY: pagehide only waits for these callbacks to settle and ignores what they resolve to. +// oxlint-disable-next-line anti-slop/no-unknown-returns -- see SAFETY above const beforeDispose = new Set<() => Promise | void>() + let nextId = 0 const port = new Promise((resolve) => { const onMessage = (event: MessageEvent) => { if (event.source !== window || event.data !== IpcTransportPort) return const value = event.ports[0] + if (!value) return window.removeEventListener("message", onMessage) + // SAFETY: the port comes from the trusted main process, whose RPC server posts only its own messages. value.addEventListener("message", (message) => receive(value, message.data as RpcMessage.FromServerEncoded)) value.start() resolve(value) } + window.addEventListener("message", onMessage) }) @@ -49,15 +63,20 @@ window.addEventListener( ) void request("DesktopEvents", null, (values) => { + // SAFETY: DesktopEvents streams desktop events, whose encoded fields are JSON-native and read as decoded. for (const value of values as ReadonlyArray) listeners.get(value._tag)?.forEach((fn) => fn(value)) }) +// SAFETY: as for `beforeDispose`, the callback's result is only awaited, never read. +// oxlint-disable-next-line anti-slop/no-unknown-returns -- see SAFETY above export function onBeforeDispose(callback: () => Promise | void) { beforeDispose.add(callback) + return () => beforeDispose.delete(callback) } export function invoke(tag: Tag, ...payload: InvokeArgs): Promise> { + // SAFETY: main replies with the success value of `tag`'s RPC; its encoding is JSON-native or bytes, as decoded. return request(tag, payload[0] ?? null) as Promise> } @@ -71,37 +90,55 @@ export function cancellable( payload: InvokeArgs[0], signal: AbortSignal | undefined, ): Promise> { + // SAFETY: as in `invoke`, the reply is the success value of `tag`'s RPC. return request(tag, payload ?? null, undefined, signal) as Promise> } export function listen(tag: Tag, listener: (value: EventValue) => void) { - const callback = listener as (value: unknown) => void + // SAFETY: events dispatch by tag, so this listener only receives `Tag` events. + const callback = listener as (value: DesktopEvent) => void const callbacks = listeners.get(tag) ?? new Set() callbacks.add(callback) listeners.set(tag, callbacks) + return () => { callbacks.delete(callback) + if (callbacks.size === 0) listeners.delete(tag) } } -function request(tag: string, payload: unknown, chunk?: Pending["chunk"], signal?: AbortSignal) { +// SAFETY: this side speaks the RPC wire format by hand, whose messages are plain tagged objects. Effect's Match, +// Predicate, and message constructors would load the Effect runtime into the renderer; see the note above. +/* oxlint-disable anti-slop-effect/no-manual-tag-comparison, anti-slop-effect/no-manual-tagged-construction -- see SAFETY above */ +function request( + tag: InvokeTag | "DesktopEvents", + payload: InvokeArgs[0] | null, + chunk?: Pending["chunk"], + signal?: AbortSignal, +) { const id = nextId++ + return new Promise((resolve, reject) => { if (signal?.aborted) return reject(signal.reason) + const abort = () => { if (!pending.delete(id)) return reject(signal?.reason) void port.then((p) => p.postMessage({ _tag: "Interrupt", requestId: id } satisfies RpcMessage.InterruptEncoded)) } + signal?.addEventListener("abort", abort, { once: true }) + const settle = (callback: (value: Value) => void) => (value: Value) => { signal?.removeEventListener("abort", abort) callback(value) } + pending.set(id, { resolve: settle(resolve), reject: settle(reject), chunk }) + const message: RpcMessage.RequestEncoded = { _tag: "Request", id, @@ -109,6 +146,7 @@ function request(tag: string, payload: unknown, chunk?: Pending["chunk"], signal payload: omitUndefined(payload), headers: [], } + void port.then((p) => p.postMessage(message)) }) } @@ -118,24 +156,33 @@ function receive(p: MessagePort, message: RpcMessage.FromServerEncoded) { case "Chunk": { pending.get(Number(message.requestId))?.chunk?.(message.values) p.postMessage({ _tag: "Ack", requestId: message.requestId } satisfies RpcMessage.AckEncoded) + return } + case "Exit": { const id = Number(message.requestId) const entry = pending.get(id) pending.delete(id) + if (!entry) return + if (message.exit._tag === "Success") return entry.resolve(message.exit.value) + return entry.reject(failure(message.exit.cause)) } + case "Defect": { const error = new Error("Desktop IPC defect", { cause: message.defect }) pending.forEach((entry) => entry.reject(error)) pending.clear() + return } + case "ClientProtocolError": { console.error("[desktop-ipc] protocol error", message.error) + return } } @@ -145,8 +192,12 @@ function receive(p: MessagePort, message: RpcMessage.FromServerEncoded) { // and interrupts surface as errors. function failure(cause: ReadonlyArray<{ readonly _tag: string; readonly error?: unknown; readonly defect?: unknown }>) { const failed = cause.find((item) => item._tag === "Fail") + if (failed) return failed.error const died = cause.find((item) => item._tag === "Die") + if (died) return new Error("Desktop IPC handler failed", { cause: died.defect }) + return new Error("Desktop IPC request interrupted") } +/* oxlint-enable anti-slop-effect/no-manual-tag-comparison, anti-slop-effect/no-manual-tagged-construction */ diff --git a/packages/desktop/src/shared/ipc-rpc.ts b/packages/desktop/src/shared/ipc-rpc.ts index 6cad9a24ff88..5475342e97bf 100644 --- a/packages/desktop/src/shared/ipc-rpc.ts +++ b/packages/desktop/src/shared/ipc-rpc.ts @@ -8,12 +8,19 @@ import { StorageRpcs } from "./ipc-rpc/storage" import { WindowRpcs } from "./ipc-rpc/window" export { AppRpcs } from "./ipc-rpc/app" + export { EventRpcs } from "./ipc-rpc/events" + export { ExtensionRpcs } from "./ipc-rpc/extensions" + export { FileRpcs } from "./ipc-rpc/files" + export { MenuRpcs } from "./ipc-rpc/menu" + export { StorageRpcs } from "./ipc-rpc/storage" + export { WindowRpcs } from "./ipc-rpc/window" export const DesktopRpcs = AppRpcs.merge(StorageRpcs, FileRpcs, WindowRpcs, MenuRpcs, EventRpcs, ExtensionRpcs) + export type DesktopRpcClient = RpcClient.FromGroup diff --git a/packages/desktop/src/shared/ipc-rpc/app.ts b/packages/desktop/src/shared/ipc-rpc/app.ts index f964e8dc1982..30496fb13a4b 100644 --- a/packages/desktop/src/shared/ipc-rpc/app.ts +++ b/packages/desktop/src/shared/ipc-rpc/app.ts @@ -6,38 +6,50 @@ const ServerReadyData = Schema.Struct({ }) export const AppAwaitInitialization = Rpc.make("AppAwaitInitialization", { success: ServerReadyData }) + export const AppReconnectService = Rpc.make("AppReconnectService", { success: ServerReadyData }) + export const AppConsumeInitialDeepLinks = Rpc.make("AppConsumeInitialDeepLinks", { success: Schema.Array(Schema.String), }) + export const AppGetDefaultServerUrl = Rpc.make("AppGetDefaultServerUrl", { success: Schema.NullOr(Schema.String), }) + export const AppSetDefaultServerUrl = Rpc.make("AppSetDefaultServerUrl", { payload: { url: Schema.NullOr(Schema.String) }, }) + export const AppIsFirstLaunchOnboardingPending = Rpc.make("AppIsFirstLaunchOnboardingPending", { success: Schema.Boolean, }) + export const AppFinishFirstLaunchOnboarding = Rpc.make("AppFinishFirstLaunchOnboarding", { payload: { createDefaultProject: Schema.Boolean }, success: Schema.NullOr(Schema.String), }) + export const AppCheckAppExists = Rpc.make("AppCheckAppExists", { payload: { appName: Schema.String }, success: Schema.Boolean, }) + export const AppResolveAppPath = Rpc.make("AppResolveAppPath", { payload: { appName: Schema.String }, success: Schema.NullOr(Schema.String), }) + export const AppSetBackgroundColor = Rpc.make("AppSetBackgroundColor", { payload: { color: Schema.String }, }) + export const AppExportDebugLogs = Rpc.make("AppExportDebugLogs", { success: Schema.String }) + export const AppSetForceFocus = Rpc.make("AppSetForceFocus", { payload: { enabled: Schema.Boolean }, }) + export const AppRecordFatalRendererError = Rpc.make("AppRecordFatalRendererError", { payload: { error: Schema.Struct({ @@ -49,10 +61,13 @@ export const AppRecordFatalRendererError = Rpc.make("AppRecordFatalRendererError }), }, }) + export const AppSetNativeTranslations = Rpc.make("AppSetNativeTranslations", { payload: { value: Schema.Unknown }, }) + export const AppRelaunch = Rpc.make("AppRelaunch") + export const AppRpcs = RpcGroup.make( AppAwaitInitialization, AppReconnectService, diff --git a/packages/desktop/src/shared/ipc-rpc/files.ts b/packages/desktop/src/shared/ipc-rpc/files.ts index ca68ba5db5a0..51e826cff238 100644 --- a/packages/desktop/src/shared/ipc-rpc/files.ts +++ b/packages/desktop/src/shared/ipc-rpc/files.ts @@ -3,64 +3,80 @@ import { Rpc, RpcGroup } from "effect/unstable/rpc" import { Transferable } from "effect/unstable/workers" const OptionalString = Schema.optional(Schema.String) + const PickerOptions = Schema.Struct({ multiple: Schema.optional(Schema.Boolean), title: OptionalString, defaultPath: OptionalString, }) + const FilePickerOptions = Schema.Struct({ multiple: Schema.optional(Schema.Boolean), title: OptionalString, defaultPath: OptionalString, extensions: Schema.optional(Schema.Array(Schema.String)), }) + const SavePickerOptions = Schema.Struct({ title: OptionalString, defaultPath: OptionalString }) + const PickedFiles = Schema.Struct({ token: Schema.String, files: Schema.Array(Schema.Struct({ path: Schema.String, name: Schema.String, size: Schema.Number })), }) + const ClipboardImage = Schema.Struct({ buffer: Transferable.Uint8Array, width: Schema.Number, height: Schema.Number }) export const FilesOpenDirectoryPicker = Rpc.make("FilesOpenDirectoryPicker", { payload: { options: Schema.optional(PickerOptions) }, success: Schema.NullOr(Schema.Union([Schema.String, Schema.Array(Schema.String)])), }) + export const FilesOpenFilePicker = Rpc.make("FilesOpenFilePicker", { payload: { options: Schema.optional(FilePickerOptions) }, success: Schema.NullOr(PickedFiles), }) + export const FilesReadPickedFile = Rpc.make("FilesReadPickedFile", { payload: { token: Schema.String, path: Schema.String }, success: Transferable.Uint8Array, }) + export const FilesReleasePickedFiles = Rpc.make("FilesReleasePickedFiles", { payload: { token: Schema.String }, }) + export const FilesSaveFile = Rpc.make("FilesSaveFile", { payload: { options: SavePickerOptions, content: Schema.String }, success: Schema.Boolean, }) + export const FilesOpenExternal = Rpc.make("FilesOpenExternal", { payload: { url: Schema.String }, }) + export const FilesOpenBrowser = Rpc.make("FilesOpenBrowser", { payload: { url: Schema.String }, success: Schema.Boolean, }) + export const FilesOpenLocalFile = Rpc.make("FilesOpenLocalFile", { payload: { url: Schema.String }, }) + export const FilesOpenPath = Rpc.make("FilesOpenPath", { payload: { path: Schema.String, application: Schema.optional(Schema.String) }, success: Schema.NullOr(Schema.String), }) + export const FilesRevealPath = Rpc.make("FilesRevealPath", { payload: { path: Schema.String }, success: Schema.Boolean, }) + export const FilesReadClipboardImage = Rpc.make("FilesReadClipboardImage", { success: Schema.NullOr(ClipboardImage), }) + export const FilesWriteClipboardText = Rpc.make("FilesWriteClipboardText", { payload: { text: Schema.String }, }) diff --git a/packages/desktop/src/shared/ipc-rpc/menu.ts b/packages/desktop/src/shared/ipc-rpc/menu.ts index def7a5b308be..a83abbaef18d 100644 --- a/packages/desktop/src/shared/ipc-rpc/menu.ts +++ b/packages/desktop/src/shared/ipc-rpc/menu.ts @@ -26,4 +26,5 @@ const DesktopMenuAction = Schema.Literals([ export const MenuRunAction = Rpc.make("MenuRunAction", { payload: { action: DesktopMenuAction }, }) + export const MenuRpcs = RpcGroup.make(MenuRunAction) diff --git a/packages/desktop/src/shared/ipc-rpc/storage.ts b/packages/desktop/src/shared/ipc-rpc/storage.ts index 2b920b3986ba..b3438f8d93da 100644 --- a/packages/desktop/src/shared/ipc-rpc/storage.ts +++ b/packages/desktop/src/shared/ipc-rpc/storage.ts @@ -6,6 +6,7 @@ export const StorageItems = Rpc.make("StorageItems", { payload: { name: Schema.String }, success: Schema.Struct({ items: Schema.Record(Schema.String, Schema.String), revision: Schema.Number }), }) + export const StorageUpdate = Rpc.make("StorageUpdate", { payload: { name: Schema.String, @@ -14,20 +15,26 @@ export const StorageUpdate = Rpc.make("StorageUpdate", { }, success: Schema.Number, }) + export const StorageClear = Rpc.make("StorageClear", { payload: { name: Schema.String } }) + export const DraftsGet = Rpc.make("DraftsGet", { payload: { key: Schema.String }, success: Schema.NullOr(Schema.String), }) + export const DraftsSet = Rpc.make("DraftsSet", { payload: { key: Schema.String, value: Schema.String, strict: Schema.Boolean }, success: Schema.Array(Schema.String), }) + export const DraftsDelete = Rpc.make("DraftsDelete", { payload: { key: Schema.String } }) + export const DraftsPutBlob = Rpc.make("DraftsPutBlob", { payload: { data: Transferable.Uint8Array }, success: Schema.String, }) + export const DraftsGetBlob = Rpc.make("DraftsGetBlob", { payload: { id: Schema.String }, success: Schema.NullOr(Transferable.Uint8Array), diff --git a/packages/desktop/src/shared/ipc-rpc/window.ts b/packages/desktop/src/shared/ipc-rpc/window.ts index bb659b61af79..e8aca73d1a45 100644 --- a/packages/desktop/src/shared/ipc-rpc/window.ts +++ b/packages/desktop/src/shared/ipc-rpc/window.ts @@ -2,20 +2,29 @@ import { Schema } from "effect" import { Rpc, RpcGroup } from "effect/unstable/rpc" export const WindowThemeReady = Rpc.make("WindowThemeReady") + export const WindowGetFocused = Rpc.make("WindowGetFocused", { success: Schema.Boolean }) + export const WindowGetFullscreen = Rpc.make("WindowGetFullscreen", { success: Schema.Boolean }) + export const WindowSetFocus = Rpc.make("WindowSetFocus") + export const WindowShow = Rpc.make("WindowShow") + export const WindowGetZoomFactor = Rpc.make("WindowGetZoomFactor", { success: Schema.Number }) + export const WindowSetZoomFactor = Rpc.make("WindowSetZoomFactor", { payload: { factor: Schema.Number }, }) + export const WindowGetPinchZoomEnabled = Rpc.make("WindowGetPinchZoomEnabled", { success: Schema.Boolean, }) + export const WindowSetPinchZoomEnabled = Rpc.make("WindowSetPinchZoomEnabled", { payload: { enabled: Schema.Boolean }, }) + export const WindowSetTitlebar = Rpc.make("WindowSetTitlebar", { payload: { theme: Schema.Struct({ @@ -24,6 +33,7 @@ export const WindowSetTitlebar = Rpc.make("WindowSetTitlebar", { }), }, }) + export const WindowRpcs = RpcGroup.make( WindowThemeReady, WindowGetFocused, diff --git a/packages/gui-extensions/src/browser/network.ts b/packages/gui-extensions/src/browser/network.ts index 4969085d1dec..35a24835b5e3 100644 --- a/packages/gui-extensions/src/browser/network.ts +++ b/packages/gui-extensions/src/browser/network.ts @@ -14,6 +14,7 @@ export const createBrowserNetwork = Effect.fn("BrowserNetwork.create")(function* partition: string }) { const options = { location: input.location } + const proxy = yield* Effect.acquireRelease( Effect.tryPromise(() => BrowserProxy.make({ @@ -37,6 +38,7 @@ export const createBrowserNetwork = Effect.fn("BrowserNetwork.create")(function* ), (proxy) => Effect.promise(() => proxy.close()), ) + const partition = session.fromPartition(input.partition) yield* Effect.addFinalizer(() => Effect.promise(() => partition.closeAllConnections())) // This is the browser's private partition, not the app/API connection. Never @@ -45,6 +47,7 @@ export const createBrowserNetwork = Effect.fn("BrowserNetwork.create")(function* partition.setProxy({ mode: "fixed_servers", proxyRules: proxy.url, proxyBypassRules: "<-loopback>" }), ) yield* Effect.tryPromise(() => partition.closeAllConnections()) + return { attach(contents: WebContents) { const login = ( @@ -64,8 +67,10 @@ export const createBrowserNetwork = Effect.fn("BrowserNetwork.create")(function* event.preventDefault() callback(proxy.credentials.username, proxy.credentials.password) } + contents.on("login", login) contents.setWebRTCIPHandlingPolicy("disable_non_proxied_udp") + return () => contents.off("login", login) }, } diff --git a/packages/gui-extensions/src/ssh/askpass.test.ts b/packages/gui-extensions/src/ssh/askpass.test.ts index 57495b9f2997..312803385be3 100644 --- a/packages/gui-extensions/src/ssh/askpass.test.ts +++ b/packages/gui-extensions/src/ssh/askpass.test.ts @@ -25,6 +25,7 @@ const request = Effect.fn("test.askpass.request")(function* ( ], { concurrency: "unbounded" }, ) + return result.text }, Effect.scoped) @@ -32,11 +33,13 @@ it.live( "per-prompt replies are isolated, including confirmation and OTP", Effect.gen(function* () { const prompts = yield* Queue.unbounded<{ id: string; text: string; confirm: boolean }>() + const bridge = yield* createAskpass({ binary: "unused", prompt: (prompt) => Queue.offer(prompts, prompt).pipe(Effect.asVoid), clear: () => Effect.void, }) + const password = yield* request(bridge.env, "Password:").pipe(Effect.forkScoped) const first = yield* Queue.take(prompts) expect(first.text).toBe("Password:") @@ -56,11 +59,13 @@ it.live( const parent = yield* Scope.Scope const scope = yield* Scope.fork(parent) const prompted = yield* Deferred.make() + const bridge = yield* createAskpass({ binary: "unused", prompt: () => Deferred.succeed(prompted, undefined).pipe(Effect.asVoid), clear: () => Effect.void, }).pipe(Scope.provide(scope)) + expect(yield* request({ ...bridge.env, OPENCODE_SSH_ASKPASS_TOKEN: "incorrect" }, "Password:")).toBe("") const reply = yield* request(bridge.env, "Trust fingerprint?", true).pipe(Effect.forkScoped) yield* Deferred.await(prompted) diff --git a/packages/gui-extensions/src/ssh/askpass.ts b/packages/gui-extensions/src/ssh/askpass.ts index bff73df673ab..6be836909119 100644 --- a/packages/gui-extensions/src/ssh/askpass.ts +++ b/packages/gui-extensions/src/ssh/askpass.ts @@ -1,5 +1,5 @@ import { NodeSocketServer } from "@effect/platform-node" -import { Deferred, Effect, Fiber, Schema, Semaphore } from "effect" +import { Deferred, Effect, Fiber, Predicate, Schema, Semaphore } from "effect" import { randomUUID } from "node:crypto" import { SshFailure } from "./command" @@ -16,24 +16,31 @@ export const createAskpass = Effect.fn("Ssh.askpass")(function* (input: { const pending = new Map>() const prompts = yield* Semaphore.make(1) const server = yield* NodeSocketServer.make({ host: "127.0.0.1", port: 0 }).pipe(Effect.mapError(SshFailure.from)) - if (server.address._tag !== "TcpAddress") return yield* Effect.fail(new SshFailure("connection")) + + if (!Predicate.isTagged(server.address, "TcpAddress")) return yield* Effect.fail(new SshFailure("connection")) const serving = yield* server .run((socket) => Effect.gen(function* () { const request = yield* Deferred.make() const state = { buffer: "", received: false } + const reader = yield* socket .runString((chunk) => { if (state.received) return Effect.fail(new SshFailure("connection")) state.buffer += chunk + if (state.buffer.length > 16_384) return Effect.fail(new SshFailure("connection")) + if (!state.buffer.includes("\n")) return Effect.void state.received = true + return Deferred.succeed(request, state.buffer.trim()) }) .pipe(Effect.ensuring(Deferred.fail(request, new SshFailure("connection"))), Effect.forkScoped) + const message = yield* Deferred.await(request).pipe(Effect.flatMap(Schema.decodeUnknownEffect(Request))) + if (message.token !== token) return // One scoped waiter per helper invocation. Disconnecting a helper or closing @@ -75,6 +82,7 @@ export const createAskpass = Effect.fn("Ssh.askpass")(function* (input: { closed: Fiber.join(serving), respond: Effect.fn("Ssh.askpass.respond")(function* (id: string, value: string) { const response = pending.get(id) + if (response) yield* Deferred.succeed(response, value) }), } diff --git a/packages/gui-extensions/src/ssh/bootstrap.ts b/packages/gui-extensions/src/ssh/bootstrap.ts index daf1258167b9..899d0feaf53b 100644 --- a/packages/gui-extensions/src/ssh/bootstrap.ts +++ b/packages/gui-extensions/src/ssh/bootstrap.ts @@ -1,4 +1,4 @@ -import { Effect, Schema } from "effect" +import { Effect, Option, Schema } from "effect" import { HttpClient } from "effect/unstable/http" import { parseTarget, quote, runSsh, sshArgs, SshFailure } from "./command" import { RemoteCli } from "./remote-cli" @@ -42,13 +42,17 @@ export function parseRegistration(output: string) { .split(/\r?\n/) .findLast((line) => line.startsWith("OPENCODE_SSH_STATUS=")) ?.slice("OPENCODE_SSH_STATUS=".length) + if (!status) return undefined + for (const match of output.matchAll( /OPENCODE_SSH_REGISTRATION_BEGIN\r?\n([\s\S]*?)\r?\nOPENCODE_SSH_REGISTRATION_END/g, )) { const result = Schema.decodeUnknownOption(Registration)(match[1]) - if (result._tag === "Some" && result.value.url === status) return result.value + + if (Option.isSome(result) && result.value.url === status) return result.value } + return undefined } @@ -58,8 +62,10 @@ export function binaryPath(version: string) { function connectionAddress(address: string, password: string) { const url = new URL(address) + if (url.protocol !== "http:" || !["127.0.0.1", "localhost", "0.0.0.0", "[::]", "[::1]"].includes(url.hostname)) throw new SshFailure("service") + return { host: url.hostname === "[::1]" ? "[::1]" : "127.0.0.1", port: Number(url.port || 80), @@ -81,30 +87,39 @@ export const bootstrap = Effect.fn("Ssh.bootstrap")(function* (input: { env: input.env, stdin: script, }) + yield* input.stage("checking") const registered = parseRegistration(yield* run(discoverScript)) + if (registered && (input.development || registered.version === input.version)) { yield* input.stage("starting") + return yield* Effect.try({ try: () => connectionAddress(registered.url, registered.password), catch: SshFailure.from, }) } + if (registered && !input.replace) return yield* Effect.fail(new SshFailure("version", registered.version)) const destination = yield* Effect.try({ try: () => binaryPath(input.version), catch: SshFailure.from }) const existing = yield* run(RemoteCli.versionScript(`"${destination}"`)) const staged = RemoteCli.parseVersion(existing) === input.version + // Source worktree versions are unpublished. Use the installer's beta channel // while retaining support for explicitly staged, matching development builds. const version = input.development && !staged ? yield* RemoteCli.latestBeta().pipe(Effect.mapError(SshFailure.from)) : input.version + const setup = { version, directory: `.opencode/desktop-ssh/${version}` } + if (!staged) { const output = yield* run(RemoteCli.probeScript).pipe(Effect.mapError(() => new SshFailure("platform"))) + const target = output .split(/\r?\n/) .findLast((line) => line.startsWith("OPENCODE_REMOTE_TARGET=")) ?.split("=")[1] + const url = yield* Effect.try({ try: () => RemoteCli.archiveUrl(target ?? "", version), catch: SshFailure.from }) yield* input.stage("downloading") yield* run(RemoteCli.installScript({ ...setup, source: { type: "download", url } })).pipe( @@ -113,6 +128,7 @@ export const bootstrap = Effect.fn("Ssh.bootstrap")(function* (input: { yield* input.stage("uploading") const http = yield* HttpClient.HttpClient const response = yield* http.get(url).pipe(Effect.mapError(SshFailure.from)) + if (response.status < 200 || response.status >= 300) return yield* Effect.fail( new SshFailure( @@ -121,6 +137,7 @@ export const bootstrap = Effect.fn("Ssh.bootstrap")(function* (input: { ), ) const archive = new Uint8Array(yield* response.arrayBuffer.pipe(Effect.mapError(SshFailure.from))) + // The upload uses stdin; the script itself must be the remote command. return yield* runSsh({ args: [ @@ -135,11 +152,15 @@ export const bootstrap = Effect.fn("Ssh.bootstrap")(function* (input: { ), ) } + yield* input.stage("starting") const registration = parseRegistration(yield* run(startScript(version, input.replace))) + if (!registration) return yield* Effect.fail(new SshFailure("service")) + if (!input.development && registration.version !== input.version) return yield* Effect.fail(new SshFailure("version", registration.version)) + return yield* Effect.try({ try: () => connectionAddress(registration.url, registration.password), catch: SshFailure.from, diff --git a/packages/gui-extensions/src/ssh/command.ts b/packages/gui-extensions/src/ssh/command.ts index 85b34cef4fcb..420ccb09b680 100644 --- a/packages/gui-extensions/src/ssh/command.ts +++ b/packages/gui-extensions/src/ssh/command.ts @@ -1,4 +1,4 @@ -import { Effect, PlatformError, Schema, Stream } from "effect" +import { Effect, PlatformError, Predicate, Schema, Stream } from "effect" import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process" import { RemoteCli } from "./remote-cli" @@ -23,17 +23,19 @@ export class SshFailure extends Schema.TaggedError()("SshFailure", { return this.detail } - static from(this: void, error: unknown) { - if (error instanceof RemoteCli.Failure) return new SshFailure(error.code, error.detail) + static from(this: void, cause: unknown) { + if (cause instanceof RemoteCli.Failure) return new SshFailure(cause.code, cause.detail) + if ( - error instanceof PlatformError.PlatformError && - error.reason._tag === "NotFound" && - error.reason.method === "spawn" + cause instanceof PlatformError.PlatformError && + Predicate.isTagged(cause.reason, "NotFound") && + cause.reason.method === "spawn" ) - return new SshFailure("ssh-missing", error.message) - return error instanceof SshFailure - ? error - : new SshFailure("connection", error instanceof Error ? error.message : String(error)) + return new SshFailure("ssh-missing", cause.message) + + return cause instanceof SshFailure + ? cause + : new SshFailure("connection", cause instanceof Error ? cause.message : String(cause)) } } @@ -44,37 +46,48 @@ export function quote(value: string) { export function parseTarget(input: string) { const tokens: string[] = [] const state = { word: "", quote: "", started: false } + for (let i = 0; i < input.length; i++) { const c = input[i] ?? "" + if (c === "\n" || c === "\r" || c === "\0") throw new SshFailure("input") + if (c === "\\" && state.quote !== "'" && i + 1 < input.length && /[\s\\"']/.test(input[i + 1] ?? "")) { state.word += input[++i] state.started = true continue } + if (state.quote) { if (c === state.quote) state.quote = "" else state.word += c continue } + if (c === "'" || c === '"') { state.quote = c state.started = true continue } + if (/\s/.test(c)) { if (state.started) tokens.push(state.word) state.word = "" state.started = false continue } + state.word += c state.started = true } + if (state.quote) throw new SshFailure("input") + if (state.started) tokens.push(state.word) + if (tokens[0] === "ssh") tokens.shift() const args: string[] = [] + const options = new Set([ "hostname", "user", @@ -87,25 +100,36 @@ export function parseTarget(input: string) { "connecttimeout", "addressfamily", ]) + while (tokens[0]?.startsWith("-")) { const token = tokens.shift() ?? "" + if (["-4", "-6", "-C", "-A", "-a"].includes(token)) { args.push(token) continue } + const flag = token.slice(0, 2) + if (!["-p", "-l", "-i", "-F", "-J", "-o"].includes(flag)) throw new SshFailure("input") const value = token.length > 2 ? token.slice(2) : tokens.shift() + if (!value || value.startsWith("-")) throw new SshFailure("input") + if (flag === "-p" && (!/^\d+$/.test(value) || Number(value) < 1 || Number(value) > 65535)) throw new SshFailure("input") + if (flag === "-o" && !options.has((value.split(/[=\s]/)[0] ?? "").toLowerCase())) throw new SshFailure("input") args.push(flag, value) } + const host = tokens[0] + if (tokens.length !== 1 || !host || !/^[a-zA-Z0-9_@.:[\]%-]+$/.test(host) || host.startsWith("-")) throw new SshFailure("input") + if (host.includes("@") && host.slice(0, host.lastIndexOf("@")).includes(":")) throw new SshFailure("input") + return { host, args } } @@ -159,6 +183,7 @@ export const runSsh = Effect.fn("Ssh.run")(function* (input: { timeout?: number }) { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner + return yield* Effect.gen(function* () { const child = yield* spawner.spawn( ChildProcess.make(sshExecutable(), input.args, { @@ -172,12 +197,13 @@ export const runSsh = Effect.fn("Ssh.run")(function* (input: { ? "ignore" : { stream: Stream.make( - typeof input.stdin === "string" ? new TextEncoder().encode(input.stdin) : input.stdin, + Predicate.isString(input.stdin) ? new TextEncoder().encode(input.stdin) : input.stdin, ), endOnDone: true, }, }), ) + const output = yield* Effect.all( { stdout: child.stdout.pipe( @@ -198,8 +224,10 @@ export const runSsh = Effect.fn("Ssh.run")(function* (input: { }, { concurrency: "unbounded" }, ) + if (output.code !== 0) return yield* Effect.fail(new SshFailure("connection", commandFailureDetail(output.code, output))) + return output.stdout }).pipe(Effect.scoped, Effect.timeout(input.timeout ?? 600_000), Effect.mapError(SshFailure.from)) }) @@ -210,5 +238,6 @@ export function commandFailureDetail(code: number | null, output: { stdout: stri const stdout = output.stdout .replace(/OPENCODE_SSH_REGISTRATION_BEGIN[\s\S]*?(?:OPENCODE_SSH_REGISTRATION_END|$)/g, "") .trim() + return [output.stderr.trim(), stdout].filter(Boolean).join("\n") || JSON.stringify({ exitCode: code }) } diff --git a/packages/gui-extensions/src/ssh/controller.ts b/packages/gui-extensions/src/ssh/controller.ts index 9accd2506426..0eb05b9927cf 100644 --- a/packages/gui-extensions/src/ssh/controller.ts +++ b/packages/gui-extensions/src/ssh/controller.ts @@ -8,6 +8,7 @@ import { Fiber, FileSystem, Path, + Predicate, PubSub, Ref, Schedule, @@ -27,6 +28,7 @@ type Connection = { ready: Deferred.Deferred respond?: (id: string, value: string) => Effect.Effect } + type Attempt = Connection & { fiber: Fiber.Fiber } export const createSshController = Effect.fn("Ssh.controller")(function* (input: { @@ -49,12 +51,14 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: const items = new Map( input.configs.map((config) => [config.id, { config, saved: true, stage: "disconnected", detail: "" }]), ) + const configs = new Map(input.configs.map((config) => [config.id, config])) const attempts = new Map() const paused = new Set(items.keys()) const failures = new Map() const lifecycle = { closed: false } const emit = PubSub.publish(changed, undefined).pipe(Effect.asVoid) + const state = (owner?: number): Effect.Effect> => Effect.sync(() => ({ servers: [...items.values()].map((item) => ({ @@ -66,12 +70,22 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: attempts.get(item.config.id)?.owner !== owner, })), })) + const update = Effect.fnUntraced(function* (id: string, value: Partial) { const item = items.get(id) + if (!item) return items.set(id, { ...item, ...value }) yield* emit }) + + // A saved server's item takes its current saved config along with the change. + const withSavedConfig = (id: string, value: Partial): Partial => { + const config = configs.get(id) + + return config ? { ...value, config } : value + } + const run = (options: Parameters[0]) => runSsh(options).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner)) @@ -79,14 +93,17 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: const target = yield* Effect.try({ try: () => parseTarget(config.target), catch: SshFailure.from }) const directory = yield* fs.makeTempDirectoryScoped({ prefix: "oc-ssh-" }) const control = path.join(directory, "s") + const helper = input.command && input.command.length > 1 && process.platform !== "win32" ? path.join(directory, "askpass") : input.binary + if (helper !== input.binary) yield* fs.writeFileString(helper, `#!/bin/sh\nexec ${input.command?.map(quote).join(" ")} "$@"\n`, { mode: 0o700, }) + if (process.platform !== "win32") { target.args.unshift("-o", "ControlMaster=auto", "-o", "ControlPersist=60", "-o", `ControlPath=${control}`) // Close only our local SSH master. The remote OpenCode service owns its @@ -97,6 +114,7 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: } const authentication = yield* Deferred.make() + const askpass = yield* createAskpass({ binary: helper, prompt: Effect.fnUntraced(function* (prompt) { @@ -104,8 +122,10 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: paused.add(config.id) yield* update(config.id, { stage: "authentication" }) yield* Deferred.succeed(authentication, undefined) + return } + yield* update(config.id, { stage: "authentication", prompt }) }), clear: (id) => @@ -113,22 +133,27 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: ? update(config.id, { prompt: undefined, stage: "connecting" }) : Effect.void, }) + connection.respond = askpass.respond yield* Effect.gen(function* () { const resolved = yield* run({ args: [...sshArgs(target), "-G", target.host], timeout: 10_000 }).pipe( Effect.orElseSucceed(() => ""), ) + const fields = new Map( resolved.split(/\r?\n/).map((line) => { const separator = line.indexOf(" ") + return [line.slice(0, separator), line.slice(separator + 1)] as const }), ) + if (fields.has("hostname")) yield* update(config.id, { destination: `${fields.get("user") ?? ""}@${fields.get("hostname")}:${fields.get("port") ?? "22"}`, }) + const remote = yield* bootstrap({ target, version: input.version, @@ -140,8 +165,10 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), Effect.provideService(HttpClient.HttpClient, httpClient), ) + const port = yield* freePort const http = { url: `http://127.0.0.1:${port}`, password: remote.password } + const tunnel = yield* spawner.spawn( ChildProcess.make(sshExecutable(), tunnelArgs(target, port, remote), { env: askpass.env, @@ -153,19 +180,24 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: forceKillAfter: "2 seconds", }), ) + const detail = yield* Ref.make("") + const stderr = yield* tunnel.stderr.pipe( Stream.decodeText(), Stream.runForEach((text) => Ref.update(detail, (tail) => (tail + text).slice(-8192))), Effect.forkScoped, ) + const closed = Effect.gen(function* () { const exitCode = yield* tunnel.exitCode yield* Fiber.join(stderr) + return yield* Effect.fail( new SshFailure("connection", (yield* Ref.get(detail)) || JSON.stringify({ exitCode })), ) }) + yield* waitReady(http, () => items.get(config.id)?.stage === "authentication").pipe( Effect.provideService(HttpClient.HttpClient, httpClient), Effect.catch(() => @@ -188,20 +220,26 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: const start = Effect.fn("Ssh.start")(function* (request: SshStart, owner?: number) { const id = request.id + if (lifecycle.closed || !/^[a-zA-Z0-9-]{1,80}$/.test(id)) return const previous = attempts.get(id) + // A second window must not replace an interactive attempt while its owner // is connecting or answering a challenge. if (previous?.owner !== undefined && previous.owner !== owner && items.get(id)?.stage !== "ready") return const config = { id, target: request.target.trim(), name: request.name.trim() } + const connection: Connection = { owner, before: items.get(id), ready: yield* Deferred.make(), } + const admitted = yield* Deferred.make() + const fiber = yield* Effect.gen(function* () { yield* Deferred.await(admitted) + if (previous) yield* Fiber.interrupt(previous.fiber) yield* connect(config, connection, request.replace) }).pipe( @@ -210,29 +248,34 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: if (Cause.hasInterruptsOnly(cause) || paused.has(id) || attempts.get(id)?.ready !== connection.ready) return const failure = SshFailure.from(Cause.squash(cause)) failures.set(id, (failures.get(id) ?? 0) + 1) + const code = /REMOTE HOST IDENTIFICATION HAS CHANGED|Host key verification failed/.test(failure.message) ? "host-key" : /spawn .*ENOENT/.test(failure.message) ? "ssh-missing" : failure.code + if ( ["version", "input", "unpublished", "platform", "host-key", "ssh-missing"].includes(code) || /Permission denied/.test(failure.message) || (failures.get(id) ?? 0) >= 5 ) paused.add(id) - yield* update(id, { - stage: code === "version" ? "incompatible" : "failed", - error: code, - detail: failure.message, - prompt: undefined, - ...(configs.has(id) ? { config: configs.get(id) } : {}), - }) + yield* update( + id, + withSavedConfig(id, { + stage: code === "version" ? "incompatible" : "failed", + error: code, + detail: failure.message, + prompt: undefined, + }), + ) }), ), Effect.ensuring( Effect.gen(function* () { yield* Deferred.succeed(connection.ready, null) + if (attempts.get(id)?.ready !== connection.ready) return attempts.delete(id) yield* update(id, { prompt: undefined }) @@ -240,6 +283,7 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: ), Effect.forkIn(lifetime, { uninterruptible: false }), ) + attempts.set(id, Object.assign(connection, { fiber })) items.set(id, { config, @@ -249,6 +293,7 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: detail: "", }) paused.delete(id) + if (!request.background) failures.delete(id) yield* emit yield* Deferred.succeed(admitted, undefined) @@ -257,15 +302,14 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: const disconnect = Effect.fn("Ssh.disconnect")(function* (id: string) { paused.add(id) const attempt = attempts.get(id) - yield* update(id, { - stage: "disconnected", - prompt: undefined, - ...(configs.has(id) ? { config: configs.get(id) } : {}), - }) + yield* update(id, withSavedConfig(id, { stage: "disconnected", prompt: undefined })) + if (attempt) yield* Fiber.interrupt(attempt.fiber) }) + const cancel = Effect.fn("Ssh.cancel")(function* (id: string, owner: number) { const attempt = attempts.get(id) + if (!attempt || attempt.owner !== owner) return paused.add(id) // Restore before interrupting: askpass cleanup must not transition the @@ -283,12 +327,14 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: }) yield* Fiber.interrupt(attempt.fiber) }) + const close = Effect.gen(function* () { if (lifecycle.closed) return lifecycle.closed = true yield* Effect.forEach([...attempts.keys()], disconnect, { concurrency: "unbounded", discard: true }) yield* Scope.close(lifetime, Exit.void) }) + yield* Effect.addFinalizer(() => close) return { @@ -297,22 +343,29 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: start, resolve: Effect.fn("Ssh.resolve")(function* (id: string) { const item = items.get(id) + if (lifecycle.closed || !item || paused.has(id)) return null + if (item.stage === "ready" && item.http) { const healthy = yield* checkHealth(item.http).pipe(Effect.provideService(HttpClient.HttpClient, httpClient)) + if (lifecycle.closed || paused.has(id)) return null + // Another window may already have replaced this tunnel during the probe. if (items.get(id) === item) { if (healthy) return item.http yield* start({ ...item.config, background: true }) } } + if (!attempts.has(id) && items.has(id)) yield* start({ ...item.config, background: true }) const attempt = attempts.get(id) + return attempt ? yield* Deferred.await(attempt.ready) : null }), respond: Effect.fn("Ssh.respond")(function* (id: string, prompt: string, value: string, owner: number) { const attempt = attempts.get(id) + if (attempt?.owner === owner && attempt.respond) yield* attempt.respond(prompt, value) }), disconnect, @@ -337,7 +390,9 @@ export const createSshController = Effect.fn("Ssh.controller")(function* (input: const freePort = Effect.gen(function* () { const server = yield* NodeSocketServer.make({ host: "127.0.0.1", port: 0 }) - if (server.address._tag !== "TcpAddress") return yield* Effect.fail(new SshFailure("connection")) + + if (!Predicate.isTagged(server.address, "TcpAddress")) return yield* Effect.fail(new SshFailure("connection")) + return server.address.port }).pipe(Effect.scoped) @@ -345,14 +400,18 @@ const waitReady = Effect.fn("Ssh.waitReady")(function* (http: SshHttp, authentic const clock = { deadline: (yield* Clock.currentTimeMillis) + 30_000 } yield* Effect.gen(function* () { const now = yield* Clock.currentTimeMillis + if (authenticating()) clock.deadline = now + 30_000 + if (now >= clock.deadline) return yield* Effect.fail(new SshFailure("service")) + return yield* checkHealth(http) }).pipe(Effect.repeat({ until: (ready) => ready, schedule: Schedule.spaced(100) })) }) const checkHealth = Effect.fn("Ssh.checkHealth")(function* (http: SshHttp) { const client = yield* HttpClient.HttpClient + return yield* client .get(`${http.url}/api/info`, { headers: { authorization: `Basic ${Buffer.from(`opencode:${http.password}`).toString("base64")}` }, diff --git a/packages/gui-extensions/src/ssh/remote-cli.ts b/packages/gui-extensions/src/ssh/remote-cli.ts index e516f558fad2..a59326684725 100644 --- a/packages/gui-extensions/src/ssh/remote-cli.ts +++ b/packages/gui-extensions/src/ssh/remote-cli.ts @@ -19,6 +19,7 @@ export function quote(value: string) { export function requireVersion(version: string) { if (version !== "local" && !/^[0-9][a-zA-Z0-9.+-]*$/.test(version)) throw new Failure({ code: "version", detail: version }) + return version } @@ -48,10 +49,13 @@ export function parseVersion(output: string) { .split(/\r?\n/) .find((line) => line.trim()) ?.trim() + if (!line) return null const marker = line.lastIndexOf(" v") const version = marker === -1 ? line : line.slice(marker + 2) + if (!version) throw new Error("V2 CLI did not provide a version") + return version } @@ -71,6 +75,7 @@ printf 'OPENCODE_REMOTE_TARGET=%s\\n' "$target" export function archiveUrl(target: string, version: string) { if (!/^(linux|darwin)-(x64-baseline|arm64)(-musl)?$/.test(target)) throw new Failure({ code: "platform", detail: target }) + return `https://registry.npmjs.org/@opencode/cli-${target}/-/cli-${target}-${requireVersion(version)}.tgz` } @@ -78,6 +83,7 @@ type Source = { type: "download"; url: string } | { type: "archive" } | { type: export function installScript(input: { version: string; directory?: string; source: Source }) { const version = requireVersion(input.version) + // The managed CLI installer also configures the user's shell PATH. Private // installations use archives so their destination and shell setup stay isolated. if (input.source.type === "installer") @@ -85,6 +91,7 @@ export function installScript(input: { version: string; directory?: string; sour curl -fsSL https://raw.githubusercontent.com/anomalyco/opencode/v2/install | bash -s -- ${input.source.binary ? `--binary ${input.source.binary}` : `--version ${quote(version)}`} ${verifyScript('"$HOME/.opencode/bin/opencode"', version)} ` + return `set -eu umask 077 destination="$HOME"/${quote(`${input.directory ?? ".opencode/bin"}/opencode`)} @@ -100,6 +107,7 @@ mv "$stage/package/bin/opencode" "$destination" function stageBinary(source: Exclude) { if (source.type === "archive") return 'cat > "$stage/archive.tgz"\ntar -xzf "$stage/archive.tgz" -C "$stage"' + return `url=${quote(source.url)} if command -v curl >/dev/null 2>&1; then curl -fsSL --connect-timeout 15 --max-time 180 "$url" -o "$stage/archive.tgz" @@ -117,11 +125,13 @@ const Beta = Schema.Struct({ version: Schema.String.check(Schema.isPattern(/^0\. export const latestBeta = Effect.fn("RemoteCli.latestBeta")(function* () { const http = yield* HttpClient.HttpClient + const metadata = yield* http.get("https://registry.npmjs.org/@opencode%2fcli/beta").pipe( Effect.flatMap(HttpClientResponse.filterStatusOk), Effect.flatMap(HttpClientResponse.schemaBodyJson(Beta)), Effect.timeout("30 seconds"), Effect.mapError(() => new Failure({ code: "install", detail: "https://registry.npmjs.org/@opencode%2fcli/beta" })), ) + return metadata.version })