ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(llm): remove the provider-error stream event

Browse source 
Delete the provider-error LLMEvent so streams carry output only and
every provider-reported failure exits through the typed error channel.
Anthropic SSE error events, OpenAI Responses response.failed/error
events, and Bedrock exception frames now fail the stream with an
LLMError from the shared classifier (carrying the provider code, e.g.
rate_limit_exceeded -> LLM.RateLimit).

Core follows: the publisher drops its provider-error case, the runner
drops held-back overflow events (overflow recovery keys off a thrown
LLM.ContextOverflow), compaction and title stop scanning events for
failures, and isContextOverflowFailure is deleted with its last
consumer. V1 packages/opencode gets minimal compile fixes only (dead
switch case, one test stream).
This commit is contained in:
Aiden Cline 2026-07-13 10:49:43 -05:00
commit 5d87c7ad1f
18 changed files with 167 additions and 240 deletions
Download patch file Download diff file Expand all files Collapse all files

5
packages/core/src/session/compaction.ts
View file

@ -247,11 +247,6 @@ const make = (dependencies: Dependencies) => {
)
.pipe(
Stream.runForEach((event) => {
if (LLMEvent.is.providerError(event))
failure = {
type: event.classification === "context-overflow" ? "provider.invalid-request" : "provider.error",
message: event.message,
}
if (LLMEvent.is.textDelta(event)) {
chunks.push(event.text)
return dependencies.events.publish(SessionEvent.Compaction.Delta, {

34
packages/core/src/session/runner/llm.ts
View file

@ -1,15 +1,6 @@
export * as SessionRunnerLLM from "./llm"
import {
LLM,
LLMClient,
LLMEvent,
Message,
SystemPart,
isContextOverflowFailure,
isLLMError,
type ProviderErrorEvent,
} from "@opencode-ai/llm"
import { LLM, LLMClient, LLMEvent, Message, SystemPart, isLLMError, type LLMError } from "@opencode-ai/llm"
import { SessionError } from "@opencode-ai/schema/session-error"
import { Money } from "@opencode-ai/schema/money"
import { Cause, Effect, Exit, Fiber, FiberSet, Layer, Option, Semaphore, Stream } from "effect"
@ -227,17 +218,10 @@ const layer = Layer.effect(
// mid-event.
const serialized = <A, E, R>(effect: Effect.Effect<A, E, R>) => publication.withPermit(effect)
const publish = (event: LLMEvent, error?: SessionError.Error) => serialized(publisher.publish(event, error))
let overflowFailure: ProviderErrorEvent | undefined
const providerStream = llm.stream(request).pipe(
Stream.runForEach((event) =>
Effect.gen(function* () {
if (overflowFailure || publisher.hasProviderError()) return
if (LLMEvent.is.providerError(event)) {
if (isContextOverflowFailure(event) && !publisher.hasRetryEvidence()) {
overflowFailure = event
return
}
}
if (publisher.hasProviderError()) return
yield* publish(event)
if (event.type !== "tool-call" || event.providerExecuted) return
if (!toolMaterialization) {
@ -317,22 +301,21 @@ const layer = Layer.effect(
// away non-interrupt failures, so both interrupt checks stay Cause-based.
const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)
const llmFailure = streamFailure !== undefined && isLLMError(streamFailure) ? streamFailure : undefined
// A context overflow before any assistant output is recoverable: compact and
// restart the step instead of surfacing the provider error.
if (
recoverOverflow &&
!publisher.hasRetryEvidence() &&
isContextOverflowFailure(overflowFailure ?? streamFailure) &&
llmFailure?._tag === "LLM.ContextOverflow" &&
(yield* restore(recoverOverflow({ sessionID: session.id, messages: context, model }))).status ===
"completed"
)
return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
// An unrecovered held-back overflow becomes the step's durable provider error. A
// thrown LLM failure records the assistant failure unless a provider error was
// already recorded from the stream. Terminal publication waits for owned tools.
if (overflowFailure) yield* publish(overflowFailure)
const llmFailure = streamFailure !== undefined && isLLMError(streamFailure) ? streamFailure : undefined
// A thrown LLM failure records the assistant failure unless a provider failure
// was already recorded from the stream. Terminal publication waits for owned tools.
if (llmFailure && !publisher.hasProviderError()) {
const error = toSessionError(llmFailure)
if (
@ -349,7 +332,8 @@ const layer = Layer.effect(
}
yield* serialized(publisher.failAssistant(error))
}
// Provider error events only arrive from the stream, so the flag is final here.
// The provider-failed flag is only set while consuming the stream (content-filter
// step finish), so it is final here.
const providerFailed = publisher.hasProviderError()
// Settle every owned tool fiber. FiberSet.join returns on the first failure, so retain

4
packages/core/src/session/runner/publish-llm-event.ts
View file

@ -438,10 +438,6 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
return
case "finish":
return
case "provider-error":
providerFailed = true
yield* failAssistant({ type: "provider.unknown", message: event.message }, true)
return
}
})

4
packages/core/src/session/title.ts
View file

@ -49,7 +49,6 @@ const make = (dependencies: Dependencies) => {
).pipe(Effect.catch(() => Effect.succeed(undefined)))
if (!resolved) return
const chunks: string[] = []
let failed = false
const streamed = yield* dependencies.llm
.stream(
LLM.request({
@ -61,14 +60,13 @@ const make = (dependencies: Dependencies) => {
)
.pipe(
Stream.runForEach((event) => {
if (LLMEvent.is.providerError(event)) failed = true
if (LLMEvent.is.textDelta(event)) chunks.push(event.text)
return Effect.void
}),
Effect.as(true),
Effect.catchIf(isLLMError, () => Effect.succeed(false)),
)
if (!streamed || failed) return
if (!streamed) return
const title = chunks
.join("")
.split("\n")

104
packages/core/test/session-runner.test.ts
View file

@ -1,5 +1,6 @@
import { describe, expect, test } from "bun:test"
import {
APIError,
BadRequest,
ConnectionError,
ContextOverflow,
@ -70,8 +71,9 @@ import { asc, eq } from "drizzle-orm"
import { testEffect } from "./lib/effect"
const requests: LLMRequest[] = []
type ScriptedResponse = LLMEvent[] | Stream.Stream<LLMEvent, LLMError>
let response: LLMEvent[] = []
let responses: LLMEvent[][] | undefined
let responses: ScriptedResponse[] | undefined
let responseStream: Stream.Stream<LLMEvent, LLMError> | undefined
let responseStreams: Stream.Stream<LLMEvent, LLMError>[] | undefined
let streamGate: Deferred.Deferred<void> | undefined
@ -94,9 +96,12 @@ const client = Layer.succeed(
responseStream = undefined
return stream
}
const scripted = responses === undefined ? response : (responses.shift() ?? [])
const events = streamFailure
? Stream.fail(streamFailure)
: Stream.fromIterable(responses === undefined ? response : (responses.shift() ?? []))
: Array.isArray(scripted)
? Stream.fromIterable(scripted)
: scripted
if (!streamGate) return events
return Stream.unwrap(
(streamStarted ? Deferred.succeed(streamStarted, undefined) : Effect.void).pipe(
@ -486,6 +491,11 @@ const setup = Effect.gen(function* () {
const providerUnavailable = () => new ConnectionError({ message: "Provider unavailable" })
const contextOverflow = () => new ContextOverflow({ message: "prompt too long" })
const failingResponse = (events: LLMEvent[], failure: LLMError): Stream.Stream<LLMEvent, LLMError> =>
Stream.fromIterable(events).pipe(Stream.concat(Stream.fail(failure)))
const invalidRequest = () => new BadRequest({ message: "Invalid request" })
const rateLimited = (retryAfterMs?: number) => new RateLimit({ message: "Rate limited", retryAfterMs })
@ -1744,14 +1754,14 @@ describe("SessionRunnerLLM", () => {
yield* admit(session, "Earlier question")
yield* session.resume(sessionID)
response = [LLMEvent.providerError({ message: "summary unavailable" })]
responseStream = Stream.fail(new APIError({ message: "summary unavailable" }))
const compaction = yield* session.compact({ sessionID })
yield* session.resume(sessionID)
expect((yield* session.messages({ sessionID })).find((message) => message.id === compaction.id)).toMatchObject({
type: "compaction",
status: "failed",
error: { type: "provider.error", message: "summary unavailable" },
error: { type: "provider.unknown", message: "summary unavailable" },
})
}),
)
@ -1861,7 +1871,7 @@ describe("SessionRunnerLLM", () => {
currentModel = compactModel
requests.length = 0
responses = [
[LLMEvent.providerError({ message: "Unsupported parameter: max_output_tokens" })],
Stream.fail(new BadRequest({ message: "Unsupported parameter: max_output_tokens" })),
reply.text("Must not run", "text-after-failed-compaction"),
]
yield* admit(session, "Recent exact request ".repeat(180))
@ -1884,10 +1894,7 @@ describe("SessionRunnerLLM", () => {
Effect.gen(function* () {
const session = yield* setupOverflowRecovery
responses = [
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" }),
],
failingResponse([LLMEvent.stepStart({ index: 0 })], contextOverflow()),
reply.text("## Objective\n- Recover overflow", "text-summary"),
reply.text("Recovered", "text-final"),
]
@ -1914,7 +1921,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setupOverflowRecovery
currentModel = model
responses = [
[LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" })],
Stream.fail(contextOverflow()),
reply.text("## Objective\n- Recover unknown limit", "text-summary-unknown-limit"),
reply.text("Recovered", "text-final-unknown-limit"),
]
@ -1934,7 +1941,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setupOverflowRecovery
currentModel = undersizedContextModel
responses = [
[LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" })],
Stream.fail(contextOverflow()),
reply.text("## Objective\n- Recover undersized limit", "text-summary-undersized-limit"),
reply.text("Recovered", "text-final-undersized-limit"),
]
@ -1952,10 +1959,7 @@ describe("SessionRunnerLLM", () => {
it.effect("persists a second context overflow after one recovery", () =>
Effect.gen(function* () {
const session = yield* setupOverflowRecovery
const overflow = () => [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" }),
]
const overflow = () => failingResponse([LLMEvent.stepStart({ index: 0 })], contextOverflow())
responses = [overflow(), reply.text("## Objective\n- Recover once", "text-summary"), overflow()]
yield* admit(session, "Continue")
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("prompt too long")
@ -1971,7 +1975,7 @@ describe("SessionRunnerLLM", () => {
it.effect("recovers once from a raw context overflow failure", () =>
Effect.gen(function* () {
const session = yield* setupOverflowRecovery
responseStream = Stream.fail(new ContextOverflow({ message: "prompt too long" }))
responseStream = Stream.fail(contextOverflow())
responses = [
reply.text("## Objective\n- Recover raw overflow", "text-summary"),
reply.text("Recovered", "text-final"),
@ -1990,10 +1994,7 @@ describe("SessionRunnerLLM", () => {
it.effect("publishes the original overflow when recovery summarization fails", () =>
Effect.gen(function* () {
const session = yield* setupOverflowRecovery
responses = [
[LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" })],
[LLMEvent.providerError({ message: "summary unavailable" })],
]
responses = [Stream.fail(contextOverflow()), Stream.fail(new APIError({ message: "summary unavailable" }))]
yield* admit(session, "Continue")
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("prompt too long")
@ -2004,7 +2005,7 @@ describe("SessionRunnerLLM", () => {
type: "compaction",
status: "failed",
reason: "auto",
error: { type: "provider.error", message: "summary unavailable" },
error: { type: "provider.unknown", message: "summary unavailable" },
}),
)
expect(context.slice(-3)).toMatchObject([
@ -2018,10 +2019,7 @@ describe("SessionRunnerLLM", () => {
it.effect("interrupts overflow recovery while the summary provider is running", () =>
Effect.gen(function* () {
const session = yield* setupOverflowRecovery
responses = [
[LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" })],
reply.text("## Objective\n- Interrupted", "text-summary"),
]
responses = [Stream.fail(contextOverflow()), reply.text("## Objective\n- Interrupted", "text-summary")]
const firstGate = yield* Deferred.make<void>()
const summaryGate = yield* Deferred.make<void>()
streamGate = firstGate
@ -3604,7 +3602,10 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
yield* admit(session, "Fail durably")
response = [LLMEvent.stepStart({ index: 0 }), LLMEvent.providerError({ message: "Provider unavailable" })]
responseStream = failingResponse(
[LLMEvent.stepStart({ index: 0 })],
new APIError({ message: "Provider unavailable" }),
)
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("Provider unavailable")
@ -3621,7 +3622,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
yield* admit(session, "Fail before step")
response = [LLMEvent.providerError({ message: "Provider unavailable" })]
responseStream = Stream.fail(new APIError({ message: "Provider unavailable" }))
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("Provider unavailable")
@ -3709,13 +3710,15 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
yield* admit(session, "Fail after output")
response = [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.textStart({ id: "text-partial" }),
LLMEvent.textDelta({ id: "text-partial", text: "Partial" }),
LLMEvent.textEnd({ id: "text-partial" }),
LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" }),
]
responseStream = failingResponse(
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.textStart({ id: "text-partial" }),
LLMEvent.textDelta({ id: "text-partial", text: "Partial" }),
LLMEvent.textEnd({ id: "text-partial" }),
],
contextOverflow(),
)
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("prompt too long")
expect(requests).toHaveLength(1)
@ -3908,11 +3911,13 @@ describe("SessionRunnerLLM", () => {
toolExecutionGate = yield* Deferred.make<void>()
toolExecutionsStarted = yield* Deferred.make<void>()
toolExecutionsReady = 1
response = [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-before-provider-error", name: "echo", input: { text: "settled" } }),
LLMEvent.providerError({ message: "Provider unavailable" }),
]
responseStream = failingResponse(
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-before-provider-error", name: "echo", input: { text: "settled" } }),
],
new APIError({ message: "Provider unavailable" }),
)
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(toolExecutionsStarted)
@ -3939,11 +3944,10 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
yield* admit(session, "Fail hosted tool durably")
response = [
LLMEvent.stepStart({ index: 0 }),
hostedCall("call-hosted-provider-error", "effect"),
LLMEvent.providerError({ message: "Provider unavailable" }),
]
responseStream = failingResponse(
[LLMEvent.stepStart({ index: 0 }), hostedCall("call-hosted-provider-error", "effect")],
new APIError({ message: "Provider unavailable" }),
)
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("Provider unavailable")
@ -3970,11 +3974,13 @@ describe("SessionRunnerLLM", () => {
Effect.gen(function* () {
const session = yield* setup
yield* admit(session, "Defect while provider fails")
response = [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-defect-provider-error", name: "defect", input: {} }),
LLMEvent.providerError({ message: "Provider unavailable" }),
]
responseStream = failingResponse(
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-defect-provider-error", name: "defect", input: {} }),
],
new APIError({ message: "Provider unavailable" }),
)
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("Provider unavailable")