Compare commits

..

1 Commits

Author SHA1 Message Date
Kit Langton 404701baac fix(util): defer module-scope randomness, I/O, and import.meta paths 2026-08-11 10:45:45 -04:00
22 changed files with 68 additions and 233 deletions
+15 -28
View File
@@ -297,7 +297,7 @@ export const classifyHttpFailure = (input: {
})
}
export const mapHttpError = (error: unknown, redactedNames: ReadonlyArray<string | RegExp>) => {
const toHttpError = (redactedNames: ReadonlyArray<string | RegExp>) => (error: unknown) => {
const transportError = (input: {
readonly message: string
readonly kind?: string | undefined
@@ -314,35 +314,23 @@ export const mapHttpError = (error: unknown, redactedNames: ReadonlyArray<string
}),
})
const cause =
HttpClientError.isHttpClientError(error) && "cause" in error.reason
? error.reason.cause
: error instanceof Error
? error.cause
: undefined
const code = [cause, error]
.map((value) => (typeof value === "object" && value !== null ? Reflect.get(value, "code") : undefined))
.find((value): value is string => typeof value === "string")
const request = HttpClientError.isHttpClientError(error) && "request" in error ? error.request : undefined
const raw = cause instanceof Error ? cause.message : error instanceof Error ? error.message : undefined
const detail = raw && request ? redactBody(raw, secretValues(request)) : raw
const message = code && detail && !detail.includes(code) ? `${code}: ${detail}` : detail
if (Cause.isTimeoutError(error) || Cause.isTimeoutError(cause))
return transportError({ message: message ?? "HTTP transport timed out", kind: code ?? "Timeout", request })
if (!HttpClientError.isHttpClientError(error)) {
return transportError({ message: message ?? "HTTP transport failed", kind: code, request })
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") {
return transportError({
message: message ?? error.reason.description ?? "HTTP transport failed",
kind: code ?? error.reason._tag,
message: error.reason.description ?? "HTTP transport failed",
kind: error.reason._tag,
request,
})
}
return transportError({
message: message ?? `HTTP transport failed: ${error.reason._tag}`,
kind: code ?? error.reason._tag,
message: `HTTP transport failed: ${error.reason._tag}`,
kind: error.reason._tag,
request,
})
}
@@ -355,16 +343,15 @@ 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((error) => mapHttpError(error, redactedNames)),
Effect.flatMap(statusError(request, redactedNames)),
)
return yield* http
.execute(request)
.pipe(Effect.mapError(toHttpError(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((error) => mapHttpError(error, redactedNames)))
).pipe(Effect.mapError(toHttpError(redactedNames)))
return yield* statusError(response.request, redactedNames)(response)
})
return Service.of({
+9 -6
View File
@@ -2,7 +2,6 @@ import { Effect, Stream } from "effect"
import { Headers, HttpClientRequest } from "effect/unstable/http"
import { Auth } from "../auth"
import { render as renderEndpoint } from "../endpoint"
import { mapHttpError } from "../executor"
import { Framing } from "../framing"
import type { HttpMiddleware, Transport, TransportPrepareInput } from "./index"
import * as ProviderShared from "../../protocols/shared"
@@ -87,16 +86,20 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
middleware: prepareInput.middleware,
}
}),
frames: (prepared, _request, runtime) =>
frames: (prepared, request, runtime) =>
Stream.unwrap(
runtime.http
.execute(prepared.request, prepared.middleware)
.pipe(
Effect.map((response) =>
Stream.unwrap(
Effect.map(Headers.CurrentRedactedNames, (redactedNames) =>
prepared.framing.frame(
response.stream.pipe(Stream.mapError((error) => mapHttpError(error, redactedNames))),
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),
),
),
),
),
+2 -2
View File
@@ -63,14 +63,14 @@ 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>, error: Error = new Error("connection reset")) =>
export const truncatedStream = (chunks: ReadonlyArray<string>) =>
dynamicResponse((input) =>
Effect.sync(() => {
const encoder = new TextEncoder()
const stream = new ReadableStream({
start(controller) {
for (const chunk of chunks) controller.enqueue(encoder.encode(chunk))
controller.error(error)
controller.error(new Error("connection reset"))
},
})
return input.respond(stream, { headers: SSE_HEADERS })
+4 -10
View File
@@ -1221,18 +1221,12 @@ 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`],
Object.assign(new Error("socket closed unexpectedly"), { code: "ECONNRESET" }),
)
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)
expect(error.reason).toMatchObject({
_tag: "Transport",
message: "ECONNRESET: socket closed unexpectedly",
kind: "ECONNRESET",
url: "https://api.openai.test/v1/chat/completions",
})
expect(error.message).toContain("Failed to read openai/openai-chat stream")
}),
)
@@ -69,63 +69,6 @@ describe("v2 session reducer", () => {
})
})
test("prefers durable selection predecessors and derives them for older events", () => {
const source: SessionMessageInfo[] = [
{ id: "msg_previous_agent", type: "agent-switched", agent: "build", time: { created: 1 } },
{
id: "msg_previous_model",
type: "model-switched",
model: { id: "old", providerID: "provider" },
time: { created: 1 },
},
]
const reducer = createV2SessionReducer()
const agent = reducer.reduce(
source,
event({
...base,
id: "evt_agent",
type: "session.agent.selected",
data: { sessionID: "ses_1", agent: "plan", previous: "review" },
}),
)
const model = reducer.reduce(
source,
event({
...base,
id: "evt_model",
type: "session.model.selected",
data: {
sessionID: "ses_1",
model: { id: "new", providerID: "provider" },
previous: { id: "durable", providerID: "provider" },
},
}),
)
const legacyAgent = reducer.reduce(
source,
event({
...base,
id: "evt_legacy_agent",
type: "session.agent.selected",
data: { sessionID: "ses_1", agent: "plan" },
}),
)
expect(agent?.messages.at(-1)).toMatchObject({ type: "agent-switched", agent: "plan", previous: "review" })
expect(model?.messages.at(-1)).toMatchObject({
type: "model-switched",
model: { id: "new" },
previous: { id: "durable" },
})
expect(legacyAgent?.messages.at(-1)).toMatchObject({
type: "agent-switched",
agent: "plan",
previous: "build",
})
})
test("folds tool, retry, and completion events", () => {
const reducer = createV2SessionReducer()
let messages: SessionMessageInfo[] = []
@@ -61,12 +61,6 @@ export function createV2SessionReducer() {
type: "agent-switched",
metadata: event.metadata,
agent: event.data.agent,
previous:
event.data.previous ??
source.findLast(
(item): item is Extract<SessionMessageInfo, { type: "agent-switched" | "assistant" }> =>
item.type === "agent-switched" || item.type === "assistant",
)?.agent,
time: { created: event.created },
})
case "session.model.selected":
@@ -75,12 +69,10 @@ export function createV2SessionReducer() {
type: "model-switched",
metadata: event.metadata,
model: event.data.model,
previous:
event.data.previous ??
source.findLast(
(item): item is Extract<SessionMessageInfo, { type: "model-switched" | "assistant" }> =>
item.type === "model-switched" || item.type === "assistant",
)?.model,
previous: source.findLast(
(item): item is Extract<SessionMessageInfo, { type: "model-switched" | "assistant" }> =>
item.type === "model-switched" || item.type === "assistant",
)?.model,
time: { created: event.created },
})
case "session.synthetic":
+2 -10
View File
@@ -339,11 +339,7 @@ export type Endpoint5_31Output =
readonly type: "session.agent.selected"
readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version }
readonly location?: Location.Ref | undefined
readonly data: {
readonly sessionID: Session.ID
readonly agent: Agent.ID
readonly previous?: Agent.ID | undefined
}
readonly data: { readonly sessionID: Session.ID; readonly agent: Agent.ID }
}
| {
readonly id: Event.ID
@@ -352,11 +348,7 @@ export type Endpoint5_31Output =
readonly type: "session.model.selected"
readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version }
readonly location?: Location.Ref | undefined
readonly data: {
readonly sessionID: Session.ID
readonly model: Model.Ref
readonly previous?: Model.Ref | undefined
}
readonly data: { readonly sessionID: Session.ID; readonly model: Model.Ref }
}
| {
readonly id: Event.ID
@@ -436,7 +436,7 @@ export type SessionAgentSelected = {
type: "session.agent.selected"
durable: { aggregateID: string; seq: number; version: 1 }
location?: LocationRef
data: { sessionID: string; agent: string; previous?: string }
data: { sessionID: string; agent: string }
}
export type SessionModelSelected = {
@@ -446,7 +446,7 @@ export type SessionModelSelected = {
type: "session.model.selected"
durable: { aggregateID: string; seq: number; version: 1 }
location?: LocationRef
data: { sessionID: string; model: ModelRef; previous?: ModelRef }
data: { sessionID: string; model: ModelRef }
}
export type SessionMoved = {
-1
View File
@@ -36,7 +36,6 @@ export const ModelsDevPlugin = define({
draft.integrationID = Integration.ID.make(provider.info.id)
})
for (const model of provider.models) {
if (model.status === "deprecated") continue
catalog.model.update(provider.info.id, model.id, (draft) => Object.assign(draft, model))
}
}
+1 -3
View File
@@ -716,11 +716,10 @@ const layer = Layer.effect(
.pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid)
}),
switchAgent: Effect.fn("Session.switchAgent")(function* (input) {
const session = yield* result.get(input.sessionID)
yield* result.get(input.sessionID)
yield* bus.publish(SessionEvent.AgentSelected, {
sessionID: input.sessionID,
agent: input.agent,
previous: session.agent,
})
}),
switchModel: Effect.fn("Session.switchModel")(function* (input) {
@@ -734,7 +733,6 @@ const layer = Layer.effect(
yield* bus.publish(SessionEvent.ModelSelected, {
sessionID: input.sessionID,
model: input.model,
previous: session.model,
})
}),
rename: Effect.fn("Session.rename")(function* (input) {
+2 -2
View File
@@ -61,7 +61,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
"session.usage.recorded": () => Effect.void,
"session.agent.selected": (event) => {
return Effect.gen(function* () {
const previous = event.data.previous ?? (yield* adapter.getAgent())
const previous = yield* adapter.getAgent()
yield* adapter.appendMessage(
SessionMessage.AgentSelected.make({
id: SessionMessage.ID.fromEvent(event.id),
@@ -76,7 +76,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
},
"session.model.selected": (event) => {
return Effect.gen(function* () {
const previous = event.data.previous ?? (yield* adapter.getModel())
const previous = yield* adapter.getModel()
yield* adapter.appendMessage(
SessionMessage.ModelSelected.make({
id: SessionMessage.ID.fromEvent(event.id),
@@ -215,66 +215,6 @@ describe("ModelsDevPlugin", () => {
}),
)
it.effect("omits deprecated models from the catalog", () =>
Effect.gen(function* () {
const integrations = yield* Integration.Service
const catalog = yield* Catalog.Service
const providerID = Provider.ID.make("acme")
const activeID = Model.ID.make("current")
const deprecatedID = Model.ID.make("legacy")
const model = {
modelID: activeID,
providerID,
name: "Current",
capabilities: { tools: true, input: [], output: [] },
variants: [],
time: { released: Date.parse("2026-01-01") },
cost: [],
status: "active",
enabled: true,
limit: { context: 128_000, output: 32_000 },
} satisfies Omit<Model.Info, "id">
const snapshots = [
{
info: {
id: providerID,
name: "Acme",
package: Provider.aisdk("@ai-sdk/openai-compatible"),
},
environment: [],
models: [
{ id: activeID, ...model },
{
id: deprecatedID,
...model,
modelID: deprecatedID,
name: "Legacy",
status: "deprecated" as const,
},
],
},
] satisfies readonly ModelsDev.Snapshot[]
yield* ModelsDevPlugin.effect(
host({
catalog: catalogHost(catalog),
integration: integrationHost(integrations),
}),
).pipe(
Effect.provideService(
ModelsDev.Service,
ModelsDev.Service.of({
get: () => Effect.succeed(snapshots),
refresh: () => Effect.void,
}),
),
)
expect(yield* catalog.model.get(providerID, activeID)).toBeDefined()
expect(yield* catalog.model.get(providerID, deprecatedID)).toBeUndefined()
}),
)
it.effect("registers key methods for providers with environment variables", () =>
Effect.gen(function* () {
const integrations = yield* Integration.Service
+3 -11
View File
@@ -654,7 +654,7 @@ describe("Session.create", () => {
expect(yield* session.get(created.id)).toMatchObject({ agent: "plan" })
expect(
Array.from(yield* logEvents(session, created.id, true).pipe(Stream.drop(1), Stream.take(1), Stream.runCollect)),
).toMatchObject([{ type: "session.agent.selected", data: { agent: "plan", previous: "build" } }])
).toMatchObject([{ type: "session.agent.selected", data: { agent: "plan" } }])
expect(yield* session.messages({ sessionID: created.id, order: "asc" })).toMatchObject([
{ type: "agent-switched", agent: "plan", previous: "build" },
])
@@ -678,12 +678,7 @@ describe("Session.create", () => {
it.effect("switches the selected model through the durable Session event", () =>
Effect.gen(function* () {
const session = yield* Session.Service
const previous = Model.Ref.make({
id: Model.ID.make("haiku"),
providerID: Provider.ID.anthropic,
variant: Model.VariantID.make("default"),
})
const created = yield* session.create({ location, model: previous })
const created = yield* session.create({ location })
const model = Model.Ref.make({
id: Model.ID.make("sonnet"),
providerID: Provider.ID.anthropic,
@@ -697,10 +692,7 @@ describe("Session.create", () => {
yield* logEvents(session, created.id, true).pipe(Stream.drop(1), Stream.take(1), Stream.runCollect),
)
expect(bus).toMatchObject([{ type: "session.model.selected" }])
expect(bus[0]?.data).toEqual({ sessionID: created.id, model, previous })
expect(yield* session.messages({ sessionID: created.id, order: "asc" })).toMatchObject([
{ type: "model-switched", model, previous },
])
expect(bus[0]?.data).toEqual({ sessionID: created.id, model })
}),
)
-6
View File
@@ -14373,9 +14373,6 @@
},
"agent": {
"type": "string"
},
"previous": {
"type": "string"
}
},
"required": ["sessionID", "agent"],
@@ -14444,9 +14441,6 @@
},
"model": {
"$ref": "#/components/schemas/Model.Ref"
},
"previous": {
"$ref": "#/components/schemas/Model.Ref"
}
},
"required": ["sessionID", "model"],
-2
View File
@@ -69,7 +69,6 @@ export const AgentSelected = Event.durable({
schema: {
...Base,
agent: Agent.ID,
previous: Agent.ID.pipe(optional),
},
})
export type AgentSelected = typeof AgentSelected.Type
@@ -80,7 +79,6 @@ export const ModelSelected = Event.durable({
schema: {
...Base,
model: Model.Ref,
previous: Model.Ref.pipe(optional),
},
})
export type ModelSelected = typeof ModelSelected.Type
+4 -2
View File
@@ -7,13 +7,15 @@ import Config from "@npmcli/config"
import { definitions, flatten, nerfDarts, shorthands } from "@npmcli/config/lib/definitions/index.js"
import { Effect } from "effect"
const npmPath = fileURLToPath(new URL("..", import.meta.url))
// Lazy: on workerd import.meta.url is undefined and constructing a URL from it
// at module scope fails startup validation; npm config is never used there.
const npmPath = () => fileURLToPath(new URL("..", import.meta.url))
export const load = (dir: string) =>
Effect.tryPromise({
try: async () => {
const config = new Config({
npmPath,
npmPath: npmPath(),
cwd: dir,
env: { ...process.env },
argv: [process.execPath, process.execPath, "--prefix", dir],
+8 -2
View File
@@ -1,6 +1,6 @@
export * as Observability from "./observability.js"
import { NodeFileSystem } from "@effect/platform-node"
import * as NodeFileSystem from "@effect/platform-node/NodeFileSystem"
import { LayerNode } from "./effect/layer-node.js"
import { Effect, Layer, Logger, References, Schema } from "effect"
import { FetchHttpClient } from "effect/unstable/http"
@@ -50,4 +50,10 @@ export function layer(
).pipe(Layer.catchCause(() => local))
}
export const node = LayerNode.make({ name: "observability", layer: layer(), deps: [] })
// Layer.suspend: constructing the loggers eagerly at module scope performs
// I/O (file logger, run id) that workerd forbids in global scope.
export const node = LayerNode.make({
name: "observability",
layer: Layer.suspend(() => layer()),
deps: [],
})
+2 -2
View File
@@ -3,7 +3,7 @@ import path from "path"
import { Global } from "../global.js"
import { runID } from "./shared.js"
function formatter(id: string = runID) {
function formatter(id: string = runID()) {
return Logger.map(Logger.formatStructured, (output) => {
const messages = Array.isArray(output.message) ? output.message : [output.message]
return [
@@ -51,7 +51,7 @@ export function file(local = true, channel = "local") {
return path.join(Global.Path.log, `opencode-${channel.replace(/[^a-zA-Z0-9._-]/g, "-")}.log`)
}
export function fileLogger(target = file(), id: string = runID) {
export function fileLogger(target = file(), id: string = runID()) {
// Do not set batchWindow to 0; it causes high idle CPU usage.
return Effect.gen(function* () {
const fs = yield* FileSystem.FileSystem
+2 -2
View File
@@ -54,8 +54,8 @@ export function resource(app: App = { client: "opencode", version: "unknown", ch
...resourceAttributes(),
"deployment.environment.name": app.channel,
"opencode.client": app.client,
"opencode.run": runID,
"service.instance.id": runID,
"opencode.run": runID(),
"service.instance.id": runID(),
},
}
}
+8 -1
View File
@@ -1 +1,8 @@
export const runID = crypto.randomUUID().slice(0, 8)
// Lazy: workerd forbids generating random values in global scope, so the id
// materializes on first call (inside a handler) and stays stable afterwards.
let generated: string | undefined
export function runID(): string {
generated ??= crypto.randomUUID().slice(0, 8)
return generated
}
-6
View File
@@ -14373,9 +14373,6 @@
},
"agent": {
"type": "string"
},
"previous": {
"type": "string"
}
},
"required": ["sessionID", "agent"],
@@ -14444,9 +14441,6 @@
},
"model": {
"$ref": "#/components/schemas/Model.Ref"
},
"previous": {
"$ref": "#/components/schemas/Model.Ref"
}
},
"required": ["sessionID", "model"],
-6
View File
@@ -14373,9 +14373,6 @@
},
"agent": {
"type": "string"
},
"previous": {
"type": "string"
}
},
"required": ["sessionID", "agent"],
@@ -14444,9 +14441,6 @@
},
"model": {
"$ref": "#/components/schemas/Model.Ref"
},
"previous": {
"$ref": "#/components/schemas/Model.Ref"
}
},
"required": ["sessionID", "model"],