Skip to content

feat(session): bi-directional cursor-based pagination (#6548) #8535

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
CasualDeveloper wants to merge 1 commit into
base: dev
Choose a base branch
from
+1,229 −25
Open

feat(session): bi-directional cursor-based pagination (#6548) #8535

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
290 changes: 286 additions & 4 deletions packages/opencode/src/cli/cmd/tui/context/sync.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,12 @@ import { useArgs } from "./args"
import { batch, onMount } from "solid-js"
import { Log } from "@/util/log"
import type { Path } from "@opencode-ai/sdk"
import { parseLinkHeader } from "@/util/link-header"

/** Maximum messages kept in memory per session */
const MAX_LOADED_MESSAGES = 500
/** Chunk size for eviction when limit exceeded */
const EVICTION_CHUNK_SIZE = 50

export const { use: useSync, provider: SyncProvider } = createSimpleContext({
name: "Sync",
Expand All @@ -48,6 +54,15 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({
}
config: Config
session: Session[]
message_page: {
[sessionID: string]: {
hasOlder: boolean
hasNewer: boolean
loading: boolean
loadingDirection?: "older" | "newer"
error?: string
}
}
session_status: {
[sessionID: string]: SessionStatus
}
Expand Down Expand Up @@ -89,6 +104,7 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({
provider: [],
provider_default: {},
session: [],
message_page: {},
session_status: {},
session_diff: {},
todo: {},
Expand Down Expand Up @@ -226,19 +242,24 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({
}

case "message.updated": {
const messages = store.message[event.properties.info.sessionID]
const sessionID = event.properties.info.sessionID
const page = store.message_page[sessionID]
const messages = store.message[sessionID]
if (!messages) {
setStore("message", event.properties.info.sessionID, [event.properties.info])
setStore("message", sessionID, [event.properties.info])
break
}
const result = Binary.search(messages, event.properties.info.id, (m) => m.id)
if (result.found) {
setStore("message", event.properties.info.sessionID, result.index, reconcile(event.properties.info))
setStore("message", sessionID, result.index, reconcile(event.properties.info))
break
}
if (page?.hasNewer) {
break
}
setStore(
"message",
event.properties.info.sessionID,
sessionID,
produce((draft) => {
draft.splice(result.index, 0, event.properties.info)
}),
Expand Down Expand Up @@ -279,6 +300,13 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({
break
}
case "message.part.updated": {
const sessionID = event.properties.part.sessionID
const page = store.message_page[sessionID]
const messages = store.message[sessionID]
const messageExists = messages?.some((m) => m.id === event.properties.part.messageID)
if (page?.hasNewer && !messageExists) {
break
}
const parts = store.part[event.properties.part.messageID]
if (!parts) {
setStore("part", event.properties.part.messageID, [event.properties.part])
Expand Down Expand Up @@ -389,6 +417,7 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({
})

const fullSyncedSessions = new Set<string>()
const loadingGuard = new Set<string>()
const result = {
data: store,
set: setStore,
Expand Down Expand Up @@ -422,6 +451,8 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({
sdk.client.session.todo({ sessionID }),
sdk.client.session.diff({ sessionID }),
])
const link = messages.response.headers.get("link") ?? ""
const hasOlder = parseLinkHeader(link).prev !== undefined
setStore(
produce((draft) => {
const match = Binary.search(draft.session, sessionID, (s) => s.id)
Expand All @@ -433,10 +464,261 @@ export const { use: useSync, provider: SyncProvider } = createSimpleContext({
draft.part[message.info.id] = message.parts
}
draft.session_diff[sessionID] = diff.data ?? []
draft.message_page[sessionID] = { hasOlder, hasNewer: false, loading: false, error: undefined }
}),
)
fullSyncedSessions.add(sessionID)
},
async loadOlder(sessionID: string) {
const page = store.message_page[sessionID]
if (page?.loading || !page?.hasOlder) return
const messages = store.message[sessionID] ?? []
const oldest = messages.at(0)
if (!oldest) return
if (loadingGuard.has(sessionID)) return
loadingGuard.add(sessionID)
try {
setStore("message_page", sessionID, { ...page, loading: true, loadingDirection: "older", error: undefined })

const res = await sdk.client.session.messages(
{ sessionID, before: oldest.id, limit: 100 },
{ throwOnError: true },
)
const link = res.response.headers.get("link") ?? ""
const hasOlder = parseLinkHeader(link).prev !== undefined
setStore(
produce((draft) => {
const existing = draft.message[sessionID] ?? []
for (const msg of res.data ?? []) {
const match = Binary.search(existing, msg.info.id, (m) => m.id)
if (!match.found) {
existing.splice(match.index, 0, msg.info)
draft.part[msg.info.id] = msg.parts
}
}
if (existing.length > MAX_LOADED_MESSAGES + EVICTION_CHUNK_SIZE) {
const evicted = existing.splice(-(existing.length - MAX_LOADED_MESSAGES))
for (const msg of evicted) delete draft.part[msg.id]
draft.message_page[sessionID] = { hasOlder, hasNewer: true, loading: false, error: undefined }
} else {
draft.message_page[sessionID] = {
hasOlder,
hasNewer: draft.message_page[sessionID]?.hasNewer ?? false,
loading: false,
error: undefined,
}
}
}),
)
} catch (e) {
const page = store.message_page[sessionID]
setStore("message_page", sessionID, {
hasOlder: page?.hasOlder ?? false,
hasNewer: page?.hasNewer ?? false,
loading: false,
error: e instanceof Error ? e.message : String(e),
})
} finally {
loadingGuard.delete(sessionID)
}
},
async loadNewer(sessionID: string) {
const page = store.message_page[sessionID]
if (page?.loading || !page?.hasNewer) return
const messages = store.message[sessionID] ?? []
const newest = messages.at(-1)
if (!newest) return
if (loadingGuard.has(sessionID)) return
loadingGuard.add(sessionID)
try {
setStore("message_page", sessionID, { ...page, loading: true, loadingDirection: "newer", error: undefined })

const res = await sdk.client.session.messages(
{ sessionID, after: newest.id, limit: 100 },
{ throwOnError: true },
)
const link = res.response.headers.get("link") ?? ""
const hasNewer = parseLinkHeader(link).next !== undefined
setStore(
produce((draft) => {
const existing = draft.message[sessionID] ?? []
for (const msg of res.data ?? []) {
const match = Binary.search(existing, msg.info.id, (m) => m.id)
if (!match.found) {
existing.splice(match.index, 0, msg.info)
draft.part[msg.info.id] = msg.parts
}
}
if (existing.length > MAX_LOADED_MESSAGES + EVICTION_CHUNK_SIZE) {
const evicted = existing.splice(0, existing.length - MAX_LOADED_MESSAGES)
for (const msg of evicted) delete draft.part[msg.id]
draft.message_page[sessionID] = { hasOlder: true, hasNewer, loading: false, error: undefined }
} else {
draft.message_page[sessionID] = {
hasOlder: draft.message_page[sessionID]?.hasOlder ?? false,
hasNewer,
loading: false,
error: undefined,
}
}
}),
)
} catch (e) {
const page = store.message_page[sessionID]
setStore("message_page", sessionID, {
hasOlder: page?.hasOlder ?? false,
hasNewer: page?.hasNewer ?? false,
loading: false,
error: e instanceof Error ? e.message : String(e),
})
} finally {
loadingGuard.delete(sessionID)
}
},
async jumpToLatest(sessionID: string) {
const page = store.message_page[sessionID]
if (page?.loading || !page?.hasNewer) return
if (loadingGuard.has(sessionID)) return
loadingGuard.add(sessionID)

try {
// Check for revert state
const session = store.session.find((s) => s.id === sessionID)
const revertMessageID = session?.revert?.messageID

setStore("message_page", sessionID, {
...page,
loading: true,
loadingDirection: "newer",
error: undefined,
})

// Fetch newest page (no cursor = newest)
const res = await sdk.client.session.messages({ sessionID, limit: 100 }, { throwOnError: true })

let messages = res.data ?? []
const link = res.response.headers.get("link") ?? ""
const hasOlder = parseLinkHeader(link).prev !== undefined

// Revert-aware: If in revert state and marker not in results, fetch it
if (revertMessageID && !messages.some((m) => m.info.id === revertMessageID)) {
try {
const revertResult = await sdk.client.session.message(
{ sessionID, messageID: revertMessageID },
{ throwOnError: true },
)
if (revertResult.data) {
// Prepend revert message (it's older than newest page)
messages = [revertResult.data, ...messages]
}
} catch (e) {
// Revert message may have been deleted, continue without it
Log.Default.info("Revert marker fetch failed (may be deleted)", {
messageID: revertMessageID,
error: e,
})
}
}

setStore(
produce((draft) => {
// Clean up parts only for messages not in new results
const oldMessages = draft.message[sessionID] ?? []
const newIds = new Set(messages.map((m) => m.info.id))
for (const msg of oldMessages) {
if (!newIds.has(msg.id)) {
delete draft.part[msg.id]
}
}

// Store new messages
draft.message[sessionID] = messages.map((m) => m.info)
for (const msg of messages) {
draft.part[msg.info.id] = msg.parts
}
draft.message_page[sessionID] = {
hasOlder,
hasNewer: false,
loading: false,
error: undefined,
}
}),
)
} catch (e) {
setStore(
produce((draft) => {
const p = draft.message_page[sessionID]
if (p) {
p.loading = false
p.error = e instanceof Error ? e.message : String(e)
}
}),
)
} finally {
loadingGuard.delete(sessionID)
}
},
async jumpToOldest(sessionID: string) {
const page = store.message_page[sessionID]
if (page?.loading || !page?.hasOlder) return
if (loadingGuard.has(sessionID)) return
loadingGuard.add(sessionID)

try {
setStore("message_page", sessionID, {
...page,
loading: true,
loadingDirection: "older",
error: undefined,
})

const res = await sdk.client.session.messages(
{ sessionID, oldest: true, limit: 100 },
{ throwOnError: true },
)

const messages = res.data ?? []
const link = res.response.headers.get("link") ?? ""
const hasNewer = parseLinkHeader(link).next !== undefined

setStore(
produce((draft) => {
// Clean up parts only for messages not in new results
const oldMessages = draft.message[sessionID] ?? []
const newIds = new Set(messages.map((m) => m.info.id))
for (const msg of oldMessages) {
if (!newIds.has(msg.id)) {
delete draft.part[msg.id]
}
}

// Store new messages
draft.message[sessionID] = messages.map((m) => m.info)
for (const msg of messages) {
draft.part[msg.info.id] = msg.parts
}
draft.message_page[sessionID] = {
hasOlder: false,
hasNewer,
loading: false,
error: undefined,
}
}),
)
} catch (e) {
setStore(
produce((draft) => {
const p = draft.message_page[sessionID]
if (p) {
p.loading = false
p.error = e instanceof Error ? e.message : String(e)
}
}),
)
} finally {
loadingGuard.delete(sessionID)
}
},
},
bootstrap,
}
Expand Down
Loading