Compare commits

..
Author SHA1 Message Date
Aiden Cline 01635b20ee fix(opencode): clarify vertex anthropic authentication 2026-05-27 23:57:28 -05:00
56 changed files with 1275 additions and 8997 deletions
+1 -1
View File
@@ -10,7 +10,7 @@
"dev:desktop": "bun --cwd packages/desktop dev",
"dev:web": "bun --cwd packages/app dev",
"dev:console": "ulimit -n 10240 2>/dev/null; bun run --cwd packages/console/app dev",
"dev:stats": "bun sst shell --stage=dev -- bun run --cwd packages/stats/app dev",
"dev:stats": "bun sst shell --stage=production -- bun run --cwd packages/stats/app dev",
"dev:storybook": "bun --cwd packages/storybook storybook",
"lint": "oxlint",
"typecheck": "bun turbo typecheck",
@@ -2,7 +2,6 @@ import { action, json, query, useAction, useSubmission } from "@solidjs/router"
import { createEffect, createMemo, createSignal, For, onCleanup, Show } from "solid-js"
import { getRequestEvent } from "solid-js/web"
import { Referral } from "@opencode-ai/console-core/referral.js"
import { Actor } from "@opencode-ai/console-core/actor.js"
import { withActor } from "~/context/auth.withActor"
import { Modal } from "~/component/modal"
import { IconCheck, IconCopy } from "~/component/icon"
@@ -10,7 +9,6 @@ import { useI18n } from "~/context/i18n"
import { useLanguage } from "~/context/language"
import { formatResetTime, liteResetTimeKeys } from "~/lib/format-reset-time"
import { queryLiteSubscription } from "~/routes/workspace/[id]/go/lite-section"
import { clearReferralCookie, referralCodeFromCookieHeader } from "~/lib/referral-invite"
import "./go-referral.css"
type GoReferralSummary = Awaited<ReturnType<typeof Referral.summary>>
@@ -27,21 +25,7 @@ const emptyUsagePreview = {
export const queryGoReferral = query(async (workspaceID: string) => {
"use server"
return withActor(async () => {
const event = getRequestEvent()
const referralCode = referralCodeFromCookieHeader(event?.request.headers.get("cookie") ?? null)
if (referralCode) {
await Referral.createFromAccount({
accountID: Actor.account(),
referralCode,
}).catch((error) => {
console.error("Referral create failed", error)
})
event?.response.headers.append("set-cookie", clearReferralCookie())
}
return Referral.summary()
}, workspaceID)
return withActor(() => Referral.summary(), workspaceID)
}, "go.referral.get")
export const queryGoReferralUsagePreview = query(async (workspaceID: string, referralID?: string) => {
@@ -81,8 +65,6 @@ function rewardDescriptionKey(source: GoReferralReward["source"]) {
function rewardActionKey(reward: GoReferralReward, hasActiveGo: boolean) {
if (reward.status === "applied") return "workspace.referral.reward.action.applied" as const
if (reward.status === "pending" && reward.source === "inviter")
return "workspace.referral.reward.source.pendingInviter" as const
if (reward.status === "pending" || !hasActiveGo) return "workspace.referral.reward.action.subscribeUnlock" as const
return "workspace.referral.reward.action.view" as const
}
@@ -1,9 +1,11 @@
import { redirect } from "@solidjs/router"
import type { APIEvent } from "@solidjs/start/server"
import { Referral } from "@opencode-ai/console-core/referral.js"
import { AuthClient } from "~/context/auth"
import { useAuthSession } from "~/context/auth"
import { i18n } from "~/i18n"
import { localeFromRequest, route } from "~/lib/language"
import { clearReferralCookie, referralCodeFromCookieHeader } from "~/lib/referral-invite"
export async function GET(input: APIEvent) {
const url = new URL(input.request.url)
@@ -17,6 +19,7 @@ export async function GET(input: APIEvent) {
if (result.err) throw new Error(result.err.message)
const decoded = AuthClient.decode(result.tokens.access, {} as any)
if (decoded.err) throw new Error(decoded.err.message)
const referralCode = referralCodeFromCookieHeader(input.request.headers.get("cookie"))
const session = await useAuthSession()
const id = decoded.subject.properties.accountID
await session.update((value) => {
@@ -32,8 +35,15 @@ export async function GET(input: APIEvent) {
current: id,
}
})
if (decoded.subject.properties.newAccount && referralCode) {
await Referral.createFromAccount({ accountID: id, referralCode }).catch((error) => {
console.error("Referral create failed", error)
})
}
const next = url.pathname === "/auth/callback" ? "/auth" : url.pathname.replace("/auth/callback", "")
return redirect(route(locale, next))
const response = redirect(route(locale, next))
if (referralCode) response.headers.append("set-cookie", clearReferralCookie())
return response
} catch (e: any) {
return new Response(
JSON.stringify({
-1
View File
@@ -17,7 +17,6 @@ export default defineConfig({
],
server: {
allowedHosts: true,
port: 3001,
},
build: {
rollupOptions: {
+4 -24
View File
@@ -1,8 +1,8 @@
import { z } from "zod"
import { and, asc, eq, inArray, isNull, sql, Database } from "./drizzle"
import { and, asc, eq, isNull, sql, Database } from "./drizzle"
import { Actor } from "./actor"
import { Identifier } from "./identifier"
import { LiteTable, PaymentTable } from "./schema/billing.sql"
import { LiteTable } from "./schema/billing.sql"
import { ReferralCodeTable, ReferralRewardTable, ReferralTable } from "./schema/referral.sql"
import { AuthTable } from "./schema/auth.sql"
import { UserTable } from "./schema/user.sql"
@@ -318,26 +318,6 @@ export namespace Referral {
.then((rows) => rows[0])
if (selfReferral) throw new Error("Self-referral is not allowed")
const workspaceIDs = await tx
.select({ workspaceID: UserTable.workspaceID })
.from(UserTable)
.where(and(eq(UserTable.accountID, input.accountID), isNull(UserTable.timeDeleted)))
.then((rows) => rows.map((row) => row.workspaceID))
if (workspaceIDs.length === 0) return
const litePayment = await tx
.select({ id: PaymentTable.id })
.from(PaymentTable)
.where(
and(
inArray(PaymentTable.workspaceID, workspaceIDs),
isNull(PaymentTable.timeDeleted),
sql`JSON_UNQUOTE(JSON_EXTRACT(${PaymentTable.enrichment}, '$.type')) = 'lite'`,
),
)
.then((rows) => rows[0])
if (litePayment) return
const referralID = Identifier.create("referral")
await tx.insert(ReferralTable).ignore().values({
workspaceID: code.workspaceID,
@@ -375,7 +355,7 @@ export namespace Referral {
.from(ReferralTable)
.where(and(eq(ReferralTable.inviteeAccountID, invitee.accountID), isNull(ReferralTable.timeDeleted)))
.then((rows) => rows[0])
if (!referral) return
if (!referral) throw new Error("Referral not found")
const result = await tx
.insert(ReferralRewardTable)
@@ -393,7 +373,7 @@ export namespace Referral {
},
])
if (result.rowsAffected === 0) return
if (result.rowsAffected === 0) throw new Error("Referral already completed")
})
}
-1
View File
@@ -29,7 +29,6 @@ export default defineConfig({
server: {
host: "0.0.0.0",
allowedHosts: true,
port: 3002,
},
worker: {
format: "es",
-42
View File
@@ -1,42 +0,0 @@
const enabled = process.env.OPENCODE_ACP_PROFILE === "1"
const started = performance.now()
export function mark(name: string, fields?: Record<string, string | number | boolean | undefined>) {
if (!enabled) return
write(`${name}.mark`, performance.now() - started, fields)
}
export function duration(
name: string,
startedAt: number,
fields?: Record<string, string | number | boolean | undefined>,
) {
if (!enabled) return
write(name, performance.now() - startedAt, fields)
}
export async function measure<T>(
name: string,
fn: () => Promise<T>,
fields?: Record<string, string | number | boolean | undefined>,
) {
if (!enabled) return fn()
const start = performance.now()
try {
return await fn()
} finally {
write(name, performance.now() - start, fields)
}
}
function write(name: string, durationMs: number, fields?: Record<string, string | number | boolean | undefined>) {
const extra = fields
? Object.entries(fields)
.filter((entry): entry is [string, string | number | boolean] => entry[1] !== undefined)
.map(([key, value]) => `${key}=${value}`)
.join(" ")
: ""
console.error(`[acp-profile] ${name} ${Math.round(durationMs)}ms${extra ? ` ${extra}` : ""}`)
}
export * as ACPNextProfile from "./profile"
+99 -338
View File
@@ -35,12 +35,9 @@ import type { Message, OpencodeClient, SessionMessageResponse } from "@opencode-
import { Context, Effect, Layer, ManagedRuntime } from "effect"
import * as ACPNextError from "./error"
import { buildConfigOptions, parseModelSelection } from "./config-option"
import { promptContentToParts } from "./content"
import { Directory } from "./directory"
import { ACPNextEvent } from "./event"
import { ACPNextSession } from "./session"
import { UsageService } from "./usage"
import { ACPNextProfile } from "./profile"
import { ModelID, ProviderID } from "@/provider/schema"
import { Provider } from "@/provider/provider"
import type { Command } from "@/command"
@@ -49,8 +46,6 @@ export const AuthMethodID = "opencode-login"
const log = Log.create({ service: "acp-next-service" })
export type Error = ACPNextError.Error
type ServiceConnection = Pick<AgentSideConnection, "sessionUpdate"> &
Partial<Pick<AgentSideConnection, "requestPermission" | "writeTextFile">>
export type Interface = {
readonly initialize: (input: InitializeRequest) => Effect.Effect<InitializeResponse, Error>
@@ -74,23 +69,21 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/AC
export function make(input: {
sdk: OpencodeClient
connection?: ServiceConnection
connection?: Pick<AgentSideConnection, "sessionUpdate"> &
Partial<Pick<AgentSideConnection, "requestPermission" | "writeTextFile">>
directory?: Directory.Interface
session?: ACPNextSession.Interface
usage?: UsageService.Interface
eventSubscription?: (subscription: ACPNextEvent.Subscription) => void
}): Interface {
const session = input.session ?? makeSessionService()
const directoryService = input.directory ?? makeDirectoryService(input.sdk)
const registeredMcp = new Map<string, Set<string>>()
const sessionSnapshots = new Map<string, Directory.Snapshot>()
const events = input.connection
? ACPNextEvent.start({ sdk: input.sdk, connection: input.connection, session })
: undefined
if (events) input.eventSubscription?.(events)
const initialize = Effect.fn("ACPNext.initialize")(function* (params: InitializeRequest) {
const started = performance.now()
const authMethod: AuthMethod = {
description: "Run `opencode auth login` in the terminal",
name: "Login with opencode",
@@ -107,7 +100,7 @@ export function make(input: {
}
}
const response = {
return {
protocolVersion: 1,
agentCapabilities: {
loadSession: true,
@@ -132,8 +125,6 @@ export function make(input: {
version: InstallationVersion,
},
}
ACPNextProfile.duration("acp.initialize", started)
return response
})
const authenticate = Effect.fn("ACPNext.authenticate")(function* (params: AuthenticateRequest) {
@@ -144,28 +135,15 @@ export function make(input: {
})
const directorySnapshot = Effect.fn("ACPNext.directorySnapshot")(function* (cwd: string) {
const started = performance.now()
const snapshot = yield* directoryService.get(cwd)
ACPNextProfile.duration("acp.directory.snapshot", started)
return snapshot
})
const configSnapshot = Effect.fn("ACPNext.configSnapshot")(function* (state: ACPNextSession.Info) {
const snapshot = sessionSnapshots.get(state.id)
if (snapshot) return snapshot
const loaded = yield* directorySnapshot(state.cwd)
sessionSnapshots.set(state.id, loaded)
return loaded
return yield* directoryService.get(cwd)
})
const newSession = Effect.fn("ACPNext.newSession")(function* (params: NewSessionRequest) {
const started = performance.now()
const snapshot = yield* directorySnapshot(params.cwd)
const selected = selectDefaultModel(snapshot)
const variant = selectVariant(snapshot, selected)
const modeId = snapshot.availableModes.length > 0 ? snapshot.defaultModeID : undefined
const created = yield* profiledRequest(
"acp.newSession.session.create",
const created = yield* request(
() =>
input.sdk.session.create(
{
@@ -189,12 +167,11 @@ export function make(input: {
variant,
modeId,
})
sessionSnapshots.set(state.id, snapshot)
yield* registerMcpServers(input.sdk, registeredMcp, params.cwd, state.id, params.mcpServers)
yield* sendAvailableCommands(input.connection, state.id, snapshot)
const response = {
return {
sessionId: state.id,
configOptions: configOptions(snapshot, {
model: state.model ?? selected,
@@ -202,8 +179,6 @@ export function make(input: {
modeId: state.modeId,
}),
}
ACPNextProfile.duration("acp.newSession", started)
return response
})
const loadSession = Effect.fn("ACPNext.loadSession")(function* (params: LoadSessionRequest) {
@@ -230,7 +205,6 @@ export function make(input: {
variant: restored.variant ?? selectVariant(snapshot, model),
modeId: restored.modeId ?? (snapshot.availableModes.length > 0 ? snapshot.defaultModeID : undefined),
})
sessionSnapshots.set(state.id, snapshot)
yield* registerMcpServers(input.sdk, registeredMcp, params.cwd, state.id, params.mcpServers)
yield* sendAvailableCommands(input.connection, state.id, snapshot)
@@ -259,35 +233,21 @@ export function make(input: {
),
"session",
)
const serverEntries = sessions.map(
(item): SessionInfo => ({
sessionId: item.id,
cwd: item.directory,
title: item.title,
updatedAt: new Date(item.time.updated).toISOString(),
}),
)
const liveEntries = (yield* session.list(params.cwd ?? undefined))
.filter((item) => !serverEntries.some((entry) => entry.sessionId === item.id))
.map(
(item): SessionInfo => ({
sessionId: item.id,
cwd: item.cwd,
updatedAt: item.createdAt.toISOString(),
}),
)
const sorted = [...liveEntries, ...serverEntries].toSorted(
(a, b) => new Date(b.updatedAt ?? 0).getTime() - new Date(a.updatedAt ?? 0).getTime(),
)
const sorted = sessions.toSorted((a, b) => b.time.updated - a.time.updated)
const filtered =
cursor === undefined || !Number.isFinite(cursor)
? sorted
: sorted.filter((item) => new Date(item.updatedAt ?? 0).getTime() < cursor)
cursor === undefined || !Number.isFinite(cursor) ? sorted : sorted.filter((item) => item.time.updated < cursor)
const page = filtered.slice(0, limit)
const last = page.at(-1)
return {
sessions: page,
...(filtered.length > limit && last ? { nextCursor: String(new Date(last.updatedAt ?? 0).getTime()) } : {}),
sessions: page.map(
(item): SessionInfo => ({
sessionId: item.id,
cwd: item.directory,
title: item.title,
updatedAt: new Date(item.time.updated).toISOString(),
}),
),
...(filtered.length > limit && last ? { nextCursor: String(last.time.updated) } : {}),
}
})
@@ -315,7 +275,6 @@ export function make(input: {
variant: restored.variant ?? selectVariant(snapshot, model),
modeId: restored.modeId ?? (snapshot.availableModes.length > 0 ? snapshot.defaultModeID : undefined),
})
sessionSnapshots.set(state.id, snapshot)
yield* registerMcpServers(input.sdk, registeredMcp, params.cwd, state.id, params.mcpServers ?? [])
yield* sendAvailableCommands(input.connection, state.id, snapshot)
@@ -333,7 +292,6 @@ export function make(input: {
const closeSession = Effect.fn("ACPNext.closeSession")(function* (params: CloseSessionRequest) {
const removed = yield* session.remove(params.sessionId)
registeredMcp.delete(params.sessionId)
sessionSnapshots.delete(params.sessionId)
if (!removed) return {}
yield* request(
@@ -377,7 +335,6 @@ export function make(input: {
variant: restored.variant ?? selectVariant(snapshot, model),
modeId: restored.modeId ?? (snapshot.availableModes.length > 0 ? snapshot.defaultModeID : undefined),
})
sessionSnapshots.set(state.id, snapshot)
yield* registerMcpServers(input.sdk, registeredMcp, params.cwd, state.id, params.mcpServers ?? [])
yield* sendAvailableCommands(input.connection, state.id, snapshot)
@@ -397,7 +354,7 @@ export function make(input: {
params: SetSessionConfigOptionRequest,
) {
const current = yield* session.get(params.sessionId)
const snapshot = yield* configSnapshot(current)
const snapshot = yield* directorySnapshot(current.cwd)
if (typeof params.value !== "string") {
return yield* new ACPNextError.InvalidConfigOptionError({ configId: params.configId })
}
@@ -452,7 +409,7 @@ export function make(input: {
const setSessionMode = Effect.fn("ACPNext.setSessionMode")(function* (params: SetSessionModeRequest) {
const current = yield* session.get(params.sessionId)
const snapshot = yield* configSnapshot(current)
const snapshot = yield* directorySnapshot(current.cwd)
if (!snapshot.availableModes.some((mode) => mode.id === params.modeId)) {
return yield* new ACPNextError.InvalidModeError({ mode: params.modeId })
}
@@ -462,7 +419,7 @@ export function make(input: {
const setSessionModel = Effect.fn("ACPNext.setSessionModel")(function* (params: SetSessionModelRequest) {
const current = yield* session.get(params.sessionId)
const snapshot = yield* configSnapshot(current)
const snapshot = yield* directorySnapshot(current.cwd)
const selected = yield* parseSelectedModel(snapshot, params.modelId)
yield* session
.setVariant(
@@ -487,81 +444,8 @@ export function make(input: {
setSessionConfigOption,
setSessionMode,
setSessionModel,
prompt: Effect.fn("ACPNext.prompt")(function* (params: PromptRequest) {
const current = yield* session.get(params.sessionId)
const snapshot = yield* directorySnapshot(current.cwd)
const selected = current.model ?? selectDefaultModel(snapshot)
if (!current.model) {
yield* session.setModel(params.sessionId, selected)
}
const variant = current.variant ?? selectVariant(snapshot, selected)
const modeId = current.modeId ?? (snapshot.availableModes.length > 0 ? snapshot.defaultModeID : undefined)
const parts = promptContentToParts(params.prompt)
const command = detectSlashCommand(parts)
if (!command) {
const response = yield* request(
() =>
input.sdk.session.prompt(
{
sessionID: current.id,
model: {
providerID: selected.providerID,
modelID: selected.modelID,
},
...(variant ? { variant } : {}),
parts,
...(modeId ? { agent: modeId } : {}),
directory: current.cwd,
},
{ throwOnError: true },
),
"session",
)
yield* sendUsageUpdate(input.usage, input.sdk, input.connection, current.id, current.cwd)
return promptResponse(response.info, params.messageId)
}
const known = snapshot.availableCommands.find((item) => item.name === command.name)
if (known) {
const response = yield* request(
() =>
input.sdk.session.command(
{
sessionID: current.id,
command: known.name,
arguments: command.args,
model: `${selected.providerID}/${selected.modelID}`,
...(variant ? { variant } : {}),
...(modeId ? { agent: modeId } : {}),
directory: current.cwd,
},
{ throwOnError: true },
),
"session",
)
yield* sendUsageUpdate(input.usage, input.sdk, input.connection, current.id, current.cwd)
return promptResponse(response.info, params.messageId)
}
if (command.name === "compact") {
yield* request(
() =>
input.sdk.session.summarize(
{
sessionID: current.id,
directory: current.cwd,
providerID: selected.providerID,
modelID: selected.modelID,
},
{ throwOnError: true },
),
"session",
)
}
yield* sendUsageUpdate(input.usage, input.sdk, input.connection, current.id, current.cwd)
return promptResponse(undefined, params.messageId)
prompt: Effect.fn("ACPNext.prompt")(function* (_input: PromptRequest) {
return yield* new ACPNextError.UnsupportedOperationError({ method: "session/prompt" })
}),
cancel: Effect.fn("ACPNext.cancel")(function* (_input: CancelNotification) {
return yield* new ACPNextError.UnsupportedOperationError({ method: "session/cancel" })
@@ -590,91 +474,6 @@ function makeDirectoryService(sdk: OpencodeClient) {
).runSync(Directory.Service.use((service) => Effect.succeed(service)))
}
function makeUsageService(sdk: OpencodeClient) {
const limits = new Map<string, Promise<number | undefined>>()
const contextLimit: UsageService.Interface["contextLimit"] = Effect.fn("ACPNext.promptUsage.contextLimit")(
function* (params) {
const key = `${params.directory}\u0000${params.providerID}\u0000${params.modelID}`
const current = limits.get(key)
if (current) return yield* Effect.promise(() => current)
const next = sdk.config
.providers({ directory: params.directory }, { throwOnError: true })
.then((response) => {
const providers = Object.fromEntries(
(response.data?.providers ?? []).map((provider) => [provider.id, provider]),
) as Record<ProviderID, Provider.Info>
return UsageService.findContextLimit(providers, params.providerID, params.modelID)
})
.catch((error: unknown) => {
log.error("failed to get providers for usage context limit", { error })
return undefined
})
limits.set(key, next)
return yield* Effect.promise(() => next)
},
)
const sendUpdate: UsageService.Interface["sendUpdate"] = Effect.fn("ACPNext.promptUsage.sendUpdate")(
function* (params) {
const messages = yield* request(
() =>
sdk.session.messages(
{
sessionID: params.sessionID,
directory: params.directory,
},
{ throwOnError: true },
),
"session",
).pipe(
Effect.map((messages) => messages as readonly UsageService.SessionMessage[]),
Effect.catch((error) =>
Effect.sync(() => {
log.error("failed to fetch messages for usage update", { error })
return undefined
}),
),
)
if (!messages) return
const message = UsageService.latestAssistantMessage(messages)
if (!message?.providerID || !message.modelID) return
const size = yield* contextLimit({
directory: params.directory,
providerID: ProviderID.make(message.providerID),
modelID: ModelID.make(message.modelID),
})
if (!size) return
yield* Effect.promise(() =>
params.connection
.sessionUpdate({
sessionId: params.sessionID,
update: {
sessionUpdate: "usage_update",
used: message.tokens.input + message.tokens.cache.read,
size,
cost: { amount: UsageService.totalSessionCost(messages), currency: "USD" },
},
})
.catch((error) => {
log.error("failed to send usage update", { error })
}),
)
},
)
return UsageService.Service.of({
buildUsage: UsageService.buildUsage,
latestAssistantMessage: UsageService.latestAssistantMessage,
totalSessionCost: UsageService.totalSessionCost,
contextLimit,
sendUpdate,
})
}
function replayMessages(subscription: ACPNextEvent.Subscription | undefined, messages: SessionMessageResponse[]) {
if (!subscription) return Effect.void
return Effect.promise(async () => {
@@ -707,8 +506,6 @@ type MessageInfo = {
readonly agent?: Message["agent"]
}
type AssistantInfo = UsageService.AssistantTokenCost | undefined
function request<T>(fn: () => Promise<T | SdkResponse<T>>, service?: string) {
return Effect.tryPromise({
try: async () => {
@@ -723,79 +520,66 @@ function request<T>(fn: () => Promise<T | SdkResponse<T>>, service?: string) {
})
}
function profiledRequest<T>(name: string, fn: () => Promise<T | SdkResponse<T>>, service?: string) {
return request(() => ACPNextProfile.measure(name, fn), service)
}
async function loadDirectorySnapshot(sdk: OpencodeClient, directory: string) {
return ACPNextProfile.measure("acp.directory.load", async () => {
const [providersResponse, agentsResponse, commandsResponse, skillsResponse, configResponse] = await Promise.all([
ACPNextProfile.measure("acp.directory.provider.list", () =>
sdk.config.providers({ directory }, { throwOnError: true }),
),
ACPNextProfile.measure("acp.directory.mode.defaultAgent.load", () =>
sdk.app.agents({ directory }, { throwOnError: true }),
),
ACPNextProfile.measure("acp.directory.command.list", () =>
sdk.command.list({ directory }, { throwOnError: true }),
),
ACPNextProfile.measure("acp.directory.skill.list", () => sdk.app.skills({ directory }, { throwOnError: true })),
ACPNextProfile.measure("acp.directory.defaultModel.config", () =>
sdk.config.get({ directory }, { throwOnError: true }).catch(() => undefined),
),
])
const providersData = providersResponse.data!
const agents = agentsResponse.data!
const commandsData = commandsResponse.data!
const skills = skillsResponse.data!
const providers = Object.fromEntries(providersData.providers.map((provider) => [provider.id, provider])) as Record<
ProviderID,
Provider.Info
>
const defaultModelStarted = performance.now()
const defaultModel = defaultModelFromConfig(configResponse?.data?.model, providers)
ACPNextProfile.duration("acp.directory.defaultModel.resolve", defaultModelStarted, { configured: !!defaultModel })
const modes = agents
.filter((agent) => agent.mode !== "subagent" && agent.hidden !== true)
.map((agent) => ({
id: agent.name,
name: agent.name,
...(agent.description ? { description: agent.description } : {}),
}))
const commands = [
...commandsData,
...skills
.filter((skill) => !commandsData.some((command) => command.name === skill.name))
.map((skill) => ({
name: skill.name,
description: skill.description,
source: "skill" as const,
template: skill.content,
hints: [],
})),
] as Command.Info[]
const [providersResponse, agentsResponse, commandsResponse, skillsResponse] = await Promise.all([
sdk.config.providers({ directory }, { throwOnError: true }),
sdk.app.agents({ directory }, { throwOnError: true }),
sdk.command.list({ directory }, { throwOnError: true }),
sdk.app.skills({ directory }, { throwOnError: true }),
])
const providersData = providersResponse.data!
const agents = agentsResponse.data!
const commandsData = commandsResponse.data!
const skills = skillsResponse.data!
const providers = Object.fromEntries(providersData.providers.map((provider) => [provider.id, provider])) as Record<
ProviderID,
Provider.Info
>
const defaultModel = await defaultModelFromSdk(sdk, directory, providers)
const modes = agents
.filter((agent) => agent.mode !== "subagent" && agent.hidden !== true)
.map((agent) => ({
id: agent.name,
name: agent.name,
...(agent.description ? { description: agent.description } : {}),
}))
const commands = [
...commandsData,
...skills
.filter((skill) => !commandsData.some((command) => command.name === skill.name))
.map((skill) => ({
name: skill.name,
description: skill.description,
source: "skill" as const,
template: skill.content,
hints: [],
})),
] as Command.Info[]
return Directory.build({
directory,
providers,
modes,
defaultModeID: agents.find((agent) => agent.mode === "primary" && agent.hidden !== true)?.name ?? "build",
commands: commands.toSorted((a, b) => a.name.localeCompare(b.name)),
...(defaultModel ? { defaultModel } : {}),
})
return Directory.build({
directory,
providers,
modes,
defaultModeID: agents.find((agent) => agent.mode === "primary" && agent.hidden !== true)?.name ?? "build",
commands: commands.toSorted((a, b) => a.name.localeCompare(b.name)),
...(defaultModel ? { defaultModel } : {}),
})
}
function defaultModelFromConfig(
configuredModel: string | undefined,
async function defaultModelFromSdk(
sdk: OpencodeClient,
directory: string,
providers: Record<ProviderID, Provider.Info>,
): Directory.DefaultModel | undefined {
const configured = configuredModel ? Provider.parseModel(configuredModel) : undefined
): Promise<Directory.DefaultModel | undefined> {
const configured = await sdk.config
.get({ directory }, { throwOnError: true })
.then((response) => (response.data?.model ? Provider.parseModel(response.data.model) : undefined))
.catch(() => undefined)
if (configured && providers[configured.providerID]?.models[configured.modelID]) return configured
// First-session ACP startup must not scan historical sessions just to infer
// a default. Configured model, opencode provider, then sorted best model keep
// the protocol response deterministic without extra session/message reads.
const lastUsed = await lastUsedModel(sdk, directory, providers)
if (lastUsed) return lastUsed
const opencodeProvider = providers[ProviderID.make("opencode")]
const opencodeModel = opencodeProvider ? Provider.sort(Object.values(opencodeProvider.models))[0] : undefined
if (opencodeProvider && opencodeModel) return { providerID: opencodeProvider.id, modelID: opencodeModel.id }
@@ -805,6 +589,30 @@ function defaultModelFromConfig(
if (configured) return configured
}
async function lastUsedModel(
sdk: OpencodeClient,
directory: string,
providers: Record<ProviderID, Provider.Info>,
): Promise<Directory.DefaultModel | undefined> {
const session = await sdk.session
.list({ directory, roots: true, limit: 1 }, { throwOnError: true })
.then((response) => response.data?.[0])
.catch(() => undefined)
if (!session) return
const lastUser = await sdk.session
.messages({ directory, sessionID: session.id, limit: 20 }, { throwOnError: true })
.then((response) => response.data?.findLast((message) => message.info.role === "user")?.info)
.catch(() => undefined)
if (lastUser?.role !== "user") return
if (!providers[ProviderID.make(lastUser.model.providerID)]?.models[ModelID.make(lastUser.model.modelID)]) return
return {
providerID: ProviderID.make(lastUser.model.providerID),
modelID: ModelID.make(lastUser.model.modelID),
}
}
function selectDefaultModel(snapshot: Directory.Snapshot) {
if (snapshot.defaultModel) return snapshot.defaultModel
const model = snapshot.modelOptions[0]
@@ -812,43 +620,6 @@ function selectDefaultModel(snapshot: Directory.Snapshot) {
return { providerID: "unknown" as ProviderID, modelID: "unknown" as ModelID }
}
function detectSlashCommand(parts: ReturnType<typeof promptContentToParts>) {
const text = parts
.filter((part): part is Extract<(typeof parts)[number], { type: "text" }> => part.type === "text")
.map((part) => part.text)
.join("")
.trim()
if (!text.startsWith("/")) return
const [name, ...rest] = text.slice(1).split(/\s+/)
if (!name) return
return { name, args: rest.join(" ").trim() }
}
function promptResponse(info: AssistantInfo, messageId: string | null | undefined): PromptResponse {
return {
stopReason: "end_turn",
...(info ? { usage: UsageService.buildUsage(info) } : {}),
...(messageId ? { userMessageId: messageId } : {}),
_meta: {},
}
}
function sendUsageUpdate(
usage: UsageService.Interface | undefined,
sdk: OpencodeClient,
connection: ServiceConnection | undefined,
sessionID: string,
directory: string,
) {
if (!connection) return Effect.void
return (usage ?? makeUsageService(sdk)).sendUpdate({
connection,
sessionID,
directory,
})
}
function selectVariant(snapshot: Directory.Snapshot, model: Directory.DefaultModel) {
const variants = Directory.variants(snapshot, model)
if (!variants) return
@@ -919,7 +690,6 @@ function registerMcpServers(
sessionId: string,
servers: readonly McpServer[],
) {
const started = performance.now()
const current = registered.get(sessionId) ?? new Set<string>()
registered.set(sessionId, current)
const pending = new Set<string>()
@@ -951,16 +721,7 @@ function registerMcpServers(
),
),
{ concurrency: "unbounded" },
).pipe(
Effect.tap(() =>
Effect.sync(() =>
ACPNextProfile.duration("acp.mcp.register", started, {
count: pending.size,
}),
),
),
Effect.asVoid,
)
).pipe(Effect.asVoid)
}
function mcpRegistrationKey(name: string, config: ReturnType<typeof mcpConfig>) {
@@ -60,7 +60,6 @@ export type PartMetadataLookupInput = {
export type Interface = {
readonly create: (input: StoreInput) => Effect.Effect<Info>
readonly load: (input: StoreInput) => Effect.Effect<Info>
readonly list: (cwd?: string) => Effect.Effect<readonly Info[]>
readonly get: (sessionId: string) => Effect.Effect<Info, ACPNextError.SessionNotFoundError>
readonly tryGet: (sessionId: string) => Effect.Effect<Info | undefined>
readonly remove: (sessionId: string) => Effect.Effect<Info | undefined>
@@ -169,12 +168,6 @@ export const layer = Layer.effect(
return Service.of({
create: store,
load: store,
list: Effect.fn("ACPNext.Session.list")(function* (cwd?: string) {
return [...(yield* Ref.get(sessions)).values()]
.filter((session) => !cwd || session.cwd === cwd)
.map(snapshot)
.toSorted((a, b) => b.createdAt.getTime() - a.createdAt.getTime())
}),
get,
tryGet,
remove,
+1 -6
View File
@@ -9,7 +9,6 @@ import { ServerAuth } from "@/server/auth"
import { createOpencodeClient } from "@opencode-ai/sdk/v2"
import { withNetworkOptions, resolveNetworkOptions } from "../network"
import { RuntimeFlags } from "@/effect/runtime-flags"
import { ACPNextProfile } from "@/acp-next/profile"
const log = Log.create({ service: "acp-command" })
@@ -24,13 +23,10 @@ export const AcpCommand = effectCmd({
})
},
handler: Effect.fn("Cli.acp")(function* (args) {
ACPNextProfile.mark("cli.acp.handler")
process.env.OPENCODE_CLIENT = "acp"
const flags = yield* RuntimeFlags.Service
const opts = yield* resolveNetworkOptions(args)
const server = yield* Effect.promise(() =>
ACPNextProfile.measure("cli.acp.server.listen", () => Server.listen(opts)),
)
const server = yield* Effect.promise(() => Server.listen(opts))
const sdk = createOpencodeClient({
baseUrl: `http://${server.hostname}:${server.port}`,
@@ -64,7 +60,6 @@ export const AcpCommand = effectCmd({
const agent = flags.acpNext ? ACPNext.init({ sdk }) : ACP.init({ sdk })
new AgentSideConnection((conn) => {
ACPNextProfile.mark("cli.acp.connection.create", { acpNext: flags.acpNext })
return agent.create(conn, { sdk })
}, stream)
@@ -460,6 +460,13 @@ export const ProvidersLoginCommand = effectCmd({
)
}
if (provider === "google-vertex") {
yield* Prompt.log.info(
"Note: this API key cannot be used with Anthropic models.\n" +
"Use `gcloud auth application-default login` or set `GOOGLE_APPLICATION_CREDENTIALS`.",
)
}
if (provider === "opencode") {
yield* Prompt.log.info("Create an api key at https://opencode.ai/auth")
}
@@ -384,6 +384,15 @@ function ApiMethod(props: ApiMethodProps) {
</text>
</box>
),
"google-vertex": (
<box gap={1}>
<text fg={theme.warning}>Note: this API key cannot be used with Anthropic models.</text>
<text fg={theme.textMuted}>
Use <span style={{ fg: theme.text }}>gcloud auth application-default login</span> or set
GOOGLE_APPLICATION_CREDENTIALS.
</text>
</box>
),
}[props.providerID] ?? undefined
}
onConfirm={async (value) => {
@@ -442,25 +442,16 @@ function ReasoningHeader(props: { toggleable: boolean; open: boolean; done: bool
: theme.warning
return (
<Switch>
<Match when={!props.done}>
<box flexDirection="row">
<Spinner color={fg()}>{props.title ? "Thinking: " + props.title : "Thinking"}</Spinner>
</box>
</Match>
<Match when={true}>
<text fg={fg()} wrapMode="none">
<Show when={props.toggleable}>
<span>{props.open ? "- " : "+ "}</span>
</Show>
<span>Thought</span>
<Show when={props.title}>
<span>: </span>
<span>{props.title}</span>
</Show>
</text>
</Match>
</Switch>
<text fg={fg()} wrapMode="none">
<Show when={props.toggleable}>
<span>{props.open ? "- " : "+ "}</span>
</Show>
<span>{props.done ? "Thought" : "Thinking"}</span>
<Show when={props.title}>
<span>: </span>
<span>{props.title}</span>
</Show>
</text>
)
}
@@ -1595,33 +1595,24 @@ function ReasoningHeader(props: {
: theme.warning
return (
<Switch>
<Match when={!props.done}>
<box flexDirection="row">
<Spinner color={fg()}>{props.title ? "Thinking: " + props.title : "Thinking"}</Spinner>
</box>
</Match>
<Match when={true}>
<text fg={fg()} wrapMode="none">
<Show when={props.toggleable}>
<span>{props.open ? "- " : "+ "}</span>
</Show>
<span>Thought</span>
<Show when={props.title || props.duration}>
<span>: </span>
</Show>
<Show when={props.title}>
<span>{props.title}</span>
</Show>
<Show when={props.duration}>
<span>
{props.title ? " · " : ""}
{props.duration}
</span>
</Show>
</text>
</Match>
</Switch>
<text fg={fg()} wrapMode="none">
<Show when={props.toggleable}>
<span>{props.open ? "- " : "+ "}</span>
</Show>
<span>{props.done ? "Thought" : "Thinking"}</span>
<Show when={props.title || props.duration}>
<span>: </span>
</Show>
<Show when={props.title}>
<span>{props.title}</span>
</Show>
<Show when={props.duration}>
<span>
{props.title ? " · " : ""}
{props.duration}
</span>
</Show>
</text>
)
}
@@ -20,9 +20,9 @@ Enabled by default on `local`, `dev`, and `beta`. On `latest` and `prod`, set `O
## Retries
- Retry WebSocket stream/setup failures up to 5 times, then use HTTP for that session until the pool entry is idle-pruned.
- `websocket_connection_limit_reached` consumes the same retry budget and HTTP fallback.
- If a WebSocket fails after its first event, fail it as retryable rather than replaying partial output in transport.
- If WebSocket setup fails or it fails before its first event, replay over HTTP and keep that session on HTTP until idle-pruned.
- If the server returns `websocket_connection_limit_reached` before output, reconnect up to 5 times, then follow the same HTTP fallback.
- If a WebSocket fails after its first event, fail the stream. Do not replay partial output.
- Abort or cancel closes the socket.
## Next Steps
+25 -42
View File
@@ -1,6 +1,5 @@
import WebSocket from "ws"
import * as Log from "@opencode-ai/core/util/log"
import { ProviderError } from "@/provider/error"
import { isRecord } from "@/util/record"
import { OpenAIWebSocket } from "./ws"
@@ -14,7 +13,7 @@ export interface CreateWebSocketFetchOptions {
connectTimeout?: number
idleTimeout?: number
maxConnectionAge?: number
streamRetries?: number
connectionLimitRetries?: number
}
interface PoolEntry {
@@ -23,7 +22,6 @@ interface PoolEntry {
lastUsedAt: number
busy: boolean
fallback: boolean
streamFailures: number
}
const DEFAULT_CONNECT_TIMEOUT = 15_000
@@ -37,7 +35,7 @@ export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) {
const connectTimeout = options?.connectTimeout ?? DEFAULT_CONNECT_TIMEOUT
const idleTimeout = options?.idleTimeout ?? DEFAULT_IDLE_TIMEOUT
const maxConnectionAge = options?.maxConnectionAge ?? DEFAULT_MAX_CONNECTION_AGE
const streamRetries = options?.streamRetries ?? 5
const connectionLimitRetries = options?.connectionLimitRetries ?? 5
const pruneTimer = setInterval(() => prune(), Math.min(idleTimeout, 60_000))
if (typeof pruneTimer === "object" && "unref" in pruneTimer && typeof pruneTimer.unref === "function") {
pruneTimer.unref()
@@ -74,7 +72,7 @@ export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) {
}
const key = `${sessionID}:conversation`
const entry = pool.get(key) ?? { lastUsedAt: Date.now(), busy: false, fallback: false, streamFailures: 0 }
const entry = pool.get(key) ?? { lastUsedAt: Date.now(), busy: false, fallback: false }
pool.set(key, entry)
if (entry.fallback) {
@@ -89,6 +87,7 @@ export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) {
entry.busy = true
entry.lastUsedAt = Date.now()
try {
let connectionLimitAttempts = 0
entry.socket = await socket(
entry,
options?.url ?? url,
@@ -112,16 +111,15 @@ export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) {
onTerminal: (event) => {
entry.busy = false
entry.lastUsedAt = Date.now()
entry.streamFailures = 0
if (event.type !== "response.completed" && event.type !== "response.done") {
log.warn("websocket terminal failure", { key, type: event.type })
invalidate(entry)
}
},
onConnectionInvalid: (error) => {
log.warn("websocket invalidated", { key, error: error.message })
log.warn("websocket invalidated", { key, error: error instanceof Error ? error.message : String(error) })
entry.busy = false
if (!entry.fallback) recordStreamFailure(entry)
entry.fallback = true
invalidate(entry)
resolveFirstEvent(false)
},
@@ -129,52 +127,51 @@ export function createWebSocketFetch(options?: CreateWebSocketFetchOptions) {
log.debug("websocket aborted", { key })
entry.busy = false
entry.lastUsedAt = Date.now()
entry.streamFailures = 0
invalidate(entry)
rejectFirstEvent(error)
},
onRetryableTerminal: async (event) => {
const error = connectionLimitError(event)
if (!error) return undefined
log.warn("websocket connection limit reached", { key })
throw error
if (connectionLimitAttempts >= connectionLimitRetries) throw error
connectionLimitAttempts++
log.warn("websocket connection limit reached", { key, attempt: connectionLimitAttempts })
invalidate(entry)
entry.socket = await socket(
entry,
options?.url ?? url,
OpenAIWebSocket.normalizeHeaders(httpInit?.headers),
connectTimeout,
maxConnectionAge,
init?.signal,
)
entry.lastUsedAt = Date.now()
return entry.socket
},
})
if (await firstEvent) return response
if (!entry.fallback) return response
log.debug("http fallback", { key, reason: "websocket_retries_exhausted" })
log.debug("http fallback", { key, reason: "websocket_failed_before_first_event" })
return httpFetch(input, httpInit)
} catch (error) {
entry.busy = false
entry.lastUsedAt = Date.now()
if (OpenAIWebSocket.isAbortError(error)) {
entry.streamFailures = 0
invalidate(entry)
throw error
}
recordStreamFailure(entry)
entry.fallback = true
log.warn("websocket setup failed", {
key,
error: error instanceof Error ? error.message : String(error),
fallback: entry.fallback ? "http" : undefined,
fallback: "http",
})
invalidate(entry)
if (entry.fallback) return httpFetch(input, httpInit)
return failedResponse(
new ProviderError.ResponseStreamError(error instanceof Error ? error.message : String(error), {
cause: error,
}),
)
return httpFetch(input, httpInit)
}
}
function recordStreamFailure(entry: PoolEntry) {
entry.streamFailures++
// Codex counts retries after the initial failed WebSocket attempt.
if (entry.streamFailures > streamRetries) entry.fallback = true
}
function prune() {
const now = Date.now()
for (const [key, entry] of pool) {
@@ -201,20 +198,6 @@ function connectionLimitError(event: Record<string, unknown>) {
return new Error(typeof event.error.message === "string" ? event.error.message : CONNECTION_LIMIT_REACHED_CODE)
}
function failedResponse(error: ProviderError.ResponseStreamError) {
return new Response(
new ReadableStream({
start(controller) {
controller.error(error)
},
}),
{
status: 200,
headers: { "content-type": "text/event-stream" },
},
)
}
async function socket(
entry: PoolEntry,
url: string,
+16 -21
View File
@@ -2,8 +2,6 @@
// fallback, and continuation state intentionally live above this file.
import WebSocket from "ws"
import { ProviderError } from "@/provider/error"
import { errorMessage } from "@/util/error"
export const PROTOCOL_HEADER = "responses_websockets=2026-02-06"
@@ -23,7 +21,7 @@ export interface StreamResponsesWebSocketOptions {
onComplete?: (event: Record<string, unknown>) => void
onTerminal?: (event: Record<string, unknown>) => void
onRetryableTerminal?: (event: Record<string, unknown>) => Promise<WebSocket | undefined>
onConnectionInvalid?: (error: ProviderError.ResponseStreamError) => void
onConnectionInvalid?: (error: Error) => void
onAbort?: (error: Error) => void
}
@@ -95,15 +93,15 @@ export function connectResponsesWebSocket(options: ConnectResponsesWebSocketOpti
resolve(socket)
}
function onError(error: unknown) {
function onError(error: Error) {
socket.on("error", () => {})
cleanup()
reject(error instanceof Error ? error : new Error(errorMessage(error), { cause: error }))
reject(error)
}
function onClose(code: number, reason: Buffer) {
cleanup()
reject(new Error(closeMessage("WebSocket closed before open", code, reason)))
reject(closeError("WebSocket closed before open", code, reason))
}
function onAbort() {
@@ -147,7 +145,7 @@ export function streamResponsesWebSocket(options: StreamResponsesWebSocketOption
controller?.close()
}
function invalidate(error: ProviderError.ResponseStreamError) {
function invalidate(error: Error) {
if (completed) return
completed = true
cleanup()
@@ -159,13 +157,16 @@ export function streamResponsesWebSocket(options: StreamResponsesWebSocketOption
if (completed) return
if (!options.idleTimeout) return
if (idleTimer) clearTimeout(idleTimer)
idleTimer = setTimeout(() => invalidate(new ProviderError.ResponseStreamError(message)), options.idleTimeout)
idleTimer = setTimeout(() => invalidate(new Error(message)), options.idleTimeout)
if (typeof idleTimer === "object" && "unref" in idleTimer && typeof idleTimer.unref === "function") {
idleTimer.unref()
}
}
async function onMessage(data: WebSocket.RawData, isBinary: boolean) {
if (completed) return
if (isBinary) {
invalidate(new ProviderError.ResponseStreamError("Unexpected binary WebSocket frame"))
invalidate(new Error("Unexpected binary WebSocket frame"))
return
}
@@ -194,11 +195,7 @@ export function streamResponsesWebSocket(options: StreamResponsesWebSocketOption
return
}
} catch (error) {
invalidate(
new ProviderError.ResponseStreamError(error instanceof Error ? error.message : String(error), {
cause: error,
}),
)
invalidate(error instanceof Error ? error : new Error(String(error)))
return
}
}
@@ -233,14 +230,12 @@ export function streamResponsesWebSocket(options: StreamResponsesWebSocketOption
}
function onError(error: Error) {
invalidate(new ProviderError.ResponseStreamError(error.message, { cause: error }))
invalidate(error)
}
function onClose(code: number, reason: Buffer) {
if (completed) return
invalidate(
new ProviderError.ResponseStreamError(closeMessage("WebSocket closed before response.completed", code, reason)),
)
invalidate(closeError("WebSocket closed before response.completed", code, reason))
}
function onAbort() {
@@ -277,7 +272,7 @@ export function streamResponsesWebSocket(options: StreamResponsesWebSocketOption
socket.send(JSON.stringify({ type: "response.create", ...payload }), (error) => {
if (completed) return
resetIdleTimeout("idle timeout waiting for websocket")
if (error) invalidate(new ProviderError.ResponseStreamError(error.message, { cause: error }))
if (error) invalidate(error)
})
}
@@ -317,11 +312,11 @@ function abortError(signal: AbortSignal | undefined) {
return new DOMException(reason instanceof Error ? reason.message : "Aborted", "AbortError")
}
function closeMessage(message: string, code: number, reason: Buffer) {
function closeError(message: string, code: number, reason: Buffer) {
const details = [`code ${code}`]
if (code === 1009) details.push("message too big")
if (reason.length > 0) details.push(reason.toString())
return `${message} (${details.join(": ")})`
return new Error(`${message} (${details.join(": ")})`)
}
export * as OpenAIWebSocket from "./ws"
-8
View File
@@ -11,14 +11,6 @@ export class HeaderTimeoutError extends Error {
}
}
export class ResponseStreamError extends Error {
public override readonly name = "ProviderResponseStreamError"
constructor(message: string, options?: ErrorOptions) {
super(message, options)
}
}
// Adapted from overflow detection patterns in:
// https://github.com/badlogic/pi-mono/blob/main/packages/ai/src/utils/overflow.ts
const OVERFLOW_PATTERNS = [
+7 -12
View File
@@ -596,16 +596,8 @@ function openaiCompatibleReasoningEfforts(id: string) {
return gpt5CodexReasoningEfforts(apiId) ?? versionedGpt5ReasoningEfforts(apiId) ?? OPENAI_EFFORTS
}
function anthropicOpus47OrLater(apiId: string) {
const version = /opus-(\d+)[.-](\d+)(?:[.-]|$)/i.exec(apiId)
if (!version) return false
const major = Number(version[1])
const minor = Number(version[2])
return major > 4 || (major === 4 && minor >= 7)
}
function anthropicAdaptiveEfforts(apiId: string): string[] | null {
if (anthropicOpus47OrLater(apiId)) {
if (["opus-4-7", "opus-4.7"].some((v) => apiId.includes(v))) {
return ["low", "medium", "high", "xhigh", "max"]
}
if (["opus-4-6", "opus-4.6", "sonnet-4-6", "sonnet-4.6"].some((v) => apiId.includes(v))) {
@@ -633,7 +625,6 @@ export function variants(model: Provider.Model): Record<string, Record<string, a
if (!model.capabilities.reasoning) return {}
const id = model.id.toLowerCase()
const adaptiveOpus = anthropicOpus47OrLater(model.api.id)
const adaptiveEfforts = anthropicAdaptiveEfforts(model.api.id)
if (
id.includes("deepseek-chat") ||
@@ -842,7 +833,9 @@ export function variants(model: Provider.Model): Record<string, Record<string, a
{
thinking: {
type: "adaptive",
...(adaptiveOpus ? { display: "summarized" } : {}),
...(model.api.id.includes("opus-4-7") || model.api.id.includes("opus-4.7")
? { display: "summarized" }
: {}),
},
effort,
},
@@ -879,7 +872,9 @@ export function variants(model: Provider.Model): Record<string, Record<string, a
reasoningConfig: {
type: "adaptive",
maxReasoningEffort: effort,
...(adaptiveOpus ? { display: "summarized" } : {}),
...(model.api.id.includes("opus-4-7") || model.api.id.includes("opus-4.7")
? { display: "summarized" }
: {}),
},
},
]),
@@ -63,11 +63,11 @@ function decodeCredential(input: string) {
Effect.match({
onFailure: emptyCredential,
onSuccess: (header) => {
const separator = header.indexOf(":")
if (separator === -1) return emptyCredential()
const parts = header.split(":")
if (parts.length !== 2) return emptyCredential()
return {
username: header.slice(0, separator),
password: Redacted.make(header.slice(separator + 1)),
username: parts[0],
password: Redacted.make(parts[1]),
}
},
}),
+16 -14
View File
@@ -1095,8 +1095,9 @@ export function latest(msgs: WithParts[]) {
export function fromError(
e: unknown,
ctx: { providerID: ProviderID; aborted?: boolean },
ctx: { model: Provider.Model; aborted?: boolean } | { providerID: ProviderID; aborted?: boolean },
): NonNullable<Assistant["error"]> {
const providerID = "model" in ctx ? ctx.model.providerID : ctx.providerID
switch (true) {
case e instanceof DOMException && e.name === "AbortError":
return new AbortedError(
@@ -1110,11 +1111,23 @@ export function fromError(
case LoadAPIKeyError.isInstance(e):
return new AuthError(
{
providerID: ctx.providerID,
providerID,
message: e.message,
},
{ cause: e },
).toObject()
case e instanceof Error &&
"model" in ctx &&
ctx.model.api.npm === "@ai-sdk/google-vertex/anthropic" &&
e.message.includes("Could not load the default credentials"):
return new AuthError(
{
providerID,
message:
"Anthropic models on Google Vertex require Google Cloud credentials. Use `gcloud auth application-default login` or set `GOOGLE_APPLICATION_CREDENTIALS`.",
},
{ cause: e },
).toObject()
case (e as SystemError)?.code === "ECONNRESET":
return new APIError(
{
@@ -1155,20 +1168,9 @@ export function fromError(
},
{ cause: e },
).toObject()
case e instanceof ProviderError.ResponseStreamError:
return new APIError(
{
message: e.message,
isRetryable: true,
metadata: {
code: e.name,
},
},
{ cause: e },
).toObject()
case APICallError.isInstance(e):
const parsed = ProviderError.parseAPICallError({
providerID: ctx.providerID,
providerID,
error: e,
})
if (parsed.type === "context_overflow") {
+1 -1
View File
@@ -124,7 +124,7 @@ export const layer = Layer.effect(
const parse = (e: unknown) =>
MessageV2.fromError(e, {
providerID: input.model.providerID,
model: input.model,
aborted,
})
@@ -4,7 +4,6 @@ import type {
ForkSessionResponse,
LoadSessionResponse,
NewSessionResponse,
SessionNotification,
ResumeSessionResponse,
SessionConfigOption,
SessionConfigSelectOption,
@@ -15,7 +14,6 @@ import { Effect, ManagedRuntime } from "effect"
import * as ACPNextService from "@/acp-next/service"
import * as ACPNextError from "@/acp-next/error"
import { ACPNextSession } from "@/acp-next/session"
import { UsageService } from "@/acp-next/usage"
import { ModelID, ProviderID } from "@/provider/schema"
import type { Provider } from "@/provider/provider"
@@ -143,14 +141,10 @@ const provider: Provider.Info = {
describe("ACP next service sessions", () => {
const makeService = (messages: readonly { info: unknown; parts: readonly unknown[] }[] = []) => {
const updates: SessionNotification[] = []
const updates: unknown[] = []
const mcpAdds: string[] = []
const aborts: string[] = []
const forks: string[] = []
const prompts: unknown[] = []
const commands: unknown[] = []
const summarizes: unknown[] = []
const usageUpdates: string[] = []
const sessions = Array.from({ length: 102 }, (_, index) => ({
id: `ses_${index + 1}`,
directory: index % 2 === 0 ? "/workspace" : "/other",
@@ -190,36 +184,6 @@ describe("ACP next service sessions", () => {
data: input.directory ? sessions.filter((session) => session.directory === input.directory) : sessions,
}),
messages: () => Promise.resolve({ data: messages }),
prompt: (input: unknown) => {
prompts.push(input)
return Promise.resolve({
data: {
info: assistantInfo({
input: 100,
output: 40,
reasoning: 7,
cache: { read: 11, write: 13 },
}),
},
})
},
command: (input: unknown) => {
commands.push(input)
return Promise.resolve({
data: {
info: assistantInfo({
input: 3,
output: 4,
reasoning: 0,
cache: { read: 0, write: 0 },
}),
},
})
},
summarize: (input: unknown) => {
summarizes.push(input)
return Promise.resolve({ data: true })
},
abort: (input: { sessionID: string }) => {
aborts.push(input.sessionID)
return Promise.resolve({ data: true })
@@ -237,33 +201,13 @@ describe("ACP next service sessions", () => {
},
} as unknown as OpencodeClient
const connection = {
sessionUpdate: (update: SessionNotification) => {
sessionUpdate: (update: unknown) => {
updates.push(update)
return Promise.resolve()
},
} as Pick<AgentSideConnection, "sessionUpdate">
const usage = UsageService.Service.of({
buildUsage: UsageService.buildUsage,
latestAssistantMessage: UsageService.latestAssistantMessage,
totalSessionCost: UsageService.totalSessionCost,
contextLimit: () => Effect.succeed(128000),
sendUpdate: (input) =>
Effect.sync(() => {
usageUpdates.push(input.sessionID)
}),
})
return {
service: ACPNextService.make({ sdk, connection, usage }),
updates,
mcpAdds,
aborts,
forks,
prompts,
commands,
summarizes,
usageUpdates,
}
return { service: ACPNextService.make({ sdk, connection }), updates, mcpAdds, aborts, forks }
}
it("creates a backed session with config options and command update", async () => {
@@ -323,15 +267,6 @@ describe("ACP next service sessions", () => {
expect(second.sessions).toEqual(first.sessions)
})
it("includes live ACP sessions before they appear in server-backed session list", async () => {
const { service } = makeService()
const created = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
const listed = await Effect.runPromise(service.listSessions({ cwd: "/workspace" }))
expect(listed.sessions[0]?.sessionId).toBe(created.sessionId)
expect(listed.sessions[0]?.cwd).toBe("/workspace")
})
it("lists all sessions with next cursor when the first page is full", async () => {
const { service } = makeService()
const first = await Effect.runPromise(service.listSessions({}))
@@ -611,46 +546,6 @@ describe("ACP next service sessions", () => {
expect(result.configOptions?.find((option) => option.id === "model")?.currentValue).toBe("test/configured-model")
})
it("does not scan last-used sessions when resolving the new session default", async () => {
const historyCalls: string[] = []
const sdk = {
config: {
providers: () => Promise.resolve({ data: { providers: [provider], default: { test: modelID } } }),
get: () => Promise.resolve({ data: {} }),
},
app: {
agents: () => Promise.resolve({ data: [{ name: "build", mode: "primary", permission: [], options: {} }] }),
skills: () => Promise.resolve({ data: [] }),
},
command: {
list: () => Promise.resolve({ data: [] }),
},
session: {
create: (input: { model?: { id?: string } }) => Promise.resolve({ data: { id: input.model?.id } }),
list: () => {
historyCalls.push("list")
return Promise.resolve({ data: [{ id: "ses_recent" }] })
},
messages: () => {
historyCalls.push("messages")
return Promise.resolve({
data: [{ info: { role: "user", model: { providerID: "test", modelID: "second-model" } } }],
})
},
},
mcp: {
add: () => Promise.resolve({ data: {} }),
},
} as unknown as OpencodeClient
const service = ACPNextService.make({ sdk })
const result = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
expect(result.sessionId).toBe("test-model")
expect(result.configOptions?.find((option) => option.id === "model")?.currentValue).toBe("test/test-model")
expect(historyCalls).toEqual([])
})
it("switches model and returns updated model and effort options", async () => {
const { service } = makeService()
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
@@ -716,35 +611,24 @@ describe("ACP next service sessions", () => {
expect(results.map((error) => error.code)).toEqual([-32602, -32602, -32602, -32602])
})
it("does not refetch providers modes or commands when switching effort from session snapshot", async () => {
const calls = {
providers: 0,
agents: 0,
commands: 0,
skills: 0,
mcpAdds: 0,
}
it("does not reload providers or commands when switching effort from a warm snapshot", async () => {
let providersCalls = 0
let commandCalls = 0
const sdk = {
config: {
providers: () => {
calls.providers++
providersCalls++
return Promise.resolve({ data: { providers: [provider], default: { test: modelID } } })
},
get: () => Promise.resolve({ data: {} }),
},
app: {
agents: () => {
calls.agents++
return Promise.resolve({ data: [{ name: "build", mode: "primary", permission: [], options: {} }] })
},
skills: () => {
calls.skills++
return Promise.resolve({ data: [] })
},
agents: () => Promise.resolve({ data: [{ name: "build", mode: "primary", permission: [], options: {} }] }),
skills: () => Promise.resolve({ data: [] }),
},
command: {
list: () => {
calls.commands++
commandCalls++
return Promise.resolve({ data: [] })
},
},
@@ -753,16 +637,14 @@ describe("ACP next service sessions", () => {
list: () => Promise.resolve({ data: [] }),
},
mcp: {
add: () => {
calls.mcpAdds++
return Promise.resolve({ data: {} })
},
add: () => Promise.resolve({ data: {} }),
},
} as unknown as OpencodeClient
const service = ACPNextService.make({ sdk })
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
expect(calls).toEqual({ providers: 1, agents: 1, commands: 1, skills: 1, mcpAdds: 0 })
expect(providersCalls).toBe(1)
expect(commandCalls).toBe(1)
await Effect.runPromise(
service.setSessionConfigOption({
@@ -772,350 +654,11 @@ describe("ACP next service sessions", () => {
}),
)
expect(calls).toEqual({ providers: 1, agents: 1, commands: 1, skills: 1, mcpAdds: 0 })
})
it("switches model against the warm provider snapshot without refetching", async () => {
const calls = {
providers: 0,
agents: 0,
commands: 0,
skills: 0,
}
const sdk = {
config: {
providers: () => {
calls.providers++
return Promise.resolve({ data: { providers: [provider], default: { test: modelID } } })
},
get: () => Promise.resolve({ data: {} }),
},
app: {
agents: () => {
calls.agents++
return Promise.resolve({ data: [{ name: "build", mode: "primary", permission: [], options: {} }] })
},
skills: () => {
calls.skills++
return Promise.resolve({ data: [] })
},
},
command: {
list: () => {
calls.commands++
return Promise.resolve({ data: [] })
},
},
session: {
create: () => Promise.resolve({ data: { id: "ses_model_fast" } }),
list: () => Promise.resolve({ data: [] }),
},
mcp: {
add: () => Promise.resolve({ data: {} }),
},
} as unknown as OpencodeClient
const service = ACPNextService.make({ sdk })
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
const updated = await Effect.runPromise(
service.setSessionConfigOption({
sessionId: session.sessionId,
configId: "model",
value: "test/second-model",
}),
)
expect(select(updated, "model")?.currentValue).toBe("test/second-model")
expect(calls).toEqual({ providers: 1, agents: 1, commands: 1, skills: 1 })
})
it("reuses the warm directory snapshot for a second new session in the same cwd", async () => {
const calls = {
providers: 0,
config: 0,
agents: 0,
commands: 0,
skills: 0,
sessionList: 0,
messages: 0,
creates: 0,
}
const sdk = {
config: {
providers: () => {
calls.providers++
return Promise.resolve({ data: { providers: [provider], default: { test: modelID } } })
},
get: () => {
calls.config++
return Promise.resolve({ data: {} })
},
},
app: {
agents: () => {
calls.agents++
return Promise.resolve({ data: [{ name: "build", mode: "primary", permission: [], options: {} }] })
},
skills: () => {
calls.skills++
return Promise.resolve({ data: [] })
},
},
command: {
list: () => {
calls.commands++
return Promise.resolve({ data: [] })
},
},
session: {
create: () => {
calls.creates++
return Promise.resolve({ data: { id: `ses_warm_${calls.creates}` } })
},
list: () => {
calls.sessionList++
return Promise.resolve({ data: [] })
},
messages: () => {
calls.messages++
return Promise.resolve({ data: [] })
},
},
mcp: {
add: () => Promise.resolve({ data: {} }),
},
} as unknown as OpencodeClient
const service = ACPNextService.make({ sdk })
const first = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
const second = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
expect(first.sessionId).toBe("ses_warm_1")
expect(second.sessionId).toBe("ses_warm_2")
expect(calls).toEqual({
providers: 1,
config: 1,
agents: 1,
commands: 1,
skills: 1,
sessionList: 0,
messages: 0,
creates: 2,
})
})
it("normal text prompt sends model variant mode and converted parts", async () => {
const { service, prompts, usageUpdates } = makeService()
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
await Effect.runPromise(
service.setSessionConfigOption({
sessionId: session.sessionId,
configId: "effort",
value: "high",
}),
)
await Effect.runPromise(
service.setSessionConfigOption({
sessionId: session.sessionId,
configId: "mode",
value: "plan",
}),
)
const result = await Effect.runPromise(
service.prompt({
sessionId: session.sessionId,
messageId: "00000000-0000-4000-8000-000000000001",
prompt: [{ type: "text", text: "hello" }],
}),
)
expect(prompts).toEqual([
{
sessionID: session.sessionId,
model: { providerID, modelID },
variant: "high",
parts: [{ type: "text", text: "hello" }],
agent: "plan",
directory: "/workspace",
},
])
expect(result).toEqual({
stopReason: "end_turn",
usage: {
inputTokens: 100,
outputTokens: 40,
thoughtTokens: 7,
cachedReadTokens: 11,
cachedWriteTokens: 13,
totalTokens: 171,
},
userMessageId: "00000000-0000-4000-8000-000000000001",
_meta: {},
})
expect(usageUpdates).toEqual([session.sessionId])
})
it("prompt maps assistant and user audience annotations", async () => {
const { service, prompts } = makeService()
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
await Effect.runPromise(
service.prompt({
sessionId: session.sessionId,
prompt: [
{ type: "text", text: "assistant context", annotations: { audience: ["assistant"] } },
{ type: "text", text: "user context", annotations: { audience: ["user"] } },
],
}),
)
expect(prompts).toContainEqual({
sessionID: session.sessionId,
model: { providerID, modelID },
variant: "default",
parts: [
{ type: "text", text: "assistant context", synthetic: true },
{ type: "text", text: "user context", ignored: true },
],
agent: "build",
directory: "/workspace",
})
})
it("prompt sends image and resource parts", async () => {
const { service, prompts } = makeService()
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
await Effect.runPromise(
service.prompt({
sessionId: session.sessionId,
prompt: [
{ type: "image", data: "AAAA", mimeType: "image/png", uri: "file:///tmp/screenshot.png" },
{
type: "resource",
resource: {
uri: "file:///tmp/report.pdf",
mimeType: "application/pdf",
blob: "JVBERg==",
},
},
],
}),
)
expect((prompts[0] as { parts?: unknown }).parts).toEqual([
{
type: "file",
url: "data:image/png;base64,AAAA",
filename: "screenshot.png",
mime: "image/png",
},
{
type: "file",
url: "data:application/pdf;base64,JVBERg==",
filename: "report.pdf",
mime: "application/pdf",
},
])
})
it("slash command prompt calls session command", async () => {
const { service, prompts, commands } = makeService()
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
const result = await Effect.runPromise(
service.prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "/init now" }] }),
)
expect(prompts).toEqual([])
expect(commands).toEqual([
{
sessionID: session.sessionId,
command: "init",
arguments: "now",
model: "test/test-model",
variant: "default",
agent: "build",
directory: "/workspace",
},
])
expect(result.usage).toEqual({ inputTokens: 3, outputTokens: 4, totalTokens: 7 })
})
it("compact slash command calls summarize path", async () => {
const { service, prompts, commands, summarizes } = makeService()
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
await Effect.runPromise(
service.prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "/compact" }] }),
)
expect(prompts).toEqual([])
expect(commands).toEqual([])
expect(summarizes).toEqual([
{
sessionID: session.sessionId,
directory: "/workspace",
providerID,
modelID,
},
])
})
it("maps prompt auth failures to auth-required request errors", async () => {
const { service } = makeService()
const session = await Effect.runPromise(service.newSession({ cwd: "/workspace", mcpServers: [] }))
const failing = ACPNextService.make({
sdk: {
config: {
providers: () => Promise.resolve({ data: { providers: [provider], default: { test: modelID } } }),
get: () => Promise.resolve({ data: {} }),
},
app: {
agents: () => Promise.resolve({ data: [{ name: "build", mode: "primary", permission: [], options: {} }] }),
skills: () => Promise.resolve({ data: [] }),
},
command: {
list: () => Promise.resolve({ data: [] }),
},
session: {
create: () => Promise.resolve({ data: { id: session.sessionId } }),
list: () => Promise.resolve({ data: [] }),
prompt: () => Promise.reject({ name: "ProviderAuthError", data: { providerID: "test" } }),
},
mcp: {
add: () => Promise.resolve({ data: {} }),
},
} as unknown as OpencodeClient,
usage: UsageService.Service.of({
buildUsage: UsageService.buildUsage,
latestAssistantMessage: UsageService.latestAssistantMessage,
totalSessionCost: UsageService.totalSessionCost,
contextLimit: () => Effect.succeed(128000),
sendUpdate: () => Effect.void,
}),
})
await Effect.runPromise(failing.newSession({ cwd: "/workspace", mcpServers: [] }))
const error = await Effect.runPromise(
failing
.prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "hello" }] })
.pipe(Effect.mapError(ACPNextError.toRequestError), Effect.flip),
)
expect(error.code).toBe(-32000)
expect(providersCalls).toBe(1)
expect(commandCalls).toBe(1)
})
})
function assistantInfo(tokens: UsageService.AssistantTokenCost["tokens"]): UsageService.AssistantMessage {
return {
role: "assistant",
providerID: "test",
modelID: "test-model",
cost: 0,
tokens,
}
}
function categories(result: NewSessionResponse | LoadSessionResponse) {
return result.configOptions?.map((option) => option.category) ?? []
}
@@ -0,0 +1,282 @@
import { describe, expect } from "bun:test"
import type {
AuthenticateResponse,
CloseSessionResponse,
InitializeResponse,
LoadSessionResponse,
NewSessionResponse,
ResumeSessionResponse,
SessionNotification,
SetSessionConfigOptionResponse,
} from "@agentclientprotocol/sdk"
import { Effect } from "effect"
import { cliIt } from "../../lib/cli-process"
import { testProviderConfig } from "../../lib/test-provider"
import { createAcpClient, expectOk, firstAlternateValue, selectConfigOption } from "../acp/acp-test-client"
describe("opencode acp-next (subprocess)", () => {
cliIt.live(
"responds to initialize behind OPENCODE_ACP_NEXT",
({ opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(yield* opencode.acp({ env: { OPENCODE_ACP_NEXT: "1" } }))
const initialized = expectOk(
yield* acp.request<InitializeResponse>("initialize", {
protocolVersion: 1,
clientCapabilities: { _meta: { "terminal-auth": true } },
}),
)
expect(initialized.protocolVersion).toBe(1)
expect(initialized.agentCapabilities?.promptCapabilities?.embeddedContext).toBe(true)
expect(initialized.agentCapabilities?.promptCapabilities?.image).toBe(true)
expect(initialized.agentCapabilities?.mcpCapabilities?.http).toBe(true)
expect(initialized.agentCapabilities?.mcpCapabilities?.sse).toBe(true)
expect(initialized.agentCapabilities?.loadSession).toBe(true)
expect(initialized.agentCapabilities?.sessionCapabilities?.close).toEqual({})
expect(initialized.agentCapabilities?.sessionCapabilities?.fork).toEqual({})
expect(initialized.agentCapabilities?.sessionCapabilities?.list).toEqual({})
expect(initialized.agentCapabilities?.sessionCapabilities?.resume).toEqual({})
expect(initialized.agentInfo?.name).toBe("OpenCode")
expect(initialized.authMethods?.[0]?.id).toBe("opencode-login")
expect(initialized.authMethods?.[0]?._meta?.["terminal-auth"]).toBeDefined()
}),
60_000,
)
cliIt.live(
"authenticate succeeds for the advertised auth method and rejects unknown methods safely",
({ opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(yield* opencode.acp({ env: { OPENCODE_ACP_NEXT: "1" } }))
const initialized = expectOk(yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 }))
const methodId = initialized.authMethods?.[0]?.id
expect(methodId).toBe("opencode-login")
expectOk(yield* acp.request<AuthenticateResponse>("authenticate", { methodId }))
const rejected = yield* acp.request<AuthenticateResponse>("authenticate", { methodId: "missing-auth-method" })
expect(errorCode(rejected.error)).toBe(-32602)
}),
60_000,
)
cliIt.live(
"creates and loads sessions behind OPENCODE_ACP_NEXT",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(
yield* opencode.acp({
env: {
OPENCODE_ACP_NEXT: "1",
OPENCODE_CONFIG_CONTENT: JSON.stringify(testProviderConfig(llm.url)),
},
}),
)
yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 })
const session = expectOk(yield* acp.request<NewSessionResponse>("session/new", { cwd: home, mcpServers: [] }))
expect(typeof session.sessionId).toBe("string")
expect(selectConfigOption(session.configOptions, "model")?.category).toBe("model")
const update = yield* acp.waitForNotification<SessionNotification>(
"session/update",
(params) =>
params.sessionId === session.sessionId && params.update.sessionUpdate === "available_commands_update",
)
expect(update.params?.sessionId).toBe(session.sessionId)
const loaded = expectOk(
yield* acp.request<LoadSessionResponse>("session/load", {
cwd: home,
sessionId: session.sessionId,
mcpServers: [],
}),
)
expect(selectConfigOption(loaded.configOptions, "model")?.category).toBe("model")
const prompt = yield* acp.request("session/prompt", {
sessionId: "ses_missing",
prompt: [{ type: "text", text: "hello" }],
})
expect(errorCode(prompt.error)).toBe(-32601)
}),
60_000,
)
cliIt.live(
"switches model through config options behind OPENCODE_ACP_NEXT",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(
yield* opencode.acp({
env: {
OPENCODE_ACP_NEXT: "1",
OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)),
},
}),
)
yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 })
const session = expectOk(yield* acp.request<NewSessionResponse>("session/new", { cwd: home, mcpServers: [] }))
const updated = expectOk(
yield* acp.request<SetSessionConfigOptionResponse>("session/set_config_option", {
sessionId: session.sessionId,
configId: "model",
value: "test/second-model",
}),
)
expect(selectConfigOption(updated.configOptions, "model")?.currentValue).toBe("test/second-model")
}),
60_000,
)
cliIt.live(
"switches effort through config options behind OPENCODE_ACP_NEXT",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(
yield* opencode.acp({
env: {
OPENCODE_ACP_NEXT: "1",
OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)),
},
}),
)
yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 })
const session = expectOk(yield* acp.request<NewSessionResponse>("session/new", { cwd: home, mcpServers: [] }))
const effort = selectConfigOption(session.configOptions, "effort")
expect(effort?.category).toBe("thought_level")
const nextEffort = effort ? firstAlternateValue(effort) : undefined
expect(nextEffort).toBe("high")
const updated = expectOk(
yield* acp.request<SetSessionConfigOptionResponse>("session/set_config_option", {
sessionId: session.sessionId,
configId: "effort",
value: nextEffort,
}),
)
expect(selectConfigOption(updated.configOptions, "effort")?.currentValue).toBe(nextEffort)
}),
60_000,
)
cliIt.live(
"advertises and supports close behind OPENCODE_ACP_NEXT",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(
yield* opencode.acp({
env: {
OPENCODE_ACP_NEXT: "1",
OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)),
},
}),
)
const initialized = expectOk(yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 }))
expect(initialized.agentCapabilities?.sessionCapabilities?.close).toEqual({})
const session = expectOk(yield* acp.request<NewSessionResponse>("session/new", { cwd: home, mcpServers: [] }))
expectOk(yield* acp.request<CloseSessionResponse>("session/close", { sessionId: session.sessionId }))
}),
60_000,
)
cliIt.live(
"advertises and supports resume behind OPENCODE_ACP_NEXT",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(
yield* opencode.acp({
env: {
OPENCODE_ACP_NEXT: "1",
OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)),
},
}),
)
const initialized = expectOk(yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 }))
expect(initialized.agentCapabilities?.sessionCapabilities?.resume).toEqual({})
const session = expectOk(yield* acp.request<NewSessionResponse>("session/new", { cwd: home, mcpServers: [] }))
const resumed = expectOk(
yield* acp.request<ResumeSessionResponse>("session/resume", {
cwd: home,
sessionId: session.sessionId,
mcpServers: [],
}),
)
expect(selectConfigOption(resumed.configOptions, "model")?.category).toBe("model")
}),
60_000,
)
cliIt.live(
"exits cleanly when flagged stdin is closed",
({ opencode }) =>
Effect.gen(function* () {
const exitedPromise = yield* Effect.scoped(
Effect.gen(function* () {
const acp = yield* opencode.acp({ env: { OPENCODE_ACP_NEXT: "1" } })
return acp.exited
}),
)
const code = yield* Effect.promise(() => exitedPromise)
expect(typeof code === "number" || code === null).toBe(true)
}),
60_000,
)
cliIt.live(
"default unflagged path still uses production ACP",
({ opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(yield* opencode.acp())
const initialized = expectOk(yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 }))
expect(initialized.agentCapabilities?.sessionCapabilities?.close).toEqual({})
expect(initialized.agentCapabilities?.sessionCapabilities?.resume).toEqual({})
}),
60_000,
)
})
function errorCode(error: unknown) {
if (!error || typeof error !== "object") return undefined
if (!("code" in error)) return undefined
return typeof error.code === "number" ? error.code : undefined
}
function verifierConfig(llmUrl: string) {
const config = testProviderConfig(llmUrl)
return {
...config,
model: "test/test-model",
provider: {
test: {
...config.provider.test,
models: {
"test-model": {
...config.provider.test.models["test-model"],
variants: {
low: {},
high: {},
},
},
"second-model": {
...config.provider.test.models["test-model"],
id: "second-model",
name: "Second Test Model",
variants: {
medium: {},
max: {},
},
},
},
},
},
}
}
@@ -1,103 +0,0 @@
import { describe, expect } from "bun:test"
import type { SetSessionConfigOptionResponse } from "@agentclientprotocol/sdk"
import { Effect } from "effect"
import { cliIt } from "../../lib/cli-process"
import { expectOk, flattenSelectOptions, selectConfigOption } from "../acp/acp-test-client"
import {
createAcpNextClient,
expectAlternateValue,
expectSelectOption,
initialize,
newSession,
verifierConfig,
} from "./helpers"
describe("opencode acp-next config option subprocess", () => {
cliIt.live(
'model option is listed with category "model"',
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)) },
)
yield* initialize(acp)
const model = expectSelectOption((yield* newSession(acp, home)).configOptions, "model")
expect(model.category).toBe("model")
expect(model.currentValue).toBe("test/test-model")
expect(flattenSelectOptions(model).length).toBeGreaterThanOrEqual(2)
}),
60_000,
)
cliIt.live(
"model switch updates currentValue",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)) },
)
yield* initialize(acp)
const session = yield* newSession(acp, home)
const model = expectSelectOption(session.configOptions, "model")
const nextModel = flattenSelectOptions(model).find((option) => option.value === "test/second-model")?.value
expect(nextModel).toBe("test/second-model")
const updated = expectOk(
yield* acp.request<SetSessionConfigOptionResponse>("session/set_config_option", {
sessionId: session.sessionId,
configId: "model",
value: nextModel,
}),
)
expect(selectConfigOption(updated.configOptions, "model")?.currentValue).toBe(nextModel)
}),
60_000,
)
cliIt.live(
'effort option is listed with category "thought_level" when selected model supports variants',
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)) },
)
yield* initialize(acp)
const effort = expectSelectOption((yield* newSession(acp, home)).configOptions, "effort")
expect(effort.category).toBe("thought_level")
expect(effort.currentValue).toBe("low")
expect(flattenSelectOptions(effort).map((option) => option.value)).toEqual(["low", "high"])
}),
60_000,
)
cliIt.live(
"effort switch updates currentValue",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)) },
)
yield* initialize(acp)
const session = yield* newSession(acp, home)
const nextEffort = expectAlternateValue(expectSelectOption(session.configOptions, "effort"))
const updated = expectOk(
yield* acp.request<SetSessionConfigOptionResponse>("session/set_config_option", {
sessionId: session.sessionId,
configId: "effort",
value: nextEffort,
}),
)
expect(selectConfigOption(updated.configOptions, "effort")?.currentValue).toBe(nextEffort)
}),
60_000,
)
})
@@ -1,103 +0,0 @@
import { expect } from "bun:test"
import type { InitializeResponse, NewSessionResponse, SessionConfigOption } from "@agentclientprotocol/sdk"
import { Effect } from "effect"
import type { CliFixture } from "../../lib/cli-process"
import { testProviderConfig } from "../../lib/test-provider"
import {
createAcpClient,
expectOk,
flattenSelectOptions,
selectConfigOption,
type AcpClient,
} from "../acp/acp-test-client"
export function createAcpNextClient(input: Pick<CliFixture, "opencode">, env?: Record<string, string>) {
return Effect.gen(function* () {
return createAcpClient(
yield* input.opencode.acp({
env: {
OPENCODE_ACP_NEXT: "1",
...env,
},
}),
)
})
}
export function initialize(acp: AcpClient) {
return Effect.gen(function* () {
return expectOk(
yield* acp.request<InitializeResponse>("initialize", {
protocolVersion: 1,
clientCapabilities: { _meta: { "terminal-auth": true } },
clientInfo: { name: "opencode-local-acp-next", version: "0.1.0" },
}),
)
})
}
export function newSession(acp: AcpClient, cwd: string) {
return Effect.gen(function* () {
return expectOk(yield* acp.request<NewSessionResponse>("session/new", { cwd, mcpServers: [] }))
})
}
export function verifierConfig(llmUrl: string, skills?: string) {
const config = testProviderConfig(llmUrl)
return {
...config,
model: "test/test-model",
...(skills ? { skills: { paths: [skills] } } : {}),
provider: {
test: {
...config.provider.test,
models: {
"test-model": {
...config.provider.test.models["test-model"],
variants: {
low: {},
high: {},
},
},
"second-model": {
...config.provider.test.models["test-model"],
id: "second-model",
name: "Second Test Model",
variants: {
medium: {},
max: {},
},
},
},
},
},
}
}
export function expectErrorCode(error: unknown, code: number) {
if (!error || typeof error !== "object" || !("code" in error)) {
expect(error).toEqual({ code })
return
}
expect(error.code).toBe(code)
}
export function expectSelectOption(options: SessionConfigOption[] | null | undefined, id: string) {
const option = selectConfigOption(options, id)
expect(option).toBeDefined()
return option!
}
export function expectAlternateValue(option: ReturnType<typeof expectSelectOption>) {
const value = flattenSelectOptions(option).find((item) => item.value !== option.currentValue)?.value
expect(value).toBeDefined()
return value!
}
export const verifierSkill = `---
name: verifier-skill
description: Verifier compatibility skill.
---
# Verifier Skill
`
@@ -1,61 +0,0 @@
import { describe, expect } from "bun:test"
import type { AuthenticateResponse, InitializeResponse } from "@agentclientprotocol/sdk"
import { Effect } from "effect"
import { cliIt } from "../../lib/cli-process"
import { createAcpNextClient, expectErrorCode, initialize } from "./helpers"
describe("opencode acp-next initialize/auth subprocess", () => {
cliIt.live(
"initialize responds with capabilities",
({ opencode }) =>
Effect.gen(function* () {
const initialized = yield* initialize(yield* createAcpNextClient({ opencode }))
expect(initialized.protocolVersion).toBe(1)
expect(initialized.agentCapabilities?.promptCapabilities?.embeddedContext).toBe(true)
expect(initialized.agentCapabilities?.promptCapabilities?.image).toBe(true)
expect(initialized.agentCapabilities?.mcpCapabilities?.http).toBe(true)
expect(initialized.agentCapabilities?.mcpCapabilities?.sse).toBe(true)
expect(initialized.agentCapabilities?.loadSession).toBe(true)
expect(initialized.agentCapabilities?.sessionCapabilities?.close).toEqual({})
expect(initialized.agentCapabilities?.sessionCapabilities?.fork).toEqual({})
expect(initialized.agentCapabilities?.sessionCapabilities?.list).toEqual({})
expect(initialized.agentCapabilities?.sessionCapabilities?.resume).toEqual({})
expect(initialized.agentInfo?.name).toBe("OpenCode")
}),
60_000,
)
cliIt.live(
"auth negotiation is explicit and safe",
({ opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient({ opencode })
const initialized = yield* initialize(acp)
expect(initialized.authMethods?.[0]?.id).toBe("opencode-login")
expect(initialized.authMethods?.[0]?._meta?.["terminal-auth"]).toBeDefined()
expect(yield* acp.request<AuthenticateResponse>("authenticate", { methodId: "opencode-login" })).toMatchObject({
result: {},
})
const rejected = yield* acp.request<AuthenticateResponse>("authenticate", { methodId: "missing-auth-method" })
expectErrorCode(rejected.error, -32602)
expect(JSON.stringify(rejected.error)).not.toContain(process.env.OPENCODE_AUTH_CONTENT ?? "not-present")
}),
60_000,
)
cliIt.live(
"initialize without terminal-auth metadata keeps auth command implicit",
({ opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient({ opencode })
const initialized = yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 })
expect(initialized.result?.authMethods?.[0]?.id).toBe("opencode-login")
expect(initialized.result?.authMethods?.[0]?._meta?.["terminal-auth"]).toBeUndefined()
}),
60_000,
)
})
@@ -1,118 +0,0 @@
import { describe, expect } from "bun:test"
import type {
CloseSessionResponse,
ListSessionsResponse,
LoadSessionResponse,
ResumeSessionResponse,
} from "@agentclientprotocol/sdk"
import { Duration, Effect } from "effect"
import { cliIt } from "../../lib/cli-process"
import { expectOk, selectConfigOption } from "../acp/acp-test-client"
import { createAcpNextClient, initialize, newSession, verifierConfig } from "./helpers"
describe("opencode acp-next lifecycle subprocess", () => {
cliIt.live(
"stdin EOF exits cleanly",
({ opencode }) =>
Effect.gen(function* () {
const acp = yield* opencode.acp({ env: { OPENCODE_ACP_NEXT: "1" } })
acp.close()
const code = yield* Effect.promise(() => acp.exited).pipe(Effect.timeout(Duration.seconds(5)))
expect(code).toBe(0)
}),
60_000,
)
cliIt.live(
"close capability and close request",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)) },
)
const initialized = yield* initialize(acp)
expect(initialized.agentCapabilities?.sessionCapabilities?.close).toEqual({})
const session = yield* newSession(acp, home)
expectOk(yield* acp.request<CloseSessionResponse>("session/close", { sessionId: session.sessionId }))
}),
60_000,
)
cliIt.live(
"loadSession capability and load request return session config options",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)) },
)
const initialized = yield* initialize(acp)
expect(initialized.agentCapabilities?.loadSession).toBe(true)
const session = yield* newSession(acp, home)
const loaded = expectOk(
yield* acp.request<LoadSessionResponse>("session/load", {
cwd: home,
sessionId: session.sessionId,
mcpServers: [],
}),
)
expect(selectConfigOption(loaded.configOptions, "model")?.category).toBe("model")
}),
60_000,
)
cliIt.live(
"list request includes a live ACP-created session",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)) },
)
yield* initialize(acp)
const session = yield* newSession(acp, home)
const listed = expectOk(yield* acp.request<ListSessionsResponse>("session/list", { cwd: home }))
expect(listed.sessions.some((item) => item.sessionId === session.sessionId)).toBe(true)
}),
60_000,
)
cliIt.live(
"resume capability advertisement",
({ opencode }) =>
Effect.gen(function* () {
const initialized = yield* initialize(yield* createAcpNextClient({ opencode }))
expect(initialized.agentCapabilities?.sessionCapabilities?.resume).toEqual({})
}),
60_000,
)
cliIt.live(
"resume request returns session config options",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url)) },
)
yield* initialize(acp)
const session = yield* newSession(acp, home)
const resumed = expectOk(
yield* acp.request<ResumeSessionResponse>("session/resume", {
cwd: home,
sessionId: session.sessionId,
mcpServers: [],
}),
)
expect(selectConfigOption(resumed.configOptions, "model")?.category).toBe("model")
}),
60_000,
)
})
@@ -1,97 +0,0 @@
import { describe, expect } from "bun:test"
import type { PromptResponse } from "@agentclientprotocol/sdk"
import { Effect } from "effect"
import { writeFile } from "node:fs/promises"
import path from "node:path"
import { pathToFileURL } from "node:url"
import { cliIt } from "../../lib/cli-process"
import { expectOk } from "../acp/acp-test-client"
import { createAcpNextClient, initialize, newSession, verifierConfig } from "./helpers"
const tinyPng = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII="
describe("opencode acp-next prompt content subprocess", () => {
cliIt.live(
"accepts embedded text resource image and file resource link prompt content",
({ home, llm, opencode }) =>
Effect.gen(function* () {
yield* Effect.promise(() => writeFile(path.join(home, "README.md"), "# ACP content smoke\n"))
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(promptContentConfig(llm.url)) },
)
yield* initialize(acp)
const session = yield* newSession(acp, home)
yield* llm.text("embedded resource accepted")
expectOk(
yield* acp.request<PromptResponse>("session/prompt", {
sessionId: session.sessionId,
prompt: [
{ type: "text", text: "Use this embedded resource." },
{
type: "resource",
resource: { uri: "file:///context.txt", mimeType: "text/plain", text: "embedded context" },
},
],
}),
)
yield* llm.text("image accepted")
expectOk(
yield* acp.request<PromptResponse>("session/prompt", {
sessionId: session.sessionId,
prompt: [
{ type: "text", text: "Use this image." },
{
type: "image",
mimeType: "image/png",
data: tinyPng,
},
],
}),
)
yield* llm.text("file link accepted")
const linked = expectOk(
yield* acp.request<PromptResponse>("session/prompt", {
sessionId: session.sessionId,
prompt: [
{ type: "text", text: "Use this linked file." },
{
type: "resource_link",
uri: pathToFileURL(path.join(home, "README.md")).href,
name: "README.md",
mimeType: "text/markdown",
},
],
}),
)
expect(linked.stopReason).toBe("end_turn")
}),
60_000,
)
})
function promptContentConfig(llmUrl: string) {
const config = verifierConfig(llmUrl)
return {
...config,
provider: {
test: {
...config.provider.test,
models: Object.fromEntries(
Object.entries(config.provider.test.models).map(([id, model]) => [
id,
{
...model,
attachment: true,
reasoning: true,
},
]),
),
},
},
}
}
@@ -1,38 +0,0 @@
import { describe, expect } from "bun:test"
import type { SessionNotification } from "@agentclientprotocol/sdk"
import { Effect } from "effect"
import { mkdir } from "node:fs/promises"
import path from "node:path"
import { cliIt } from "../../lib/cli-process"
import { createAcpNextClient, initialize, newSession, verifierConfig, verifierSkill } from "./helpers"
describe("opencode acp-next skills subprocess", () => {
cliIt.live(
"skill slash command appears through available_commands_update",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const skills = path.join(home, "skills")
yield* Effect.promise(() => mkdir(path.join(skills, "verifier-skill"), { recursive: true }))
yield* Effect.promise(() => Bun.write(path.join(skills, "verifier-skill", "SKILL.md"), verifierSkill))
const acp = yield* createAcpNextClient(
{ opencode },
{ OPENCODE_CONFIG_CONTENT: JSON.stringify(verifierConfig(llm.url, skills)) },
)
yield* initialize(acp)
const session = yield* newSession(acp, home)
const update = yield* acp.waitForNotification<SessionNotification>(
"session/update",
(params) =>
params.sessionId === session.sessionId &&
params.update.sessionUpdate === "available_commands_update" &&
params.update.availableCommands.some(
(command) => command.name === "verifier-skill" && command.description.length > 0,
),
)
expect(update.params?.sessionId).toBe(session.sessionId)
}),
60_000,
)
})
@@ -40,7 +40,7 @@ describe("opencode acp verifier compatibility baseline", () => {
)
cliIt.live(
"first session returns model options",
"first session timing diagnostic stays bounded and returns model options",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(
@@ -50,6 +50,7 @@ describe("opencode acp verifier compatibility baseline", () => {
},
}),
)
const started = Date.now()
yield* acp.request<InitializeResponse>("initialize", {
protocolVersion: 1,
clientCapabilities: {},
@@ -61,6 +62,9 @@ describe("opencode acp verifier compatibility baseline", () => {
mcpServers: [],
}),
)
const durationMs = Date.now() - started
expect(durationMs).toBeLessThan(15_000)
const model = selectConfigOption(session.configOptions, "model")
expect(model?.category).toBe("model")
expect(model?.currentValue).toBe("test/test-model")
@@ -70,7 +74,7 @@ describe("opencode acp verifier compatibility baseline", () => {
)
cliIt.live(
"newSession can be called repeatedly",
"warm newSession timing diagnostic stays bounded",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(
@@ -83,19 +87,22 @@ describe("opencode acp verifier compatibility baseline", () => {
yield* acp.request<InitializeResponse>("initialize", { protocolVersion: 1 })
yield* acp.request<NewSessionResponse>("session/new", { cwd: home, mcpServers: [] })
const started = Date.now()
const session = expectOk(
yield* acp.request<NewSessionResponse>("session/new", {
cwd: home,
mcpServers: [],
}),
)
const durationMs = Date.now() - started
expect(durationMs).toBeLessThan(15_000)
expect(session.sessionId).toBeTruthy()
}),
60_000,
)
cliIt.live(
"model switch updates currentValue",
"model switch timing diagnostic updates currentValue",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const acp = createAcpClient(
@@ -114,6 +121,7 @@ describe("opencode acp verifier compatibility baseline", () => {
: undefined
expect(nextModel).toBe("test/second-model")
const started = Date.now()
const updated = expectOk(
yield* acp.request<SetSessionConfigOptionResponse>("session/set_config_option", {
sessionId: session.sessionId,
@@ -121,7 +129,9 @@ describe("opencode acp verifier compatibility baseline", () => {
value: nextModel,
}),
)
const durationMs = Date.now() - started
expect(durationMs).toBeLessThan(15_000)
expect(selectConfigOption(updated.configOptions, "model")?.currentValue).toBe(nextModel)
}),
60_000,
@@ -179,7 +189,7 @@ describe("opencode acp verifier compatibility baseline", () => {
)
cliIt.live(
"skill slash command appears through available_commands_update",
"skill slash command timing diagnostic appears through available_commands_update",
({ home, llm, opencode }) =>
Effect.gen(function* () {
const skills = path.join(home, "skills")
@@ -204,6 +214,20 @@ describe("opencode acp verifier compatibility baseline", () => {
)
expect(update.params?.sessionId).toBe(session.sessionId)
const secondSession = expectOk(
yield* acp.request<NewSessionResponse>("session/new", { cwd: home, mcpServers: [] }),
)
const started = Date.now()
yield* acp.waitForNotification<SessionNotification>(
"session/update",
(params) =>
params.sessionId === secondSession.sessionId &&
params.update.sessionUpdate === "available_commands_update" &&
params.update.availableCommands.some((command) => command.name === "verifier-skill"),
)
const durationMs = Date.now() - started
expect(durationMs).toBeLessThan(15_000)
}),
60_000,
)
+150 -239
View File
@@ -1,9 +1,8 @@
import { describe, expect, test } from "bun:test"
import { EventEmitter } from "node:events"
import { createServer, type IncomingMessage, type Server as HttpServer } from "node:http"
import type { IncomingMessage } from "node:http"
import net, { type AddressInfo, type Socket } from "node:net"
import WebSocket, { WebSocketServer } from "ws"
import { ProviderError } from "../../src/provider/error"
import { OpenAIWebSocket } from "../../src/plugin/openai/ws"
import { OpenAIWebSocketPool, TITLE_HEADER } from "../../src/plugin/openai/ws-pool"
@@ -39,17 +38,6 @@ describe("plugin.openai.ws", () => {
).rejects.toThrow("WebSocket connect timed out")
})
test("surfaces websocket upgrade rejection messages", async () => {
await using server = await createRejectingWebSocketServer(() => {})
await expect(
OpenAIWebSocket.connectResponsesWebSocket({
url: server.wsUrl,
headers: {},
}),
).rejects.toThrow("Expected 101 status code")
})
test("enforces websocket send idle timeout", async () => {
const socket = new (class extends EventEmitter {
send(_data: string, _callback: (error?: Error) => void) {}
@@ -62,7 +50,7 @@ describe("plugin.openai.ws", () => {
onConnectionInvalid: (error) => invalid.push(error.message),
})
expect((await readTextError(response.text())).message).toContain("idle timeout sending websocket request")
await expect(response.text()).rejects.toThrow("idle timeout sending websocket request")
expect(invalid).toEqual(["idle timeout sending websocket request"])
})
@@ -97,7 +85,7 @@ describe("plugin.openai.ws", () => {
})
test("errors the SSE stream when the server closes before a terminal event", async () => {
const invalid: Error[] = []
const invalid: string[] = []
await using server = await createWebSocketServer((socket) => {
socket.once("message", () => {
socket.close(1009, "payload too large")
@@ -108,14 +96,13 @@ describe("plugin.openai.ws", () => {
const response = OpenAIWebSocket.streamResponsesWebSocket({
socket,
body: { stream: true, input: "hi" },
onConnectionInvalid: (error) => invalid.push(error),
onConnectionInvalid: (error) => invalid.push(error.message),
})
expect((await readTextError(response.text())).message).toContain(
await expect(response.text()).rejects.toThrow(
"WebSocket closed before response.completed (code 1009: message too big: payload too large)",
)
expect(invalid[0]).toBeInstanceOf(ProviderError.ResponseStreamError)
expect(invalid.map((error) => error.message)).toEqual([
expect(invalid).toEqual([
"WebSocket closed before response.completed (code 1009: message too big: payload too large)",
])
})
@@ -135,7 +122,7 @@ describe("plugin.openai.ws", () => {
onConnectionInvalid: (error) => invalid.push(error.message),
})
expect((await readTextError(response.text())).message).toContain("Unexpected binary WebSocket frame")
await expect(response.text()).rejects.toThrow("Unexpected binary WebSocket frame")
expect(invalid).toEqual(["Unexpected binary WebSocket frame"])
})
})
@@ -153,12 +140,13 @@ describe("plugin.openai.ws-pool", () => {
})
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
httpFetch: mockFetch(async () => new Response("http")),
})
const first = await fetch(server.url, streamRequest())
const first = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await first.text()).toContain("data: [DONE]")
const second = await fetch(server.url, streamRequest())
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await second.text()).toContain("data: [DONE]")
expect(connections).toBe(1)
expect(messages).toBe(2)
@@ -175,59 +163,41 @@ describe("plugin.openai.ws-pool", () => {
})
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
httpFetch: mockFetch(async () => new Response("http")),
maxConnectionAge: 0,
})
const first = await fetch(server.url, streamRequest())
const first = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await first.text()).toContain("data: [DONE]")
const second = await fetch(server.url, streamRequest())
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await second.text()).toContain("data: [DONE]")
expect(connections).toBe(2)
fetch.close()
})
test("falls back to HTTP after websocket setup retries are exhausted", async () => {
test("falls back to HTTP when websocket setup fails and keeps the fallback sticky", async () => {
const attempts: string[] = []
await using server = await createRejectingWebSocketServer(() => attempts.push("websocket"))
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
connectTimeout: 100,
streamRetries: 1,
})
const first = await fetch(server.url, streamRequest({ [TITLE_HEADER]: "false" }))
expect(await readTextError(first.text())).toBeInstanceOf(ProviderError.ResponseStreamError)
const second = await fetch(server.url, streamRequest({ [TITLE_HEADER]: "false" }))
const third = await fetch(server.url, streamRequest({ [TITLE_HEADER]: "false" }))
const first = await fetch("https://api.openai.com/v1/responses", streamRequest({ [TITLE_HEADER]: "false" }))
const second = await fetch("https://api.openai.com/v1/responses", streamRequest({ [TITLE_HEADER]: "false" }))
expect(await second.text()).toBe("http")
expect(await third.text()).toBe("http")
expect(attempts).toEqual(["websocket", "websocket"])
expect(server.httpRequests).toHaveLength(2)
expect(server.httpRequests[0]?.headers[TITLE_HEADER]).toBeUndefined()
expect(server.httpRequests[1]?.headers[TITLE_HEADER]).toBeUndefined()
fetch.close()
})
test("prunes HTTP fallback after its idle timeout", async () => {
let websocketAttempts = 0
await using server = await createRejectingWebSocketServer(() => websocketAttempts++)
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
connectTimeout: 100,
idleTimeout: 20,
streamRetries: 0,
})
const first = await fetch(server.url, streamRequest())
expect(await first.text()).toBe("http")
await new Promise((resolve) => setTimeout(resolve, 50))
const second = await fetch(server.url, streamRequest())
expect(await second.text()).toBe("http")
expect(websocketAttempts).toBe(2)
expect(server.httpRequests).toHaveLength(2)
expect(attempts).toEqual(["websocket"])
expect(httpRequests).toHaveLength(2)
expect(httpRequests[0]?.get(TITLE_HEADER)).toBeNull()
expect(httpRequests[1]?.get(TITLE_HEADER)).toBeNull()
fetch.close()
})
@@ -239,21 +209,26 @@ describe("plugin.openai.ws-pool", () => {
socket.send(JSON.stringify({ type: connections === 1 ? "response.failed" : "response.completed" }))
})
})
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest())
const first = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await first.text()).toContain('data: {"type":"response.failed"}')
const second = await fetch(server.url, streamRequest())
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await second.text()).toContain('data: {"type":"response.completed"}')
expect(connections).toBe(2)
expect(server.httpRequests).toHaveLength(0)
expect(httpRequests).toHaveLength(0)
fetch.close()
})
test("retries websocket connection limit errors on the next stream attempt", async () => {
test("reconnects and replays after websocket connection limit errors", async () => {
let connections = 0
let messages = 0
await using server = await createWebSocketServer((socket) => {
@@ -277,21 +252,24 @@ describe("plugin.openai.ws-pool", () => {
socket.send(JSON.stringify({ type: "response.completed", response: { id: "resp_retry" } }))
})
})
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest())
expect((await readTextError(first.text())).message).toContain("Responses websocket connection limit reached")
const second = await fetch(server.url, streamRequest())
const text = await second.text()
const response = await fetch("https://api.openai.com/v1/responses", streamRequest())
const text = await response.text()
expect(text).not.toContain("websocket_connection_limit_reached")
expect(text).toContain('data: {"type":"response.completed","response":{"id":"resp_retry"}}')
expect(text).toContain("data: [DONE]")
expect(connections).toBe(2)
expect(messages).toBe(2)
expect(server.httpRequests).toHaveLength(0)
expect(httpRequests).toHaveLength(0)
fetch.close()
})
@@ -313,159 +291,98 @@ describe("plugin.openai.ws-pool", () => {
)
})
})
let httpRequests = 0
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
streamRetries: 2,
connectionLimitRetries: 2,
httpFetch: mockFetch(async () => {
httpRequests += 1
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest())
expect((await readTextError(first.text())).message).toContain("Responses websocket connection limit reached")
const second = await fetch(server.url, streamRequest())
expect((await readTextError(second.text())).message).toContain("Responses websocket connection limit reached")
const third = await fetch(server.url, streamRequest())
const fourth = await fetch(server.url, streamRequest())
expect(await third.text()).toBe("http")
expect(await fourth.text()).toBe("http")
expect(connections).toBe(3)
expect(server.httpRequests).toHaveLength(2)
fetch.close()
})
test("shares the websocket retry budget across stream and connection limit failures", async () => {
let connections = 0
await using server = await createWebSocketServer((socket) => {
connections += 1
socket.once("message", () => {
if (connections === 1) {
socket.send(JSON.stringify({ type: "response.output_text.delta", delta: "started" }))
socket.terminate()
return
}
socket.send(
JSON.stringify({
type: "error",
error: {
code: "websocket_connection_limit_reached",
message: "Responses websocket connection limit reached",
},
}),
)
})
})
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
streamRetries: 1,
})
const first = await fetch(server.url, streamRequest())
expect((await readTextError(first.text())).message).toContain("WebSocket closed before response.completed")
const second = await fetch(server.url, streamRequest())
const first = await fetch("https://api.openai.com/v1/responses", streamRequest())
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await first.text()).toBe("http")
expect(await second.text()).toBe("http")
expect(connections).toBe(2)
expect(server.httpRequests).toHaveLength(1)
expect(connections).toBe(3)
expect(httpRequests).toBe(2)
fetch.close()
})
test("retries websocket idle failures before first event then falls back to HTTP", async () => {
test("replays over HTTP when websocket idles before its first event", async () => {
let connections = 0
await using server = await createWebSocketServer((socket) => {
connections += 1
socket.once("message", () => {})
})
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
idleTimeout: 20,
streamRetries: 1,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest())
expect((await readTextError(first.text())).message).toContain("idle timeout waiting for websocket")
const second = await fetch(server.url, streamRequest())
const third = await fetch(server.url, streamRequest())
const first = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await first.text()).toBe("http")
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await second.text()).toBe("http")
expect(await third.text()).toBe("http")
expect(connections).toBe(2)
expect(server.httpRequests).toHaveLength(2)
expect(connections).toBe(1)
expect(httpRequests).toHaveLength(2)
fetch.close()
})
test("retries failed websocket streams before using HTTP fallback", async () => {
test("does not replay over HTTP after a websocket event was emitted", async () => {
await using server = await createWebSocketServer((socket) => {
socket.once("message", () => {
socket.send(JSON.stringify({ type: "response.output_text.delta", delta: "started" }))
})
})
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
idleTimeout: 20,
streamRetries: 1,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest())
expect((await readTextError(first.text())).message).toContain("idle timeout waiting for websocket")
const second = await fetch(server.url, streamRequest())
expect((await readTextError(second.text())).message).toContain("idle timeout waiting for websocket")
const third = await fetch(server.url, streamRequest())
const first = await fetch("https://api.openai.com/v1/responses", streamRequest())
await expect(first.text()).rejects.toThrow("idle timeout waiting for websocket")
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await third.text()).toBe("http")
expect(server.httpRequests).toHaveLength(1)
fetch.close()
})
test("resets websocket stream failures after a completed response", async () => {
let connections = 0
let requests = 0
await using server = await createWebSocketServer((socket) => {
connections += 1
socket.on("message", () => {
requests += 1
if (requests === 1 || requests === 3) {
socket.send(JSON.stringify({ type: "response.output_text.delta", delta: "started" }))
socket.terminate()
return
}
socket.send(JSON.stringify({ type: "response.completed", response: { id: `resp_${requests}` } }))
})
})
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
streamRetries: 1,
})
const first = await fetch(server.url, streamRequest())
expect((await readTextError(first.text())).message).toContain("WebSocket closed before response.completed")
const second = await fetch(server.url, streamRequest())
expect(await second.text()).toContain("data: [DONE]")
const third = await fetch(server.url, streamRequest())
expect((await readTextError(third.text())).message).toContain("WebSocket closed before response.completed")
const fourth = await fetch(server.url, streamRequest())
expect(await fourth.text()).toContain("data: [DONE]")
expect(connections).toBe(3)
expect(requests).toBe(4)
expect(server.httpRequests).toHaveLength(0)
expect(await second.text()).toBe("http")
expect(httpRequests).toHaveLength(1)
fetch.close()
})
test("falls back to HTTP for missing session and title requests", async () => {
await using server = await createWebSocketServer(() => {})
const fetch = OpenAIWebSocketPool.createWebSocketFetch()
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const missingSession = await fetch(server.url, {
const missingSession = await fetch("https://api.openai.com/v1/responses", {
method: "POST",
headers: { [TITLE_HEADER]: "false" },
body: JSON.stringify({ stream: true }),
})
const title = await fetch(server.url, streamRequest({ [TITLE_HEADER]: "true" }))
const title = await fetch("https://api.openai.com/v1/responses", streamRequest({ [TITLE_HEADER]: "true" }))
expect(await missingSession.text()).toBe("http")
expect(await title.text()).toBe("http")
expect(server.httpRequests).toHaveLength(2)
expect(server.httpRequests[0]?.headers[TITLE_HEADER]).toBeUndefined()
expect(server.httpRequests[1]?.headers[TITLE_HEADER]).toBeUndefined()
expect(httpRequests).toHaveLength(2)
expect(httpRequests[0]?.get(TITLE_HEADER)).toBeNull()
expect(httpRequests[1]?.get(TITLE_HEADER)).toBeNull()
fetch.close()
})
@@ -478,44 +395,52 @@ describe("plugin.openai.ws-pool", () => {
})
})
const abort = new AbortController()
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest({}, abort.signal))
const first = await fetch("https://api.openai.com/v1/responses", streamRequest({}, abort.signal))
const firstText = first.text()
await waitFor(() => connections === 1, "websocket did not connect")
const second = await fetch(server.url, streamRequest())
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await second.text()).toBe("http")
expect(server.httpRequests).toHaveLength(1)
expect(httpRequests).toHaveLength(1)
expect(connections).toBe(1)
abort.abort(new Error("stop"))
expect((await readTextError(firstText)).message).toContain("stop")
await expect(firstText).rejects.toThrow("stop")
fetch.close()
})
test("reserves a websocket lane while its socket is connecting", async () => {
await using server = await createHangingTcpServer()
await using fallback = await createHttpServer()
let httpRequests = 0
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
connectTimeout: 20,
streamRetries: 0,
httpFetch: mockFetch(async () => {
httpRequests += 1
return new Response("http")
}),
})
const first = fetch(fallback.url, streamRequest())
const first = fetch("https://api.openai.com/v1/responses", streamRequest())
await waitFor(() => server.connections() === 1, "first websocket did not begin connecting")
const second = fetch(fallback.url, streamRequest())
const second = fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await (await second).text()).toBe("http")
expect(await (await first).text()).toBe("http")
expect(server.connections()).toBe(1)
expect(fallback.httpRequests).toHaveLength(2)
expect(httpRequests).toBe(2)
fetch.close()
})
test("retries unexpected closes before first event then falls back to HTTP", async () => {
test("replays over HTTP after an unexpected close before the first event", async () => {
let connections = 0
await using server = await createWebSocketServer((socket) => {
connections += 1
@@ -523,20 +448,22 @@ describe("plugin.openai.ws-pool", () => {
socket.close(1001, "server shutdown")
})
})
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
streamRetries: 1,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest())
expect((await readTextError(first.text())).message).toContain("WebSocket closed before response.completed")
const second = await fetch(server.url, streamRequest())
const third = await fetch(server.url, streamRequest())
const first = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await first.text()).toBe("http")
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await second.text()).toBe("http")
expect(await third.text()).toBe("http")
expect(connections).toBe(2)
expect(server.httpRequests).toHaveLength(2)
expect(connections).toBe(1)
expect(httpRequests).toHaveLength(2)
fetch.close()
})
@@ -552,22 +479,27 @@ describe("plugin.openai.ws-pool", () => {
socket.send(JSON.stringify({ type: "response.completed", response: { id: "resp_456" } }))
})
})
const httpRequests: Headers[] = []
const abort = new AbortController()
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest({}, abort.signal))
const first = await fetch("https://api.openai.com/v1/responses", streamRequest({}, abort.signal))
const firstText = first.text()
await waitFor(() => connections === 1, "first websocket did not connect")
abort.abort(new Error("stop"))
expect((await readTextError(firstText)).message).toContain("stop")
await expect(firstText).rejects.toThrow("stop")
const second = await fetch(server.url, streamRequest())
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await second.text()).toContain("data: [DONE]")
expect(connections).toBe(2)
expect(server.httpRequests).toHaveLength(0)
expect(httpRequests).toHaveLength(0)
fetch.close()
})
@@ -583,19 +515,24 @@ describe("plugin.openai.ws-pool", () => {
socket.send(JSON.stringify({ type: "response.completed", response: { id: "resp_after_cancel" } }))
})
})
const httpRequests: Headers[] = []
const fetch = OpenAIWebSocketPool.createWebSocketFetch({
url: server.url,
httpFetch: mockFetch(async (_input, init) => {
httpRequests.push(new Headers(init?.headers))
return new Response("http")
}),
})
const first = await fetch(server.url, streamRequest())
const first = await fetch("https://api.openai.com/v1/responses", streamRequest())
await waitFor(() => connections === 1, "first websocket did not connect")
await first.body!.cancel("stop")
const second = await fetch(server.url, streamRequest())
const second = await fetch("https://api.openai.com/v1/responses", streamRequest())
expect(await second.text()).toContain("data: [DONE]")
expect(connections).toBe(2)
expect(server.httpRequests).toHaveLength(0)
expect(httpRequests).toHaveLength(0)
fetch.close()
})
})
@@ -613,24 +550,20 @@ function streamRequest(headers?: Record<string, string>, signal?: AbortSignal):
}
}
async function readTextError(promise: Promise<string>) {
// Bun 1.3.14 hangs on expect(response.text()).rejects for streams errored from ws callbacks.
return promise.then(
() => {
throw new Error("Expected response text to reject")
},
(error) => {
expect(error).toBeInstanceOf(Error)
return error as Error
},
)
function mockFetch(
fn: (
input: Parameters<typeof globalThis.fetch>[0],
init: Parameters<typeof globalThis.fetch>[1],
) => ReturnType<typeof globalThis.fetch>,
): typeof globalThis.fetch {
return Object.assign(fn, { preconnect: globalThis.fetch.preconnect })
}
async function createWebSocketServer(onConnection: (socket: WebSocket, request: IncomingMessage) => void) {
const http = await createHttpServer()
const server = new WebSocketServer({ server: http.server })
const server = new WebSocketServer({ host: "127.0.0.1", port: 0 })
server.on("connection", onConnection)
return websocketServerHandle(server, http)
await new Promise<void>((resolve) => server.once("listening", resolve))
return websocketServerHandle(server)
}
async function createHangingTcpServer() {
@@ -655,53 +588,31 @@ async function createHangingTcpServer() {
}
async function createRejectingWebSocketServer(onAttempt: () => void) {
const http = await createHttpServer()
const server = new WebSocketServer({
server: http.server,
host: "127.0.0.1",
port: 0,
verifyClient(_info, callback) {
onAttempt()
callback(false, 401, "denied")
},
})
return websocketServerHandle(server, http)
await new Promise<void>((resolve) => server.once("listening", resolve))
return websocketServerHandle(server)
}
async function createHttpServer() {
const httpRequests: IncomingMessage[] = []
const server = createServer((request, response) => {
httpRequests.push(request)
response.writeHead(200, { "content-type": "text/plain" })
response.end("http")
})
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve))
function websocketServerHandle(server: WebSocketServer) {
const address = server.address() as AddressInfo
const url = `http://127.0.0.1:${address.port}/v1/responses`
return {
server,
httpRequests,
url: `http://127.0.0.1:${address.port}/v1/responses`,
async [Symbol.asyncDispose]() {
await closeHttpServer(server)
},
}
}
function websocketServerHandle(server: WebSocketServer, http: Awaited<ReturnType<typeof createHttpServer>>) {
return {
url: http.url,
wsUrl: http.url.replace(/^http/, "ws"),
httpRequests: http.httpRequests,
url,
wsUrl: url.replace(/^http/, "ws"),
async [Symbol.asyncDispose]() {
for (const socket of server.clients) socket.terminate()
server.close()
http.server.close()
},
}
}
function closeHttpServer(server: HttpServer) {
return new Promise<void>((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))
}
async function waitFor(predicate: () => boolean, message: string) {
const started = Date.now()
while (!predicate()) {
@@ -3223,12 +3223,6 @@ describe("ProviderTransform.variants", () => {
efforts: ["low", "medium", "high", "xhigh", "max"],
expectedHigh: { thinking: { type: "adaptive", display: "summarized" }, effort: "high" },
},
{
name: "opus 4.8",
apiIds: ["claude-opus-4-8", "claude-opus-4.8"],
efforts: ["low", "medium", "high", "xhigh", "max"],
expectedHigh: { thinking: { type: "adaptive", display: "summarized" }, effort: "high" },
},
]) {
for (const apiId of testCase.apiIds) {
test(`${testCase.name} ${apiId} returns supported reasoning efforts`, () => {
@@ -3347,28 +3341,6 @@ describe("ProviderTransform.variants", () => {
})
})
test("anthropic opus 4.8 returns adaptive reasoning options with xhigh", () => {
const result = ProviderTransform.variants(
createMockModel({
id: "bedrock/anthropic-claude-opus-4.8",
providerID: "bedrock",
api: {
id: "anthropic.claude-opus-4.8",
url: "https://bedrock.amazonaws.com",
npm: "@ai-sdk/amazon-bedrock",
},
}),
)
expect(Object.keys(result)).toEqual(["low", "medium", "high", "xhigh", "max"])
expect(result.high).toEqual({
reasoningConfig: {
type: "adaptive",
maxReasoningEffort: "high",
display: "summarized",
},
})
})
test("returns WIDELY_SUPPORTED_EFFORTS with reasoningConfig", () => {
const model = createMockModel({
id: "bedrock/llama-4",
@@ -406,20 +406,6 @@ describe("HttpApi UI fallback", () => {
}),
)
it.live("accepts basic auth passwords containing colons for the web UI", () =>
Effect.gen(function* () {
const response = yield* uiApp({
password: "sec:ret",
username: "opencode",
disableEmbeddedWebUi: true,
}).request("/", {
headers: { authorization: `Basic ${btoa("opencode:sec:ret")}` },
})
expect(response.status).toBe(200)
}),
)
// Regression for #25698 (Ope): the browser fetches the PWA manifest and
// its icons via flows that don't carry app-managed credentials (the
// `<link rel="manifest">` request is not under page-auth control), so the
@@ -1509,6 +1509,37 @@ describe("session.message-v2.fromError", () => {
})
})
test("explains ADC requirements when Vertex Anthropic credentials are missing", () => {
const vertexAnthropicModel = {
...model,
providerID: ProviderID.googleVertex,
api: { ...model.api, npm: "@ai-sdk/google-vertex/anthropic" },
}
const result = MessageV2.fromError(
new Error(
"Could not load the default credentials. Browse to https://cloud.google.com/docs/authentication/getting-started for more information.",
),
{ model: vertexAnthropicModel },
)
expect(result).toStrictEqual({
name: "ProviderAuthError",
data: {
providerID: "google-vertex",
message:
"Anthropic models on Google Vertex require Google Cloud credentials. Use `gcloud auth application-default login` or set `GOOGLE_APPLICATION_CREDENTIALS`.",
},
})
})
test("does not present Anthropic ADC guidance for Gemini Vertex credentials errors", () => {
const result = MessageV2.fromError(new Error("Could not load the default credentials."), {
model: { ...model, providerID: ProviderID.googleVertex, api: { ...model.api, npm: "@ai-sdk/google-vertex" } },
})
expect(result.name).toBe("UnknownError")
})
test("serializes tagged errors with their message", () => {
const result = MessageV2.fromError(new Question.RejectedError(), { providerID })
@@ -172,17 +172,6 @@ describe("session.retry.retryable", () => {
})
})
test("retries websocket stream transport errors", () => {
const request = MessageV2.fromError(
new ProviderError.ResponseStreamError("WebSocket closed before response.completed (code 1006: Connection ended)"),
{ providerID },
)
expect(MessageV2.APIError.isInstance(request)).toBe(true)
expect(SessionRetry.retryable(request, retryProvider)).toEqual({
message: "WebSocket closed before response.completed (code 1006: Connection ended)",
})
})
test("does not retry context overflow errors", () => {
const error = new MessageV2.ContextOverflowError({
message: "Input exceeds context window of this model",
+1 -2
View File
@@ -17,6 +17,5 @@
"paths": {
"~/*": ["./src/*"]
}
},
"include": ["*.ts", "src", "../core/src/resource.d.ts"]
}
}
@@ -1,5 +0,0 @@
ALTER TABLE `geo_stat` ADD `provider` varchar(128) DEFAULT 'all' NOT NULL;--> statement-breakpoint
ALTER TABLE `geo_stat` ADD `model` varchar(256) DEFAULT 'all' NOT NULL;--> statement-breakpoint
ALTER TABLE `geo_stat` DROP INDEX `uniq_country_period`;--> statement-breakpoint
CREATE UNIQUE INDEX `uniq_country_period` ON `geo_stat` (`grain`,`period_start`,`dataset`,`tier`,`client`,`source`,`provider`,`model`,`country`);--> statement-breakpoint
CREATE INDEX `idx_country_model` ON `geo_stat` (`model`,`country`,`grain`,`period_start`);
File diff suppressed because it is too large Load Diff
@@ -1,3 +0,0 @@
ALTER TABLE `geo_stat` ADD `period_key` varchar(32) NOT NULL;--> statement-breakpoint
ALTER TABLE `model_stat` ADD `period_key` varchar(32) NOT NULL;--> statement-breakpoint
ALTER TABLE `provider_stat` ADD `period_key` varchar(32) NOT NULL;
File diff suppressed because it is too large Load Diff
@@ -1,6 +0,0 @@
ALTER TABLE `geo_stat` DROP COLUMN `period_start`;--> statement-breakpoint
ALTER TABLE `geo_stat` DROP COLUMN `period_end`;--> statement-breakpoint
ALTER TABLE `model_stat` DROP COLUMN `period_start`;--> statement-breakpoint
ALTER TABLE `model_stat` DROP COLUMN `period_end`;--> statement-breakpoint
ALTER TABLE `provider_stat` DROP COLUMN `period_start`;--> statement-breakpoint
ALTER TABLE `provider_stat` DROP COLUMN `period_end`;
File diff suppressed because it is too large Load Diff
+15 -19
View File
@@ -16,7 +16,7 @@ export const modelStat = mysqlTable(
(table) => [
uniqueIndex("uniq_model_period").on(
table.grain,
table.period_key,
table.period_start,
table.dataset,
table.tier,
table.client,
@@ -24,8 +24,8 @@ export const modelStat = mysqlTable(
table.provider,
table.model,
),
index("idx_leaderboard_tokens").on(table.grain, table.period_key, table.dataset, table.tier, table.total_tokens),
index("idx_model").on(table.model, table.grain, table.period_key),
index("idx_leaderboard_tokens").on(table.grain, table.period_start, table.dataset, table.tier, table.total_tokens),
index("idx_model").on(table.model, table.grain, table.period_start),
],
)
@@ -45,7 +45,7 @@ export const providerStat = mysqlTable(
(table) => [
uniqueIndex("uniq_provider_period").on(
table.grain,
table.period_key,
table.period_start,
table.dataset,
table.tier,
table.client,
@@ -54,20 +54,20 @@ export const providerStat = mysqlTable(
),
index("idx_provider_leaderboard_tokens").on(
table.grain,
table.period_key,
table.period_start,
table.dataset,
table.tier,
table.total_tokens,
),
index("idx_provider_market_share").on(
table.grain,
table.period_key,
table.period_start,
table.dataset,
table.tier,
table.market_share_tokens,
),
index("idx_provider_rank").on(table.grain, table.period_key, table.dataset, table.tier, table.rank_by_tokens),
index("idx_provider").on(table.provider, table.grain, table.period_key),
index("idx_provider_rank").on(table.grain, table.period_start, table.dataset, table.tier, table.rank_by_tokens),
index("idx_provider").on(table.provider, table.grain, table.period_start),
],
)
@@ -75,8 +75,6 @@ export const geoStat = mysqlTable(
"geo_stat",
{
...periodColumns(),
provider: varchar({ length: 128 }).notNull().default("all"),
model: varchar({ length: 256 }).notNull().default("all"),
country: char({ length: 2 }).notNull(),
continent: varchar({ length: 8 }).notNull().default(""),
...metricColumns(),
@@ -90,20 +88,17 @@ export const geoStat = mysqlTable(
(table) => [
uniqueIndex("uniq_country_period").on(
table.grain,
table.period_key,
table.period_start,
table.dataset,
table.tier,
table.client,
table.source,
table.provider,
table.model,
table.country,
),
index("idx_country_map_tokens").on(table.grain, table.period_key, table.dataset, table.tier, table.total_tokens),
index("idx_country_rank").on(table.grain, table.period_key, table.dataset, table.tier, table.rank_by_tokens),
index("idx_country").on(table.country, table.grain, table.period_key),
index("idx_continent").on(table.continent, table.grain, table.period_key),
index("idx_country_model").on(table.model, table.country, table.grain, table.period_key),
index("idx_country_map_tokens").on(table.grain, table.period_start, table.dataset, table.tier, table.total_tokens),
index("idx_country_rank").on(table.grain, table.period_start, table.dataset, table.tier, table.rank_by_tokens),
index("idx_country").on(table.country, table.grain, table.period_start),
index("idx_continent").on(table.continent, table.grain, table.period_start),
],
)
@@ -111,7 +106,8 @@ function periodColumns() {
return {
id: bigint({ mode: "number" }).autoincrement().primaryKey(),
grain: varchar({ length: 16 }).notNull(),
period_key: varchar({ length: 32 }).notNull(),
period_start: datetime({ mode: "date" }).notNull(),
period_end: datetime({ mode: "date" }).notNull(),
dataset: varchar({ length: 64 }).notNull().default("all"),
tier: varchar({ length: 64 }).notNull().default("all"),
client: varchar({ length: 64 }).notNull().default("all"),
+22 -54
View File
@@ -8,7 +8,6 @@ import {
collapseRows,
inserted,
rankRowsWithMarketShare,
statPeriodKey,
synthesizeAllTierRows,
toStatBaseRow,
UPSERT_CHUNK_SIZE,
@@ -16,18 +15,11 @@ import {
} from "./stat"
export type GeoStatRow = typeof geoStat.$inferInsert
export type GeoStatAggregate = StatBaseAggregate & {
provider: string
model: string
country: string
continent: string
}
export type GeoStatAggregate = StatBaseAggregate & { country: string; continent: string }
export type GeoStatMetric = {
periodKey: string
updatedAt: Date
periodStart: Date
periodEnd: Date
tier: string
provider: string
model: string
country: string
continent: string
totalTokens: number
@@ -38,13 +30,11 @@ export declare namespace GeoStatRepo {
readonly listDaily: () => Effect.Effect<GeoStatMetric[], DatabaseError>
readonly listByPeriod: (opts: {
readonly grain: string
readonly periodKey: string
readonly periodStart: Date
readonly dataset?: string
readonly tier?: string
readonly client?: string
readonly source?: string
readonly provider?: string
readonly model?: string
}) => Effect.Effect<GeoStatRow[], DatabaseError>
readonly upsert: (rows: GeoStatRow[]) => Effect.Effect<void, DatabaseError>
}
@@ -61,39 +51,27 @@ export class GeoStatRepo extends Context.Service<GeoStatRepo, GeoStatRepo.Servic
try: () =>
db
.select({
periodKey: geoStat.period_key,
updatedAt: geoStat.updated_at,
periodStart: geoStat.period_start,
periodEnd: geoStat.period_end,
tier: geoStat.tier,
provider: geoStat.provider,
model: geoStat.model,
country: geoStat.country,
continent: geoStat.continent,
totalTokens: geoStat.total_tokens,
})
.from(geoStat)
.where(
and(
eq(geoStat.grain, "day"),
eq(geoStat.client, "all"),
eq(geoStat.source, "all"),
eq(geoStat.provider, "all"),
eq(geoStat.model, "all"),
),
)
.orderBy(asc(geoStat.period_key)),
.where(and(eq(geoStat.grain, "day"), eq(geoStat.client, "all"), eq(geoStat.source, "all")))
.orderBy(asc(geoStat.period_start)),
catch: (cause) => DatabaseError.make({ cause }),
})
})
const listByPeriod = Effect.fn("GeoStatRepo.listByPeriod")(function* (opts: {
readonly grain: string
readonly periodKey: string
readonly periodStart: Date
readonly dataset?: string
readonly tier?: string
readonly client?: string
readonly source?: string
readonly provider?: string
readonly model?: string
}) {
return yield* Effect.tryPromise({
try: () =>
@@ -103,13 +81,11 @@ export class GeoStatRepo extends Context.Service<GeoStatRepo, GeoStatRepo.Servic
.where(
and(
eq(geoStat.grain, opts.grain),
eq(geoStat.period_key, opts.periodKey),
eq(geoStat.period_start, opts.periodStart),
eq(geoStat.dataset, opts.dataset ?? "zen"),
eq(geoStat.tier, opts.tier ?? "all"),
eq(geoStat.client, opts.client ?? "all"),
eq(geoStat.source, opts.source ?? "all"),
eq(geoStat.provider, opts.provider ?? "all"),
eq(geoStat.model, opts.model ?? "all"),
),
),
catch: (cause) => DatabaseError.make({ cause }),
@@ -127,6 +103,7 @@ export class GeoStatRepo extends Context.Service<GeoStatRepo, GeoStatRepo.Servic
.values(chunk)
.onDuplicateKeyUpdate({
set: {
period_end: inserted("period_end"),
continent: inserted("continent"),
sessions: inserted("sessions"),
requests: inserted("requests"),
@@ -169,35 +146,26 @@ export class GeoStatRepo extends Context.Service<GeoStatRepo, GeoStatRepo.Servic
}
export function rowsFromAggregates(aggregates: GeoStatAggregate[]) {
return rankRowsWithMarketShare(
[
...synthesizeAllTierRows(
collapseRows(aggregates.filter((item) => item.grain === "week").map(toRow), dimensionKey),
dimensionKey,
),
...synthesizeAllTierRows(
collapseRows(aggregates.filter((item) => item.grain === "day").map(toRow), dimensionKey),
dimensionKey,
),
],
marketShareKey,
)
return rankRowsWithMarketShare([
...synthesizeAllTierRows(
collapseRows(aggregates.filter((item) => item.grain === "week").map(toRow), dimensionKey),
dimensionKey,
),
...synthesizeAllTierRows(
collapseRows(aggregates.filter((item) => item.grain === "day").map(toRow), dimensionKey),
dimensionKey,
),
])
}
function toRow(data: GeoStatAggregate): GeoStatRow {
return {
...toStatBaseRow(data),
provider: data.provider,
model: data.model,
country: data.country,
continent: data.continent,
}
}
function dimensionKey(row: GeoStatRow) {
return [row.provider, row.model, row.country].join("\u0000")
}
function marketShareKey(row: GeoStatRow) {
return [statPeriodKey(row), row.provider, row.model].join("\u0000")
return row.country
}
+21 -37
View File
@@ -28,17 +28,17 @@ const TOKEN_SCALE = 1_000_000
const DOLLARS_PER_MICROCENT = 1 / 100_000_000
const months = ["JAN", "FEB", "MAR", "APR", "MAY", "JUN", "JUL", "AUG", "SEP", "OCT", "NOV", "DEC"] as const
type StatMetricRow = Omit<ModelStatMetric, "updatedAt"> & {
type StatMetricRow = Omit<ModelStatMetric, "periodStart" | "periodEnd"> & {
periodStart: number
updatedAt: number
periodEnd: number
}
type ProviderMetricRow = Omit<ProviderStatMetric, "updatedAt"> & {
type ProviderMetricRow = Omit<ProviderStatMetric, "periodStart" | "periodEnd"> & {
periodStart: number
updatedAt: number
periodEnd: number
}
type GeoMetricRow = Omit<GeoStatMetric, "updatedAt"> & {
type GeoMetricRow = Omit<GeoStatMetric, "periodStart" | "periodEnd"> & {
periodStart: number
updatedAt: number
periodEnd: number
}
type DateWindow = { start: number; end: number; previousStart: number; previousEnd: number }
@@ -85,10 +85,10 @@ function buildStatsHomeData(
const earliest = Math.min(...periods.map((row) => row.periodStart))
const latest = Math.max(...periods.map((row) => row.periodStart))
const latestUpdate = Math.max(...periods.map((row) => row.updatedAt))
const latestEnd = Math.max(...periods.map((row) => row.periodEnd))
return {
updatedAt: new Date(latestUpdate).toISOString(),
updatedAt: new Date(latestEnd).toISOString(),
usage: createUsageProductRecord((product) =>
createRangeRecord((range) => buildUsagePoints(normalized, product, range, getWindow(range, earliest, latest))),
),
@@ -361,14 +361,14 @@ function createRangeRecord<T>(value: (range: UsageRange) => T): Record<UsageRang
}
function normalizeStatRow(row: ModelStatMetric): StatMetricRow[] {
const periodStart = periodKeyTime(row.periodKey)
const updatedAt = dateTime(row.updatedAt)
if (!Number.isFinite(periodStart) || !Number.isFinite(updatedAt)) return []
const periodStart = dateTime(row.periodStart)
const periodEnd = dateTime(row.periodEnd)
if (!Number.isFinite(periodStart) || !Number.isFinite(periodEnd)) return []
return [
{
...row,
periodStart,
updatedAt,
periodEnd,
tier: normalizeTier(row.tier),
provider: row.provider || "unknown",
model: row.model || "unknown",
@@ -377,14 +377,14 @@ function normalizeStatRow(row: ModelStatMetric): StatMetricRow[] {
}
function normalizeProviderRow(row: ProviderStatMetric): ProviderMetricRow[] {
const periodStart = periodKeyTime(row.periodKey)
const updatedAt = dateTime(row.updatedAt)
if (!Number.isFinite(periodStart) || !Number.isFinite(updatedAt)) return []
const periodStart = dateTime(row.periodStart)
const periodEnd = dateTime(row.periodEnd)
if (!Number.isFinite(periodStart) || !Number.isFinite(periodEnd)) return []
return [
{
...row,
periodStart,
updatedAt,
periodEnd,
tier: normalizeTier(row.tier),
provider: row.provider || "unknown",
},
@@ -392,17 +392,15 @@ function normalizeProviderRow(row: ProviderStatMetric): ProviderMetricRow[] {
}
function normalizeGeoRow(row: GeoStatMetric): GeoMetricRow[] {
const periodStart = periodKeyTime(row.periodKey)
const updatedAt = dateTime(row.updatedAt)
if (!Number.isFinite(periodStart) || !Number.isFinite(updatedAt)) return []
const periodStart = dateTime(row.periodStart)
const periodEnd = dateTime(row.periodEnd)
if (!Number.isFinite(periodStart) || !Number.isFinite(periodEnd)) return []
return [
{
...row,
periodStart,
updatedAt,
periodEnd,
tier: normalizeTier(row.tier),
provider: row.provider || "all",
model: row.model || "all",
country: row.country || "ZZ",
continent: row.continent || "",
},
@@ -422,12 +420,6 @@ function dateTime(value: Date | string) {
return (value instanceof Date ? value : new Date(value)).getTime()
}
function periodKeyTime(value: string) {
const match = /^(\d{4})-(\d{2})-(\d{2})$/.exec(value)
if (!match) return Number.NaN
return Date.UTC(Number(match[1]), Number(match[2]) - 1, Number(match[3]))
}
function formatBucketLabel(value: number, range: UsageRange) {
const date = new Date(value)
if (range === "YTD") return months[date.getUTCMonth()]
@@ -441,19 +433,11 @@ function formatBucketLabel(value: number, range: UsageRange) {
function formatProvider(provider: string) {
const known: Record<string, string> = {
anthropic: "Anthropic",
deepseek: "DeepSeek",
google: "Google",
minimax: "MiniMax",
moonshot: "Moonshot",
moonshotai: "Moonshot",
nvidia: "NVIDIA",
opencode: "opencode",
nvidia: "Nvidia",
openai: "OpenAI",
qwen: "Qwen",
tencent: "Tencent",
xai: "xAI",
xiaomi: "Xiaomi",
zhipu: "Zhipu",
zhipuai: "Zhipu",
}
const normalized = provider.toLowerCase().replace(/[^a-z0-9]/g, "")
@@ -1,66 +0,0 @@
import { describe, expect, test } from "bun:test"
import { toModelAggregate } from "./inference"
import { modelAuthor, normalizeInferenceModel } from "./model-normalization"
describe("inference stat normalization", () => {
test("normalizes model suffixes used by router/provider variants", () => {
expect(normalizeInferenceModel("deepseek-v4-flash-free")).toBe("deepseek-v4-flash")
expect(normalizeInferenceModel("deepseek-v4-flash:global")).toBe("deepseek-v4-flash")
expect(normalizeInferenceModel("mimo-v2.5-free")).toBe("mimo-v2.5")
expect(normalizeInferenceModel("nemotron-3-super-free")).toBe("nemotron-3-super")
expect(normalizeInferenceModel("mimo-v2.5-free:global")).toBe("mimo-v2.5")
})
test("maps normalized model ids to public authors", () => {
expect(modelAuthor("big-pickle")).toBe("opencode")
expect(modelAuthor("claude-sonnet-4-5")).toBe("anthropic")
expect(modelAuthor("deepseek-v4-pro")).toBe("deepseek")
expect(modelAuthor("gemini-3.5-flash")).toBe("google")
expect(modelAuthor("glm-5.1")).toBe("zhipu")
expect(modelAuthor("gpt-5.5-pro")).toBe("openai")
expect(modelAuthor("grok-build-0.1")).toBe("xai")
expect(modelAuthor("hy3-preview")).toBe("tencent")
expect(modelAuthor("kimi-k2.6")).toBe("moonshot")
expect(modelAuthor("mimo-v2-omni")).toBe("xiaomi")
expect(modelAuthor("minimax-m2.7")).toBe("minimax")
expect(modelAuthor("nemotron-3-super-free")).toBe("nvidia")
expect(modelAuthor("qwen3.7-max")).toBe("qwen")
expect(modelAuthor("alpha-gpt-next")).toBeUndefined()
})
test("model aggregates ignore datalake provider and use normalized author/model", () => {
expect(toModelAggregate(aggregate("alpha-gpt-next", "openai"))).toEqual([])
expect(toModelAggregate(aggregate("deepseek-v4-flash-free", "not-public-provider"))).toMatchObject([
{
period_key: "2026-05-20",
provider: "deepseek",
model: "deepseek-v4-flash",
},
])
})
test("model aggregates use ISO week period keys", () => {
expect(
toModelAggregate({
...aggregate("gpt-5.5-pro", "openai"),
grain: "week",
period_key: "2026-W20",
}),
).toMatchObject([{ period_key: "2026-W20" }])
})
})
function aggregate(model: string, provider: string) {
return {
grain: "day",
period_key: "2026-05-20",
dataset: "zen",
tier: "Paid",
provider,
model,
sessions: "1",
requests: "1",
sample_count: "1",
}
}
+46 -86
View File
@@ -2,17 +2,10 @@ import { Resource } from "sst/resource"
import type { AthenaData } from "../athena"
import type { GeoStatAggregate } from "./geo"
import type { ModelStatAggregate } from "./model"
import {
EXCLUDED_MODELS,
MODEL_AUTHOR_OVERRIDES,
MODEL_AUTHOR_RULES,
modelAuthor,
normalizeInferenceModel,
} from "./model-normalization"
import type { ProviderStatAggregate } from "./provider"
import { normalizeCountry, normalizeTier, type StatBaseAggregate } from "./stat"
export type StatDimension = "model" | "provider" | "geo" | "geo_model"
export type StatDimension = "model" | "provider" | "geo"
export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: StatDimension) {
const periodStartValue = sqlString(periodStart.toISOString())
@@ -27,13 +20,8 @@ export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: S
groupBy: "provider, model",
}
if (dimension === "provider") return { select: "provider", groupBy: "provider" }
if (dimension === "geo_model")
return {
select: "provider, model, country, COALESCE(MAX(NULLIF(continent, '')), '') AS continent",
groupBy: "provider, model, country",
}
return {
select: "'all' AS provider, 'all' AS model, country, COALESCE(MAX(NULLIF(continent, '')), '') AS continent",
select: "country, COALESCE(MAX(NULLIF(continent, '')), '') AS continent",
groupBy: "country",
}
})()
@@ -60,56 +48,32 @@ export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: S
COUNT(*) AS sample_count`
return `
WITH normalized AS (
WITH filtered AS (
SELECT
from_iso8601_timestamp(event_timestamp) AS event_time,
model AS raw_model,
COALESCE(NULLIF(regexp_replace(model, '(-free|:global)+$', ''), ''), 'unknown') AS model,
CASE
WHEN source = 'lite' THEN 'Go'
WHEN model IN ('gpt-5-nano', 'grok-code', 'big-pickle') OR model LIKE '%-free' THEN 'Free'
ELSE 'Paid'
END AS tier,
COALESCE(NULLIF(
CASE
WHEN starts_with(provider, 'minimax-plan') THEN 'minimax-plan'
WHEN starts_with(provider, 'zai-plan') THEN 'zai-plan'
WHEN starts_with(provider, 'azure-databricks') THEN 'azure-databricks'
WHEN regexp_like(provider, '^azure[0-9]+') THEN 'azure-openai'
ELSE provider
END,
''
), 'unknown') AS provider,
COALESCE(NULLIF(provider_model, ''), '') AS provider_model,
COALESCE(NULLIF(model, ''), 'unknown') AS model,
UPPER(COALESCE(NULLIF(cf_country, ''), 'ZZ')) AS country,
COALESCE(NULLIF(cf_continent, ''), '') AS continent,
session,
status,
duration AS duration_ms,
time_to_first_byte AS ttfb_ms,
timestamp_first_byte,
timestamp_last_byte,
tokens_input,
tokens_output,
tokens_reasoning,
tokens_cache_read,
tokens_cache_write_5m,
cost_input_microcents,
cost_output_microcents,
cost_total_microcents,
cost_input,
cost_output,
cost_total,
source
FROM ${sourceTable}
WHERE event_type = 'completions'
AND model IS NOT NULL
AND model <> ''
AND (strpos(COALESCE(user_agent, ''), 'ai-sdk') > 0 OR strpos(COALESCE(user_agent, ''), 'opencode') > 0)
AND event_timestamp >= ${periodStartValue}
AND event_timestamp < ${periodEndValue}
), filtered AS (
SELECT
event_time,
CASE
WHEN source = 'lite' THEN 'Go'
WHEN model IN ('gpt-5-nano', 'grok-code', 'big-pickle') OR regexp_like(raw_model, '-free(:global)?$') THEN 'Free'
ELSE 'Paid'
END AS tier,
${modelAuthorSql("model")} AS provider,
provider_model,
model,
country,
continent,
session,
status,
duration_ms,
ttfb_ms,
CASE
WHEN timestamp_last_byte - timestamp_first_byte < 100 THEN null
ELSE CAST(tokens_output AS double) / (timestamp_last_byte - timestamp_first_byte) * 1000
@@ -122,47 +86,50 @@ WITH normalized AS (
COALESCE(cost_input_microcents, cost_input * 1000000) AS cost_input_microcents,
COALESCE(cost_output_microcents, cost_output * 1000000) AS cost_output_microcents,
COALESCE(cost_total_microcents, cost_total * 1000000) AS cost_total_microcents
FROM normalized
WHERE lower(model) NOT IN (${[...EXCLUDED_MODELS].map(sqlString).join(", ")})
), weekly AS (
SELECT
concat(CAST(year_of_week(event_time) AS varchar), '-W', lpad(CAST(week(event_time) AS varchar), 2, '0')) AS week_key,
*
FROM filtered
FROM ${sourceTable}
WHERE event_type = 'completions'
AND model IS NOT NULL
AND model <> ''
AND (strpos(COALESCE(user_agent, ''), 'ai-sdk') > 0 OR strpos(COALESCE(user_agent, ''), 'opencode') > 0)
AND event_timestamp >= ${periodStartValue}
AND event_timestamp < ${periodEndValue}
), daily AS (
SELECT substr(to_iso8601(date_trunc('day', event_time)), 1, 10) AS day_key, *
SELECT date_trunc('day', event_time) AS day, *
FROM filtered
)
SELECT
'week' AS grain,
week_key AS period_key,
${periodStartValue} AS period_start,
${periodEndValue} AS period_end,
${sqlString(Resource.StatsSyncConfig.dataset)} AS dataset,
tier,
${dimensionSql.select},
${aggregateColumns}
FROM weekly
GROUP BY week_key, tier, ${dimensionSql.groupBy}
FROM filtered
GROUP BY tier, ${dimensionSql.groupBy}
UNION ALL
SELECT
'day' AS grain,
day_key AS period_key,
to_iso8601(day) AS period_start,
to_iso8601(least(day + INTERVAL '1' DAY, from_iso8601_timestamp(${periodEndValue}))) AS period_end,
${sqlString(Resource.StatsSyncConfig.dataset)} AS dataset,
tier,
${dimensionSql.select},
${aggregateColumns}
FROM daily
GROUP BY day_key, tier, ${dimensionSql.groupBy}
ORDER BY grain, period_key, total_tokens DESC
GROUP BY day, tier, ${dimensionSql.groupBy}
ORDER BY grain, period_start, total_tokens DESC
`
}
export function toModelAggregate(data: AthenaData): ModelStatAggregate[] {
const model = normalizeInferenceModel(data.model)
const author = modelAuthor(model)
if (!author) return []
return toStatBaseAggregate(data).flatMap((base) => [
{ ...base, provider: author, model, provider_model: data.provider_model || "" },
{
...base,
provider: data.provider || "unknown",
model: data.model || "unknown",
provider_model: data.provider_model || "",
},
])
}
@@ -174,8 +141,6 @@ export function toGeoAggregate(data: AthenaData): GeoStatAggregate[] {
return toStatBaseAggregate(data).flatMap((base) => [
{
...base,
provider: data.provider || "all",
model: normalizeInferenceModel(data.model || "all"),
country: normalizeCountry(data.country),
continent: data.continent || "",
},
@@ -184,12 +149,15 @@ export function toGeoAggregate(data: AthenaData): GeoStatAggregate[] {
function toStatBaseAggregate(data: AthenaData): StatBaseAggregate[] {
const grain = data.grain === "day" || data.grain === "week" ? data.grain : undefined
if (!grain || !data.period_key) return []
const periodStart = new Date(data.period_start ?? "")
const periodEnd = new Date(data.period_end ?? "")
if (!grain || Number.isNaN(periodStart.getTime()) || Number.isNaN(periodEnd.getTime())) return []
return [
{
grain,
period_key: data.period_key,
period_start: periodStart,
period_end: periodEnd,
dataset: data.dataset || Resource.StatsSyncConfig.dataset,
tier: normalizeTier(data.tier || "unknown"),
sessions: integer(data, "sessions"),
@@ -242,11 +210,3 @@ function sqlIdentifier(value: string) {
function sqlString(value: string) {
return `'${value.replace(/'/g, "''")}'`
}
function modelAuthorSql(model: string) {
return `CASE
${MODEL_AUTHOR_OVERRIDES.map((item) => ` WHEN lower(${model}) = ${sqlString(item.model)} THEN ${sqlString(item.author)}`).join("\n")}
${MODEL_AUTHOR_RULES.map((item) => ` WHEN strpos(lower(${model}), ${sqlString(item.match)}) > 0 THEN ${sqlString(item.author)}`).join("\n")}
ELSE 'unknown'
END`
}
@@ -1,30 +0,0 @@
export const MODEL_AUTHOR_OVERRIDES = [{ model: "big-pickle", author: "opencode" }] as const
export const MODEL_AUTHOR_RULES = [
{ match: "claude", author: "anthropic" },
{ match: "gemini", author: "google" },
{ match: "deepseek", author: "deepseek" },
{ match: "glm", author: "zhipu" },
{ match: "gpt", author: "openai" },
{ match: "grok", author: "xai" },
{ match: "hy3", author: "tencent" },
{ match: "kimi", author: "moonshot" },
{ match: "mimo", author: "xiaomi" },
{ match: "minimax", author: "minimax" },
{ match: "nemotron", author: "nvidia" },
{ match: "qwen", author: "qwen" },
] as const
export const EXCLUDED_MODELS = new Set(["alpha-gpt-next"])
export function normalizeInferenceModel(value: string | undefined) {
return (value || "unknown").replace(/(-free|:global)+$/, "") || "unknown"
}
export function modelAuthor(value: string | undefined) {
const model = normalizeInferenceModel(value).toLowerCase()
if (EXCLUDED_MODELS.has(model)) return undefined
const override = MODEL_AUTHOR_OVERRIDES.find((item) => item.model === model)
if (override) return override.author
return MODEL_AUTHOR_RULES.find((item) => model.includes(item.match))?.author ?? "unknown"
}
+6 -5
View File
@@ -19,8 +19,8 @@ export type ModelStatRow = typeof modelStat.$inferInsert
export type ModelStatAggregate = StatBaseAggregate & { provider: string; model: string; provider_model: string }
export type ModelStatMetric = {
periodKey: string
updatedAt: Date
periodStart: Date
periodEnd: Date
tier: string
provider: string
model: string
@@ -55,8 +55,8 @@ export class ModelStatRepo extends Context.Service<ModelStatRepo, ModelStatRepo.
try: () =>
db
.select({
periodKey: modelStat.period_key,
updatedAt: modelStat.updated_at,
periodStart: modelStat.period_start,
periodEnd: modelStat.period_end,
tier: modelStat.tier,
provider: modelStat.provider,
model: modelStat.model,
@@ -72,7 +72,7 @@ export class ModelStatRepo extends Context.Service<ModelStatRepo, ModelStatRepo.
})
.from(modelStat)
.where(and(eq(modelStat.grain, "day"), eq(modelStat.client, "all"), eq(modelStat.source, "all")))
.orderBy(asc(modelStat.period_key)),
.orderBy(asc(modelStat.period_start)),
catch: (cause) => DatabaseError.make({ cause }),
})
})
@@ -88,6 +88,7 @@ export class ModelStatRepo extends Context.Service<ModelStatRepo, ModelStatRepo.
.values(chunk)
.onDuplicateKeyUpdate({
set: {
period_end: inserted("period_end"),
provider_model: inserted("provider_model"),
sessions: inserted("sessions"),
requests: inserted("requests"),
+9 -8
View File
@@ -17,8 +17,8 @@ import {
export type ProviderStatRow = typeof providerStat.$inferInsert
export type ProviderStatAggregate = StatBaseAggregate & { provider: string }
export type ProviderStatMetric = {
periodKey: string
updatedAt: Date
periodStart: Date
periodEnd: Date
tier: string
provider: string
totalTokens: number
@@ -29,7 +29,7 @@ export declare namespace ProviderStatRepo {
readonly listDaily: () => Effect.Effect<ProviderStatMetric[], DatabaseError>
readonly listByPeriod: (opts: {
readonly grain: string
readonly periodKey: string
readonly periodStart: Date
readonly dataset?: string
readonly tier?: string
readonly client?: string
@@ -52,22 +52,22 @@ export class ProviderStatRepo extends Context.Service<ProviderStatRepo, Provider
try: () =>
db
.select({
periodKey: providerStat.period_key,
updatedAt: providerStat.updated_at,
periodStart: providerStat.period_start,
periodEnd: providerStat.period_end,
tier: providerStat.tier,
provider: providerStat.provider,
totalTokens: providerStat.total_tokens,
})
.from(providerStat)
.where(and(eq(providerStat.grain, "day"), eq(providerStat.client, "all"), eq(providerStat.source, "all")))
.orderBy(asc(providerStat.period_key)),
.orderBy(asc(providerStat.period_start)),
catch: (cause) => DatabaseError.make({ cause }),
})
})
const listByPeriod = Effect.fn("ProviderStatRepo.listByPeriod")(function* (opts: {
readonly grain: string
readonly periodKey: string
readonly periodStart: Date
readonly dataset?: string
readonly tier?: string
readonly client?: string
@@ -81,7 +81,7 @@ export class ProviderStatRepo extends Context.Service<ProviderStatRepo, Provider
.where(
and(
eq(providerStat.grain, opts.grain),
eq(providerStat.period_key, opts.periodKey),
eq(providerStat.period_start, opts.periodStart),
eq(providerStat.dataset, opts.dataset ?? "zen"),
eq(providerStat.tier, opts.tier ?? "all"),
eq(providerStat.client, opts.client ?? "all"),
@@ -103,6 +103,7 @@ export class ProviderStatRepo extends Context.Service<ProviderStatRepo, Provider
.values(chunk)
.onDuplicateKeyUpdate({
set: {
period_end: inserted("period_end"),
sessions: inserted("sessions"),
requests: inserted("requests"),
input_tokens: inserted("input_tokens"),
+27 -40
View File
@@ -1,13 +1,13 @@
import { sql } from "drizzle-orm"
export const UPSERT_CHUNK_SIZE = 500
const DAY_MS = 86_400_000
export type StatGrain = "day" | "week"
export type StatBaseAggregate = {
grain: StatGrain
period_key: string
period_start: Date
period_end: Date
dataset: string
tier: string
sessions: number
@@ -34,7 +34,8 @@ export type StatBaseAggregate = {
export type StatBaseRow = {
grain: string
period_key: string
period_start: Date
period_end: Date
dataset?: string
tier?: string
client?: string
@@ -64,7 +65,8 @@ export type StatBaseRow = {
export function toStatBaseRow(data: StatBaseAggregate) {
return {
grain: data.grain,
period_key: data.period_key,
period_start: data.period_start,
period_end: data.period_end,
dataset: data.dataset,
tier: data.tier,
client: "all",
@@ -97,7 +99,14 @@ export function synthesizeAllTierRows<T extends StatBaseRow>(rows: T[], dimensio
...rows,
...Object.values(
rows.reduce<Record<string, T>>((result, row) => {
const key = [row.grain, row.period_key, row.dataset, row.client, row.source, dimensionKey(row)].join("\u0000")
const key = [
row.grain,
row.period_start.toISOString(),
row.dataset,
row.client,
row.source,
dimensionKey(row),
].join("\u0000")
result[key] = result[key] ? combineRows(result[key], row) : { ...row, tier: "all" }
return result
}, {}),
@@ -108,9 +117,15 @@ export function synthesizeAllTierRows<T extends StatBaseRow>(rows: T[], dimensio
export function collapseRows<T extends StatBaseRow>(rows: T[], dimensionKey: (row: T) => string) {
return Object.values(
rows.reduce<Record<string, T>>((result, row) => {
const key = [row.grain, row.period_key, row.dataset, row.tier, row.client, row.source, dimensionKey(row)].join(
"\u0000",
)
const key = [
row.grain,
row.period_start.toISOString(),
row.dataset,
row.tier,
row.client,
row.source,
dimensionKey(row),
].join("\u0000")
result[key] = result[key] ? combineRows(result[key], row) : row
return result
}, {}),
@@ -120,6 +135,7 @@ export function collapseRows<T extends StatBaseRow>(rows: T[], dimensionKey: (ro
export function combineRows<T extends StatBaseRow>(left: T, right: T): T {
return {
...left,
period_end: right.period_end > left.period_end ? right.period_end : left.period_end,
sessions: (left.sessions ?? 0) + (right.sessions ?? 0),
requests: (left.requests ?? 0) + (right.requests ?? 0),
input_tokens: (left.input_tokens ?? 0) + (right.input_tokens ?? 0),
@@ -144,46 +160,17 @@ export function combineRows<T extends StatBaseRow>(left: T, right: T): T {
}
export function statPeriodKey(row: StatBaseRow) {
return [row.grain, row.period_key, row.dataset, row.tier, row.client, row.source].join("\u0000")
}
export function periodKeyFor(grain: StatGrain, periodStart: Date) {
if (grain === "week") return isoWeekId(periodStart)
return utcDateId(periodStart)
}
export function startOfUtcDay(value: Date) {
return new Date(Date.UTC(value.getUTCFullYear(), value.getUTCMonth(), value.getUTCDate()))
}
export function startOfIsoWeek(value: Date) {
return new Date(
Date.UTC(value.getUTCFullYear(), value.getUTCMonth(), value.getUTCDate() - (value.getUTCDay() || 7) + 1),
)
}
export function isoWeekId(value: Date) {
const thursday = new Date(
Date.UTC(value.getUTCFullYear(), value.getUTCMonth(), value.getUTCDate() + 4 - (value.getUTCDay() || 7)),
)
return `${thursday.getUTCFullYear()}-W${String(Math.ceil(((thursday.getTime() - Date.UTC(thursday.getUTCFullYear(), 0, 1)) / DAY_MS + 1) / 7)).padStart(2, "0")}`
}
function utcDateId(value: Date) {
return `${value.getUTCFullYear()}-${String(value.getUTCMonth() + 1).padStart(2, "0")}-${String(value.getUTCDate()).padStart(2, "0")}`
return [row.grain, row.period_start.toISOString(), row.dataset, row.tier, row.client, row.source].join("\u0000")
}
export function rankBy<T extends StatBaseRow>(rows: T[], value: (row: T) => number) {
return new Map(rows.toSorted((a, b) => value(b) - value(a)).map((row, index) => [row, index + 1]))
}
export function rankRowsWithMarketShare<T extends StatBaseRow>(
rows: T[],
groupKey: (row: T) => string = statPeriodKey,
) {
export function rankRowsWithMarketShare<T extends StatBaseRow>(rows: T[]) {
return Object.values(
rows.reduce<Record<string, T[]>>((result, row) => {
const key = groupKey(row)
const key = statPeriodKey(row)
result[key] = [...(result[key] ?? []), row]
return result
}, {}),
File diff suppressed because it is too large Load Diff
+5 -8
View File
@@ -6,10 +6,8 @@ import { GeoStatRepo, rowsFromAggregates as geoRowsFromAggregates } from "./doma
import { buildStatsQuery, toGeoAggregate, toModelAggregate, toProviderAggregate } from "./domain/inference"
import { ModelStatRepo, rowsFromAggregates as modelRowsFromAggregates } from "./domain/model"
import { ProviderStatRepo, rowsFromAggregates as providerRowsFromAggregates } from "./domain/provider"
import { startOfIsoWeek } from "./domain/stat"
const DATALAKE_INGESTION_LAG_MS = 5 * 60_000
const WEEK_MS = 7 * 86_400_000
export type SyncStatsResult = { ok: true; rows: number; startedAt: string; periodStart: string; periodEnd: string }
export type SyncStatsError = AthenaQueryError | AthenaQueryTimeoutError | DatabaseError
@@ -21,7 +19,9 @@ export const syncStats: () => Effect.Effect<
> = Effect.fn("StatSync.sync")(function* () {
const startedAt = yield* DateTime.nowAsDate
const periodEnd = new Date(Math.floor((startedAt.getTime() - DATALAKE_INGESTION_LAG_MS) / 60_000) * 60_000)
const periodStart = new Date(startOfIsoWeek(periodEnd).getTime() - WEEK_MS)
const periodStart = new Date(
Date.UTC(periodEnd.getUTCFullYear(), periodEnd.getUTCMonth(), periodEnd.getUTCDate() - 6),
)
const athena = yield* Athena
const modelStats = yield* ModelStatRepo
const providerStats = yield* ProviderStatRepo
@@ -29,7 +29,7 @@ export const syncStats: () => Effect.Effect<
yield* logRuntimeCheck()
const [modelAggregates, providerAggregates, geoAggregates, geoModelAggregates] = yield* Effect.all(
const [modelAggregates, providerAggregates, geoAggregates] = yield* Effect.all(
[
athena
.query(buildStatsQuery(periodStart, periodEnd, "model"))
@@ -40,15 +40,12 @@ export const syncStats: () => Effect.Effect<
athena
.query(buildStatsQuery(periodStart, periodEnd, "geo"))
.pipe(Effect.map((rows) => rows.flatMap(toGeoAggregate))),
athena
.query(buildStatsQuery(periodStart, periodEnd, "geo_model"))
.pipe(Effect.map((rows) => rows.flatMap(toGeoAggregate))),
],
{ concurrency: "unbounded" },
)
const modelRows = modelRowsFromAggregates(modelAggregates)
const providerRows = providerRowsFromAggregates(providerAggregates)
const geoRows = geoRowsFromAggregates([...geoAggregates, ...geoModelAggregates])
const geoRows = geoRowsFromAggregates(geoAggregates)
yield* Effect.all([modelStats.upsert(modelRows), providerStats.upsert(providerRows), geoStats.upsert(geoRows)], {
concurrency: "unbounded",
+1 -1
View File
@@ -116,7 +116,7 @@ For project-specific TUI settings, add `tui.json` alongside it.
Place project specific config in the root of your project.
:::
When OpenCode starts up, it first looks for a config file in the current directory, then traverses up to the nearest Git directory.
When OpenCode starts up, it looks for a config file in the current directory or traverse up to the nearest Git directory.
This is also safe to be checked into Git and uses the same schema as the global one.
+48 -8
View File
@@ -26,14 +26,6 @@ declare module "sst" {
"AuthApi": import("@cloudflare/workers-types").Service
"AuthStorage": import("@cloudflare/workers-types").KVNamespace
"Bucket": import("@cloudflare/workers-types").R2Bucket
"CLOUDFLARE_API_TOKEN": {
"type": "sst.sst.Secret"
"value": string
}
"CLOUDFLARE_DEFAULT_ACCOUNT_ID": {
"type": "sst.sst.Secret"
"value": string
}
"Console": {
"type": "sst.cloudflare.SolidStart"
"url": string
@@ -99,6 +91,37 @@ declare module "sst" {
"type": "random.index/randomPassword.RandomPassword"
"value": string
}
"InferenceEvent": {
"catalog": string
"database": string
"region": string
"table": string
"tableBucket": string
"type": "sst.sst.Linkable"
"workgroup": string
}
"LakeIngest": {
"secret": string
"type": "sst.sst.Linkable"
"url": string
}
"LakeIngestConfig": {
"secret": string
"streamName": string
"type": "sst.sst.Linkable"
}
"LakeIngestSecret": {
"type": "random.index/randomPassword.RandomPassword"
"value": string
}
"LakeIngestService": {
"service": string
"type": "sst.aws.Service"
"url": string
}
"LakeVpc": {
"type": "sst.aws.Vpc"
}
"LogProcessor": import("@cloudflare/workers-types").Service
"R2AccessKey": {
"type": "sst.sst.Secret"
@@ -133,6 +156,23 @@ declare module "sst" {
"value": string
}
"Stat": import("@cloudflare/workers-types").Service
"StatsDatabase": {
"database": string
"host": string
"password": string
"port": number
"type": "sst.sst.Linkable"
"url": string
"username": string
}
"StatsSyncConfig": {
"dataset": string
"type": "sst.sst.Linkable"
}
"StatsSyncService": {
"service": string
"type": "sst.aws.Service"
}
"Teams": {
"type": "sst.cloudflare.SolidStart"
"url": string