perf: move conversation persistence, refresh and agent events out of App
Persistence and history loading, refresh scheduling and the agent event handler work against the conversation store outside React; App.tsx drops to 9,566 lines and the lint ceiling to 9,600.
This commit is contained in:
mesalogo
@@ -4842,6 +4842,34 @@ describe("App", () => {
|
||||
expect(sidebarRenderProbes.row.mock.calls.length).toBeLessThanOrEqual(4);
|
||||
});
|
||||
|
||||
it("does not re-render the App root when the active conversation's streaming text is flushed", async () => {
|
||||
render(<App />);
|
||||
const composer = await screen.findByLabelText("向 GoodBuddy 提问");
|
||||
fireEvent.change(composer, { target: { value: "Render counter" } });
|
||||
fireEvent.click(screen.getByLabelText("发送"));
|
||||
await waitFor(() => expect(run).toHaveBeenCalledOnce());
|
||||
const requestId = run.mock.calls[0]![0].requestId;
|
||||
act(() => agentListener?.({ requestId, type: "text", delta: "First chunk" }));
|
||||
await screen.findByText("First chunk");
|
||||
// Let the first flush land and the run settle into a steady state.
|
||||
await act(() => new Promise((resolve) => setTimeout(resolve, 400)));
|
||||
// useUnviewedCompletions runs once per App render: it counts App renders.
|
||||
assistantTasksProbe.mockClear();
|
||||
historyRenderProbes.pane.mockClear();
|
||||
for (const delta of [" second", " third"]) {
|
||||
act(() => agentListener?.({ requestId, type: "text", delta }));
|
||||
}
|
||||
expect(screen.getByText("First chunk second third")).toBeInTheDocument();
|
||||
// The 250 ms flush absorbs the deltas into the conversation store.
|
||||
await act(() => new Promise((resolve) => setTimeout(resolve, 400)));
|
||||
expect(screen.getByText("First chunk second third")).toBeInTheDocument();
|
||||
expect(assistantTasksProbe).not.toHaveBeenCalled();
|
||||
// The history pane takes the absorbed conversation through its own subscription.
|
||||
expect(historyRenderProbes.pane).toHaveBeenCalled();
|
||||
fireEvent.click(screen.getByLabelText("停止生成"));
|
||||
await waitFor(() => expect(api.agent.cancel).toHaveBeenCalledWith(requestId));
|
||||
});
|
||||
|
||||
it("isolates history and task strips while typing during a running request", async () => {
|
||||
const conversationId = crypto.randomUUID();
|
||||
vi.mocked(api.conversations.list).mockResolvedValueOnce([{
|
||||
@@ -12608,7 +12636,8 @@ describe("App", () => {
|
||||
expect(within(process).queryByText(`Progress snapshot ${step - 1}`)).not.toBeInTheDocument();
|
||||
expect(within(process).queryByText(`Reading file ${step - 1}`)).not.toBeInTheDocument();
|
||||
}
|
||||
expect(assistantTasksProbe).toHaveBeenCalled();
|
||||
// Progress changes only the message: App may skip rendering entirely
|
||||
// (PERF-13), and any render it does make keeps the same task list.
|
||||
for (const [tasks] of assistantTasksProbe.mock.calls) {
|
||||
expect(tasks).toBe(runningTasks);
|
||||
}
|
||||
|
||||
+175
-1545
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,121 @@
|
||||
import { memo, useEffect, useMemo, useState, type ComponentProps } from "react";
|
||||
import { useTranslation } from "react-i18next";
|
||||
import type {
|
||||
AssistantProject,
|
||||
AssistantSchedule,
|
||||
AssistantTask,
|
||||
InteractiveWorkMode,
|
||||
} from "../../shared/assistant-contracts";
|
||||
import { normalizeInteractiveWorkMode } from "../../shared/assistant-contracts";
|
||||
import { ChatHistoryPane } from "./ChatHistoryPane";
|
||||
import type { Conversation } from "./chat-conversation";
|
||||
import { useConversation, type ConversationStore } from "./conversation-store";
|
||||
import { ConversationTaskStrip } from "./ConversationTaskStrip";
|
||||
import { displayErrorMessage } from "./error-message";
|
||||
import { RuntimeChecklistStrip } from "./RuntimeChecklistStrip";
|
||||
|
||||
function ConversationHistoryLoader({ conversationId, active, load }: {
|
||||
conversationId: string;
|
||||
active: boolean;
|
||||
load: (id: string) => Promise<Conversation>;
|
||||
}): React.JSX.Element {
|
||||
const { t } = useTranslation("app");
|
||||
const [error, setError] = useState<string>();
|
||||
const [attempt, setAttempt] = useState(0);
|
||||
useEffect(() => {
|
||||
let mounted = true;
|
||||
void load(conversationId).catch(reason => {
|
||||
if (mounted) setError(displayErrorMessage(reason, t("conversation.historyLoadFailed")));
|
||||
});
|
||||
return () => { mounted = false; };
|
||||
}, [conversationId, load, attempt, t]);
|
||||
return <section hidden={!active} className="empty-state" aria-busy={!error}>
|
||||
<p role={error ? "alert" : "status"}>{error ?? t("conversation.historyLoading")}</p>
|
||||
{error && <button type="button" className="secondary-button" onClick={() => {
|
||||
setError(undefined);
|
||||
setAttempt(value => value + 1);
|
||||
}}>{t("conversation.historyRetry")}</button>}
|
||||
</section>;
|
||||
}
|
||||
|
||||
type PaneProps = Omit<ComponentProps<typeof ChatHistoryPane>, "conversation" | "taskStrip">;
|
||||
|
||||
export type ConversationHistorySlotProps = PaneProps & {
|
||||
conversationId: string;
|
||||
store: ConversationStore;
|
||||
loadHistory: (conversationId: string) => Promise<Conversation>;
|
||||
/** The active conversation's effective mode; others use their own or the project's. */
|
||||
workModeOverride?: InteractiveWorkMode;
|
||||
projects: readonly AssistantProject[];
|
||||
tasks: AssistantTask[];
|
||||
schedules: AssistantSchedule[];
|
||||
selectedAssistantTaskId?: string;
|
||||
onRemoveSchedule: (scheduleId: string) => Promise<void>;
|
||||
onRunSchedule: (scheduleId: string) => Promise<void>;
|
||||
onSelectTask: (taskId: string) => void;
|
||||
onSetScheduleEnabled: (scheduleId: string, enabled: boolean) => Promise<void>;
|
||||
};
|
||||
|
||||
/**
|
||||
* One kept conversation pane. It subscribes to its own conversation in the
|
||||
* store, so message updates (streaming flushes, tool events) re-render only
|
||||
* this slot, never App; the task strip element is rebuilt only when its
|
||||
* inputs change, keeping the memoized pane skipped otherwise.
|
||||
*/
|
||||
export const ConversationHistorySlot = memo(function ConversationHistorySlot({
|
||||
conversationId,
|
||||
store,
|
||||
loadHistory,
|
||||
workModeOverride,
|
||||
projects,
|
||||
tasks,
|
||||
schedules,
|
||||
selectedAssistantTaskId,
|
||||
onRemoveSchedule,
|
||||
onRunSchedule,
|
||||
onSelectTask,
|
||||
onSetScheduleEnabled,
|
||||
...paneProps
|
||||
}: ConversationHistorySlotProps): React.JSX.Element | null {
|
||||
const conversation = useConversation(store, conversationId);
|
||||
const remote = Boolean(conversation?.remote);
|
||||
const conversationMode = workModeOverride ?? normalizeInteractiveWorkMode(
|
||||
conversation?.workMode ??
|
||||
projects.find(project => project.id === conversation?.projectId)?.defaultWorkMode,
|
||||
);
|
||||
const selectedTaskId = tasks.some(task => task.id === selectedAssistantTaskId)
|
||||
? selectedAssistantTaskId : undefined;
|
||||
const messages = conversation?.messages;
|
||||
const activeMessageId = conversation?.activeRequest?.messageId;
|
||||
const { locale } = paneProps;
|
||||
const taskStrip = useMemo(() => messages && (
|
||||
<div className="conversation-context-strips">
|
||||
{!remote && (
|
||||
<ConversationTaskStrip
|
||||
conversationMode={conversationMode}
|
||||
locale={locale}
|
||||
onRemoveSchedule={onRemoveSchedule}
|
||||
onRunSchedule={onRunSchedule}
|
||||
onSelectTask={onSelectTask}
|
||||
onSetScheduleEnabled={onSetScheduleEnabled}
|
||||
schedules={schedules}
|
||||
selectedTaskId={selectedTaskId}
|
||||
tasks={tasks}
|
||||
/>
|
||||
)}
|
||||
<RuntimeChecklistStrip messages={messages} activeMessageId={activeMessageId} />
|
||||
</div>
|
||||
), [remote, conversationMode, locale, onRemoveSchedule, onRunSchedule, onSelectTask,
|
||||
onSetScheduleEnabled, schedules, selectedTaskId, tasks, messages, activeMessageId]);
|
||||
if (!conversation) return null;
|
||||
if (conversation.messageSummary) {
|
||||
return (
|
||||
<ConversationHistoryLoader
|
||||
conversationId={conversation.id}
|
||||
active={paneProps.active}
|
||||
load={loadHistory}
|
||||
/>
|
||||
);
|
||||
}
|
||||
return <ChatHistoryPane {...paneProps} conversation={conversation} taskStrip={taskStrip} />;
|
||||
});
|
||||
@@ -0,0 +1,905 @@
|
||||
import type { Dispatch, SetStateAction } from "react";
|
||||
import type { TFunction } from "i18next";
|
||||
import type { AgentEvent } from "../../shared/contracts";
|
||||
import type {
|
||||
AssistantArtifact,
|
||||
AssistantTask,
|
||||
ConversationContextCompressionMarker,
|
||||
ConversationMessageBlock,
|
||||
} from "../../shared/assistant-contracts";
|
||||
import { appendConversationQuestionBlock } from "../../shared/conversation-question-blocks";
|
||||
import { knowledgeReferenceKey } from "../../shared/knowledge-reference";
|
||||
import type { ActivityRecord } from "./activity-store";
|
||||
import type { Message, SubagentActivity, ToolActivity } from "./ChatTimeline";
|
||||
import type { ConversationStore } from "./conversation-store";
|
||||
import type { LiveMessageStore } from "./live-message-store";
|
||||
import type { AppNotificationInput } from "./notifications";
|
||||
|
||||
/**
|
||||
* Applies agent run events (IPC `agent.onEvent`) to the conversation store,
|
||||
* the live-message store and App's task/activity/artifact state.
|
||||
*
|
||||
* Moved out of App unchanged; App passes its current setters and refs. Text
|
||||
* and reasoning deltas go to the live-message store only, so they never
|
||||
* touch App state; everything else is a store or state update, in the order
|
||||
* the events arrive.
|
||||
*/
|
||||
|
||||
export type ActiveRun = {
|
||||
conversationId: string;
|
||||
messageId: string;
|
||||
taskId?: string;
|
||||
projectId?: string;
|
||||
runtimeSelectionKey: string;
|
||||
};
|
||||
|
||||
type Ref<T> = { current: T };
|
||||
|
||||
export type AgentEventDependencies = {
|
||||
activeRuns: Ref<Map<string, ActiveRun>>;
|
||||
activeConversationIdRef: Ref<string>;
|
||||
activeProjectIdRef: Ref<string>;
|
||||
assistantTasksRef: Ref<AssistantTask[]>;
|
||||
hydratingArtifactIds: Ref<Set<string>>;
|
||||
/** Set when a run ends: persist with the next list commit. */
|
||||
requestPersistenceFlush: () => void;
|
||||
tRef: Ref<TFunction<"app">>;
|
||||
conversationStore: ConversationStore;
|
||||
liveMessages: LiveMessageStore;
|
||||
setConversations: ConversationStore["set"];
|
||||
setAssistantTasks: Dispatch<SetStateAction<AssistantTask[]>>;
|
||||
setAssistantArtifacts: Dispatch<SetStateAction<AssistantArtifact[]>>;
|
||||
setActivityRecords: Dispatch<SetStateAction<ActivityRecord[]>>;
|
||||
setUnreadConversationIds: Dispatch<SetStateAction<Set<string>>>;
|
||||
notify: (notification: AppNotificationInput) => void;
|
||||
updateMessage: (conversationId: string, messageId: string, update: (message: Message) => Message) => void;
|
||||
recordActivity: (record: Omit<ActivityRecord, "id" | "createdAt" | "scope">) => void;
|
||||
updateRequestActivity: (requestId: string, status: ActivityRecord["status"], detail?: string) => void;
|
||||
loadWorkspaceChanges: (projectId: string) => Promise<void>;
|
||||
markConversationCompleted: (conversationId: string) => void;
|
||||
setConversationActivity: (conversationId: string, active: boolean) => void;
|
||||
releaseConversationQueueAfterRun: (run: Pick<ActiveRun, "conversationId" | "projectId">) => void;
|
||||
};
|
||||
|
||||
export function mergeArtifacts(
|
||||
current: AssistantArtifact[],
|
||||
incoming: AssistantArtifact[],
|
||||
): AssistantArtifact[] {
|
||||
const merged = new Map(current.map((artifact) => [artifact.id, artifact]));
|
||||
for (const artifact of incoming) {
|
||||
const existing = merged.get(artifact.id);
|
||||
merged.set(artifact.id, {
|
||||
...existing,
|
||||
...artifact,
|
||||
content: artifact.content ?? existing?.content,
|
||||
});
|
||||
}
|
||||
return [...merged.values()].sort((left, right) =>
|
||||
right.createdAt.localeCompare(left.createdAt),
|
||||
);
|
||||
}
|
||||
|
||||
function upsertMessageToolBlock(
|
||||
blocks: ConversationMessageBlock[] | undefined,
|
||||
tool: ToolActivity,
|
||||
): ConversationMessageBlock[] | undefined {
|
||||
if (!blocks) {
|
||||
return blocks;
|
||||
}
|
||||
const callId = tool.callId;
|
||||
const index = callId
|
||||
? blocks.findIndex(
|
||||
(block) => block.type === "tool" && block.tool.callId === callId,
|
||||
)
|
||||
: -1;
|
||||
if (index >= 0) {
|
||||
return blocks.map((block, blockIndex) =>
|
||||
blockIndex === index && block.type === "tool"
|
||||
? { ...block, tool }
|
||||
: block,
|
||||
);
|
||||
}
|
||||
return [
|
||||
...blocks,
|
||||
{
|
||||
id: crypto.randomUUID(),
|
||||
type: "tool",
|
||||
tool,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
function upsertMessageSubagentBlock(
|
||||
blocks: ConversationMessageBlock[] | undefined,
|
||||
childTaskId: string,
|
||||
runtimeCallId?: string,
|
||||
): ConversationMessageBlock[] | undefined {
|
||||
if (!blocks) {
|
||||
return blocks;
|
||||
}
|
||||
if (
|
||||
blocks.some(
|
||||
(block) => block.type === "subagent" && block.childTaskId === childTaskId,
|
||||
)
|
||||
) {
|
||||
return blocks;
|
||||
}
|
||||
const provisionalIndex = runtimeCallId
|
||||
? blocks.findIndex(
|
||||
(block) => block.type === "tool" && block.tool.callId === runtimeCallId,
|
||||
)
|
||||
: -1;
|
||||
if (provisionalIndex >= 0) {
|
||||
return blocks.map((block, index) =>
|
||||
index === provisionalIndex
|
||||
? {
|
||||
id: block.id,
|
||||
type: "subagent" as const,
|
||||
childTaskId,
|
||||
}
|
||||
: block,
|
||||
);
|
||||
}
|
||||
return [
|
||||
...blocks,
|
||||
{
|
||||
id: crypto.randomUUID(),
|
||||
type: "subagent",
|
||||
childTaskId,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
function terminalizeMessageToolBlocks(
|
||||
blocks: ConversationMessageBlock[] | undefined,
|
||||
state: "failed" | "cancelled" | "interrupted",
|
||||
): ConversationMessageBlock[] | undefined {
|
||||
return blocks?.map((block) =>
|
||||
block.type === "tool" &&
|
||||
(block.tool.state === "pending" || block.tool.state === "running")
|
||||
? {
|
||||
...block,
|
||||
tool: {
|
||||
...block.tool,
|
||||
state,
|
||||
},
|
||||
}
|
||||
: block,
|
||||
);
|
||||
}
|
||||
|
||||
function isErrorRepresentedByFailedTool(
|
||||
tools: ToolActivity[] | undefined,
|
||||
errorMessage: string,
|
||||
): boolean {
|
||||
return Boolean(
|
||||
tools?.some(
|
||||
(tool) =>
|
||||
tool.state === "failed" &&
|
||||
((tool.error && errorMessage.includes(tool.error)) ||
|
||||
(tool.callId && errorMessage.includes(tool.callId))),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
export function handleAgentEvent(event: AgentEvent, deps: AgentEventDependencies): void {
|
||||
const {
|
||||
activeRuns, activeConversationIdRef, activeProjectIdRef, assistantTasksRef, hydratingArtifactIds,
|
||||
requestPersistenceFlush, tRef, conversationStore, liveMessages, setConversations,
|
||||
setAssistantTasks, setAssistantArtifacts, setActivityRecords, setUnreadConversationIds, notify,
|
||||
updateMessage, recordActivity, updateRequestActivity, loadWorkspaceChanges, markConversationCompleted,
|
||||
setConversationActivity, releaseConversationQueueAfterRun,
|
||||
} = deps;
|
||||
const run = activeRuns.current.get(event.requestId);
|
||||
if (!run) {
|
||||
if (event.type !== "approval") {
|
||||
return;
|
||||
}
|
||||
const attachScheduledApproval = (
|
||||
task: AssistantTask | undefined,
|
||||
): void => {
|
||||
if (!task?.conversationId || task.origin !== "schedule") {
|
||||
return;
|
||||
}
|
||||
setAssistantTasks((current) =>
|
||||
current.map((candidate) =>
|
||||
candidate.id === task.id
|
||||
? { ...candidate, status: "waiting_approval" }
|
||||
: candidate,
|
||||
),
|
||||
);
|
||||
if (activeConversationIdRef.current !== task.conversationId) {
|
||||
setUnreadConversationIds((current) => {
|
||||
const next = new Set(current);
|
||||
next.add(task.conversationId!);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
setConversations((current) =>
|
||||
current.map((conversation) => {
|
||||
if (conversation.id !== task.conversationId) {
|
||||
return conversation;
|
||||
}
|
||||
const existing = conversation.messages.find(
|
||||
(message) => message.approval?.id === event.approvalId,
|
||||
);
|
||||
if (existing) {
|
||||
return conversation;
|
||||
}
|
||||
return {
|
||||
...conversation,
|
||||
updatedAt: Date.now(),
|
||||
messages: [
|
||||
...conversation.messages,
|
||||
{
|
||||
id: crypto.randomUUID(),
|
||||
role: "assistant",
|
||||
content: event.title,
|
||||
createdAt: Date.now(),
|
||||
state: "complete",
|
||||
task: {
|
||||
id: task.id,
|
||||
title: task.title,
|
||||
},
|
||||
approval: {
|
||||
id: event.approvalId,
|
||||
title: event.title,
|
||||
description: event.description,
|
||||
toolName: event.toolName,
|
||||
argumentSummary: event.argumentSummary,
|
||||
allowPermanent: event.allowPermanent,
|
||||
},
|
||||
},
|
||||
],
|
||||
};
|
||||
}),
|
||||
);
|
||||
};
|
||||
const task = assistantTasksRef.current.find(
|
||||
(candidate) => candidate.id === event.requestId,
|
||||
);
|
||||
if (task) {
|
||||
attachScheduledApproval(task);
|
||||
} else {
|
||||
void window.goodbuddy.tasks
|
||||
.list()
|
||||
.then((tasks) => {
|
||||
setAssistantTasks(tasks);
|
||||
assistantTasksRef.current = tasks;
|
||||
attachScheduledApproval(
|
||||
tasks.find((candidate) => candidate.id === event.requestId),
|
||||
);
|
||||
})
|
||||
.catch(() => undefined);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const liveMessage = conversationStore.getState()
|
||||
.find((conversation) => conversation.id === run.conversationId)
|
||||
?.messages.find((message) => message.id === run.messageId);
|
||||
if (event.type === "question-resolved") {
|
||||
updateMessage(run.conversationId, run.messageId, (message) => {
|
||||
if (!message.pendingQuestions?.some((question) => question.questionId === event.questionId)) return message;
|
||||
const pendingQuestions = message.pendingQuestions.filter((question) => question.questionId !== event.questionId);
|
||||
if (!pendingQuestions.length && !message.approval && message.state === "streaming") {
|
||||
setAssistantTasks((current) => current.map((task) =>
|
||||
task.id === event.requestId && task.status === "waiting_approval" ? { ...task, status: "running" } : task,
|
||||
));
|
||||
}
|
||||
return { ...message, pendingQuestions };
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (event.type === "question" && (
|
||||
liveMessage?.state !== "streaming" ||
|
||||
liveMessage.pendingQuestions?.some((question) => question.questionId === event.questionId) ||
|
||||
liveMessage.answeredQuestions?.some((question) => question.questionId === event.questionId)
|
||||
)) return;
|
||||
setAssistantTasks((current) => {
|
||||
let changed = false;
|
||||
const updated = current.map((task) => {
|
||||
if (task.id !== event.requestId) {
|
||||
return task;
|
||||
}
|
||||
const status: AssistantTask["status"] =
|
||||
event.type === "approval" || event.type === "question"
|
||||
? "waiting_approval"
|
||||
: event.type === "done"
|
||||
? "completed"
|
||||
: event.type === "error"
|
||||
? event.status
|
||||
: liveMessage?.pendingQuestions?.length || liveMessage?.approval
|
||||
? "waiting_approval"
|
||||
: "running";
|
||||
const completedAt =
|
||||
event.type === "done" || event.type === "error"
|
||||
? new Date().toISOString()
|
||||
: task.completedAt;
|
||||
const error = event.type === "error" ? event.message : task.error;
|
||||
if (
|
||||
task.status === status &&
|
||||
task.completedAt === completedAt &&
|
||||
task.error === error
|
||||
) {
|
||||
return task;
|
||||
}
|
||||
changed = true;
|
||||
return {
|
||||
...task,
|
||||
status,
|
||||
completedAt,
|
||||
error,
|
||||
};
|
||||
});
|
||||
return changed ? updated : current;
|
||||
});
|
||||
if (event.type === "done") {
|
||||
if (run.projectId && activeProjectIdRef.current === run.projectId) {
|
||||
void loadWorkspaceChanges(run.projectId).catch(() =>
|
||||
notify({
|
||||
tone: "error",
|
||||
message: tRef.current("notices.workspaceChangesReadFailed"),
|
||||
}),
|
||||
);
|
||||
}
|
||||
void window.goodbuddy.artifacts
|
||||
.list()
|
||||
.then((artifacts) =>
|
||||
setAssistantArtifacts((current) =>
|
||||
mergeArtifacts(current, artifacts),
|
||||
),
|
||||
)
|
||||
.catch(() =>
|
||||
notify({
|
||||
tone: "error",
|
||||
message: tRef.current("notices.resultsRefreshFailed"),
|
||||
}),
|
||||
);
|
||||
} else if (event.type === "artifact") {
|
||||
hydratingArtifactIds.current.add(event.artifactId);
|
||||
void window.goodbuddy.artifacts
|
||||
.get(event.artifactId)
|
||||
.then((artifact) =>
|
||||
setAssistantArtifacts((current) =>
|
||||
mergeArtifacts(current, [artifact]),
|
||||
),
|
||||
)
|
||||
.catch(() =>
|
||||
notify({
|
||||
tone: "error",
|
||||
message: tRef.current("notices.generatedImageReadFailed"),
|
||||
}),
|
||||
)
|
||||
.finally(() => {
|
||||
hydratingArtifactIds.current.delete(event.artifactId);
|
||||
});
|
||||
}
|
||||
|
||||
if (event.type === "checklist") {
|
||||
updateMessage(run.conversationId, run.messageId, (message) =>
|
||||
message.state === "streaming"
|
||||
? { ...message, runtimeChecklist: event.checklist }
|
||||
: message,
|
||||
);
|
||||
} else if (event.type === "text" || event.type === "reasoning") {
|
||||
// Rendered by the message row from the live store; App state absorbs
|
||||
// the deltas later (see live-message-store.ts).
|
||||
liveMessages.append(
|
||||
run.conversationId,
|
||||
run.messageId,
|
||||
event.type,
|
||||
event.delta,
|
||||
liveMessage,
|
||||
);
|
||||
} else if (event.type === "context-metrics") {
|
||||
setConversations((current) =>
|
||||
current.map((conversation) =>
|
||||
conversation.id === run.conversationId
|
||||
? {
|
||||
...conversation,
|
||||
contextMetrics: {
|
||||
contextTokens: event.contextTokens,
|
||||
source: event.source,
|
||||
basis: "model-call",
|
||||
runtimeSelectionKey: run.runtimeSelectionKey,
|
||||
},
|
||||
}
|
||||
: conversation,
|
||||
),
|
||||
);
|
||||
} else if (event.type === "context-compression") {
|
||||
const estimatedAfterTokens = event.estimatedAfterTokens;
|
||||
const conversationScoped = event.scope !== "agent-run";
|
||||
const scope = event.scope ?? "conversation";
|
||||
const marker: ConversationContextCompressionMarker = {
|
||||
state: event.state === "started" ? "compressing" : event.state,
|
||||
scope,
|
||||
estimatedBeforeTokens: event.estimatedBeforeTokens,
|
||||
estimatedAfterTokens: event.estimatedAfterTokens,
|
||||
compressionCount: event.compressionCount,
|
||||
};
|
||||
updateMessage(run.conversationId, run.messageId, (message) => {
|
||||
const current =
|
||||
message.contextCompressions ??
|
||||
(message.contextCompression ? [message.contextCompression] : []);
|
||||
const existingIndex = current.findIndex(
|
||||
(compression) => (compression.scope ?? "conversation") === scope,
|
||||
);
|
||||
const contextCompressions =
|
||||
existingIndex >= 0
|
||||
? [
|
||||
...current.filter(
|
||||
(_compression, index) => index !== existingIndex,
|
||||
),
|
||||
marker,
|
||||
]
|
||||
: [...current, marker];
|
||||
return {
|
||||
...message,
|
||||
contextCompression: undefined,
|
||||
contextCompressions,
|
||||
};
|
||||
});
|
||||
if (
|
||||
conversationScoped &&
|
||||
event.state === "completed" &&
|
||||
estimatedAfterTokens !== undefined
|
||||
) {
|
||||
setConversations((current) =>
|
||||
current.map((conversation) =>
|
||||
conversation.id === run.conversationId
|
||||
? {
|
||||
...conversation,
|
||||
contextMetrics: {
|
||||
runtimeSelectionKey: run.runtimeSelectionKey,
|
||||
contextTokens: estimatedAfterTokens,
|
||||
source: "estimated",
|
||||
basis: "conversation",
|
||||
},
|
||||
contextCompressionState:
|
||||
event.conversationState ??
|
||||
conversation.contextCompressionState,
|
||||
}
|
||||
: conversation,
|
||||
),
|
||||
);
|
||||
} else if (conversationScoped && event.conversationState) {
|
||||
setConversations((current) =>
|
||||
current.map((conversation) =>
|
||||
conversation.id === run.conversationId
|
||||
? {
|
||||
...conversation,
|
||||
contextCompressionState: event.conversationState,
|
||||
}
|
||||
: conversation,
|
||||
),
|
||||
);
|
||||
}
|
||||
} else if (event.type === "status") {
|
||||
updateMessage(run.conversationId, run.messageId, (message) => ({
|
||||
...message,
|
||||
status: event.message,
|
||||
}));
|
||||
} else if (event.type === "tool") {
|
||||
recordActivity({
|
||||
conversationId: run.conversationId,
|
||||
requestId: event.requestId,
|
||||
callId: event.callId.slice(0, 256),
|
||||
kind: "tool",
|
||||
title: event.name,
|
||||
detail: [event.summary, event.error].filter(Boolean).join("\n"),
|
||||
status:
|
||||
event.state === "pending"
|
||||
? "pending"
|
||||
: event.state === "running"
|
||||
? "running"
|
||||
: event.state === "failed"
|
||||
? "failed"
|
||||
: "completed",
|
||||
});
|
||||
updateMessage(run.conversationId, run.messageId, (message) => {
|
||||
const tools = [...(message.tools ?? [])];
|
||||
const index = tools.findIndex(
|
||||
(tool) => tool.callId === event.callId.slice(0, 256),
|
||||
);
|
||||
const tool = {
|
||||
callId: event.callId.slice(0, 256),
|
||||
name: event.name,
|
||||
state: event.state,
|
||||
summary: event.summary,
|
||||
input: event.input,
|
||||
output: event.output,
|
||||
error: event.error,
|
||||
};
|
||||
if (index >= 0) {
|
||||
tools[index] = tool;
|
||||
} else {
|
||||
tools.push(tool);
|
||||
}
|
||||
const blocks = upsertMessageToolBlock(message.blocks, tool);
|
||||
return {
|
||||
...message,
|
||||
tools,
|
||||
blocks,
|
||||
status: undefined,
|
||||
};
|
||||
});
|
||||
} else if (event.type === "subagent") {
|
||||
const actor =
|
||||
"actor" in event
|
||||
? event.actor
|
||||
: {
|
||||
kind: "expert" as const,
|
||||
expertId: event.expertId,
|
||||
expertName: event.expertName,
|
||||
};
|
||||
const actorLabel =
|
||||
actor.kind === "direct-model"
|
||||
? tRef.current("chat.subagents.directModelLabel")
|
||||
: actor.expertName;
|
||||
const childStatus = event.state;
|
||||
const completedAt =
|
||||
event.state === "completed" ||
|
||||
event.state === "failed" ||
|
||||
event.state === "cancelled"
|
||||
? new Date().toISOString()
|
||||
: undefined;
|
||||
if (actor.kind === "expert") {
|
||||
setAssistantTasks((current) => {
|
||||
const existing = current.find(
|
||||
(task) => task.id === event.childTaskId,
|
||||
);
|
||||
const childTask: AssistantTask = {
|
||||
id: event.childTaskId,
|
||||
projectId: run.projectId,
|
||||
conversationId: run.conversationId,
|
||||
parentTaskId: event.requestId,
|
||||
expertId: actor.expertId,
|
||||
routingMode:
|
||||
event.routingMode === "native" ? undefined : event.routingMode,
|
||||
title: actor.expertName,
|
||||
instructions:
|
||||
event.reason ??
|
||||
tRef.current("chat.subagents.fallbackTask", {
|
||||
name: actor.expertName,
|
||||
}),
|
||||
origin: "subagent",
|
||||
status: childStatus,
|
||||
createdAt: existing?.createdAt ?? new Date().toISOString(),
|
||||
startedAt:
|
||||
event.state === "running"
|
||||
? (existing?.startedAt ?? new Date().toISOString())
|
||||
: existing?.startedAt,
|
||||
completedAt: completedAt ?? existing?.completedAt,
|
||||
error: event.error,
|
||||
};
|
||||
if (
|
||||
existing &&
|
||||
(Object.keys(childTask) as (keyof AssistantTask)[]).every(
|
||||
(key) => existing[key] === childTask[key],
|
||||
)
|
||||
) {
|
||||
return current;
|
||||
}
|
||||
return existing
|
||||
? current.map((task) =>
|
||||
task.id === event.childTaskId ? childTask : task,
|
||||
)
|
||||
: [...current, childTask];
|
||||
});
|
||||
}
|
||||
if (event.runtimeCallId) {
|
||||
setActivityRecords((current) => {
|
||||
const remaining = current.filter(
|
||||
(record) =>
|
||||
!(
|
||||
record.requestId === event.requestId &&
|
||||
record.kind === "tool" &&
|
||||
record.callId === event.runtimeCallId
|
||||
),
|
||||
);
|
||||
return remaining.length === current.length ? current : remaining;
|
||||
});
|
||||
}
|
||||
recordActivity({
|
||||
conversationId: run.conversationId,
|
||||
requestId: event.requestId,
|
||||
callId: event.childTaskId,
|
||||
kind: "subagent",
|
||||
title: actorLabel,
|
||||
detail: [
|
||||
actor.kind === "direct-model"
|
||||
? tRef.current("chat.subagents.directModel")
|
||||
: event.routingMode === "smart"
|
||||
? tRef.current("chat.subagents.smart")
|
||||
: event.routingMode === "native"
|
||||
? tRef.current("chat.subagents.native")
|
||||
: tRef.current("chat.subagents.manual"),
|
||||
actor.kind === "direct-model" && event.workMode
|
||||
? event.workMode === "execute"
|
||||
? "Execute"
|
||||
: "Ask"
|
||||
: undefined,
|
||||
event.reason,
|
||||
event.error,
|
||||
]
|
||||
.filter(Boolean)
|
||||
.join(" · "),
|
||||
status: event.state === "queued" ? "pending" : event.state,
|
||||
});
|
||||
updateMessage(run.conversationId, run.messageId, (message) => {
|
||||
const subagents = [...(message.subagents ?? [])];
|
||||
const index = subagents.findIndex(
|
||||
(subagent) => subagent.childTaskId === event.childTaskId,
|
||||
);
|
||||
const commonSubagent = {
|
||||
childTaskId: event.childTaskId,
|
||||
routingMode: event.routingMode,
|
||||
runtimeCallId: event.runtimeCallId,
|
||||
workMode: event.workMode,
|
||||
state: event.state,
|
||||
reason: event.reason,
|
||||
progress: event.progress,
|
||||
output: event.output,
|
||||
error: event.error,
|
||||
};
|
||||
const subagent: SubagentActivity =
|
||||
"actor" in event
|
||||
? { ...commonSubagent, actor: event.actor }
|
||||
: {
|
||||
...commonSubagent,
|
||||
expertId: event.expertId,
|
||||
expertName: event.expertName,
|
||||
};
|
||||
if (index >= 0) {
|
||||
subagents[index] = subagent;
|
||||
} else {
|
||||
subagents.push(subagent);
|
||||
}
|
||||
const runtimeCallId = event.runtimeCallId;
|
||||
const blocks = upsertMessageSubagentBlock(
|
||||
message.blocks,
|
||||
event.childTaskId,
|
||||
runtimeCallId,
|
||||
);
|
||||
return {
|
||||
...message,
|
||||
subagents,
|
||||
tools: runtimeCallId
|
||||
? message.tools?.filter((tool) => tool.callId !== runtimeCallId)
|
||||
: message.tools,
|
||||
blocks,
|
||||
};
|
||||
});
|
||||
} else if (event.type === "approval") {
|
||||
recordActivity({
|
||||
conversationId: run.conversationId,
|
||||
requestId: event.requestId,
|
||||
kind: "approval",
|
||||
title: event.title,
|
||||
detail: event.description,
|
||||
status: "pending",
|
||||
});
|
||||
updateMessage(run.conversationId, run.messageId, (message) => ({
|
||||
...message,
|
||||
status: undefined,
|
||||
approval: {
|
||||
id: event.approvalId,
|
||||
taskId: run.taskId,
|
||||
title: event.title,
|
||||
description: event.description,
|
||||
toolName: event.toolName,
|
||||
argumentSummary: event.argumentSummary,
|
||||
allowPermanent: event.allowPermanent,
|
||||
},
|
||||
}));
|
||||
} else if (event.type === "question") {
|
||||
updateMessage(run.conversationId, run.messageId, (message) => {
|
||||
if (message.state !== "streaming" ||
|
||||
message.pendingQuestions?.some((question) => question.questionId === event.questionId) ||
|
||||
message.answeredQuestions?.some((question) => question.questionId === event.questionId)) {
|
||||
return message;
|
||||
}
|
||||
return {
|
||||
...message,
|
||||
status: undefined,
|
||||
blocks: appendConversationQuestionBlock(message.blocks, event.questionId),
|
||||
pendingQuestions: [...(message.pendingQuestions ?? []), event],
|
||||
};
|
||||
});
|
||||
} else if (event.type === "artifact") {
|
||||
updateMessage(run.conversationId, run.messageId, (message) => ({
|
||||
...message,
|
||||
artifactIds: [
|
||||
...new Set([...(message.artifactIds ?? []), event.artifactId]),
|
||||
].slice(-8),
|
||||
imageContextNotice: event.imageContextNotice,
|
||||
status: tRef.current("chat.status.savingImage"),
|
||||
}));
|
||||
} else if (event.type === "knowledge-retrieval") {
|
||||
updateMessage(run.conversationId, run.messageId, (message) => ({
|
||||
...message,
|
||||
knowledgeRetrieval: {
|
||||
mode: event.mode,
|
||||
state: event.state,
|
||||
libraryCount: event.libraryCount,
|
||||
resultCount: event.resultCount,
|
||||
durationMs: event.durationMs,
|
||||
usedChannels: event.usedChannels,
|
||||
warnings: event.warnings,
|
||||
},
|
||||
status:
|
||||
event.state === "searching"
|
||||
? tRef.current("chat.knowledgeRetrieval.searching")
|
||||
: undefined,
|
||||
}));
|
||||
} else if (event.type === "source-references") {
|
||||
updateMessage(run.conversationId, run.messageId, (message) => {
|
||||
const incoming = [
|
||||
...new Map(
|
||||
event.references.map((reference) => [
|
||||
knowledgeReferenceKey(reference),
|
||||
reference,
|
||||
]),
|
||||
).values(),
|
||||
];
|
||||
const incomingKeys = new Set(incoming.map(knowledgeReferenceKey));
|
||||
const references = [
|
||||
...incoming,
|
||||
...(message.sourceReferences ?? []).filter(
|
||||
(reference) => !incomingKeys.has(knowledgeReferenceKey(reference)),
|
||||
),
|
||||
].slice(0, 20);
|
||||
return {
|
||||
...message,
|
||||
sourceReferences: references,
|
||||
};
|
||||
});
|
||||
} else if (event.type === "done" || event.type === "error") {
|
||||
const terminalStatus =
|
||||
event.type === "error"
|
||||
? event.status === "cancelled"
|
||||
? "cancelled"
|
||||
: "failed"
|
||||
: "completed";
|
||||
const incompleteSubagentError = tRef.current(
|
||||
"chat.subagents.incomplete",
|
||||
);
|
||||
const incompleteActivityDetail = tRef.current(
|
||||
"chat.status.activityIncomplete",
|
||||
);
|
||||
updateRequestActivity(
|
||||
event.requestId,
|
||||
terminalStatus,
|
||||
event.type === "error"
|
||||
? event.message
|
||||
: tRef.current("chat.status.taskCompleted"),
|
||||
);
|
||||
setActivityRecords((current) =>
|
||||
current.map((record) =>
|
||||
record.requestId === event.requestId &&
|
||||
record.kind !== "request" &&
|
||||
(record.status === "pending" || record.status === "running")
|
||||
? {
|
||||
...record,
|
||||
status:
|
||||
event.type === "done" ? "interrupted" : terminalStatus,
|
||||
detail: `${record.detail}\n${
|
||||
event.type === "done"
|
||||
? incompleteActivityDetail
|
||||
: event.message
|
||||
}`,
|
||||
}
|
||||
: record,
|
||||
),
|
||||
);
|
||||
recordActivity({
|
||||
conversationId: run.conversationId,
|
||||
requestId: event.requestId,
|
||||
kind: "result",
|
||||
title:
|
||||
event.type === "error"
|
||||
? tRef.current(event.status === "cancelled"
|
||||
? "chat.status.taskCancelled" : "chat.status.taskFailed")
|
||||
: tRef.current("chat.status.taskCompleted"),
|
||||
detail:
|
||||
event.type === "error"
|
||||
? event.message
|
||||
: tRef.current("chat.status.runtimeCompleted"),
|
||||
status: terminalStatus,
|
||||
});
|
||||
setAssistantTasks((current) =>
|
||||
current.map((task) =>
|
||||
task.parentTaskId === event.requestId &&
|
||||
(task.status === "queued" || task.status === "running")
|
||||
? {
|
||||
...task,
|
||||
status: event.type === "done" ? "failed" : terminalStatus,
|
||||
completedAt: new Date().toISOString(),
|
||||
error:
|
||||
event.type === "error"
|
||||
? event.message
|
||||
: incompleteSubagentError,
|
||||
}
|
||||
: task,
|
||||
),
|
||||
);
|
||||
updateMessage(run.conversationId, run.messageId, (message) => {
|
||||
const representedToolError =
|
||||
event.type === "error" &&
|
||||
isErrorRepresentedByFailedTool(message.tools, event.message);
|
||||
const toolTerminalState =
|
||||
event.type === "error"
|
||||
? event.status === "cancelled"
|
||||
? ("cancelled" as const)
|
||||
: ("failed" as const)
|
||||
: ("interrupted" as const);
|
||||
return {
|
||||
...message,
|
||||
state: event.type === "error" ? "error" : "complete",
|
||||
terminalStatus: event.type === "error"
|
||||
? event.status === "cancelled" ? "cancelled" : "failed"
|
||||
: undefined,
|
||||
status:
|
||||
event.type === "error" && (event.status === "cancelled" || !representedToolError)
|
||||
? event.message
|
||||
: event.type === "done"
|
||||
? tRef.current("chat.status.taskCompleted")
|
||||
: undefined,
|
||||
contextCompression:
|
||||
event.type === "error" &&
|
||||
message.contextCompression?.state === "compressing"
|
||||
? {
|
||||
...message.contextCompression,
|
||||
state: "failed" as const,
|
||||
}
|
||||
: message.contextCompression,
|
||||
contextCompressions:
|
||||
event.type === "error"
|
||||
? message.contextCompressions?.map((compression) =>
|
||||
compression.state === "compressing"
|
||||
? {
|
||||
...compression,
|
||||
state: "failed" as const,
|
||||
}
|
||||
: compression,
|
||||
)
|
||||
: message.contextCompressions,
|
||||
approval: undefined,
|
||||
pendingQuestions: undefined,
|
||||
tools: toolTerminalState
|
||||
? message.tools?.map((tool) =>
|
||||
tool.state === "pending" || tool.state === "running"
|
||||
? { ...tool, state: toolTerminalState }
|
||||
: tool,
|
||||
)
|
||||
: message.tools,
|
||||
subagents: message.subagents?.map((subagent) =>
|
||||
subagent.state === "queued" || subagent.state === "running"
|
||||
? {
|
||||
...subagent,
|
||||
state:
|
||||
event.type === "error"
|
||||
? event.status === "cancelled"
|
||||
? ("cancelled" as const)
|
||||
: ("failed" as const)
|
||||
: ("failed" as const),
|
||||
...(event.type === "error" && event.status !== "cancelled"
|
||||
? { error: event.message.slice(0, 1_000) }
|
||||
: event.type === "done"
|
||||
? { error: incompleteSubagentError }
|
||||
: {}),
|
||||
}
|
||||
: subagent,
|
||||
),
|
||||
blocks: terminalizeMessageToolBlocks(message.blocks, toolTerminalState),
|
||||
};
|
||||
});
|
||||
activeRuns.current.delete(event.requestId);
|
||||
if (event.type === "done") markConversationCompleted(run.conversationId);
|
||||
setConversationActivity(run.conversationId, false);
|
||||
releaseConversationQueueAfterRun(run);
|
||||
requestPersistenceFlush();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,242 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import type { ConversationListSnapshot, LocalConversationSaveBatch } from "../../shared/assistant-contracts";
|
||||
import type { Message } from "./ChatTimeline";
|
||||
import type { Conversation } from "./chat-conversation";
|
||||
import { createConversationPersistence } from "./conversation-persistence";
|
||||
import { startConversationRefresh, type ConversationRefreshApi } from "./conversation-refresh";
|
||||
import { createConversationStore } from "./conversation-store";
|
||||
import { createLiveMessageStore } from "./live-message-store";
|
||||
|
||||
function message(id: string, overrides: Partial<Message> = {}): Message {
|
||||
return { id, role: "assistant", content: id, createdAt: 1, state: "complete", ...overrides } as Message;
|
||||
}
|
||||
|
||||
function conversation(id: string, messages: Message[] = [message(`${id}-m`)], overrides: Partial<Conversation> = {}): Conversation {
|
||||
return { id, title: id, updatedAt: 1, messages, ...overrides } as Conversation;
|
||||
}
|
||||
|
||||
function deferred<T = void>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((done) => { resolve = done; });
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
const remoteSource = { channel: "wecom", accountDisplay: "bot", conversationType: "direct" } as const;
|
||||
|
||||
type SaveLocal = (batch: LocalConversationSaveBatch) => Promise<unknown>;
|
||||
|
||||
function setup(initial: Conversation[], saveLocal = vi.fn<SaveLocal>(async () => undefined)) {
|
||||
const store = createConversationStore(initial, createLiveMessageStore());
|
||||
const get = vi.fn(async (id: string) => ({ ...conversation(id), updatedAt: 2 }) as ConversationListSnapshot);
|
||||
const persistence = createConversationPersistence({ store, api: { saveLocal, get }, intervalMs: 500 });
|
||||
return { store, persistence, saveLocal, get };
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
describe("conversation persistence", () => {
|
||||
it("saves only changed local conversations and only their changed messages", async () => {
|
||||
const remote = conversation("r", undefined, { remote: remoteSource });
|
||||
const { store, persistence, saveLocal } = setup([conversation("a"), conversation("b"), remote]);
|
||||
persistence.persist();
|
||||
await persistence.idle();
|
||||
expect(saveLocal).toHaveBeenCalledTimes(1);
|
||||
expect(saveLocal.mock.calls[0]![0].map((entry) => entry.header.id)).toEqual(["a", "b"]);
|
||||
|
||||
store.set((current) => current.map((item) => item.id === "a"
|
||||
? { ...item, updatedAt: 2, messages: [...item.messages, message("a-new")] }
|
||||
: item));
|
||||
persistence.persist();
|
||||
await persistence.idle();
|
||||
expect(saveLocal).toHaveBeenCalledTimes(2);
|
||||
const [entry] = saveLocal.mock.calls[1]![0];
|
||||
expect(entry!.header.id).toBe("a");
|
||||
expect(entry!.messages.map((item) => item.id)).toEqual(["a-new"]);
|
||||
|
||||
persistence.persist();
|
||||
await persistence.idle();
|
||||
expect(saveLocal).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("serializes saves with queued tasks and skips paused or deleting conversations", async () => {
|
||||
const gate = deferred();
|
||||
const order: string[] = [];
|
||||
const saveLocal = vi.fn<SaveLocal>(async (batch) => {
|
||||
order.push(`save:${batch.map((entry) => entry.header.id).join(",")}`);
|
||||
await gate.promise;
|
||||
});
|
||||
const { store, persistence } = setup([conversation("a"), conversation("b")], saveLocal);
|
||||
persistence.markDeleting("b", true);
|
||||
persistence.persist();
|
||||
const queued = persistence.enqueue(async () => { order.push("task"); });
|
||||
await vi.waitFor(() => expect(order).toEqual(["save:a"]));
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
// The task waits for the pending save.
|
||||
expect(order).toEqual(["save:a"]);
|
||||
gate.resolve();
|
||||
await queued;
|
||||
expect(order).toEqual(["save:a", "task"]);
|
||||
|
||||
persistence.setPaused(true);
|
||||
store.set((current) => current.map((item) => ({ ...item, updatedAt: 3 })));
|
||||
persistence.persist();
|
||||
await persistence.idle();
|
||||
expect(saveLocal).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("saves on the interval, after a requested commit flush and once more on quit", async () => {
|
||||
vi.useFakeTimers();
|
||||
const { store, persistence, saveLocal } = setup([conversation("a")]);
|
||||
persistence.flushAfterCommit();
|
||||
persistence.requestFlushAfterCommit();
|
||||
persistence.flushAfterCommit();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
// Not started yet: a commit flush waits for the store to be ready.
|
||||
expect(saveLocal).not.toHaveBeenCalled();
|
||||
|
||||
const stop = persistence.start();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(saveLocal).toHaveBeenCalledTimes(1);
|
||||
|
||||
store.set((current) => current.map((item) => ({ ...item, updatedAt: 2 })));
|
||||
await vi.advanceTimersByTimeAsync(500);
|
||||
expect(saveLocal).toHaveBeenCalledTimes(2);
|
||||
|
||||
store.set((current) => current.map((item) => ({ ...item, updatedAt: 3 })));
|
||||
persistence.requestFlushAfterCommit();
|
||||
persistence.flushAfterCommit();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(saveLocal).toHaveBeenCalledTimes(3);
|
||||
|
||||
store.set((current) => current.map((item) => ({ ...item, updatedAt: 4 })));
|
||||
await persistence.flushForQuit();
|
||||
expect(saveLocal).toHaveBeenCalledTimes(4);
|
||||
expect(saveLocal.mock.calls[3]![0][0]!.header.updatedAt).toBe(4);
|
||||
stop();
|
||||
});
|
||||
|
||||
it("releases acknowledged idle history that is not retained", async () => {
|
||||
const { store, persistence } = setup([conversation("a"), conversation("b"), conversation("c", [message("c-m", { state: "streaming" })])]);
|
||||
persistence.connect({
|
||||
retainedConversationIds: () => new Set(["a"]),
|
||||
hasBusyTask: () => false,
|
||||
pinState: () => ({ revision: 0, pending: false }),
|
||||
historyUnavailableError: () => new Error("unavailable"),
|
||||
onSaveFailed: () => undefined,
|
||||
});
|
||||
persistence.persist();
|
||||
await persistence.idle();
|
||||
const byId = new Map(store.getState().map((item) => [item.id, item]));
|
||||
expect(byId.get("a")!.messageSummary).toBeUndefined();
|
||||
expect(byId.get("b")!.messageSummary).toEqual({ count: 1, firstRole: "assistant" });
|
||||
expect(byId.get("b")!.messages).toEqual([]);
|
||||
expect(byId.get("c")!.messageSummary).toBeUndefined();
|
||||
});
|
||||
|
||||
it("loads summarized history once and merges it into the store", async () => {
|
||||
const summary = { ...conversation("a", []), messageSummary: { count: 1 } } as Conversation;
|
||||
const { store, persistence, get } = setup([summary]);
|
||||
const [first, second] = await Promise.all([persistence.ensureHistory("a"), persistence.ensureHistory("a")]);
|
||||
expect(get).toHaveBeenCalledTimes(1);
|
||||
expect(first).toBe(second);
|
||||
expect(store.getConversation("a")!.messageSummary).toBeUndefined();
|
||||
expect(store.getConversation("a")!.messages.map((item) => item.id)).toEqual(["a-m"]);
|
||||
expect(await persistence.ensureHistory("a")).toBe(store.getConversation("a"));
|
||||
});
|
||||
|
||||
it("reports a failed save", async () => {
|
||||
const onSaveFailed = vi.fn();
|
||||
const { persistence } = setup([conversation("a")], vi.fn(async () => { throw new Error("disk"); }));
|
||||
persistence.connect({
|
||||
retainedConversationIds: () => new Set(),
|
||||
hasBusyTask: () => false,
|
||||
pinState: () => ({ revision: 0, pending: false }),
|
||||
historyUnavailableError: () => new Error("unavailable"),
|
||||
onSaveFailed,
|
||||
});
|
||||
persistence.persist();
|
||||
await persistence.idle();
|
||||
await Promise.resolve();
|
||||
expect(onSaveFailed).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("conversation refresh", () => {
|
||||
function refreshApi(summaries: () => ConversationListSnapshot[]) {
|
||||
let changed: (() => void) | undefined;
|
||||
const api: ConversationRefreshApi = {
|
||||
listTasks: vi.fn(async () => []),
|
||||
listSchedules: vi.fn(async () => []),
|
||||
listSummaries: vi.fn(async () => summaries()),
|
||||
onConversationsChanged: vi.fn((listener) => { changed = listener; return () => { changed = undefined; }; }),
|
||||
onQueueChanged: vi.fn(() => () => undefined),
|
||||
};
|
||||
return { api, emit: () => changed?.() };
|
||||
}
|
||||
|
||||
it("merges summaries on change notifications, marks remote updates unread and settles finished runs", async () => {
|
||||
const { store, persistence } = setup([
|
||||
conversation("local", [message("l1", { state: "streaming" })], { updatedAt: 1 }),
|
||||
conversation("remote", undefined, { updatedAt: 1, remote: remoteSource }),
|
||||
]);
|
||||
let persisted: ConversationListSnapshot[] = [];
|
||||
const { api, emit } = refreshApi(() => persisted);
|
||||
const activeRuns = new Map([["req", { conversationId: "local", messageId: "l1" }]]);
|
||||
const onUnread = vi.fn();
|
||||
const onRunSettled = vi.fn();
|
||||
const onTasks = vi.fn();
|
||||
const stop = startConversationRefresh({
|
||||
store, persistence, api, activeRuns,
|
||||
activeConversationId: () => "local",
|
||||
retainedConversationIds: () => new Set(["local"]),
|
||||
pinState: () => ({ revision: 0, pending: false }),
|
||||
onTasks, onSchedules: vi.fn(), onUnread, onRunSettled, onError: vi.fn(), delayMs: 0,
|
||||
});
|
||||
persisted = [
|
||||
{ id: "local", title: "Local", updatedAt: 1, messages: [message("l1", { content: "done", state: "complete" })] },
|
||||
{ id: "remote", title: "Remote", updatedAt: 5, remote: remoteSource, messages: [message("r1")] },
|
||||
{ id: "new", title: "New", updatedAt: 3, messages: [message("n1")] },
|
||||
] as ConversationListSnapshot[];
|
||||
emit();
|
||||
emit();
|
||||
await vi.waitFor(() => expect(onRunSettled).toHaveBeenCalled());
|
||||
expect(api.listSummaries).toHaveBeenCalledTimes(1);
|
||||
expect(api.listSummaries).toHaveBeenCalledWith(["local"]);
|
||||
expect(onTasks).toHaveBeenCalledWith([]);
|
||||
expect(onUnread.mock.calls[0]![0].map((item: ConversationListSnapshot) => item.id)).toEqual(["remote"]);
|
||||
expect(onRunSettled).toHaveBeenCalledWith("req", { conversationId: "local", messageId: "l1" }, "complete");
|
||||
expect(activeRuns.size).toBe(0);
|
||||
const byId = new Map(store.getState().map((item) => [item.id, item]));
|
||||
expect(byId.get("local")!.messages[0]!.content).toBe("done");
|
||||
expect(byId.get("remote")!.title).toBe("Remote");
|
||||
expect(byId.get("new")!.title).toBe("New");
|
||||
// The persisted copy is acknowledged, so it is not saved back.
|
||||
expect(persistence.acknowledged().get("local")).toBe(persisted[0]);
|
||||
stop();
|
||||
expect(api.onConversationsChanged).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("drops a reply when a pin edit changed meanwhile and refreshes again", async () => {
|
||||
const { store, persistence } = setup([conversation("a")]);
|
||||
const reply = deferred<ConversationListSnapshot[]>();
|
||||
const { api, emit } = refreshApi(() => []);
|
||||
vi.mocked(api.listSummaries).mockReturnValueOnce(reply.promise);
|
||||
let revision = 0;
|
||||
const stop = startConversationRefresh({
|
||||
store, persistence, api, activeRuns: new Map(),
|
||||
activeConversationId: () => "a",
|
||||
retainedConversationIds: () => new Set(),
|
||||
pinState: () => ({ revision, pending: false }),
|
||||
onTasks: vi.fn(), onSchedules: vi.fn(), onUnread: vi.fn(), onRunSettled: vi.fn(), onError: vi.fn(), delayMs: 0,
|
||||
});
|
||||
emit();
|
||||
await vi.waitFor(() => expect(api.listSummaries).toHaveBeenCalledTimes(1));
|
||||
revision = 1;
|
||||
reply.resolve([{ id: "a", title: "Stale", updatedAt: 9, messages: [] } as ConversationListSnapshot]);
|
||||
await vi.waitFor(() => expect(api.listSummaries).toHaveBeenCalledTimes(2));
|
||||
expect(store.getConversation("a")!.title).toBe("a");
|
||||
stop();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,425 @@
|
||||
import type {
|
||||
ConversationListSnapshot,
|
||||
ConversationMessage,
|
||||
ConversationSnapshot,
|
||||
LocalConversationHeader,
|
||||
LocalConversationSaveBatch,
|
||||
} from "../../shared/assistant-contracts";
|
||||
import type { Message } from "./ChatTimeline";
|
||||
import { sortConversationsForDisplay, type Conversation } from "./chat-conversation";
|
||||
import type { ConversationStore } from "./conversation-store";
|
||||
import { mergeMessageImageState } from "./message-image-state";
|
||||
|
||||
/**
|
||||
* Local conversation persistence and history loading, outside React.
|
||||
*
|
||||
* Works against the conversation store directly (getState/set), so App never
|
||||
* re-renders for its bookkeeping: the acknowledged snapshots, the serialized
|
||||
* save queue, in-flight history requests and the "persist after commit" flag
|
||||
* all live here. App keeps aliases of the ref-like fields for the few
|
||||
* operations (delete, branch, clear data) that coordinate with the queue.
|
||||
*/
|
||||
|
||||
export function toConversationMessage(message: Message): ConversationMessage {
|
||||
return {
|
||||
id: message.id,
|
||||
queueItemId: message.queueItemId,
|
||||
role: message.role,
|
||||
content: message.content,
|
||||
reasoning: message.reasoning,
|
||||
blocks: message.blocks,
|
||||
displayCaptureTruncated: message.displayCaptureTruncated,
|
||||
createdAt: message.createdAt,
|
||||
state: message.state,
|
||||
terminalStatus: message.terminalStatus,
|
||||
status: message.status,
|
||||
runtimeChecklist: message.runtimeChecklist,
|
||||
contextCompression: message.contextCompression,
|
||||
contextCompressions: message.contextCompressions,
|
||||
tools: message.tools,
|
||||
subagents: message.subagents,
|
||||
sources: message.sources,
|
||||
sourceReferences: message.sourceReferences,
|
||||
knowledgeRetrieval: message.knowledgeRetrieval,
|
||||
artifactIds: message.artifactIds,
|
||||
imageOperations: message.imageOperations,
|
||||
imageSourceArtifactIds: message.imageSourceArtifactIds,
|
||||
imageContextNotice: message.imageContextNotice,
|
||||
task: message.task,
|
||||
attachments: message.attachments,
|
||||
answeredQuestions: message.answeredQuestions,
|
||||
};
|
||||
}
|
||||
|
||||
export function toLocalConversationHeader(
|
||||
conversation: Conversation,
|
||||
): LocalConversationHeader {
|
||||
return {
|
||||
id: conversation.id,
|
||||
projectId: conversation.projectId,
|
||||
runtimeSelection: conversation.runtimeSelection,
|
||||
workMode: conversation.workMode,
|
||||
knowledgeLibraryIds: conversation.knowledgeLibraryIds,
|
||||
knowledgeRetrievalMode: conversation.knowledgeRetrievalMode,
|
||||
contextMetrics: conversation.contextMetrics,
|
||||
contextCompressionState: conversation.contextCompressionState,
|
||||
...(conversation.branch ? { branch: conversation.branch } : {}),
|
||||
title: conversation.title,
|
||||
updatedAt: conversation.updatedAt,
|
||||
};
|
||||
}
|
||||
|
||||
export function createLocalConversationSaveBatch(
|
||||
conversations: readonly Conversation[],
|
||||
persisted: ReadonlyMap<string, Conversation>,
|
||||
deletingConversationIds: ReadonlySet<string>,
|
||||
): {
|
||||
batch: LocalConversationSaveBatch;
|
||||
acknowledgements: Conversation[];
|
||||
} {
|
||||
const batch: LocalConversationSaveBatch = [];
|
||||
const acknowledgements: Conversation[] = [];
|
||||
for (const conversation of conversations) {
|
||||
if (
|
||||
conversation.remote ||
|
||||
deletingConversationIds.has(conversation.id) ||
|
||||
persisted.get(conversation.id) === conversation
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const previous = persisted.get(conversation.id);
|
||||
const previousMessages = new Map(
|
||||
previous?.messages.map((message) => [message.id, message]) ?? [],
|
||||
);
|
||||
batch.push({
|
||||
header: toLocalConversationHeader(conversation),
|
||||
messages: conversation.messages
|
||||
.filter((message) => previousMessages.get(message.id) !== message)
|
||||
.map(toConversationMessage),
|
||||
});
|
||||
acknowledgements.push(conversation);
|
||||
if (batch.length === 100) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return { batch, acknowledgements };
|
||||
}
|
||||
|
||||
/**
|
||||
* The persisted store is the authority for a message's terminal state.
|
||||
* A locally streaming message (or a locally interrupted one that the
|
||||
* remote recovery later completed) must never shadow a persisted
|
||||
* terminal state, even when the local conversation was touched later.
|
||||
*/
|
||||
function persistedTerminalStateOverridesLocal(
|
||||
local: Message,
|
||||
persisted: Message,
|
||||
): boolean {
|
||||
return persisted.state !== "streaming" && local.state !== "complete";
|
||||
}
|
||||
|
||||
export function withRecoveredQuestions(conversation: Conversation): Conversation {
|
||||
const active = conversation.activeRequest;
|
||||
if (!active) return conversation;
|
||||
return {
|
||||
...conversation,
|
||||
messages: conversation.messages.map(message =>
|
||||
message.id === active.messageId && message.state === "streaming"
|
||||
? { ...message, pendingQuestions: active.questions }
|
||||
: message,
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
export function mergePersistedConversations(
|
||||
current: readonly Conversation[],
|
||||
incoming: readonly ConversationListSnapshot[],
|
||||
persistedLocal: Map<string, Conversation>,
|
||||
retainDetails: ReadonlySet<string> = new Set(),
|
||||
): Conversation[] {
|
||||
const incomingById = new Map(
|
||||
incoming.map((conversation) => [conversation.id, conversation]),
|
||||
);
|
||||
const currentById = new Map(
|
||||
current.map((conversation) => [conversation.id, conversation]),
|
||||
);
|
||||
const merged = incoming.map((conversation): Conversation => {
|
||||
if (conversation.messageSummary) {
|
||||
const local = currentById.get(conversation.id);
|
||||
const acknowledged = persistedLocal.get(conversation.id);
|
||||
const dirtyMessages = local &&
|
||||
local.messages.some(message => !acknowledged?.messages.includes(message));
|
||||
const keepMessages = local && (dirtyMessages ||
|
||||
local.messages.some(message => message.approval || message.pendingQuestions?.length || message.state === "streaming") ||
|
||||
(!local.messageSummary && retainDetails.has(conversation.id)));
|
||||
const next = local && local.updatedAt > conversation.updatedAt
|
||||
? { ...local, pinned: conversation.pinned, messageSummary: conversation.messageSummary, messages: [] }
|
||||
: conversation;
|
||||
if (keepMessages) {
|
||||
// A list reply started before navigation must not unload the newly
|
||||
// opened history, or any messages not acknowledged by Main yet.
|
||||
return {
|
||||
...next, messages: local.messages,
|
||||
messageSummary: local.messageSummary ? conversation.messageSummary : undefined,
|
||||
};
|
||||
}
|
||||
if (!conversation.remote) persistedLocal.set(conversation.id, conversation);
|
||||
return next;
|
||||
}
|
||||
if (conversation.remote) {
|
||||
return conversation;
|
||||
}
|
||||
const local = currentById.get(conversation.id);
|
||||
if (!local || local.remote) {
|
||||
persistedLocal.set(conversation.id, conversation);
|
||||
return withRecoveredQuestions(conversation);
|
||||
}
|
||||
const localMessageById = new Map(
|
||||
local.messages.map((message) => [message.id, message]),
|
||||
);
|
||||
const localIsNewer = local.updatedAt > conversation.updatedAt;
|
||||
const serverMessageIds = new Set(
|
||||
conversation.messages.map((message) => message.id),
|
||||
);
|
||||
const messages = [
|
||||
...conversation.messages.map((persistedMessage) => {
|
||||
const localMessage = localMessageById.get(persistedMessage.id);
|
||||
const message = mergeMessageImageState(persistedMessage, persistedMessage, localMessage);
|
||||
// A newer conversation timestamp can still contain an older message
|
||||
// snapshot while a terminal event is being persisted.
|
||||
if (localMessage && localMessage.state !== "streaming" && message.state === "streaming") {
|
||||
return mergeMessageImageState(localMessage, persistedMessage, localMessage);
|
||||
}
|
||||
if (!conversation.activeRequest && local.activeRequest?.messageId === message.id) {
|
||||
return { ...message, pendingQuestions: undefined };
|
||||
}
|
||||
if (!localIsNewer) {
|
||||
// Pending prompts are live-only and are omitted from persisted messages.
|
||||
if (localMessage?.state === "streaming" && message.state === "streaming") {
|
||||
return mergeMessageImageState(localMessage, persistedMessage, localMessage);
|
||||
}
|
||||
return message;
|
||||
}
|
||||
if (
|
||||
!localMessage ||
|
||||
persistedTerminalStateOverridesLocal(localMessage, message)
|
||||
) {
|
||||
return message;
|
||||
}
|
||||
return mergeMessageImageState(localMessage, persistedMessage, localMessage);
|
||||
}),
|
||||
...local.messages.filter((message) => !serverMessageIds.has(message.id)),
|
||||
];
|
||||
const next = localIsNewer
|
||||
? { ...local, pinned: conversation.pinned, messages, messageSummary: undefined, activeRequest: conversation.activeRequest }
|
||||
: { ...conversation, messages };
|
||||
persistedLocal.set(conversation.id, conversation);
|
||||
return withRecoveredQuestions(next);
|
||||
});
|
||||
for (const conversation of current) {
|
||||
if (!incomingById.has(conversation.id)) {
|
||||
merged.push(conversation);
|
||||
}
|
||||
}
|
||||
return sortConversationsForDisplay(merged);
|
||||
}
|
||||
|
||||
|
||||
export type ConversationPersistenceApi = {
|
||||
saveLocal: (batch: LocalConversationSaveBatch) => Promise<unknown>;
|
||||
get: (conversationId: string) => Promise<ConversationSnapshot>;
|
||||
};
|
||||
|
||||
/** What persistence needs from App; connected after each commit. */
|
||||
export type ConversationPersistenceHost = {
|
||||
/** IDs whose loaded history must stay in memory (open panes, runs). */
|
||||
retainedConversationIds: () => Set<string>;
|
||||
/** Whether a task of this conversation is still running or waiting. */
|
||||
hasBusyTask: (conversationId: string) => boolean;
|
||||
/** Pin edits in flight: a history reply must not overwrite them. */
|
||||
pinState: () => { revision: number; pending: boolean };
|
||||
historyUnavailableError: () => Error;
|
||||
onSaveFailed: () => void;
|
||||
};
|
||||
|
||||
export type ConversationPersistenceOptions = {
|
||||
store: ConversationStore;
|
||||
/** Defaults to the preload bridge. */
|
||||
api?: ConversationPersistenceApi;
|
||||
intervalMs: number;
|
||||
};
|
||||
|
||||
export type ConversationPersistence = ReturnType<typeof createConversationPersistence>;
|
||||
|
||||
const detachedHost: ConversationPersistenceHost = {
|
||||
retainedConversationIds: () => new Set(),
|
||||
hasBusyTask: () => false,
|
||||
pinState: () => ({ revision: 0, pending: false }),
|
||||
historyUnavailableError: () => new Error("Conversation history is unavailable"),
|
||||
onSaveFailed: () => undefined,
|
||||
};
|
||||
|
||||
/**
|
||||
* Local conversation persistence. Owns the acknowledged snapshots, the
|
||||
* serialized save queue (every write to Main's local copy goes through it,
|
||||
* in order), the delete/pause guards and in-flight history requests.
|
||||
*/
|
||||
export function createConversationPersistence(options: ConversationPersistenceOptions) {
|
||||
const { store } = options;
|
||||
const api: ConversationPersistenceApi = options.api ?? {
|
||||
saveLocal: (batch) => window.goodbuddy.conversations.saveLocal(batch),
|
||||
get: (conversationId) => window.goodbuddy.conversations.get(conversationId),
|
||||
};
|
||||
let host = detachedHost;
|
||||
/** The last snapshot of each local conversation acknowledged by Main. */
|
||||
let acknowledged = new Map<string, Conversation>();
|
||||
/** Serializes every local save; await it before reading Main's copy. */
|
||||
let queue: Promise<void> = Promise.resolve();
|
||||
let paused = false;
|
||||
const deleting = new Set<string>();
|
||||
/** Set by run completion; the next list commit persists at once. */
|
||||
let flushRequested = false;
|
||||
const historyRequests = new Map<string, Promise<Conversation>>();
|
||||
let started = false;
|
||||
|
||||
const releaseUnretainedHistories = (): void => {
|
||||
const retained = host.retainedConversationIds();
|
||||
store.set(current => {
|
||||
let changed = false;
|
||||
const next = current.map(conversation => {
|
||||
if (conversation.messageSummary || retained.has(conversation.id) ||
|
||||
historyRequests.has(conversation.id) ||
|
||||
conversation.activeRequest ||
|
||||
(!conversation.remote && acknowledged.get(conversation.id) !== conversation) ||
|
||||
conversation.messages.some(message => message.state === "streaming" ||
|
||||
message.approval || message.pendingQuestions?.length) ||
|
||||
host.hasBusyTask(conversation.id)) return conversation;
|
||||
changed = true;
|
||||
const summary: Conversation = {
|
||||
...conversation, messages: [],
|
||||
messageSummary: { count: conversation.messages.length, firstRole: conversation.messages[0]?.role },
|
||||
};
|
||||
if (!conversation.remote) acknowledged.set(conversation.id, summary);
|
||||
return summary;
|
||||
});
|
||||
return changed ? next : current;
|
||||
});
|
||||
};
|
||||
|
||||
/** Queues a save of every changed local conversation, then releases idle history. */
|
||||
const persist = (): void => {
|
||||
const operation = queue.then(async () => {
|
||||
if (paused) {
|
||||
return;
|
||||
}
|
||||
const { batch, acknowledgements } = createLocalConversationSaveBatch(
|
||||
store.getState(),
|
||||
acknowledged,
|
||||
deleting,
|
||||
);
|
||||
if (batch.length === 0) {
|
||||
releaseUnretainedHistories();
|
||||
return;
|
||||
}
|
||||
await api.saveLocal(batch);
|
||||
for (const conversation of acknowledgements) {
|
||||
acknowledged.set(conversation.id, conversation);
|
||||
}
|
||||
releaseUnretainedHistories();
|
||||
});
|
||||
queue = operation.catch(() => undefined);
|
||||
void operation.catch(() => host.onSaveFailed());
|
||||
};
|
||||
|
||||
const ensureHistory = (conversationId: string): Promise<Conversation> => {
|
||||
const current = store.getConversation(conversationId);
|
||||
if (!current) return Promise.reject(host.historyUnavailableError());
|
||||
if (!current.messageSummary) return Promise.resolve(current);
|
||||
const pending = historyRequests.get(conversationId);
|
||||
if (pending) return pending;
|
||||
const pinRevision = host.pinState().revision;
|
||||
const request = api.get(conversationId).then(snapshot => {
|
||||
const latest = store.getConversation(conversationId);
|
||||
if (!latest || deleting.has(conversationId)) {
|
||||
throw host.historyUnavailableError();
|
||||
}
|
||||
if (!latest.messageSummary) return latest;
|
||||
const pins = host.pinState();
|
||||
if (pinRevision !== pins.revision || pins.pending) {
|
||||
snapshot = { ...snapshot, pinned: latest.pinned };
|
||||
}
|
||||
const next = mergePersistedConversations(
|
||||
store.getState(), [snapshot], acknowledged,
|
||||
host.retainedConversationIds(),
|
||||
);
|
||||
store.set(next);
|
||||
return next.find(item => item.id === conversationId)!;
|
||||
}).finally(() => { historyRequests.delete(conversationId); });
|
||||
historyRequests.set(conversationId, request);
|
||||
return request;
|
||||
};
|
||||
|
||||
return {
|
||||
connect(next: ConversationPersistenceHost): void {
|
||||
host = next;
|
||||
},
|
||||
/** Acknowledged snapshots; merges update it in place. */
|
||||
acknowledged: (): Map<string, Conversation> => acknowledged,
|
||||
replaceAcknowledged(next: Map<string, Conversation>): void {
|
||||
acknowledged = next;
|
||||
},
|
||||
/** Resolves after every save queued so far. */
|
||||
idle: (): Promise<void> => queue,
|
||||
/** Runs `task` after every queued save, as part of the queue. */
|
||||
enqueue<T>(task: () => Promise<T>): Promise<T> {
|
||||
const operation = queue.then(task);
|
||||
queue = operation.then(() => undefined, () => undefined);
|
||||
return operation;
|
||||
},
|
||||
/** Makes the queue wait for `operation` (initial load or migration). */
|
||||
chain(operation: Promise<unknown>): void {
|
||||
queue = operation.then(() => undefined, () => undefined);
|
||||
},
|
||||
setPaused(value: boolean): void {
|
||||
paused = value;
|
||||
},
|
||||
markDeleting(conversationId: string, value: boolean): void {
|
||||
if (value) deleting.add(conversationId);
|
||||
else deleting.delete(conversationId);
|
||||
},
|
||||
/** A run ended: persist with the next committed list change. */
|
||||
requestFlushAfterCommit(): void {
|
||||
flushRequested = true;
|
||||
},
|
||||
persist,
|
||||
ensureHistory,
|
||||
releaseUnretainedHistories,
|
||||
/**
|
||||
* Starts periodic saves once the store holds Main's conversations.
|
||||
* The returned stop function saves once more.
|
||||
*/
|
||||
start(): () => void {
|
||||
started = true;
|
||||
persist();
|
||||
const interval = setInterval(persist, options.intervalMs);
|
||||
return () => {
|
||||
clearInterval(interval);
|
||||
started = false;
|
||||
persist();
|
||||
};
|
||||
},
|
||||
/** Called after each committed list change. */
|
||||
flushAfterCommit(): void {
|
||||
if (!started || !flushRequested) return;
|
||||
flushRequested = false;
|
||||
persist();
|
||||
},
|
||||
/** Saves what is pending and resolves when the queue is drained. */
|
||||
async flushForQuit(): Promise<void> {
|
||||
if (!started) return;
|
||||
flushRequested = false;
|
||||
persist();
|
||||
await queue;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,204 @@
|
||||
import type {
|
||||
AssistantSchedule,
|
||||
AssistantTask,
|
||||
ConversationListSnapshot,
|
||||
} from "../../shared/assistant-contracts";
|
||||
import type { Message } from "./ChatTimeline";
|
||||
import type { ConversationStore } from "./conversation-store";
|
||||
import { mergePersistedConversations, type ConversationPersistence } from "./conversation-persistence";
|
||||
|
||||
/** A run App started and has not seen finish yet. */
|
||||
export type ConversationRefreshRun = {
|
||||
conversationId: string;
|
||||
messageId: string;
|
||||
projectId?: string;
|
||||
};
|
||||
|
||||
export type ConversationRefreshApi = {
|
||||
listTasks: () => Promise<AssistantTask[]>;
|
||||
listSchedules: () => Promise<AssistantSchedule[]>;
|
||||
listSummaries: (detailIds: string[]) => Promise<ConversationListSnapshot[]>;
|
||||
/** Main reports that persisted conversations, tasks or schedules changed. */
|
||||
onConversationsChanged: (listener: () => void) => () => void;
|
||||
onQueueChanged: (listener: () => void) => () => void;
|
||||
};
|
||||
|
||||
export type ConversationRefreshDependencies = {
|
||||
store: ConversationStore;
|
||||
persistence: Pick<ConversationPersistence, "acknowledged">;
|
||||
/** Defaults to the preload bridge. */
|
||||
api?: ConversationRefreshApi;
|
||||
activeRuns: Map<string, ConversationRefreshRun>;
|
||||
activeConversationId: () => string;
|
||||
retainedConversationIds: () => Set<string>;
|
||||
pinState: () => { revision: number; pending: boolean };
|
||||
onTasks: (tasks: AssistantTask[]) => void;
|
||||
onSchedules: (schedules: AssistantSchedule[]) => void;
|
||||
/** Remote conversations updated while another conversation is open. */
|
||||
onUnread: (conversations: ConversationListSnapshot[]) => void;
|
||||
/** A run whose message reached a persisted terminal state. */
|
||||
onRunSettled: (requestId: string, run: ConversationRefreshRun, state: Message["state"]) => void;
|
||||
onError: (kind: "tasks" | "schedules" | "conversations") => void;
|
||||
/** Coalescing delay before a refresh; tests pass 0. */
|
||||
delayMs?: number;
|
||||
target?: Pick<Window, "addEventListener" | "removeEventListener">;
|
||||
document?: Pick<Document, "addEventListener" | "removeEventListener" | "visibilityState">;
|
||||
};
|
||||
|
||||
/**
|
||||
* Re-reads tasks, schedules and conversation summaries whenever Main reports a
|
||||
* change, the queue changes, or the window regains focus, and merges the
|
||||
* summaries into the conversation store. Refreshes are coalesced, never run
|
||||
* concurrently, and stale replies (a newer change arrived meanwhile) are
|
||||
* dropped. Returns the stop function.
|
||||
*/
|
||||
function desktopRefreshApi(): ConversationRefreshApi {
|
||||
const api = window.goodbuddy;
|
||||
return {
|
||||
listTasks: () => api.tasks.list(),
|
||||
listSchedules: () => api.schedules.list(),
|
||||
listSummaries: (detailIds) => api.conversations.listSummaries(detailIds),
|
||||
onConversationsChanged: (listener) => api.conversations.onChanged(listener),
|
||||
onQueueChanged: (listener) => api.conversationQueue.onChanged(() => listener()),
|
||||
};
|
||||
}
|
||||
|
||||
export function startConversationRefresh(deps: ConversationRefreshDependencies): () => void {
|
||||
const { store } = deps;
|
||||
const api = deps.api ?? desktopRefreshApi();
|
||||
const target = deps.target ?? window;
|
||||
const doc = deps.document ?? document;
|
||||
const delayMs = deps.delayMs ?? 50;
|
||||
let active = true;
|
||||
let refreshSequence = 0;
|
||||
let refreshTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
let refreshInFlight = false;
|
||||
let refreshQueued = false;
|
||||
|
||||
const queueRefresh = (): void => {
|
||||
refreshSequence += 1;
|
||||
refreshQueued = true;
|
||||
if (refreshInFlight || refreshTimer !== undefined) {
|
||||
return;
|
||||
}
|
||||
refreshTimer = setTimeout(() => {
|
||||
refreshTimer = undefined;
|
||||
refresh();
|
||||
}, delayMs);
|
||||
};
|
||||
|
||||
// A run whose assistant message has already reached a persisted terminal
|
||||
// state (for example a remote task that finished while the desktop was
|
||||
// disconnected) must stop counting as in-flight, otherwise the renderer
|
||||
// keeps showing "processing" forever.
|
||||
const settleActiveRunsFromPersistedMessages = (
|
||||
persisted: readonly ConversationListSnapshot[],
|
||||
): void => {
|
||||
if (deps.activeRuns.size === 0) {
|
||||
return;
|
||||
}
|
||||
const persistedMessageStates = new Map<string, Message["state"]>();
|
||||
for (const conversation of persisted) {
|
||||
for (const message of conversation.messages) {
|
||||
persistedMessageStates.set(`${conversation.id}\0${message.id}`, message.state);
|
||||
}
|
||||
}
|
||||
for (const [requestId, run] of deps.activeRuns) {
|
||||
const state = persistedMessageStates.get(`${run.conversationId}\0${run.messageId}`);
|
||||
if (state !== undefined && state !== "streaming") {
|
||||
deps.activeRuns.delete(requestId);
|
||||
deps.onRunSettled(requestId, run, state);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const refresh = (): void => {
|
||||
if (refreshInFlight) {
|
||||
refreshQueued = true;
|
||||
return;
|
||||
}
|
||||
refreshInFlight = true;
|
||||
refreshQueued = false;
|
||||
const sequence = refreshSequence;
|
||||
const current = (): boolean => active && sequence === refreshSequence;
|
||||
const pinRevision = deps.pinState().revision;
|
||||
const tasksRefresh = api.listTasks()
|
||||
.then((tasks) => {
|
||||
if (current()) deps.onTasks(tasks);
|
||||
})
|
||||
.catch(() => {
|
||||
if (current()) deps.onError("tasks");
|
||||
});
|
||||
const schedulesRefresh = api.listSchedules()
|
||||
.then((schedules) => {
|
||||
if (current()) deps.onSchedules(schedules);
|
||||
})
|
||||
.catch(() => {
|
||||
if (current()) deps.onError("schedules");
|
||||
});
|
||||
const conversationsRefresh = api.listSummaries([...deps.retainedConversationIds()])
|
||||
.then((persisted) => {
|
||||
if (!current()) {
|
||||
return;
|
||||
}
|
||||
const pins = deps.pinState();
|
||||
if (pinRevision !== pins.revision) {
|
||||
queueRefresh();
|
||||
return;
|
||||
}
|
||||
if (pins.pending) {
|
||||
const currentPins = new Map(store.getState().map(item => [item.id, item.pinned]));
|
||||
persisted = persisted.map(item => ({ ...item, pinned: currentPins.has(item.id) ? currentPins.get(item.id) : item.pinned }));
|
||||
}
|
||||
const previousById = new Map(store.getState().map((conversation) => [conversation.id, conversation]));
|
||||
const updated = persisted.filter((conversation) => {
|
||||
if (!conversation.remote) return false;
|
||||
const previous = previousById.get(conversation.id);
|
||||
return previous === undefined || conversation.updatedAt > previous.updatedAt;
|
||||
});
|
||||
const activeId = deps.activeConversationId();
|
||||
const unread = updated.filter((conversation) => conversation.id !== activeId);
|
||||
if (unread.length > 0) {
|
||||
deps.onUnread(unread);
|
||||
}
|
||||
store.set((list) =>
|
||||
mergePersistedConversations(
|
||||
list,
|
||||
persisted,
|
||||
deps.persistence.acknowledged(),
|
||||
deps.retainedConversationIds(),
|
||||
),
|
||||
);
|
||||
settleActiveRunsFromPersistedMessages(persisted);
|
||||
})
|
||||
.catch(() => {
|
||||
if (current()) deps.onError("conversations");
|
||||
});
|
||||
void Promise.allSettled([tasksRefresh, schedulesRefresh, conversationsRefresh]).finally(() => {
|
||||
refreshInFlight = false;
|
||||
if (active && refreshQueued) {
|
||||
queueRefresh();
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
const refreshWhenVisible = (): void => {
|
||||
if (doc.visibilityState !== "hidden") {
|
||||
queueRefresh();
|
||||
}
|
||||
};
|
||||
const removeConversations = api.onConversationsChanged(queueRefresh);
|
||||
const removeQueue = api.onQueueChanged(queueRefresh);
|
||||
target.addEventListener("focus", refreshWhenVisible);
|
||||
doc.addEventListener("visibilitychange", refreshWhenVisible);
|
||||
return () => {
|
||||
active = false;
|
||||
if (refreshTimer !== undefined) {
|
||||
clearTimeout(refreshTimer);
|
||||
}
|
||||
removeConversations();
|
||||
removeQueue();
|
||||
target.removeEventListener("focus", refreshWhenVisible);
|
||||
doc.removeEventListener("visibilitychange", refreshWhenVisible);
|
||||
};
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
import { useCallback } from "react";
|
||||
import type { AssistantTask } from "../../shared/assistant-contracts";
|
||||
import type { Message } from "./ChatTimeline";
|
||||
import { getConversationDisplayTitle, type Conversation } from "./chat-conversation";
|
||||
import { getConversationDisplayTitle, isUnusedConversation, type Conversation } from "./chat-conversation";
|
||||
import {
|
||||
deriveConversationActivity,
|
||||
sameActivitySummary,
|
||||
@@ -121,6 +121,107 @@ export function useConversationActivitySummary(
|
||||
return useConversationStoreSelector(store, selector, sameActivitySummary);
|
||||
}
|
||||
|
||||
/**
|
||||
* What App shows of the active conversation: its header fields plus a few
|
||||
* values derived from its messages. Message content is not part of it, so a
|
||||
* streaming flush or tool update in the active conversation does not
|
||||
* re-render App; the history pane subscribes to the conversation itself.
|
||||
*/
|
||||
export type ActiveConversationView = Pick<
|
||||
Conversation,
|
||||
| "id" | "title" | "projectId" | "workMode" | "runtimeSelection" | "knowledgeLibraryIds"
|
||||
| "knowledgeRetrievalMode" | "remote" | "branch" | "contextMetrics" | "contextCompressionState"
|
||||
> & {
|
||||
/** Untouched greeting-only conversation: shown with the default title. */
|
||||
unused: boolean;
|
||||
/** Messages are loaded (not a summary placeholder). */
|
||||
historyLoaded: boolean;
|
||||
/** Any message is still streaming. */
|
||||
running: boolean;
|
||||
messageCount: number;
|
||||
/** Artifact IDs referenced by messages, for hydration. */
|
||||
artifactIds: readonly string[];
|
||||
};
|
||||
|
||||
const artifactIdsByMessages = new WeakMap<readonly Message[], string[]>();
|
||||
function referencedArtifactIds(messages: readonly Message[]): string[] {
|
||||
let ids = artifactIdsByMessages.get(messages);
|
||||
if (!ids) {
|
||||
ids = [...new Set(messages.flatMap((message) => [
|
||||
...(message.artifactIds ?? []),
|
||||
...(message.imageOperations?.flatMap((operation) => operation.artifactIds) ?? []),
|
||||
]))];
|
||||
artifactIdsByMessages.set(messages, ids);
|
||||
}
|
||||
return ids;
|
||||
}
|
||||
|
||||
export function selectActiveConversationView(
|
||||
conversation: Conversation | undefined,
|
||||
): ActiveConversationView | undefined {
|
||||
if (!conversation) return undefined;
|
||||
return {
|
||||
id: conversation.id,
|
||||
title: conversation.title,
|
||||
projectId: conversation.projectId,
|
||||
workMode: conversation.workMode,
|
||||
runtimeSelection: conversation.runtimeSelection,
|
||||
knowledgeLibraryIds: conversation.knowledgeLibraryIds,
|
||||
knowledgeRetrievalMode: conversation.knowledgeRetrievalMode,
|
||||
remote: conversation.remote,
|
||||
branch: conversation.branch,
|
||||
contextMetrics: conversation.contextMetrics,
|
||||
contextCompressionState: conversation.contextCompressionState,
|
||||
unused: isUnusedConversation(conversation),
|
||||
historyLoaded: !conversation.messageSummary,
|
||||
running: conversation.messages.some((message) => message.state === "streaming"),
|
||||
messageCount: conversation.messageSummary?.count ?? conversation.messages.length,
|
||||
artifactIds: referencedArtifactIds(conversation.messages),
|
||||
};
|
||||
}
|
||||
|
||||
// Small plain-data fields: persisted refreshes replace them with equal copies.
|
||||
function sameData(left: unknown, right: unknown): boolean {
|
||||
return left === right || JSON.stringify(left) === JSON.stringify(right);
|
||||
}
|
||||
|
||||
export function sameActiveConversationView(
|
||||
left: ActiveConversationView | undefined,
|
||||
right: ActiveConversationView | undefined,
|
||||
): boolean {
|
||||
if (left === right) return true;
|
||||
if (!left || !right) return false;
|
||||
return left.id === right.id && left.title === right.title &&
|
||||
left.projectId === right.projectId && left.workMode === right.workMode &&
|
||||
left.knowledgeRetrievalMode === right.knowledgeRetrievalMode &&
|
||||
left.unused === right.unused && left.historyLoaded === right.historyLoaded &&
|
||||
left.running === right.running && left.messageCount === right.messageCount &&
|
||||
sameArrayItems(left.knowledgeLibraryIds ?? [], right.knowledgeLibraryIds ?? []) &&
|
||||
(left.knowledgeLibraryIds === undefined) === (right.knowledgeLibraryIds === undefined) &&
|
||||
sameArrayItems(left.artifactIds, right.artifactIds) &&
|
||||
sameData(left.runtimeSelection, right.runtimeSelection) &&
|
||||
sameData(left.remote, right.remote) && sameData(left.branch, right.branch) &&
|
||||
sameData(left.contextMetrics, right.contextMetrics) &&
|
||||
sameData(left.contextCompressionState, right.contextCompressionState);
|
||||
}
|
||||
|
||||
export function useActiveConversationView(
|
||||
store: ConversationStore,
|
||||
conversationId: string,
|
||||
): ActiveConversationView | undefined {
|
||||
const selector = useCallback(
|
||||
(conversations: Conversation[]) => selectActiveConversationView(
|
||||
conversations.find((conversation) => conversation.id === conversationId),
|
||||
),
|
||||
[conversationId],
|
||||
);
|
||||
const subscribe = useCallback(
|
||||
(listener: () => void) => store.subscribeConversation(conversationId, listener),
|
||||
[conversationId, store],
|
||||
);
|
||||
return useConversationStoreSelector(store, selector, sameActiveConversationView, subscribe);
|
||||
}
|
||||
|
||||
/** The conversations with these IDs, in this order; stable while unchanged. */
|
||||
export function useConversationsById(
|
||||
store: ConversationStore,
|
||||
|
||||
@@ -12,7 +12,9 @@ import {
|
||||
type ConversationStore,
|
||||
} from "./conversation-store";
|
||||
import {
|
||||
sameActiveConversationView,
|
||||
selectPendingSidebarApprovals,
|
||||
useActiveConversationView,
|
||||
usePendingSidebarApprovals,
|
||||
useConversationsById,
|
||||
} from "./conversation-selectors";
|
||||
@@ -179,6 +181,33 @@ describe("conversation store hooks", () => {
|
||||
expect(view.container.textContent).toBe("A,B");
|
||||
});
|
||||
|
||||
it("re-renders the active conversation view only for header or derived changes", () => {
|
||||
const store = setup([conversation("a", [message({ state: "complete", content: "x" })]), conversation("b")]);
|
||||
const seen: unknown[] = [];
|
||||
function Header() {
|
||||
const view = useActiveConversationView(store, "a");
|
||||
seen.push(view);
|
||||
return <span>{view?.title}:{String(view?.running)}</span>;
|
||||
}
|
||||
const view = render(<Header />);
|
||||
const editMessage = (content: string) => store.set((current) => current.map((item) => item.id === "a"
|
||||
? { ...item, updatedAt: item.updatedAt + 1, messages: item.messages.map((entry) => ({ ...entry, content })) }
|
||||
: item));
|
||||
act(() => editMessage("streamed text"));
|
||||
act(() => store.set((current) => current.map((item) => item.id === "a"
|
||||
? { ...item, runtimeSelection: item.runtimeSelection ? { ...item.runtimeSelection } : undefined }
|
||||
: item)));
|
||||
expect(seen).toHaveLength(1);
|
||||
act(() => store.set((current) => current.map((item) => item.id === "a" ? { ...item, title: "A" } : item)));
|
||||
expect(view.container.textContent).toBe("A:false");
|
||||
act(() => store.set((current) => current.map((item) => item.id === "a"
|
||||
? { ...item, messages: [...item.messages, message({ id: "m2", state: "streaming" })] }
|
||||
: item)));
|
||||
expect(view.container.textContent).toBe("A:true");
|
||||
expect(seen).toHaveLength(3);
|
||||
expect(sameActiveConversationView(seen[1] as never, seen[2] as never)).toBe(false);
|
||||
});
|
||||
|
||||
it("prunes absorbed deltas and runs the commit callback after list commits", () => {
|
||||
const live = createLiveMessageStore({ createId: () => "b" });
|
||||
const store = createConversationStore([conversation("c1", [message()])], live);
|
||||
|
||||
Reference in New Issue
Block a user