refactor: address PR review — remove sync Effect wrappers, add Stream pagination

- Remove partsEffect/getEffect/pageEffect; call sync functions directly
- Convert streamMessages to lazy Stream.paginate (chunk-efficient)
- Add filterCompactedMessages using Stream.takeUntil on the stream
- Make toModelMessagesEffect the primary impl, derive promise version

Author: kitlangton

packages/opencode/src/session/index.ts

@@ -33,7 +33,7 @@ import { ModelID, ProviderID } from "@/provider/schema"
 import { Permission } from "@/permission"
 import { Global } from "@/global"
 import type { LanguageModelV2Usage } from "@ai-sdk/provider"
-import { Effect, Layer, Scope, ServiceMap } from "effect"
+import { Effect, Layer, Scope, ServiceMap, Stream } from "effect"
 import { makeRuntime } from "@/effect/run-service"
 
 export namespace Session {
@@ -594,12 +594,10 @@ export namespace Session {
 
       const messages = Effect.fn("Session.messages")(function* (input: { sessionID: SessionID; limit?: number }) {
         if (input.limit) {
-          const result = yield* MessageV2.pageEffect({ sessionID: input.sessionID, limit: input.limit })
-          return result.items
+          return MessageV2.pageSync({ sessionID: input.sessionID, limit: input.limit }).items
         }
-        const all = yield* MessageV2.streamEffect(input.sessionID)
-        all.reverse()
-        return all
+        const all = yield* Stream.runCollect(MessageV2.streamMessages(input.sessionID))
+        return Array.from(all).reverse()
       })
 
       const removeMessage = Effect.fn("Session.removeMessage")(function* (input: {

packages/opencode/src/session/message-v2.ts

@@ -15,7 +15,7 @@ import { errorMessage } from "@/util/error"
 import type { SystemError } from "bun"
 import type { Provider } from "@/provider/provider"
 import { ModelID, ProviderID } from "@/provider/schema"
-import { Effect } from "effect"
+import { Effect, Option, Stream } from "effect"
 
 /** Error shape thrown by Bun's fetch() when gzip/br decompression fails mid-stream */
 interface FetchDecompressionError extends Error {
@@ -574,11 +574,11 @@ export namespace MessageV2 {
     }))
   }
 
-  export async function toModelMessages(
+  export const toModelMessagesEffect = Effect.fnUntraced(function* (
     input: WithParts[],
     model: Provider.Model,
     options?: { stripMedia?: boolean },
-  ): Promise<ModelMessage[]> {
+  ) {
     const result: UIMessage[] = []
     const toolNames = new Set<string>()
     // Track media from tool results that need to be injected as user messages
@@ -801,24 +801,26 @@ export namespace MessageV2 {
 
     const tools = Object.fromEntries(Array.from(toolNames).map((toolName) => [toolName, { toModelOutput }]))
 
-    return await convertToModelMessages(
-      result.filter((msg) => msg.parts.some((part) => part.type !== "step-start")),
-      {
-        //@ts-expect-error (convertToModelMessages expects a ToolSet but only actually needs tools[name]?.toModelOutput)
-        tools,
-      },
+    return yield* Effect.promise(() =>
+      convertToModelMessages(
+        result.filter((msg) => msg.parts.some((part) => part.type !== "step-start")),
+        {
+          //@ts-expect-error (convertToModelMessages expects a ToolSet but only actually needs tools[name]?.toModelOutput)
+          tools,
+        },
+      ),
     )
-  }
+  })
 
-  export const toModelMessagesEffect = Effect.fnUntraced(function* (
+  export function toModelMessages(
     input: WithParts[],
     model: Provider.Model,
     options?: { stripMedia?: boolean },
-  ) {
-    return yield* Effect.promise(() => toModelMessages(input, model, options))
-  })
+  ): Promise<ModelMessage[]> {
+    return Effect.runPromise(toModelMessagesEffect(input, model, options))
+  }
 
-  function pageSync(input: { sessionID: SessionID; limit: number; before?: string }) {
+  export function pageSync(input: { sessionID: SessionID; limit: number; before?: string }) {
     const before = input.before ? cursor.decode(input.before) : undefined
     const where = before
       ? and(eq(MessageTable.session_id, input.sessionID), older(before))
@@ -878,7 +880,7 @@ export namespace MessageV2 {
     }
   })
 
-  function partsSync(message_id: MessageID) {
+  export function partsSync(message_id: MessageID) {
     const rows = Database.use((db) =>
       db.select().from(PartTable).where(eq(PartTable.message_id, message_id)).orderBy(PartTable.id).all(),
     )
@@ -895,7 +897,7 @@ export namespace MessageV2 {
 
   export const parts = fn(MessageID.zod, async (message_id) => partsSync(message_id))
 
-  function getSync(input: { sessionID: SessionID; messageID: MessageID }): WithParts {
+  export function getSync(input: { sessionID: SessionID; messageID: MessageID }): WithParts {
     const row = Database.use((db) =>
       db
         .select()
@@ -918,54 +920,30 @@ export namespace MessageV2 {
     async (input): Promise<WithParts> => getSync(input),
   )
 
-  export const partsEffect = Effect.fnUntraced(function* (id: MessageID) {
-    return partsSync(id)
-  })
-
-  export const getEffect = Effect.fnUntraced(function* (input: { sessionID: SessionID; messageID: MessageID }) {
-    return getSync(input)
-  })
-
-  export const pageEffect = Effect.fnUntraced(function* (input: {
-    sessionID: SessionID
-    limit: number
-    before?: string
-  }) {
-    return pageSync(input)
-  })
 
-  export const streamEffect = Effect.fnUntraced(function* (sessionID: SessionID) {
-    const result: WithParts[] = []
-    const size = 50
-    let before: string | undefined
-    while (true) {
-      const next = pageSync({ sessionID, limit: size, before })
-      if (next.items.length === 0) break
-      for (let i = next.items.length - 1; i >= 0; i--) {
-        result.push(next.items[i])
-      }
-      if (!next.more || !next.cursor) break
-      before = next.cursor
-    }
-    return result
-  })
+  export const streamMessages = (sessionID: SessionID): Stream.Stream<WithParts> =>
+    Stream.paginate(undefined as string | undefined, (before) =>
+      Effect.sync(() => {
+        const next = pageSync({ sessionID, limit: 50, before })
+        const items = next.items.toReversed()
+        const nextCursor = next.more && next.cursor ? Option.some(next.cursor) : Option.none()
+        return [items, nextCursor] as const
+      }),
+    )
 
-  function applyCompactionFilter(msgs: MessageV2.WithParts[]) {
-    const result = [] as MessageV2.WithParts[]
+  export const filterCompactedMessages = (sessionID: SessionID): Stream.Stream<WithParts> => {
     const completed = new Set<string>()
-    for (const msg of msgs) {
-      result.push(msg)
-      if (
-        msg.info.role === "user" &&
-        completed.has(msg.info.id) &&
-        msg.parts.some((part) => part.type === "compaction")
-      )
-        break
-      if (msg.info.role === "assistant" && msg.info.summary && msg.info.finish && !msg.info.error)
-        completed.add(msg.info.parentID)
-    }
-    result.reverse()
-    return result
+    return streamMessages(sessionID).pipe(
+      Stream.takeUntil((msg) => {
+        if (msg.info.role === "assistant" && msg.info.summary && msg.info.finish && !msg.info.error)
+          completed.add(msg.info.parentID)
+        return (
+          msg.info.role === "user" &&
+          completed.has(msg.info.id) &&
+          msg.parts.some((part) => part.type === "compaction")
+        )
+      }),
+    )
   }
 
   export async function filterCompacted(stream: AsyncIterable<MessageV2.WithParts>) {
@@ -987,7 +965,8 @@ export namespace MessageV2 {
   }
 
   export const filterCompactedEffect = Effect.fnUntraced(function* (sessionID: SessionID) {
-    return applyCompactionFilter(yield* streamEffect(sessionID))
+    const all = yield* Stream.runCollect(filterCompactedMessages(sessionID))
+    return Array.from(all).reverse()
   })
 
   export function fromError(

packages/opencode/src/session/processor.ts

@@ -180,7 +180,7 @@ export namespace SessionProcessor {
                 metadata: value.providerMetadata,
               } satisfies MessageV2.ToolPart)
 
-              const parts = yield* MessageV2.partsEffect(ctx.assistantMessage.id)
+              const parts = MessageV2.partsSync(ctx.assistantMessage.id)
               const recentParts = parts.slice(-DOOM_LOOP_THRESHOLD)
 
               if (
@@ -392,7 +392,7 @@ export namespace SessionProcessor {
           }
           ctx.reasoningMap = {}
 
-          const parts = yield* MessageV2.partsEffect(ctx.assistantMessage.id)
+          const parts = MessageV2.partsSync(ctx.assistantMessage.id)
           for (const part of parts) {
             if (part.type !== "tool" || part.state.status === "completed" || part.state.status === "error") continue
             yield* session.updatePart({

packages/opencode/test/session/processor-effect.test.ts

@@ -207,7 +207,7 @@ it.live("session.processor effect tests capture llm input cleanly", () =>
         } satisfies LLM.StreamInput
 
         const value = yield* handle.process(input)
-        const parts = yield* MessageV2.partsEffect(msg.id)
+        const parts = MessageV2.partsSync(msg.id)
         const calls = yield* llm.calls
 
         expect(value).toBe("continue")
@@ -254,7 +254,7 @@ it.live("session.processor effect tests stop after token overflow requests compa
           tools: {},
         })
 
-        const parts = yield* MessageV2.partsEffect(msg.id)
+        const parts = MessageV2.partsSync(msg.id)
 
         expect(value).toBe("compact")
         expect(parts.some((part) => part.type === "text" && part.text === "after")).toBe(true)
@@ -299,7 +299,7 @@ it.live("session.processor effect tests capture reasoning from http mock", () =>
           tools: {},
         })
 
-        const parts = yield* Effect.promise(() => MessageV2.parts(msg.id))
+        const parts = MessageV2.partsSync(msg.id)
         const reasoning = parts.find((part): part is MessageV2.ReasoningPart => part.type === "reasoning")
         const text = parts.find((part): part is MessageV2.TextPart => part.type === "text")
 
@@ -347,7 +347,7 @@ it.live("session.processor effect tests reset reasoning state across retries", (
           tools: {},
         })
 
-        const parts = yield* MessageV2.partsEffect(msg.id)
+        const parts = MessageV2.partsSync(msg.id)
         const reasoning = parts.filter((part): part is MessageV2.ReasoningPart => part.type === "reasoning")
 
         expect(value).toBe("continue")
@@ -438,7 +438,7 @@ it.live("session.processor effect tests retry recognized structured json errors"
           tools: {},
         })
 
-        const parts = yield* MessageV2.partsEffect(msg.id)
+        const parts = MessageV2.partsSync(msg.id)
 
         expect(value).toBe("continue")
         expect(yield* llm.calls).toBe(2)
@@ -596,7 +596,7 @@ it.live("session.processor effect tests mark pending tools as aborted on cleanup
         if (Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)) {
           yield* handle.abort()
         }
-        const parts = yield* MessageV2.partsEffect(msg.id)
+        const parts = MessageV2.partsSync(msg.id)
         const call = parts.find((part): part is MessageV2.ToolPart => part.type === "tool")
 
         expect(Exit.isFailure(exit)).toBe(true)
@@ -669,7 +669,7 @@ it.live("session.processor effect tests record aborted errors and idle state", (
           yield* handle.abort()
         }
         yield* Effect.promise(() => seen.promise)
-        const stored = yield* MessageV2.getEffect({ sessionID: chat.id, messageID: msg.id })
+        const stored = MessageV2.getSync({ sessionID: chat.id, messageID: msg.id })
         const state = yield* sts.get(chat.id)
         off()
 
@@ -731,7 +731,7 @@ it.live("session.processor effect tests mark interruptions aborted without manua
         yield* Fiber.interrupt(run)
 
         const exit = yield* Fiber.await(run)
-        const stored = yield* MessageV2.getEffect({ sessionID: chat.id, messageID: msg.id })
+        const stored = MessageV2.getSync({ sessionID: chat.id, messageID: msg.id })
         const state = yield* sts.get(chat.id)
 
         expect(Exit.isFailure(exit)).toBe(true)