From af2830db9b611a6459f4bd8f46c564b32cd4537a Mon Sep 17 00:00:00 2001 From: 7418 <7418@openclaw.ai> Date: Fri, 13 Feb 2026 01:24:27 +0800 Subject: [PATCH] feat: add cursor-based message pagination (P0-3) Backend returns paginated messages with hasMore flag. Frontend shows "Load earlier messages" button with scroll anchor preservation. Co-Authored-By: Claude Opus 4.6 --- .../api/chat/sessions/[id]/messages/route.ts | 13 +- src/app/chat/[id]/page.tsx | 12 +- src/components/chat/ChatView.tsx | 242 ++++++------------ src/components/chat/MessageList.tsx | 46 +++- src/lib/db.ts | 36 ++- src/types/index.ts | 1 + 6 files changed, 180 insertions(+), 170 deletions(-) diff --git a/src/app/api/chat/sessions/[id]/messages/route.ts b/src/app/api/chat/sessions/[id]/messages/route.ts index 368c1a6..a2b6ba5 100644 --- a/src/app/api/chat/sessions/[id]/messages/route.ts +++ b/src/app/api/chat/sessions/[id]/messages/route.ts @@ -3,7 +3,7 @@ import { getMessages, getSession } from '@/lib/db'; import type { MessagesResponse } from '@/types'; export async function GET( - _request: NextRequest, + request: NextRequest, { params }: { params: Promise<{ id: string }> } ) { try { @@ -13,8 +13,15 @@ export async function GET( return Response.json({ error: 'Session not found' }, { status: 404 }); } - const messages = getMessages(id); - const response: MessagesResponse = { messages }; + const searchParams = request.nextUrl.searchParams; + const limitParam = searchParams.get('limit'); + const beforeParam = searchParams.get('before'); + + const limit = limitParam ? Math.min(Math.max(parseInt(limitParam, 10) || 100, 1), 500) : 100; + const beforeRowId = beforeParam ? parseInt(beforeParam, 10) || undefined : undefined; + + const { messages, hasMore } = getMessages(id, { limit, beforeRowId }); + const response: MessagesResponse = { messages, hasMore }; return Response.json(response); } catch (error) { const message = error instanceof Error ? error.message : 'Failed to fetch messages'; diff --git a/src/app/chat/[id]/page.tsx b/src/app/chat/[id]/page.tsx index b455ff3..1b83ae5 100644 --- a/src/app/chat/[id]/page.tsx +++ b/src/app/chat/[id]/page.tsx @@ -1,6 +1,7 @@ 'use client'; import { useEffect, useState, use } from 'react'; +import Link from 'next/link'; import type { Message, MessagesResponse, ChatSession } from '@/types'; import { ChatView } from '@/components/chat/ChatView'; import { HugeiconsIcon } from "@hugeicons/react"; @@ -14,6 +15,7 @@ interface ChatSessionPageProps { export default function ChatSessionPage({ params }: ChatSessionPageProps) { const { id } = use(params); const [messages, setMessages] = useState([]); + const [hasMore, setHasMore] = useState(false); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); const [sessionTitle, setSessionTitle] = useState(''); @@ -52,12 +54,13 @@ export default function ChatSessionPage({ params }: ChatSessionPageProps) { setLoading(true); setError(null); setMessages([]); + setHasMore(false); let cancelled = false; async function loadMessages() { try { - const res = await fetch(`/api/chat/sessions/${id}/messages`); + const res = await fetch(`/api/chat/sessions/${id}/messages?limit=100`); if (cancelled) return; if (!res.ok) { if (res.status === 404) { @@ -69,6 +72,7 @@ export default function ChatSessionPage({ params }: ChatSessionPageProps) { const data: MessagesResponse = await res.json(); if (cancelled) return; setMessages(data.messages); + setHasMore(data.hasMore ?? false); } catch (err) { if (cancelled) return; setError(err instanceof Error ? err.message : 'Failed to load messages'); @@ -95,9 +99,9 @@ export default function ChatSessionPage({ params }: ChatSessionPageProps) {
); @@ -116,7 +120,7 @@ export default function ChatSessionPage({ params }: ChatSessionPageProps) { )} - + ); } diff --git a/src/components/chat/ChatView.tsx b/src/components/chat/ChatView.tsx index d7e670f..8be4111 100644 --- a/src/components/chat/ChatView.tsx +++ b/src/components/chat/ChatView.tsx @@ -1,10 +1,11 @@ 'use client'; import { useState, useCallback, useEffect, useRef } from 'react'; -import type { Message, SSEEvent, TokenUsage, PermissionRequestEvent, FileAttachment } from '@/types'; +import type { Message, MessagesResponse, PermissionRequestEvent, FileAttachment } from '@/types'; import { MessageList } from './MessageList'; import { MessageInput } from './MessageInput'; import { usePanel } from '@/hooks/usePanel'; +import { consumeSSEStream } from '@/hooks/useSSEStream'; interface ToolUseInfo { id: string; @@ -20,13 +21,16 @@ interface ToolResultInfo { interface ChatViewProps { sessionId: string; initialMessages?: Message[]; + initialHasMore?: boolean; modelName?: string; initialMode?: string; } -export function ChatView({ sessionId, initialMessages = [], modelName, initialMode }: ChatViewProps) { +export function ChatView({ sessionId, initialMessages = [], initialHasMore = false, modelName, initialMode }: ChatViewProps) { const { setStreamingSessionId, workingDirectory, setWorkingDirectory, setPanelOpen, setPendingApprovalSessionId } = usePanel(); const [messages, setMessages] = useState(initialMessages); + const [hasMore, setHasMore] = useState(initialHasMore); + const [loadingMore, setLoadingMore] = useState(false); const [streamingContent, setStreamingContent] = useState(''); const [isStreaming, setIsStreaming] = useState(false); const [toolUses, setToolUses] = useState([]); @@ -103,6 +107,31 @@ export function ChatView({ sessionId, initialMessages = [], modelName, initialMo } }, [initialMode]); + // Sync hasMore when initial data loads + useEffect(() => { + setHasMore(initialHasMore); + }, [initialHasMore]); + + const loadEarlierMessages = useCallback(async () => { + if (loadingMore || !hasMore || messages.length === 0) return; + setLoadingMore(true); + try { + // Use _rowid of the earliest message as cursor + const earliest = messages[0]; + const earliestRowId = (earliest as Message & { _rowid?: number })._rowid; + if (!earliestRowId) return; + const res = await fetch(`/api/chat/sessions/${sessionId}/messages?limit=100&before=${earliestRowId}`); + if (!res.ok) return; + const data: MessagesResponse = await res.json(); + setHasMore(data.hasMore ?? false); + if (data.messages.length > 0) { + setMessages(prev => [...data.messages, ...prev]); + } + } finally { + setLoadingMore(false); + } + }, [sessionId, messages, hasMore, loadingMore]); + const stopStreaming = useCallback(() => { abortControllerRef.current?.abort(); abortControllerRef.current = null; @@ -199,161 +228,57 @@ export function ChatView({ sessionId, initialMessages = [], modelName, initialMo const reader = response.body?.getReader(); if (!reader) throw new Error('No response stream'); - const decoder = new TextDecoder(); - let tokenUsage: TokenUsage | null = null; - let buffer = ''; - - while (true) { - const { done, value } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split('\n'); - // Keep the last potentially incomplete line in the buffer - buffer = lines.pop() || ''; - - for (const line of lines) { - if (!line.startsWith('data: ')) continue; - - try { - const event: SSEEvent = JSON.parse(line.slice(6)); - - switch (event.type) { - case 'text': { - accumulated += event.data; - accumulatedRef.current = accumulated; - setStreamingContent(accumulated); - break; - } - - case 'tool_use': { - try { - const toolData = JSON.parse(event.data); - // Clear streaming output for new tool - setStreamingToolOutput(''); - setToolUses((prev) => { - // Avoid duplicates - if (prev.some((t) => t.id === toolData.id)) return prev; - return [...prev, { - id: toolData.id, - name: toolData.name, - input: toolData.input, - }]; - }); - } catch { - // skip malformed tool_use data - } - break; - } - - case 'tool_result': { - try { - const resultData = JSON.parse(event.data); - setStreamingToolOutput(''); - setToolResults((prev) => [...prev, { - tool_use_id: resultData.tool_use_id, - content: resultData.content, - }]); - } catch { - // skip malformed tool_result data - } - break; - } - - case 'tool_output': { - // Check if this is a progress heartbeat or real stderr output - try { - const parsed = JSON.parse(event.data); - if (parsed._progress) { - // SDK tool_progress event — update status with elapsed time - setStatusText(`Running ${parsed.tool_name}... (${Math.round(parsed.elapsed_time_seconds)}s)`); - break; - } - } catch { - // Not JSON — it's raw stderr output, fall through - } - // Real-time stderr output from tool execution - setStreamingToolOutput((prev) => { - const next = prev + (prev ? '\n' : '') + event.data; - return next.length > 5000 ? next.slice(-5000) : next; - }); - break; - } - - case 'status': { - try { - const statusData = JSON.parse(event.data); - if (statusData.session_id) { - // Init event — show briefly then clear so tool status can take over - setStatusText(`Connected (${statusData.model || 'claude'})`); - setTimeout(() => setStatusText(undefined), 2000); - } else if (statusData.notification) { - // Notification from SDK hooks — show as progress - setStatusText(statusData.message || statusData.title || undefined); - } else { - setStatusText(typeof event.data === 'string' ? event.data : undefined); - } - } catch { - setStatusText(event.data || undefined); - } - break; - } - - case 'result': { - try { - const resultData = JSON.parse(event.data); - if (resultData.usage) { - tokenUsage = resultData.usage; - } - } catch { - // skip - } - setStatusText(undefined); - break; - } - - case 'permission_request': { - try { - const permData: PermissionRequestEvent = JSON.parse(event.data); - setPendingPermission(permData); - setPermissionResolved(null); - setPendingApprovalSessionId(sessionId); - } catch { - // skip malformed permission_request data - } - break; - } - - case 'tool_timeout': { - try { - const timeoutData = JSON.parse(event.data); - toolTimeoutRef.current = { - toolName: timeoutData.tool_name, - elapsedSeconds: timeoutData.elapsed_seconds, - }; - } catch { - // skip malformed timeout data - } - break; - } - - case 'error': { - accumulated += '\n\n**Error:** ' + event.data; - accumulatedRef.current = accumulated; - setStreamingContent(accumulated); - break; - } - - case 'done': { - // Stream complete - break; - } - } - } catch { - // skip malformed SSE lines + const result = await consumeSSEStream(reader, { + onText: (acc) => { + accumulated = acc; + accumulatedRef.current = acc; + setStreamingContent(acc); + }, + onToolUse: (tool) => { + setStreamingToolOutput(''); + setToolUses((prev) => { + if (prev.some((t) => t.id === tool.id)) return prev; + return [...prev, tool]; + }); + }, + onToolResult: (res) => { + setStreamingToolOutput(''); + setToolResults((prev) => [...prev, res]); + }, + onToolOutput: (data) => { + setStreamingToolOutput((prev) => { + const next = prev + (prev ? '\n' : '') + data; + return next.length > 5000 ? next.slice(-5000) : next; + }); + }, + onToolProgress: (toolName, elapsed) => { + setStatusText(`Running ${toolName}... (${elapsed}s)`); + }, + onStatus: (text) => { + if (text?.startsWith('Connected (')) { + setStatusText(text); + setTimeout(() => setStatusText(undefined), 2000); + } else { + setStatusText(text); } - } - } + }, + onResult: () => { /* token usage captured by consumeSSEStream */ }, + onPermissionRequest: (permData) => { + setPendingPermission(permData); + setPermissionResolved(null); + setPendingApprovalSessionId(sessionId); + }, + onToolTimeout: (toolName, elapsedSeconds) => { + toolTimeoutRef.current = { toolName, elapsedSeconds }; + }, + onError: (acc) => { + accumulated = acc; + accumulatedRef.current = acc; + setStreamingContent(acc); + }, + }); + + accumulated = result.accumulated; // Add the assistant message to the list if (accumulated.trim()) { @@ -363,7 +288,7 @@ export function ChatView({ sessionId, initialMessages = [], modelName, initialMo role: 'assistant', content: accumulated.trim(), created_at: new Date().toISOString(), - token_usage: tokenUsage ? JSON.stringify(tokenUsage) : null, + token_usage: result.tokenUsage ? JSON.stringify(result.tokenUsage) : null, }; setMessages((prev) => [...prev, assistantMessage]); } @@ -541,6 +466,9 @@ export function ChatView({ sessionId, initialMessages = [], modelName, initialMo onPermissionResponse={handlePermissionResponse} permissionResolved={permissionResolved} onForceStop={stopStreaming} + hasMore={hasMore} + loadingMore={loadingMore} + onLoadMore={loadEarlierMessages} /> void; permissionResolved?: 'allow' | 'deny' | null; onForceStop?: () => void; + hasMore?: boolean; + loadingMore?: boolean; + onLoadMore?: () => void; } export function MessageList({ @@ -49,7 +53,34 @@ export function MessageList({ onPermissionResponse, permissionResolved, onForceStop, + hasMore, + loadingMore, + onLoadMore, }: MessageListProps) { + // Scroll anchor: preserve position when older messages are prepended + const anchorIdRef = useRef(null); + const prevMessageCountRef = useRef(messages.length); + + // Before loading more, record the first visible message ID + const handleLoadMore = () => { + if (messages.length > 0) { + anchorIdRef.current = messages[0].id; + } + onLoadMore?.(); + }; + + // After messages are prepended, scroll the anchor element back into view + useEffect(() => { + if (anchorIdRef.current && messages.length > prevMessageCountRef.current) { + const el = document.getElementById(`msg-${anchorIdRef.current}`); + if (el) { + el.scrollIntoView({ block: 'start' }); + } + anchorIdRef.current = null; + } + prevMessageCountRef.current = messages.length; + }, [messages]); + if (messages.length === 0 && !isStreaming) { return (
@@ -65,8 +96,21 @@ export function MessageList({ return ( + {hasMore && ( +
+ +
+ )} {messages.map((message) => ( - +
+ +
))} {isStreaming && ( diff --git a/src/lib/db.ts b/src/lib/db.ts index a13411b..2c93cd5 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -1,17 +1,17 @@ import Database from 'better-sqlite3'; import path from 'path'; import crypto from 'crypto'; +import fs from 'fs'; +import os from 'os'; import type { ChatSession, Message, SettingsMap, TaskItem, TaskStatus, ApiProvider, CreateProviderRequest, UpdateProviderRequest } from '@/types'; -const dataDir = process.env.CLAUDE_GUI_DATA_DIR || path.join(require('os').homedir(), '.codepilot'); +const dataDir = process.env.CLAUDE_GUI_DATA_DIR || path.join(os.homedir(), '.codepilot'); const DB_PATH = path.join(dataDir, 'codepilot.db'); let db: Database.Database | null = null; export function getDb(): Database.Database { if (!db) { - const fs = require('fs'); - const os = require('os'); const dir = path.dirname(DB_PATH); if (!fs.existsSync(dir)) { fs.mkdirSync(dir, { recursive: true }); @@ -276,9 +276,35 @@ export function updateSessionMode(id: string, mode: string): void { // Message Operations // ========================================== -export function getMessages(sessionId: string): Message[] { +export function getMessages( + sessionId: string, + options?: { limit?: number; beforeRowId?: number }, +): { messages: Message[]; hasMore: boolean } { const db = getDb(); - return db.prepare('SELECT * FROM messages WHERE session_id = ? ORDER BY created_at ASC').all(sessionId) as Message[]; + const limit = options?.limit ?? 100; + const beforeRowId = options?.beforeRowId; + + let rows: Message[]; + if (beforeRowId) { + // Fetch `limit + 1` rows before the cursor to detect if there are more + rows = db.prepare( + 'SELECT *, rowid as _rowid FROM messages WHERE session_id = ? AND rowid < ? ORDER BY rowid DESC LIMIT ?' + ).all(sessionId, beforeRowId, limit + 1) as Message[]; + } else { + // Fetch the most recent `limit + 1` messages + rows = db.prepare( + 'SELECT *, rowid as _rowid FROM messages WHERE session_id = ? ORDER BY rowid DESC LIMIT ?' + ).all(sessionId, limit + 1) as Message[]; + } + + const hasMore = rows.length > limit; + if (hasMore) { + rows = rows.slice(0, limit); + } + + // Reverse to chronological order (ASC) + rows.reverse(); + return { messages: rows, hasMore }; } export function addMessage( diff --git a/src/types/index.ts b/src/types/index.ts index 31d5b7d..06e26f1 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -243,6 +243,7 @@ export interface SessionResponse { export interface MessagesResponse { messages: Message[]; + hasMore?: boolean; } export interface SuccessResponse {