vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-08-31 06:24:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(core): batch initial streamed delta

Browse source
This commit is contained in:
Dax Raad 2026-08-15 19:26:16 -04:00
parent 42e345e1bc
commit 8251934007
3 changed files with 10 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -150,6 +150,10 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
if (!current) return yield* Effect.die(new Error(`${name} delta before start: ${id}`))
if (!current.pending) return undefined
const now = yield* Clock.currentTimeMillis
if (!force && current.publishedAt === undefined) {
current.publishedAt = now
return undefined
}
if (!force && current.publishedAt !== undefined && now - current.publishedAt < deltaBatchInterval)
return undefined
yield* delta(id, current.pending, current.ordinal)

11
packages/core/test/session-runner-tool-events.test.ts
View file

@ -217,16 +217,13 @@ it.effect("batches text deltas and flushes pending text before the terminal even
{ discard: true },
)
expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
{ delta: "one" },
])
expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
yield* TestClock.adjust("99 millis")
expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(1)
expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
yield* TestClock.adjust("1 millis")
yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " four" }))
expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
{ delta: "one" },
{ delta: " two three four" },
{ delta: "one two three four" },
])
yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " five" }))
@ -253,7 +250,7 @@ it.effect("batches reasoning deltas and flushes pending reasoning before the ter
expect(
published.filter((event) => event.type === "session.reasoning.delta").map((event) => event.data),
).toMatchObject([{ delta: "one" }, { delta: " two three" }])
).toMatchObject([{ delta: "one two three" }])
expect(published.slice(-2).map((event) => event.type)).toEqual([
"session.reasoning.delta",
"session.reasoning.ended.1",

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

@ -767,7 +767,7 @@ const verifyEphemeralDeltas = (kind: FragmentKind) =>
yield* admit(session, prompt)
const bus = yield* Bus.Service
const live = fixture.delta
? yield* bus.subscribe(fixture.delta).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
? yield* bus.subscribe(fixture.delta).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
: undefined
yield* Effect.yieldNow
yield* TestLLM.push(fixture.completeEvents)
@ -785,7 +785,7 @@ const verifyEphemeralDeltas = (kind: FragmentKind) =>
: []
if (live) {
const streamed = Array.from(yield* Fiber.join(live))
expect(streamed).toHaveLength(2)
expect(streamed).toHaveLength(1)
expect(
streamed
.map((event) => {