From 20e37e71229246d6786d55fa8b76633df156d9e1 Mon Sep 17 00:00:00 2001 From: Dax Raad Date: Fri, 10 Jul 2026 17:18:03 -0400 Subject: [PATCH] fix(plugin): adapt promise host wire values --- .../client/src/promise/generated/client.ts | 4 +- packages/core/src/plugin/promise.ts | 150 ++++++++++++++++-- packages/httpapi-codegen/src/index.ts | 2 +- 3 files changed, 137 insertions(+), 19 deletions(-) diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 28c223e6b5..c030946153 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -193,12 +193,12 @@ import { ClientError } from "./client-error" export interface ClientOptions { readonly baseUrl: string readonly fetch?: typeof globalThis.fetch - readonly headers?: HeadersInit + readonly headers?: RequestInit["headers"] } export interface RequestOptions { readonly signal?: AbortSignal - readonly headers?: HeadersInit + readonly headers?: RequestInit["headers"] } interface RequestDescriptor { diff --git a/packages/core/src/plugin/promise.ts b/packages/core/src/plugin/promise.ts index 716d69c10d..d89505b7fe 100644 --- a/packages/core/src/plugin/promise.ts +++ b/packages/core/src/plugin/promise.ts @@ -2,13 +2,26 @@ export * as PluginPromise from "./promise" import { Plugin } from "@opencode-ai/plugin/v2/effect" import type { AnyTool } from "@opencode-ai/plugin/v2/tool" -import { Effect, Scope, Stream } from "effect" +import { Agent } from "@opencode-ai/schema/agent" +import { Integration } from "@opencode-ai/schema/integration" +import { Location } from "@opencode-ai/schema/location" +import { Model } from "@opencode-ai/schema/model" +import { Provider } from "@opencode-ai/schema/provider" +import { AbsolutePath } from "@opencode-ai/schema/schema" +import { Session } from "@opencode-ai/schema/session" +import { SessionMessage } from "@opencode-ai/schema/session-message" +import { Workspace } from "@opencode-ai/schema/workspace" +import { DateTime, Effect, Scope, Stream } from "effect" import { Tool } from "../tool/tool" type HostRegistration = { readonly dispose: Effect.Effect } type Registration = { readonly dispose: () => Promise } type PromisePlugin = import("@opencode-ai/plugin/v2/plugin").Plugin type PromisePluginContext = import("@opencode-ai/plugin/v2/plugin").Context +type PromiseEvent = ReturnType extends AsyncIterable + ? Event + : never +type JsonValue = null | boolean | number | string | Array | { [key: string]: JsonValue } /** * Adapts a Promise plugin into an Effect plugin so the existing Effect-only @@ -33,7 +46,7 @@ export function fromPromise(plugin: PromisePlugin) { dispose: () => Effect.runPromiseWith(context)(registration.dispose), })) - const run = (effect: Effect.Effect) => Effect.runPromiseWith(context)(effect) + const run = (effect: Effect.Effect) => Effect.runPromiseWith(context)(effect).then(wire) const transform = (domain: { @@ -60,11 +73,12 @@ export function fromPromise(plugin: PromisePlugin) { catalog: { provider: { list: (input) => run(host.catalog.provider.list(input)), - get: (input) => run(host.catalog.provider.get(input)), + get: (input) => run(host.catalog.provider.get({ ...input, providerID: Provider.ID.make(input.providerID) })), }, model: { list: (input) => run(host.catalog.model.list(input)), - default: (input) => run(host.catalog.model.default(input)), + default: (input) => + run(host.catalog.model.default(input)).then((result) => ({ ...result, data: result.data ?? null })), }, transform: transform(host.catalog), reload: () => run(host.catalog.reload()), @@ -75,19 +89,48 @@ export function fromPromise(plugin: PromisePlugin) { reload: () => run(host.command.reload()), }, event: { - subscribe: () => Stream.toAsyncIterable(host.event.subscribe()), + subscribe: () => Stream.toAsyncIterable(host.event.subscribe().pipe(Stream.map(wireEvent))), }, integration: { list: (input) => run(host.integration.list(input)), - get: (input) => run(host.integration.get(input)), + get: (input) => + run(host.integration.get({ ...input, integrationID: Integration.ID.make(input.integrationID) })).then( + (result) => ({ ...result, data: result.data ?? null }), + ), connect: { - key: (input) => run(host.integration.connect.key(input)), - oauth: (input) => run(host.integration.connect.oauth(input)), + key: (input) => + run(host.integration.connect.key({ ...input, integrationID: Integration.ID.make(input.integrationID) })), + oauth: (input) => + run( + host.integration.connect.oauth({ + ...input, + integrationID: Integration.ID.make(input.integrationID), + methodID: Integration.MethodID.make(input.methodID), + }), + ), }, attempt: { - status: (input) => run(host.integration.attempt.status(input)), - complete: (input) => run(host.integration.attempt.complete(input)), - cancel: (input) => run(host.integration.attempt.cancel(input)), + status: (input) => + run( + host.integration.attempt.status({ + ...input, + attemptID: Integration.AttemptID.make(input.attemptID), + }), + ), + complete: (input) => + run( + host.integration.attempt.complete({ + ...input, + attemptID: Integration.AttemptID.make(input.attemptID), + }), + ), + cancel: (input) => + run( + host.integration.attempt.cancel({ + ...input, + attemptID: Integration.AttemptID.make(input.attemptID), + }), + ), }, transform: transform(host.integration), reload: () => run(host.integration.reload()), @@ -122,11 +165,53 @@ export function fromPromise(plugin: PromisePlugin) { register(host.tool.hook(name, (event) => Effect.promise(() => Promise.resolve(callback(event))))), }, session: { - create: (input) => run(host.session.create(input)), - get: (input) => run(host.session.get(input)), - prompt: (input) => run(host.session.prompt(input)), - command: (input) => run(host.session.command(input)), - interrupt: (input) => run(host.session.interrupt(input)), + create: (input) => + run( + host.session.create( + input === undefined + ? undefined + : { + id: input.id == null ? undefined : Session.ID.make(input.id), + agent: input.agent == null ? undefined : Agent.ID.make(input.agent), + model: input.model == null ? undefined : model(input.model), + location: + input.location == null + ? undefined + : Location.Ref.make({ + directory: AbsolutePath.make(input.location.directory), + workspaceID: + input.location.workspaceID === undefined + ? undefined + : Workspace.ID.make(input.location.workspaceID), + }), + }, + ), + ), + get: (input) => run(host.session.get({ sessionID: Session.ID.make(input.sessionID) })), + prompt: (input) => + run( + host.session.prompt({ + ...input, + sessionID: Session.ID.make(input.sessionID), + id: input.id == null ? undefined : SessionMessage.ID.make(input.id), + delivery: input.delivery ?? undefined, + resume: input.resume ?? undefined, + }), + ), + command: (input) => + run( + host.session.command({ + ...input, + sessionID: Session.ID.make(input.sessionID), + id: input.id == null ? undefined : SessionMessage.ID.make(input.id), + agent: input.agent == null ? undefined : Agent.ID.make(input.agent), + model: input.model == null ? undefined : model(input.model), + arguments: input.arguments ?? undefined, + delivery: input.delivery ?? undefined, + resume: input.resume ?? undefined, + }), + ), + interrupt: (input) => run(host.session.interrupt({ sessionID: Session.ID.make(input.sessionID) })), }, } @@ -137,6 +222,39 @@ export function fromPromise(plugin: PromisePlugin) { }) } +function model(input: { readonly id: string; readonly providerID: string; readonly variant?: string }) { + return Model.Ref.make({ + id: Model.ID.make(input.id), + providerID: Provider.ID.make(input.providerID), + variant: input.variant === undefined ? undefined : Model.VariantID.make(input.variant), + }) +} + +type Wire = unknown extends Value + ? JsonValue + : Value extends string | number | boolean | bigint | symbol | null | undefined + ? Value + : Value extends DateTime.DateTime + ? number + : Value extends ReadonlyArray + ? Array> + : Value extends object + ? { -readonly [Key in keyof Value]: Wire } + : Value + +function wire(value: Value): Wire +function wire(value: unknown): unknown { + if (DateTime.isDateTime(value)) return DateTime.toEpochMillis(value) + if (Array.isArray(value)) return value.map(wire) + if (typeof value !== "object" || value === null) return value + return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, wire(item)])) +} + +function wireEvent(value: unknown): PromiseEvent +function wireEvent(value: unknown): unknown { + return wire(value) +} + function fromPromiseTool(tool: AnyTool) { if ("jsonSchema" in tool) return Tool.make({ diff --git a/packages/httpapi-codegen/src/index.ts b/packages/httpapi-codegen/src/index.ts index 60548f247a..069da2c1e7 100644 --- a/packages/httpapi-codegen/src/index.ts +++ b/packages/httpapi-codegen/src/index.ts @@ -722,7 +722,7 @@ function renderPromiseClient(groups: ReadonlyArray) { if (group.endpoints[0]?.topLevel) return fields return `${JSON.stringify(group.identifier)}: { ${fields} }` }) - return `import type { ${imports.join(", ")} } from "./types"\nimport { ClientError } from "./client-error"\n\nexport interface ClientOptions {\n readonly baseUrl: string\n readonly fetch?: typeof globalThis.fetch\n readonly headers?: HeadersInit\n}\n\nexport interface RequestOptions {\n readonly signal?: AbortSignal\n readonly headers?: HeadersInit\n}\n\ninterface RequestDescriptor {\n readonly method: string\n readonly path: string\n readonly query?: Record\n readonly headers?: Record\n readonly body?: unknown\n readonly successStatus: number\n readonly declaredStatuses: ReadonlyArray\n readonly empty: boolean\n}\n\nexport function make(options: ClientOptions) {\n const fetch = options.fetch ?? globalThis.fetch\n\n const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n const url = new URL(descriptor.path, options.baseUrl)\n for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n const headers = new Headers(options.headers)\n for (const [key, value] of Object.entries(descriptor.headers ?? {})) {\n if (value !== undefined && value !== null) headers.set(key, String(value))\n }\n for (const [key, value] of new Headers(requestOptions?.headers)) headers.set(key, value)\n if (descriptor.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json")\n return {\n url,\n init: {\n method: descriptor.method,\n signal: requestOptions?.signal,\n headers,\n body: descriptor.body === undefined ? undefined : JSON.stringify(descriptor.body),\n } satisfies RequestInit,\n }\n }\n\n const execute = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n try {\n const prepared = prepare(descriptor, requestOptions)\n return await fetch(prepared.url, prepared.init)\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n }\n\n const responseError = async (response: Response, descriptor: RequestDescriptor): Promise => {\n if (descriptor.declaredStatuses.includes(response.status)) throw await json(response)\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnexpectedStatus", { cause: { status: response.status } })\n }\n\n const request = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions): Promise => {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) return responseError(response, descriptor)\n if (descriptor.empty) {\n try {\n await response.body?.cancel()\n } catch {}\n return undefined as A\n }\n return await json(response) as A\n }\n\n const sse = (descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable => ({\n async *[Symbol.asyncIterator]() {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) await responseError(response, descriptor)\n if (!isContentType(response, "text/event-stream")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n if (response.body === null) throw new ClientError("MalformedResponse")\n const reader = response.body.getReader()\n const decoder = new TextDecoder()\n let buffer = ""\n try {\n while (true) {\n let next: ReadableStreamReadResult\n try {\n next = await reader.read()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n buffer += decoder.decode(next.value, { stream: !next.done })\n if (buffer.length > 1_048_576) throw new ClientError("MalformedResponse")\n const trailingCarriageReturn = !next.done && buffer.endsWith("\\r")\n if (trailingCarriageReturn) buffer = buffer.slice(0, -1)\n buffer = buffer.replaceAll("\\r\\n", "\\n").replaceAll("\\r", "\\n")\n if (trailingCarriageReturn) buffer += "\\r"\n if (next.done && buffer !== "") buffer += "\\n\\n"\n let boundary = buffer.indexOf("\\n\\n")\n while (boundary >= 0) {\n const block = buffer.slice(0, boundary)\n buffer = buffer.slice(boundary + 2)\n const data = block.split("\\n").flatMap((line) => line.startsWith("data:") ? [line.slice(5).trimStart()] : []).join("\\n")\n if (data !== "") {\n try {\n yield JSON.parse(data) as A\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n }\n boundary = buffer.indexOf("\\n\\n")\n }\n if (next.done) return\n }\n } finally {\n try {\n await reader.cancel()\n } catch {}\n reader.releaseLock()\n }\n },\n })\n\n return { ${fields.join(", ")} }\n}\n\nfunction appendQuery(params: URLSearchParams, key: string, value: unknown): void {\n if (value === undefined) return\n if (value === null) {\n params.append(key, "null")\n return\n }\n if (Array.isArray(value)) {\n for (const item of value) appendQuery(params, key, item)\n return\n }\n if (typeof value === "object") {\n for (const [child, item] of Object.entries(value)) appendQuery(params, \`\${key}[\${child}]\`, item)\n return\n }\n params.append(key, String(value))\n}\n\nasync function json(response: Response): Promise {\n if (!isContentType(response, "application/json") && !response.headers.get("content-type")?.includes("+json")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n let text: string\n try {\n text = await response.text()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (text === "") throw new ClientError("MalformedResponse")\n try {\n return JSON.parse(text)\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n}\n\nfunction isContentType(response: Response, expected: string) {\n return response.headers.get("content-type")?.split(";", 1)[0]?.trim().toLowerCase() === expected\n}\n` + return `import type { ${imports.join(", ")} } from "./types"\nimport { ClientError } from "./client-error"\n\nexport interface ClientOptions {\n readonly baseUrl: string\n readonly fetch?: typeof globalThis.fetch\n readonly headers?: RequestInit["headers"]\n}\n\nexport interface RequestOptions {\n readonly signal?: AbortSignal\n readonly headers?: RequestInit["headers"]\n}\n\ninterface RequestDescriptor {\n readonly method: string\n readonly path: string\n readonly query?: Record\n readonly headers?: Record\n readonly body?: unknown\n readonly successStatus: number\n readonly declaredStatuses: ReadonlyArray\n readonly empty: boolean\n}\n\nexport function make(options: ClientOptions) {\n const fetch = options.fetch ?? globalThis.fetch\n\n const prepare = (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n const url = new URL(descriptor.path, options.baseUrl)\n for (const [key, value] of Object.entries(descriptor.query ?? {})) appendQuery(url.searchParams, key, value)\n const headers = new Headers(options.headers)\n for (const [key, value] of Object.entries(descriptor.headers ?? {})) {\n if (value !== undefined && value !== null) headers.set(key, String(value))\n }\n for (const [key, value] of new Headers(requestOptions?.headers)) headers.set(key, value)\n if (descriptor.body !== undefined && !headers.has("content-type")) headers.set("content-type", "application/json")\n return {\n url,\n init: {\n method: descriptor.method,\n signal: requestOptions?.signal,\n headers,\n body: descriptor.body === undefined ? undefined : JSON.stringify(descriptor.body),\n } satisfies RequestInit,\n }\n }\n\n const execute = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions) => {\n try {\n const prepared = prepare(descriptor, requestOptions)\n return await fetch(prepared.url, prepared.init)\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n }\n\n const responseError = async (response: Response, descriptor: RequestDescriptor): Promise => {\n if (descriptor.declaredStatuses.includes(response.status)) throw await json(response)\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnexpectedStatus", { cause: { status: response.status } })\n }\n\n const request = async (descriptor: RequestDescriptor, requestOptions?: RequestOptions): Promise => {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) return responseError(response, descriptor)\n if (descriptor.empty) {\n try {\n await response.body?.cancel()\n } catch {}\n return undefined as A\n }\n return await json(response) as A\n }\n\n const sse = (descriptor: RequestDescriptor, requestOptions?: RequestOptions): AsyncIterable => ({\n async *[Symbol.asyncIterator]() {\n const response = await execute(descriptor, requestOptions)\n if (response.status !== descriptor.successStatus) await responseError(response, descriptor)\n if (!isContentType(response, "text/event-stream")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n if (response.body === null) throw new ClientError("MalformedResponse")\n const reader = response.body.getReader()\n const decoder = new TextDecoder()\n let buffer = ""\n try {\n while (true) {\n let next: ReadableStreamReadResult\n try {\n next = await reader.read()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n buffer += decoder.decode(next.value, { stream: !next.done })\n if (buffer.length > 1_048_576) throw new ClientError("MalformedResponse")\n const trailingCarriageReturn = !next.done && buffer.endsWith("\\r")\n if (trailingCarriageReturn) buffer = buffer.slice(0, -1)\n buffer = buffer.replaceAll("\\r\\n", "\\n").replaceAll("\\r", "\\n")\n if (trailingCarriageReturn) buffer += "\\r"\n if (next.done && buffer !== "") buffer += "\\n\\n"\n let boundary = buffer.indexOf("\\n\\n")\n while (boundary >= 0) {\n const block = buffer.slice(0, boundary)\n buffer = buffer.slice(boundary + 2)\n const data = block.split("\\n").flatMap((line) => line.startsWith("data:") ? [line.slice(5).trimStart()] : []).join("\\n")\n if (data !== "") {\n try {\n yield JSON.parse(data) as A\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n }\n boundary = buffer.indexOf("\\n\\n")\n }\n if (next.done) return\n }\n } finally {\n try {\n await reader.cancel()\n } catch {}\n reader.releaseLock()\n }\n },\n })\n\n return { ${fields.join(", ")} }\n}\n\nfunction appendQuery(params: URLSearchParams, key: string, value: unknown): void {\n if (value === undefined) return\n if (value === null) {\n params.append(key, "null")\n return\n }\n if (Array.isArray(value)) {\n for (const item of value) appendQuery(params, key, item)\n return\n }\n if (typeof value === "object") {\n for (const [child, item] of Object.entries(value)) appendQuery(params, \`\${key}[\${child}]\`, item)\n return\n }\n params.append(key, String(value))\n}\n\nasync function json(response: Response): Promise {\n if (!isContentType(response, "application/json") && !response.headers.get("content-type")?.includes("+json")) {\n try {\n await response.body?.cancel()\n } catch {}\n throw new ClientError("UnsupportedContentType")\n }\n let text: string\n try {\n text = await response.text()\n } catch (cause) {\n throw new ClientError("Transport", { cause })\n }\n if (text === "") throw new ClientError("MalformedResponse")\n try {\n return JSON.parse(text)\n } catch (cause) {\n throw new ClientError("MalformedResponse", { cause })\n }\n}\n\nfunction isContentType(response: Response, expected: string) {\n return response.headers.get("content-type")?.split(";", 1)[0]?.trim().toLowerCase() === expected\n}\n` } function promiseTypePrefix(group: string, path: ReadonlyArray) {