mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-14 13:46:22 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a9c3d484ba |
@@ -188,12 +188,15 @@
|
||||
"@types/bun": "catalog:",
|
||||
"@typescript/native-preview": "catalog:",
|
||||
"effect": "catalog:",
|
||||
"solid-js": "catalog:",
|
||||
},
|
||||
"peerDependencies": {
|
||||
"effect": "4.0.0-beta.101",
|
||||
"solid-js": ">=1.9.0",
|
||||
},
|
||||
"optionalPeers": [
|
||||
"effect",
|
||||
"solid-js",
|
||||
],
|
||||
},
|
||||
"packages/codemode": {
|
||||
|
||||
@@ -11,6 +11,9 @@ import { extractPromptFromParts } from "@/utils/prompt"
|
||||
import type { TextPart as SDKTextPart } from "@/types"
|
||||
import { base64Encode } from "@opencode-ai/core/util/encode"
|
||||
import { useLanguage } from "@/context/language"
|
||||
import { useServerSDK } from "@/context/server-sdk"
|
||||
import { sessionHref } from "@/utils/session-route"
|
||||
import { ServerConnection } from "@/context/servers"
|
||||
|
||||
interface ForkableMessage {
|
||||
id: string
|
||||
@@ -27,6 +30,7 @@ export const DialogFork: Component = () => {
|
||||
const navigate = useNavigate()
|
||||
const sync = useSync()
|
||||
const sdk = useSDK()
|
||||
const serverSDK = useServerSDK()
|
||||
const prompt = usePrompt()
|
||||
const dialog = useDialog()
|
||||
const language = useLanguage()
|
||||
@@ -73,7 +77,7 @@ export const DialogFork: Component = () => {
|
||||
.then((forked) => {
|
||||
dialog.close()
|
||||
prompt.set(restored, undefined, { dir, id: forked.id })
|
||||
navigate(`/${dir}/session/${forked.id}`)
|
||||
navigate(sessionHref(ServerConnection.key(serverSDK.server), forked.id))
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
const message = err instanceof Error ? err.message : String(err)
|
||||
|
||||
@@ -4,7 +4,7 @@ import { useDialog } from "@opencode-ai/ui/context/dialog"
|
||||
import { ProviderIcon } from "@opencode-ai/ui/provider-icon"
|
||||
import { showToast } from "@/utils/toast"
|
||||
import { popularProviders, useProviders } from "@/hooks/use-providers"
|
||||
import { createMemo, type Component, For, Show } from "solid-js"
|
||||
import { createMemo, type Component, For, Show, createEffect } from "solid-js"
|
||||
import { useLanguage } from "@/context/language"
|
||||
import { useServerSDK } from "@/context/server-sdk"
|
||||
import { useServerSync } from "@/context/server-sync"
|
||||
@@ -14,6 +14,7 @@ import { SettingsServerScope } from "../settings-server-picker"
|
||||
import { InlineServerSelect } from "./parts/server-select"
|
||||
import { SettingsListV2 } from "./parts/list"
|
||||
import "./settings-v2.css"
|
||||
import { useData } from "@/context/server"
|
||||
|
||||
type ProviderSource = "env" | "api" | "config" | "custom"
|
||||
type ProviderItem = ReturnType<ReturnType<typeof useProviders>["connected"]>[number]
|
||||
@@ -35,6 +36,7 @@ export const SettingsProvidersV2: Component<{
|
||||
directory: string | undefined
|
||||
onBack?: () => void
|
||||
}> = (props) => {
|
||||
const data = useData()
|
||||
const dialog = useDialog()
|
||||
const language = useLanguage()
|
||||
const serverSdk = useServerSDK()
|
||||
|
||||
@@ -7,6 +7,7 @@ import { useServerHealth } from "@/utils/server-health"
|
||||
import { createServerSdkContext } from "./server-sdk"
|
||||
import { createServerSyncContext } from "./server-sync"
|
||||
import { getOwner } from "solid-js/web"
|
||||
import { createServerData } from "@opencode-ai/client/solid"
|
||||
import type { ServerScope } from "@/utils/server-scope"
|
||||
import { createServerPermissionState } from "./permission"
|
||||
import { createServerNotificationState } from "./notification"
|
||||
@@ -102,6 +103,12 @@ function createServerController(
|
||||
const sync = createServerSyncContext(sdk)
|
||||
const permission = createServerPermissionState({ sdk, sync })
|
||||
const notification = createServerNotificationState({ sdk, sync, key: connKey })
|
||||
const data = createServerData({
|
||||
api: () => sdk.api,
|
||||
event: sdk.event,
|
||||
connection: sdk.connection,
|
||||
directory: "",
|
||||
})
|
||||
|
||||
function enrich(project: { worktree: string; expanded: boolean }) {
|
||||
const [childStore] = sync.child(project.worktree, { bootstrap: false })
|
||||
@@ -134,6 +141,7 @@ function createServerController(
|
||||
(conn?.type === "sidecar" && conn.variant === "base") || (conn?.type === "http" && isLocalHost(conn.http.url))
|
||||
|
||||
return {
|
||||
data,
|
||||
sdk,
|
||||
sync,
|
||||
isLocal,
|
||||
|
||||
@@ -278,7 +278,7 @@ export function createServerNotificationState(input: { sdk: ServerSDK; sync: Ser
|
||||
})
|
||||
}
|
||||
|
||||
const unsub = input.sdk.event.listen((e) => {
|
||||
const unsub = input.sdk.eventByDir.listen((e) => {
|
||||
const event = e.details
|
||||
if (
|
||||
event.type !== "session.execution.succeeded" &&
|
||||
|
||||
@@ -52,7 +52,7 @@ function hasPermissionPromptRules(permission: unknown) {
|
||||
return Object.values(config).some(isNonAllowRule)
|
||||
}
|
||||
|
||||
type PermissionEvent = Parameters<Parameters<ServerSDK["event"]["listen"]>[0]>[0]
|
||||
type PermissionEvent = Parameters<Parameters<ServerSDK["eventByDir"]["listen"]>[0]>[0]
|
||||
|
||||
export function createServerPermissionState(input: { sdk: ServerSDK; sync: ServerSync }) {
|
||||
const [store, setStore, _, ready] = persisted(
|
||||
@@ -197,7 +197,7 @@ export function createServerPermissionState(input: { sdk: ServerSDK; sync: Serve
|
||||
void respondPending(event.properties, e.name)
|
||||
}
|
||||
|
||||
const unsubscribe = input.sdk.event.listen((event) => {
|
||||
const unsubscribe = input.sdk.eventByDir.listen((event) => {
|
||||
if (ready()) {
|
||||
handlePermission(event)
|
||||
return
|
||||
|
||||
@@ -1,18 +1,6 @@
|
||||
import { describe, expect, test } from "bun:test"
|
||||
import type { OpenCodeEvent } from "@opencode-ai/client/promise"
|
||||
import { adaptServerEvent, coalesceServerEvents, enqueueServerEvent, resumeStreamAfterPageShow } from "./server-sdk"
|
||||
|
||||
describe("resumeStreamAfterPageShow", () => {
|
||||
test("restarts a stream only after a back-forward cache restore", () => {
|
||||
let starts = 0
|
||||
const start = () => starts++
|
||||
|
||||
resumeStreamAfterPageShow({ persisted: false } as PageTransitionEvent, start)
|
||||
resumeStreamAfterPageShow({ persisted: true } as PageTransitionEvent, start)
|
||||
|
||||
expect(starts).toBe(1)
|
||||
})
|
||||
})
|
||||
import { adaptServerEvent } from "./server-sdk"
|
||||
|
||||
describe("adaptServerEvent", () => {
|
||||
test("preserves current permission requests", () => {
|
||||
@@ -43,57 +31,3 @@ describe("adaptServerEvent", () => {
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe("current event buffering", () => {
|
||||
const delta = (id: string, value: string, ordinal = 0) => ({
|
||||
directory: "/repo",
|
||||
payload: adaptServerEvent({
|
||||
id,
|
||||
created: 1,
|
||||
type: "session.text.delta",
|
||||
location: { directory: "/repo" },
|
||||
data: { sessionID: "ses", assistantMessageID: "msg", ordinal, delta: value },
|
||||
} as OpenCodeEvent),
|
||||
})
|
||||
|
||||
test("merges adjacent text deltas for the same message and ordinal", () => {
|
||||
const result = coalesceServerEvents([delta("evt_1", "hello "), delta("evt_2", "world")])
|
||||
|
||||
expect(result).toHaveLength(1)
|
||||
expect(result[0]?.payload.current).toMatchObject({ id: "evt_2", data: { delta: "hello world" } })
|
||||
expect(result[0]?.payload.properties).toMatchObject({ delta: "hello world" })
|
||||
})
|
||||
|
||||
test("coalesces current tool input deltas by tool ID", () => {
|
||||
const current = (eventID: string, id: string, delta: string) =>
|
||||
adaptServerEvent({
|
||||
id: eventID,
|
||||
created: 1,
|
||||
type: "session.tool.input.delta",
|
||||
location: { directory: "/repo" },
|
||||
data: { sessionID: "ses", assistantMessageID: "msg", id, delta },
|
||||
} as OpenCodeEvent)
|
||||
const result = coalesceServerEvents([
|
||||
{ directory: "/repo", payload: current("evt_1", "call_1", "{") },
|
||||
{ directory: "/repo", payload: current("evt_2", "call_1", "}") },
|
||||
{ directory: "/repo", payload: current("evt_3", "call_2", "[]") },
|
||||
])
|
||||
|
||||
expect(result).toHaveLength(2)
|
||||
expect(result[0]?.payload.current).toMatchObject({ id: "evt_2", data: { id: "call_1", delta: "{}" } })
|
||||
expect(result[1]?.payload.current).toMatchObject({ id: "evt_3", data: { id: "call_2", delta: "[]" } })
|
||||
})
|
||||
|
||||
test("preserves boundaries between distinct delta streams", () => {
|
||||
const events = [delta("evt_1", "a"), delta("evt_2", "b", 1), delta("evt_3", "c")]
|
||||
|
||||
expect(coalesceServerEvents(events).map((event) => event.payload.current?.id)).toEqual(["evt_1", "evt_2", "evt_3"])
|
||||
})
|
||||
|
||||
test("preserves current event order when enqueuing", () => {
|
||||
const events: Parameters<typeof enqueueServerEvent>[0] = []
|
||||
;[delta("evt_1", "a"), delta("evt_2", "b", 1)].forEach((event) => enqueueServerEvent(events, event))
|
||||
|
||||
expect(events.map((event) => event.payload.current?.id)).toEqual(["evt_1", "evt_2"])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
import type { OpenCodeEvent } from "@opencode-ai/client/promise"
|
||||
import { createClientConnection, type ClientConnectionStatus } from "@opencode-ai/client/solid"
|
||||
import type { Event } from "@/types"
|
||||
import { createGlobalEmitter } from "@solid-primitives/event-bus"
|
||||
import { makeEventListener } from "@solid-primitives/event-listener"
|
||||
import { type Accessor, batch, onCleanup, onMount } from "solid-js"
|
||||
import { createStore } from "solid-js/store"
|
||||
import { type Accessor, onCleanup } from "solid-js"
|
||||
import { createApiForServer, type ServerApi } from "@/utils/server"
|
||||
import { usePlatform } from "./platform"
|
||||
import { ServerConnection } from "./servers"
|
||||
@@ -12,85 +11,15 @@ import { ServerScope } from "@/utils/server-scope"
|
||||
import { useServer } from "./server"
|
||||
|
||||
export type ServerEvent = Event & { id?: string; current?: OpenCodeEvent }
|
||||
type QueuedServerEvent = { directory: string; payload: ServerEvent }
|
||||
type CurrentDelta = Extract<
|
||||
OpenCodeEvent,
|
||||
{ type: "session.text.delta" | "session.reasoning.delta" | "session.tool.input.delta" | "session.compaction.delta" }
|
||||
>
|
||||
|
||||
export function adaptServerEvent(event: OpenCodeEvent): ServerEvent {
|
||||
return { id: event.id, type: event.type, properties: event.data, current: event } as ServerEvent
|
||||
}
|
||||
|
||||
export function enqueueServerEvent(queue: QueuedServerEvent[], event: QueuedServerEvent) {
|
||||
queue.push(event)
|
||||
return true
|
||||
}
|
||||
|
||||
export function coalesceServerEvents(events: QueuedServerEvent[]) {
|
||||
const output: QueuedServerEvent[] = []
|
||||
events.forEach((event) => {
|
||||
const current = currentDelta(event.payload.current)
|
||||
if (current) {
|
||||
const previous = output[output.length - 1]
|
||||
const prior = currentDelta(previous?.payload.current)
|
||||
if (
|
||||
previous &&
|
||||
prior &&
|
||||
previous.directory === event.directory &&
|
||||
currentDeltaKey(prior) === currentDeltaKey(current)
|
||||
) {
|
||||
const fragment = currentDeltaFragment(prior) + currentDeltaFragment(current)
|
||||
const data =
|
||||
current.type === "session.compaction.delta"
|
||||
? { ...current.data, text: fragment }
|
||||
: { ...current.data, delta: fragment }
|
||||
output[output.length - 1] = {
|
||||
directory: event.directory,
|
||||
payload: {
|
||||
...event.payload,
|
||||
properties: data,
|
||||
current: { ...current, data } as CurrentDelta,
|
||||
} as ServerEvent,
|
||||
}
|
||||
return
|
||||
}
|
||||
output.push(event)
|
||||
return
|
||||
}
|
||||
output.push(event)
|
||||
})
|
||||
return output
|
||||
}
|
||||
|
||||
function currentDelta(event: OpenCodeEvent | undefined): CurrentDelta | undefined {
|
||||
if (
|
||||
event?.type === "session.text.delta" ||
|
||||
event?.type === "session.reasoning.delta" ||
|
||||
event?.type === "session.tool.input.delta" ||
|
||||
event?.type === "session.compaction.delta"
|
||||
)
|
||||
return event
|
||||
}
|
||||
|
||||
function currentDeltaKey(event: CurrentDelta) {
|
||||
if (event.type === "session.tool.input.delta")
|
||||
return `${event.type}:${event.data.sessionID}:${event.data.assistantMessageID}:${event.data.id}`
|
||||
if (event.type === "session.compaction.delta") return `${event.type}:${event.data.sessionID}`
|
||||
return `${event.type}:${event.data.sessionID}:${event.data.assistantMessageID}:${event.data.ordinal}`
|
||||
}
|
||||
|
||||
function currentDeltaFragment(event: CurrentDelta) {
|
||||
return event.type === "session.compaction.delta" ? event.data.text : event.data.delta
|
||||
}
|
||||
|
||||
export function resumeStreamAfterPageShow(event: PageTransitionEvent, start: () => unknown) {
|
||||
if (!event.persisted) return
|
||||
start()
|
||||
}
|
||||
|
||||
type ServerEventEmitter = ReturnType<typeof createGlobalEmitter<{ [key: string]: ServerEvent }>>
|
||||
export type ServerConnectionStatus = "connecting" | "connected" | "reconnecting"
|
||||
type CurrentEventMap = { [Type in OpenCodeEvent["type"]]: Extract<OpenCodeEvent, { type: Type }> }
|
||||
type CurrentEventEmitter = ReturnType<typeof createGlobalEmitter<CurrentEventMap>>
|
||||
export type ServerConnectionStatus = ClientConnectionStatus
|
||||
type ServerSDKBase = {
|
||||
server: ServerConnection.Any
|
||||
scope: ServerScope
|
||||
@@ -101,231 +30,51 @@ type ServerSDKBase = {
|
||||
attempt: Accessor<number>
|
||||
error: Accessor<string | undefined>
|
||||
}
|
||||
event: {
|
||||
eventByDir: {
|
||||
on: ServerEventEmitter["on"]
|
||||
listen: ServerEventEmitter["listen"]
|
||||
}
|
||||
event: {
|
||||
on: CurrentEventEmitter["on"]
|
||||
listen: CurrentEventEmitter["listen"]
|
||||
}
|
||||
}
|
||||
|
||||
function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerScope): ServerSDKBase {
|
||||
const platform = usePlatform()
|
||||
const abort = new AbortController()
|
||||
const api = createApiForServer({ server: server.http, fetch: platform.fetch })
|
||||
const dirEmitter = createGlobalEmitter<{ [key: string]: ServerEvent }>()
|
||||
const emitter = createGlobalEmitter<CurrentEventMap>()
|
||||
|
||||
const eventFetch = (() => {
|
||||
if (!platform.fetch || !server) return
|
||||
try {
|
||||
const url = new URL(server.http.url)
|
||||
const loopback = url.hostname === "localhost" || url.hostname === "127.0.0.1" || url.hostname === "::1"
|
||||
if (url.protocol === "http:" && !loopback) return platform.fetch
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
})()
|
||||
|
||||
const eventApi = createApiForServer({ server: server.http, fetch: eventFetch })
|
||||
const emitter = createGlobalEmitter<{
|
||||
[key: string]: ServerEvent
|
||||
}>()
|
||||
|
||||
type Queued = QueuedServerEvent
|
||||
const FLUSH_FRAME_MS = 16
|
||||
const STREAM_YIELD_MS = 8
|
||||
const CONNECT_TIMEOUT_MS = 2_000
|
||||
const RECONNECT_DELAY_MS = 1_000
|
||||
|
||||
let queue: Queued[] = []
|
||||
let buffer: Queued[] = []
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
let last = 0
|
||||
|
||||
function flush() {
|
||||
if (timer) clearTimeout(timer)
|
||||
timer = undefined
|
||||
|
||||
if (queue.length === 0) return
|
||||
|
||||
const events = queue
|
||||
queue = buffer
|
||||
buffer = events
|
||||
queue.length = 0
|
||||
|
||||
last = Date.now()
|
||||
const output = coalesceServerEvents(events)
|
||||
batch(() => {
|
||||
output.forEach((event) => emitter.emit(event.directory, event.payload))
|
||||
})
|
||||
|
||||
buffer.length = 0
|
||||
}
|
||||
|
||||
function schedule() {
|
||||
if (timer) return
|
||||
const elapsed = Date.now() - last
|
||||
timer = setTimeout(flush, Math.max(0, FLUSH_FRAME_MS - elapsed))
|
||||
}
|
||||
|
||||
function publish(event: OpenCodeEvent) {
|
||||
const directory = event.location?.directory ?? "global"
|
||||
if (enqueueServerEvent(queue, { directory, payload: adaptServerEvent(event) })) schedule()
|
||||
}
|
||||
|
||||
function wait(delay: number, signal: AbortSignal) {
|
||||
return new Promise<void>((resolve) => {
|
||||
const timer = setTimeout(done, delay)
|
||||
signal.addEventListener("abort", done, { once: true })
|
||||
function done() {
|
||||
clearTimeout(timer)
|
||||
signal.removeEventListener("abort", done)
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
}
|
||||
let attempt: AbortController | undefined
|
||||
let run: Promise<void> | undefined
|
||||
let started = false
|
||||
let generation = 0
|
||||
const [connection, setConnection] = createStore<{
|
||||
status: ServerConnectionStatus
|
||||
attempt: number
|
||||
error?: string
|
||||
}>({ status: "connecting", attempt: 0 })
|
||||
|
||||
async function connect(signal: AbortSignal): Promise<{ error: unknown; connectedAt: number | undefined }> {
|
||||
let connectedAt: number | undefined
|
||||
|
||||
// Bound the initial handshake and tie this request to the stream lifetime.
|
||||
const request = new AbortController()
|
||||
const cancel = () => request.abort(signal.reason)
|
||||
const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), CONNECT_TIMEOUT_MS)
|
||||
signal.addEventListener("abort", cancel, { once: true })
|
||||
|
||||
try {
|
||||
// Open the event stream and validate its initial handshake.
|
||||
const iterator = eventApi.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]()
|
||||
const first = await iterator.next()
|
||||
|
||||
if (signal.aborted) return { error: undefined, connectedAt }
|
||||
if (first.done) {
|
||||
const error =
|
||||
request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected")
|
||||
return { error, connectedAt }
|
||||
}
|
||||
if (first.value.type !== "server.connected")
|
||||
return { error: new Error("Event stream did not start with server.connected"), connectedAt }
|
||||
|
||||
// Publish the connected state before forwarding live events.
|
||||
clearTimeout(timeout)
|
||||
publish(first.value)
|
||||
connectedAt = Date.now()
|
||||
setConnection({ status: "connected", attempt: 0, error: undefined })
|
||||
|
||||
// Forward events until the stream closes or this connection is cancelled.
|
||||
let yielded = Date.now()
|
||||
while (!signal.aborted) {
|
||||
const event = await iterator.next()
|
||||
if (signal.aborted) return { error: undefined, connectedAt }
|
||||
if (event.done) return { error: new Error("Event stream disconnected"), connectedAt }
|
||||
publish(event.value)
|
||||
if (Date.now() - yielded < STREAM_YIELD_MS) continue
|
||||
yielded = Date.now()
|
||||
await wait(0, signal)
|
||||
}
|
||||
return { error: undefined, connectedAt }
|
||||
} catch (error) {
|
||||
return { error, connectedAt }
|
||||
} finally {
|
||||
request.abort()
|
||||
clearTimeout(timeout)
|
||||
signal.removeEventListener("abort", cancel)
|
||||
}
|
||||
}
|
||||
|
||||
async function runStream(active: number) {
|
||||
let retries = 0
|
||||
// oxlint-disable-next-line no-unmodified-loop-condition -- stop() changes the lifecycle flags and aborts the active request
|
||||
while (!abort.signal.aborted && started && generation === active) {
|
||||
setConnection({ status: retries === 0 ? "connecting" : "reconnecting", attempt: retries, error: undefined })
|
||||
const controller = new AbortController()
|
||||
attempt = controller
|
||||
const onAbort = () => controller.abort()
|
||||
abort.signal.addEventListener("abort", onAbort)
|
||||
const result = await connect(controller.signal)
|
||||
abort.signal.removeEventListener("abort", onAbort)
|
||||
|
||||
if (abort.signal.aborted || !started || generation !== active) {
|
||||
if (attempt === controller) attempt = undefined
|
||||
return
|
||||
}
|
||||
if (result.connectedAt !== undefined && Date.now() - result.connectedAt >= 1_000) retries = 0
|
||||
retries += 1
|
||||
const message =
|
||||
result.error === undefined
|
||||
? undefined
|
||||
: result.error instanceof Error
|
||||
? result.error.message
|
||||
: String(result.error)
|
||||
console.info("[global-sdk] event stream disconnected", {
|
||||
url: server.http.url,
|
||||
fetch: eventFetch ? "platform" : "webview",
|
||||
attempt: retries,
|
||||
error: message,
|
||||
})
|
||||
setConnection({ status: "reconnecting", attempt: retries, error: message })
|
||||
await wait(RECONNECT_DELAY_MS, controller.signal)
|
||||
if (attempt === controller) attempt = undefined
|
||||
}
|
||||
}
|
||||
|
||||
function start() {
|
||||
if (started) return run
|
||||
started = true
|
||||
const active = ++generation
|
||||
const previous = run
|
||||
const current = (async () => {
|
||||
if (previous) await previous
|
||||
await runStream(active)
|
||||
})().finally(() => {
|
||||
if (run !== current) return
|
||||
run = undefined
|
||||
flush()
|
||||
})
|
||||
run = current
|
||||
return run
|
||||
}
|
||||
|
||||
function stop() {
|
||||
started = false
|
||||
generation++
|
||||
attempt?.abort()
|
||||
}
|
||||
|
||||
onMount(() => {
|
||||
makeEventListener(window, "pagehide", stop)
|
||||
makeEventListener(window, "pageshow", (event) => resumeStreamAfterPageShow(event, start))
|
||||
void start()
|
||||
const connection = createClientConnection(api, {
|
||||
flushInterval: 16,
|
||||
pageLifecycle: true,
|
||||
onEvent(event) {
|
||||
emitter.emit(event.type, event)
|
||||
dirEmitter.emit(event.location?.directory ?? "global", adaptServerEvent(event))
|
||||
},
|
||||
log: {
|
||||
info(message, data) {
|
||||
if (message !== "event stream disconnected") return
|
||||
console.info("[global-sdk] event stream disconnected", { url: server.http.url, ...data })
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
onCleanup(() => {
|
||||
stop()
|
||||
abort.abort()
|
||||
if (timer) clearTimeout(timer)
|
||||
timer = undefined
|
||||
queue = []
|
||||
buffer = []
|
||||
dirEmitter.clear()
|
||||
emitter.clear()
|
||||
})
|
||||
|
||||
const api = createApiForServer({ server: server.http, fetch: platform.fetch })
|
||||
|
||||
return {
|
||||
server,
|
||||
scope,
|
||||
url: server.http.url,
|
||||
api,
|
||||
connection: {
|
||||
status: () => connection.status,
|
||||
attempt: () => connection.attempt,
|
||||
error: () => connection.error,
|
||||
connection,
|
||||
eventByDir: {
|
||||
on: dirEmitter.on.bind(dirEmitter),
|
||||
listen: dirEmitter.listen.bind(dirEmitter),
|
||||
},
|
||||
event: {
|
||||
on: emitter.on.bind(emitter),
|
||||
@@ -365,7 +114,7 @@ export type DirectorySDK = {
|
||||
function createDirSdkContext(directory: string, serverSDK: ServerSDKBase): DirectorySDK {
|
||||
const emitter = createGlobalEmitter<SDKEventMap>()
|
||||
|
||||
const unsub = serverSDK.event.on(directory, (event) => {
|
||||
const unsub = serverSDK.eventByDir.on(directory, (event) => {
|
||||
emitter.emit(event.type, event)
|
||||
})
|
||||
onCleanup(unsub)
|
||||
|
||||
@@ -54,7 +54,7 @@ import { createServerSession, type ServerSession } from "./server-session"
|
||||
import { createCatalogSync } from "./server-sync/catalog"
|
||||
import { createConnectionSync } from "./server-sync/connection"
|
||||
import { usePlatform } from "./platform"
|
||||
import { useServer } from "./server"
|
||||
import { useData, useServer } from "./server"
|
||||
|
||||
export function shouldRefreshWorkspaceSessions(event: ServerEvent) {
|
||||
const type = event.current?.type ?? event.type
|
||||
@@ -568,7 +568,7 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) {
|
||||
return event
|
||||
}
|
||||
|
||||
const unsub = serverSDK.event.listen((e) => {
|
||||
const unsub = serverSDK.eventByDir.listen((e) => {
|
||||
const directory = e.name
|
||||
const key = directoryKey(directory)
|
||||
const event = e.details
|
||||
|
||||
@@ -22,3 +22,8 @@ export const { use: useServer, provider: ServerProvider } = createSimpleContext(
|
||||
}
|
||||
},
|
||||
})
|
||||
|
||||
export const useData = () => {
|
||||
const server = useServer()
|
||||
return server.ctx.data
|
||||
}
|
||||
|
||||
@@ -101,7 +101,7 @@ import { diffs as list } from "@/utils/diffs"
|
||||
import { Persist, persisted } from "@/utils/persist"
|
||||
import { extractPromptFromParts } from "@/utils/prompt"
|
||||
import { formatServerError, isLocalSessionNotFoundError, isSessionNotFoundError } from "@/utils/server-errors"
|
||||
import { legacySessionHref, requireServerKey, sessionHref } from "@/utils/session-route"
|
||||
import { requireServerKey, sessionHref } from "@/utils/session-route"
|
||||
import { useUsageExceededDialogs } from "./session/usage-exceeded-dialogs"
|
||||
import { createSessionLineage } from "./session/session-lineage"
|
||||
|
||||
@@ -2128,9 +2128,12 @@ export default function Page() {
|
||||
const id = controller.data.parentID()
|
||||
if (!id) return
|
||||
navigate(
|
||||
controller.identity.params.serverKey
|
||||
? sessionHref(requireServerKey(controller.identity.params.serverKey), id)
|
||||
: legacySessionHref(sdk().directory, id),
|
||||
sessionHref(
|
||||
controller.identity.params.serverKey
|
||||
? requireServerKey(controller.identity.params.serverKey)
|
||||
: ServerConnection.key(serverSDK.server),
|
||||
id,
|
||||
),
|
||||
)
|
||||
},
|
||||
setPromptRef: (el) => {
|
||||
|
||||
@@ -15,13 +15,15 @@ import { useSDK } from "@/context/sdk"
|
||||
import { useSync } from "@/context/sync"
|
||||
import { useTabs } from "@/context/tabs"
|
||||
import type { SessionController } from "@/pages/session/session-controller"
|
||||
import { legacySessionHref, requireServerKey, sessionHref } from "@/utils/session-route"
|
||||
import { requireServerKey, sessionHref } from "@/utils/session-route"
|
||||
import { useServerSDK } from "@/context/server-sdk"
|
||||
import { sessionTitle } from "@/utils/session-title"
|
||||
import { downloadSessionExport, fetchSessionExport, sessionExportFilename } from "@/utils/session-export"
|
||||
import { showToast } from "@/utils/toast"
|
||||
import { timelineChildTitle, timelineRemovedSessionIDs } from "./controller-projection"
|
||||
import { createTimelineProjection } from "./projection"
|
||||
import { useServer } from "@/context/server"
|
||||
import { ServerConnection } from "@/context/servers"
|
||||
|
||||
const emptyMessages: Message[] = []
|
||||
const emptyParts: Part[] = []
|
||||
@@ -46,6 +48,7 @@ export function createTimelineController(input: {
|
||||
}) {
|
||||
const navigate = useNavigate()
|
||||
const sdk = useSDK()
|
||||
const serverSDK = useServerSDK()
|
||||
const sync = useSync()
|
||||
const server = useServer()
|
||||
const settings = useSettings()
|
||||
@@ -146,19 +149,22 @@ export function createTimelineController(input: {
|
||||
if (!id || pending.unshare || !shareEnabled()) return
|
||||
}
|
||||
const href = (id: string) =>
|
||||
input.session.identity.params.serverKey
|
||||
? sessionHref(requireServerKey(input.session.identity.params.serverKey), id)
|
||||
: legacySessionHref(sdk().directory, id)
|
||||
sessionHref(
|
||||
input.session.identity.params.serverKey
|
||||
? requireServerKey(input.session.identity.params.serverKey)
|
||||
: ServerConnection.key(serverSDK.server),
|
||||
id,
|
||||
)
|
||||
const navigateAfterRemoval = (id: string, parent?: string, next?: string) => {
|
||||
if (input.session.identity.params.id !== id) return
|
||||
if (parent) return navigate(href(parent))
|
||||
if (next) return navigate(href(next))
|
||||
if (input.session.identity.params.serverKey)
|
||||
return tabs.newDraft({
|
||||
server: requireServerKey(input.session.identity.params.serverKey),
|
||||
directory: sdk().directory,
|
||||
})
|
||||
navigate(`/${input.session.identity.params.dir}/session`)
|
||||
return tabs.newDraft({
|
||||
server: input.session.identity.params.serverKey
|
||||
? requireServerKey(input.session.identity.params.serverKey)
|
||||
: ServerConnection.key(serverSDK.server),
|
||||
directory: sdk().directory,
|
||||
})
|
||||
}
|
||||
const exportSession = async (id: string) => {
|
||||
try {
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
"./promise": "./src/promise/index.ts",
|
||||
"./promise/api": "./src/promise/api.ts",
|
||||
"./service": "./src/promise/service.ts",
|
||||
"./solid": "./src/solid/index.ts",
|
||||
"./effect": "./src/effect/index.ts",
|
||||
"./effect/api": "./src/effect/api.ts",
|
||||
"./effect/service": "./src/effect/service.ts"
|
||||
@@ -36,11 +37,15 @@
|
||||
"@opencode-ai/protocol": "workspace:*"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"effect": "4.0.0-beta.101"
|
||||
"effect": "4.0.0-beta.101",
|
||||
"solid-js": ">=1.9.0"
|
||||
},
|
||||
"peerDependenciesMeta": {
|
||||
"effect": {
|
||||
"optional": true
|
||||
},
|
||||
"solid-js": {
|
||||
"optional": true
|
||||
}
|
||||
},
|
||||
"devDependencies": {
|
||||
@@ -49,6 +54,7 @@
|
||||
"@tsconfig/bun": "catalog:",
|
||||
"@types/bun": "catalog:",
|
||||
"@typescript/native-preview": "catalog:",
|
||||
"effect": "catalog:"
|
||||
"effect": "catalog:",
|
||||
"solid-js": "catalog:"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,269 @@
|
||||
import { batch, onCleanup, onMount } from "solid-js"
|
||||
import { createStore } from "solid-js/store"
|
||||
import type { OpenCodeClient, OpenCodeEvent } from "../promise"
|
||||
|
||||
export type ClientConnectionStatus = "connected" | "connecting" | "reconnecting"
|
||||
export type ClientConnectionEvent = {
|
||||
readonly type: "client.connection"
|
||||
readonly created: number
|
||||
readonly data: {
|
||||
readonly status: "connecting" | "connected" | "disconnected" | "reconnecting"
|
||||
readonly attempt: number
|
||||
readonly error?: string
|
||||
}
|
||||
}
|
||||
|
||||
export type ClientConnectionOptions = {
|
||||
readonly reconnect?: (signal: AbortSignal) => Promise<OpenCodeClient>
|
||||
readonly onEvent: (event: OpenCodeEvent) => void
|
||||
readonly flushInterval?: number
|
||||
readonly pageLifecycle?: boolean
|
||||
readonly log?: {
|
||||
readonly debug?: (message: string, data?: Readonly<Record<string, unknown>>) => void
|
||||
readonly info?: (message: string, data?: Readonly<Record<string, unknown>>) => void
|
||||
}
|
||||
}
|
||||
|
||||
const connectTimeout = 2_000
|
||||
const reconnectDelay = 1_000
|
||||
const connectionHistoryLimit = 50
|
||||
|
||||
type CurrentDelta = Extract<
|
||||
OpenCodeEvent,
|
||||
{ type: "session.text.delta" | "session.reasoning.delta" | "session.tool.input.delta" | "session.compaction.delta" }
|
||||
>
|
||||
|
||||
export function coalesceClientEvents(events: OpenCodeEvent[]) {
|
||||
return events.reduce<OpenCodeEvent[]>((output, event) => {
|
||||
const current = currentDelta(event)
|
||||
const previous = output[output.length - 1]
|
||||
const prior = currentDelta(previous)
|
||||
if (
|
||||
!current ||
|
||||
!prior ||
|
||||
previous?.location?.directory !== event.location?.directory ||
|
||||
currentDeltaKey(prior) !== currentDeltaKey(current)
|
||||
) {
|
||||
output.push(event)
|
||||
return output
|
||||
}
|
||||
const fragment = currentDeltaFragment(prior) + currentDeltaFragment(current)
|
||||
output[output.length - 1] = {
|
||||
...current,
|
||||
data:
|
||||
current.type === "session.compaction.delta"
|
||||
? { ...current.data, text: fragment }
|
||||
: { ...current.data, delta: fragment },
|
||||
} as CurrentDelta
|
||||
return output
|
||||
}, [])
|
||||
}
|
||||
|
||||
function currentDelta(event: OpenCodeEvent | undefined): CurrentDelta | undefined {
|
||||
if (
|
||||
event?.type === "session.text.delta" ||
|
||||
event?.type === "session.reasoning.delta" ||
|
||||
event?.type === "session.tool.input.delta" ||
|
||||
event?.type === "session.compaction.delta"
|
||||
)
|
||||
return event
|
||||
}
|
||||
|
||||
function currentDeltaKey(event: CurrentDelta) {
|
||||
if (event.type === "session.tool.input.delta")
|
||||
return `${event.type}:${event.data.sessionID}:${event.data.assistantMessageID}:${event.data.id}`
|
||||
if (event.type === "session.compaction.delta") return `${event.type}:${event.data.sessionID}`
|
||||
return `${event.type}:${event.data.sessionID}:${event.data.assistantMessageID}:${event.data.ordinal}`
|
||||
}
|
||||
|
||||
function currentDeltaFragment(event: CurrentDelta) {
|
||||
return event.type === "session.compaction.delta" ? event.data.text : event.data.delta
|
||||
}
|
||||
|
||||
export function createClientConnection(initialApi: OpenCodeClient, options: ClientConnectionOptions) {
|
||||
const abort = new AbortController()
|
||||
const history: ClientConnectionEvent[] = []
|
||||
const [connection, setConnection] = createStore<{
|
||||
status: ClientConnectionStatus
|
||||
attempt: number
|
||||
error?: string
|
||||
}>({ status: "connecting", attempt: 0 })
|
||||
let api = initialApi
|
||||
let pending: OpenCodeEvent[] = []
|
||||
let flushTimer: ReturnType<typeof setTimeout> | undefined
|
||||
let stream: AbortController | undefined
|
||||
let run: Promise<void> | undefined
|
||||
let started = false
|
||||
let generation = 0
|
||||
|
||||
function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) {
|
||||
history.push({ type: "client.connection", created: Date.now(), data: { status, attempt, error } })
|
||||
if (history.length > connectionHistoryLimit) history.shift()
|
||||
}
|
||||
|
||||
function publish(event: OpenCodeEvent) {
|
||||
pending.push(event)
|
||||
if (flushTimer) return
|
||||
flushTimer = setTimeout(() => {
|
||||
flushTimer = undefined
|
||||
const events = pending
|
||||
pending = []
|
||||
batch(() => coalesceClientEvents(events).forEach(options.onEvent))
|
||||
}, options.flushInterval ?? 10)
|
||||
}
|
||||
|
||||
async function connect(signal: AbortSignal, attempt: number) {
|
||||
let connectedAt: number | undefined
|
||||
const request = new AbortController()
|
||||
const cancel = () => request.abort(signal.reason)
|
||||
const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout)
|
||||
signal.addEventListener("abort", cancel, { once: true })
|
||||
|
||||
try {
|
||||
record(attempt === 0 ? "connecting" : "reconnecting", attempt)
|
||||
options.log?.info?.("event stream connecting", { attempt })
|
||||
const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]()
|
||||
const first = await iterator.next()
|
||||
if (signal.aborted) return { error: undefined, connectedAt }
|
||||
if (first.done)
|
||||
return {
|
||||
error: request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected"),
|
||||
connectedAt,
|
||||
}
|
||||
if (first.value.type !== "server.connected")
|
||||
return { error: new Error("Event stream did not start with server.connected"), connectedAt }
|
||||
|
||||
clearTimeout(timeout)
|
||||
record("connected", attempt)
|
||||
connectedAt = Date.now()
|
||||
options.log?.info?.("event stream connected")
|
||||
publish(first.value)
|
||||
setConnection({ status: "connected", attempt: 0, error: undefined })
|
||||
|
||||
while (!signal.aborted) {
|
||||
const event = await iterator.next()
|
||||
if (signal.aborted) return { error: undefined, connectedAt }
|
||||
if (event.done) return { error: new Error("Event stream disconnected"), connectedAt }
|
||||
if ("durable" in event.value)
|
||||
options.log?.debug?.("event", {
|
||||
type: event.value.type,
|
||||
aggregateID: event.value.durable.aggregateID,
|
||||
seq: event.value.durable.seq,
|
||||
})
|
||||
publish(event.value)
|
||||
}
|
||||
return { error: undefined, connectedAt }
|
||||
} catch (error) {
|
||||
return { error, connectedAt }
|
||||
} finally {
|
||||
request.abort()
|
||||
clearTimeout(timeout)
|
||||
signal.removeEventListener("abort", cancel)
|
||||
}
|
||||
}
|
||||
|
||||
async function runStream(active: number) {
|
||||
let attempt = 0
|
||||
while (!abort.signal.aborted && started && generation === active) {
|
||||
setConnection({ status: attempt === 0 ? "connecting" : "reconnecting", attempt, error: undefined })
|
||||
const controller = new AbortController()
|
||||
stream = controller
|
||||
const cancel = () => controller.abort(abort.signal.reason)
|
||||
abort.signal.addEventListener("abort", cancel)
|
||||
const result = await connect(controller.signal, attempt)
|
||||
abort.signal.removeEventListener("abort", cancel)
|
||||
if (abort.signal.aborted || !started || generation !== active) return
|
||||
if (result.connectedAt !== undefined && Date.now() - result.connectedAt >= reconnectDelay) attempt = 0
|
||||
attempt += 1
|
||||
const message = errorMessage(result.error)
|
||||
record("disconnected", attempt, message)
|
||||
options.log?.info?.("event stream disconnected", { attempt, error: message })
|
||||
setConnection({ status: "reconnecting", attempt, error: message })
|
||||
|
||||
if (options.reconnect) {
|
||||
const next = await options.reconnect(controller.signal).catch((error) => {
|
||||
if (!controller.signal.aborted)
|
||||
options.log?.info?.("server resolution failed", { attempt, error: errorMessage(error) })
|
||||
})
|
||||
if (abort.signal.aborted || controller.signal.aborted || !started || generation !== active) return
|
||||
if (next) {
|
||||
api = next
|
||||
if (attempt === 1) continue
|
||||
}
|
||||
}
|
||||
await wait(reconnectDelay, controller.signal)
|
||||
}
|
||||
}
|
||||
|
||||
function start() {
|
||||
if (started) return run
|
||||
started = true
|
||||
const active = ++generation
|
||||
const previous = run
|
||||
const current = (async () => {
|
||||
if (previous) await previous
|
||||
await runStream(active)
|
||||
})().finally(() => {
|
||||
if (run !== current) return
|
||||
run = undefined
|
||||
})
|
||||
run = current
|
||||
return run
|
||||
}
|
||||
|
||||
function stop() {
|
||||
started = false
|
||||
generation += 1
|
||||
stream?.abort()
|
||||
}
|
||||
|
||||
onMount(() => {
|
||||
if (options.pageLifecycle) {
|
||||
const pagehide = () => stop()
|
||||
const pageshow = (event: PageTransitionEvent) => {
|
||||
if (event.persisted) void start()
|
||||
}
|
||||
window.addEventListener("pagehide", pagehide)
|
||||
window.addEventListener("pageshow", pageshow)
|
||||
onCleanup(() => {
|
||||
window.removeEventListener("pagehide", pagehide)
|
||||
window.removeEventListener("pageshow", pageshow)
|
||||
})
|
||||
}
|
||||
void start()
|
||||
})
|
||||
|
||||
onCleanup(() => {
|
||||
stop()
|
||||
abort.abort()
|
||||
if (flushTimer) clearTimeout(flushTimer)
|
||||
pending = []
|
||||
})
|
||||
|
||||
return {
|
||||
status: () => connection.status,
|
||||
attempt: () => connection.attempt,
|
||||
error: () => connection.error,
|
||||
internal: {
|
||||
history: () => history.slice(),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function errorMessage(error: unknown) {
|
||||
if (error === undefined) return undefined
|
||||
if (error instanceof Error) return error.message
|
||||
return String(error)
|
||||
}
|
||||
|
||||
function wait(delay: number, signal: AbortSignal) {
|
||||
return new Promise<void>((resolve) => {
|
||||
const timer = setTimeout(done, delay)
|
||||
signal.addEventListener("abort", done, { once: true })
|
||||
function done() {
|
||||
clearTimeout(timer)
|
||||
signal.removeEventListener("abort", done)
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,2 @@
|
||||
export * from "./data"
|
||||
export * from "./connection"
|
||||
@@ -0,0 +1,44 @@
|
||||
import { describe, expect, test } from "bun:test"
|
||||
import type { OpenCodeEvent } from "../src/promise"
|
||||
import { coalesceClientEvents } from "../src/solid/connection"
|
||||
|
||||
describe("coalesceClientEvents", () => {
|
||||
const delta = (id: string, value: string, ordinal = 0) =>
|
||||
({
|
||||
id,
|
||||
created: 1,
|
||||
type: "session.text.delta",
|
||||
location: { directory: "/repo" },
|
||||
data: { sessionID: "ses", assistantMessageID: "msg", ordinal, delta: value },
|
||||
}) as OpenCodeEvent
|
||||
|
||||
test("merges adjacent deltas for the same stream", () => {
|
||||
const result = coalesceClientEvents([delta("evt_1", "hello "), delta("evt_2", "world")])
|
||||
expect(result).toHaveLength(1)
|
||||
expect(result[0]).toMatchObject({ id: "evt_2", data: { delta: "hello world" } })
|
||||
})
|
||||
|
||||
test("coalesces tool input deltas by tool ID", () => {
|
||||
const current = (eventID: string, id: string, value: string) =>
|
||||
({
|
||||
id: eventID,
|
||||
created: 1,
|
||||
type: "session.tool.input.delta",
|
||||
location: { directory: "/repo" },
|
||||
data: { sessionID: "ses", assistantMessageID: "msg", id, delta: value },
|
||||
}) as OpenCodeEvent
|
||||
const result = coalesceClientEvents([
|
||||
current("evt_1", "call_1", "{"),
|
||||
current("evt_2", "call_1", "}"),
|
||||
current("evt_3", "call_2", "[]"),
|
||||
])
|
||||
expect(result).toHaveLength(2)
|
||||
expect(result[0]).toMatchObject({ id: "evt_2", data: { id: "call_1", delta: "{}" } })
|
||||
expect(result[1]).toMatchObject({ id: "evt_3", data: { id: "call_2", delta: "[]" } })
|
||||
})
|
||||
|
||||
test("preserves boundaries between distinct delta streams", () => {
|
||||
const events = [delta("evt_1", "a"), delta("evt_2", "b", 1), delta("evt_3", "c")]
|
||||
expect(coalesceClientEvents(events).map((event) => event.id)).toEqual(["evt_1", "evt_2", "evt_3"])
|
||||
})
|
||||
})
|
||||
@@ -41,7 +41,7 @@ export async function startBackgroundCli(logger: Logger) {
|
||||
? join(app.getPath("userData"), "opencode", "service-local.json")
|
||||
: undefined,
|
||||
version: cli.version,
|
||||
command: [...cli.command, "serve", "--service"],
|
||||
command: [...cli.command, "serve", "--service", ...(isolated ? ["--port", "0"] : [])],
|
||||
onStart: (reason, previousVersion) => logger.log("v2 CLI background service starting", { reason, previousVersion }),
|
||||
})
|
||||
if (service.auth?.type !== "basic") throw new Error("V2 CLI background service did not provide authentication")
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
import { Session } from "@opencode-ai/core/session"
|
||||
import { Agent } from "@opencode-ai/core/agent"
|
||||
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
|
||||
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
|
||||
import { SessionTransfer } from "@opencode-ai/core/session/transfer"
|
||||
import { InstructionEntry } from "@opencode-ai/core/session/instruction-entry"
|
||||
import { DateTime, Effect, Stream } from "effect"
|
||||
@@ -94,24 +91,14 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
|
||||
.handle(
|
||||
"session.create",
|
||||
Effect.fn(function* (ctx) {
|
||||
const location = ctx.payload.location ?? { directory: AbsolutePath.make(process.cwd()) }
|
||||
const selectedAgent = ctx.payload.model ? undefined : ctx.payload.agent
|
||||
const agentModel = selectedAgent
|
||||
? yield* Effect.gen(function* () {
|
||||
const plugins = yield* PluginSupervisor.Service
|
||||
yield* plugins.flush
|
||||
const agents = yield* Agent.Service
|
||||
return (yield* agents.get(selectedAgent))?.model
|
||||
}).pipe(Effect.provide(LocationServiceMap.Service.get(location)))
|
||||
: undefined
|
||||
return {
|
||||
data: yield* session
|
||||
.create({
|
||||
id: ctx.payload.id,
|
||||
title: ctx.payload.title,
|
||||
agent: ctx.payload.agent,
|
||||
model: ctx.payload.model ?? agentModel,
|
||||
location,
|
||||
model: ctx.payload.model,
|
||||
location: ctx.payload.location ?? { directory: AbsolutePath.make(process.cwd()) },
|
||||
})
|
||||
.pipe(Effect.orDie),
|
||||
}
|
||||
|
||||
@@ -1,64 +0,0 @@
|
||||
import fs from "node:fs/promises"
|
||||
import path from "node:path"
|
||||
import { expect } from "bun:test"
|
||||
import { Effect } from "effect"
|
||||
import { HttpServer } from "effect/unstable/http"
|
||||
import { tmpdir } from "../../core/test/fixture/tmpdir"
|
||||
import { it } from "../../core/test/lib/effect"
|
||||
import { ServerProcess } from "../src/process"
|
||||
|
||||
it.live("waits for plugin initialization before applying the selected agent model", () =>
|
||||
Effect.acquireUseRelease(
|
||||
Effect.promise(() => tmpdir("opencode-session-endpoint-")),
|
||||
(tmp) =>
|
||||
Effect.gen(function* () {
|
||||
yield* Effect.promise(() =>
|
||||
fs.writeFile(
|
||||
path.join(tmp.path, "opencode.json"),
|
||||
JSON.stringify({
|
||||
agents: {
|
||||
modelprobe: {
|
||||
description: "Model resolution probe",
|
||||
mode: "primary",
|
||||
model: "opencode/nemotron-3.5-lightning-free",
|
||||
},
|
||||
},
|
||||
}),
|
||||
),
|
||||
)
|
||||
const server = yield* ServerProcess.start<never, never>({
|
||||
hostname: "127.0.0.1",
|
||||
port: 0,
|
||||
password: "secret",
|
||||
app: { version: "test-version" },
|
||||
database: { path: ":memory:" },
|
||||
config: { directory: tmp.path },
|
||||
fs: { filewatcher: false },
|
||||
})
|
||||
const response = yield* Effect.promise(() =>
|
||||
fetch(new URL("/api/session", HttpServer.formatAddress(server.address)), {
|
||||
method: "POST",
|
||||
headers: {
|
||||
authorization: `Basic ${btoa("opencode:secret")}`,
|
||||
"content-type": "application/json",
|
||||
},
|
||||
body: JSON.stringify({ agent: "modelprobe", location: { directory: tmp.path } }),
|
||||
}),
|
||||
)
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
const body: unknown = yield* Effect.promise(() => response.json())
|
||||
if (!isRecord(body) || !isRecord(body["data"])) throw new Error("Expected a created session")
|
||||
expect(body["data"]["model"]).toEqual({
|
||||
providerID: "opencode",
|
||||
id: "nemotron-3.5-lightning-free",
|
||||
variant: "default",
|
||||
})
|
||||
}),
|
||||
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
|
||||
),
|
||||
)
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value)
|
||||
}
|
||||
@@ -1,177 +1,39 @@
|
||||
import type { OpenCodeClient, OpenCodeEvent } from "@opencode-ai/client"
|
||||
import { createClientConnection } from "@opencode-ai/client/solid"
|
||||
import { createGlobalEmitter } from "@solid-primitives/event-bus"
|
||||
import { batch, onCleanup, onMount } from "solid-js"
|
||||
import { createStore } from "solid-js/store"
|
||||
import { errorMessage } from "../util/error"
|
||||
import { onCleanup } from "solid-js"
|
||||
import { createSimpleContext } from "./helper"
|
||||
import { useLog } from "./log"
|
||||
|
||||
export type ClientConnectionStatus = "connected" | "connecting" | "reconnecting"
|
||||
export type ClientConnectionEvent = {
|
||||
readonly type: "client.connection"
|
||||
readonly created: number
|
||||
readonly data: {
|
||||
readonly status: "connecting" | "connected" | "disconnected" | "reconnecting"
|
||||
readonly attempt: number
|
||||
readonly error?: string
|
||||
}
|
||||
}
|
||||
|
||||
type ManagedService = {
|
||||
reconnect: (signal: AbortSignal) => Promise<{ api: OpenCodeClient }>
|
||||
restart: () => Promise<void>
|
||||
}
|
||||
|
||||
type ClientEventMap = { [Type in OpenCodeEvent["type"]]: Extract<OpenCodeEvent, { type: Type }> }
|
||||
const connectTimeout = 2_000
|
||||
const connectionHistoryLimit = 50
|
||||
const eventFlushInterval = 10
|
||||
|
||||
export const { use: useClient, provider: ClientProvider } = createSimpleContext({
|
||||
name: "Client",
|
||||
init: (props: { api: OpenCodeClient; service?: ManagedService }) => {
|
||||
const log = useLog({ component: "client" })
|
||||
const abort = new AbortController()
|
||||
const history: ClientConnectionEvent[] = []
|
||||
let api = props.api
|
||||
const service = props.service
|
||||
const events = createGlobalEmitter<ClientEventMap>()
|
||||
let pending: OpenCodeEvent[] = []
|
||||
let flushTimer: ReturnType<typeof setTimeout> | undefined
|
||||
const [connection, setConnection] = createStore<{
|
||||
status: ClientConnectionStatus
|
||||
attempt: number
|
||||
error?: string
|
||||
}>({
|
||||
status: "connecting",
|
||||
attempt: 0,
|
||||
})
|
||||
let stream: AbortController | undefined
|
||||
let api = props.api
|
||||
|
||||
function record(status: ClientConnectionEvent["data"]["status"], attempt: number, error?: string) {
|
||||
history.push({ type: "client.connection", created: Date.now(), data: { status, attempt, error } })
|
||||
if (history.length > connectionHistoryLimit) history.shift()
|
||||
}
|
||||
|
||||
function flushEvents() {
|
||||
flushTimer = undefined
|
||||
const queued = pending
|
||||
pending = []
|
||||
batch(() => queued.forEach((event) => events.emit(event.type, event)))
|
||||
}
|
||||
|
||||
function emit(event: OpenCodeEvent) {
|
||||
pending.push(event)
|
||||
if (flushTimer) return
|
||||
flushTimer = setTimeout(flushEvents, eventFlushInterval)
|
||||
}
|
||||
|
||||
async function connect(signal: AbortSignal, attempt: number) {
|
||||
let connectedAt: number | undefined
|
||||
|
||||
// Bound the initial handshake and tie this request to the stream lifetime.
|
||||
const request = new AbortController()
|
||||
const cancel = () => request.abort(signal.reason)
|
||||
const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout)
|
||||
signal.addEventListener("abort", cancel, { once: true })
|
||||
|
||||
try {
|
||||
// Open the event stream and validate its initial handshake.
|
||||
record(attempt === 0 ? "connecting" : "reconnecting", attempt)
|
||||
log.info("event stream connecting", { attempt })
|
||||
|
||||
const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]()
|
||||
const first = await iterator.next()
|
||||
|
||||
if (signal.aborted) return { error: undefined, connectedAt }
|
||||
if (first.done) {
|
||||
const error =
|
||||
request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected")
|
||||
return { error, connectedAt }
|
||||
}
|
||||
if (first.value.type !== "server.connected")
|
||||
return { error: new Error("Event stream did not start with server.connected"), connectedAt }
|
||||
|
||||
// Publish the connected state before forwarding live events.
|
||||
clearTimeout(timeout)
|
||||
record("connected", attempt)
|
||||
connectedAt = Date.now()
|
||||
log.info("event stream connected")
|
||||
emit(first.value)
|
||||
setConnection({ status: "connected", attempt: 0, error: undefined })
|
||||
|
||||
// Forward events until the stream closes or this connection is cancelled.
|
||||
while (!signal.aborted) {
|
||||
const event = await iterator.next()
|
||||
|
||||
if (signal.aborted) return { error: undefined, connectedAt }
|
||||
if (event.done) return { error: new Error("Event stream disconnected"), connectedAt }
|
||||
|
||||
if ("durable" in event.value)
|
||||
log.debug("event", {
|
||||
type: event.value.type,
|
||||
aggregateID: event.value.durable.aggregateID,
|
||||
seq: event.value.durable.seq,
|
||||
})
|
||||
|
||||
emit(event.value)
|
||||
}
|
||||
|
||||
return { error: undefined, connectedAt }
|
||||
} catch (error) {
|
||||
return { error, connectedAt }
|
||||
} finally {
|
||||
request.abort()
|
||||
clearTimeout(timeout)
|
||||
signal.removeEventListener("abort", cancel)
|
||||
}
|
||||
}
|
||||
|
||||
function start() {
|
||||
stream?.abort()
|
||||
const controller = new AbortController()
|
||||
stream = controller
|
||||
void (async () => {
|
||||
let attempt = 0
|
||||
while (!abort.signal.aborted && !controller.signal.aborted) {
|
||||
const result = await connect(controller.signal, attempt)
|
||||
if (abort.signal.aborted || controller.signal.aborted) return
|
||||
if (result.connectedAt !== undefined && Date.now() - result.connectedAt >= 1_000) attempt = 0
|
||||
attempt += 1
|
||||
const message = errorMessage(result.error)
|
||||
record("disconnected", attempt, message)
|
||||
log.info("event stream disconnected", {
|
||||
attempt,
|
||||
error: message,
|
||||
})
|
||||
setConnection({ status: "reconnecting", attempt, error: message })
|
||||
// Re-resolve the transport before retrying: the server may have
|
||||
// moved (service restarted on a new port) or need starting. Static
|
||||
// transports (--server, standalone) resolve to the same address.
|
||||
if (props.service) {
|
||||
const next = await props.service.reconnect(controller.signal).catch((error) => {
|
||||
if (!controller.signal.aborted)
|
||||
log.info("server resolution failed", {
|
||||
attempt,
|
||||
error: errorMessage(error),
|
||||
})
|
||||
})
|
||||
if (abort.signal.aborted || controller.signal.aborted) return
|
||||
if (next) {
|
||||
api = next.api
|
||||
if (attempt === 1) continue
|
||||
}
|
||||
const connection = createClientConnection(api, {
|
||||
reconnect: service
|
||||
? async (signal) => {
|
||||
api = (await service.reconnect(signal)).api
|
||||
return api
|
||||
}
|
||||
await wait(1_000, controller.signal)
|
||||
}
|
||||
})()
|
||||
}
|
||||
: undefined,
|
||||
onEvent(event) {
|
||||
events.emit(event.type, event)
|
||||
},
|
||||
log,
|
||||
})
|
||||
|
||||
onMount(start)
|
||||
onCleanup(() => {
|
||||
abort.abort()
|
||||
stream?.abort()
|
||||
if (flushTimer) clearTimeout(flushTimer)
|
||||
pending = []
|
||||
events.clear()
|
||||
})
|
||||
|
||||
@@ -183,35 +45,8 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
|
||||
on: events.on,
|
||||
listen: events.listen,
|
||||
},
|
||||
connection: {
|
||||
status() {
|
||||
return connection.status
|
||||
},
|
||||
attempt() {
|
||||
return connection.attempt
|
||||
},
|
||||
error() {
|
||||
return connection.error
|
||||
},
|
||||
internal: {
|
||||
history() {
|
||||
return history.slice()
|
||||
},
|
||||
},
|
||||
},
|
||||
restart: props.service?.restart,
|
||||
connection,
|
||||
restart: service?.restart,
|
||||
}
|
||||
},
|
||||
})
|
||||
|
||||
function wait(delay: number, signal: AbortSignal) {
|
||||
return new Promise<void>((resolve) => {
|
||||
const timer = setTimeout(done, delay)
|
||||
signal.addEventListener("abort", done, { once: true })
|
||||
function done() {
|
||||
clearTimeout(timer)
|
||||
signal.removeEventListener("abort", done)
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -37,7 +37,6 @@ import type { Plugin } from "@opencode-ai/plugin/tui"
|
||||
import { createStore, produce, reconcile } from "solid-js/store"
|
||||
import { createSimpleContext } from "./helper"
|
||||
import { useClient } from "./client"
|
||||
import { nonEmptyToolContent } from "../util/tool-display"
|
||||
import type { SessionInbox } from "@opencode-ai/schema/session-inbox"
|
||||
import { Worktree } from "@opencode-ai/schema/worktree"
|
||||
import { createEffect, createSignal, onCleanup } from "solid-js"
|
||||
|
||||
@@ -1873,15 +1873,14 @@ test("refreshes effective catalog data after catalog updates", async () => {
|
||||
|
||||
test("refreshes agents after agent updates", async () => {
|
||||
const events = createEventStream()
|
||||
let requests = 0
|
||||
let agentID = "build"
|
||||
const calls = createFetch((url) => {
|
||||
if (url.pathname !== "/api/agent") return
|
||||
requests++
|
||||
return json({
|
||||
location: { directory, project: { id: "proj_test", directory } },
|
||||
data: [
|
||||
{
|
||||
id: requests === 1 ? "build" : "reviewer",
|
||||
id: agentID,
|
||||
request: { headers: {}, body: {} },
|
||||
mode: "primary",
|
||||
hidden: false,
|
||||
@@ -1911,6 +1910,11 @@ test("refreshes agents after agent updates", async () => {
|
||||
|
||||
try {
|
||||
await wait(() => data.location.agent.list()?.[0]?.id === "build")
|
||||
await Bun.sleep(20)
|
||||
events.emit({ id: "evt_agent_unlocated", created: 0, type: "agent.updated", data: {} })
|
||||
await Bun.sleep(20)
|
||||
expect(data.location.agent.list()?.[0]?.id).toBe("build")
|
||||
agentID = "reviewer"
|
||||
emitEvent(events, { id: "evt_agent", created: 0, type: "agent.updated", data: {} })
|
||||
await wait(() => data.location.agent.list()?.[0]?.id === "reviewer")
|
||||
} finally {
|
||||
@@ -2801,7 +2805,7 @@ test("renders admitted prompts immediately and tracks them until promoted", asyn
|
||||
await mounted
|
||||
const received: string[] = []
|
||||
const unsubscribe = sync.listen((event) => received.push(event.name))
|
||||
emitEvent(events, {
|
||||
events.emit({
|
||||
id: "evt_admitted_1",
|
||||
created: 0,
|
||||
type: "session.inbox.enqueued",
|
||||
|
||||
Reference in New Issue
Block a user