feat(coding): add renderer conversation store

This commit is contained in:
2026-08-23 22:04:55 +08:00
parent a0722574f3
commit 36626c88b5
14 changed files with 2703 additions and 1227 deletions

View File

@@ -0,0 +1,657 @@
import { useStore } from 'zustand';
import { createStore, type StoreApi } from 'zustand/vanilla';
import {
createConversationReducerState,
reduceConversationPatch,
replaceConversationSnapshot,
type ConversationReducerState,
} from '../../shared/coding-conversation-reducer';
import { AppError } from '@/lib/error-model';
import {
getCodingConversationSnapshot,
openCodingConversationEvents,
submitCodingConversationPrompt,
type SubmitCodingConversationPromptInput,
} from '@/lib/coding-conversations';
import type {
CodingConversationDraft,
CodingConversationPatchEvent,
CodingConversationSnapshotEvent,
CodingConversationSummary,
CodingDraftAttachment,
CodingPromptRequestState,
ConversationMessageNode,
ConversationPatchEnvelope,
ConversationSnapshot,
PromptAcceptance,
PromptMode,
} from '@/types/coding-conversation';
export type CodingConversationLoadState =
| 'empty'
| 'loading'
| 'live'
| 'recovering'
| 'error';
export interface CodingConversationEntry {
reducer: ConversationReducerState;
loadState: CodingConversationLoadState;
error: string | null;
unread: boolean;
}
export interface SubmitCodingPromptInput {
conversationId: string;
mode: PromptMode;
text?: string;
attachments?: CodingDraftAttachment[];
}
interface CodingConversationStoreDependencies {
getSnapshot(conversationId: string): Promise<ConversationSnapshot>;
openEvents(conversationId?: string): Promise<EventSource>;
submitPrompt(input: SubmitCodingConversationPromptInput): Promise<PromptAcceptance>;
createId(kind: 'request' | 'node'): string;
}
export interface CodingConversationStoreState {
selectedConversationId: string | null;
entriesByConversationId: Record<string, CodingConversationEntry>;
summariesByConversationId: Record<string, CodingConversationSummary>;
draftsByConversationId: Record<string, CodingConversationDraft>;
requestsByConversationId: Record<string, Record<string, CodingPromptRequestState>>;
connectionState: 'disconnected' | 'connecting' | 'live' | 'reconnecting' | 'error';
globalError: string | null;
selectConversation(conversationId: string): Promise<void>;
loadSnapshot(conversationId: string, recovering?: boolean): Promise<ConversationSnapshot>;
connectEvents(): Promise<void>;
disconnectEvents(): void;
setDraft(conversationId: string, text: string, attachments?: CodingDraftAttachment[]): void;
markUnread(conversationId: string, unread: boolean): void;
submitPrompt(input: SubmitCodingPromptInput): Promise<PromptAcceptance>;
applySnapshotEvent(event: CodingConversationSnapshotEvent): void;
applyPatchEvent(event: CodingConversationPatchEvent): void;
}
const DEFAULT_DRAFT: CodingConversationDraft = {
text: '',
attachments: [],
revision: 0,
};
const EMPTY_REQUESTS: Record<string, CodingPromptRequestState> = {};
function cloneDraft(draft: CodingConversationDraft): CodingConversationDraft {
return {
text: draft.text,
attachments: draft.attachments.map((attachment) => ({ ...attachment })),
revision: draft.revision,
};
}
function emptyEntry(): CodingConversationEntry {
return {
reducer: createConversationReducerState(),
loadState: 'empty',
error: null,
unread: false,
};
}
function errorDetails(error: unknown): { code?: string; message: string } {
if (error instanceof AppError) {
return {
...(typeof error.details?.backendCode === 'string'
? { code: error.details.backendCode }
: {}),
message: error.message,
};
}
return { message: error instanceof Error ? error.message : String(error) };
}
function summaryOf(entry: CodingConversationEntry): CodingConversationSummary | null {
const snapshot = entry.reducer.snapshot;
if (!snapshot) return null;
return {
conversationId: snapshot.conversation.id,
title: snapshot.conversation.title,
runStatus: snapshot.run.status,
runError: snapshot.run.error ?? null,
workerStatus: snapshot.worker.status,
model: snapshot.conversation.model,
unread: entry.unread,
};
}
function summariesWithEntry(
summaries: Record<string, CodingConversationSummary>,
entry: CodingConversationEntry,
): Record<string, CodingConversationSummary> {
const next = summaryOf(entry);
if (!next) return summaries;
const current = summaries[next.conversationId];
if (current
&& current.title === next.title
&& current.runStatus === next.runStatus
&& current.runError === next.runError
&& current.workerStatus === next.workerStatus
&& current.model === next.model
&& current.unread === next.unread) {
return summaries;
}
return { ...summaries, [next.conversationId]: next };
}
function requestIdsInSnapshot(snapshot: ConversationSnapshot): Set<string> {
return new Set(snapshot.nodes.flatMap((node) => (
node.kind === 'message' && node.clientRequestId ? [node.clientRequestId] : []
)));
}
function withoutReconciledRequests(
requests: Record<string, CodingPromptRequestState> | undefined,
snapshot: ConversationSnapshot,
): Record<string, CodingPromptRequestState> {
const reconciled = requestIdsInSnapshot(snapshot);
if (!requests || reconciled.size === 0) return requests ?? {};
return Object.fromEntries(
Object.entries(requests).filter(([clientRequestId]) => !reconciled.has(clientRequestId)),
);
}
function optimisticNode(
nodeId: string,
clientRequestId: string,
draft: CodingConversationDraft,
): ConversationMessageNode {
return {
kind: 'message',
id: nodeId,
clientRequestId,
role: 'user',
status: 'optimistic',
blocks: [
...(draft.text ? [{
kind: 'text' as const,
id: `${nodeId}:text`,
text: draft.text,
status: 'complete' as const,
}] : []),
...draft.attachments.map((attachment, index) => ({
kind: 'image' as const,
id: `${nodeId}:attachment:${index}`,
attachmentId: attachment.attachmentId,
mime: attachment.mime,
})),
],
};
}
function withOptimisticNode(
reducer: ConversationReducerState,
node: ConversationMessageNode,
): ConversationReducerState {
const snapshot = reducer.snapshot;
if (!snapshot) return reducer;
return replaceConversationSnapshot(reducer, {
...snapshot,
nodes: [...snapshot.nodes, node],
});
}
function withRejectedNode(
reducer: ConversationReducerState,
nodeId: string | undefined,
): ConversationReducerState {
const snapshot = reducer.snapshot;
if (!snapshot || !nodeId) return reducer;
return replaceConversationSnapshot(reducer, {
...snapshot,
nodes: snapshot.nodes.map((node) => (
node.kind === 'message' && node.id === nodeId
? { ...node, status: 'error' as const }
: node
)),
});
}
function snapshotIsOlder(
current: ConversationSnapshot | null,
incoming: ConversationSnapshot,
): boolean {
if (!current) return false;
if (incoming.cursor.workerGeneration !== current.cursor.workerGeneration) {
return incoming.cursor.workerGeneration < current.cursor.workerGeneration;
}
return incoming.cursor.seq < current.cursor.seq;
}
function defaultDependencies(): CodingConversationStoreDependencies {
return {
getSnapshot: getCodingConversationSnapshot,
openEvents: openCodingConversationEvents,
submitPrompt: submitCodingConversationPrompt,
createId: () => crypto.randomUUID(),
};
}
export function createCodingConversationStore(
dependencies: Partial<CodingConversationStoreDependencies> = {},
): StoreApi<CodingConversationStoreState> {
const deps = { ...defaultDependencies(), ...dependencies };
const snapshotLoads = new Map<string, Promise<ConversationSnapshot>>();
let eventSource: EventSource | null = null;
let connectFlight: Promise<void> | null = null;
let connectionGeneration = 0;
return createStore<CodingConversationStoreState>((set, get) => ({
selectedConversationId: null,
entriesByConversationId: {},
summariesByConversationId: {},
draftsByConversationId: {},
requestsByConversationId: {},
connectionState: 'disconnected',
globalError: null,
async selectConversation(conversationId) {
set((state) => {
const current = state.entriesByConversationId[conversationId] ?? emptyEntry();
const entry = current.unread ? { ...current, unread: false } : current;
return {
selectedConversationId: conversationId,
entriesByConversationId: {
...state.entriesByConversationId,
[conversationId]: entry,
},
summariesByConversationId: summariesWithEntry(state.summariesByConversationId, entry),
};
});
const entry = get().entriesByConversationId[conversationId];
if (!entry?.reducer.snapshot || entry.reducer.invalidation) {
await get().loadSnapshot(conversationId, Boolean(entry?.reducer.invalidation));
}
await get().connectEvents();
},
loadSnapshot(conversationId, recovering = false) {
const prior = snapshotLoads.get(conversationId);
if (prior) return prior;
set((state) => {
const entry = state.entriesByConversationId[conversationId] ?? emptyEntry();
return {
entriesByConversationId: {
...state.entriesByConversationId,
[conversationId]: {
...entry,
loadState: recovering ? 'recovering' : 'loading',
error: null,
},
},
};
});
const flight = deps.getSnapshot(conversationId)
.then((snapshot) => {
get().applySnapshotEvent({
type: 'snapshot',
conversationId,
workerGeneration: snapshot.cursor.workerGeneration,
seq: snapshot.cursor.seq,
snapshot,
});
return snapshot;
})
.catch((error) => {
const failure = errorDetails(error);
set((state) => {
const entry = state.entriesByConversationId[conversationId] ?? emptyEntry();
return {
entriesByConversationId: {
...state.entriesByConversationId,
[conversationId]: { ...entry, loadState: 'error', error: failure.message },
},
};
});
throw error;
})
.finally(() => {
if (snapshotLoads.get(conversationId) === flight) snapshotLoads.delete(conversationId);
});
snapshotLoads.set(conversationId, flight);
return flight;
},
async connectEvents() {
if (eventSource) return;
if (connectFlight) return await connectFlight;
const generation = ++connectionGeneration;
set({ connectionState: 'connecting', globalError: null });
let flight: Promise<void>;
flight = deps.openEvents()
.then((source) => {
if (generation !== connectionGeneration) {
source.close();
return;
}
eventSource = source;
const parse = <T,>(event: Event): T => JSON.parse((event as MessageEvent<string>).data) as T;
source.addEventListener('snapshot', (event) => {
try {
get().applySnapshotEvent(parse<CodingConversationSnapshotEvent>(event));
} catch {
set({ connectionState: 'error', globalError: 'Conversation 事件快照无法读取。' });
}
});
source.addEventListener('patch', (event) => {
try {
get().applyPatchEvent(parse<CodingConversationPatchEvent>(event));
} catch {
set({ connectionState: 'error', globalError: 'Conversation 事件更新无法读取。' });
}
});
source.onopen = () => {
if (eventSource === source) set({ connectionState: 'live', globalError: null });
};
source.onerror = () => {
if (eventSource === source) {
set({
connectionState: 'reconnecting',
globalError: 'Conversation 事件连接中断,正在重连。',
});
}
};
})
.catch((error) => {
if (generation !== connectionGeneration) return;
set({ connectionState: 'error', globalError: errorDetails(error).message });
throw error;
})
.finally(() => {
if (connectFlight === flight) connectFlight = null;
});
connectFlight = flight;
await flight;
},
disconnectEvents() {
connectionGeneration += 1;
eventSource?.close();
eventSource = null;
connectFlight = null;
set({ connectionState: 'disconnected', globalError: null });
},
setDraft(conversationId, text, attachments = []) {
set((state) => {
const current = state.draftsByConversationId[conversationId] ?? DEFAULT_DRAFT;
return {
draftsByConversationId: {
...state.draftsByConversationId,
[conversationId]: {
text,
attachments: attachments.map((attachment) => ({ ...attachment })),
revision: current.revision + 1,
},
},
};
});
},
markUnread(conversationId, unread) {
set((state) => {
const current = state.entriesByConversationId[conversationId] ?? emptyEntry();
if (current.unread === unread) return state;
const entry = { ...current, unread };
return {
entriesByConversationId: {
...state.entriesByConversationId,
[conversationId]: entry,
},
summariesByConversationId: summariesWithEntry(state.summariesByConversationId, entry),
};
});
},
async submitPrompt(input) {
let entry = get().entriesByConversationId[input.conversationId];
if (!entry?.reducer.snapshot || entry.reducer.invalidation) {
await get().loadSnapshot(input.conversationId, Boolean(entry?.reducer.invalidation));
entry = get().entriesByConversationId[input.conversationId];
}
if (!entry?.reducer.snapshot) throw new Error('Conversation snapshot is unavailable');
const existingDraft = get().draftsByConversationId[input.conversationId] ?? DEFAULT_DRAFT;
const submittedDraft: CodingConversationDraft = {
text: input.text ?? existingDraft.text,
attachments: (input.attachments ?? existingDraft.attachments).map((attachment) => ({ ...attachment })),
revision: existingDraft.revision,
};
const clientRequestId = deps.createId('request');
const nodeId = input.mode === 'prompt' ? deps.createId('node') : undefined;
const clearedRevision = existingDraft.revision + 1;
const request: CodingPromptRequestState = {
clientRequestId,
mode: input.mode,
...(nodeId ? { nodeId } : {}),
status: 'pending',
submittedDraft: cloneDraft(submittedDraft),
};
set((state) => {
const currentEntry = state.entriesByConversationId[input.conversationId] ?? emptyEntry();
const reducer = nodeId
? withOptimisticNode(
currentEntry.reducer,
optimisticNode(nodeId, clientRequestId, submittedDraft),
)
: currentEntry.reducer;
const nextEntry = { ...currentEntry, reducer, error: null };
return {
entriesByConversationId: {
...state.entriesByConversationId,
[input.conversationId]: nextEntry,
},
summariesByConversationId: summariesWithEntry(
state.summariesByConversationId,
nextEntry,
),
draftsByConversationId: {
...state.draftsByConversationId,
[input.conversationId]: { text: '', attachments: [], revision: clearedRevision },
},
requestsByConversationId: {
...state.requestsByConversationId,
[input.conversationId]: {
...(state.requestsByConversationId[input.conversationId] ?? {}),
[clientRequestId]: request,
},
},
};
});
try {
const acceptance = await deps.submitPrompt({
conversationId: input.conversationId,
clientRequestId,
mode: input.mode,
text: submittedDraft.text,
attachments: submittedDraft.attachments.map(({ attachmentId }) => ({ attachmentId })),
});
set((state) => {
const requests = state.requestsByConversationId[input.conversationId] ?? {};
const current = requests[clientRequestId];
if (!current) return state;
return {
requestsByConversationId: {
...state.requestsByConversationId,
[input.conversationId]: {
...requests,
[clientRequestId]: { ...current, status: 'accepted' },
},
},
};
});
return acceptance;
} catch (error) {
const failure = errorDetails(error);
const uncertain = failure.code === 'CODING_REQUEST_UNCERTAIN';
set((state) => {
const currentEntry = state.entriesByConversationId[input.conversationId] ?? emptyEntry();
const requests = state.requestsByConversationId[input.conversationId] ?? {};
const currentRequest = requests[clientRequestId] ?? request;
const currentDraft = state.draftsByConversationId[input.conversationId] ?? DEFAULT_DRAFT;
const canRestore = currentDraft.revision === clearedRevision
&& !currentDraft.text
&& currentDraft.attachments.length === 0;
const reducer = uncertain
? currentEntry.reducer
: withRejectedNode(currentEntry.reducer, nodeId);
const nextEntry = { ...currentEntry, reducer, error: failure.message };
return {
entriesByConversationId: {
...state.entriesByConversationId,
[input.conversationId]: nextEntry,
},
summariesByConversationId: summariesWithEntry(
state.summariesByConversationId,
nextEntry,
),
draftsByConversationId: canRestore
? {
...state.draftsByConversationId,
[input.conversationId]: {
...cloneDraft(submittedDraft),
revision: clearedRevision + 1,
},
}
: state.draftsByConversationId,
requestsByConversationId: {
...state.requestsByConversationId,
[input.conversationId]: {
...requests,
[clientRequestId]: {
...currentRequest,
status: uncertain ? 'uncertain' : 'rejected',
...(failure.code ? { errorCode: failure.code } : {}),
errorMessage: failure.message,
},
},
},
};
});
throw error;
}
},
applySnapshotEvent(event) {
const snapshot = event.snapshot;
if (event.type !== 'snapshot'
|| event.conversationId !== snapshot.conversation.id
|| event.workerGeneration !== snapshot.cursor.workerGeneration
|| event.seq !== snapshot.cursor.seq) {
return;
}
set((state) => {
const current = state.entriesByConversationId[event.conversationId] ?? emptyEntry();
if (!current.reducer.invalidation && snapshotIsOlder(current.reducer.snapshot, snapshot)) {
return state;
}
const reducer = replaceConversationSnapshot(current.reducer, snapshot);
const entry: CodingConversationEntry = {
...current,
reducer,
loadState: reducer.invalidation ? 'error' : 'live',
error: reducer.invalidation?.reason ?? null,
unread: state.selectedConversationId === event.conversationId ? false : current.unread,
};
return {
entriesByConversationId: {
...state.entriesByConversationId,
[event.conversationId]: entry,
},
summariesByConversationId: summariesWithEntry(state.summariesByConversationId, entry),
requestsByConversationId: {
...state.requestsByConversationId,
[event.conversationId]: withoutReconciledRequests(
state.requestsByConversationId[event.conversationId],
snapshot,
),
},
};
});
},
applyPatchEvent(event) {
if (event.type !== 'patch') return;
let recover = false;
set((state) => {
const current = state.entriesByConversationId[event.conversationId] ?? emptyEntry();
const reducer = reduceConversationPatch(current.reducer, event as ConversationPatchEnvelope);
if (reducer === current.reducer) return state;
recover = Boolean(reducer.invalidation);
const incomingMessage = event.patch.op === 'message.upsert' ? event.patch.node : null;
const unread = state.selectedConversationId !== event.conversationId
&& incomingMessage?.role === 'assistant'
? true
: current.unread;
const entry: CodingConversationEntry = {
...current,
reducer,
loadState: reducer.invalidation ? 'recovering' : 'live',
error: reducer.invalidation?.reason ?? null,
unread,
};
const requests = state.requestsByConversationId[event.conversationId] ?? {};
const nextRequests = incomingMessage?.clientRequestId
? Object.fromEntries(
Object.entries(requests).filter(([id]) => id !== incomingMessage.clientRequestId),
)
: requests;
return {
entriesByConversationId: {
...state.entriesByConversationId,
[event.conversationId]: entry,
},
summariesByConversationId: summariesWithEntry(state.summariesByConversationId, entry),
requestsByConversationId: nextRequests === requests
? state.requestsByConversationId
: {
...state.requestsByConversationId,
[event.conversationId]: nextRequests,
},
};
});
if (recover) {
void get().loadSnapshot(event.conversationId, true).catch(() => undefined);
}
},
}));
}
export const codingConversationStore = createCodingConversationStore();
export function useCodingConversationStore<T>(
selector: (state: CodingConversationStoreState) => T,
): T {
return useStore(codingConversationStore, selector);
}
export const selectCodingConversationSnapshot = (conversationId: string) => (
state: CodingConversationStoreState,
): ConversationSnapshot | null => (
state.entriesByConversationId[conversationId]?.reducer.snapshot ?? null
);
export const selectCodingConversationSummary = (conversationId: string) => (
state: CodingConversationStoreState,
): CodingConversationSummary | null => state.summariesByConversationId[conversationId] ?? null;
export const selectCodingConversationDraft = (conversationId: string) => (
state: CodingConversationStoreState,
): CodingConversationDraft => state.draftsByConversationId[conversationId] ?? DEFAULT_DRAFT;
export const selectCodingConversationRequests = (conversationId: string) => (
state: CodingConversationStoreState,
): Record<string, CodingPromptRequestState> => (
state.requestsByConversationId[conversationId] ?? EMPTY_REQUESTS
);