admin/DeskClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix startup chat history recovery for Issue #816 (#821)

Browse Source
This commit is contained in:
Lingxuan Zuo
2026-04-12 15:30:11 +08:00
committed by GitHub
Unverified
parent 87ab12849c
commit 4ff6861042
6 changed files with 783 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
src/stores/chat.ts
View File

@@ -8,6 +8,14 @@ import { hostApiFetch } from '@/lib/host-api';
import { useGatewayStore } from './gateway';
import { useAgentsStore } from './agents';
import { buildCronSessionHistoryPath, isCronSessionKey } from './chat/cron-session-utils';
import {
CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS,
classifyHistoryStartupRetryError,
getHistoryLoadingSafetyTimeout,
getStartupHistoryTimeoutOverride,
shouldRetryStartupHistoryLoad,
sleep,
} from './chat/history-startup-retry';
import {
DEFAULT_CANONICAL_PREFIX,
DEFAULT_SESSION_KEY,
@@ -52,6 +60,7 @@ let _loadSessionsInFlight: Promise<void> | null = null;
let _lastLoadSessionsAt = 0;
const _historyLoadInFlight = new Map<string, Promise<void>>();
const _lastHistoryLoadAtBySession = new Map<string, number>();
const _foregroundHistoryLoadSeen = new Set<string>();
const SESSION_LOAD_MIN_INTERVAL_MS = 1_200;
const HISTORY_LOAD_MIN_INTERVAL_MS = 800;
const HISTORY_POLL_SILENCE_WINDOW_MS = 2_500;
@@ -1304,6 +1313,8 @@ export const useChatStore = create<ChatState>((set, get) => ({
loadHistory: async (quiet = false) => {
const { currentSessionKey } = get();
const isInitialForegroundLoad = !quiet && !_foregroundHistoryLoadSeen.has(currentSessionKey);
const historyTimeoutOverride = getStartupHistoryTimeoutOverride(isInitialForegroundLoad);
const existingLoad = _historyLoadInFlight.get(currentSessionKey);
if (existingLoad) {
await existingLoad;
@@ -1323,7 +1334,7 @@ export const useChatStore = create<ChatState>((set, get) => ({
const loadingSafetyTimer = quiet ? null : setTimeout(() => {
loadingTimedOut = true;
set({ loading: false });
}, 15_000);
}, getHistoryLoadingSafetyTimeout(isInitialForegroundLoad));
const loadPromise = (async () => {
const isCurrentSession = () => get().currentSessionKey === currentSessionKey;
@@ -1367,7 +1378,7 @@ export const useChatStore = create<ChatState>((set, get) => ({
// Guard: if the user switched sessions while this async load was in
// flight, discard the result to prevent overwriting the new session's
// messages with stale data from the old session.
if (!isCurrentSession()) return;
if (!isCurrentSession()) return false;
// Before filtering: attach images/files from tool_result messages to the next assistant message
const messagesWithToolImages = enrichWithToolResultFiles(rawMessages);
@@ -1466,13 +1477,53 @@ export const useChatStore = create<ChatState>((set, get) => ({
set({ sending: false, activeRunId: null, pendingFinal: false });
}
}
return true;
};
try {
const data = await useGatewayStore.getState().rpc<Record<string, unknown>>(
'chat.history',
{ sessionKey: currentSessionKey, limit: 200 },
);
let data: Record<string, unknown> | null = null;
let lastError: unknown = null;
for (let attempt = 0; attempt <= CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS.length; attempt += 1) {
if (!isCurrentSession()) {
break;
}
try {
data = await useGatewayStore.getState().rpc<Record<string, unknown>>(
'chat.history',
{ sessionKey: currentSessionKey, limit: 200 },
historyTimeoutOverride,
);
lastError = null;
break;
} catch (error) {
lastError = error;
}
if (!isCurrentSession()) {
break;
}
const errorKind = classifyHistoryStartupRetryError(lastError);
const shouldRetry = isInitialForegroundLoad
&& attempt < CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS.length
&& shouldRetryStartupHistoryLoad(useGatewayStore.getState().status, errorKind);
if (!shouldRetry) {
break;
}
console.warn('[chat.history] startup retry scheduled', {
sessionKey: currentSessionKey,
attempt: attempt + 1,
gatewayState: useGatewayStore.getState().status.state,
errorKind,
error: String(lastError),
});
await sleep(CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS[attempt]!);
}
if (data) {
let rawMessages = Array.isArray(data.messages) ? data.messages as RawMessage[] : [];
const thinkingLevel = data.thinkingLevel ? String(data.thinkingLevel) : null;
@@ -1480,20 +1531,40 @@ export const useChatStore = create<ChatState>((set, get) => ({
rawMessages = await loadCronFallbackMessages(currentSessionKey, 200);
}
applyLoadedMessages(rawMessages, thinkingLevel);
const applied = applyLoadedMessages(rawMessages, thinkingLevel);
if (applied && isInitialForegroundLoad) {
_foregroundHistoryLoadSeen.add(currentSessionKey);
}
} else {
if (isCurrentSession() && isInitialForegroundLoad && classifyHistoryStartupRetryError(lastError)) {
console.warn('[chat.history] startup retry exhausted', {
sessionKey: currentSessionKey,
gatewayState: useGatewayStore.getState().status.state,
error: String(lastError),
});
}
const fallbackMessages = await loadCronFallbackMessages(currentSessionKey, 200);
if (fallbackMessages.length > 0) {
applyLoadedMessages(fallbackMessages, null);
const applied = applyLoadedMessages(fallbackMessages, null);
if (applied && isInitialForegroundLoad) {
_foregroundHistoryLoadSeen.add(currentSessionKey);
}
} else {
applyLoadFailure('Failed to load chat history');
applyLoadFailure(
(lastError instanceof Error ? lastError.message : String(lastError))
|| 'Failed to load chat history',
);
}
}
} catch (err) {
console.warn('Failed to load chat history:', err);
const fallbackMessages = await loadCronFallbackMessages(currentSessionKey, 200);
if (fallbackMessages.length > 0) {
applyLoadedMessages(fallbackMessages, null);
const applied = applyLoadedMessages(fallbackMessages, null);
if (applied && isInitialForegroundLoad) {
_foregroundHistoryLoadSeen.add(currentSessionKey);
}
} else {
applyLoadFailure(String(err));
}

109
src/stores/chat/history-actions.ts
View File

@@ -1,5 +1,6 @@
import { invokeIpc } from '@/lib/api-client';
import { hostApiFetch } from '@/lib/host-api';
import { useGatewayStore } from '@/stores/gateway';
import {
clearHistoryPoll,
enrichWithCachedImages,
@@ -12,9 +13,18 @@ import {
toMs,
} from './helpers';
import { buildCronSessionHistoryPath, isCronSessionKey } from './cron-session-utils';
import {
CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS,
classifyHistoryStartupRetryError,
getStartupHistoryTimeoutOverride,
shouldRetryStartupHistoryLoad,
sleep,
} from './history-startup-retry';
import type { RawMessage } from './types';
import type { ChatGet, ChatSet, SessionHistoryActions } from './store-api';
const foregroundHistoryLoadSeen = new Set<string>();
async function loadCronFallbackMessages(sessionKey: string, limit = 200): Promise<RawMessage[]> {
if (!isCronSessionKey(sessionKey)) return [];
try {
@@ -35,6 +45,8 @@ export function createHistoryActions(
return {
loadHistory: async (quiet = false) => {
const { currentSessionKey } = get();
const isInitialForegroundLoad = !quiet && !foregroundHistoryLoadSeen.has(currentSessionKey);
const historyTimeoutOverride = getStartupHistoryTimeoutOverride(isInitialForegroundLoad);
if (!quiet) set({ loading: true, error: null });
const isCurrentSession = () => get().currentSessionKey === currentSessionKey;
@@ -75,7 +87,7 @@ export function createHistoryActions(
};
const applyLoadedMessages = (rawMessages: RawMessage[], thinkingLevel: string | null) => {
if (!isCurrentSession()) return;
if (!isCurrentSession()) return false;
// Before filtering: attach images/files from tool_result messages to the next assistant message
const messagesWithToolImages = enrichWithToolResultFiles(rawMessages);
const filteredMessages = messagesWithToolImages.filter((msg) => !isToolResultRole(msg.role) && !isInternalMessage(msg));
@@ -173,36 +185,103 @@ export function createHistoryActions(
set({ sending: false, activeRunId: null, pendingFinal: false });
}
}
return true;
};
try {
const result = await invokeIpc(
'gateway:rpc',
'chat.history',
{ sessionKey: currentSessionKey, limit: 200 }
) as { success: boolean; result?: Record<string, unknown>; error?: string };
let result: { success: boolean; result?: Record<string, unknown>; error?: string } | null = null;
let lastError: unknown = null;
if (result.success && result.result) {
for (let attempt = 0; attempt <= CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS.length; attempt += 1) {
if (!isCurrentSession()) {
break;
}
try {
result = await invokeIpc(
'gateway:rpc',
'chat.history',
{ sessionKey: currentSessionKey, limit: 200 },
...(historyTimeoutOverride != null ? [historyTimeoutOverride] as const : []),
) as { success: boolean; result?: Record<string, unknown>; error?: string };
if (result.success) {
lastError = null;
break;
}
lastError = new Error(result.error || 'Failed to load chat history');
} catch (error) {
lastError = error;
}
if (!isCurrentSession()) {
break;
}
const errorKind = classifyHistoryStartupRetryError(lastError);
const shouldRetry = result?.success !== true
&& isInitialForegroundLoad
&& attempt < CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS.length
&& shouldRetryStartupHistoryLoad(useGatewayStore.getState().status, errorKind);
if (!shouldRetry) {
break;
}
console.warn('[chat.history] startup retry scheduled', {
sessionKey: currentSessionKey,
attempt: attempt + 1,
gatewayState: useGatewayStore.getState().status.state,
errorKind,
error: String(lastError),
});
await sleep(CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS[attempt]!);
}
if (result?.success && result.result) {
const data = result.result;
let rawMessages = Array.isArray(data.messages) ? data.messages as RawMessage[] : [];
const thinkingLevel = data.thinkingLevel ? String(data.thinkingLevel) : null;
if (rawMessages.length === 0 && isCronSessionKey(currentSessionKey)) {
rawMessages = await loadCronFallbackMessages(currentSessionKey, 200);
}
applyLoadedMessages(rawMessages, thinkingLevel);
} else {
const fallbackMessages = await loadCronFallbackMessages(currentSessionKey, 200);
if (fallbackMessages.length > 0) {
applyLoadedMessages(fallbackMessages, null);
} else {
applyLoadFailure(result.error || 'Failed to load chat history');
const applied = applyLoadedMessages(rawMessages, thinkingLevel);
if (applied && isInitialForegroundLoad) {
foregroundHistoryLoadSeen.add(currentSessionKey);
}
return;
}
if (isCurrentSession() && isInitialForegroundLoad && classifyHistoryStartupRetryError(lastError)) {
console.warn('[chat.history] startup retry exhausted', {
sessionKey: currentSessionKey,
gatewayState: useGatewayStore.getState().status.state,
error: String(lastError),
});
}
const fallbackMessages = await loadCronFallbackMessages(currentSessionKey, 200);
if (fallbackMessages.length > 0) {
const applied = applyLoadedMessages(fallbackMessages, null);
if (applied && isInitialForegroundLoad) {
foregroundHistoryLoadSeen.add(currentSessionKey);
}
} else {
applyLoadFailure(
result?.error
|| (lastError instanceof Error ? lastError.message : String(lastError))
|| 'Failed to load chat history',
);
}
} catch (err) {
console.warn('Failed to load chat history:', err);
const fallbackMessages = await loadCronFallbackMessages(currentSessionKey, 200);
if (fallbackMessages.length > 0) {
applyLoadedMessages(fallbackMessages, null);
const applied = applyLoadedMessages(fallbackMessages, null);
if (applied && isInitialForegroundLoad) {
foregroundHistoryLoadSeen.add(currentSessionKey);
}
} else {
applyLoadFailure(String(err));
}

79
src/stores/chat/history-startup-retry.ts Normal file
View File

@@ -0,0 +1,79 @@
import type { GatewayStatus } from '@/types/gateway';
export const CHAT_HISTORY_RPC_TIMEOUT_MS = 35_000;
export const CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS = [600] as const;
export const CHAT_HISTORY_STARTUP_CONNECTION_GRACE_MS = 15_000;
export const CHAT_HISTORY_STARTUP_RUNNING_WINDOW_MS =
CHAT_HISTORY_RPC_TIMEOUT_MS + CHAT_HISTORY_STARTUP_CONNECTION_GRACE_MS;
export const CHAT_HISTORY_DEFAULT_LOADING_SAFETY_TIMEOUT_MS = 15_000;
export const CHAT_HISTORY_LOADING_SAFETY_TIMEOUT_MS =
CHAT_HISTORY_RPC_TIMEOUT_MS * (CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS.length + 1)
+ CHAT_HISTORY_STARTUP_RETRY_DELAYS_MS.reduce((sum, delay) => sum + delay, 0)
+ 2_000;
export type HistoryRetryErrorKind = 'timeout' | 'gateway_unavailable';
export function classifyHistoryStartupRetryError(error: unknown): HistoryRetryErrorKind | null {
const message = String(error).toLowerCase();
if (
message.includes('rpc timeout: chat.history')
|| message.includes('gateway rpc timeout: chat.history')
|| message.includes('gateway ws timeout: chat.history')
|| message.includes('request timed out')
) {
return 'timeout';
}
if (
message.includes('gateway not connected')
|| message.includes('gateway socket is not connected')
|| message.includes('gateway is unavailable')
|| message.includes('service channel unavailable')
|| message.includes('websocket closed before handshake')
|| message.includes('connect handshake timeout')
|| message.includes('gateway ws connect timeout')
|| message.includes('gateway connection closed')
) {
return 'gateway_unavailable';
}
return null;
}
export function shouldRetryStartupHistoryLoad(
gatewayStatus: GatewayStatus | undefined,
errorKind: HistoryRetryErrorKind | null,
): boolean {
if (!gatewayStatus || !errorKind) return false;
if (gatewayStatus.state === 'starting') {
return true;
}
if (gatewayStatus.state !== 'running') {
return false;
}
if (gatewayStatus.connectedAt == null) {
return true;
}
return Date.now() - gatewayStatus.connectedAt <= CHAT_HISTORY_STARTUP_RUNNING_WINDOW_MS;
}
export async function sleep(ms: number): Promise<void> {
await new Promise((resolve) => setTimeout(resolve, ms));
}
export function getStartupHistoryTimeoutOverride(
isInitialForegroundLoad: boolean,
): number | undefined {
return isInitialForegroundLoad ? CHAT_HISTORY_RPC_TIMEOUT_MS : undefined;
}
export function getHistoryLoadingSafetyTimeout(isInitialForegroundLoad: boolean): number {
return isInitialForegroundLoad
? CHAT_HISTORY_LOADING_SAFETY_TIMEOUT_MS
: CHAT_HISTORY_DEFAULT_LOADING_SAFETY_TIMEOUT_MS;
}