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

feat(llm): enforce terminal stream contract for every route

Browse source 
LLMClient.stream now guarantees, for native protocol routes and the
synthetic AI SDK route alike: a successful stream emits exactly one
terminal finish event and nothing after it. EOF before finish means the
provider stream was truncated (proxy cut, silent drop) and fails as
LLM.MalformedResponse instead of letting a partial response settle as
complete; output after finish also fails. Enforcement applies after
protocol parsing so stream.onHalt flushes remain subject to it, with
per-subscription state.
This commit is contained in:
Aiden Cline 2026-07-13 10:50:49 -05:00
commit 49e86dd8ef
2 changed files with 39 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

34
packages/llm/src/route/client.ts
View file

@ -229,6 +229,35 @@ const streamError = (route: string, message: string, cause: Cause.Cause<unknown>
return ProviderShared.eventError(route, message, Cause.pretty(cause))
}
/**
* Terminal contract for every route, native or synthetic: a successful
* stream emits exactly one `finish`, and nothing after it. EOF before
* `finish` means the provider stream was truncated (proxy cut, silent
* drop) and must fail rather than let a partial response settle as
* complete. Applied after protocol parsing so `stream.onHalt` flushes are
* still subject to it.
*/
const enforceTerminal = (route: string) => (events: Stream.Stream<LLMEvent, LLMError>) => {
let finished = false
return events.pipe(
Stream.mapEffect((event) => {
if (finished)
return Effect.fail(
ProviderShared.eventError(route, `Provider emitted ${event.type} after the terminal finish event`),
)
if (event.type === "finish") finished = true
return Effect.succeed(event)
}),
Stream.concat(
Stream.suspend(() =>
finished
? Stream.empty
: Stream.fail(ProviderShared.eventError(route, "Provider stream ended without a terminal finish event")),
),
),
)
}
function makeFromTransport<Body, Prepared, Frame, Event, State>(
input: MakeTransportInput<Body, Prepared, Frame, Event, State>,
): Route<Body, Prepared> {
@ -383,7 +412,10 @@ const streamRequestWith = (runtime: TransportRuntime) => (request: LLMRequest) =
Stream.unwrap(
Effect.gen(function* () {
const compiled = yield* compile(request)
return compiled.route.streamPrepared(compiled.prepared, compiled.request, runtime)
const route = `${compiled.request.model.provider}/${compiled.route.id}`
return compiled.route
.streamPrepared(compiled.prepared, compiled.request, runtime)
.pipe(enforceTerminal(route))
}),
)

8
packages/llm/test/provider/openai-chat.test.ts
View file

@ -614,8 +614,11 @@ describe("OpenAI Chat route", () => {
const input = LLM.updateRequest(request, {
tools: [{ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } }],
})
const events = Array.from(
yield* LLMClient.stream(input).pipe(Stream.runCollect, Effect.provide(fixedResponse(body))),
const events: LLMEvent[] = []
const streamError = yield* LLMClient.stream(input).pipe(
Stream.runForEach((event) => Effect.sync(() => events.push(event))),
Effect.flip,
Effect.provide(fixedResponse(body)),
)
const error = yield* LLMClient.generate(input).pipe(Effect.provide(fixedResponse(body)), Effect.flip)
@ -626,6 +629,7 @@ describe("OpenAI Chat route", () => {
{ type: "tool-input-delta", id: "call_1", name: "lookup", text: ':"weather"}' },
])
expect(events.filter(LLMEvent.is.toolCall)).toEqual([])
expect(streamError.message).toContain("Provider stream ended without a terminal finish event")
expect(error.message).toContain("Provider stream ended without a terminal finish event")
}),
)