mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-12 04:29:50 -04:00
Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f2bce90b67 | |||
| 01cbdcb8e4 | |||
| bf751a907d | |||
| 9d63ca8f90 | |||
| e3bd82013c | |||
| 90b6fa0eab | |||
| 5b7b1830d2 | |||
| a8fc664b6d |
@@ -573,7 +573,10 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (
|
||||
cache_control: cacheControl(breakpoints, part.cache),
|
||||
})
|
||||
}
|
||||
messages.push({ role: "user", content })
|
||||
const previous = messages.at(-1)
|
||||
if (previous?.role === "user" && previous.content.every((block) => block.type === "tool_result"))
|
||||
messages[messages.length - 1] = { role: "user", content: [...previous.content, ...content] }
|
||||
else messages.push({ role: "user", content })
|
||||
}
|
||||
|
||||
return messages
|
||||
|
||||
@@ -157,7 +157,6 @@ const OpenAIChatUsage = Schema.StructWithRest(
|
||||
prompt_tokens: optionalNull(Schema.Number),
|
||||
completion_tokens: optionalNull(Schema.Number),
|
||||
total_tokens: optionalNull(Schema.Number),
|
||||
cost: optionalNull(Schema.Number),
|
||||
prompt_tokens_details: optionalNull(
|
||||
Schema.StructWithRest(
|
||||
Schema.Struct({
|
||||
@@ -596,7 +595,6 @@ const mapUsage = (usage: OpenAIChatEvent["usage"]): Usage | undefined => {
|
||||
cacheWriteInputTokens: cacheWrite,
|
||||
reasoningTokens: reasoning,
|
||||
totalTokens: ProviderShared.totalTokens(input, output, usage.total_tokens ?? undefined),
|
||||
cost: usage.cost ?? undefined,
|
||||
providerMetadata: { openai: usage },
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { Cause, Context, Effect, Layer, Option, Schema } from "effect"
|
||||
import { Cause, Context, Effect, Layer, Option, Schema, Stream } from "effect"
|
||||
import {
|
||||
FetchHttpClient,
|
||||
Headers,
|
||||
@@ -297,44 +297,86 @@ export const classifyHttpFailure = (input: {
|
||||
})
|
||||
}
|
||||
|
||||
const toHttpError = (redactedNames: ReadonlyArray<string | RegExp>) => (error: unknown) => {
|
||||
const transportError = (input: {
|
||||
readonly message: string
|
||||
readonly kind?: string | undefined
|
||||
readonly request?: HttpClientRequest.HttpClientRequest | undefined
|
||||
}) =>
|
||||
type HttpOperation = "request" | "read"
|
||||
|
||||
const NativeTransportFailure = Schema.Struct({
|
||||
message: Schema.String,
|
||||
code: Schema.optionalKey(Schema.String),
|
||||
cause: Schema.optionalKey(Schema.Unknown),
|
||||
})
|
||||
const decodeNativeTransportFailure = Schema.decodeUnknownOption(NativeTransportFailure)
|
||||
|
||||
const nativeTransportFailure = (error: unknown) => {
|
||||
const failure = Option.getOrUndefined(decodeNativeTransportFailure(error))
|
||||
if (!failure) return undefined
|
||||
if (failure.code !== undefined) return failure
|
||||
const cause = Option.getOrUndefined(decodeNativeTransportFailure(failure.cause))
|
||||
if (cause?.code !== undefined) return cause
|
||||
return failure
|
||||
}
|
||||
|
||||
const httpError = (input: {
|
||||
readonly error: unknown
|
||||
readonly request: HttpClientRequest.HttpClientRequest
|
||||
readonly operation: HttpOperation
|
||||
readonly redactedNames: ReadonlyArray<string | RegExp>
|
||||
}) => {
|
||||
const request = HttpClientError.isHttpClientError(input.error) ? input.error.request : input.request
|
||||
const transportError = (failure: { readonly message: string; readonly code?: string | undefined }) =>
|
||||
new AIError({
|
||||
module: "RequestExecutor",
|
||||
method: "execute",
|
||||
method: input.operation,
|
||||
reason: new TransportReason({
|
||||
message: input.message,
|
||||
kind: input.kind,
|
||||
url: input.request ? redactUrl(input.request.url) : undefined,
|
||||
http: input.request ? new HttpContext({ request: requestDetails(input.request, redactedNames) }) : undefined,
|
||||
message: failure.message,
|
||||
transport: "http",
|
||||
operation: input.operation,
|
||||
code: failure.code,
|
||||
url: redactUrl(request.url),
|
||||
http: new HttpContext({ request: requestDetails(request, input.redactedNames) }),
|
||||
}),
|
||||
})
|
||||
|
||||
if (Cause.isTimeoutError(error)) {
|
||||
return transportError({ message: error.message, kind: "Timeout" })
|
||||
}
|
||||
if (!HttpClientError.isHttpClientError(error)) {
|
||||
return transportError({ message: error instanceof Error ? error.message : "HTTP transport failed" })
|
||||
}
|
||||
const request = "request" in error ? error.request : undefined
|
||||
if (error.reason._tag === "TransportError") {
|
||||
const source =
|
||||
HttpClientError.isHttpClientError(input.error) && "cause" in input.error.reason
|
||||
? input.error.reason.cause
|
||||
: input.error
|
||||
const native = nativeTransportFailure(source)
|
||||
const code = native?.code
|
||||
const raw = native?.message ?? (input.error instanceof Error ? input.error.message : undefined)
|
||||
const detail = raw ? redactBody(raw, secretValues(request)) : undefined
|
||||
const message = code && detail && !detail.includes(code) ? `${code}: ${detail}` : detail
|
||||
|
||||
if (Cause.isTimeoutError(input.error) || Cause.isTimeoutError(source))
|
||||
return transportError({ message: message ?? "HTTP transport timed out", code: code ?? "Timeout" })
|
||||
if (!HttpClientError.isHttpClientError(input.error))
|
||||
return transportError({ message: message ?? "HTTP transport failed", code })
|
||||
if (input.error.reason._tag === "TransportError") {
|
||||
return transportError({
|
||||
message: error.reason.description ?? "HTTP transport failed",
|
||||
kind: error.reason._tag,
|
||||
request,
|
||||
message: message ?? input.error.reason.description ?? "HTTP transport failed",
|
||||
code: code ?? input.error.reason._tag,
|
||||
})
|
||||
}
|
||||
return transportError({
|
||||
message: `HTTP transport failed: ${error.reason._tag}`,
|
||||
kind: error.reason._tag,
|
||||
request,
|
||||
message: message ?? `HTTP transport failed: ${input.error.reason._tag}`,
|
||||
code: code ?? input.error.reason._tag,
|
||||
})
|
||||
}
|
||||
|
||||
export const stream = (
|
||||
executor: Interface,
|
||||
request: HttpClientRequest.HttpClientRequest,
|
||||
middleware?: HttpMiddleware,
|
||||
): Stream.Stream<Uint8Array, AIError> =>
|
||||
Stream.unwrap(
|
||||
Effect.gen(function* () {
|
||||
const redactedNames = yield* Headers.CurrentRedactedNames
|
||||
const response = yield* executor.execute(request, middleware)
|
||||
return response.stream.pipe(
|
||||
Stream.mapError((error) => httpError({ error, request: response.request, operation: "read", redactedNames })),
|
||||
)
|
||||
}),
|
||||
)
|
||||
|
||||
export const layer: Layer.Layer<Service, never, HttpClient.HttpClient> = Layer.effect(
|
||||
Service,
|
||||
Effect.gen(function* () {
|
||||
@@ -343,15 +385,16 @@ export const layer: Layer.Layer<Service, never, HttpClient.HttpClient> = Layer.e
|
||||
Effect.gen(function* () {
|
||||
const redactedNames = yield* Headers.CurrentRedactedNames
|
||||
if (!middleware)
|
||||
return yield* http
|
||||
.execute(request)
|
||||
.pipe(Effect.mapError(toHttpError(redactedNames)), Effect.flatMap(statusError(request, redactedNames)))
|
||||
return yield* http.execute(request).pipe(
|
||||
Effect.mapError((error) => httpError({ error, request, operation: "request", redactedNames })),
|
||||
Effect.flatMap(statusError(request, redactedNames)),
|
||||
)
|
||||
|
||||
const response = yield* middleware(request, (input) =>
|
||||
http
|
||||
.execute(input)
|
||||
.pipe(Effect.mapError((cause) => (cause instanceof Error ? cause : new Error(String(cause))))),
|
||||
).pipe(Effect.mapError(toHttpError(redactedNames)))
|
||||
).pipe(Effect.mapError((error) => httpError({ error, request, operation: "request", redactedNames })))
|
||||
return yield* statusError(response.request, redactedNames)(response)
|
||||
})
|
||||
return Service.of({
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { Effect, Stream } from "effect"
|
||||
import { Effect } from "effect"
|
||||
import { Headers, HttpClientRequest } from "effect/unstable/http"
|
||||
import { Auth } from "../auth.js"
|
||||
import { render as renderEndpoint } from "../endpoint.js"
|
||||
@@ -6,6 +6,7 @@ import { Framing } from "../framing.js"
|
||||
import type { HttpMiddleware, Transport, TransportPrepareInput } from "./index.js"
|
||||
import * as ProviderShared from "../../protocols/shared.js"
|
||||
import { mergeJsonRecords, type LLMRequest } from "../../schema/index.js"
|
||||
import { RequestExecutor } from "../executor.js"
|
||||
|
||||
export type JsonRequestInput<Body> = TransportPrepareInput<Body>
|
||||
|
||||
@@ -86,26 +87,8 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
|
||||
middleware: prepareInput.middleware,
|
||||
}
|
||||
}),
|
||||
frames: (prepared, request, runtime) =>
|
||||
Stream.unwrap(
|
||||
runtime.http
|
||||
.execute(prepared.request, prepared.middleware)
|
||||
.pipe(
|
||||
Effect.map((response) =>
|
||||
prepared.framing.frame(
|
||||
response.stream.pipe(
|
||||
Stream.mapError((error) =>
|
||||
ProviderShared.eventError(
|
||||
`${request.model.provider}/${request.model.route.id}`,
|
||||
`Failed to read ${request.model.provider}/${request.model.route.id} stream`,
|
||||
ProviderShared.errorText(error),
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
frames: (prepared, _request, runtime) =>
|
||||
prepared.framing.frame(RequestExecutor.stream(runtime.http, prepared.request, prepared.middleware)),
|
||||
})
|
||||
|
||||
export const sseJson = {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { Cause, Context, Effect, Layer, Queue, Stream } from "effect"
|
||||
import { Headers } from "effect/unstable/http"
|
||||
import { AIError, TransportReason } from "../../schema/index.js"
|
||||
import { AIError, TransportReason, type TransportOperation } from "../../schema/index.js"
|
||||
import * as HttpTransport from "./http.js"
|
||||
import type { Transport } from "./index.js"
|
||||
|
||||
@@ -29,12 +29,18 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/AI
|
||||
const transportError = (
|
||||
method: string,
|
||||
message: string,
|
||||
input: { readonly url?: string; readonly kind?: string } = {},
|
||||
input: { readonly operation: TransportOperation; readonly url?: string; readonly code?: string },
|
||||
) =>
|
||||
new AIError({
|
||||
module: "WebSocketExecutor",
|
||||
method,
|
||||
reason: new TransportReason({ message, url: input.url, kind: input.kind }),
|
||||
reason: new TransportReason({
|
||||
message,
|
||||
transport: "websocket",
|
||||
operation: input.operation,
|
||||
url: input.url,
|
||||
code: input.code,
|
||||
}),
|
||||
})
|
||||
|
||||
const eventMessage = (event: Event) => {
|
||||
@@ -55,7 +61,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
|
||||
return Effect.fail(
|
||||
transportError("open", `WebSocket closed before opening (state ${ws.readyState})`, {
|
||||
url: input.url,
|
||||
kind: "open",
|
||||
operation: "request",
|
||||
code: "closed",
|
||||
}),
|
||||
)
|
||||
}
|
||||
@@ -79,7 +86,10 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
|
||||
cleanup()
|
||||
resume(
|
||||
Effect.fail(
|
||||
transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, { url: input.url, kind: "open" }),
|
||||
transportError("open", `Failed to open WebSocket: ${eventMessage(event)}`, {
|
||||
url: input.url,
|
||||
operation: "request",
|
||||
}),
|
||||
),
|
||||
)
|
||||
}
|
||||
@@ -89,7 +99,8 @@ const waitOpen = (ws: globalThis.WebSocket, input: WebSocketRequest) => {
|
||||
Effect.fail(
|
||||
transportError("open", `WebSocket closed before opening with code ${event.code}`, {
|
||||
url: input.url,
|
||||
kind: "open",
|
||||
operation: "request",
|
||||
code: String(event.code),
|
||||
}),
|
||||
),
|
||||
)
|
||||
@@ -118,7 +129,8 @@ const webSocketUrl = (value: string) =>
|
||||
catch: (error) =>
|
||||
transportError("prepare", error instanceof Error ? error.message : "Invalid WebSocket URL", {
|
||||
url: value,
|
||||
kind: "websocket",
|
||||
operation: "request",
|
||||
code: "invalid-url",
|
||||
}),
|
||||
})
|
||||
|
||||
@@ -129,7 +141,7 @@ export const open = (input: WebSocketRequest) =>
|
||||
catch: (error) =>
|
||||
transportError("open", error instanceof Error ? error.message : "Failed to construct WebSocket", {
|
||||
url: input.url,
|
||||
kind: "open",
|
||||
operation: "request",
|
||||
}),
|
||||
}).pipe(Effect.flatMap((ws) => fromWebSocket(ws, input)))
|
||||
|
||||
@@ -150,7 +162,10 @@ export const fromWebSocket = (
|
||||
Queue.failCauseUnsafe(
|
||||
messages,
|
||||
Cause.fail(
|
||||
transportError("message", "Unsupported WebSocket message payload", { url: input.url, kind: "message" }),
|
||||
transportError("message", "Unsupported WebSocket message payload", {
|
||||
url: input.url,
|
||||
operation: "read",
|
||||
}),
|
||||
),
|
||||
)
|
||||
}
|
||||
@@ -158,7 +173,10 @@ export const fromWebSocket = (
|
||||
Queue.failCauseUnsafe(
|
||||
messages,
|
||||
Cause.fail(
|
||||
transportError("message", `WebSocket error: ${eventMessage(event)}`, { url: input.url, kind: "message" }),
|
||||
transportError("message", `WebSocket error: ${eventMessage(event)}`, {
|
||||
url: input.url,
|
||||
operation: "read",
|
||||
}),
|
||||
),
|
||||
)
|
||||
}
|
||||
@@ -167,7 +185,11 @@ export const fromWebSocket = (
|
||||
Queue.failCauseUnsafe(
|
||||
messages,
|
||||
Cause.fail(
|
||||
transportError("message", `WebSocket closed with code ${event.code}`, { url: input.url, kind: "close" }),
|
||||
transportError("message", `WebSocket closed with code ${event.code}`, {
|
||||
url: input.url,
|
||||
operation: "read",
|
||||
code: String(event.code),
|
||||
}),
|
||||
),
|
||||
)
|
||||
}
|
||||
@@ -188,7 +210,7 @@ export const fromWebSocket = (
|
||||
catch: (error) =>
|
||||
transportError("sendText", error instanceof Error ? error.message : "Failed to send WebSocket message", {
|
||||
url: input.url,
|
||||
kind: "write",
|
||||
operation: "write",
|
||||
}),
|
||||
}),
|
||||
messages: Stream.fromQueue(messages),
|
||||
@@ -243,7 +265,8 @@ export const json = <Body, Message>(input: JsonInput<Body, Message>): JsonTransp
|
||||
return Stream.fail(
|
||||
transportError("json", "WebSocket JSON transport requires WebSocketExecutor.Service", {
|
||||
url: prepared.url,
|
||||
kind: "websocket",
|
||||
operation: "request",
|
||||
code: "unavailable",
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -92,10 +92,18 @@ export class ProviderInternalReason extends Schema.Class<ProviderInternalReason>
|
||||
http: Schema.optional(HttpContext),
|
||||
}) {}
|
||||
|
||||
export const TransportType = Schema.Literals(["http", "websocket"])
|
||||
export type TransportType = typeof TransportType.Type
|
||||
|
||||
export const TransportOperation = Schema.Literals(["request", "read", "write"])
|
||||
export type TransportOperation = typeof TransportOperation.Type
|
||||
|
||||
export class TransportReason extends Schema.Class<TransportReason>("AI.Error.Transport")({
|
||||
_tag: Schema.tag("Transport"),
|
||||
message: Schema.String,
|
||||
kind: Schema.optional(Schema.String),
|
||||
transport: TransportType,
|
||||
operation: TransportOperation,
|
||||
code: Schema.optional(Schema.String),
|
||||
url: Schema.optional(Schema.String),
|
||||
http: Schema.optional(HttpContext),
|
||||
}) {}
|
||||
|
||||
@@ -56,8 +56,6 @@ export class Usage extends Schema.Class<Usage>("AI.Usage")({
|
||||
cacheWriteInputTokens: Schema.optional(Schema.Number),
|
||||
reasoningTokens: Schema.optional(Schema.Number),
|
||||
totalTokens: Schema.optional(Schema.Number),
|
||||
/** Provider-reported cost for this physical request, normalized to USD. */
|
||||
cost: Schema.optional(Schema.Number),
|
||||
providerMetadata: Schema.optional(ProviderMetadata),
|
||||
}) {
|
||||
/**
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
import { describe, expect } from "bun:test"
|
||||
import { Effect, Layer, Ref } from "effect"
|
||||
import { Headers, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
|
||||
import { Effect, Layer, Ref, Stream } from "effect"
|
||||
import { Headers, HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
|
||||
import { LLM, AIError } from "../src/index.js"
|
||||
import { LLMClient, RequestExecutor } from "../src/route.js"
|
||||
import * as OpenAIChat from "../src/protocols/openai-chat.js"
|
||||
import { dynamicResponse } from "./lib/http.js"
|
||||
import { dynamicResponse, systemError } from "./lib/http.js"
|
||||
import { deltaChunk } from "./lib/openai-chunks.js"
|
||||
import { sseRaw } from "./lib/sse.js"
|
||||
import { it } from "./lib/effect.js"
|
||||
@@ -67,6 +67,62 @@ const expectAIError = (error: unknown) => {
|
||||
const errorHttp = (error: AIError) => ("http" in error.reason ? error.reason.http : undefined)
|
||||
|
||||
describe("RequestExecutor", () => {
|
||||
it.effect("parses response body failures at the executor seam", () =>
|
||||
Effect.gen(function* () {
|
||||
const executor = yield* RequestExecutor.Service
|
||||
const error = yield* RequestExecutor.stream(executor, secretRequest).pipe(Stream.runDrain, Effect.flip)
|
||||
|
||||
expectAIError(error)
|
||||
expect(error.reason).toMatchObject({
|
||||
_tag: "Transport",
|
||||
message: "ECONNRESET: disconnected <redacted> <redacted>",
|
||||
transport: "http",
|
||||
operation: "read",
|
||||
code: "ECONNRESET",
|
||||
url: "https://provider.test/v1/chat?api_key=%3Credacted%3E&debug=1",
|
||||
})
|
||||
}).pipe(
|
||||
Effect.provide(
|
||||
responsesLayer([
|
||||
new Response(
|
||||
new ReadableStream({
|
||||
start(controller) {
|
||||
controller.error(systemError("ECONNRESET", "disconnected query-secret-123 header-secret-456"))
|
||||
},
|
||||
}),
|
||||
),
|
||||
]),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("unwraps native transport failure causes", () =>
|
||||
Effect.gen(function* () {
|
||||
const executor = yield* RequestExecutor.Service
|
||||
const error = yield* RequestExecutor.stream(executor, secretRequest).pipe(Stream.runDrain, Effect.flip)
|
||||
|
||||
expectAIError(error)
|
||||
expect(error.reason).toMatchObject({
|
||||
_tag: "Transport",
|
||||
message: "ECONNRESET: socket closed",
|
||||
operation: "read",
|
||||
code: "ECONNRESET",
|
||||
})
|
||||
}).pipe(
|
||||
Effect.provide(
|
||||
responsesLayer([
|
||||
new Response(
|
||||
new ReadableStream({
|
||||
pull(controller) {
|
||||
controller.error(new TypeError("fetch failed", { cause: systemError("ECONNRESET", "socket closed") }))
|
||||
},
|
||||
}),
|
||||
),
|
||||
]),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("preserves middleware error messages", () =>
|
||||
Effect.gen(function* () {
|
||||
const executor = yield* RequestExecutor.Service
|
||||
@@ -79,6 +135,48 @@ describe("RequestExecutor", () => {
|
||||
}).pipe(Effect.provide(responsesLayer([]))),
|
||||
)
|
||||
|
||||
it.effect("reports the request sent by middleware", () =>
|
||||
Effect.gen(function* () {
|
||||
const executor = yield* RequestExecutor.Service
|
||||
const error = yield* executor
|
||||
.execute(request, (original, handler) =>
|
||||
handler(
|
||||
original.pipe(
|
||||
HttpClientRequest.setUrl("https://proxy.test/v1/chat?api_key=proxy-secret"),
|
||||
HttpClientRequest.setHeader("authorization", "Bearer proxy-secret"),
|
||||
),
|
||||
),
|
||||
)
|
||||
.pipe(Effect.flip)
|
||||
|
||||
expectAIError(error)
|
||||
expect(error.reason).toMatchObject({
|
||||
_tag: "Transport",
|
||||
message: "ECONNRESET: proxy disconnected <redacted>",
|
||||
url: "https://proxy.test/v1/chat?api_key=%3Credacted%3E",
|
||||
http: {
|
||||
request: {
|
||||
url: "https://proxy.test/v1/chat?api_key=%3Credacted%3E",
|
||||
headers: { authorization: "<redacted>" },
|
||||
},
|
||||
},
|
||||
})
|
||||
}).pipe(
|
||||
Effect.provide(
|
||||
dynamicResponse((input) =>
|
||||
Effect.fail(
|
||||
new HttpClientError.HttpClientError({
|
||||
reason: new HttpClientError.TransportError({
|
||||
request: input.request,
|
||||
cause: systemError("ECONNRESET", "proxy disconnected proxy-secret"),
|
||||
}),
|
||||
}),
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("classifies context overflow responses", () =>
|
||||
Effect.gen(function* () {
|
||||
const executor = yield* RequestExecutor.Service
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { Effect, Layer, Ref } from "effect"
|
||||
import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
|
||||
import { HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
|
||||
import { LLMClient, RequestExecutor, WebSocketExecutor } from "../../src/route.js"
|
||||
import type { Service as LLMClientService } from "../../src/route/client.js"
|
||||
import type { Service as RequestExecutorService } from "../../src/route/executor.js"
|
||||
@@ -14,7 +14,9 @@ export type HandlerInput = {
|
||||
) => HttpClientResponse.HttpClientResponse
|
||||
}
|
||||
|
||||
export type Handler = (input: HandlerInput) => Effect.Effect<HttpClientResponse.HttpClientResponse>
|
||||
export type Handler = (
|
||||
input: HandlerInput,
|
||||
) => Effect.Effect<HttpClientResponse.HttpClientResponse, HttpClientError.HttpClientError>
|
||||
|
||||
const handlerLayer = (handler: Handler): Layer.Layer<HttpClient.HttpClient> =>
|
||||
Layer.succeed(
|
||||
@@ -34,6 +36,12 @@ const handlerLayer = (handler: Handler): Layer.Layer<HttpClient.HttpClient> =>
|
||||
|
||||
export type RuntimeEnv = RequestExecutorService | WebSocketExecutorService | LLMClientService
|
||||
|
||||
export interface SystemError extends Error {
|
||||
readonly code: string
|
||||
}
|
||||
|
||||
export const systemError = (code: string, message: string): SystemError => Object.assign(new Error(message), { code })
|
||||
|
||||
export const runtimeLayer = (layer: Layer.Layer<HttpClient.HttpClient>): Layer.Layer<RuntimeEnv> => {
|
||||
const requestExecutorLayer = RequestExecutor.layer.pipe(Layer.provide(layer))
|
||||
const deps = Layer.mergeAll(requestExecutorLayer, WebSocketExecutor.layer)
|
||||
@@ -63,14 +71,20 @@ export const dynamicResponse = (handler: Handler) => runtimeLayer(handlerLayer(h
|
||||
* Layer that emits the supplied SSE chunks and then aborts mid-stream. Used to
|
||||
* exercise transport errors that surface during parsing.
|
||||
*/
|
||||
export const truncatedStream = (chunks: ReadonlyArray<string>) =>
|
||||
export const truncatedStream = (chunks: ReadonlyArray<string>, error: Error = new Error("connection reset")) =>
|
||||
dynamicResponse((input) =>
|
||||
Effect.sync(() => {
|
||||
const encoder = new TextEncoder()
|
||||
let index = 0
|
||||
const stream = new ReadableStream({
|
||||
start(controller) {
|
||||
for (const chunk of chunks) controller.enqueue(encoder.encode(chunk))
|
||||
controller.error(new Error("connection reset"))
|
||||
pull(controller) {
|
||||
const chunk = chunks[index]
|
||||
if (chunk !== undefined) {
|
||||
index++
|
||||
controller.enqueue(encoder.encode(chunk))
|
||||
return
|
||||
}
|
||||
controller.error(error)
|
||||
},
|
||||
})
|
||||
return input.respond(stream, { headers: SSE_HEADERS })
|
||||
|
||||
@@ -271,6 +271,47 @@ describe("Anthropic Messages route", () => {
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("batches parallel tool results into one Anthropic user message", () =>
|
||||
Effect.gen(function* () {
|
||||
const prepared = yield* compileRequest(
|
||||
LLM.request({
|
||||
model,
|
||||
messages: [
|
||||
Message.user("Check both cities."),
|
||||
Message.assistant([
|
||||
{ type: "text", text: "I'll check both." },
|
||||
ToolCallPart.make({ id: "call_paris", name: "weather", input: { city: "Paris" } }),
|
||||
ToolCallPart.make({ id: "call_london", name: "weather", input: { city: "London" } }),
|
||||
]),
|
||||
Message.tool({ id: "call_paris", name: "weather", result: { temperature: 22 } }),
|
||||
Message.tool({ id: "call_london", name: "weather", result: { temperature: 18 } }),
|
||||
],
|
||||
cache: "none",
|
||||
}),
|
||||
)
|
||||
|
||||
expect(prepared.body.messages).toMatchObject([
|
||||
{ role: "user", content: [{ type: "text", text: "Check both cities." }] },
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
{ type: "text", text: "I'll check both." },
|
||||
{ type: "tool_use", id: "call_paris", name: "weather", input: { city: "Paris" } },
|
||||
{ type: "tool_use", id: "call_london", name: "weather", input: { city: "London" } },
|
||||
],
|
||||
},
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "tool_result", tool_use_id: "call_paris", content: '{"temperature":22}' },
|
||||
{ type: "tool_result", tool_use_id: "call_london", content: '{"temperature":18}' },
|
||||
],
|
||||
},
|
||||
])
|
||||
expect(prepared.body.messages).toHaveLength(3)
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("keeps tools and sends tool_choice none", () =>
|
||||
Effect.gen(function* () {
|
||||
const prepared = yield* compileRequest(
|
||||
@@ -915,6 +956,54 @@ describe("Anthropic Messages route", () => {
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("assembles and persists multiple tool calls from one Anthropic response", () =>
|
||||
Effect.gen(function* () {
|
||||
const response = yield* LLMClient.generate(request).pipe(
|
||||
Effect.provide(
|
||||
fixedResponse(
|
||||
sseEvents(
|
||||
{ type: "message_start", message: { usage: { input_tokens: 5 } } },
|
||||
{
|
||||
type: "content_block_start",
|
||||
index: 0,
|
||||
content_block: { type: "tool_use", id: "call_paris", name: "weather", input: {} },
|
||||
},
|
||||
{
|
||||
type: "content_block_delta",
|
||||
index: 0,
|
||||
delta: { type: "input_json_delta", partial_json: '{"city":"Paris"}' },
|
||||
},
|
||||
{ type: "content_block_stop", index: 0 },
|
||||
{
|
||||
type: "content_block_start",
|
||||
index: 1,
|
||||
content_block: { type: "tool_use", id: "call_london", name: "weather", input: {} },
|
||||
},
|
||||
{
|
||||
type: "content_block_delta",
|
||||
index: 1,
|
||||
delta: { type: "input_json_delta", partial_json: '{"city":"London"}' },
|
||||
},
|
||||
{ type: "content_block_stop", index: 1 },
|
||||
{ type: "message_delta", delta: { stop_reason: "tool_use" }, usage: { output_tokens: 2 } },
|
||||
{ type: "message_stop" },
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
expect(response.toolCalls).toMatchObject([
|
||||
{ id: "call_paris", name: "weather", input: { city: "Paris" } },
|
||||
{ id: "call_london", name: "weather", input: { city: "London" } },
|
||||
])
|
||||
expect(response.message.content).toMatchObject([
|
||||
{ type: "tool-call", id: "call_paris", name: "weather", input: { city: "Paris" } },
|
||||
{ type: "tool-call", id: "call_london", name: "weather", input: { city: "London" } },
|
||||
])
|
||||
expect(response.finishReason).toEqual({ normalized: "tool-calls", raw: "tool_use" })
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("keeps malformed server tool input terminal", () =>
|
||||
Effect.gen(function* () {
|
||||
const body = sseEvents(
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { describe, expect } from "bun:test"
|
||||
import { Effect, Schema, Stream } from "effect"
|
||||
import { Effect, Ref, Schema, Stream } from "effect"
|
||||
import { HttpClientRequest } from "effect/unstable/http"
|
||||
import {
|
||||
HttpOptions,
|
||||
@@ -22,7 +22,7 @@ import { ProviderShared } from "../../src/protocols/shared.js"
|
||||
import { Auth, LLMClient } from "../../src/route.js"
|
||||
import { compileRequest } from "../../src/route/client.js"
|
||||
import { it } from "../lib/effect.js"
|
||||
import { dynamicResponse, fixedResponse, truncatedStream } from "../lib/http.js"
|
||||
import { dynamicResponse, fixedResponse, systemError, truncatedStream } from "../lib/http.js"
|
||||
import { deltaChunk, usageChunk } from "../lib/openai-chunks.js"
|
||||
import { sseEvents } from "../lib/sse.js"
|
||||
|
||||
@@ -1221,12 +1221,44 @@ describe("OpenAI Chat route", () => {
|
||||
|
||||
it.effect("surfaces transport errors that occur mid-stream", () =>
|
||||
Effect.gen(function* () {
|
||||
const layer = truncatedStream([
|
||||
`data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`,
|
||||
])
|
||||
const error = yield* LLMClient.generate(request).pipe(Effect.provide(layer), Effect.flip)
|
||||
const layer = truncatedStream(
|
||||
[`data: ${JSON.stringify(deltaChunk({ role: "assistant", content: "Hello" }))}\n\n`],
|
||||
systemError("ECONNRESET", "socket closed unexpectedly"),
|
||||
)
|
||||
const events = yield* Ref.make<ReadonlyArray<LLMEvent>>([])
|
||||
const error = yield* LLMClient.stream(request).pipe(
|
||||
Stream.tap((event) => Ref.update(events, (current) => [...current, event])),
|
||||
Stream.runDrain,
|
||||
Effect.provide(layer),
|
||||
Effect.flip,
|
||||
)
|
||||
|
||||
expect(error.message).toContain("Failed to read openai/openai-chat stream")
|
||||
expect((yield* Ref.get(events)).some((event) => event.type === "text-delta")).toBeTrue()
|
||||
expect(error.reason).toMatchObject({
|
||||
_tag: "Transport",
|
||||
message: "ECONNRESET: socket closed unexpectedly",
|
||||
transport: "http",
|
||||
operation: "read",
|
||||
code: "ECONNRESET",
|
||||
url: "https://api.openai.test/v1/chat/completions",
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("surfaces transport errors before the first stream frame", () =>
|
||||
Effect.gen(function* () {
|
||||
const error = yield* LLMClient.generate(request).pipe(
|
||||
Effect.provide(truncatedStream([], systemError("ECONNRESET", "socket closed before output"))),
|
||||
Effect.flip,
|
||||
)
|
||||
|
||||
expect(error.reason).toMatchObject({
|
||||
_tag: "Transport",
|
||||
message: "ECONNRESET: socket closed before output",
|
||||
transport: "http",
|
||||
operation: "read",
|
||||
code: "ECONNRESET",
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
@@ -246,24 +246,6 @@ describe("OpenRouter", () => {
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("reports OpenRouter's streamed USD cost", () =>
|
||||
Effect.gen(function* () {
|
||||
const model = OpenRouter.configure({ apiKey: "test-key" }).model("openai/gpt-4o-mini")
|
||||
const response = yield* LLMClient.generate(LLM.request({ model, prompt: "Say hello." })).pipe(
|
||||
Effect.provide(
|
||||
fixedResponse(
|
||||
sseEvents({
|
||||
choices: [{ delta: { content: "Hello" }, finish_reason: "stop" }],
|
||||
usage: { prompt_tokens: 10, completion_tokens: 2, total_tokens: 12, cost: 0.00123 },
|
||||
}),
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
expect(response.usage?.cost).toBe(0.00123)
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("fails on a mid-stream provider error", () =>
|
||||
Effect.gen(function* () {
|
||||
const model = OpenRouter.configure({ apiKey: "test-key" }).model("openai/gpt-4o-mini")
|
||||
|
||||
+11
-37
@@ -343,8 +343,7 @@ function modelFromLanguage(info: Info, language: LanguageModelV3) {
|
||||
model: (input) =>
|
||||
LanguageModel.make({ ...input, provider: "provider" in input ? input.provider : info.providerID, route }),
|
||||
prepareTransport: (body) => Effect.succeed(body),
|
||||
streamPrepared: (prepared) =>
|
||||
streamLanguage(language, prepared as LanguageModelV3CallOptions, info.providerID === Provider.ID.githubCopilot),
|
||||
streamPrepared: (prepared) => streamLanguage(language, prepared as LanguageModelV3CallOptions),
|
||||
}
|
||||
return LanguageModel.make({
|
||||
id: info.modelID ?? info.id,
|
||||
@@ -428,7 +427,6 @@ function callOptions(request: LLMRequest): LanguageModelV3CallOptions {
|
||||
toolChoice: toolChoice(request.toolChoice),
|
||||
headers: request.http?.headers,
|
||||
providerOptions: providerOptions(request.providerOptions),
|
||||
includeRawChunks: request.model.provider === ProviderID.make(Provider.ID.githubCopilot),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -549,15 +547,8 @@ function providerOptions(input: LLMRequest["providerOptions"]): SharedV3Provider
|
||||
return Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsonObject(value)]))
|
||||
}
|
||||
|
||||
interface StreamState {
|
||||
step: number
|
||||
toolNames: Record<string, string>
|
||||
copilot: boolean
|
||||
cost?: number
|
||||
}
|
||||
|
||||
function streamLanguage(language: LanguageModelV3, options: LanguageModelV3CallOptions, copilot: boolean) {
|
||||
const state: StreamState = { step: 0, toolNames: {}, copilot }
|
||||
function streamLanguage(language: LanguageModelV3, options: LanguageModelV3CallOptions) {
|
||||
const state = { step: 0, toolNames: {} as Record<string, string> }
|
||||
return Stream.concat(
|
||||
Stream.make(LLMEvent.stepStart({ index: state.step })),
|
||||
Stream.unwrap(
|
||||
@@ -580,19 +571,17 @@ function streamLanguage(language: LanguageModelV3, options: LanguageModelV3CallO
|
||||
}
|
||||
|
||||
function streamPartEvents(
|
||||
state: StreamState,
|
||||
state: { step: number; toolNames: Record<string, string> },
|
||||
event: LanguageModelV3StreamPart,
|
||||
): Effect.Effect<ReadonlyArray<LLMEvent>, AIError> {
|
||||
switch (event.type) {
|
||||
case "stream-start":
|
||||
case "response-metadata":
|
||||
case "raw":
|
||||
case "file":
|
||||
case "source":
|
||||
case "tool-approval-request":
|
||||
return Effect.succeed([])
|
||||
case "raw":
|
||||
if (state.copilot) state.cost = copilotCost(event.rawValue) ?? state.cost
|
||||
return Effect.succeed([])
|
||||
case "text-start":
|
||||
return Effect.succeed([
|
||||
LLMEvent.textStart({ id: event.id, providerMetadata: providerMetadata(event.providerMetadata) }),
|
||||
@@ -683,18 +672,16 @@ function streamPartEvents(
|
||||
}),
|
||||
])
|
||||
case "finish":
|
||||
const normalized = usage(event.usage, state.cost)
|
||||
state.cost = undefined
|
||||
return Effect.succeed([
|
||||
LLMEvent.stepFinish({
|
||||
index: state.step++,
|
||||
reason: { normalized: finishReason(event.finishReason), raw: event.finishReason.raw },
|
||||
usage: normalized,
|
||||
usage: usage(event.usage),
|
||||
providerMetadata: providerMetadata(event.providerMetadata),
|
||||
}),
|
||||
LLMEvent.finish({
|
||||
reason: { normalized: finishReason(event.finishReason), raw: event.finishReason.raw },
|
||||
usage: normalized,
|
||||
usage: usage(event.usage),
|
||||
providerMetadata: providerMetadata(event.providerMetadata),
|
||||
}),
|
||||
])
|
||||
@@ -703,10 +690,7 @@ function streamPartEvents(
|
||||
}
|
||||
}
|
||||
|
||||
function usage(
|
||||
input: Extract<LanguageModelV3StreamPart, { type: "finish" }>["usage"],
|
||||
cost?: number,
|
||||
): UsageInput | undefined {
|
||||
function usage(input: Extract<LanguageModelV3StreamPart, { type: "finish" }>["usage"]): UsageInput | undefined {
|
||||
const output = {
|
||||
inputTokens: input.inputTokens.total,
|
||||
nonCachedInputTokens: input.inputTokens.noCache,
|
||||
@@ -718,22 +702,10 @@ function usage(
|
||||
input.inputTokens.total === undefined || input.outputTokens.total === undefined
|
||||
? undefined
|
||||
: input.inputTokens.total + input.outputTokens.total,
|
||||
cost,
|
||||
}
|
||||
return Object.values(output).some((value) => value !== undefined) ? output : undefined
|
||||
}
|
||||
|
||||
function copilotCost(input: unknown): number | undefined {
|
||||
if (!ProviderShared.isRecord(input)) return undefined
|
||||
const raw = input
|
||||
const response = ProviderShared.isRecord(raw.response) ? raw.response : undefined
|
||||
const usage = raw.copilot_usage ?? response?.copilot_usage
|
||||
if (!ProviderShared.isRecord(usage)) return undefined
|
||||
const total = usage.total_nano_aiu
|
||||
if (typeof total !== "number" || !Number.isFinite(total) || total < 0) return undefined
|
||||
return total / 100_000_000_000
|
||||
}
|
||||
|
||||
function finishReason(value: LanguageModelV3FinishReason): FinishReason {
|
||||
return value.unified === "other" ? "unknown" : value.unified
|
||||
}
|
||||
@@ -791,7 +763,9 @@ function apiCallErrorReason(error: APICallError) {
|
||||
if (error.statusCode !== undefined || !error.isRetryable) return reason
|
||||
return new TransportReason({
|
||||
message: reason.message,
|
||||
kind: error.name,
|
||||
transport: "http",
|
||||
operation: "request",
|
||||
code: error.name,
|
||||
url: error.url,
|
||||
http: "http" in reason ? reason.http : undefined,
|
||||
})
|
||||
|
||||
@@ -75,7 +75,9 @@ export const create = (
|
||||
const outputFileParts = outputFiles(content)
|
||||
if (outputFileParts.length > 0)
|
||||
yield* Ref.update(files, (items) => [...items, { index, files: outputFileParts }])
|
||||
return executed.output
|
||||
if (executed.output !== undefined) return executed.output
|
||||
const text = content.flatMap((part) => (part.type === "text" ? [part.text] : [])).join("\n")
|
||||
return text === "" ? null : text
|
||||
}),
|
||||
{
|
||||
onToolCallStart: ({ index, name, input }) => {
|
||||
@@ -155,7 +157,7 @@ function runtime(
|
||||
tools[path] = Tool.make({
|
||||
description: child.description,
|
||||
input: child.inputSchema,
|
||||
output: child.outputSchema,
|
||||
output: child.outputSchema ?? Schema.NullOr(Schema.String),
|
||||
execute: (input) => executeTool(name, registration, input),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -272,8 +272,10 @@ const layer = Layer.effect(
|
||||
snapshot: startSnapshot,
|
||||
assistantMessageID,
|
||||
})
|
||||
const stepUsage = (finish: NonNullable<StepRecord["finish"]>) =>
|
||||
SessionUsage.record(finish.usage, resolved.cost)
|
||||
const stepUsage = (finish: NonNullable<StepRecord["finish"]>) => ({
|
||||
cost: SessionUsage.calculateCost(resolved.cost, finish.tokens),
|
||||
tokens: finish.tokens,
|
||||
})
|
||||
|
||||
const captureStepEnd = Effect.fnUntraced(function* () {
|
||||
const snapshot = yield* snapshots.capture()
|
||||
|
||||
@@ -35,7 +35,7 @@ export interface StepRecord {
|
||||
/** Present once the provider finished the step normally. */
|
||||
readonly finish?: {
|
||||
readonly finish: Extract<LLMEvent, { type: "step-finish" }>["reason"]["normalized"]
|
||||
readonly usage: Extract<LLMEvent, { type: "step-finish" }>["usage"]
|
||||
readonly tokens: ReturnType<typeof SessionUsage.tokens>
|
||||
}
|
||||
readonly calls: ReadonlyArray<{
|
||||
readonly id: string
|
||||
@@ -495,7 +495,7 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
|
||||
case "step-finish":
|
||||
yield* flush()
|
||||
if (stepSettlement) return yield* Effect.die(new Error("Duplicate step finish"))
|
||||
stepSettlement = { finish: event.reason.normalized, usage: event.usage }
|
||||
stepSettlement = { finish: event.reason.normalized, tokens: SessionUsage.tokens(event.usage) }
|
||||
if (event.reason.normalized === "content-filter") {
|
||||
providerFailed = true
|
||||
yield* failAssistant({ type: "provider.content-filter", message: "Provider blocked the response" })
|
||||
|
||||
@@ -48,10 +48,12 @@ export const schedule = (
|
||||
assistantMessageID: () => SessionMessage.ID,
|
||||
) =>
|
||||
Schedule.max([Schedule.exponential("2 seconds"), Schedule.recurs(4)]).pipe(
|
||||
Schedule.jittered,
|
||||
Schedule.setInputType<RetryableFailure>(),
|
||||
Schedule.modifyDelay(({ input: failure, duration: delay }) => {
|
||||
const minimum = retryAfter(failure)
|
||||
return Effect.succeed(minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum)))
|
||||
const duration = minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum))
|
||||
return Effect.succeed(Duration.millis(Math.ceil(Duration.toMillis(duration))))
|
||||
}),
|
||||
Schedule.tap((metadata) =>
|
||||
bus.publish(SessionEvent.RetryScheduled, {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { Message, ToolCallPart, ToolResultPart, type ContentPart, type ProviderMetadata } from "@opencode-ai/ai"
|
||||
import { Option, Schema } from "effect"
|
||||
import { fileURLToPath } from "url"
|
||||
import type { Model } from "../../model.js"
|
||||
import { SessionMessage } from "../message.js"
|
||||
import type { FileAttachment } from "@opencode-ai/schema/prompt"
|
||||
@@ -14,6 +15,17 @@ const media = (file: FileAttachment): ContentPart => ({
|
||||
metadata: file.description === undefined ? undefined : { description: file.description },
|
||||
})
|
||||
|
||||
const attachmentLocation = (file: FileAttachment) => {
|
||||
if (file.source.type !== "uri") return undefined
|
||||
const url = URL.parse(file.source.uri)
|
||||
if (url?.protocol !== "file:") return undefined
|
||||
try {
|
||||
return fileURLToPath(url)
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
const textAttachment = (file: FileAttachment): ContentPart => ({
|
||||
type: "text",
|
||||
text: `\n\n${[
|
||||
@@ -36,7 +48,7 @@ const textAttachment = (file: FileAttachment): ContentPart => ({
|
||||
const directoryAttachment = (file: FileAttachment): ContentPart => ({
|
||||
type: "text",
|
||||
text: `\n\n${[
|
||||
`Attached directory: ${file.name ?? (file.source.type === "uri" ? file.source.uri : "directory")}`,
|
||||
`Attached directory: ${attachmentLocation(file) ?? file.name ?? (file.source.type === "uri" ? file.source.uri : "directory")}`,
|
||||
file.description === undefined ? undefined : `Description: ${file.description}`,
|
||||
file.data.length === 0 ? undefined : "",
|
||||
file.data.length === 0 ? undefined : Buffer.from(file.data, "base64").toString("utf8"),
|
||||
@@ -55,7 +67,10 @@ const directoryAttachment = (file: FileAttachment): ContentPart => ({
|
||||
const attachmentContent = (file: FileAttachment): ContentPart[] => {
|
||||
if (file.mime === "text/plain") return [textAttachment(file)]
|
||||
if (file.mime === "application/x-directory") return [directoryAttachment(file)]
|
||||
if (imageMimes.has(file.mime)) return [media(file)]
|
||||
if (imageMimes.has(file.mime)) {
|
||||
const location = attachmentLocation(file)
|
||||
return [...(location === undefined ? [] : [Message.text(`Attached file: ${location}`)]), media(file)]
|
||||
}
|
||||
return []
|
||||
}
|
||||
|
||||
|
||||
@@ -17,6 +17,7 @@ export const tokens = (usage: Usage | undefined): TokenUsage.Info => ({
|
||||
},
|
||||
})
|
||||
|
||||
// TODO(#35765): Use Copilot's reported billed amount once billing has a dedicated typed runtime contract.
|
||||
export function calculateCost(costs: Model.Info["cost"], usage: TokenUsage.Info) {
|
||||
const context = usage.input + usage.cache.read + usage.cache.write
|
||||
const tier = costs
|
||||
@@ -37,14 +38,7 @@ export type Recorded = { readonly tokens: TokenUsage.Info; readonly cost: Money.
|
||||
|
||||
export const record = (usage: Usage | undefined, costs: Model.Info["cost"]): Recorded => {
|
||||
const normalized = tokens(usage)
|
||||
const reported = usage?.cost
|
||||
return {
|
||||
tokens: normalized,
|
||||
cost:
|
||||
reported !== undefined && Number.isFinite(reported) && reported >= 0
|
||||
? Money.USD.make(reported)
|
||||
: calculateCost(costs, normalized),
|
||||
}
|
||||
return { tokens: normalized, cost: calculateCost(costs, normalized) }
|
||||
}
|
||||
|
||||
export const add = (a: Recorded, b: Recorded): Recorded => ({
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { APICallError } from "@ai-sdk/provider"
|
||||
import type { LanguageModelV3, LanguageModelV3CallOptions, LanguageModelV3StreamPart } from "@ai-sdk/provider"
|
||||
import type { LanguageModelV3, LanguageModelV3StreamPart } from "@ai-sdk/provider"
|
||||
import { AISDK } from "@opencode-ai/core/aisdk"
|
||||
import { SessionRunnerRetry } from "@opencode-ai/core/session/runner/retry"
|
||||
import { toSessionError } from "@opencode-ai/core/session/to-session-error"
|
||||
@@ -23,26 +23,21 @@ const model = (packageName: string, settings: Record<string, unknown> = {}) =>
|
||||
limit: { context: 100, output: 20 },
|
||||
})
|
||||
|
||||
const streamModel = (
|
||||
events: ReadonlyArray<LanguageModelV3StreamPart>,
|
||||
inspect?: (options: LanguageModelV3CallOptions) => void,
|
||||
): LanguageModelV3 => ({
|
||||
const streamModel = (events: ReadonlyArray<LanguageModelV3StreamPart>): LanguageModelV3 => ({
|
||||
specificationVersion: "v3",
|
||||
provider: "test",
|
||||
modelId: "test",
|
||||
supportedUrls: {},
|
||||
doGenerate: () => Promise.reject(new Error("Unexpected non-streaming request")),
|
||||
doStream: (options) => {
|
||||
inspect?.(options)
|
||||
return Promise.resolve({
|
||||
doStream: () =>
|
||||
Promise.resolve({
|
||||
stream: new ReadableStream({
|
||||
start(controller) {
|
||||
events.forEach((event) => controller.enqueue(event))
|
||||
controller.close()
|
||||
},
|
||||
}),
|
||||
})
|
||||
},
|
||||
}),
|
||||
})
|
||||
|
||||
const usage = {
|
||||
@@ -54,7 +49,9 @@ const client = LLMClient.layer.pipe(
|
||||
Layer.provide(
|
||||
Layer.succeed(
|
||||
RequestExecutor.Service,
|
||||
RequestExecutor.Service.of({ execute: () => Effect.die("Unexpected HTTP request") }),
|
||||
RequestExecutor.Service.of({
|
||||
execute: () => Effect.die("Unexpected HTTP request"),
|
||||
}),
|
||||
),
|
||||
),
|
||||
)
|
||||
@@ -380,41 +377,6 @@ it.effect("emits malformed AI SDK tool input without executing it", () =>
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("normalizes Copilot billed usage to USD", () =>
|
||||
Effect.gen(function* () {
|
||||
const aisdk = yield* AISDK.Service
|
||||
let options: LanguageModelV3CallOptions | undefined
|
||||
yield* aisdk.hook.sdk((event) => {
|
||||
event.sdk = {
|
||||
languageModel: () =>
|
||||
streamModel(
|
||||
[
|
||||
{
|
||||
type: "raw",
|
||||
rawValue: { type: "message_delta", copilot_usage: { total_nano_aiu: 4_473_525_000 } },
|
||||
},
|
||||
{ type: "finish", finishReason: { unified: "stop", raw: "end_turn" }, usage },
|
||||
],
|
||||
(input) => {
|
||||
options = input
|
||||
},
|
||||
),
|
||||
}
|
||||
})
|
||||
|
||||
const resolved = yield* aisdk.model({
|
||||
...model("@ai-sdk/github-copilot"),
|
||||
providerID: Provider.ID.githubCopilot,
|
||||
})
|
||||
const response = yield* LLMClient.generate(LLM.request({ model: resolved, prompt: "Hello" })).pipe(
|
||||
Effect.provide(client),
|
||||
)
|
||||
|
||||
expect(options?.includeRawChunks).toBeTrue()
|
||||
expect(response.usage?.cost).toBeCloseTo(0.04473525)
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("keeps malformed provider-executed AI SDK input terminal", () =>
|
||||
Effect.gen(function* () {
|
||||
const aisdk = yield* AISDK.Service
|
||||
@@ -582,7 +544,12 @@ it.effect("retries status-less AI SDK transport failures", () =>
|
||||
isRetryable: true,
|
||||
}),
|
||||
)
|
||||
expect(error.reason).toMatchObject({ _tag: "Transport", kind: "AI_APICallError" })
|
||||
expect(error.reason).toMatchObject({
|
||||
_tag: "Transport",
|
||||
transport: "http",
|
||||
operation: "request",
|
||||
code: "AI_APICallError",
|
||||
})
|
||||
expect(SessionRunnerRetry.isRetryable(error)).toBeTrue()
|
||||
expect("http" in error.reason ? error.reason.http?.request.url : undefined).toBe("https://api.example.com/chat")
|
||||
}),
|
||||
|
||||
@@ -248,6 +248,12 @@ const mcp = Layer.mock(MCP.Service, {
|
||||
required: ["ok"],
|
||||
},
|
||||
}),
|
||||
new MCP.Tool({
|
||||
server: MCP.ServerName.make("demo"),
|
||||
name: "status",
|
||||
description: "Status",
|
||||
inputSchema: { type: "object", properties: {} },
|
||||
}),
|
||||
new MCP.Tool({
|
||||
server: MCP.ServerName.make("direct"),
|
||||
name: "lookup",
|
||||
@@ -290,6 +296,13 @@ const mcp = Layer.mock(MCP.Service, {
|
||||
{ type: "media", data: "aGVsbG8=", mimeType: "image/png" },
|
||||
],
|
||||
})
|
||||
if (input.name === "status")
|
||||
return new MCP.ToolResult({
|
||||
server: MCP.ServerName.make(input.server),
|
||||
tool: input.name,
|
||||
isError: false,
|
||||
content: [{ type: "text", text: "hello" }],
|
||||
})
|
||||
return new MCP.ToolResult({
|
||||
server: MCP.ServerName.make(input.server),
|
||||
tool: input.name,
|
||||
@@ -984,6 +997,31 @@ it.effect("advertises MCP output schemas to Code Mode", () =>
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("returns content-only MCP results through Code Mode", () =>
|
||||
Effect.gen(function* () {
|
||||
assertion = yield* Deferred.make<Permission.AssertInput>()
|
||||
decision = Effect.void
|
||||
const registry = yield* Tool.Service
|
||||
const toolSet = yield* waitForCodeModeTool(registry, "demo.status")
|
||||
|
||||
const execution = yield* toolSet.execute({
|
||||
sessionID: Session.ID.make("ses_mcp_content_only"),
|
||||
...toolIdentity,
|
||||
call: {
|
||||
type: "tool-call",
|
||||
id: "call_mcp_content_only",
|
||||
name: "execute",
|
||||
input: { code: "return await tools.demo.status({})" },
|
||||
},
|
||||
})
|
||||
|
||||
expect(execution).toMatchObject({
|
||||
output: { output: "hello", toolCalls: [{ tool: "demo.status", status: "completed" }] },
|
||||
content: [{ type: "text", text: "hello" }],
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("advertises MCP tools directly when Code Mode is disabled for the server", () =>
|
||||
Effect.gen(function* () {
|
||||
const registry = yield* Tool.Service
|
||||
|
||||
@@ -393,4 +393,45 @@ describe("fromPromise", () => {
|
||||
expect(progress).toEqual([{ phase: "greeting" }])
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("returns content-only plugin results through Code Mode", () =>
|
||||
Effect.gen(function* () {
|
||||
const plugins = yield* Plugin.Service
|
||||
const registry = yield* Tool.Service
|
||||
const host = yield* PluginHost.make(plugins)
|
||||
const promisePlugin = define({
|
||||
id: "content-only-tool",
|
||||
setup: async (ctx) => {
|
||||
await ctx.tool.transform((tools) => {
|
||||
tools.add({
|
||||
name: "demo_status",
|
||||
description: "Returns a status string",
|
||||
input: Schema.Struct({}),
|
||||
execute: async () => ({ content: [{ type: "text", text: "hello" }] }),
|
||||
options: { codemode: true },
|
||||
})
|
||||
})
|
||||
},
|
||||
})
|
||||
|
||||
yield* PluginPromise.fromPromise(promisePlugin).effect(host)
|
||||
|
||||
const toolSet = yield* registry.snapshot()
|
||||
const throughCodeMode = yield* toolSet.execute({
|
||||
sessionID: Session.ID.make("ses_content_only_tool"),
|
||||
agent: Agent.ID.make("build"),
|
||||
messageID: SessionMessage.ID.make("msg_content_only_tool"),
|
||||
call: {
|
||||
type: "tool-call",
|
||||
id: "call_content_only_tool",
|
||||
name: "execute",
|
||||
input: { code: "return await tools.demo_status({})" },
|
||||
},
|
||||
})
|
||||
expect(throughCodeMode).toMatchObject({
|
||||
output: { output: "hello", toolCalls: [{ tool: "demo_status", status: "completed" }] },
|
||||
content: [{ type: "text", text: "hello" }],
|
||||
})
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
@@ -39,7 +39,9 @@ describe("toSessionError", () => {
|
||||
)
|
||||
expect(toSessionError(llm(new QuotaExceededReason({ message: "quota" }))).type).toBe("provider.quota")
|
||||
expect(toSessionError(llm(new ContentPolicyReason({ message: "blocked" }))).type).toBe("provider.content-filter")
|
||||
expect(toSessionError(llm(new TransportReason({ message: "transport" }))).type).toBe("provider.transport")
|
||||
expect(
|
||||
toSessionError(llm(new TransportReason({ message: "transport", transport: "http", operation: "request" }))).type,
|
||||
).toBe("provider.transport")
|
||||
expect(toSessionError(llm(new ProviderInternalReason({ message: "internal", status: 500 }))).type).toBe(
|
||||
"provider.internal",
|
||||
)
|
||||
@@ -111,7 +113,7 @@ describe("toSessionError", () => {
|
||||
const eligible = [
|
||||
llm(new RateLimitReason({ message: "rate" })),
|
||||
llm(new ProviderInternalReason({ message: "internal", status: 500 })),
|
||||
llm(new TransportReason({ message: "transport" })),
|
||||
llm(new TransportReason({ message: "transport", transport: "http", operation: "request" })),
|
||||
]
|
||||
const ineligible = [
|
||||
llm(new AuthenticationReason({ message: "auth", kind: "invalid" })),
|
||||
|
||||
@@ -32,7 +32,7 @@ describe("SessionExecution lifecycle", () => {
|
||||
new AIError({
|
||||
module: "test",
|
||||
method: "stream",
|
||||
reason: new TransportReason({ message: "Disconnected" }),
|
||||
reason: new TransportReason({ message: "Disconnected", transport: "http", operation: "request" }),
|
||||
}),
|
||||
),
|
||||
),
|
||||
|
||||
@@ -11,6 +11,8 @@ import { Shell } from "@opencode-ai/schema/shell"
|
||||
import { Location } from "@opencode-ai/schema/location"
|
||||
import { AbsolutePath } from "@opencode-ai/schema/schema"
|
||||
import { DateTime } from "effect"
|
||||
import path from "path"
|
||||
import { pathToFileURL } from "url"
|
||||
|
||||
const created = DateTime.makeUnsafe(0)
|
||||
const id = (value: string) => SessionMessage.ID.make(`msg_${value}`)
|
||||
@@ -267,12 +269,13 @@ Recent work
|
||||
])
|
||||
})
|
||||
|
||||
test("lowers directory attachments as directory context", () => {
|
||||
test("exposes admitted reference directory source paths in model context", () => {
|
||||
const location = path.resolve("/references/harness-engineering")
|
||||
const directory = FileAttachment.make({
|
||||
data: Base64.make(Buffer.from("lib/\nindex.ts").toString("base64")),
|
||||
mime: "application/x-directory",
|
||||
source: { type: "uri", uri: "file:///project/src" },
|
||||
name: "src/",
|
||||
source: { type: "uri", uri: pathToFileURL(location).href },
|
||||
name: "harness-engineering",
|
||||
})
|
||||
const messages = toLLMMessages(
|
||||
[
|
||||
@@ -295,14 +298,15 @@ Recent work
|
||||
{ type: "text", text: "Review this directory" },
|
||||
{
|
||||
type: "text",
|
||||
text: "\n\nAttached directory: src/\n\nlib/\nindex.ts",
|
||||
metadata: { attachment: { source: directory.source, name: "src/" } },
|
||||
text: `\n\nAttached directory: ${location}\n\nlib/\nindex.ts`,
|
||||
metadata: { attachment: { source: directory.source, name: "harness-engineering" } },
|
||||
},
|
||||
],
|
||||
})
|
||||
})
|
||||
|
||||
test("preserves attachment order after the prompt", () => {
|
||||
const directory = path.resolve("/project/src")
|
||||
const messages = toLLMMessages(
|
||||
[
|
||||
SessionMessage.User.make({
|
||||
@@ -313,7 +317,7 @@ Recent work
|
||||
FileAttachment.make({
|
||||
data: Base64.make(Buffer.from("index.ts").toString("base64")),
|
||||
mime: "application/x-directory",
|
||||
source: { type: "uri", uri: "file:///project/src" },
|
||||
source: { type: "uri", uri: pathToFileURL(directory).href },
|
||||
name: "src/",
|
||||
}),
|
||||
FileAttachment.make({
|
||||
@@ -332,12 +336,13 @@ Recent work
|
||||
expect(messages).toHaveLength(1)
|
||||
expect(messages[0]?.content.map((part) => (part.type === "text" ? part.text : part.type))).toEqual([
|
||||
"Review these attachments",
|
||||
"\n\nAttached directory: src/\n\nindex.ts",
|
||||
`\n\nAttached directory: ${directory}\n\nindex.ts`,
|
||||
"\n\nAttached file: main.ts\n\nexport const value = 1",
|
||||
])
|
||||
})
|
||||
|
||||
test("omits empty prompt text before an attachment", () => {
|
||||
const directory = path.resolve("/project/src")
|
||||
const messages = toLLMMessages(
|
||||
[
|
||||
SessionMessage.User.make({
|
||||
@@ -348,7 +353,7 @@ Recent work
|
||||
FileAttachment.make({
|
||||
data: Base64.make(Buffer.from("index.ts").toString("base64")),
|
||||
mime: "application/x-directory",
|
||||
source: { type: "uri", uri: "file:///project/src" },
|
||||
source: { type: "uri", uri: pathToFileURL(directory).href },
|
||||
name: "src/",
|
||||
}),
|
||||
],
|
||||
@@ -359,7 +364,9 @@ Recent work
|
||||
)
|
||||
|
||||
expect(messages).toHaveLength(1)
|
||||
expect(messages[0]?.content).toMatchObject([{ type: "text", text: "\n\nAttached directory: src/\n\nindex.ts" }])
|
||||
expect(messages[0]?.content).toMatchObject([
|
||||
{ type: "text", text: `\n\nAttached directory: ${directory}\n\nindex.ts` },
|
||||
])
|
||||
})
|
||||
|
||||
test("uses materialized image data as provider media and drops unsupported attachments", () => {
|
||||
@@ -391,6 +398,108 @@ Recent work
|
||||
])
|
||||
})
|
||||
|
||||
test("exposes admitted local image source paths before provider media", () => {
|
||||
const data = Base64.make("AAECAw==")
|
||||
const location = path.resolve("/project/IMG_3480.JPG")
|
||||
const image = FileAttachment.make({
|
||||
data,
|
||||
mime: "image/png",
|
||||
source: { type: "uri", uri: pathToFileURL(location).href },
|
||||
name: "IMG_3480.JPG",
|
||||
})
|
||||
|
||||
const messages = toLLMMessages(
|
||||
[
|
||||
SessionMessage.User.make({
|
||||
id: id("user-local-image-path"),
|
||||
type: "user",
|
||||
text: "Inspect this image",
|
||||
files: [image],
|
||||
time: { created },
|
||||
}),
|
||||
],
|
||||
model,
|
||||
)
|
||||
|
||||
expect(messages[0]?.content).toEqual([
|
||||
{ type: "text", text: "Inspect this image" },
|
||||
{ type: "text", text: `Attached file: ${location}` },
|
||||
{ type: "media", mediaType: "image/png", data, filename: "IMG_3480.JPG" },
|
||||
])
|
||||
})
|
||||
|
||||
test("falls back to attachment names for invalid local source paths", () => {
|
||||
const data = Base64.make("AAECAw==")
|
||||
const messages = toLLMMessages(
|
||||
[
|
||||
SessionMessage.User.make({
|
||||
id: id("user-invalid-local-paths"),
|
||||
type: "user",
|
||||
text: "Inspect these attachments",
|
||||
files: [
|
||||
FileAttachment.make({
|
||||
data: Base64.make(Buffer.from("index.ts").toString("base64")),
|
||||
mime: "application/x-directory",
|
||||
source: { type: "uri", uri: "file:///project/src%2Flib" },
|
||||
name: "src/",
|
||||
}),
|
||||
FileAttachment.make({
|
||||
data,
|
||||
mime: "image/png",
|
||||
source: { type: "uri", uri: "file:///project/image%2Fpreview.png" },
|
||||
name: "preview.png",
|
||||
}),
|
||||
],
|
||||
time: { created },
|
||||
}),
|
||||
],
|
||||
model,
|
||||
)
|
||||
|
||||
expect(messages[0]?.content).toEqual([
|
||||
{ type: "text", text: "Inspect these attachments" },
|
||||
{
|
||||
type: "text",
|
||||
text: "\n\nAttached directory: src/\n\nindex.ts",
|
||||
metadata: {
|
||||
attachment: {
|
||||
source: { type: "uri", uri: "file:///project/src%2Flib" },
|
||||
name: "src/",
|
||||
},
|
||||
},
|
||||
},
|
||||
{ type: "media", mediaType: "image/png", data, filename: "preview.png" },
|
||||
])
|
||||
})
|
||||
|
||||
test("does not add attachment location text for non-local provider media", () => {
|
||||
const data = Base64.make("AAECAw==")
|
||||
const messages = toLLMMessages(
|
||||
[
|
||||
SessionMessage.User.make({
|
||||
id: id("user-remote-image"),
|
||||
type: "user",
|
||||
text: "Inspect this image",
|
||||
files: [
|
||||
FileAttachment.make({
|
||||
data,
|
||||
mime: "image/png",
|
||||
source: { type: "uri", uri: "https://example.com/image.png" },
|
||||
name: "image.png",
|
||||
}),
|
||||
],
|
||||
time: { created },
|
||||
}),
|
||||
],
|
||||
model,
|
||||
)
|
||||
|
||||
expect(messages[0]?.content).toEqual([
|
||||
{ type: "text", text: "Inspect this image" },
|
||||
{ type: "media", mediaType: "image/png", data, filename: "image.png" },
|
||||
])
|
||||
})
|
||||
|
||||
test("deduplicates provider media while preserving durable attachment references", () => {
|
||||
const data = Base64.make("AAECAw==")
|
||||
const messages = toLLMMessages(
|
||||
@@ -468,7 +577,7 @@ Recent work
|
||||
FileAttachment.make({
|
||||
data,
|
||||
mime: "image/png",
|
||||
source: { type: "uri", uri: "file:///project/image.png" },
|
||||
source: { type: "uri", uri: pathToFileURL(path.resolve("/project/image.png")).href },
|
||||
name: "image.png",
|
||||
mention: { start: 0, end: 9, text: "[Image 1]" },
|
||||
}),
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { expect, test } from "bun:test"
|
||||
import { Cause, Effect, Exit, Schema } from "effect"
|
||||
import { LLMEvent } from "@opencode-ai/ai"
|
||||
import { Money } from "@opencode-ai/schema/money"
|
||||
import { Bus } from "@opencode-ai/core/bus"
|
||||
import { Event } from "@opencode-ai/schema/event"
|
||||
import { Agent } from "@opencode-ai/core/agent"
|
||||
@@ -12,7 +13,6 @@ import { Provider } from "@opencode-ai/core/provider"
|
||||
import { RelativePath } from "@opencode-ai/core/schema"
|
||||
import { Snapshot } from "@opencode-ai/core/snapshot"
|
||||
import { createLLMEventPublisher } from "@opencode-ai/core/session/runner/publish-llm-event"
|
||||
import { SessionUsage } from "@opencode-ai/core/session/usage"
|
||||
|
||||
const sessionID = Session.ID.make("ses_tool_event_test")
|
||||
const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB"
|
||||
@@ -280,7 +280,6 @@ test("content-filter finish retains failure evidence until step closeout", async
|
||||
nonCachedInputTokens: 8,
|
||||
outputTokens: 3,
|
||||
reasoningTokens: 1,
|
||||
cost: 1.25,
|
||||
},
|
||||
}),
|
||||
),
|
||||
@@ -290,13 +289,13 @@ test("content-filter finish retains failure evidence until step closeout", async
|
||||
const settlement = publisher.record().finish
|
||||
expect(settlement).toMatchObject({
|
||||
finish: "content-filter",
|
||||
usage: { nonCachedInputTokens: 8, outputTokens: 3, reasoningTokens: 1, cost: 1.25 },
|
||||
tokens: { input: 8, output: 2, reasoning: 1 },
|
||||
})
|
||||
if (!settlement) throw new Error("Expected content-filter settlement")
|
||||
const recorded = SessionUsage.record(settlement.usage, [])
|
||||
await Effect.runPromise(
|
||||
publisher.publishStepFailure({
|
||||
...recorded,
|
||||
cost: Money.USD.make(1.25),
|
||||
tokens: settlement.tokens,
|
||||
snapshot: Snapshot.ID.make("tree-end"),
|
||||
files: [RelativePath.make("src/changed.ts")],
|
||||
}),
|
||||
|
||||
@@ -515,7 +515,11 @@ const providerUnavailable = () =>
|
||||
new AIError({
|
||||
module: "test",
|
||||
method: "stream",
|
||||
reason: new TransportReason({ message: "Provider unavailable" }),
|
||||
reason: new TransportReason({
|
||||
message: "Provider unavailable",
|
||||
transport: "http",
|
||||
operation: "request",
|
||||
}),
|
||||
})
|
||||
|
||||
const incompleteStream = () =>
|
||||
@@ -3947,7 +3951,7 @@ describe("SessionRunnerLLM", () => {
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("retries eligible pre-output failures after exponential backoff", () =>
|
||||
it.effect("bounds jittered exponential backoff for eligible pre-output failures", () =>
|
||||
Effect.gen(function* () {
|
||||
const session = yield* setup
|
||||
yield* admit(session, "Retry transport")
|
||||
@@ -3956,9 +3960,9 @@ describe("SessionRunnerLLM", () => {
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
yield* TestClock.adjust("1999 millis")
|
||||
yield* TestClock.adjust("1599 millis")
|
||||
expect(requests).toHaveLength(1)
|
||||
yield* TestClock.adjust("1 millis")
|
||||
yield* TestClock.adjust("801 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(requests).toHaveLength(2)
|
||||
@@ -3983,7 +3987,7 @@ describe("SessionRunnerLLM", () => {
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(requests).toHaveLength(2)
|
||||
@@ -4028,7 +4032,7 @@ describe("SessionRunnerLLM", () => {
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(requests).toHaveLength(2)
|
||||
@@ -4085,7 +4089,7 @@ describe("SessionRunnerLLM", () => {
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(requests[1]?.messages.at(-2)).toMatchObject({
|
||||
@@ -4126,7 +4130,7 @@ describe("SessionRunnerLLM", () => {
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
while (!(yield* recordedEventTypes(sessionID)).includes("session.retry.scheduled.1")) yield* Effect.yieldNow
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(executions).toEqual(["settled"])
|
||||
@@ -4165,7 +4169,7 @@ describe("SessionRunnerLLM", () => {
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
while (!(yield* recordedEventTypes(sessionID)).includes("session.retry.scheduled.1")) yield* Effect.yieldNow
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(messageRoles(requests[1])).toEqual(["user", "assistant", "tool", "user"])
|
||||
@@ -4203,7 +4207,7 @@ describe("SessionRunnerLLM", () => {
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
for (const [index, delay] of [2_000, 4_000, 8_000, 16_000].entries()) {
|
||||
for (const [index, delay] of [2_400, 4_800, 9_600, 19_200].entries()) {
|
||||
yield* TestClock.adjust(delay)
|
||||
yield* TestLLM.wait(index + 2)
|
||||
}
|
||||
@@ -4224,7 +4228,7 @@ describe("SessionRunnerLLM", () => {
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
for (const [index, delay] of [2_000, 4_000, 8_000, 16_000].entries()) {
|
||||
for (const [index, delay] of [2_400, 4_800, 9_600, 19_200].entries()) {
|
||||
yield* TestClock.adjust(delay)
|
||||
yield* TestLLM.wait(index + 2)
|
||||
}
|
||||
@@ -4239,12 +4243,15 @@ describe("SessionRunnerLLM", () => {
|
||||
.orderBy(asc(EventTable.seq))
|
||||
.all()
|
||||
.pipe(Effect.orDie)
|
||||
expect(retries.map((event) => event.data)).toMatchObject([
|
||||
{ attempt: 2, at: 2_000 },
|
||||
{ attempt: 3, at: 6_000 },
|
||||
{ attempt: 4, at: 14_000 },
|
||||
{ attempt: 5, at: 30_000 },
|
||||
])
|
||||
for (const [index, range] of [
|
||||
[1_600, 2_400],
|
||||
[4_800, 7_200],
|
||||
[11_200, 16_800],
|
||||
[24_000, 36_000],
|
||||
].entries()) {
|
||||
expect(retries[index]?.data.at).toBeGreaterThanOrEqual(range[0]!)
|
||||
expect(retries[index]?.data.at).toBeLessThanOrEqual(range[1]!)
|
||||
}
|
||||
expect((yield* recordedEventTypes(sessionID)).filter((type) => type === "session.step.started.1")).toHaveLength(5)
|
||||
const assistant = requireAssistant(yield* session.context(sessionID))
|
||||
expect(yield* recordedStepSettlementEvents(sessionID, assistant.id)).toMatchObject([
|
||||
@@ -4274,7 +4281,7 @@ describe("SessionRunnerLLM", () => {
|
||||
|
||||
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
|
||||
yield* TestLLM.wait(1)
|
||||
yield* TestClock.adjust("2 seconds")
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
expect(requests).toHaveLength(3)
|
||||
|
||||
@@ -1,28 +0,0 @@
|
||||
import { expect, test } from "bun:test"
|
||||
import { Usage } from "@opencode-ai/ai"
|
||||
import { Money } from "@opencode-ai/schema/money"
|
||||
import { SessionUsage } from "@opencode-ai/core/session/usage"
|
||||
|
||||
const costs = [
|
||||
{
|
||||
input: Money.USDPerMillionTokens.make(1),
|
||||
output: Money.USDPerMillionTokens.make(2),
|
||||
cache: { read: Money.USDPerMillionTokens.zero, write: Money.USDPerMillionTokens.zero },
|
||||
},
|
||||
]
|
||||
|
||||
test("prefers provider-reported cost", () => {
|
||||
expect(SessionUsage.record(new Usage({ nonCachedInputTokens: 1_000_000, cost: 0.25 }), costs).cost).toBe(
|
||||
Money.USD.make(0.25),
|
||||
)
|
||||
expect(SessionUsage.record(new Usage({ nonCachedInputTokens: 1_000_000, cost: 0 }), costs).cost).toBe(Money.USD.zero)
|
||||
})
|
||||
|
||||
test("falls back to catalog pricing for invalid reported cost", () => {
|
||||
expect(SessionUsage.record(new Usage({ nonCachedInputTokens: 1_000_000, cost: Number.NaN }), costs).cost).toBe(
|
||||
Money.USD.make(1),
|
||||
)
|
||||
expect(SessionUsage.record(new Usage({ nonCachedInputTokens: 1_000_000, cost: -1 }), costs).cost).toBe(
|
||||
Money.USD.make(1),
|
||||
)
|
||||
})
|
||||
@@ -7,15 +7,6 @@ import { isAllowedCorsOrigin } from "./cors"
|
||||
import { createRoutes } from "./routes"
|
||||
import type { ServerOptions } from "./options"
|
||||
|
||||
export interface BootOptions {
|
||||
/**
|
||||
* Resumes execution-journaled Sessions once the application layer boots. Pair with
|
||||
* `SessionExecution.configured({ suspendOnStart: true })` on runtimes that can die without
|
||||
* teardown, so turns orphaned by a hard death replay on the next boot.
|
||||
*/
|
||||
readonly resumeSuspendedSessions?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds a web-standard fetch handler — `(request: Request) => Promise<Response>` — serving the
|
||||
* same HttpApi routes as the Node server process without binding a port, owning a listener, or
|
||||
@@ -32,13 +23,17 @@ export interface BootOptions {
|
||||
* Auth follows `createRoutes` semantics: `options.password` enforces Basic auth; omitting it
|
||||
* serves unauthenticated, so an embedder without a password must front the handler with its own
|
||||
* access control.
|
||||
*
|
||||
* Sessions whose execution claim was never released resume once the layer is built, exactly as
|
||||
* the Node server process does: a runtime that dies without teardown — an evicted Durable
|
||||
* Object leaves the same durable signature as a killed process — replays orphaned turns on the
|
||||
* next boot, and the sweep is a no-op when nothing is suspended.
|
||||
*/
|
||||
export const make = Effect.fn("ServerFetch.make")(function* (options: ServerOptions = {}, boot: BootOptions = {}) {
|
||||
export const make = Effect.fn("ServerFetch.make")(function* (options: ServerOptions = {}) {
|
||||
const context = yield* Layer.build(createRoutes(options, () => []).pipe(Layer.provide(HttpServer.layerServices)))
|
||||
// Forked so the returned handler is never delayed; resumed drains are already
|
||||
// logged and durably recorded by the execution layer.
|
||||
if (boot.resumeSuspendedSessions)
|
||||
yield* Effect.forkDetach(Context.get(context, SessionRestart.Service).resumeSuspendedSessions)
|
||||
yield* Effect.forkDetach(Context.get(context, SessionRestart.Service).resumeSuspendedSessions)
|
||||
return Context.get(context, HttpRouter.HttpRouter)
|
||||
.asHttpEffect()
|
||||
.pipe(
|
||||
|
||||
@@ -1814,7 +1814,7 @@ function CompactionMessage(props: { message: Extract<SessionMessageInfo, { type:
|
||||
streaming={true}
|
||||
internalBlockMode="top-level"
|
||||
content={content()}
|
||||
tableOptions={{ style: "grid" }}
|
||||
tableOptions={{ style: "grid", cellPaddingX: 1 }}
|
||||
conceal={ctx.markdownMode() === "rendered"}
|
||||
fg={theme.markdown.text}
|
||||
bg={theme.background.default}
|
||||
@@ -2264,7 +2264,7 @@ function TextPart(props: { last: boolean; part: SessionMessageAssistantText }) {
|
||||
streaming={true}
|
||||
internalBlockMode="top-level"
|
||||
content={props.part.text.trim()}
|
||||
tableOptions={{ style: "grid" }}
|
||||
tableOptions={{ style: "grid", cellPaddingX: 1 }}
|
||||
conceal={ctx.markdownMode() === "rendered"}
|
||||
fg={theme.markdown.text}
|
||||
bg={theme.background.default}
|
||||
|
||||
Reference in New Issue
Block a user