perf(core): keep text deltas ephemeral

kitlangton · Jun 4, 2026 · Jun 2, 2026 · Jun 2, 2026 · Jun 2, 2026 · Jun 2, 2026
commit befbbc3627e991fae5023853146c08c788c161fd
diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts
@@ -256,10 +256,13 @@ export const layer = Layer.effect(
 
     function publishEvent<D extends Definition>(event: Payload<D>) {
       return Effect.gen(function* () {
-        for (const sync of syncHandlers) {
-          yield* sync(event as Payload)
+        const durable = registry.get(event.type)?.sync !== undefined
+        if (durable) {
+          for (const sync of syncHandlers) {
+            yield* sync(event as Payload)
+          }
+          yield* commitSyncEvent(event as Payload)
         }
-        yield* commitSyncEvent(event as Payload)
         for (const listener of listeners) {
           yield* listener(event as Payload)
         }

diff --git a/packages/core/src/session/event.ts b/packages/core/src/session/event.ts
@@ -159,9 +159,9 @@ export namespace Text {
   })
   export type Started = typeof Started.Type
 
+  // Stream fragments are live-only; Text.Ended is the replayable full-value boundary.
   export const Delta = EventV2.define({
     type: "session.next.text.delta",
-    ...options,
     schema: {
       ...Base,
       textID: Schema.String,

diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts
@@ -332,7 +332,6 @@ export const layer = Layer.effectDiscard(
     yield* events.project(SessionEvent.Step.Ended, (event) => run(db, event))
     yield* events.project(SessionEvent.Step.Failed, (event) => run(db, event))
     yield* events.project(SessionEvent.Text.Started, (event) => run(db, event))
-    yield* events.project(SessionEvent.Text.Delta, (event) => run(db, event))
     yield* events.project(SessionEvent.Text.Ended, (event) => run(db, event))
     yield* events.project(SessionEvent.Tool.Input.Started, (event) => run(db, event))
     yield* events.project(SessionEvent.Tool.Input.Delta, (event) => run(db, event))

diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts
@@ -100,7 +100,7 @@ export const layer = Layer.effect(
             let settledLocalTool = false
             const context = yield* getContext(session.id)
             const request = LLM.request({ model, messages: toLLMMessages(context), tools: yield* tools.definitions() })
-            const publishLLMEvent = createLLMEventPublisher(events, {
+            const publisher = createLLMEventPublisher(events, {
               sessionID: session.id,
               agent: session.agent ?? "build",
               model: {
@@ -110,10 +110,10 @@ export const layer = Layer.effect(
               },
             })
             const publication = Semaphore.makeUnsafe(1)
-            const publish = (event: LLMEvent) => publication.withPermit(publishLLMEvent(event))
+            const publish = (event: LLMEvent) => publication.withPermit(publisher.publish(event))
 
-            yield* llm.stream(request).pipe(
-              Stream.runForEach((event) =>
+             yield* llm.stream(request).pipe(
+               Stream.runForEach((event) =>
                 Effect.gen(function* () {
                    yield* publish(event)
                    if (event.type !== "tool-call" || event.providerExecuted) return
@@ -127,6 +127,7 @@ export const layer = Layer.effect(
                      )
                  }),
                ),
+               Effect.ensuring(publication.withPermit(publisher.flushText())),
              )
              yield* FiberSet.awaitEmpty(settlements)
              return settledLocalTool

diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts
@@ -64,7 +64,7 @@ const output = (result: ToolResultValue): ToolOutput => {
 
 /** Persist one provider turn without executing tools or starting a continuation turn. */
 export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) => {
-  const text = new Map<string, string>()
+  const text = new Map<string, string[]>()
   const reasoning = new Map<string, string>()
   const tools = new Map<
     string,
@@ -78,6 +78,22 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
       ? Effect.die("Tool event before assistant step start")
       : Effect.succeed(assistantMessageID)
 
+  const endText = Effect.fnUntraced(function* (textID: string) {
+    const chunks = text.get(textID)
+    if (!chunks) return yield* Effect.die(`Text end before start: ${textID}`)
+    yield* events.publish(SessionEvent.Text.Ended, {
+      sessionID: input.sessionID,
+      timestamp: yield* timestamp,
+      textID,
+      text: chunks.join(""),
+    })
+    text.delete(textID)
+  })
+
+  const flushText = Effect.fn("SessionRunner.flushText")(function* () {
+    for (const textID of text.keys()) yield* endText(textID)
+  })
+
   const startToolInput = Effect.fnUntraced(function* (event: { readonly id: string; readonly name: string }) {
     if (tools.has(event.id)) return yield* Effect.die(`Duplicate tool input start: ${event.id}`)
     const assistantMessageID = yield* currentAssistantMessageID()
@@ -106,37 +122,32 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
     })
   })
 
-  return Effect.fn("SessionRunner.publishLLMEvent")(function* (event: LLMEvent) {
+  const publish = Effect.fn("SessionRunner.publishLLMEvent")(function* (event: LLMEvent) {
     switch (event.type) {
       case "step-start":
         assistantMessageID = (yield* events.publish(SessionEvent.Step.Started, { ...input, timestamp: yield* timestamp })).id
         return
       case "text-start":
-        text.set(event.id, "")
+        if (text.has(event.id)) return yield* Effect.die(`Duplicate text start: ${event.id}`)
+        text.set(event.id, [])
         yield* events.publish(SessionEvent.Text.Started, { sessionID: input.sessionID, timestamp: yield* timestamp, textID: event.id })
         return
       case "text-delta":
-        if (!text.has(event.id)) return yield* Effect.die(`Text delta before start: ${event.id}`)
-        text.set(event.id, `${text.get(event.id)}${event.text}`)
+        {
+          const chunks = text.get(event.id)
+          if (!chunks) return yield* Effect.die(`Text delta before start: ${event.id}`)
+          chunks.push(event.text)
+        }
         yield* events.publish(SessionEvent.Text.Delta, {
           sessionID: input.sessionID,
           timestamp: yield* timestamp,
           textID: event.id,
           delta: event.text,
         })
         return
-      case "text-end": {
-        const value = text.get(event.id)
-        if (value === undefined) return yield* Effect.die(`Text end before start: ${event.id}`)
-        text.delete(event.id)
-        yield* events.publish(SessionEvent.Text.Ended, {
-          sessionID: input.sessionID,
-          timestamp: yield* timestamp,
-          textID: event.id,
-          text: value,
-        })
+      case "text-end":
+        yield* endText(event.id)
         return
-      }
       case "reasoning-start":
         if (reasoning.has(event.id)) return yield* Effect.die(`Duplicate reasoning start: ${event.id}`)
         reasoning.set(event.id, "")
@@ -266,6 +277,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
         return
       }
       case "step-finish":
+        yield* flushText()
         yield* events.publish(SessionEvent.Step.Ended, {
           sessionID: input.sessionID,
           timestamp: yield* timestamp,
@@ -277,6 +289,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
       case "finish":
         return
       case "provider-error":
+        yield* flushText()
         yield* events.publish(SessionEvent.Step.Failed, {
           sessionID: input.sessionID,
           timestamp: yield* timestamp,
@@ -285,4 +298,6 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
         return
     }
   })
+
+  return { publish, flushText }
 }
diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts
@@ -218,6 +218,24 @@ describe("EventV2", () => {
     }),
   )
 
+  it.effect("does not synchronize live-only events", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const synchronized = new Array<string>()
+      const unsubscribe = yield* events.sync((event) =>
+        Effect.sync(() => {
+          synchronized.push(event.type)
+        }),
+      )
+      yield* Effect.addFinalizer(() => unsubscribe)
+
+      yield* events.publish(Message, { text: "live only" })
+      yield* events.publish(SyncMessage, { id: "one", text: "durable" })
+
+      expect(synchronized).toEqual([SyncMessage.type])
+    }),
+  )
+
   it.effect("inserts sync event rows on publish", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service

diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts
@@ -91,8 +91,6 @@ describe("SessionRunnerLLM recorded", () => {
         "session.next.prompted.1",
         "session.next.step.started.1",
         "session.next.text.started.1",
-        "session.next.text.delta.1",
-        "session.next.text.delta.1",
         "session.next.text.ended.1",
         "session.next.step.ended.1",
       ])

diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts
@@ -8,6 +8,7 @@ import { Project } from "@opencode-ai/core/project"
 import { ProjectTable } from "@opencode-ai/core/project/sql"
 import { AbsolutePath } from "@opencode-ai/core/schema"
 import { SessionV2 } from "@opencode-ai/core/session"
+import { SessionEvent } from "@opencode-ai/core/session/event"
 import { Prompt } from "@opencode-ai/core/session/prompt"
 import { SessionProjector } from "@opencode-ai/core/session/projector"
 import { SessionExecution } from "@opencode-ai/core/session/execution"
@@ -749,6 +750,153 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("broadcasts provider text deltas without storing projection rewrites", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Stream many chunks" }), resume: false })
+      const live: SessionEvent.Text.Delta[] = []
+      const unsubscribe = yield* EventV2.Service.pipe(
+        Effect.flatMap((events) => events.listen((event) => Effect.sync(() => {
+          if (event.type === SessionEvent.Text.Delta.type) live.push(event as SessionEvent.Text.Delta)
+        }))),
+      )
+      yield* Effect.addFinalizer(() => unsubscribe)
+
+      responses = undefined
+      streamGate = undefined
+      streamStarted = undefined
+      response = [
+        LLMEvent.stepStart({ index: 0 }),
+        LLMEvent.textStart({ id: "text-many" }),
+        ...Array.from({ length: 32 }, (_, index) => LLMEvent.textDelta({ id: "text-many", text: `${index},` })),
+        LLMEvent.textEnd({ id: "text-many" }),
+        LLMEvent.stepFinish({ index: 0, reason: "stop" }),
+        LLMEvent.finish({ reason: "stop" }),
+      ]
+
+      yield* session.resume(sessionID)
+
+      const { db } = yield* Database.Service
+      const events = yield* EventV2.Service
+      const deltas = yield* db
+        .select({ type: EventTable.type })
+        .from(EventTable)
+        .where(eq(EventTable.type, EventV2.versionedType(SessionEvent.Text.Delta.type, 1)))
+        .all()
+        .pipe(Effect.orDie)
+      expect(live).toHaveLength(32)
+      expect(deltas).toHaveLength(0)
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user", text: "Stream many chunks" },
+        {
+          type: "assistant",
+          finish: "stop",
+          content: [{ type: "text", id: "text-many", text: Array.from({ length: 32 }, (_, index) => `${index},`).join("") }],
+        },
+      ])
+      const recorded = yield* db
+        .select()
+        .from(EventTable)
+        .where(eq(EventTable.aggregate_id, sessionID))
+        .orderBy(asc(EventTable.seq))
+        .all()
+        .pipe(Effect.orDie)
+      yield* events.remove(sessionID)
+      yield* db.delete(SessionMessageTable).where(eq(SessionMessageTable.session_id, sessionID)).run().pipe(Effect.orDie)
+      yield* events.replayAll(
+        recorded.map((event) => ({
+          id: event.id,
+          aggregateID: event.aggregate_id,
+          seq: event.seq,
+          type: event.type,
+          data: event.data,
+        })),
+      )
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user", text: "Stream many chunks" },
+        {
+          type: "assistant",
+          finish: "stop",
+          content: [{ type: "text", id: "text-many", text: Array.from({ length: 32 }, (_, index) => `${index},`).join("") }],
+        },
+      ])
+    }),
+  )
+
+  it.effect("durably closes partial text when the provider stream fails", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Fail after text" }), resume: false })
+      const failure = new LLMError({
+        module: "test",
+        method: "stream",
+        reason: new TransportReason({ message: "Provider unavailable" }),
+      })
+
+      responses = undefined
+      streamGate = undefined
+      streamStarted = undefined
+      responseStream = Stream.concat(
+        Stream.fromIterable([
+          LLMEvent.stepStart({ index: 0 }),
+          LLMEvent.textStart({ id: "text-partial" }),
+          LLMEvent.textDelta({ id: "text-partial", text: "Partial" }),
+        ]),
+        Stream.fail(failure),
+      )
+
+      expect(yield* session.resume(sessionID).pipe(Effect.flip)).toBe(failure)
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user", text: "Fail after text" },
+        { type: "assistant", content: [{ type: "text", id: "text-partial", text: "Partial" }] },
+      ])
+    }),
+  )
+
+  it.effect("durably closes partial text when the provider stream is interrupted", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Interrupt after text" }), resume: false })
+      const streamed = yield* Deferred.make<void>()
+
+      responses = undefined
+      streamGate = undefined
+      streamStarted = undefined
+      responseStream = Stream.concat(
+        Stream.fromIterable([
+          LLMEvent.stepStart({ index: 0 }),
+          LLMEvent.textStart({ id: "text-interrupted" }),
+          LLMEvent.textDelta({ id: "text-interrupted", text: "Partial" }),
+        ]),
+        Stream.fromEffect(Deferred.succeed(streamed, undefined)).pipe(Stream.flatMap(() => Stream.never)),
+      )
+
+      const fiber = yield* session.resume(sessionID).pipe(Effect.forkChild)
+      yield* Deferred.await(streamed)
+      yield* Fiber.interrupt(fiber)
+      expect(yield* session.context(sessionID)).toMatchObject([
+        { type: "user", text: "Interrupt after text" },
+        { type: "assistant", content: [{ type: "text", id: "text-interrupted", text: "Partial" }] },
+      ])
+    }),
+  )
+
+  it.effect("rejects duplicate streamed text starts", () =>
+    Effect.gen(function* () {
+      yield* setup
+      const session = yield* SessionV2.Service
+      responses = undefined
+      streamGate = undefined
+      streamStarted = undefined
+      response = [LLMEvent.textStart({ id: "text-1" }), LLMEvent.textStart({ id: "text-1" })]
+
+      expect(yield* session.resume(sessionID).pipe(Effect.catchDefect(Effect.succeed))).toBe("Duplicate text start: text-1")
+    }),
+  )
+
   it.effect("rejects malformed streamed tool input ordering", () =>
     Effect.gen(function* () {
       yield* setup