From 94c5234828e96e885e81e3b47f46bfa05f73527b Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 31 Jul 2026 15:44:37 -0300 Subject: [PATCH 1/3] fix(session): bound compacted history hydration --- packages/opencode/src/session/message-v2.ts | 126 +++++++++++++----- .../test/session/messages-pagination.test.ts | 47 +++++-- 2 files changed, 130 insertions(+), 43 deletions(-) diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index 1bea9f52c3ec..e0c6dd8f9380 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -122,6 +122,52 @@ function hydrate(db: Database.Interface["db"], rows: (typeof MessageTable.$infer }) } +function messageRows( + db: Database.Interface["db"], + input: { + sessionID: SessionID + limit: number + before?: string + }, +) { + return Effect.gen(function* () { + const before = input.before ? cursor.decode(input.before) : undefined + const where = before + ? and(eq(MessageTable.session_id, input.sessionID), older(before)) + : eq(MessageTable.session_id, input.sessionID) + const rows = yield* db + .select() + .from(MessageTable) + .where(where) + .orderBy(desc(MessageTable.time_created), desc(MessageTable.id)) + .limit(input.limit + 1) + .all() + .pipe(Effect.orDie) + if (rows.length === 0) { + const row = yield* db + .select({ id: SessionTable.id }) + .from(SessionTable) + .where(eq(SessionTable.id, input.sessionID)) + .get() + .pipe(Effect.orDie) + if (!row) return yield* new NotFoundError({ message: `Session not found: ${input.sessionID}` }) + return { + rows: [], + more: false, + } + } + + const more = rows.length > input.limit + const slice = more ? rows.slice(0, input.limit) : rows + const tail = slice.at(-1) + return { + rows: slice, + more, + cursor: more && tail ? cursor.encode({ id: tail.id, time: tail.time_created }) : undefined, + } + }) +} + function providerMeta(metadata: Record | undefined) { if (!metadata) return undefined const { providerExecuted: _, ...rest } = metadata @@ -428,41 +474,13 @@ export const page = Effect.fn("MessageV2.page")(function* (input: { before?: string }) { const { db } = yield* Database.Service - const before = input.before ? cursor.decode(input.before) : undefined - const where = before - ? and(eq(MessageTable.session_id, input.sessionID), older(before)) - : eq(MessageTable.session_id, input.sessionID) - const rows = yield* db - .select() - .from(MessageTable) - .where(where) - .orderBy(desc(MessageTable.time_created), desc(MessageTable.id)) - .limit(input.limit + 1) - .all() - .pipe(Effect.orDie) - if (rows.length === 0) { - const row = yield* db - .select({ id: SessionTable.id }) - .from(SessionTable) - .where(eq(SessionTable.id, input.sessionID)) - .get() - .pipe(Effect.orDie) - if (!row) return yield* new NotFoundError({ message: `Session not found: ${input.sessionID}` }) - return { - items: [] as WithParts[], - more: false, - } - } - - const more = rows.length > input.limit - const slice = more ? rows.slice(0, input.limit) : rows - const items = yield* hydrate(db, slice) + const result = yield* messageRows(db, input) + const items = yield* hydrate(db, result.rows) items.reverse() - const tail = slice.at(-1) return { items, - more, - cursor: more && tail ? cursor.encode({ id: tail.id, time: tail.time_created }) : undefined, + more: result.more, + cursor: result.cursor, } }) @@ -572,7 +590,49 @@ export function filterCompacted(msgs: Iterable) { } export const filterCompactedEffect = Effect.fnUntraced(function* (sessionID: SessionID) { - return filterCompacted(yield* stream(sessionID)) + const { db } = yield* Database.Service + const size = 50 + const rows = [] as (typeof MessageTable.$inferSelect)[] + const completed = new Set() + let retain: MessageID | undefined + let before: string | undefined + + pages: while (true) { + const next = yield* messageRows(db, { sessionID, limit: size, before }).pipe( + Effect.catchIf( + (error): error is NotFoundError => NotFoundError.isInstance(error), + () => Effect.succeed({ rows: [] as (typeof MessageTable.$inferSelect)[], more: false, cursor: undefined }), + ), + ) + if (next.rows.length === 0) break + for (const row of next.rows) { + rows.push(row) + const message = info(row) + if (retain) { + if (message.id === retain) break pages + continue + } + if (message.role === "user" && completed.has(message.id)) { + const compaction = (yield* parts(message.id)).find((item): item is CompactionPart => item.type === "compaction") + if (!compaction) continue + if (!compaction.tail_start_id) break pages + retain = compaction.tail_start_id + if (message.id === retain) break pages + continue + } + if (message.role === "assistant" && message.summary && message.finish && !message.error) { + completed.add(message.parentID) + } + } + if (!next.more || !next.cursor) break + before = next.cursor + } + + const result = [] as WithParts[] + for (let index = 0; index < rows.length; index += size) { + result.push(...(yield* hydrate(db, rows.slice(index, index + size)))) + } + return filterCompacted(result) }) // filterCompacted reorders messages for model consumption diff --git a/packages/opencode/test/session/messages-pagination.test.ts b/packages/opencode/test/session/messages-pagination.test.ts index b67c982ebcd3..a5f6d6cf7df3 100644 --- a/packages/opencode/test/session/messages-pagination.test.ts +++ b/packages/opencode/test/session/messages-pagination.test.ts @@ -2,7 +2,9 @@ import { describe, expect, test } from "bun:test" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { SessionV1 } from "@opencode-ai/core/v1/session" import { SessionProjector } from "@opencode-ai/core/session/projector" +import { Database } from "@opencode-ai/core/database/database" import { Effect, Option } from "effect" +import { sql } from "drizzle-orm" import { Session as SessionNs } from "@/session/session" import { MessageV2 } from "../../src/session/message-v2" import { MessageID, PartID, type SessionID } from "../../src/session/schema" @@ -602,7 +604,7 @@ describe("MessageV2.filterCompacted", () => { Effect.gen(function* () { const ids = yield* fill(sessionID, 5) - const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID)) + const result = yield* MessageV2.filterCompactedEffect(sessionID) expect(result).toHaveLength(5) // reversed from newest-first to chronological expect(result.map((item) => item.info.id)).toEqual(ids) @@ -636,7 +638,7 @@ describe("MessageV2.filterCompacted", () => { text: "new response", }) - const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID)) + const result = yield* MessageV2.filterCompactedEffect(sessionID) // Includes compaction boundary: u1, a1, u2, a2 expect(result[0].info.id).toBe(u1) expect(result.length).toBe(4) @@ -658,7 +660,7 @@ describe("MessageV2.filterCompacted", () => { yield* addCompactionPart(sessionID, u1) yield* addUser(sessionID, "world") - const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID)) + const result = yield* MessageV2.filterCompactedEffect(sessionID) expect(result).toHaveLength(2) }), ), @@ -677,7 +679,7 @@ describe("MessageV2.filterCompacted", () => { yield* addAssistant(sessionID, u1, { summary: true, finish: "end_turn", error }) yield* addUser(sessionID, "retry") - const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID)) + const result = yield* MessageV2.filterCompactedEffect(sessionID) // Error assistant doesn't add to completed, so compaction boundary never triggers expect(result).toHaveLength(3) }), @@ -694,7 +696,7 @@ describe("MessageV2.filterCompacted", () => { yield* addAssistant(sessionID, u1, { summary: true }) yield* addUser(sessionID, "next") - const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID)) + const result = yield* MessageV2.filterCompactedEffect(sessionID) expect(result).toHaveLength(3) }), ), @@ -744,7 +746,7 @@ describe("MessageV2.filterCompacted", () => { text: "third reply", }) - const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID)) + const result = yield* MessageV2.filterCompactedEffect(sessionID) expect(result.map((item) => item.info.id)).toEqual([c1, s1, u2, a2, u3, a3]) }), @@ -797,11 +799,11 @@ describe("MessageV2.filterCompacted", () => { text: "third reply", }) - const parentFiltered = MessageV2.filterCompacted(yield* MessageV2.stream(created.id)) + const parentFiltered = yield* MessageV2.filterCompactedEffect(created.id) expect(parentFiltered.map((item) => item.info.id)).toEqual([c1, s1, u2, a2, u3, a3]) const forked = yield* session.fork({ sessionID: created.id }) - const childFiltered = MessageV2.filterCompacted(yield* MessageV2.stream(forked.id)) + const childFiltered = yield* MessageV2.filterCompactedEffect(forked.id) expect(childFiltered).toHaveLength(parentFiltered.length) const tailPart = childFiltered.flatMap((m) => m.parts).find((p) => p.type === "compaction") @@ -867,7 +869,7 @@ describe("MessageV2.filterCompacted", () => { text: "third reply", }) - const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID)) + const result = yield* MessageV2.filterCompactedEffect(sessionID) expect(result.map((item) => item.info.id)).toEqual([c1, s1, a3, u3, a4]) }), @@ -939,13 +941,38 @@ describe("MessageV2.filterCompacted", () => { text: "fourth reply", }) - const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID)) + const result = yield* MessageV2.filterCompactedEffect(sessionID) expect(result.map((item) => item.info.id)).toEqual([c2, s2, u3, a3, u4, a4]) }), ), ) + it.instance("does not hydrate parts before a completed compaction", () => + withSession(({ session, sessionID }) => + Effect.gen(function* () { + const old = yield* addUser(sessionID, "old prompt") + const compaction = yield* addUser(sessionID) + yield* addCompactionPart(sessionID, compaction) + const summary = yield* addAssistant(sessionID, compaction, { summary: true, finish: "end_turn" }) + yield* session.updatePart({ + id: PartID.ascending(), + sessionID, + messageID: summary, + type: "text", + text: "summary", + }) + const recent = yield* fill(sessionID, 60) + + const { db } = yield* Database.Service + yield* db.run(sql`UPDATE part SET data = 'invalid json' WHERE message_id = ${old}`) + + const result = yield* MessageV2.filterCompactedEffect(sessionID) + expect(result.map((item) => item.info.id)).toEqual([compaction, summary, ...recent]) + }), + ), + ) + test("works with array input", () => { // filterCompacted accepts any Iterable, not just generators const id = MessageID.ascending() From 4ff112ab1999b82c10775ea0676d708d43e82aff Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Thu, 6 Aug 2026 20:11:19 -0300 Subject: [PATCH 2/3] fix(session): preserve history for incomplete compactions --- packages/opencode/src/session/message-v2.ts | 4 ++-- .../test/session/messages-pagination.test.ts | 20 +++++++++++++++++-- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index e0c6dd8f9380..c6bbb83ebad6 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -549,7 +549,7 @@ export function filterCompacted(msgs: Iterable) { 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 + if (!part.tail_start_id) continue retain = part.tail_start_id if (msg.info.id === retain) break continue @@ -615,7 +615,7 @@ export const filterCompactedEffect = Effect.fnUntraced(function* (sessionID: Ses if (message.role === "user" && completed.has(message.id)) { const compaction = (yield* parts(message.id)).find((item): item is CompactionPart => item.type === "compaction") if (!compaction) continue - if (!compaction.tail_start_id) break pages + if (!compaction.tail_start_id) continue retain = compaction.tail_start_id if (message.id === retain) break pages continue diff --git a/packages/opencode/test/session/messages-pagination.test.ts b/packages/opencode/test/session/messages-pagination.test.ts index a5f6d6cf7df3..a5c922a3cb32 100644 --- a/packages/opencode/test/session/messages-pagination.test.ts +++ b/packages/opencode/test/session/messages-pagination.test.ts @@ -666,6 +666,21 @@ describe("MessageV2.filterCompacted", () => { ), ) + it.instance("retains history for an incomplete compaction", () => + withSession(({ sessionID }) => + Effect.gen(function* () { + const before = yield* fill(sessionID, 2) + const compaction = yield* addUser(sessionID) + yield* addCompactionPart(sessionID, compaction) + const summary = yield* addAssistant(sessionID, compaction, { summary: true, finish: "end_turn" }) + const after = yield* fill(sessionID, 60) + + const result = yield* MessageV2.filterCompactedEffect(sessionID) + expect(result.map((item) => item.info.id)).toEqual([...before, compaction, summary, ...after]) + }), + ), + ) + it.instance("skips assistant with error even if marked as summary", () => withSession(({ sessionID }) => Effect.gen(function* () { @@ -952,8 +967,9 @@ describe("MessageV2.filterCompacted", () => { withSession(({ session, sessionID }) => Effect.gen(function* () { const old = yield* addUser(sessionID, "old prompt") + const tail = yield* addUser(sessionID, "retained prompt") const compaction = yield* addUser(sessionID) - yield* addCompactionPart(sessionID, compaction) + yield* addCompactionPart(sessionID, compaction, tail) const summary = yield* addAssistant(sessionID, compaction, { summary: true, finish: "end_turn" }) yield* session.updatePart({ id: PartID.ascending(), @@ -968,7 +984,7 @@ describe("MessageV2.filterCompacted", () => { yield* db.run(sql`UPDATE part SET data = 'invalid json' WHERE message_id = ${old}`) const result = yield* MessageV2.filterCompactedEffect(sessionID) - expect(result.map((item) => item.info.id)).toEqual([compaction, summary, ...recent]) + expect(result.map((item) => item.info.id)).toEqual([compaction, summary, tail, ...recent]) }), ), ) From 69d3227b52c4f97edc468584ebd983e27ac784a8 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 14 Aug 2026 14:04:24 -0300 Subject: [PATCH 3/3] fix(session): preserve full compaction boundaries --- packages/opencode/src/session/message-v2.ts | 4 ++-- .../opencode/test/session/messages-pagination.test.ts | 11 +++++------ 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index c6bbb83ebad6..e0c6dd8f9380 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -549,7 +549,7 @@ export function filterCompacted(msgs: Iterable) { 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) continue + if (!part.tail_start_id) break retain = part.tail_start_id if (msg.info.id === retain) break continue @@ -615,7 +615,7 @@ export const filterCompactedEffect = Effect.fnUntraced(function* (sessionID: Ses if (message.role === "user" && completed.has(message.id)) { const compaction = (yield* parts(message.id)).find((item): item is CompactionPart => item.type === "compaction") if (!compaction) continue - if (!compaction.tail_start_id) continue + if (!compaction.tail_start_id) break pages retain = compaction.tail_start_id if (message.id === retain) break pages continue diff --git a/packages/opencode/test/session/messages-pagination.test.ts b/packages/opencode/test/session/messages-pagination.test.ts index a5c922a3cb32..e2a2169761a9 100644 --- a/packages/opencode/test/session/messages-pagination.test.ts +++ b/packages/opencode/test/session/messages-pagination.test.ts @@ -672,11 +672,11 @@ describe("MessageV2.filterCompacted", () => { const before = yield* fill(sessionID, 2) const compaction = yield* addUser(sessionID) yield* addCompactionPart(sessionID, compaction) - const summary = yield* addAssistant(sessionID, compaction, { summary: true, finish: "end_turn" }) + const incomplete = yield* addAssistant(sessionID, compaction, { finish: "end_turn" }) const after = yield* fill(sessionID, 60) const result = yield* MessageV2.filterCompactedEffect(sessionID) - expect(result.map((item) => item.info.id)).toEqual([...before, compaction, summary, ...after]) + expect(result.map((item) => item.info.id)).toEqual([...before, compaction, incomplete, ...after]) }), ), ) @@ -963,13 +963,12 @@ describe("MessageV2.filterCompacted", () => { ), ) - it.instance("does not hydrate parts before a completed compaction", () => + it.instance("does not hydrate parts before a completed full compaction", () => withSession(({ session, sessionID }) => Effect.gen(function* () { const old = yield* addUser(sessionID, "old prompt") - const tail = yield* addUser(sessionID, "retained prompt") const compaction = yield* addUser(sessionID) - yield* addCompactionPart(sessionID, compaction, tail) + yield* addCompactionPart(sessionID, compaction) const summary = yield* addAssistant(sessionID, compaction, { summary: true, finish: "end_turn" }) yield* session.updatePart({ id: PartID.ascending(), @@ -984,7 +983,7 @@ describe("MessageV2.filterCompacted", () => { yield* db.run(sql`UPDATE part SET data = 'invalid json' WHERE message_id = ${old}`) const result = yield* MessageV2.filterCompactedEffect(sessionID) - expect(result.map((item) => item.info.id)).toEqual([compaction, summary, tail, ...recent]) + expect(result.map((item) => item.info.id)).toEqual([compaction, summary, ...recent]) }), ), )