fix(session): avoid full history hydration after compaction

Factor the compaction filter into an incremental scan and have
filterCompactedEffect page backward through messages, stopping as soon as a
completed compaction boundary (and any retained tail) is found, instead of
hydrating the entire session history first.

Ported from anomalyco#31638 (already Effect-based; applied as-is).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: randomvariable

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

@@ -518,30 +518,45 @@ export const get = Effect.fn("MessageV2.get")(function* (input: { sessionID: Ses
   }
 })
 
-export function filterCompacted(msgs: Iterable<WithParts>) {
-  const result = [] as WithParts[]
-  const completed = new Set<string>()
-  let retain: MessageID | undefined
-  for (const msg of msgs) {
-    result.push(msg)
-    if (retain) {
-      if (msg.info.id === retain) break
-      continue
-    }
-    if (msg.info.role === "user" && completed.has(msg.info.id)) {
-      const part = msg.parts.find((item): item is CompactionPart => item.type === "compaction")
-      if (!part) continue
-      if (!part.tail_start_id) break
-      retain = part.tail_start_id
-      if (msg.info.id === retain) break
-      continue
+type CompactedScan = {
+  result: WithParts[]
+  completed: Set<string>
+  retain: MessageID | undefined
+  done: boolean
+}
+
+function compactedScan() {
+  return {
+    result: [] as WithParts[],
+    completed: new Set<string>(),
+    retain: undefined,
+    done: false,
+  } satisfies CompactedScan
+}
+
+function scanCompacted(scan: CompactedScan, msg: WithParts) {
+  scan.result.push(msg)
+  if (scan.retain) {
+    if (msg.info.id === scan.retain) scan.done = true
+    return
+  }
+  if (msg.info.role === "user" && scan.completed.has(msg.info.id)) {
+    const part = msg.parts.find((item): item is CompactionPart => item.type === "compaction")
+    if (!part) return
+    if (!part.tail_start_id) {
+      scan.done = true
+      return
     }
-    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)
+    scan.retain = part.tail_start_id
+    if (msg.info.id === scan.retain) scan.done = true
+    return
   }
-  result.reverse()
+  if (msg.info.role === "assistant" && msg.info.summary && msg.info.finish && !msg.info.error)
+    scan.completed.add(msg.info.parentID)
+}
+
+function finishCompactedScan(scan: CompactedScan) {
+  const result = scan.result.reverse()
   const compactionIndex = result.findLastIndex(
     (msg) =>
       msg.info.role === "user" &&
@@ -571,8 +586,36 @@ export function filterCompacted(msgs: Iterable<WithParts>) {
   return result
 }
 
+export function filterCompacted(msgs: Iterable<WithParts>) {
+  const scan = compactedScan()
+  for (const msg of msgs) {
+    scanCompacted(scan, msg)
+    if (scan.done) break
+  }
+  return finishCompactedScan(scan)
+}
+
 export const filterCompactedEffect = Effect.fnUntraced(function* (sessionID: SessionID) {
-  return filterCompacted(yield* stream(sessionID))
+  const scan = compactedScan()
+  const size = 50
+  let before: string | undefined
+  while (!scan.done) {
+    const next = yield* page({ sessionID, limit: size, before }).pipe(
+      Effect.catchIf(NotFoundError.isInstance, () =>
+        Effect.succeed({ items: [] as WithParts[], more: false, cursor: undefined }),
+      ),
+    )
+    if (next.items.length === 0) break
+    for (let i = next.items.length - 1; i >= 0; i--) {
+      const item = next.items[i]
+      if (!item) continue
+      scanCompacted(scan, item)
+      if (scan.done) break
+    }
+    if (scan.done || !next.more || !next.cursor) break
+    before = next.cursor
+  }
+  return finishCompactedScan(scan)
 })
 
 // filterCompacted reorders messages for model consumption

packages/opencode/test/session/messages-pagination.test.ts

@@ -700,6 +700,190 @@ describe("MessageV2.filterCompacted", () => {
     ),
   )
 
+  it.instance("filterCompactedEffect matches full stream filtering with tail retention", () =>
+    withSession(({ session, sessionID }) =>
+      Effect.gen(function* () {
+        const old = yield* fill(sessionID, 75, (i: number) => Date.now() - 100_000 + i)
+
+        const u1 = yield* addUser(sessionID, "first")
+        const a1 = yield* addAssistant(sessionID, u1, { finish: "end_turn" })
+        yield* session.updatePart({
+          id: PartID.ascending(),
+          sessionID,
+          messageID: a1,
+          type: "text",
+          text: "first reply",
+        })
+
+        const u2 = yield* addUser(sessionID, "second")
+        const a2 = yield* addAssistant(sessionID, u2, { finish: "end_turn" })
+        yield* session.updatePart({
+          id: PartID.ascending(),
+          sessionID,
+          messageID: a2,
+          type: "text",
+          text: "second reply",
+        })
+
+        const c1 = yield* addUser(sessionID)
+        yield* addCompactionPart(sessionID, c1, u2)
+        const s1 = yield* addAssistant(sessionID, c1, { summary: true, finish: "end_turn" })
+        yield* session.updatePart({
+          id: PartID.ascending(),
+          sessionID,
+          messageID: s1,
+          type: "text",
+          text: "summary",
+        })
+
+        const u3 = yield* addUser(sessionID, "third")
+        const a3 = yield* addAssistant(sessionID, u3, { finish: "end_turn" })
+        yield* session.updatePart({
+          id: PartID.ascending(),
+          sessionID,
+          messageID: a3,
+          type: "text",
+          text: "third reply",
+        })
+
+        const full = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
+        const optimized = yield* MessageV2.filterCompactedEffect(sessionID)
+
+        expect(optimized.map((item) => item.info.id)).toEqual(full.map((item) => item.info.id))
+        expect(optimized.map((item) => item.info.id)).toEqual([c1, s1, u2, a2, u3, a3])
+        expect(optimized.some((item) => old.includes(item.info.id))).toBe(false)
+      }),
+    ),
+  )
+
+  it.instance("filterCompactedEffect keeps retained tail across page boundaries", () =>
+    withSession(({ session, sessionID }) =>
+      Effect.gen(function* () {
+        const old = yield* fill(sessionID, 10, (i: number) => Date.now() - 200_000 + i)
+        const tail = yield* fill(sessionID, 60, (i: number) => Date.now() - 100_000 + i)
+        const tailStart = tail[0]
+        if (!tailStart) throw new Error("expected retained tail")
+        const tailEnd = tail[tail.length - 1]
+        if (!tailEnd) throw new Error("expected retained tail end")
+
+        const c1 = yield* addUser(sessionID)
+        yield* addCompactionPart(sessionID, c1, tailStart)
+        const s1 = yield* addAssistant(sessionID, c1, { summary: true, finish: "end_turn" })
+        yield* session.updatePart({
+          id: PartID.ascending(),
+          sessionID,
+          messageID: s1,
+          type: "text",
+          text: "summary",
+        })
+
+        const u1 = yield* addUser(sessionID, "next")
+        const a1 = yield* addAssistant(sessionID, u1, { finish: "end_turn" })
+        yield* session.updatePart({
+          id: PartID.ascending(),
+          sessionID,
+          messageID: a1,
+          type: "text",
+          text: "next reply",
+        })
+
+        const full = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
+        const optimized = yield* MessageV2.filterCompactedEffect(sessionID)
+        const ids = optimized.map((item) => item.info.id)
+
+        expect(ids).toEqual(full.map((item) => item.info.id))
+        expect(ids.slice(0, 2)).toEqual([c1, s1])
+        expect(ids).toContain(tailStart)
+        expect(ids).toContain(tailEnd)
+        expect(ids.slice(-2)).toEqual([u1, a1])
+        expect(ids.some((item) => old.includes(item))).toBe(false)
+      }),
+    ),
+  )
+
+  it.instance("filterCompactedEffect stops at completed compaction without retained tail", () =>
+    withSession(({ session, sessionID }) =>
+      Effect.gen(function* () {
+        const old = yield* fill(sessionID, 75, (i: number) => Date.now() - 100_000 + i)
+
+        const c1 = yield* addUser(sessionID)
+        yield* addCompactionPart(sessionID, c1)
+        const s1 = yield* addAssistant(sessionID, c1, { summary: true, finish: "end_turn" })
+        yield* session.updatePart({
+          id: PartID.ascending(),
+          sessionID,
+          messageID: s1,
+          type: "text",
+          text: "summary",
+        })
+
+        const u1 = yield* addUser(sessionID, "next")
+        const a1 = yield* addAssistant(sessionID, u1, { finish: "end_turn" })
+        yield* session.updatePart({
+          id: PartID.ascending(),
+          sessionID,
+          messageID: a1,
+          type: "text",
+          text: "next reply",
+        })
+
+        const full = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
+        const optimized = yield* MessageV2.filterCompactedEffect(sessionID)
+
+        expect(optimized.map((item) => item.info.id)).toEqual(full.map((item) => item.info.id))
+        expect(optimized.map((item) => item.info.id)).toEqual([c1, s1, u1, a1])
+        expect(optimized.some((item) => old.includes(item.info.id))).toBe(false)
+      }),
+    ),
+  )
+
+  it.instance("filterCompactedEffect keeps full history without compaction", () =>
+    withSession(({ sessionID }) =>
+      Effect.gen(function* () {
+        yield* fill(sessionID, 75, (i: number) => Date.now() - 100_000 + i)
+
+        const full = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
+        const optimized = yield* MessageV2.filterCompactedEffect(sessionID)
+
+        expect(optimized.map((item) => item.info.id)).toEqual(full.map((item) => item.info.id))
+        expect(optimized).toHaveLength(full.length)
+      }),
+    ),
+  )
+
+  it.instance("filterCompactedEffect keeps full history without completed compaction", () =>
+    withSession(({ sessionID }) =>
+      Effect.gen(function* () {
+        const old = yield* fill(sessionID, 75, (i: number) => Date.now() - 100_000 + i)
+
+        const missing = yield* addUser(sessionID)
+        yield* addCompactionPart(sessionID, missing)
+
+        const errorParent = yield* addUser(sessionID)
+        yield* addCompactionPart(sessionID, errorParent)
+        const error = new SessionV1.APIError({
+          message: "boom",
+          isRetryable: true,
+        }).toObject() as SessionV1.Assistant["error"]
+        yield* addAssistant(sessionID, errorParent, { summary: true, finish: "end_turn", error })
+
+        const unfinished = yield* addUser(sessionID)
+        yield* addCompactionPart(sessionID, unfinished)
+        yield* addAssistant(sessionID, unfinished, { summary: true })
+
+        const full = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
+        const optimized = yield* MessageV2.filterCompactedEffect(sessionID)
+        const ids = optimized.map((item) => item.info.id)
+
+        expect(ids).toEqual(full.map((item) => item.info.id))
+        expect(ids).toContain(missing)
+        expect(ids).toContain(errorParent)
+        expect(ids).toContain(unfinished)
+        expect(ids).toContain(old[0])
+      }),
+    ),
+  )
+
   it.instance("retains original tail when compaction stores tail_start_id", () =>
     withSession(({ session, sessionID }) =>
       Effect.gen(function* () {
@@ -745,8 +929,10 @@ describe("MessageV2.filterCompacted", () => {
         })
 
         const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
+        const optimized = yield* MessageV2.filterCompactedEffect(sessionID)
 
         expect(result.map((item) => item.info.id)).toEqual([c1, s1, u2, a2, u3, a3])
+        expect(optimized.map((item) => item.info.id)).toEqual(result.map((item) => item.info.id))
       }),
     ),
   )
@@ -802,7 +988,9 @@ describe("MessageV2.filterCompacted", () => {
 
       const forked = yield* session.fork({ sessionID: created.id })
       const childFiltered = MessageV2.filterCompacted(yield* MessageV2.stream(forked.id))
+      const childOptimized = yield* MessageV2.filterCompactedEffect(forked.id)
       expect(childFiltered).toHaveLength(parentFiltered.length)
+      expect(childOptimized.map((item) => item.info.id)).toEqual(childFiltered.map((item) => item.info.id))
 
       const tailPart = childFiltered.flatMap((m) => m.parts).find((p) => p.type === "compaction")
       expect(tailPart?.type).toBe("compaction")
@@ -868,8 +1056,10 @@ describe("MessageV2.filterCompacted", () => {
         })
 
         const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
+        const optimized = yield* MessageV2.filterCompactedEffect(sessionID)
 
         expect(result.map((item) => item.info.id)).toEqual([c1, s1, a3, u3, a4])
+        expect(optimized.map((item) => item.info.id)).toEqual(result.map((item) => item.info.id))
       }),
     ),
   )
@@ -940,8 +1130,10 @@ describe("MessageV2.filterCompacted", () => {
         })
 
         const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
+        const optimized = yield* MessageV2.filterCompactedEffect(sessionID)
 
         expect(result.map((item) => item.info.id)).toEqual([c2, s2, u3, a3, u4, a4])
+        expect(optimized.map((item) => item.info.id)).toEqual(result.map((item) => item.info.id))
       }),
     ),
   )