d8dfun/clawX
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refine chat history timeout (#1081)

Browse Source
This commit is contained in:
paisley
2026-05-29 13:29:56 +08:00
committed by GitHub
parent 97dca873a1
commit 6e305e6dba
7 changed files with 430 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
harness/specs/tasks/fix-chat-history-gateway-timeout.md
View File

@@ -19,8 +19,12 @@ touchedAreas:
- src/stores/chat/session-label-hydration.ts
- src/stores/chat/store-api.ts
- src/stores/chat/types.ts
- tests/setup.ts
- tests/e2e/chat-history-startup-retry.spec.ts
- tests/unit/chat-store-history-retry.test.ts
- tests/unit/chat-store-session-label-fetch.test.ts
- tests/unit/gateway-rpc-backpressure.test.ts
- tests/unit/history-startup-retry.test.ts
- tests/unit/session-label-fetch.test.ts
- tests/unit/session-summaries-route.test.ts
expectedUserBehavior:
@@ -38,6 +42,9 @@ requiredRules:
- api-client-transport-policy
requiredTests:
- pnpm run typecheck
- tests/e2e/chat-history-startup-retry.spec.ts
- tests/unit/chat-store-history-retry.test.ts
- tests/unit/history-startup-retry.test.ts
- tests/unit/chat-store-session-label-fetch.test.ts
- tests/unit/gateway-rpc-backpressure.test.ts
- tests/unit/session-label-fetch.test.ts
@@ -48,6 +55,7 @@ acceptance:
- Startup/restart no longer fans out sidebar label chat.history calls before the visible session history finishes loading.
- Sidebar label hydration no longer depends on gateway chat.history full-session scans.
- Foreground history uses a bounded startup RPC wait and falls back to local transcript reads instead of surfacing transient RPC timeout errors.
- Foreground startup history can show local transcript data while chat.history is pending, then replace it with Gateway history without disabling startup retry.
- Main-process chat.history RPCs are single-flighted/backpressured before reaching the Gateway.
docs:
required: false

102
src/stores/chat.ts
View File

@@ -12,6 +12,7 @@ import { buildCronSessionHistoryPath, isCronSessionKey } from './chat/cron-sessi
import { pickStartupSessionFallback } from './chat/session-selection';
import {
CHAT_HISTORY_DISK_FALLBACK_TIMEOUT_MS,
CHAT_HISTORY_STARTUP_FALLBACK_RACE_MS,
CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS,
classifyHistoryStartupRetryError,
getHistoryLoadingSafetyTimeout,
@@ -2981,6 +2982,51 @@ export const useChatStore = create<ChatState>((set, get) => ({
return true;
};
let localFallbackApplied = false;
let gatewayHistorySettled = false;
const applyLocalFallbackMessages = async (
options: { onlyWhileGatewayPending?: boolean; logTimeout?: boolean } = {},
): Promise<boolean> => {
const fallbackMessages = await loadLocalHistoryFallback(currentSessionKey, 200, {
logTimeout: options.logTimeout,
});
if (
fallbackMessages.length === 0
|| !isCurrentSession()
|| (options.onlyWhileGatewayPending && gatewayHistorySettled)
) {
return false;
}
const applied = applyLoadedMessages(fallbackMessages, null);
if (!applied) return false;
localFallbackApplied = true;
set({ hasMoreHistory: fallbackMessages.length >= HISTORY_PAGE_SIZE });
if (isInitialForegroundLoad) {
_foregroundHistoryLoadSeen.add(foregroundLoadKey);
void refreshVisibleSessionSummaries(set, get);
}
return true;
};
const applyStartupFallbackAfterGrace = async (): Promise<'fallback' | 'none'> => {
if (!isInitialForegroundLoad || !shouldShowForegroundLoading) {
return 'none';
}
await sleep(CHAT_HISTORY_STARTUP_FALLBACK_RACE_MS);
if (!isCurrentSession() || gatewayHistorySettled) {
return 'none';
}
const applied = await applyLocalFallbackMessages({
onlyWhileGatewayPending: true,
logTimeout: false,
});
return applied ? 'fallback' : 'none';
};
const loadGatewayHistory = async (): Promise<void> => {
try {
const fallbackMessages: RawMessage[] = [];
const gatewayRpc = useGatewayStore.getState().rpc.bind(useGatewayStore.getState());
@@ -3049,6 +3095,11 @@ export const useChatStore = create<ChatState>((set, get) => ({
);
}
if (rawMessages.length === 0 && localFallbackApplied && !isCronSessionKey(currentSessionKey)) {
set({ loading: false });
return;
}
const applied = applyLoadedMessages(rawMessages, thinkingLevel);
if (applied) {
set({ hasMoreHistory: rawMessages.length >= HISTORY_PAGE_SIZE });
@@ -3067,18 +3118,20 @@ export const useChatStore = create<ChatState>((set, get) => ({
});
}
const lateFallbackMessages = fallbackMessages.length > 0
? fallbackMessages
: await loadLocalHistoryFallback(currentSessionKey, 200);
if (lateFallbackMessages.length > 0) {
const applied = applyLoadedMessages(lateFallbackMessages, null);
if (applied) {
set({ hasMoreHistory: lateFallbackMessages.length >= HISTORY_PAGE_SIZE });
}
if (applied && isInitialForegroundLoad) {
_foregroundHistoryLoadSeen.add(foregroundLoadKey);
void refreshVisibleSessionSummaries(set, get);
const appliedLateFallback = fallbackMessages.length > 0
? applyLoadedMessages(fallbackMessages, null)
: await applyLocalFallbackMessages();
if (appliedLateFallback) {
if (fallbackMessages.length > 0) {
localFallbackApplied = true;
set({ hasMoreHistory: fallbackMessages.length >= HISTORY_PAGE_SIZE });
if (isInitialForegroundLoad) {
_foregroundHistoryLoadSeen.add(foregroundLoadKey);
void refreshVisibleSessionSummaries(set, get);
}
}
} else if (localFallbackApplied) {
set({ loading: false });
} else if (errorKind === 'timeout' && isInitialForegroundLoad) {
// Keep startup usable while Gateway RPC routing catches up. The
// Sidebar/gateway event refreshes will retry quietly instead of
@@ -3093,20 +3146,25 @@ export const useChatStore = create<ChatState>((set, get) => ({
}
} catch (err) {
console.warn('Failed to load chat history:', err);
const fallbackMessages = await loadLocalHistoryFallback(currentSessionKey, 200);
if (fallbackMessages.length > 0) {
const applied = applyLoadedMessages(fallbackMessages, null);
if (applied) {
set({ hasMoreHistory: fallbackMessages.length >= HISTORY_PAGE_SIZE });
}
if (applied && isInitialForegroundLoad) {
_foregroundHistoryLoadSeen.add(foregroundLoadKey);
void refreshVisibleSessionSummaries(set, get);
}
} else {
const applied = await applyLocalFallbackMessages();
if (!applied && localFallbackApplied) {
set({ loading: false });
} else if (!applied) {
applyLoadFailure(String(err));
}
} finally {
gatewayHistorySettled = true;
}
};
const gatewayLoadPromise = loadGatewayHistory();
if (isInitialForegroundLoad && shouldShowForegroundLoading) {
await Promise.race([
gatewayLoadPromise.then(() => 'gateway' as const),
applyStartupFallbackAfterGrace(),
]);
}
await gatewayLoadPromise;
})();
_historyLoadInFlight.set(currentSessionKey, loadPromise);

4
src/stores/chat/history-startup-retry.ts
View File

@@ -5,6 +5,7 @@ export const CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS = [800, 2_000, 4_000, 8_000] a
export const CHAT_HISTORY_STARTUP_CONNECTION_GRACE_MS = 30_000;
export const CHAT_HISTORY_STARTUP_RUNNING_WINDOW_MS =
CHAT_HISTORY_RPC_TIMEOUT_MS + CHAT_HISTORY_STARTUP_CONNECTION_GRACE_MS;
export const CHAT_HISTORY_STARTUP_FALLBACK_RACE_MS = 1_500;
export const CHAT_HISTORY_DISK_FALLBACK_TIMEOUT_MS = 4_000;
export const CHAT_HISTORY_DEFAULT_LOADING_SAFETY_TIMEOUT_MS = 15_000;
export const CHAT_HISTORY_LOADING_SAFETY_TIMEOUT_MS =
@@ -28,8 +29,11 @@ export function classifyHistoryStartupRetryError(error: unknown): HistoryRetryEr
if (
message.includes('rpc timeout: chat.history')
|| message.includes('rpc timeout: chat:history')
|| message.includes('gateway rpc timeout: chat.history')
|| message.includes('gateway rpc timeout: chat:history')
|| message.includes('gateway ws timeout: chat.history')
|| message.includes('gateway ws timeout: chat:history')
|| message.includes('request timed out')
) {
return 'timeout';

96
tests/e2e/chat-history-startup-retry.spec.ts
View File

@@ -99,4 +99,100 @@ test.describe('ClawX startup chat history recovery', () => {
await closeElectronApp(app);
}
});
test('renders local transcript while initial chat.history is still pending', async ({ launchElectronApp }) => {
const app = await launchElectronApp({ skipSetup: true });
try {
await installIpcMocks(app, {
gatewayStatus: { state: 'running', port: 18789, pid: 12345, connectedAt: Date.now() },
gatewayRpc: {},
hostApi: {
[stableStringify(['/api/gateway/status', 'GET'])]: {
ok: true,
data: {
status: 200,
ok: true,
json: { state: 'running', port: 18789, pid: 12345, connectedAt: Date.now() },
},
},
[stableStringify(['/api/agents', 'GET'])]: {
ok: true,
data: {
status: 200,
ok: true,
json: { success: true, agents: [{ id: 'main', name: 'main' }] },
},
},
[stableStringify(['/api/sessions/transcript?sessionKey=agent%3Amain%3Amain&limit=200', 'GET'])]: {
ok: true,
data: {
status: 200,
ok: true,
json: {
messages: [
{ role: 'assistant', content: 'local transcript while gateway is pending', timestamp: 1000 },
],
},
},
},
},
});
await app.evaluate(async ({ app: _app }) => {
const { ipcMain } = process.mainModule!.require('electron') as typeof import('electron');
ipcMain.removeHandler('gateway:rpc');
ipcMain.handle('gateway:rpc', async (_event: unknown, method: string, payload: unknown) => {
const stableStringify = (value: unknown): string => {
if (value == null || typeof value !== 'object') return JSON.stringify(value);
if (Array.isArray(value)) return `[${value.map((item) => stableStringify(item)).join(',')}]`;
const entries = Object.entries(value as Record<string, unknown>)
.sort(([left], [right]) => left.localeCompare(right))
.map(([key, entryValue]) => `${JSON.stringify(key)}:${stableStringify(entryValue)}`);
return `{${entries.join(',')}}`;
};
const key = stableStringify([method, payload ?? null]);
if (key === stableStringify(['sessions.list', {}])) {
return {
success: true,
result: {
sessions: [{ key: 'agent:main:main', displayName: 'main' }],
},
};
}
if (key === stableStringify(['chat.history', { sessionKey: 'agent:main:main', limit: 200, maxChars: 500000 }])) {
await new Promise((resolve) => setTimeout(resolve, 5_000));
return {
success: true,
result: {
messages: [
{ role: 'assistant', content: 'gateway authoritative history after delay', timestamp: 1001 },
],
},
};
}
return { success: true, result: {} };
});
});
const page = await getStableWindow(app);
try {
await page.reload();
} catch (error) {
if (!String(error).includes('ERR_FILE_NOT_FOUND')) {
throw error;
}
}
await expect(page.getByTestId('main-layout')).toBeVisible();
await expect(page.getByText('local transcript while gateway is pending')).toBeVisible({ timeout: 10_000 });
await expect(page.getByText('gateway authoritative history after delay')).toBeVisible({ timeout: 10_000 });
await expect(page.getByText('local transcript while gateway is pending')).toHaveCount(0);
await expect(page.getByText('RPC timeout: chat.history')).toHaveCount(0);
} finally {
await closeElectronApp(app);
}
});
});

38
tests/setup.ts
View File

@@ -50,6 +50,44 @@ if (typeof window !== 'undefined') {
});
}
// Vitest/jsdom can expose an opaque-origin window where localStorage is absent.
// The renderer stores use it for small UI caches, so provide the browser shape
// expected by unit tests when the environment does not.
if (typeof window !== 'undefined') {
let needsLocalStorageMock: boolean;
try {
needsLocalStorageMock = !window.localStorage;
} catch {
needsLocalStorageMock = true;
}
if (needsLocalStorageMock) {
const storage = new Map<string, string>();
const localStorageMock: Storage = {
get length() {
return storage.size;
},
clear: vi.fn(() => storage.clear()),
getItem: vi.fn((key: string) => storage.get(key) ?? null),
key: vi.fn((index: number) => Array.from(storage.keys())[index] ?? null),
removeItem: vi.fn((key: string) => {
storage.delete(key);
}),
setItem: vi.fn((key: string, value: string) => {
storage.set(key, String(value));
}),
};
Object.defineProperty(window, 'localStorage', {
value: localStorageMock,
configurable: true,
});
Object.defineProperty(globalThis, 'localStorage', {
value: localStorageMock,
configurable: true,
});
}
}
// Mock matchMedia
if (typeof window !== 'undefined') {
Object.defineProperty(window, 'matchMedia', {

193
tests/unit/chat-store-history-retry.test.ts
View File

@@ -97,6 +97,199 @@ describe('useChatStore startup history retry', () => {
setTimeoutSpy.mockRestore();
});
it('renders local transcript fallback while the initial gateway history request is still pending', async () => {
const { useChatStore } = await import('@/stores/chat');
useChatStore.setState({
currentSessionKey: 'agent:main:main',
currentAgentId: 'main',
sessions: [{ key: 'agent:main:main' }],
messages: [],
sessionLabels: {},
sessionLastActivity: {},
sending: false,
activeRunId: null,
streamingText: '',
streamingMessage: null,
streamingTools: [],
pendingFinal: false,
lastUserMessageAt: null,
pendingToolImages: [],
error: null,
loading: false,
thinkingLevel: null,
});
let resolveGatewayHistory: ((value: { messages: Array<{ role: string; content: string; timestamp: number }> }) => void) | null = null;
gatewayRpcMock.mockImplementation(async (method: string) => {
if (method === 'config.get') return {};
if (method !== 'chat.history') {
throw new Error(`Unexpected gateway RPC: ${method}`);
}
return await new Promise<{ messages: Array<{ role: string; content: string; timestamp: number }> }>((resolve) => {
resolveGatewayHistory = resolve;
});
});
hostApiFetchMock.mockImplementation(async (url: string) => {
if (url.startsWith('/api/sessions/transcript')) {
return {
messages: [{ role: 'assistant', content: 'local transcript first', timestamp: 1000 }],
};
}
return { messages: [] };
});
const loadPromise = useChatStore.getState().loadHistory(false);
await vi.advanceTimersByTimeAsync(1_500);
expect(useChatStore.getState().loading).toBe(false);
expect(useChatStore.getState().messages.map((message) => message.content)).toEqual([
'local transcript first',
]);
expect(gatewayRpcMock).toHaveBeenCalledWith(
'chat.history',
chatHistoryRpcParams('agent:main:main', 200),
35_000,
);
const quietReloadWhileGatewayPending = useChatStore.getState().loadHistory(true);
await Promise.resolve();
expect(gatewayRpcMock.mock.calls.filter(([method]) => method === 'chat.history')).toHaveLength(1);
resolveGatewayHistory?.({
messages: [{ role: 'assistant', content: 'gateway authoritative history', timestamp: 1001 }],
});
await loadPromise;
await quietReloadWhileGatewayPending;
await vi.waitFor(() => {
expect(useChatStore.getState().messages.map((message) => message.content)).toEqual([
'gateway authoritative history',
]);
});
});
it('keeps startup retry active after rendering local transcript fallback', async () => {
const { useChatStore } = await import('@/stores/chat');
useChatStore.setState({
currentSessionKey: 'agent:main:main',
currentAgentId: 'main',
sessions: [{ key: 'agent:main:main' }],
messages: [],
sessionLabels: {},
sessionLastActivity: {},
sending: false,
activeRunId: null,
streamingText: '',
streamingMessage: null,
streamingTools: [],
pendingFinal: false,
lastUserMessageAt: null,
pendingToolImages: [],
error: null,
loading: false,
thinkingLevel: null,
});
let rejectFirstGatewayHistory: ((reason?: unknown) => void) | null = null;
let chatHistoryCalls = 0;
gatewayRpcMock.mockImplementation(async (method: string) => {
if (method === 'config.get') return {};
if (method !== 'chat.history') {
throw new Error(`Unexpected gateway RPC: ${method}`);
}
chatHistoryCalls += 1;
if (chatHistoryCalls === 1) {
return await new Promise((_resolve, reject) => {
rejectFirstGatewayHistory = reject;
});
}
return {
messages: [{ role: 'assistant', content: 'gateway history after retry', timestamp: 1001 }],
};
});
hostApiFetchMock.mockImplementation(async (url: string) => {
if (url.startsWith('/api/sessions/transcript')) {
return {
messages: [{ role: 'assistant', content: 'local transcript first', timestamp: 1000 }],
};
}
return { messages: [] };
});
const loadPromise = useChatStore.getState().loadHistory(false);
await vi.advanceTimersByTimeAsync(1_500);
expect(useChatStore.getState().messages.map((message) => message.content)).toEqual([
'local transcript first',
]);
rejectFirstGatewayHistory?.(new Error('RPC timeout: chat.history'));
await Promise.resolve();
await vi.advanceTimersByTimeAsync(800);
await loadPromise;
expect(gatewayRpcMock.mock.calls.filter(([method]) => method === 'chat.history')).toHaveLength(2);
expect(useChatStore.getState().messages.map((message) => message.content)).toEqual([
'gateway history after retry',
]);
});
it('keeps local transcript fallback when gateway returns empty history', async () => {
const { useChatStore } = await import('@/stores/chat');
useChatStore.setState({
currentSessionKey: 'agent:main:main',
currentAgentId: 'main',
sessions: [{ key: 'agent:main:main' }],
messages: [],
sessionLabels: {},
sessionLastActivity: {},
sending: false,
activeRunId: null,
streamingText: '',
streamingMessage: null,
streamingTools: [],
pendingFinal: false,
lastUserMessageAt: null,
pendingToolImages: [],
error: null,
loading: false,
thinkingLevel: null,
});
let resolveGatewayHistory: ((value: { messages: Array<unknown> }) => void) | null = null;
gatewayRpcMock.mockImplementation(async (method: string) => {
if (method === 'config.get') return {};
if (method !== 'chat.history') {
throw new Error(`Unexpected gateway RPC: ${method}`);
}
return await new Promise<{ messages: Array<unknown> }>((resolve) => {
resolveGatewayHistory = resolve;
});
});
hostApiFetchMock.mockImplementation(async (url: string) => {
if (url.startsWith('/api/sessions/transcript')) {
return {
messages: [{ role: 'assistant', content: 'local transcript remains', timestamp: 1000 }],
};
}
return { messages: [] };
});
const loadPromise = useChatStore.getState().loadHistory(false);
await vi.advanceTimersByTimeAsync(1_500);
expect(useChatStore.getState().messages.map((message) => message.content)).toEqual([
'local transcript remains',
]);
resolveGatewayHistory?.({ messages: [] });
await loadPromise;
expect(useChatStore.getState().loading).toBe(false);
expect(useChatStore.getState().messages.map((message) => message.content)).toEqual([
'local transcript remains',
]);
});
it('forces the internal final-message reload through the quiet history cooldown', async () => {
const { useChatStore } = await import('@/stores/chat');
useChatStore.setState({

11
tests/unit/history-startup-retry.test.ts Normal file
View File

@@ -0,0 +1,11 @@
import { describe, expect, it } from 'vitest';
import { classifyHistoryStartupRetryError } from '@/stores/chat/history-startup-retry';
describe('history startup retry classification', () => {
it('treats dotted and colon chat history timeout names as retryable timeouts', () => {
expect(classifyHistoryStartupRetryError(new Error('RPC timeout: chat.history'))).toBe('timeout');
expect(classifyHistoryStartupRetryError(new Error('RPC timeout: chat:history'))).toBe('timeout');
expect(classifyHistoryStartupRetryError(new Error('Gateway RPC timeout: chat:history'))).toBe('timeout');
expect(classifyHistoryStartupRetryError(new Error('Gateway WS timeout: chat:history'))).toBe('timeout');
});
});