Skip to content

fix(session): bound compacted history hydration #39930

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
brauliobo wants to merge 3 commits into
base: dev
Choose a base branch
from
Open

fix(session): bound compacted history hydration #39930

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
94c5234
fix(session): bound compacted history hydration
brauliobo Jul 31, 2026
4ff112a
fix(session): preserve history for incomplete compactions
brauliobo Aug 6, 2026
69d3227
fix(session): preserve full compaction boundaries
brauliobo Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
126 changes: 93 additions & 33 deletions packages/opencode/src/session/message-v2.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, any> | undefined) {
if (!metadata) return undefined
const { providerExecuted: _, ...rest } = metadata
Expand Down Expand Up @@ -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,
}
})

Expand Down Expand Up @@ -572,7 +590,49 @@ export function filterCompacted(msgs: Iterable<WithParts>) {
}

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<string>()
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
Expand Down
62 changes: 52 additions & 10 deletions packages/opencode/test/session/messages-pagination.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -658,12 +660,27 @@ 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)
}),
),
)

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 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, incomplete, ...after])
}),
),
)

it.instance("skips assistant with error even if marked as summary", () =>
withSession(({ sessionID }) =>
Effect.gen(function* () {
Expand All @@ -677,7 +694,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)
}),
Expand All @@ -694,7 +711,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)
}),
),
Expand Down Expand Up @@ -744,7 +761,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])
}),
Expand Down Expand Up @@ -797,11 +814,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")
Expand Down Expand Up @@ -867,7 +884,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])
}),
Expand Down Expand Up @@ -939,13 +956,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 full 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()
Expand Down
Loading