Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 108 Packages Projects Releases Wiki Activity

refactor(app): simplify server sync lifecycle

Browse Source
This commit is contained in:
LukeParkerDev
2026-08-13 12:45:32 +10:00
parent d236368eae
commit 93f06f06c2
14 changed files with 119 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/e2e/utils/mock-server.ts
+5 -5
View File
@@ -47,7 +47,6 @@ type MockStreamWindow = Window & {
export async function mockOpenCodeServer(page: Page, config: MockServerConfig) {
const cursors = new Map<string, string>()
let nextCursor = 0
const streamPort = process.env.PLAYWRIGHT_SERVER_PORT ?? "4096"
await page.addInitScript(
({ port, retry }) => {
@@ -64,11 +63,12 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) {
host.__mockServerStream = {
push(payloads: unknown[]) {
const frames = payloads.map(frame)
if (!state.controller) {
const controller = state.controller
if (!controller) {
state.buffer.push(...frames)
return
}
for (const item of frames) state.controller.enqueue(encoder.encode(item))
frames.forEach((item) => controller.enqueue(encoder.encode(item)))
},
}
const fetch = (input: RequestInfo | URL, init?: RequestInit) => {
@@ -87,7 +87,7 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) {
controller.enqueue(
encoder.encode(frame({ id: `evt_mock_connected_${id}`, type: "server.connected", data: {} })),
)
for (const item of state.buffer.splice(0)) controller.enqueue(encoder.encode(item))
state.buffer.splice(0).forEach((item) => controller.enqueue(encoder.encode(item)))
request.signal.addEventListener(
"abort",
() => {
@@ -114,7 +114,7 @@ export async function mockOpenCodeServer(page: Page, config: MockServerConfig) {
}
Object.defineProperty(window, "fetch", { configurable: true, writable: true, value: fetch })
},
{ port: streamPort, retry: config.eventRetry },
{ port: process.env.PLAYWRIGHT_SERVER_PORT ?? "4096", retry: config.eventRetry },
)
if (config.events) {
packages/app/src/components/dialog-connect-provider.tsx
+13 -18
View File
@@ -139,9 +139,9 @@ export const DialogConnectProvider: Component<{
function ProviderPicker(props: { directory?: string; onSelect: (provider: string) => void; onPrepare?: () => void }) {
const settings = useSettings()
const integrations = useIntegrations(() => props.directory)
if (settings.general.newLayoutDesigns())
return <ProviderPickerV2 integrations={integrations.list} onSelect={props.onSelect} onPrepare={props.onPrepare} />
return <ProviderPickerV2 directory={props.directory} onSelect={props.onSelect} onPrepare={props.onPrepare} />
const integrations = useIntegrations(() => props.directory)
const language = useLanguage()
const popularGroup = () => language.t("dialog.provider.group.popular")
const otherGroup = () => language.t("dialog.provider.group.other")
@@ -208,11 +208,8 @@ function ProviderPicker(props: { directory?: string; onSelect: (provider: string
)
}
function ProviderPickerV2(props: {
integrations: ReturnType<typeof useIntegrations>["list"]
onSelect: (provider: string) => void
onPrepare?: () => void
}) {
function ProviderPickerV2(props: { directory?: string; onSelect: (provider: string) => void; onPrepare?: () => void }) {
const integrations = useIntegrations(() => props.directory)
const language = useLanguage()
const [store, setStore] = createStore({
filter: "",
@@ -224,7 +221,7 @@ function ProviderPickerV2(props: {
const all = createMemo(() => {
language.locale()
const query = store.filter.trim().toLowerCase()
const values = [custom(), ...props.integrations()]
const values = [custom(), ...integrations.list()]
if (!query) return values
return values.filter((provider) => `${provider.id} ${provider.name}`.toLowerCase().includes(query))
})
@@ -387,16 +384,14 @@ function ProviderConnection(props: {
},
})
const provider = createMemo(
() =>
providers.all().get(props.provider) ??
serverSync().data.provider.all.get(props.provider) ?? {
id: props.provider,
name: controller.integration()?.name ?? props.provider,
source: "custom" as const,
env: [],
options: {},
models: {},
},
() => ({
id: props.provider,
name:
providers.all().get(props.provider)?.name ??
serverSync().data.provider.all.get(props.provider)?.name ??
controller.integration()?.name ??
props.provider,
}),
)
const methodLabel = (value?: { type?: string; label?: string }) => {
if (!value) return ""
packages/app/src/context/global-sync/event-reducer.ts
-2
View File
@@ -42,8 +42,6 @@ export function applyGlobalEvent(input: {
input.refresh()
return
}
if (input.event.type === "server.connected") return
if (input.event.type !== "project.updated") return
const properties = input.event.properties as Project
const result = Binary.search(input.project, properties.id, (s) => s.id)
packages/app/src/context/server-sdk.tsx
+20 -13
View File
@@ -140,7 +140,7 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
let timer: ReturnType<typeof setTimeout> | undefined
let last = 0
const flush = () => {
function flush() {
if (timer) clearTimeout(timer)
timer = undefined
@@ -160,18 +160,20 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
buffer.length = 0
}
const schedule = () => {
function schedule() {
if (timer) return
const elapsed = Date.now() - last
timer = setTimeout(flush, Math.max(0, FLUSH_FRAME_MS - elapsed))
}
const publish = (event: OpenCodeEvent) => {
function publish(event: OpenCodeEvent) {
const directory = event.location?.directory ?? "global"
if (enqueueServerEvent(queue, { directory, payload: adaptServerEvent(event) })) schedule()
}
const wait = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms))
function wait(ms: number) {
return new Promise<void>((resolve) => setTimeout(resolve, ms))
}
let attempt: AbortController | undefined
let run: Promise<void> | undefined
let started = false
@@ -182,15 +184,17 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
error?: string
}>({ status: "connecting", attempt: 0 })
const connect = async (signal: AbortSignal): Promise<{ error: unknown; connectedAt: number | undefined }> => {
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()
@@ -203,11 +207,13 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
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()
@@ -228,17 +234,18 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
}
}
const main = async (active: number) => {
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 })
attempt = new AbortController()
const onAbort = () => attempt?.abort()
const controller = new AbortController()
attempt = controller
const onAbort = () => controller.abort()
abort.signal.addEventListener("abort", onAbort)
const result = await connect(attempt.signal)
const result = await connect(controller.signal)
abort.signal.removeEventListener("abort", onAbort)
attempt = undefined
if (attempt === controller) attempt = undefined
if (abort.signal.aborted || !started || generation !== active) return
if (result.connectedAt !== undefined && Date.now() - result.connectedAt >= 1_000) retries = 0
@@ -260,14 +267,14 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
}
}
const start = () => {
function start() {
if (started) return run
started = true
const active = ++generation
const previous = run
const current = (async () => {
if (previous) await previous
await main(active)
await runStream(active)
})().finally(() => {
if (run !== current) return
run = undefined
@@ -277,7 +284,7 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
return run
}
const stop = () => {
function stop() {
started = false
generation++
attempt?.abort()
packages/app/src/context/server-session-v2-reducer.test.ts
+4 -3
View File
@@ -211,7 +211,7 @@ describe("v2 session reducer", () => {
expect(result).toMatchObject({ sessionID: "ses_1", missing: "msg_user", touched: [] })
})
test("projects non-initial instruction updates", () => {
test("projects rendered instruction updates", () => {
const reducer = createV2SessionReducer()
const result = reducer.reduce(
[],
@@ -219,7 +219,7 @@ describe("v2 session reducer", () => {
...base,
id: "evt_instructions",
type: "session.instructions.updated",
data: { sessionID: "ses_1", delta: { agents: "hash" } },
data: { sessionID: "ses_1", delta: { agents: "hash" }, text: "Changed instructions" },
}),
)
@@ -227,7 +227,8 @@ describe("v2 session reducer", () => {
{
id: "msg_instructions",
type: "system",
text: "Instructions updated: agents",
text: "Changed instructions",
description: "Instructions updated: agents",
metadata: undefined,
time: { created: 1 },
},
packages/app/src/context/server-session-v2-reducer.ts
+4 -11
View File
@@ -111,23 +111,16 @@ export function createV2SessionReducer() {
)?.model,
time: { created: event.created },
})
case "session.instructions.updated": {
const instructions = event.metadata?.instructions
if (
typeof instructions === "object" &&
instructions !== null &&
"initial" in instructions &&
instructions.initial === true
)
return
case "session.instructions.updated":
if (event.data.text === undefined) return
return append({
id: messageID(event.id),
type: "system",
text: `Instructions updated: ${Object.keys(event.data.delta).join(", ")}`,
text: event.data.text,
description: `Instructions updated: ${Object.keys(event.data.delta).join(", ")}`,
metadata: event.metadata,
time: { created: event.created },
})
}
case "session.synthetic":
return append({
id: messageID(event.id),
packages/app/src/context/server-sync.tsx
+12 -21
View File
@@ -167,12 +167,10 @@ export function reconcileActiveSessionStatuses(
session: Pick<ServerSession, "data" | "set">,
active: SessionActiveOutput,
) {
for (const sessionID of Object.keys(session.data.session_status)) {
if (active[sessionID] !== undefined) continue
if (session.data.session_status[sessionID]?.type !== "idle")
session.set("session_status", sessionID, { type: "idle" })
}
for (const sessionID of Object.keys(active)) session.set("session_status", sessionID, { type: "busy" })
Object.keys(session.data.session_status)
.filter((sessionID) => active[sessionID] === undefined && session.data.session_status[sessionID]?.type !== "idle")
.forEach((sessionID) => session.set("session_status", sessionID, { type: "idle" }))
Object.keys(active).forEach((sessionID) => session.set("session_status", sessionID, { type: "busy" }))
}
function makeQueryOptionsApi(scope: ServerScope, serverAPI: ServerApi) {
@@ -227,9 +225,9 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) {
active: async () => {
const active = await serverSDK.api.session.active()
reconcileActiveSessionStatuses(session, active)
for (const sessionID of Object.keys(active)) {
Object.keys(active).forEach((sessionID) => {
void Promise.all([session.resolve(sessionID), hydrateSessionState(sessionID)]).catch(() => undefined)
}
})
return active
},
}),
@@ -362,21 +360,14 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) {
if (info.reconnect) void session.refreshPinned()
if (activeSessionsQuery.data !== undefined && !activeSessionsQuery.isFetching) void activeSessionsQuery.refetch()
if (bootstrap.data !== undefined && !bootstrap.isFetching) void bootstrap.refetch()
for (const directory of Object.keys(children.children)) {
if (children.active(directory)) queue.push(directory)
}
Object.keys(children.children).filter(children.active).forEach(queue.push)
},
})
const location = createLocationSync({
scope: serverSDK.scope,
queryClient,
active: () => Object.keys(children.children).filter(children.active).map(pathKey),
info: (directory) => serverSDK.api.location.get({ location: { directory } }),
vcs: (directory) => serverSDK.api.vcs.get({ location: { directory } }).then((result) => result.data),
skill: (directory) => serverSDK.api.skill.list({ location: { directory } }).then((result) => result.data),
websearch: (directory) =>
serverSDK.api.websearch.providers({ location: { directory } }).then((result) => result.data),
shell: (directory) => serverSDK.api.shell.list({ location: { directory } }).then((result) => result.data),
api: serverSDK.api,
})
async function loadSessions(directory: string, options?: { limit?: number }) {
@@ -552,9 +543,9 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) {
})
}
homeSessions.refresh(event.type)
catalog.main({ type: eventType, directory })
connection.main({ type: eventType, directory })
if (event.current) location.main(event.current, directory)
catalog.handleEvent({ type: eventType, directory })
connection.handleEvent({ type: eventType, directory })
if (event.current) location.handleEvent(event.current, directory)
if (directory === "global") {
applyGlobalEvent({
@@ -570,7 +561,7 @@ export function createServerSyncContextInner(serverSDK: ServerSDK) {
)
bootstrap.refetch()
if (eventType === "global.disposed")
for (const directory of Object.keys(children.children)) if (children.active(directory)) queue.push(directory)
Object.keys(children.children).filter(children.active).forEach(queue.push)
return
}
packages/app/src/context/server-sync/catalog.test.ts
+2 -2
View File
@@ -17,7 +17,7 @@ test("invalidates the catalog for the event location", async () => {
load: async () => {},
})
catalog.main({ type: "catalog.updated", directory: "/one" })
catalog.handleEvent({ type: "catalog.updated", directory: "/one" })
await Bun.sleep(0)
expect(queryClient.getQueryState(one)?.isInvalidated).toBe(true)
@@ -39,7 +39,7 @@ test("invalidates global and active catalogs after connection", async () => {
load: async () => {},
})
catalog.main({ type: "server.connected", directory: "global" })
catalog.handleEvent({ type: "server.connected", directory: "global" })
await Bun.sleep(0)
expect(queryClient.getQueryState(global)?.isInvalidated).toBe(true)
packages/app/src/context/server-sync/catalog.ts
+2 -2
View File
@@ -13,7 +13,7 @@ export function createCatalogSync(input: {
active: () => PathKey[]
load: (directory: PathKey | null) => Promise<void>
}) {
function main(event: CatalogEvent) {
function handleEvent(event: CatalogEvent) {
if (event.type === "server.connected") {
void refreshActive()
return
@@ -42,7 +42,7 @@ export function createCatalogSync(input: {
}
return {
main,
handleEvent,
refresh,
refreshActive,
}
packages/app/src/context/server-sync/connection.test.ts
+2 -2
View File
@@ -12,9 +12,9 @@ test("invalidates disconnected data and synchronizes after the handshake", () =>
connected: () => calls.push("connected"),
})
connection.main({ type: "server.connected", directory: "global" })
connection.handleEvent({ type: "server.connected", directory: "global" })
expect(calls).toContain("connected")
connection.main({ type: "server.connected", directory: "/repo" })
connection.handleEvent({ type: "server.connected", directory: "/repo" })
expect(calls.filter((call) => call === "connected")).toHaveLength(1)
setStatus("connected")
return dispose
packages/app/src/context/server-sync/connection.ts
+2 -2
View File
@@ -12,11 +12,11 @@ export function createConnectionSync(input: {
})
let connectedOnce = false
function main(event: { type: string; directory: string }) {
function handleEvent(event: { type: string; directory: string }) {
if (event.directory !== "global" || event.type !== "server.connected") return
input.connected({ reconnect: connectedOnce })
connectedOnce = true
}
return { main }
return { handleEvent }
}
packages/app/src/context/server-sync/location.test.ts
+23 -11
View File
@@ -9,21 +9,33 @@ test("projects shell events and refreshes location catalogs", async () => {
const queryClient = new QueryClient()
const directory = pathKey("/repo")
const calls: string[] = []
const location = createLocationSync({
const location = { directory, project: { id: "project", directory, canonical: directory } }
const sync = createLocationSync({
scope: ServerScope.local,
queryClient,
active: () => [directory],
info: async () => calls.push("info"),
vcs: async () => calls.push("vcs"),
skill: async () => calls.push("skill"),
websearch: async () => calls.push("websearch"),
shell: async () => {
calls.push("shell")
return []
api: {
location: { get: async () => Promise.reject(new Error("Unexpected location refresh")) },
vcs: { get: async () => Promise.reject(new Error("Unexpected VCS refresh")) },
skill: {
list: async () => {
calls.push("skill")
return { location, data: [] }
},
},
websearch: {
providers: async () => {
calls.push("websearch")
return { location, data: [] }
},
},
shell: {
list: async () => Promise.reject(new Error("Unexpected shell refresh")),
},
},
})
location.main(
sync.handleEvent(
{
type: "shell.created",
id: "evt_1",
@@ -42,8 +54,8 @@ test("projects shell events and refreshes location catalogs", async () => {
} as OpenCodeEvent,
directory,
)
location.main({ type: "skill.updated", id: "evt_2", data: {} } as OpenCodeEvent, directory)
location.main({ type: "websearch.updated", id: "evt_3", data: {} } as OpenCodeEvent, directory)
sync.handleEvent({ type: "skill.updated", id: "evt_2", data: {} } as OpenCodeEvent, directory)
sync.handleEvent({ type: "websearch.updated", id: "evt_3", data: {} } as OpenCodeEvent, directory)
await Bun.sleep(0)
expect(queryClient.getQueryData<unknown[]>([ServerScope.local, directory, "shell"])).toHaveLength(1)
packages/app/src/context/server-sync/location.ts
+29 -12
View File
@@ -2,18 +2,23 @@ import type { OpenCodeEvent, ShellInfo } from "@opencode-ai/client/promise"
import type { QueryClient } from "@tanstack/solid-query"
import { pathKey, type PathKey } from "@/utils/path-key"
import type { ServerScope } from "@/utils/server-scope"
import type { ServerApi } from "@/utils/server"
type LocationApi = {
location: Pick<ServerApi["location"], "get">
vcs: Pick<ServerApi["vcs"], "get">
skill: Pick<ServerApi["skill"], "list">
websearch: Pick<ServerApi["websearch"], "providers">
shell: Pick<ServerApi["shell"], "list">
}
export function createLocationSync(input: {
scope: ServerScope
queryClient: QueryClient
active: () => PathKey[]
info: (directory: PathKey) => Promise<unknown>
vcs: (directory: PathKey) => Promise<unknown>
skill: (directory: PathKey) => Promise<unknown>
websearch: (directory: PathKey) => Promise<unknown>
shell: (directory: PathKey) => Promise<ShellInfo[]>
api: LocationApi
}) {
function main(event: OpenCodeEvent, directory: string) {
function handleEvent(event: OpenCodeEvent, directory: string) {
if (event.type === "server.connected") {
void refreshActive()
return
@@ -42,23 +47,35 @@ export function createLocationSync(input: {
}
async function refreshInfo(directory: PathKey) {
input.queryClient.setQueryData(infoKey(directory), await input.info(directory))
input.queryClient.setQueryData(infoKey(directory), await input.api.location.get({ location: { directory } }))
}
async function refreshVcs(directory: PathKey) {
input.queryClient.setQueryData(vcsKey(directory), await input.vcs(directory))
input.queryClient.setQueryData(
vcsKey(directory),
(await input.api.vcs.get({ location: { directory } })).data,
)
}
async function refreshSkill(directory: PathKey) {
input.queryClient.setQueryData(skillKey(directory), await input.skill(directory))
input.queryClient.setQueryData(
skillKey(directory),
(await input.api.skill.list({ location: { directory } })).data,
)
}
async function refreshWebsearch(directory: PathKey) {
input.queryClient.setQueryData(websearchKey(directory), await input.websearch(directory))
input.queryClient.setQueryData(
websearchKey(directory),
(await input.api.websearch.providers({ location: { directory } })).data,
)
}
async function refreshShell(directory: PathKey) {
input.queryClient.setQueryData(shellKey(directory), await input.shell(directory))
input.queryClient.setQueryData(
shellKey(directory),
(await input.api.shell.list({ location: { directory } })).data,
)
}
function rememberShell(directory: PathKey, shell: ShellInfo) {
@@ -80,5 +97,5 @@ export function createLocationSync(input: {
const websearchKey = (directory: PathKey) => [input.scope, directory, "websearch"] as const
const shellKey = (directory: PathKey) => [input.scope, directory, "shell"] as const
return { main }
return { handleEvent }
}
packages/app/src/hooks/use-providers.ts
+1 -5
View File
@@ -1,6 +1,4 @@
import { useServerSync } from "@/context/server-sync"
import { decode64 } from "@/utils/base64"
import { useParams } from "@solidjs/router"
import { Iterable, pipe } from "effect"
import { createEffect, createMemo, type Accessor } from "solid-js"
import { selectProviderCatalog } from "./provider-catalog"
@@ -19,8 +17,7 @@ const popularProviderSet = new Set(popularProviders)
export function useProviders(directory: Accessor<string | undefined>) {
const serverSync = useServerSync()
const params = useParams()
const dir = () => (directory ? directory() : decode64(params.dir))
const dir = directory
const providers = () => {
const value = dir()
const projectStore = value ? serverSync().child(value)[0] : undefined
@@ -41,7 +38,6 @@ export function useProviders(directory: Accessor<string | undefined>) {
return {
ready: () => {
const value = dir()
if (!directory) return true
if (!value) return false
return serverSync().child(value)[0].provider_ready
},