Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
01635b20ee |
+1
-1
@@ -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({
|
||||
|
||||
@@ -17,7 +17,6 @@ export default defineConfig({
|
||||
],
|
||||
server: {
|
||||
allowedHosts: true,
|
||||
port: 3001,
|
||||
},
|
||||
build: {
|
||||
rollupOptions: {
|
||||
|
||||
@@ -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")
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -29,7 +29,6 @@ export default defineConfig({
|
||||
server: {
|
||||
host: "0.0.0.0",
|
||||
allowedHosts: true,
|
||||
port: 3002,
|
||||
},
|
||||
worker: {
|
||||
format: "es",
|
||||
|
||||
@@ -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"
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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 = [
|
||||
|
||||
@@ -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]),
|
||||
}
|
||||
},
|
||||
}),
|
||||
|
||||
@@ -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") {
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
@@ -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"),
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
}
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
@@ -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"),
|
||||
|
||||
@@ -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"),
|
||||
|
||||
@@ -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
@@ -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",
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
Vendored
+48
-8
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user