mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-09-04 10:11:46 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(webui): reconcile threads after browser resume

Browse Source
This commit is contained in:
chengyongru
2026-07-28 16:25:08 +08:00
committed by chengyongru
parent 78cf68c291
commit ae089aa3ae
39 changed files with 6151 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
webui/src/hooks/useNanobotStream.ts
+121 -12
View File
@@ -540,6 +540,8 @@ export function useNanobotStream(
) => SubmittedTurn | null;
transcribeAudio: (dataUrl: string, options?: { durationMs?: number }) => Promise<string>;
stop: () => void;
/** Mark an accepted canonical snapshot as the definitive end of the active turn. */
reconcileTurnComplete: () => void;
setMessages: React.Dispatch<React.SetStateAction<UIMessage[]>>;
/** Latest transport-level fault raised since the last ``dismissStreamError``.
* ``null`` when there is nothing to show. */
@@ -581,10 +583,6 @@ export function useNanobotStream(
* backend changes. */
const streamEndTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
useEffect(() => {
return client.onError((err) => setStreamError(err));
}, [client]);
const dismissStreamError = useCallback(() => setStreamError(null), []);
const clearPendingStreamWork = useCallback(() => {
@@ -654,6 +652,66 @@ export function useNanobotStream(
return !!closedStreamId;
}, []);
const applyStreamError = useCallback((err: StreamError) => {
// One multiplexed client serves every thread. A correlated send fault
// belongs only to its target chat. An uncorrelated transport close can
// still be shown in the mounted thread, but cannot roll back any turn.
if (!chatId || (err.chatId && err.chatId !== chatId)) return;
setStreamError(err);
if (!err.turnId) return;
const rejectedTurnId = err.turnId;
pendingStreamEventsRef.current = pendingStreamEventsRef.current.filter(
(event) => event.turn.turnId !== rejectedTurnId,
);
sideChannelTurnIdsRef.current.delete(rejectedTurnId);
cancelStreamEndTimer();
setMessages((prev) => {
const rejectedRows = prev.filter((message) => message.turnId === rejectedTurnId);
if (rejectedRows.length === 0) return prev;
const rejectedIds = new Set(rejectedRows.map((message) => message.id));
const rejectedSegments = new Set(
rejectedRows
.map((message) => message.activitySegmentId)
.filter((segmentId): segmentId is string => typeof segmentId === "string"),
);
if (
activeAssistantRef.current
&& rejectedIds.has(activeAssistantRef.current.id)
) {
activeAssistantRef.current = null;
}
if (buffer.current && rejectedIds.has(buffer.current.messageId)) {
buffer.current = null;
}
for (const id of rejectedIds) closedAssistantStreamIdsRef.current.delete(id);
if (
activitySegmentRef.current
&& rejectedSegments.has(activitySegmentRef.current)
) {
activitySegmentRef.current = null;
}
if (
fileEditSegmentRef.current
&& rejectedSegments.has(fileEditSegmentRef.current)
) {
fileEditSegmentRef.current = null;
}
return prev.filter((message) => message.turnId !== rejectedTurnId);
});
const remainingStartedAt = client.getRunStartedAt(chatId);
const hasRemainingRun = (
remainingStartedAt !== null
|| client.hasUnsettledRun(chatId)
);
setRunStartedAt(remainingStartedAt);
setIsStreaming(hasRemainingRun);
if (!hasRemainingRun) suppressStreamUntilTurnEndRef.current = false;
}, [cancelStreamEndTimer, chatId, client]);
useEffect(() => client.onError(applyStreamError), [applyStreamError, client]);
const resolveActiveAssistantIndex = useCallback((
prev: UIMessage[],
turn: UIMessageTurnFields = {},
@@ -849,6 +907,15 @@ export function useNanobotStream(
return () => document.removeEventListener("visibilitychange", flushOnReturn);
}, [flushPendingStreamEvents]);
useEffect(() => {
return client.onStatus((status) => {
if (status !== "reconnecting" && status !== "closed") return;
// A transport drop does not prove the backend turn completed. Keep the
// semantic running state intact so queued guidance is not flushed early.
cancelStreamEndTimer();
});
}, [cancelStreamEndTimer, client]);
// Reset local state when switching chats. Do not reset on every
// ``initialMessages`` update: a brand-new chat can receive an empty/404
// history response after the optimistic first message has already rendered.
@@ -883,6 +950,31 @@ export function useNanobotStream(
if (!chatId) return;
const handle = (ev: InboundEvent) => {
if (ev.event === "error") {
if (ev.detail === "message_too_big") {
applyStreamError({
kind: "message_too_big",
chatId,
turnId: ev.turn_id,
});
} else if (ev.detail === "workspace_scope_rejected") {
applyStreamError({
kind: "workspace_scope_rejected",
reason: ev.reason,
chatId,
turnId: ev.turn_id,
});
} else if (ev.turn_id) {
applyStreamError({
kind: "turn_rejected",
detail: ev.detail,
reason: ev.reason,
chatId,
turnId: ev.turn_id,
});
}
return;
}
const sideChannelEvent = isSideChannelEvent(ev);
if (
streamEndTimerRef.current !== null
@@ -1187,8 +1279,7 @@ export function useNanobotStream(
});
return;
}
// ``attached`` / ``error`` frames aren't actionable here; the client
// shell handles them separately.
// ``attached`` frames aren't actionable here.
};
const unsub = client.onChat(chatId, handle);
@@ -1202,6 +1293,7 @@ export function useNanobotStream(
cancelStreamEndTimer();
};
}, [
applyStreamError,
cancelStreamEndTimer,
chatId,
client,
@@ -1271,12 +1363,16 @@ export function useNanobotStream(
});
if (!sideChannel) setIsStreaming(true);
const wireMedia = hasAttachments ? images!.map((i) => i.media) : undefined;
const wireOptions = { ...options, turnId };
delete wireOptions.quotedContext;
delete wireOptions.sideChannel;
delete wireOptions.finalizeActiveTurn;
delete wireOptions.continueActiveTurn;
client.sendMessage(chatId, outboundContent, wireMedia, wireOptions);
const clientOptions = {
...options,
turnId,
...((sideChannel || continueActiveTurn) ? { startsNewRun: false } : {}),
};
delete clientOptions.quotedContext;
delete clientOptions.sideChannel;
delete clientOptions.finalizeActiveTurn;
delete clientOptions.continueActiveTurn;
client.sendMessage(chatId, outboundContent, wireMedia, clientOptions);
return { turnId, userMessageId, sideChannel };
},
[cancelStreamEndTimer, chatId, clearActivitySegment, client, flushPendingStreamEvents],
@@ -1297,6 +1393,18 @@ export function useNanobotStream(
client.sendMessage(chatId, "/stop");
}, [chatId, clearActivitySegment, client, flushPendingStreamEvents]);
const reconcileTurnComplete = useCallback(() => {
cancelStreamEndTimer();
clearPendingStreamWork();
buffer.current = null;
activeAssistantRef.current = null;
closedAssistantStreamIdsRef.current.clear();
clearActivitySegment();
suppressStreamUntilTurnEndRef.current = false;
setRunStartedAt(null);
setIsStreaming(false);
}, [cancelStreamEndTimer, clearActivitySegment, clearPendingStreamWork]);
const transcribeAudio = useCallback(
(dataUrl: string, options?: { durationMs?: number }) =>
client.transcribeAudio(dataUrl, options),
@@ -1312,6 +1420,7 @@ export function useNanobotStream(
send,
transcribeAudio,
stop,
reconcileTurnComplete,
setMessages,
streamError,
dismissStreamError,
webui/src/hooks/useSessions.ts
+201 -53
View File
@@ -24,6 +24,8 @@ const INITIAL_HISTORY_PAGE_LIMIT = 160;
const OLDER_HISTORY_PAGE_LIMIT = 120;
const CHAT_CREATE_TIMEOUT_MS = 60_000;
export type SessionHistoryContinuity = "initial" | "overlap" | "reset";
function persistedMessagesToUi(messages: UIMessage[]): UIMessage[] {
return messages.map((m, idx) => ({
...m,
@@ -32,6 +34,63 @@ function persistedMessagesToUi(messages: UIMessage[]): UIMessage[] {
}));
}
function sameSemanticMessage(a: UIMessage, b: UIMessage): boolean {
return (
a.role === b.role
&& (a.kind ?? "") === (b.kind ?? "")
&& a.content === b.content
&& (!a.turnId || !b.turnId || a.turnId === b.turnId)
);
}
function longestSemanticOverlap(previous: UIMessage[], latest: UIMessage[]): number {
const maxOverlap = Math.min(previous.length, latest.length);
for (let overlap = maxOverlap; overlap > 0; overlap -= 1) {
const previousStart = previous.length - overlap;
let matches = true;
for (let index = 0; index < overlap; index += 1) {
if (!sameSemanticMessage(previous[previousStart + index], latest[index])) {
matches = false;
break;
}
}
if (matches) return overlap;
}
return 0;
}
function mergeLatestHistory(
previous: UIMessage[],
latest: UIMessage[],
initial: boolean,
): {
continuity: SessionHistoryContinuity;
messages: UIMessage[];
retainedPrefixLength: number;
} {
if (initial) {
return {
continuity: "initial",
messages: latest,
retainedPrefixLength: 0,
};
}
const overlapLength = longestSemanticOverlap(previous, latest);
if (overlapLength === 0) {
return {
continuity: "reset",
messages: latest,
retainedPrefixLength: 0,
};
}
const retainedPrefixLength = previous.length - overlapLength;
return {
continuity: "overlap",
messages: [...previous.slice(0, retainedPrefixLength), ...latest],
retainedPrefixLength,
};
}
function hasPendingToolCallsFromThread(
body: Awaited<ReturnType<typeof fetchWebuiThread>>,
messages: UIMessage[],
@@ -42,6 +101,17 @@ function hasPendingToolCallsFromThread(
return hasPendingAgentActivity(messages);
}
function completedTurnIdsFromThread(
body: Awaited<ReturnType<typeof fetchWebuiThread>>,
): string[] {
if (!Array.isArray(body?.completed_turn_ids)) return [];
return Array.from(new Set(
body.completed_turn_ids.filter(
(turnId): turnId is string => typeof turnId === "string" && turnId.length > 0,
),
));
}
/** Sidebar state: fetches the full session list and exposes create / delete actions. */
export function useSessions(): {
sessions: ChatSummary[];
@@ -191,11 +261,20 @@ export function useSessionHistory(key: string | null): {
userMessageOffset: number;
version: number;
forkBoundaryMessageCount: number | null;
/** ``true`` when the replayed transcript ends with a trace row (turn still in flight). */
/** ``true`` when the server reports that the turn is still in flight. */
hasPendingToolCalls: boolean;
/** Turn identities backed by explicit persisted completion events. */
completedTurnIds: string[];
/** Relationship between the latest canonical page and its predecessor. */
continuity: SessionHistoryContinuity;
/** Stable across overlapping latest pages; changes on initial load or reset. */
lineage: number;
/** Exact active turn when supplied by a current gateway. */
activeTurnId: string | null;
} {
const { token } = useClient();
const loadingOlderRef = useRef(false);
const historyVersionRef = useRef(0);
const [refreshSeq, setRefreshSeq] = useState(0);
const refresh = useCallback(() => {
setRefreshSeq((value) => value + 1);
@@ -207,11 +286,15 @@ export function useSessionHistory(key: string | null): {
loadingOlder: boolean;
error: string | null;
hasPendingToolCalls: boolean;
completedTurnIds: string[];
forkBoundaryMessageCount: number | null;
beforeCursor: string | null;
hasMoreBefore: boolean;
userMessageOffset: number;
version: number;
continuity: SessionHistoryContinuity;
lineage: number;
activeTurnId: string | null;
}>({
key: null,
messages: [],
@@ -219,11 +302,15 @@ export function useSessionHistory(key: string | null): {
loadingOlder: false,
error: null,
hasPendingToolCalls: false,
completedTurnIds: [],
forkBoundaryMessageCount: null,
beforeCursor: null,
hasMoreBefore: false,
userMessageOffset: 0,
version: 0,
continuity: "initial",
lineage: 0,
activeTurnId: null,
});
useEffect(() => {
@@ -235,11 +322,15 @@ export function useSessionHistory(key: string | null): {
loadingOlder: false,
error: null,
hasPendingToolCalls: false,
completedTurnIds: [],
forkBoundaryMessageCount: null,
beforeCursor: null,
hasMoreBefore: false,
userMessageOffset: 0,
version: 0,
continuity: "initial",
lineage: 0,
activeTurnId: null,
});
return;
}
@@ -255,11 +346,15 @@ export function useSessionHistory(key: string | null): {
loadingOlder: false,
error: null,
hasPendingToolCalls: false,
completedTurnIds: [],
forkBoundaryMessageCount: null,
beforeCursor: null,
hasMoreBefore: false,
userMessageOffset: 0,
version: 0,
continuity: "initial",
lineage: 0,
activeTurnId: null,
});
(async () => {
try {
@@ -268,56 +363,83 @@ export function useSessionHistory(key: string | null): {
direction: "latest",
});
if (cancelled) return;
if (!body?.messages?.length) {
setState((prev) => ({
historyVersionRef.current += 1;
const responseVersion = historyVersionRef.current;
const completedTurnIds = completedTurnIdsFromThread(body);
const ui = persistedMessagesToUi(body?.messages ?? []);
const hasPending = hasPendingToolCallsFromThread(body, ui);
const forkBoundary = typeof body?.fork_boundary_message_count === "number"
? Math.max(0, Math.min(body.fork_boundary_message_count, ui.length))
: null;
setState((prev) => {
const merged = prev.key === key
? mergeLatestHistory(prev.messages, ui, prev.lineage === 0)
: mergeLatestHistory([], ui, true);
const retainedPrefix = merged.retainedPrefixLength > 0;
const retainedForkBoundary = (
retainedPrefix
&& prev.forkBoundaryMessageCount !== null
&& prev.forkBoundaryMessageCount <= merged.retainedPrefixLength
)
? prev.forkBoundaryMessageCount
: null;
return {
key,
messages: [],
messages: merged.messages,
loading: false,
loadingOlder: false,
error: null,
hasPendingToolCalls: false,
forkBoundaryMessageCount: null,
beforeCursor: null,
hasMoreBefore: false,
userMessageOffset: 0,
version: prev.key === key ? prev.version + 1 : 1,
}));
return;
}
const ui = persistedMessagesToUi(body.messages);
const hasPending = hasPendingToolCallsFromThread(body, ui);
const forkBoundary = typeof body.fork_boundary_message_count === "number"
? Math.max(0, Math.min(body.fork_boundary_message_count, ui.length))
: null;
setState((prev) => ({
key,
messages: ui,
loading: false,
loadingOlder: false,
error: null,
hasPendingToolCalls: hasPending,
forkBoundaryMessageCount: forkBoundary,
beforeCursor: body.page?.before_cursor ?? null,
hasMoreBefore: body.page?.has_more_before === true,
userMessageOffset: Math.max(0, body.page?.user_message_offset ?? 0),
version: prev.key === key ? prev.version + 1 : 1,
}));
hasPendingToolCalls: hasPending,
completedTurnIds,
forkBoundaryMessageCount: forkBoundary === null
? retainedForkBoundary
: forkBoundary + merged.retainedPrefixLength,
beforeCursor: retainedPrefix
? prev.beforeCursor
: body?.page?.before_cursor ?? null,
hasMoreBefore: retainedPrefix
? prev.hasMoreBefore
: body?.page?.has_more_before === true,
userMessageOffset: retainedPrefix
? prev.userMessageOffset
: Math.max(0, body?.page?.user_message_offset ?? 0),
version: responseVersion,
continuity: merged.continuity,
lineage: merged.continuity === "overlap"
? prev.lineage
: responseVersion,
activeTurnId: typeof body?.active_turn_id === "string"
? body.active_turn_id
: null,
};
});
} catch (e) {
if (cancelled) return;
if (e instanceof ApiError && e.status === 404) {
setState((prev) => ({
key,
messages: [],
loading: false,
loadingOlder: false,
error: null,
hasPendingToolCalls: false,
forkBoundaryMessageCount: null,
beforeCursor: null,
hasMoreBefore: false,
userMessageOffset: 0,
version: prev.key === key ? prev.version + 1 : 1,
}));
historyVersionRef.current += 1;
const responseVersion = historyVersionRef.current;
setState((prev) => {
const continuity = prev.key === key && prev.lineage > 0
? "reset"
: "initial";
return {
key,
messages: [],
loading: false,
loadingOlder: false,
error: null,
hasPendingToolCalls: false,
completedTurnIds: [],
forkBoundaryMessageCount: null,
beforeCursor: null,
hasMoreBefore: false,
userMessageOffset: 0,
version: responseVersion,
continuity,
lineage: responseVersion,
activeTurnId: null,
};
});
} else {
setState((prev) => ({
key,
@@ -326,11 +448,15 @@ export function useSessionHistory(key: string | null): {
loadingOlder: false,
error: (e as Error).message,
hasPendingToolCalls: false,
completedTurnIds: [],
forkBoundaryMessageCount: null,
beforeCursor: null,
hasMoreBefore: false,
userMessageOffset: 0,
version: prev.key === key ? prev.version : 0,
continuity: prev.key === key ? prev.continuity : "initial",
lineage: prev.key === key ? prev.lineage : 0,
activeTurnId: prev.key === key ? prev.activeTurnId : null,
}));
}
}
@@ -342,17 +468,26 @@ export function useSessionHistory(key: string | null): {
const loadOlder = useCallback(async () => {
if (!key || loadingOlderRef.current) return;
const before = state.key === key ? state.beforeCursor : null;
if (!before || !state.hasMoreBefore) return;
const requestKey = key;
const requestLineage = state.key === requestKey ? state.lineage : 0;
const beforeCursor = state.key === requestKey ? state.beforeCursor : null;
if (!beforeCursor || !state.hasMoreBefore || requestLineage === 0) return;
const matchesRequest = (candidate: typeof state) => (
candidate.key === requestKey
&& candidate.lineage === requestLineage
&& candidate.beforeCursor === beforeCursor
);
loadingOlderRef.current = true;
setState((prev) => prev.key === key ? { ...prev, loadingOlder: true, error: null } : prev);
setState((prev) => matchesRequest(prev)
? { ...prev, loadingOlder: true, error: null }
: prev);
try {
const body = await fetchWebuiThread(token, key, {
const body = await fetchWebuiThread(token, requestKey, {
limit: OLDER_HISTORY_PAGE_LIMIT,
before,
before: beforeCursor,
});
setState((prev) => {
if (prev.key !== key) return prev;
if (!matchesRequest(prev)) return prev;
if (!body?.messages?.length) {
return {
...prev,
@@ -369,21 +504,21 @@ export function useSessionHistory(key: string | null): {
? null
: prev.forkBoundaryMessageCount + older.length;
const nextMessages = [...older, ...prev.messages];
// An older page cannot change the authoritative latest-turn lifecycle
// state or masquerade as a completed latest-page refresh.
return {
...prev,
messages: nextMessages,
loadingOlder: false,
error: null,
hasPendingToolCalls: hasPendingAgentActivity(nextMessages),
forkBoundaryMessageCount: olderBoundary ?? shiftedBoundary,
beforeCursor: body.page?.before_cursor ?? null,
hasMoreBefore: body.page?.has_more_before === true,
userMessageOffset: Math.max(0, body.page?.user_message_offset ?? 0),
version: prev.version + 1,
};
});
} catch (e) {
setState((prev) => prev.key === key
setState((prev) => matchesRequest(prev)
? {
...prev,
loadingOlder: false,
@@ -398,6 +533,7 @@ export function useSessionHistory(key: string | null): {
state.beforeCursor,
state.hasMoreBefore,
state.key,
state.lineage,
token,
]);
@@ -414,6 +550,10 @@ export function useSessionHistory(key: string | null): {
version: 0,
forkBoundaryMessageCount: null,
hasPendingToolCalls: false,
completedTurnIds: [],
continuity: "initial",
lineage: 0,
activeTurnId: null,
};
}
@@ -432,6 +572,10 @@ export function useSessionHistory(key: string | null): {
version: 0,
forkBoundaryMessageCount: null,
hasPendingToolCalls: false,
completedTurnIds: [],
continuity: "initial",
lineage: 0,
activeTurnId: null,
};
}
@@ -447,6 +591,10 @@ export function useSessionHistory(key: string | null): {
version: state.version,
forkBoundaryMessageCount: state.forkBoundaryMessageCount,
hasPendingToolCalls: state.hasPendingToolCalls,
completedTurnIds: state.completedTurnIds,
continuity: state.continuity,
lineage: state.lineage,
activeTurnId: state.activeTurnId,
};
}