diff --git a/apps/cli/src/lib/loro/doc.ts b/apps/cli/src/lib/loro/doc.ts index 8ad032ae6..c3df73c84 100644 --- a/apps/cli/src/lib/loro/doc.ts +++ b/apps/cli/src/lib/loro/doc.ts @@ -2579,6 +2579,13 @@ export class SessionDocument implements LoroDocument { if (!this.mirror) { return []; diff --git a/packages/components/src/atoms/runtime.ts b/packages/components/src/atoms/runtime.ts index b95434613..ae076d84d 100644 --- a/packages/components/src/atoms/runtime.ts +++ b/packages/components/src/atoms/runtime.ts @@ -67,6 +67,7 @@ import type { } from '@lody/shared'; import type { LocalProjectGitStateRpcResponse } from '@lody/loro-streams-rpc'; import type { WorkspaceWriter } from '../providers/workspace-writer'; +import type { ConversationView } from '../lib/conversation-view'; import type { CodeCollabFileIndexCache } from '@/lib/code-collab-file-index-cache'; import { readStoredAuthToken } from '@/lib/auth-bootstrap'; import type { RoomSyncState } from '@/lib/room-sync-state'; @@ -92,6 +93,14 @@ export type SessionDocStore = { acquireSync: () => () => void; getSyncState: () => RoomSyncState; subscribeSyncState: (listener: (state: RoomSyncState) => void) => () => void; + /** + * Windowed history reader, present when the store was built with the + * control-plane Mirror (`isConversationViewEnabled()`); `null` on the + * full-Mirror rollback path. When present, `getState().history` is a lazy + * bridge over `conversationView.readAll()` and history writes must go + * through the history writer (`WorkspaceWriter` does), never `setState`. + */ + readonly conversationView: ConversationView | null; getState: () => SessionDocState; setState: (updater: SessionDocUpdater) => void; subscribe: (listener: (state: SessionDocState) => void) => () => void; diff --git a/packages/components/src/components/ai-gui/index.tsx b/packages/components/src/components/ai-gui/index.tsx index 1cdf37e9c..159fba8e1 100644 --- a/packages/components/src/components/ai-gui/index.tsx +++ b/packages/components/src/components/ai-gui/index.tsx @@ -30,6 +30,8 @@ import { } from './view'; import { buildChatStreamItems, type BuildChatStreamItemsCache } from './build-chat-stream-items'; import { useStableCallback } from '@/hooks/use-stable-callback'; +import { useConversationViewSelector } from '@/hooks/use-conversation-view-selector'; +import type { ConversationView } from '@/lib/conversation-view'; import { useCloudQuery } from '@lody/platform/react'; import type { SessionNavigationTarget } from '@/lib/session-navigation'; import type { @@ -38,6 +40,14 @@ import type { } from '@/components/sessions/session-fork-destination-menu'; const emptyHistory = [] as SessionDoc['history']; + +const findLastUserTurnId = (view: ConversationView): string | null => { + for (let index = view.turnCount - 1; index >= 0; index -= 1) { + const row = view.index(index); + if (row?.role === 'user') return row.id ?? null; + } + return null; +}; const CHAT_STREAM_ITEMS_CACHE_LIMIT = 20; const chatStreamItemsCacheBySessionId = new Map(); @@ -75,6 +85,12 @@ export interface SessionChatStreamProps { sessionId: SessionId; workspaceId?: WorkspaceId | null; sessionDoc: SessionDoc; + /** + * Windowed history reader for `sessionDoc`'s session. Present on the + * ConversationView path; `null` (or omitted) on the full-Mirror rollback + * path, where `sessionDoc.history` is the only source. + */ + conversationView?: ConversationView | null; sessionCreatedAt?: string; dividerLabel?: string; className?: string; @@ -159,6 +175,7 @@ const SessionChatStreamImpl = forwardRef { + const fallbackLastUserMessageId = useMemo(() => { + if (conversationView) return null; for (let index = sessionHistory.length - 1; index >= 0; index -= 1) { if (sessionHistory[index]?.role === 'user') return sessionHistory[index]?.id ?? null; } return null; - }, [sessionHistory]); + }, [conversationView, sessionHistory]); + // Index rows carry `role` and `id`, so this never hydrates a turn. + const lastUserMessageId = useConversationViewSelector( + conversationView, + findLastUserTurnId, + fallbackLastUserMessageId + ); const renderMessageRow = useCallback( ({ diff --git a/packages/components/src/components/sessions/managed-preview-surface.tsx b/packages/components/src/components/sessions/managed-preview-surface.tsx index ea47ede47..4f4b8f373 100644 --- a/packages/components/src/components/sessions/managed-preview-surface.tsx +++ b/packages/components/src/components/sessions/managed-preview-surface.tsx @@ -11,7 +11,6 @@ import { Loader2, Send, X } from 'lucide-react'; import { useTranslation } from 'react-i18next'; import { useAtomValue } from 'jotai'; import { - resolveActiveAssistantTurnId, type SessionMeta, type VisualAnnotationReferencePayload, } from '@lody/shared'; @@ -55,6 +54,7 @@ import { } from '@/components/chat/visual-annotation-reference-state'; import { usePreviewVisualCommentDoc } from '@/hooks/use-preview-visual-comment-doc'; import { useSessionDoc } from '@/hooks/use-session-doc'; +import { useActiveAssistantTurnId } from '@/hooks/use-session-turn-selectors'; import { useStableCallback } from '@/hooks/use-stable-callback'; import { observeResizeOnAnimationFrame } from '@/lib/resize-observer'; import { @@ -349,8 +349,8 @@ export function ManagedPreviewSurface({ ) .map((comment) => comment.id); }, [comments, visualAnnotationReferenceKeys]); - const commentTurnId = - resolveActiveAssistantTurnId(sessionDoc.doc.history) ?? session.latestUserMsgId ?? session.id; + const activeAssistantTurnId = useActiveAssistantTurnId(sessionDoc); + const commentTurnId = activeAssistantTurnId ?? session.latestUserMsgId ?? session.id; const trackedAnchors = useMemo(() => { const next = comments.map((comment) => ({ diff --git a/packages/components/src/components/sessions/session-chat-interface.tsx b/packages/components/src/components/sessions/session-chat-interface.tsx index 7029fcbb4..214fa7998 100644 --- a/packages/components/src/components/sessions/session-chat-interface.tsx +++ b/packages/components/src/components/sessions/session-chat-interface.tsx @@ -110,7 +110,6 @@ import { resolveSessionConversationConfig, resolveSessionConversationSourceFence, resolveVisibleSessionGoal, - resolveActiveAssistantTurnId, resolveBaseBranchPreference, resolveProjectGitHubRepo, } from '@lody/shared'; @@ -187,6 +186,7 @@ import { Input } from '@/ui/input'; import { Separator } from '@/ui/separator'; import { Tooltip, TooltipContent, TooltipTrigger } from '@/ui/tooltip'; import { useSessionDoc } from '@/hooks/use-session-doc'; +import { useActiveAssistantTurnId } from '@/hooks/use-session-turn-selectors'; import { useSessionActions } from '@/hooks/use-session-actions'; import { useWorkspaceMembers, type WorkspaceMember } from '@/hooks/use-workspace-members'; import { UserAvatar } from '@/components/user-avatar'; @@ -2052,6 +2052,7 @@ export const SessionChatInterface = memo( const [pendingRemoteHtmlFileName, setPendingRemoteHtmlFileName] = useState(null); const { doc: sessionDoc, + conversationView, addHistory: addSessionHistory, pushMessageQueue, removeMessageQueueItem, @@ -2753,9 +2754,10 @@ export const SessionChatInterface = memo( return resolveActivityFromHistory(sessionHistory); }, [liveSessionStatus, sessionHistory]); - const activeAssistantTurnId = useMemo(() => { - return resolveActiveAssistantTurnId(sessionHistory); - }, [sessionHistory]); + const activeAssistantTurnId = useActiveAssistantTurnId({ + doc: sessionDoc, + conversationView, + }); const messageQueue = useMemo( () => (sessionDoc?.mq ?? []) as MessageQueueItem[], [sessionDoc?.mq] @@ -4246,13 +4248,16 @@ export const SessionChatInterface = memo( const handleScrollToMessage = useCallback( (historyId: string) => { - const history = (sessionDoc?.history as SessionHistory[] | undefined) ?? []; - const index = history.findIndex((h) => h.id === historyId); + const index = conversationView + ? conversationView.indexOf(historyId) + : ((sessionDoc?.history as SessionHistory[] | undefined) ?? []).findIndex( + (h) => h.id === historyId + ); if (index >= 0) { chatStreamRef.current?.scrollToIndex(index); } }, - [sessionDoc?.history] + [conversationView, sessionDoc?.history] ); const createPrPrompt = t('sessions.prompts.createPr', CREATE_PR_PROMPT); @@ -5836,6 +5841,7 @@ export const SessionChatInterface = memo( sessionId={session?.id} workspaceId={workspaceId} sessionDoc={sessionDoc} + conversationView={conversationView} sessionCreatedAt={session?.createdAt} dividerLabel={sessionDividerLabel} className="h-full" diff --git a/packages/components/src/components/sessions/session-detail.tsx b/packages/components/src/components/sessions/session-detail.tsx index ee0cce359..88b4d513d 100644 --- a/packages/components/src/components/sessions/session-detail.tsx +++ b/packages/components/src/components/sessions/session-detail.tsx @@ -238,6 +238,7 @@ import { } from '@/lib/session-file-provider-open-result'; import { canOpenHistoricalSessionDiffs } from '@/lib/session-file-provider'; import { useSessionDoc, useSessionDocSyncState } from '@/hooks/use-session-doc'; +import { useSessionSystemNotice } from '@/hooks/use-session-turn-selectors'; import { useDelayedFlag } from '@/hooks/use-delayed-flag'; import { isSyncingRoomSyncState } from '@/lib/room-sync-state'; import { @@ -462,7 +463,11 @@ function PendingWorktreeForkObserver({ onCompleted: () => void; onFailed: (message: string) => void; }) { - const { doc, ready } = useSessionDoc(targetSessionId, { syncEnabled: true }); + const sessionDoc = useSessionDoc(targetSessionId, { syncEnabled: true }); + const { doc, ready } = sessionDoc; + // The origin notice is the system turn the fork appends last, so this reads + // system turns only and never hydrates the copied transcript. + const completed = useSessionSystemNotice(sessionDoc, 'session_fork_origin'); const terminalRef = useRef(false); useEffect(() => { if (!ready || terminalRef.current) return; @@ -472,16 +477,11 @@ function PendingWorktreeForkObserver({ onFailed(operation.data.error?.message ?? 'Unable to create the fork worktree'); return; } - const completed = doc.history.some((entry) => - (entry.items ?? []).some( - (item) => item.type === 'system_notice' && item.name === 'session_fork_origin' - ) - ); if (!operation.success && completed) { terminalRef.current = true; onCompleted(); } - }, [doc.forkOperation, doc.history, onCompleted, onFailed, ready]); + }, [completed, doc.forkOperation, onCompleted, onFailed, ready]); return null; } diff --git a/packages/components/src/components/sessions/use-session-diff-summary.ts b/packages/components/src/components/sessions/use-session-diff-summary.ts index 996c7405e..b84a5e2f7 100644 --- a/packages/components/src/components/sessions/use-session-diff-summary.ts +++ b/packages/components/src/components/sessions/use-session-diff-summary.ts @@ -2,6 +2,7 @@ import { normalizeFileDiff, type FileDiff, type SessionId } from '@lody/shared'; import { useAtomValue } from 'jotai'; import { useEffect, useRef, useState } from 'react'; import { activeWorkspaceRuntimeAtom } from '@/atoms/runtime'; +import { readDiffInputsFromView } from '@/lib/conversation-view'; import type { SessionFileChangedFilesResult, SessionFileChangeEntry, @@ -401,7 +402,13 @@ export function useSessionDiffSummary( } releaseSync = store.acquireSync(); - const initialHistory = store.getState().history; + // On the view path every turn's `fileDiff` comes from its own small + // container (index rows carry id/role), so the summary never hydrates + // message items; the rollback path reads the Mirror's history array. + const view = store.conversationView; + const readDiffInputs = (): SessionHistoryInput => + view ? readDiffInputsFromView(view) : store.getState().history; + const initialHistory = readDiffInputs(); historyRef.current = initialHistory; diffInputsFingerprintRef.current = computeSessionDiffInputsFingerprint(initialHistory); setDiffInputsVersion((prev) => prev + 1); @@ -427,18 +434,19 @@ export function useSessionDiffSummary( // ignore }); - unsubscribe = store.subscribe((nextState) => { - const nextFingerprint = computeSessionDiffInputsFingerprint(nextState.history); + const handleHistoryChange = () => { + const nextHistory = readDiffInputs(); + const nextFingerprint = computeSessionDiffInputsFingerprint(nextHistory); if (nextFingerprint === diffInputsFingerprintRef.current) { return; } diffInputsFingerprintRef.current = nextFingerprint; - historyRef.current = nextState.history; + historyRef.current = nextHistory; setDiffInputsVersion((prev) => prev + 1); if (!shouldUpdateFallbackSummary()) { return; } - const nextSummary = buildSessionDiffSummary(nextState.history); + const nextSummary = buildSessionDiffSummary(nextHistory); setState((prev) => { if (areSessionDiffSummariesEqual(prev.summary, nextSummary)) { if (prev.source === 'fallback') { @@ -456,7 +464,10 @@ export function useSessionDiffSummary( unavailableMessage: undefined, }; }); - }); + }; + unsubscribe = view + ? view.subscribe(handleHistoryChange) + : store.subscribe(handleHistoryChange); } catch (error) { console.error('Failed to load session diff summary', { sessionId, error }); } diff --git a/packages/components/src/hooks/AGENTS.md b/packages/components/src/hooks/AGENTS.md index 0811952fa..f6b5ff16c 100644 --- a/packages/components/src/hooks/AGENTS.md +++ b/packages/components/src/hooks/AGENTS.md @@ -84,7 +84,14 @@ this file; edit `AGENTS.md` only. queue minutes of React work behind a long active turn and retain every obsolete history tree. Control state (`session`, message queue, fork, preview, external cursor) stays synchronous; do not delay it behind transcript rendering or - restore a direct `setState` for history-only mirror events. + restore a direct `setState` for history-only mirror events. It also returns + `conversationView` (the store's windowed history reader, `null` on the + full-Mirror rollback path). Prefer it over `doc.history`: on the view path + that property is a lazy bridge whose first read hydrates the whole + transcript. `use-conversation-view-selector.ts` derives values from the + view per version and takes the doc-state fallback from `useViewFallback`, + which only evaluates while there is no view; `use-session-turn-selectors.ts` + hosts the shared ones (active assistant turn, system notices). - Code Collab file-index hooks borrow owner-session resources from the workspace-owned Effect `ScopedCache`; do not open, scan, subscribe, or join the same Flock once per React mount. The resource subscribes before its cold diff --git a/packages/components/src/hooks/use-conversation-view-selector.ts b/packages/components/src/hooks/use-conversation-view-selector.ts new file mode 100644 index 000000000..2d0aa8476 --- /dev/null +++ b/packages/components/src/hooks/use-conversation-view-selector.ts @@ -0,0 +1,74 @@ +import { useCallback, useMemo, useRef, useSyncExternalStore } from 'react'; +import type { ConversationView } from '@/lib/conversation-view'; + +const noopSubscribe = () => () => {}; + +/** + * Derive a value from a `ConversationView` and re-render when the view + * changes. `select` runs at most once per view version (and per `revision`, + * for callers that hydrate turns themselves, which does not bump the view); + * its result is reused until then, which is what `useSyncExternalStore` + * needs from a snapshot. Pass `isEqual` for derived objects so an unchanged + * answer keeps its identity across versions. + * + * With no view (flag off, or the store not loaded yet) the hook returns + * `fallback`, which the caller derives from the doc state instead. + */ +export function useConversationViewSelector( + view: ConversationView | null | undefined, + select: (view: ConversationView) => T, + fallback: T, + options: { isEqual?: (previous: T, next: T) => boolean; revision?: number } = {} +): T { + const { isEqual = Object.is, revision = 0 } = options; + const selectRef = useRef(select); + selectRef.current = select; + const isEqualRef = useRef(isEqual); + isEqualRef.current = isEqual; + const cacheRef = useRef<{ + view: ConversationView; + version: number; + revision: number; + value: T; + } | null>(null); + + const subscribe = useCallback( + (onStoreChange: () => void) => (view ? view.subscribe(() => onStoreChange()) : noopSubscribe()), + [view] + ); + const getSnapshot = useCallback((): T => { + if (!view) return fallback; + const cached = cacheRef.current; + if ( + cached && + cached.view === view && + cached.version === view.version && + cached.revision === revision + ) { + return cached.value; + } + const next = selectRef.current(view); + const value = + cached && cached.view === view && isEqualRef.current(cached.value, next) + ? cached.value + : next; + cacheRef.current = { view, version: view.version, revision, value }; + return value; + }, [fallback, revision, view]); + + return useSyncExternalStore(subscribe, getSnapshot, getSnapshot); +} + +/** + * Evaluate `computeFallback` only while there is no view, so a bridged + * `doc.history` getter is never touched on the view path. `deps` are the + * doc-state inputs the fallback reads. + */ +export function useViewFallback( + view: ConversationView | null | undefined, + computeFallback: () => T, + deps: readonly unknown[] +): T | undefined { + // eslint-disable-next-line react-hooks/exhaustive-deps + return useMemo(() => (view ? undefined : computeFallback()), [view, ...deps]); +} diff --git a/packages/components/src/hooks/use-remove-local-project.ts b/packages/components/src/hooks/use-remove-local-project.ts index 1a9ea2f09..da6cc052f 100644 --- a/packages/components/src/hooks/use-remove-local-project.ts +++ b/packages/components/src/hooks/use-remove-local-project.ts @@ -10,7 +10,6 @@ import { getServerNow, isActiveSessionStatus, machineFlockKeys, - resolveActiveAssistantTurnId, type LocalProjectId, type LocalProjectMeta, type LocalProjectWorktreeCleanupPreflightResult, @@ -27,6 +26,7 @@ import { } from '@/hooks/use-machine-flock-rows'; import { useVisibleSessionMetas } from '@/hooks/use-visible-session-metas'; import { useSessionActions } from '@/hooks/use-session-actions'; +import { readActiveAssistantTurnId } from '@/lib/session-store-history'; import { getLocalProjectVisibilityKey } from '@/lib/visible-local-project-index'; export type RemoveLocalProjectTarget = { @@ -150,7 +150,7 @@ export function useRemoveLocalProject() { const sessionId = session.id as SessionId; const activeAssistantTurnId = await runtime.withSessionStore( sessionId, - (sessionStore) => resolveActiveAssistantTurnId(sessionStore.getState().history) + (sessionStore) => readActiveAssistantTurnId(sessionStore) ); if (!activeAssistantTurnId) return; await requestSessionCancel(sessionId, activeAssistantTurnId); diff --git a/packages/components/src/hooks/use-session-actions.ts b/packages/components/src/hooks/use-session-actions.ts index eb5f4cb35..f8cf7713f 100644 --- a/packages/components/src/hooks/use-session-actions.ts +++ b/packages/components/src/hooks/use-session-actions.ts @@ -56,6 +56,8 @@ import { getRpcDeliveredTurnKey, rpcDeliveredTurnsAtom, } from '@/atoms/session-dispatch-delivery'; +import { patchHistoryEntry } from '@/lib/conversation-view'; +import { readSessionUserTurn } from '@/lib/session-store-history'; import { resolveSessionCreateRepoFullName } from '@/lib/session-repo'; import { collectSessionLifecycleIds } from '@/lib/session-lifecycle'; import { capturePostHogEvent } from '@/lib/posthog-analytics'; @@ -812,9 +814,7 @@ export function useSessionActions(): SessionActions { throw new Error('Runtime not ready'); } const entry = await runtime.withSessionStore(sessionId, (sessionStore) => - sessionStore - .getState() - .history.find((item) => item.id === userTurnId && item.role === 'user') + readSessionUserTurn(sessionStore, userTurnId) ); const inputConfig = options?.inputConfig ?? normalizeSessionTurnInputConfig(entry?.inputConfig); @@ -920,9 +920,7 @@ export function useSessionActions(): SessionActions { throw new Error('Runtime not ready'); } const entry = await runtime.withSessionStore(sessionId, (sessionStore) => - sessionStore - .getState() - .history.find((item) => item.id === userTurnId && item.role === 'user') + readSessionUserTurn(sessionStore, userTurnId) ); const inputConfig = normalizeSessionTurnInputConfig(entry?.inputConfig); const userId = entry?.userId?.trim(); @@ -960,6 +958,21 @@ export function useSessionActions(): SessionActions { // Re-acquire the store for the write: the steer RPC above can run long, // and we must not hold a store ref across it. const promoted = await runtime.withSessionStore(sessionId, (sessionStore) => { + const view = sessionStore.conversationView; + if (view) { + // `status` is an index field, so the precondition reads the doc's + // current value without hydrating the turn; the patch writes only + // the two scalars, leaving the turn's items untouched. + const index = view.indexOf(userTurnId); + const row = index >= 0 ? view.index(index) : undefined; + if (!row || row.role !== 'user' || row.status !== 'pending_apply') return false; + return patchHistoryEntry( + sessionStore.doc, + userTurnId, + { status: 'pending', read: false }, + index + ); + } let didPromote = false; sessionStore.setState((draft: SessionDocMeta) => { const pendingEntry = draft.history.find( diff --git a/packages/components/src/hooks/use-session-doc.ts b/packages/components/src/hooks/use-session-doc.ts index 86e1f2873..a9e919ae0 100644 --- a/packages/components/src/hooks/use-session-doc.ts +++ b/packages/components/src/hooks/use-session-doc.ts @@ -16,8 +16,11 @@ import { type SessionDocStore, } from '@/atoms/runtime'; import { browserOnlineAtom } from '@/atoms/control-connection'; +import type { ConversationView } from '@/lib/conversation-view'; import type { RoomSyncState } from '@/lib/room-sync-state'; import { subscribeLatestOnAnimationFrame } from '@/lib/latest-frame-subscription'; +import { readSessionHistoryEntry, readSessionHistoryLength } from '@/lib/session-store-history'; +import { readSessionDocHistoryRevision } from '../providers/session-doc-state-source'; declare global { interface Window { @@ -39,6 +42,13 @@ export type PushMessageQueueInput = Omit< export type UseSessionDocResult = { doc: SessionDocState; + /** + * Windowed history reader for this session, once the store is loaded and + * the store runs on the control-plane Mirror; `null` on the full-Mirror + * rollback path. Prefer it over `doc.history`, which on the view path is a + * lazy bridge that materializes the whole transcript on first access. + */ + conversationView: ConversationView | null; addHistory: ( history: Omit & { id?: string }, options?: { dispatch?: boolean } @@ -88,12 +98,14 @@ export function sessionMetaSuggestsHistory(session: SessionHistoryHint | null | ); } +// Compared through the revision marker, never `history` itself: on the view +// path that property is a getter whose first read hydrates the transcript. const updateOnlyChangesHistory = ( previous: SessionDocState | undefined, next: SessionDocState ): boolean => previous !== undefined && - previous.history !== next.history && + readSessionDocHistoryRevision(previous) !== readSessionDocHistoryRevision(next) && previous.session === next.session && previous.mq === next.mq && previous.forkOperation === next.forkOperation && @@ -371,14 +383,11 @@ export function useSessionDoc( // The updater is a function that can't cross the intent wire; resolve it to // the concrete replacement entry against the current snapshot and send that // through the writer seam. Preserve the "not found → no-op" short-circuit. - const history = await withStore( - (store) => (store.getState().history ?? []) as SessionHistoryInput[] - ); - const index = history.findIndex((entry) => entry.id === historyId); - if (index < 0) { + const current = await withStore((store) => readSessionHistoryEntry(store, historyId)); + if (!current) { return; } - const nextEntry = updater(history[index] as SessionHistoryInput); + const nextEntry = updater(current as SessionHistoryInput); if (!runtime) { throw new Error('Runtime not ready'); } @@ -397,6 +406,7 @@ export function useSessionDoc( return { doc: state, + conversationView: loadedStore?.conversationView ?? null, addHistory, pushMessageQueue, removeMessageQueueItem, @@ -447,14 +457,14 @@ export function useSessionDocSyncState( return; } - const readHasLocalHistory = (state: SessionDocState) => (state.history?.length ?? 0) > 0; - setHasLocalHistory(readHasLocalHistory(store.getState())); + const readHasLocalHistory = () => readSessionHistoryLength(store) > 0; + setHasLocalHistory(readHasLocalHistory()); setSyncState(store.getSyncState()); setReady(true); releaseSync = store.acquireSync(); - unsubscribeStore = store.subscribe((nextState) => { + unsubscribeStore = store.subscribe(() => { if (!cancelled) { - const nextHasLocalHistory = readHasLocalHistory(nextState); + const nextHasLocalHistory = readHasLocalHistory(); setHasLocalHistory((prev) => prev === nextHasLocalHistory ? prev : nextHasLocalHistory ); diff --git a/packages/components/src/hooks/use-session-turn-selectors.ts b/packages/components/src/hooks/use-session-turn-selectors.ts new file mode 100644 index 000000000..4da9f6409 --- /dev/null +++ b/packages/components/src/hooks/use-session-turn-selectors.ts @@ -0,0 +1,77 @@ +import { useEffect, useState } from 'react'; +import { resolveActiveAssistantTurnId, type SessionHistory } from '@lody/shared'; +import type { SessionDocState } from '@/atoms/runtime'; +import { + findSystemNotice, + resolveActiveAssistantTurnIdFromView, + type ConversationView, + type SystemNoticeSearch, +} from '@/lib/conversation-view'; +import { useConversationViewSelector, useViewFallback } from './use-conversation-view-selector'; + +type SessionDocSource = { + doc: SessionDocState; + conversationView?: ConversationView | null; +}; + +/** + * The open assistant turn's id (see `resolveActiveAssistantTurnId`), from + * index rows on the view path and from `doc.history` on the rollback path. + */ +export function useActiveAssistantTurnId(source: SessionDocSource): string | undefined { + const view = source.conversationView ?? null; + const fallback = useViewFallback( + view, + () => resolveActiveAssistantTurnId(source.doc.history), + [source.doc] + ); + return useConversationViewSelector(view, resolveActiveAssistantTurnIdFromView, fallback); +} + +const NOT_FOUND: SystemNoticeSearch = { found: false, unhydratedSystemTurnIndex: null }; +const FOUND: SystemNoticeSearch = { found: true }; + +const hasSystemNoticeInHistory = (history: readonly SessionHistory[], name: string): boolean => + history.some((entry) => + (entry.items ?? []).some((item) => item.type === 'system_notice' && item.name === name) + ); + +const searchEqual = (left: SystemNoticeSearch, right: SystemNoticeSearch): boolean => + left.found === right.found && + (left.found || right.found || left.unhydratedSystemTurnIndex === right.unhydratedSystemTurnIndex); + +/** + * Whether the session carries the `system_notice` named `name`. Only system + * turns are inspected; one that is not hydrated is hydrated on its own + * (one `toJSON`, no version bump) and the search re-runs. + */ +export function useSessionSystemNotice(source: SessionDocSource, name: string): boolean { + const view = source.conversationView ?? null; + const fallback = useViewFallback( + view, + () => + hasSystemNoticeInHistory(source.doc.history as readonly SessionHistory[], name) + ? FOUND + : NOT_FOUND, + [source.doc, name] + ); + const [hydrationRevision, setHydrationRevision] = useState(0); + const search = useConversationViewSelector( + view, + (current) => findSystemNotice(current, name), + fallback ?? NOT_FOUND, + { isEqual: searchEqual, revision: hydrationRevision } + ); + const pendingIndex = search.found ? null : search.unhydratedSystemTurnIndex; + useEffect(() => { + if (!view || pendingIndex === null) return undefined; + let cancelled = false; + void view.ensureRange(pendingIndex, pendingIndex + 1).then(() => { + if (!cancelled) setHydrationRevision((revision) => revision + 1); + }); + return () => { + cancelled = true; + }; + }, [pendingIndex, view]); + return search.found; +} diff --git a/packages/components/src/hooks/use-task-actions.ts b/packages/components/src/hooks/use-task-actions.ts index 3d283fef6..12349a0c4 100644 --- a/packages/components/src/hooks/use-task-actions.ts +++ b/packages/components/src/hooks/use-task-actions.ts @@ -29,6 +29,7 @@ import { type SessionId, } from '@lody/shared'; import { userAtom } from '@/atoms'; +import { readSessionHistoryEntry } from '@/lib/session-store-history'; import { activeWorkspaceRuntimeAtom, type TaskDocStore, type WorkspaceRuntime } from '@/atoms/runtime'; import { taskIndexRowsAtom, taskListAtom } from '@/atoms/tasks'; @@ -448,7 +449,7 @@ export function useTaskActions() { return; } const entry = await runtime.withSessionStore(sessionId, (sessionStore) => - sessionStore.getState().history.find((item) => item.id === entryId) + readSessionHistoryEntry(sessionStore, entryId) ); if (!entry) { return; diff --git a/packages/components/src/lib/conversation-view/AGENTS.md b/packages/components/src/lib/conversation-view/AGENTS.md new file mode 100644 index 000000000..7001e274b --- /dev/null +++ b/packages/components/src/lib/conversation-view/AGENTS.md @@ -0,0 +1,41 @@ +# `lib/conversation-view` — windowed session history + +`CLAUDE.md` is a symlink to this file. Edit `AGENTS.md` only. Package +[AGENTS.md](../../../AGENTS.md) applies. + +`ConversationView` is the client's read model over a session doc's `history` +list, and `history-writer.ts` is its only write path. Together they let a +session store run on `sessionControlDocSchema` (history ignored), so opening a +long conversation never materializes every turn through a Mirror. + +## Contracts + +- `index(i)` is always loaded and comes from the turn map's shallow value plus + `summary`, `itemCount` and `planCount`. Add a field to `TURN_INDEX_FIELDS` + only when a reader that must stay O(1) needs it; every field costs one + shallow read per turn at open. +- `turn(i)` is synchronous only for hydrated turns. Hydration is per-turn + `toJSON()`; the LRU never evicts the tail (`tailKeep`), a `retain()`ed + range, or the range an `ensureRange()` call just asked for. A caller that + awaits `ensureRange(a, b)` must find every turn in it. +- `fileDiff(i)` reads one small container per turn and is cached until that + turn's `fileDiff` changes, so the diff summary never hydrates items. +- `readAll()` is the deliberate full-transcript read (markdown export, search + index, the `doc.history` bridge). First read is O(n) `toJSON`; later reads + re-materialize only changed turns and hand back the same object for every + other turn, because `buildChatStreamItems` and the outline are keyed on + entry identity. It holds one object per turn for the life of the view. +- Doc events keep the index and hydrated turns current and bump `version` on + every change; `subscribe` listeners see `index` / `range` / `tail`. + Hydration itself never bumps `version` — hooks that hydrate on their own + pass a `revision` to `useConversationViewSelector`. +- Writers (`appendHistoryEntry`, `replaceHistoryEntry`, `patchHistoryEntry`, + `respondHistoryPermission`) produce exactly the container shapes a + full-schema `Mirror.setState` produces (pinned by + `tests/conversation-view-history-writer.test.ts`). `replaceHistoryEntry` + drops and recreates the nested containers it carries; use + `patchHistoryEntry` for a scalar flip so items keep their containers. +- `sessionControlDocSchema` keeps `history` declared as `Ignore` so + `ignoreUnknownProperties` root mirroring does not re-materialize it. A + Mirror `setState` that reaches an ignored field is memory-only, which is why + `providers/session-doc-state-source.ts` throws on it. diff --git a/packages/components/src/lib/conversation-view/CLAUDE.md b/packages/components/src/lib/conversation-view/CLAUDE.md new file mode 120000 index 000000000..47dc3e3d8 --- /dev/null +++ b/packages/components/src/lib/conversation-view/CLAUDE.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file diff --git a/packages/components/src/lib/conversation-view/control-doc-schema.ts b/packages/components/src/lib/conversation-view/control-doc-schema.ts new file mode 100644 index 000000000..9f3e7ddec --- /dev/null +++ b/packages/components/src/lib/conversation-view/control-doc-schema.ts @@ -0,0 +1,25 @@ +import { schema, type ContainerSchemaType } from 'loro-mirror'; +import { sessionDocSchema } from '@lody/shared'; + +/** + * The session doc schema with `history` declared as an ignored root. + * + * A Mirror over the full `sessionDocSchema` materializes every history + * container at construction (357–521 ms on a 171-turn doc with 70k + * containers), which is what makes long conversations freeze on open. The + * renderer reads history through `ConversationView` instead, so the Mirror + * that backs the control plane (session, mq, preview, fork, runtime config) + * must never touch the history list. `schema.Ignore()` keeps the key declared + * — so `ignoreUnknownProperties` root mirroring does not re-materialize it — + * while `buildRootStateSnapshot` skips it. + * + * Writes to history must go through `HistoryWriter`: a `setState` that + * touches an ignored field is memory-only and would silently not persist. + */ +export const sessionControlDocSchema = schema({ + ...sessionDocSchema.definition, + // `RootSchemaDefinition` is typed as containers only, but the runtime treats + // an `ignore` root exactly like an ignored map field (memory-only, skipped by + // `buildRootStateSnapshot`); `control-mirror.test.ts` pins that behavior. + history: schema.Ignore() as unknown as ContainerSchemaType, +}); diff --git a/packages/components/src/lib/conversation-view/control-mirror.ts b/packages/components/src/lib/conversation-view/control-mirror.ts new file mode 100644 index 000000000..abced4a53 --- /dev/null +++ b/packages/components/src/lib/conversation-view/control-mirror.ts @@ -0,0 +1,51 @@ +import { Mirror } from 'loro-mirror'; +import type { LoroDoc, LoroEventBatch } from 'loro-crdt'; +import type { SessionId } from '@lody/shared'; +import { sessionControlDocSchema } from './control-doc-schema'; + +export type SessionControlMirror = Mirror; + +const HISTORY_ROOT = 'history'; + +const isHistoryEvent = (event: LoroEventBatch['events'][number], historyRootId: string): boolean => + event.target === historyRootId || + (Array.isArray(event.path) && event.path[0] === HISTORY_ROOT); + +/** + * The control-plane Mirror: `sessionControlDocSchema` over the session doc. + * + * loro-mirror 2.3.1 honors an `Ignore` root when it builds its initial + * snapshot, but its incremental event path does not: a doc event under + * `history` is applied to the state anyway, which materializes the event's + * delta into a partial `history` array and registers every container the + * delta carries. Over a streaming session that is the whole transcript, + * item by item, plus a registry that grows with every new container — the + * exact cost the control schema exists to avoid, and a partial array whose + * indices do not match the doc's. This installs an instance-level filter on + * the two methods the Mirror runs on every batch so history events never + * reach them. `ConversationView` owns those events. + */ +export function createSessionControlMirror(doc: LoroDoc, sessionId: SessionId): SessionControlMirror { + const mirror = new Mirror({ + doc, + schema: sessionControlDocSchema, + // Tolerate root keys written by peers running a newer schema version. + ignoreUnknownProperties: true, + initialState: { session: { id: sessionId } }, + debug: false, + }); + const historyRootId = doc.getList(HISTORY_ROOT).id as string; + const withoutHistory = (batch: LoroEventBatch): LoroEventBatch => ({ + ...batch, + events: batch.events.filter((event) => !isHistoryEvent(event, historyRootId)), + }); + const internals = mirror as unknown as { + normalizeLoroEventBatch: (batch: LoroEventBatch) => LoroEventBatch; + registerContainersFromLoroEvent: (batch: LoroEventBatch) => void; + }; + const normalize = internals.normalizeLoroEventBatch.bind(mirror); + const register = internals.registerContainersFromLoroEvent.bind(mirror); + internals.normalizeLoroEventBatch = (batch) => normalize(withoutHistory(batch)); + internals.registerContainersFromLoroEvent = (batch) => register(withoutHistory(batch)); + return mirror; +} diff --git a/packages/components/src/lib/conversation-view/conversation-view.ts b/packages/components/src/lib/conversation-view/conversation-view.ts new file mode 100644 index 000000000..53f9674c7 --- /dev/null +++ b/packages/components/src/lib/conversation-view/conversation-view.ts @@ -0,0 +1,404 @@ +import type { LoroDoc, LoroEventBatch, LoroList, LoroMap } from 'loro-crdt'; +import type { FileDiff, SessionHistory, SessionId, TurnSummary } from '@lody/shared'; + +/** + * Fields of a turn that are always loaded, straight from the turn map's shallow + * value (one cheap wasm call per turn). Everything the outline rail, the + * folded-turn header and Virtua's height estimate need lives here. + */ +export const TURN_INDEX_FIELDS = [ + 'id', + 'role', + 'timestamp', + 'status', + 'finished', + 'endedAt', + 'sendStatus', + 'userTurnId', + 'acpTurnId', +] as const; + +export type TurnIndexRow = Pick & { + summary?: TurnSummary; + itemCount?: number; + planCount?: number; +}; + +export type ConversationViewChange = + | { kind: 'index' } + | { kind: 'range'; from: number; to: number } + | { kind: 'tail'; from: number; to: number }; + +/** + * Windowed read model over a session doc's `history` list. + * + * `index(i)` is always available; `turn(i)` is synchronous only for hydrated + * turns. Turns are keyed by container id, never by position, so a concurrent + * insert in the middle of the list shifts indices without invalidating + * hydrated data. + */ +export interface ConversationView { + readonly sessionId: SessionId; + readonly turnCount: number; + /** Bumps on any structural or index-field change. */ + readonly version: number; + index(i: number): TurnIndexRow | undefined; + indexOf(turnId: string): number; + turn(i: number): SessionHistory | undefined; + isHydrated(i: number): boolean; + /** + * The turn's `fileDiff` alone, read from its own small container and cached + * per turn until that turn changes. Lets the diff summary see every turn's + * edits without hydrating a single message item. + */ + fileDiff(i: number): FileDiff[] | undefined; + ensureRange(from: number, to: number): Promise; + /** + * Keep `[from, to)` out of LRU eviction until the returned release runs. A + * mounted renderer window retains what it shows; an active search retains + * everything it hydrated. + */ + retain(from: number, to: number): () => void; + release(from: number, to: number): void; + /** + * Every turn, hydrated. O(n) in turns on the first read; later reads only + * re-hydrate the turns that changed and hand back the same objects for the + * rest, so identity-keyed caches downstream keep hitting. Cached per + * `version`, so repeated reads between changes are free. Still the + * deliberate full-transcript escape hatch (markdown export, search index): + * it holds one JS object per turn for the life of the view. + */ + readAll(): SessionHistory[]; + subscribe(listener: (change: ConversationViewChange) => void): () => void; + dispose(): void; +} + +export type CreateConversationViewOptions = { + sessionId: SessionId; + /** Full turns kept in memory beyond subscribed ranges and the tail. */ + maxHydrated?: number; + /** Trailing turns that are always hydrated and never evicted. */ + tailKeep?: number; +}; + +const DEFAULT_MAX_HYDRATED = 200; +const DEFAULT_TAIL_KEEP = 20; + +type HydratedTurn = { value: SessionHistory; lastUsed: number }; + +const isContainerId = (value: unknown): value is string => + typeof value === 'string' && value.startsWith('cid:'); + +const readTurnMap = (doc: LoroDoc, cid: string): LoroMap | null => { + const container = doc.getContainerById(cid as never); + return container && container.kind() === 'Map' ? (container as LoroMap) : null; +}; + +const containerLength = (doc: LoroDoc, value: unknown): number | undefined => { + if (isContainerId(value)) { + const container = doc.getContainerById(value as never); + return container && container.kind() === 'List' ? (container as LoroList).length : undefined; + } + return Array.isArray(value) ? value.length : undefined; +}; + +/** + * Same normalization `SessionDocument.getHistory()` applies on the CLI: the + * Mirror state never carries `undefined` for absent optional fields either. + */ +const toSessionHistory = (value: Record): SessionHistory => + value as unknown as SessionHistory; + +export function createConversationViewFromDoc( + doc: LoroDoc, + options: CreateConversationViewOptions +): ConversationView { + const maxHydrated = options.maxHydrated ?? DEFAULT_MAX_HYDRATED; + const tailKeep = options.tailKeep ?? DEFAULT_TAIL_KEEP; + const list = doc.getList('history') as LoroList; + + let ids: (string | null)[] = []; + let indexRows: (TurnIndexRow | undefined)[] = []; + const positionById = new Map(); + const hydrated = new Map(); + /** + * Turn objects handed out by `readAll()`, kept beside the LRU so a full read + * never re-materializes an unchanged turn and never changes its identity. + * Entries leave only when their turn changes or its container leaves the + * list. Empty until the first full read. + */ + const snapshotByCid = new Map(); + const fileDiffByCid = new Map(); + const listeners = new Set<(change: ConversationViewChange) => void>(); + const subscribedRanges = new Set<{ from: number; to: number }>(); + let version = 0; + let clock = 0; + let allCache: { version: number; value: SessionHistory[] } | null = null; + let disposed = false; + + const readIndexRow = (cid: string): TurnIndexRow | undefined => { + const map = readTurnMap(doc, cid); + if (!map) return undefined; + const shallow = map.getShallowValue() as Record; + const row: Record = {}; + for (const field of TURN_INDEX_FIELDS) { + const value = shallow[field]; + if (value !== undefined && !isContainerId(value)) row[field] = value; + } + const summary = shallow.summary; + if (summary !== undefined) { + row.summary = isContainerId(summary) + ? (doc.getContainerById(summary as never)?.toJSON() as TurnSummary | undefined) + : summary; + } + const itemCount = containerLength(doc, shallow.items); + if (itemCount !== undefined) row.itemCount = itemCount; + const planCount = containerLength(doc, shallow.plan); + if (planCount !== undefined) row.planCount = planCount; + return row as TurnIndexRow; + }; + + const rebuildIndex = () => { + const shallow = list.getShallowValue() as unknown[]; + const nextIds: (string | null)[] = new Array(shallow.length); + const nextRows: (TurnIndexRow | undefined)[] = new Array(shallow.length); + positionById.clear(); + for (let i = 0; i < shallow.length; i += 1) { + const cid = shallow[i]; + if (!isContainerId(cid)) { + nextIds[i] = null; + nextRows[i] = undefined; + continue; + } + nextIds[i] = cid; + // Reuse the previous row when the container did not move so a structural + // change costs one shallow read for the list, not one per turn. + const previousPosition = ids.indexOf(cid); + const row = + previousPosition >= 0 && previousPosition === i + ? (indexRows[previousPosition] ?? readIndexRow(cid)) + : readIndexRow(cid); + nextRows[i] = row; + if (row?.id) positionById.set(row.id, i); + } + // Drop hydrated turns whose container left the list. + const live = new Set(nextIds.filter((cid): cid is string => cid !== null)); + for (const cid of hydrated.keys()) { + if (!live.has(cid)) hydrated.delete(cid); + } + for (const cid of snapshotByCid.keys()) { + if (!live.has(cid)) snapshotByCid.delete(cid); + } + for (const cid of fileDiffByCid.keys()) { + if (!live.has(cid)) fileDiffByCid.delete(cid); + } + ids = nextIds; + indexRows = nextRows; + }; + + const materialize = (i: number): SessionHistory | undefined => { + const cid = ids[i]; + if (!cid) return undefined; + const map = readTurnMap(doc, cid); + if (!map) return undefined; + return toSessionHistory(map.toJSON() as Record); + }; + + const hydrateOne = (i: number): SessionHistory | undefined => { + const cid = ids[i]; + const value = materialize(i); + if (!cid || !value) return undefined; + hydrated.set(cid, { value, lastUsed: (clock += 1) }); + return value; + }; + + const isProtected = (i: number, keep?: { from: number; to: number }): boolean => { + if (i >= ids.length - tailKeep) return true; + if (keep && i >= keep.from && i < keep.to) return true; + for (const range of subscribedRanges) { + if (i >= range.from && i < range.to) return true; + } + return false; + }; + + /** + * Trim to `maxHydrated`, least recently used first. The tail, subscribed + * ranges and the range a caller just asked for (`keep`) are never evicted: + * a caller that awaits `ensureRange(a, b)` must find every turn in it. + */ + const evict = (keep?: { from: number; to: number }) => { + if (hydrated.size <= maxHydrated) return; + const candidates: { cid: string; lastUsed: number }[] = []; + for (const [cid, entry] of hydrated) { + const position = ids.indexOf(cid); + if (position >= 0 && isProtected(position, keep)) continue; + candidates.push({ cid, lastUsed: entry.lastUsed }); + } + candidates.sort((left, right) => left.lastUsed - right.lastUsed); + let excess = hydrated.size - maxHydrated; + for (const candidate of candidates) { + if (excess <= 0) break; + hydrated.delete(candidate.cid); + excess -= 1; + } + }; + + const bump = (change: ConversationViewChange) => { + version += 1; + allCache = null; + for (const listener of listeners) listener(change); + }; + + const refreshTurn = (i: number, fieldTouched: string | null) => { + const cid = ids[i]; + if (!cid) return; + if ( + fieldTouched === null || + (TURN_INDEX_FIELDS as readonly string[]).includes(fieldTouched) || + fieldTouched === 'summary' || + fieldTouched === 'items' || + fieldTouched === 'plan' + ) { + indexRows[i] = readIndexRow(cid); + } + if (fieldTouched === null || fieldTouched === 'fileDiff') fileDiffByCid.delete(cid); + // Any change to the turn invalidates the object a full read handed out; + // the next `readAll()` re-materializes exactly this turn. + snapshotByCid.delete(cid); + if (hydrated.has(cid)) hydrateOne(i); + }; + + const handleEvent = (batch: LoroEventBatch) => { + if (disposed) return; + let structural = false; + let rangeFrom = Number.POSITIVE_INFINITY; + let rangeTo = -1; + for (const event of batch.events) { + const path = event.path as unknown[]; + if (path[0] !== 'history') continue; + if (path.length === 1) { + structural = true; + continue; + } + const position = path[1]; + if (typeof position !== 'number') continue; + const fieldTouched = path.length >= 3 && typeof path[2] === 'string' ? path[2] : null; + // A path of exactly ['history', i] is the turn map itself (a scalar + // field set), which may include index fields. + refreshTurn(position, path.length === 2 ? null : fieldTouched); + rangeFrom = Math.min(rangeFrom, position); + rangeTo = Math.max(rangeTo, position + 1); + } + if (structural) { + rebuildIndex(); + bump({ kind: 'index' }); + return; + } + if (rangeTo >= 0) { + const isTail = rangeTo > ids.length - tailKeep; + bump({ kind: isTail ? 'tail' : 'range', from: rangeFrom, to: rangeTo }); + } + }; + + rebuildIndex(); + // The tail is what streaming touches; keep it warm from the start. + for (let i = Math.max(0, ids.length - tailKeep); i < ids.length; i += 1) hydrateOne(i); + const unsubscribeDoc = doc.subscribe(handleEvent); + + const view: ConversationView = { + sessionId: options.sessionId, + get turnCount() { + return ids.length; + }, + get version() { + return version; + }, + index: (i) => indexRows[i], + indexOf: (turnId) => positionById.get(turnId) ?? -1, + turn: (i) => { + const cid = ids[i]; + if (!cid) return undefined; + const entry = hydrated.get(cid); + if (!entry) return undefined; + entry.lastUsed = clock += 1; + return entry.value; + }, + isHydrated: (i) => { + const cid = ids[i]; + return cid !== null && cid !== undefined && hydrated.has(cid); + }, + fileDiff: (i) => { + const cid = ids[i]; + if (!cid) return undefined; + const cached = fileDiffByCid.get(cid); + if (cached) return cached; + const map = readTurnMap(doc, cid); + if (!map) return undefined; + const raw = map.get('fileDiff'); + const value = + raw && typeof raw === 'object' && 'kind' in (raw as object) + ? (raw as LoroList).toJSON() + : raw; + const fileDiff = (Array.isArray(value) ? value : []) as FileDiff[]; + fileDiffByCid.set(cid, fileDiff); + return fileDiff; + }, + ensureRange: async (from, to) => { + const start = Math.max(0, from); + const end = Math.min(ids.length, to); + for (let i = start; i < end; i += 1) { + const cid = ids[i]; + if (cid && !hydrated.has(cid)) hydrateOne(i); + } + evict({ from: start, to: end }); + }, + retain: (from, to) => { + const range = { from: Math.max(0, from), to }; + subscribedRanges.add(range); + return () => { + subscribedRanges.delete(range); + }; + }, + release: (from, to) => { + for (let i = Math.max(0, from); i < Math.min(ids.length, to); i += 1) { + const cid = ids[i]; + if (cid && !isProtected(i)) hydrated.delete(cid); + } + }, + readAll: () => { + if (allCache && allCache.version === version) return allCache.value; + const value: SessionHistory[] = []; + for (let i = 0; i < ids.length; i += 1) { + const cid = ids[i]; + if (!cid) continue; + // The LRU holds the freshest object for a hydrated turn (events + // re-hydrate in place); the snapshot holds what an earlier full read + // saw for everything else and is cleared per turn on change. + let entry = hydrated.get(cid)?.value ?? snapshotByCid.get(cid); + if (!entry) entry = materialize(i); + if (!entry) continue; + snapshotByCid.set(cid, entry); + value.push(entry); + } + allCache = { version, value }; + return value; + }, + subscribe: (listener) => { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + dispose: () => { + disposed = true; + unsubscribeDoc(); + listeners.clear(); + hydrated.clear(); + snapshotByCid.clear(); + fileDiffByCid.clear(); + subscribedRanges.clear(); + allCache = null; + }, + }; + return view; +} diff --git a/packages/components/src/lib/conversation-view/history-writer.ts b/packages/components/src/lib/conversation-view/history-writer.ts new file mode 100644 index 000000000..191b5650d --- /dev/null +++ b/packages/components/src/lib/conversation-view/history-writer.ts @@ -0,0 +1,376 @@ +import { + LoroList, + LoroMap, + LoroMovableList, + LoroText, + type Container, + type LoroDoc, +} from 'loro-crdt'; +import { sessionDocSchema, type SessionHistoryInput } from '@lody/shared'; + +/** + * Writes history entries straight into the Loro doc with the exact container + * shapes a full-schema `Mirror.setState` would have produced, so peers that + * still read through the full `sessionDocSchema` (the CLI, older clients) see + * byte-for-byte the same structure. + * + * The rules are loro-mirror's `initializeContainer` rules, restated: + * - `undefined` and `$cid` keys are skipped; + * - a declared container field (`loro-map` / `loro-list` / `loro-text`) becomes + * that container when the value has the matching shape, otherwise a plain + * value; + * - an `any` field infers: plain object → Map, array → List (MovableList when + * `defaultMovableList`), string → Text only when that `any` sets + * `defaultLoroText`; the `any`'s options are inherited by everything nested + * under it; + * - a map without a schema (created by inference) keeps inferring with the + * inherited options; a schema'd map's undeclared keys use its catchall, or a + * plain value when there is none; + * - primitives go through the field's `transform.encode` when one exists. + */ + +type InferOptions = { defaultLoroText?: boolean; defaultMovableList?: boolean }; + +type SchemaLike = { + type: string; + definition?: Record; + catchallType?: SchemaLike; + itemSchema?: SchemaLike; + options?: { + defaultLoroText?: boolean; + defaultMovableList?: boolean; + transform?: { encode?: (value: unknown) => unknown }; + }; + getContainerType?: () => string | null; +}; + +type ContainerKind = 'Map' | 'List' | 'MovableList' | 'Text'; + +const HISTORY_LIST_SCHEMA = ( + sessionDocSchema as unknown as { definition: Record } +).definition.history!; +const HISTORY_ENTRY_SCHEMA = HISTORY_LIST_SCHEMA.itemSchema!; + +const isPlainObject = (value: unknown): value is Record => + typeof value === 'object' && + value !== null && + !Array.isArray(value) && + !(value instanceof Date) && + !(value instanceof RegExp); + +const inferKind = (value: unknown, infer: InferOptions | undefined): ContainerKind | undefined => { + if (isPlainObject(value)) return 'Map'; + if (Array.isArray(value)) return infer?.defaultMovableList ? 'MovableList' : 'List'; + if (typeof value === 'string') return infer?.defaultLoroText ? 'Text' : undefined; + return undefined; +}; + +const schemaKind = (schema: SchemaLike): ContainerKind | undefined => { + switch (schema.type) { + case 'loro-map': + return 'Map'; + case 'loro-list': + return 'List'; + case 'loro-movable-list': + return 'MovableList'; + case 'loro-text': + return 'Text'; + default: + return undefined; + } +}; + +const matchesKind = (kind: ContainerKind, value: unknown): boolean => + kind === 'Map' + ? isPlainObject(value) + : kind === 'Text' + ? typeof value === 'string' + : Array.isArray(value); + +const inferFromAny = (schema: SchemaLike, base: InferOptions | undefined): InferOptions => ({ + ...base, + defaultLoroText: schema.options?.defaultLoroText ?? false, + ...(schema.options?.defaultMovableList !== undefined + ? { defaultMovableList: schema.options.defaultMovableList } + : {}), +}); + +const encode = (schema: SchemaLike | undefined, value: unknown): unknown => { + if (value === null || value === undefined) return value; + const transform = schema?.options?.transform; + return transform?.encode ? transform.encode(value) : value; +}; + +const newContainer = (kind: ContainerKind): Container => { + switch (kind) { + case 'Map': + return new LoroMap(); + case 'List': + return new LoroList(); + case 'MovableList': + return new LoroMovableList(); + case 'Text': + return new LoroText(); + default: { + const unreachable: never = kind; + throw new Error(`Unknown container kind: ${String(unreachable)}`); + } + } +}; + +/** Decide how one value is stored under `schema` (or inferred under `infer`). */ +const resolveWrite = ( + schema: SchemaLike | undefined, + value: unknown, + infer: InferOptions | undefined +): + | { mode: 'plain'; value: unknown } + | { + mode: 'container'; + kind: ContainerKind; + schema: SchemaLike | undefined; + infer: InferOptions | undefined; + } => { + if (schema?.type === 'any') { + const childInfer = inferFromAny(schema, infer); + const kind = inferKind(value, childInfer); + return kind + ? { mode: 'container', kind, schema: undefined, infer: childInfer } + : { mode: 'plain', value }; + } + if (schema) { + const kind = schemaKind(schema); + if (kind && matchesKind(kind, value)) { + return { mode: 'container', kind, schema, infer: undefined }; + } + return { mode: 'plain', value: encode(schema, value) }; + } + const kind = inferKind(value, infer); + return kind ? { mode: 'container', kind, schema: undefined, infer } : { mode: 'plain', value }; +}; + +const fillContainer = ( + container: Container, + schema: SchemaLike | undefined, + value: unknown, + infer: InferOptions | undefined +): void => { + switch (container.kind()) { + case 'Map': + writeMapEntries(container as LoroMap, value, schema, infer); + return; + case 'List': + case 'MovableList': + writeListItems(container as LoroList | LoroMovableList, value, schema, infer); + return; + case 'Text': + if (typeof value === 'string' && value.length > 0) { + (container as LoroText).insert(0, value); + } + return; + default: + return; + } +}; + +const setMapField = ( + map: LoroMap, + key: string, + value: unknown, + fieldSchema: SchemaLike | undefined, + infer: InferOptions | undefined +): void => { + const decision = resolveWrite(fieldSchema, value, infer); + if (decision.mode === 'plain') { + map.set(key, decision.value as never); + return; + } + const child = map.setContainer(key, newContainer(decision.kind) as never) as Container; + fillContainer(child, decision.schema, value, decision.infer); +}; + +const writeMapEntries = ( + map: LoroMap, + value: unknown, + mapSchema: SchemaLike | undefined, + infer: InferOptions | undefined +): void => { + if (!isPlainObject(value)) return; + const schema = mapSchema?.type === 'loro-map' ? mapSchema : undefined; + for (const [key, entry] of Object.entries(value)) { + if (key === '$cid' || entry === undefined) continue; + if (schema) { + const fieldSchema = schema.definition?.[key] ?? schema.catchallType; + // A schema'd map with neither a declared field nor a catchall stores the + // value as-is, without inference (loro-mirror `initializeContainer`). + if (!fieldSchema) { + map.set(key, entry as never); + continue; + } + setMapField(map, key, entry, fieldSchema, undefined); + continue; + } + setMapField(map, key, entry, undefined, infer); + } +}; + +const writeListItems = ( + list: LoroList | LoroMovableList, + value: unknown, + listSchema: SchemaLike | undefined, + infer: InferOptions | undefined +): void => { + if (!Array.isArray(value)) return; + const itemSchema = + listSchema && (listSchema.type === 'loro-list' || listSchema.type === 'loro-movable-list') + ? listSchema.itemSchema + : undefined; + for (const item of value) { + const decision = resolveWrite(itemSchema, item, infer); + if (decision.mode === 'plain') { + list.push(decision.value as never); + continue; + } + const child = list.pushContainer(newContainer(decision.kind) as never) as Container; + fillContainer(child, decision.schema, item, decision.infer); + } +}; + +const historyList = (doc: LoroDoc): LoroList => doc.getList('history'); + +/** A map field's value, resolving a nested container (e.g. an inferred `LoroText`) to its JSON. */ +const fieldValue = (map: LoroMap, key: string): unknown => { + const value = map.get(key); + return value && typeof value === 'object' && 'kind' in (value as object) + ? (value as Container).toJSON() + : value; +}; + +const turnMapAt = (doc: LoroDoc, index: number): LoroMap | null => { + const cid = (historyList(doc).getShallowValue() as unknown[])[index]; + if (typeof cid !== 'string' || !cid.startsWith('cid:')) return null; + const container = doc.getContainerById(cid as never); + return container && container.kind() === 'Map' ? (container as LoroMap) : null; +}; + +/** Position of the entry whose `id` is `entryId`, scanning shallow values. */ +export function findHistoryIndex(doc: LoroDoc, entryId: string): number { + const cids = historyList(doc).getShallowValue() as unknown[]; + for (let i = 0; i < cids.length; i += 1) { + const cid = cids[i]; + if (typeof cid !== 'string' || !cid.startsWith('cid:')) continue; + const map = doc.getContainerById(cid as never); + if (!map || map.kind() !== 'Map') continue; + if (fieldValue(map as LoroMap, 'id') === entryId) return i; + } + return -1; +} + +export function appendHistoryEntry(doc: LoroDoc, entry: SessionHistoryInput): void { + const map = historyList(doc).pushContainer(new LoroMap()) as LoroMap; + writeMapEntries(map, entry, HISTORY_ENTRY_SCHEMA, undefined); + doc.commit(); +} + +/** + * Replace the entry with `entryId` in place: fields the new entry carries are + * rewritten (a nested container is dropped and recreated), fields it omits are + * deleted. The turn map keeps its container id, so readers keyed by turn keep + * their identity. + */ +export function replaceHistoryEntry( + doc: LoroDoc, + entryId: string, + entry: SessionHistoryInput, + indexHint?: number +): boolean { + const map = resolveTurnMap(doc, entryId, indexHint); + if (!map) return false; + const next = entry as unknown as Record; + for (const key of map.keys()) { + if (next[key] === undefined) map.delete(key); + } + writeEntryFields(map, next); + doc.commit(); + return true; +} + +/** + * Set only the fields in `patch` on the entry with `entryId`; an explicit + * `undefined` deletes that field, and untouched fields keep their containers. + * The scalar-field counterpart of `replaceHistoryEntry`, for status flips + * that must not rewrite a turn's items. + */ +export function patchHistoryEntry( + doc: LoroDoc, + entryId: string, + patch: Partial, + indexHint?: number +): boolean { + const map = resolveTurnMap(doc, entryId, indexHint); + if (!map) return false; + const next = patch as unknown as Record; + for (const [key, value] of Object.entries(next)) { + if (key !== '$cid' && value === undefined) map.delete(key); + } + writeEntryFields(map, next); + doc.commit(); + return true; +} + +/** The entry's map: the hinted position when it still holds `entryId`, else a scan. */ +const resolveTurnMap = (doc: LoroDoc, entryId: string, indexHint?: number): LoroMap | null => { + if (indexHint !== undefined && indexHint >= 0) { + const hinted = turnMapAt(doc, indexHint); + if (hinted !== null && fieldValue(hinted, 'id') === entryId) return hinted; + } + const index = findHistoryIndex(doc, entryId); + return index < 0 ? null : turnMapAt(doc, index); +}; + +const writeEntryFields = (map: LoroMap, fields: Record): void => { + for (const [key, value] of Object.entries(fields)) { + if (key === '$cid' || value === undefined) continue; + const fieldSchema = HISTORY_ENTRY_SCHEMA.definition?.[key] ?? HISTORY_ENTRY_SCHEMA.catchallType; + if (!fieldSchema) { + map.set(key, value as never); + continue; + } + setMapField(map, key, value, fieldSchema, undefined); + } +}; + +/** + * Record a permission outcome on the item carrying `requestId`. Mirrors the + * previous `setState` path, which set `permissionRequest.outcome` on the item. + */ +export function respondHistoryPermission( + doc: LoroDoc, + requestId: string, + outcome: unknown, + indexHint?: number +): boolean { + const cids = historyList(doc).getShallowValue() as unknown[]; + const order = + indexHint !== undefined && indexHint >= 0 && indexHint < cids.length + ? [indexHint, ...cids.map((_, i) => i).filter((i) => i !== indexHint)] + : cids.map((_, i) => i); + for (const turnIndex of order) { + const map = turnMapAt(doc, turnIndex); + if (!map) continue; + const items = map.get('items'); + if (!(items instanceof LoroList)) continue; + for (let i = 0; i < items.length; i += 1) { + const item = items.get(i); + if (!(item instanceof LoroMap)) continue; + const request = item.get('permissionRequest'); + if (!(request instanceof LoroMap) || fieldValue(request, 'requestId') !== requestId) continue; + // The item map is inferred under the catchall `any` with defaultLoroText, + // so its nested writes keep inferring with that option. + setMapField(request, 'outcome', outcome, undefined, { defaultLoroText: true }); + doc.commit(); + return true; + } + } + return false; +} diff --git a/packages/components/src/lib/conversation-view/index.ts b/packages/components/src/lib/conversation-view/index.ts new file mode 100644 index 000000000..19dfa3ff1 --- /dev/null +++ b/packages/components/src/lib/conversation-view/index.ts @@ -0,0 +1,27 @@ +export { sessionControlDocSchema } from './control-doc-schema'; +export { createSessionControlMirror, type SessionControlMirror } from './control-mirror'; +export { + createConversationViewFromDoc, + TURN_INDEX_FIELDS, + type ConversationView, + type ConversationViewChange, + type CreateConversationViewOptions, + type TurnIndexRow, +} from './conversation-view'; +export { + appendHistoryEntry, + findHistoryIndex, + patchHistoryEntry, + replaceHistoryEntry, + respondHistoryPermission, +} from './history-writer'; +export { + ensureTurnById, + findPermissionRequestTurnIndex, + findSystemNotice, + readDiffInputsFromView, + readTurnById, + resolveActiveAssistantTurnIdFromView, + type SystemNoticeSearch, + type TurnDiffInput, +} from './turn-selectors'; diff --git a/packages/components/src/lib/conversation-view/turn-selectors.ts b/packages/components/src/lib/conversation-view/turn-selectors.ts new file mode 100644 index 000000000..308ff3bce --- /dev/null +++ b/packages/components/src/lib/conversation-view/turn-selectors.ts @@ -0,0 +1,108 @@ +import type { FileDiff, SessionHistory } from '@lody/shared'; +import type { ConversationView } from './conversation-view'; + +/** + * Index- and tail-only readers over a `ConversationView`. Each one answers a + * question the app used to ask of the full `history` array without hydrating + * a single turn it does not have to. + */ + +/** + * The open assistant turn, if the newest assistant turn is still running: + * the same rule as `resolveActiveAssistantTurnId` in `@lody/shared`, read from + * index rows (`role`, `finished`, `endedAt` are index fields) from the tail. + */ +export function resolveActiveAssistantTurnIdFromView(view: ConversationView): string | undefined { + for (let i = view.turnCount - 1; i >= 0; i -= 1) { + const row = view.index(i); + if (!row || row.role !== 'assistant') continue; + if (row.finished === true || typeof row.endedAt === 'number') return undefined; + return row.id; + } + return undefined; +} + +/** The hydrated turn with `id`, or the index it lives at when not hydrated. */ +export function readTurnById( + view: ConversationView, + turnId: string +): { index: number; turn: SessionHistory | undefined } { + const index = view.indexOf(turnId); + return { index, turn: index >= 0 ? view.turn(index) : undefined }; +} + +/** Hydrate one turn by id and return it. */ +export async function ensureTurnById( + view: ConversationView, + turnId: string +): Promise { + const index = view.indexOf(turnId); + if (index < 0) return undefined; + await view.ensureRange(index, index + 1); + return view.turn(index); +} + +/** + * Which turn carries the permission request `requestId`, searching the + * hydrated tail first. Returns -1 when no hydrated turn carries it; the writer + * then scans the doc itself, which needs no hydration. + */ +export function findPermissionRequestTurnIndex(view: ConversationView, requestId: string): number { + for (let i = view.turnCount - 1; i >= 0; i -= 1) { + if (!view.isHydrated(i)) continue; + const turn = view.turn(i); + if (!turn || !Array.isArray(turn.items)) continue; + for (const item of turn.items as unknown[]) { + const request = (item as { permissionRequest?: { requestId?: string } } | null) + ?.permissionRequest; + if (request?.requestId === requestId) return i; + } + } + return -1; +} + +export type SystemNoticeSearch = + | { found: true } + | { found: false; unhydratedSystemTurnIndex: number | null }; + +/** + * Whether any turn carries the `system_notice` named `name`. Only `system` + * turns can, so non-system turns are skipped from the index alone; a system + * turn that is not hydrated is reported so the caller can hydrate it and ask + * again, instead of hydrating the whole transcript. + */ +export function findSystemNotice(view: ConversationView, name: string): SystemNoticeSearch { + let unhydrated: number | null = null; + for (let i = view.turnCount - 1; i >= 0; i -= 1) { + const row = view.index(i); + if (!row || row.role !== 'system') continue; + const turn = view.turn(i); + if (!turn) { + unhydrated ??= i; + continue; + } + const items = Array.isArray(turn.items) ? (turn.items as unknown[]) : []; + for (const item of items) { + const notice = item as { type?: string; name?: string } | null; + if (notice?.type === 'system_notice' && notice.name === name) return { found: true }; + } + } + return { found: false, unhydratedSystemTurnIndex: unhydrated }; +} + +export type TurnDiffInput = { id: string; role: SessionHistory['role']; fileDiff: FileDiff[] }; + +/** + * One light `{ id, role, fileDiff }` per turn, built from index rows and the + * per-turn `fileDiff` cache, so the diff summary sees every turn's edits + * without hydrating any message items. + */ +export function readDiffInputsFromView(view: ConversationView): TurnDiffInput[] { + const inputs: TurnDiffInput[] = []; + for (let i = 0; i < view.turnCount; i += 1) { + const row = view.index(i); + if (!row?.id) continue; + inputs.push({ id: row.id, role: row.role, fileDiff: view.fileDiff(i) ?? [] }); + } + return inputs; +} diff --git a/packages/components/src/lib/session-store-history.ts b/packages/components/src/lib/session-store-history.ts new file mode 100644 index 000000000..df0bf1b43 --- /dev/null +++ b/packages/components/src/lib/session-store-history.ts @@ -0,0 +1,41 @@ +import { resolveActiveAssistantTurnId, type SessionHistory } from '@lody/shared'; +import type { SessionDocStore } from '@/atoms/runtime'; +import { ensureTurnById, resolveActiveAssistantTurnIdFromView } from './conversation-view'; + +/** + * History reads against a `SessionDocStore` that stay O(1) or O(one turn) + * when the store carries a `ConversationView`, and fall back to the full + * `getState().history` array on the rollback path. Hooks that used to + * `getState().history.find(...)` go through here so the flag decides the + * cost, not the call site. + */ + +export async function readSessionHistoryEntry( + store: SessionDocStore, + entryId: string +): Promise { + const view = store.conversationView; + if (view) return await ensureTurnById(view, entryId); + return (store.getState().history as SessionHistory[]).find((entry) => entry.id === entryId); +} + +/** The user turn with `userTurnId`, or `undefined` when absent or not a user turn. */ +export async function readSessionUserTurn( + store: SessionDocStore, + userTurnId: string +): Promise { + const entry = await readSessionHistoryEntry(store, userTurnId); + return entry?.role === 'user' ? entry : undefined; +} + +export function readActiveAssistantTurnId(store: SessionDocStore): string | undefined { + const view = store.conversationView; + if (view) return resolveActiveAssistantTurnIdFromView(view); + return resolveActiveAssistantTurnId(store.getState().history); +} + +export function readSessionHistoryLength(store: SessionDocStore): number { + const view = store.conversationView; + if (view) return view.turnCount; + return store.getState().history?.length ?? 0; +} diff --git a/packages/components/src/providers/AGENTS.md b/packages/components/src/providers/AGENTS.md index 31ba7d4f9..ff1eb0a77 100644 --- a/packages/components/src/providers/AGENTS.md +++ b/packages/components/src/providers/AGENTS.md @@ -30,3 +30,23 @@ again. Contract test: `packages/shared/tests/session-doc-forward-compat.test.ts` projection and disables queries, Machine Flock, sharing, and eager-sync inputs. Provider- external consumers such as `RuntimeProvider` retain their existing default behavior. Explicit `workspaceId` / `enabled` options remain fenced by the route scope and cannot reopen stale work. + +## Session store history + +- `createSessionStore` composes its state through `session-doc-state-source.ts`. + With `isConversationViewEnabled()` (on unless `VITE_LODY_CONVERSATION_VIEW=0` + or `localStorage['lody:conversationView']='0'`) the Mirror uses + `sessionControlDocSchema` and `SessionDocStore.conversationView` is the + history reader; off is the untouched full-Mirror path, kept as rollback. +- On the view path `getState().history` is a lazy bridge over + `conversationView.readAll()`, bound once per snapshot and memoized per + (control root, view version). Compare snapshots through + `readSessionDocHistoryRevision`, never by touching `history`: the first read + materializes the transcript. A history-only doc event never produces a new + control root, so control-plane identity checks keep working. +- `setState` must not reach `history` on the view path; the store throws + `SessionHistoryWriteThroughMirrorError` instead of persisting nothing. Every + history write goes through `WorkspaceWriter` (`workspace-writer-impl.ts`), + which routes to `lib/conversation-view/history-writer` when the store has a + view and to `setState` otherwise. Store-level reads that hooks need + (`lib/session-store-history.ts`) branch the same way. diff --git a/packages/components/src/providers/create-workspace-runtime.ts b/packages/components/src/providers/create-workspace-runtime.ts index 216da9e33..15f711473 100644 --- a/packages/components/src/providers/create-workspace-runtime.ts +++ b/packages/components/src/providers/create-workspace-runtime.ts @@ -35,7 +35,6 @@ import { SESSION_DOC_PREFIX, type SessionStatus, LORO_STREAMS_BUCKET_ID, - sessionDocSchema, ClientToServerSchema, ServerToClientSchema, type ClientToServer, @@ -75,6 +74,7 @@ import { } from './workspace-target-router'; import { mergePresenceSnapshots } from './presence-snapshot-merge'; import { Mirror } from 'loro-mirror'; +import { createSessionDocStateSource } from './session-doc-state-source'; import { LoroDoc, EphemeralStore } from 'loro-crdt'; import { WorkspaceRuntime, @@ -306,6 +306,21 @@ const isElectronLocalDataPlaneEnabled = (): boolean => { } }; +/** + * Session docs read history through `ConversationView` (control-plane Mirror, + * O(window) hydration) unless switched off. Off is the full-Mirror path, + * untouched, as the rollback: `VITE_LODY_CONVERSATION_VIEW=0` at build time or + * `localStorage['lody:conversationView'] = '0'` at runtime. + */ +const isConversationViewEnabled = (): boolean => { + if (import.meta.env.VITE_LODY_CONVERSATION_VIEW === '0') return false; + try { + return globalThis.localStorage?.getItem('lody:conversationView') !== '0'; + } catch { + return true; + } +}; + // Escape hatch for the Machine RPC response live transport. Unset (the normal // case) leaves the SSE-first policy in charge; setting it pins one transport and // disables the fallback/probe logic. @@ -3622,14 +3637,10 @@ export async function createWorkspaceRuntime(deps: RuntimeDeps): Promise { listener(syncLeaseCount > 0 || roomSub || syncJoinPromise ? state : 'idle'); }), - getState: () => mirror.getState(), - setState: (updater) => { - mirror.setState(updater as never); - }, - subscribe: (listener) => mirror.subscribe(listener), + conversationView: stateSource.conversationView, + getState: stateSource.getState, + setState: stateSource.setState, + subscribe: stateSource.subscribe, dispose: () => { disposed = true; stopSyncNow(); syncTracker.dispose(); - mirror.dispose(); + stateSource.dispose(); }, waitUntilSynced: async (signal?: AbortSignal) => { await transportReady.promise; diff --git a/packages/components/src/providers/session-doc-state-source.ts b/packages/components/src/providers/session-doc-state-source.ts new file mode 100644 index 000000000..c6ca702d0 --- /dev/null +++ b/packages/components/src/providers/session-doc-state-source.ts @@ -0,0 +1,232 @@ +import { Mirror } from 'loro-mirror'; +import type { LoroDoc } from 'loro-crdt'; +import { sessionDocSchema, type SessionId } from '@lody/shared'; +import type { SessionDocState, SessionDocUpdater } from '../atoms/runtime'; +import { + createConversationViewFromDoc, + createSessionControlMirror, + type ConversationView, +} from '../lib/conversation-view'; + +/** + * The part of a `SessionDocStore` that turns a Loro doc into React-readable + * state: `getState` / `setState` / `subscribe`, plus the `ConversationView` + * when history is read through it. + * + * Two compositions, selected once at store creation: + * + * - **full Mirror** (`conversationViewEnabled: false`): one `Mirror` over the + * whole `sessionDocSchema`, exactly what every client ran before. This is + * the rollback path and must stay byte-for-byte the old behavior. + * - **control Mirror + view** (`conversationViewEnabled: true`): the Mirror is + * built with `sessionControlDocSchema`, so it never materializes history, + * and `history` on the returned state is a lazy getter over + * `conversationView.readAll()`. Readers that still take `doc.history` keep + * working and pay one incremental full read on first access instead of the + * Mirror init; readers on the view pay nothing. History WRITES must go + * through `lib/conversation-view/history-writer`: an ignored field is + * memory-only in loro-mirror, so a `setState` that reaches `history` would + * silently persist nothing. The guard below turns that into an error. + */ +export type SessionDocStateSource = { + readonly conversationView: ConversationView | null; + getState: () => SessionDocState; + setState: (updater: SessionDocUpdater) => void; + subscribe: (listener: (state: SessionDocState) => void) => () => void; + dispose: () => void; +}; + +/** + * Non-enumerable marker on a bridged state: the view version its `history` + * getter is bound to. `use-session-doc.ts` compares this instead of touching + * `history`, because reading the getter is what materializes the transcript. + */ +export const SESSION_DOC_HISTORY_REVISION: unique symbol = Symbol('sessionDocHistoryRevision'); + +type BridgedSessionDocState = SessionDocState & { + readonly [SESSION_DOC_HISTORY_REVISION]?: number; +}; + +/** + * What distinguishes two session-doc snapshots as far as history goes: the + * bridged revision when present, else the `history` array identity the full + * Mirror keeps stable across unrelated updates. + */ +export function readSessionDocHistoryRevision(state: SessionDocState): unknown { + const revision = (state as BridgedSessionDocState)[SESSION_DOC_HISTORY_REVISION]; + return revision !== undefined ? revision : state.history; +} + +export class SessionHistoryWriteThroughMirrorError extends Error { + constructor() { + super( + 'Session history cannot be written through setState while ConversationView is enabled: ' + + 'the control-plane Mirror ignores `history`, so the write would only change memory. ' + + 'Use appendHistoryEntry / replaceHistoryEntry / respondHistoryPermission from ' + + '@/lib/conversation-view (WorkspaceWriter routes there).' + ); + this.name = 'SessionHistoryWriteThroughMirrorError'; + } +} + +const HISTORY_KEY = 'history'; + +/** + * Wrap a mutable draft so any touch of `history` throws before loro-mirror can + * swallow it. Reads throw too: `draft.history.push(...)` would otherwise fail + * with an unrelated "cannot read property of undefined". + */ +const guardHistoryOnDraft = (draft: T): T => + new Proxy(draft, { + get(target, property, receiver) { + if (property === HISTORY_KEY) throw new SessionHistoryWriteThroughMirrorError(); + return Reflect.get(target, property, receiver); + }, + set(target, property, value, receiver) { + if (property === HISTORY_KEY) throw new SessionHistoryWriteThroughMirrorError(); + return Reflect.set(target, property, value, receiver); + }, + deleteProperty(target, property) { + if (property === HISTORY_KEY) throw new SessionHistoryWriteThroughMirrorError(); + return Reflect.deleteProperty(target, property); + }, + }); + +const touchesHistory = (value: unknown): boolean => + typeof value === 'object' && value !== null && HISTORY_KEY in value; + +/** Same own keys with identical values: the Mirror root did not change. */ +const shallowEqualRoot = (left: object, right: object): boolean => { + if (left === right) return true; + const leftKeys = Object.keys(left); + const rightKeys = Object.keys(right); + if (leftKeys.length !== rightKeys.length) return false; + for (const key of leftKeys) { + if ( + !Object.prototype.hasOwnProperty.call(right, key) || + (left as Record)[key] !== (right as Record)[key] + ) { + return false; + } + } + return true; +}; + +export function createSessionDocStateSource(options: { + doc: LoroDoc; + sessionId: SessionId; + conversationViewEnabled: boolean; +}): SessionDocStateSource { + const { doc, sessionId } = options; + + if (!options.conversationViewEnabled) { + const mirror = new Mirror({ + doc, + schema: sessionDocSchema, + // Tolerate root keys written by peers running a newer schema version. + ignoreUnknownProperties: true, + // Plan is now stored per-turn on history entries, not at root level + initialState: { session: { id: sessionId }, history: [] }, + debug: false, + }); + return { + conversationView: null, + getState: () => mirror.getState(), + setState: (updater) => { + mirror.setState(updater as never); + }, + subscribe: (listener) => mirror.subscribe(listener), + dispose: () => { + mirror.dispose(); + }, + }; + } + + const mirror = createSessionControlMirror(doc, sessionId); + const view = createConversationViewFromDoc(doc, { sessionId }); + const listeners = new Set<(state: SessionDocState) => void>(); + + // The Mirror notifies on every doc batch, including history-only ones its + // control schema does not reflect. Only a root whose own values changed is + // a control-plane change; the rest reuse the current control snapshot. + let controlState: object = mirror.getState() as object; + let cached: { control: object; version: number; value: SessionDocState } | null = null; + + const bridge = (control: object, version: number): SessionDocState => { + const state = { ...(control as Record) }; + let history: SessionDocState['history'] | undefined; + Object.defineProperty(state, HISTORY_KEY, { + enumerable: true, + configurable: false, + get: () => { + // Bound once per snapshot: a held snapshot never changes underneath + // its reader, and repeated access within a render is free. + history ??= view.readAll() as SessionDocState['history']; + return history; + }, + }); + Object.defineProperty(state, SESSION_DOC_HISTORY_REVISION, { + enumerable: false, + configurable: false, + value: version, + }); + return state as unknown as SessionDocState; + }; + + const getState = (): SessionDocState => { + const version = view.version; + if (cached && cached.control === controlState && cached.version === version) { + return cached.value; + } + const value = bridge(controlState, version); + cached = { control: controlState, version, value }; + return value; + }; + + const notify = () => { + const state = getState(); + for (const listener of listeners) listener(state); + }; + + const unsubscribeMirror = mirror.subscribe((next) => { + if (shallowEqualRoot(controlState, next as object)) return; + controlState = next as object; + notify(); + }); + const unsubscribeView = view.subscribe(() => { + notify(); + }); + + return { + conversationView: view, + getState, + setState: (updater) => { + if (typeof updater !== 'function') { + if (touchesHistory(updater)) throw new SessionHistoryWriteThroughMirrorError(); + mirror.setState(updater as never); + return; + } + mirror.setState(((draft: object) => { + const guarded = guardHistoryOnDraft(draft); + const result = (updater as (state: object) => unknown)(guarded); + // A mutative updater that returns its own draft is still mutative. + if (result === undefined || result === guarded || result === draft) return undefined; + if (touchesHistory(result)) throw new SessionHistoryWriteThroughMirrorError(); + return result; + }) as never); + }, + subscribe: (listener) => { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + dispose: () => { + unsubscribeMirror(); + unsubscribeView(); + listeners.clear(); + view.dispose(); + mirror.dispose(); + }, + }; +} diff --git a/packages/components/src/providers/workspace-writer-impl.ts b/packages/components/src/providers/workspace-writer-impl.ts index 78863a264..5406b3079 100644 --- a/packages/components/src/providers/workspace-writer-impl.ts +++ b/packages/components/src/providers/workspace-writer-impl.ts @@ -5,10 +5,17 @@ import { type MessageQueueItem, type PreviewVisualCommentDocInput, type SessionDocMeta, + type SessionHistoryInput, } from '@lody/shared'; import type { SessionId } from '@lody/shared/ids'; import type { LoroRepo } from 'loro-repo'; import type { PreviewVisualCommentDocStore, SessionDocStore } from '../atoms/runtime'; +import { + appendHistoryEntry, + findPermissionRequestTurnIndex, + replaceHistoryEntry, + respondHistoryPermission, +} from '../lib/conversation-view'; import type { WorkspaceWriter } from './workspace-writer'; // # WorkspaceWriter implementation @@ -31,6 +38,23 @@ export type DirectWorkspaceWriterDeps = { * repo / session stores. This is exactly what the hooks did before the seam, so * there is zero behavior change in cloud mode. */ +/** + * Every history write lands here. A store carrying a `ConversationView` runs + * on the control-plane Mirror, whose schema ignores `history`, so its writes + * go straight to the doc through `lib/conversation-view/history-writer` + * (same container shapes the full-schema Mirror would produce). Without a + * view this is the untouched `setState` path. + */ +const appendHistory = (store: SessionDocStore, entry: Record): void => { + if (store.conversationView) { + appendHistoryEntry(store.doc, entry as unknown as SessionHistoryInput); + return; + } + store.setState((draft: SessionDocMeta) => { + draft.history.push(entry as SessionDocMeta['history'][number]); + }); +}; + export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): WorkspaceWriter { const withSessionStore = async ( sessionId: string, @@ -77,9 +101,7 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo meta as Parameters[1] ), withSessionStore(sessionId, (store) => { - store.setState((draft: SessionDocMeta) => { - draft.history.push(entry as SessionDocMeta['history'][number]); - }); + appendHistory(store, entry); }), ]); void dispatch; @@ -120,9 +142,7 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo async appendSessionTurn(sessionId, entry, dispatch) { await withSessionStore(sessionId, (store) => { - store.setState((draft: SessionDocMeta) => { - draft.history.push(entry as SessionDocMeta['history'][number]); - }); + appendHistory(store, entry); }); // Dispatch stays the caller's sibling side effect (Machine RPC / durable // pointer), matching the send hot path. @@ -131,14 +151,24 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo async appendSessionHistory(sessionId, entry) { await withSessionStore(sessionId, (store) => { - store.setState((draft: SessionDocMeta) => { - draft.history.push(entry as SessionDocMeta['history'][number]); - }); + appendHistory(store, entry); }); }, async updateSessionHistory(sessionId, entryId, entry) { await withSessionStore(sessionId, (store) => { + const view = store.conversationView; + if (view) { + // Not found stays a no-op, as on the Mirror path. + const index = view.indexOf(entryId); + replaceHistoryEntry( + store.doc, + entryId, + entry as unknown as SessionHistoryInput, + index >= 0 ? index : undefined + ); + return; + } store.setState((draft: SessionDocMeta) => { const history = draft.history as SessionDocMeta['history']; const idx = history.findIndex((h) => (h as { id?: string }).id === entryId); @@ -150,6 +180,14 @@ export function createDirectWorkspaceWriter(deps: DirectWorkspaceWriterDeps): Wo async respondSessionPermission(sessionId, requestId, outcome) { await withSessionStore(sessionId, (store) => { + const view = store.conversationView; + if (view) { + // The request almost always sits on the open tail turn; the writer + // scans the doc itself when the hydrated tail does not carry it. + const hint = findPermissionRequestTurnIndex(view, requestId); + respondHistoryPermission(store.doc, requestId, outcome, hint >= 0 ? hint : undefined); + return; + } store.setState((draft: SessionDocMeta) => { for (const entry of draft.history as SessionDocMeta['history']) { const items = (entry as { items?: unknown[] }).items; diff --git a/packages/components/tests/conversation-view-history-writer.test.ts b/packages/components/tests/conversation-view-history-writer.test.ts new file mode 100644 index 000000000..d24901f71 --- /dev/null +++ b/packages/components/tests/conversation-view-history-writer.test.ts @@ -0,0 +1,165 @@ +import { describe, expect, it } from 'vitest'; +import { LoroDoc } from 'loro-crdt'; +import { Mirror } from 'loro-mirror'; +import { sessionDocSchema, type SessionHistoryInput, type SessionId } from '@lody/shared'; + +import { + appendHistoryEntry, + findHistoryIndex, + replaceHistoryEntry, + respondHistoryPermission, +} from '../src/lib/conversation-view/history-writer'; + +const sessionId = 'session-1' as SessionId; + +/** Container-kind tree of the doc, with values, e.g. `{ text: "hi" }`. */ +function shapeOf(doc: LoroDoc): unknown { + const kindOf = (cid: string) => cid.slice(cid.lastIndexOf(':') + 1); + const walk = (node: unknown): unknown => { + if (node && typeof node === 'object' && 'cid' in (node as object)) { + const { value, cid } = node as { value: unknown; cid: string }; + return { $kind: kindOf(cid), value: walk(value) }; + } + if (Array.isArray(node)) return node.map(walk); + if (node && typeof node === 'object') { + return Object.fromEntries( + Object.entries(node as Record).map(([k, v]) => [k, walk(v)]) + ); + } + return node; + }; + return walk((doc.getDeepValueWithID() as { history: unknown }).history); +} + +function viaMirror(history: SessionHistoryInput[]): LoroDoc { + const doc = new LoroDoc(); + const mirror = new Mirror({ + doc, + schema: sessionDocSchema, + ignoreUnknownProperties: true, + initialState: { session: { id: sessionId }, history: [] }, + }); + mirror.setState((prev) => ({ ...prev, history: history as never })); + doc.commit(); + return doc; +} + +function viaWriter(history: SessionHistoryInput[]): LoroDoc { + const doc = new LoroDoc(); + for (const entry of history) appendHistoryEntry(doc, entry); + return doc; +} + +const userTurn = (id: string): SessionHistoryInput => ({ + id, + role: 'user', + timestamp: '2026-01-01T00:00:00.000Z', + status: 'seen', + read: true, + finished: true, + fileDiff: [], + items: [{ type: 'text', text: `hello ${id}` }] as never, + inputConfig: { + prompt: `hello ${id}`, + cliType: 'builtin', + agentType: 'claude', + inputBlocks: [{ type: 'text', text: `hello ${id}` }], + mcpServerIds: [], + configOptionValues: { reasoning_effort: 'high' }, + } as never, +}); + +const assistantTurn = (id: string): SessionHistoryInput => ({ + id, + role: 'assistant', + timestamp: '2026-01-01T00:00:01.000Z', + finished: true, + endedAt: 1_700_000_000_000, + fileDiff: [{ filePath: 'a.ts', additions: 1, deletions: 0 }], + modelInfo: { id: 'model', name: 'Model' } as never, + plan: [{ content: 'step', status: 'completed', priority: 'medium' }] as never, + items: [ + { type: 'thought', text: 'thinking' }, + { + type: 'tool_call', + toolCallId: 'tc1', + title: 'Run ls', + kind: 'execute', + status: 'completed', + locations: [{ path: 'a.ts' }], + content: [ + { type: 'terminal_command', command: 'ls', cwd: '/x', args: ['-la'] }, + { + type: 'terminal_output', + output: 'a\nb', + stream: 'combined', + truncated: false, + exitStatus: { exitCode: 0, signal: null }, + }, + ], + permissionRequest: { requestId: 'req-1', options: [{ optionId: 'allow', name: 'Allow' }] }, + }, + { type: 'text', text: 'done' }, + ] as never, +}); + +describe('HistoryWriter', () => { + it('writes the same container shapes as a full-schema Mirror', () => { + const history = [userTurn('u1'), assistantTurn('a1'), userTurn('u2')]; + expect(shapeOf(viaWriter(history))).toEqual(shapeOf(viaMirror(history))); + }); + + it('reads back through the full-schema Mirror as the same state', () => { + const history = [userTurn('u1'), assistantTurn('a1')]; + const doc = viaWriter(history); + const mirror = new Mirror({ doc, schema: sessionDocSchema, ignoreUnknownProperties: true }); + const expected = new Mirror({ + doc: viaMirror(history), + schema: sessionDocSchema, + ignoreUnknownProperties: true, + }); + expect(JSON.parse(JSON.stringify(mirror.getState().history))).toEqual( + JSON.parse(JSON.stringify(expected.getState().history)) + ); + }); + + it('replaces an entry in place and keeps the turn container id', () => { + const doc = viaWriter([userTurn('u1'), assistantTurn('a1')]); + const before = (doc.getList('history').getShallowValue() as string[])[1]; + const changed = { ...assistantTurn('a1'), finished: false, endedAt: undefined, status: 'seen' }; + expect(replaceHistoryEntry(doc, 'a1', changed as SessionHistoryInput)).toBe(true); + const after = (doc.getList('history').getShallowValue() as string[])[1]; + expect(after).toBe(before); + const state = new Mirror({ + doc, + schema: sessionDocSchema, + ignoreUnknownProperties: true, + }).getState().history as unknown as Record[]; + expect(state[1]).toMatchObject({ id: 'a1', finished: false, status: 'seen' }); + expect(state[1]).not.toHaveProperty('endedAt'); + expect(findHistoryIndex(doc, 'a1')).toBe(1); + expect(findHistoryIndex(doc, 'missing')).toBe(-1); + }); + + it('records a permission outcome the same way the Mirror path did', () => { + const history = [assistantTurn('a1')]; + const outcome = { outcome: 'selected', optionId: 'allow' }; + const docW = viaWriter(history); + expect(respondHistoryPermission(docW, 'req-1', outcome)).toBe(true); + expect(respondHistoryPermission(docW, 'req-missing', outcome)).toBe(false); + + const docM = viaMirror(history); + const mirror = new Mirror({ + doc: docM, + schema: sessionDocSchema, + ignoreUnknownProperties: true, + }); + mirror.setState((draft) => { + const items = (draft.history[0] as { items: { permissionRequest?: { outcome?: unknown } }[] }) + .items; + items[1]!.permissionRequest!.outcome = outcome; + }); + docM.commit(); + expect(shapeOf(docW)).toEqual(shapeOf(docM)); + }); +}); diff --git a/packages/components/tests/conversation-view-turn-selectors.test.ts b/packages/components/tests/conversation-view-turn-selectors.test.ts new file mode 100644 index 000000000..5c7b6b714 --- /dev/null +++ b/packages/components/tests/conversation-view-turn-selectors.test.ts @@ -0,0 +1,177 @@ +import { describe, expect, it } from 'vitest'; +import { LoroDoc } from 'loro-crdt'; +import { + resolveActiveAssistantTurnId, + type SessionHistoryInput, + type SessionId, +} from '@lody/shared'; +import { + appendHistoryEntry, + createConversationViewFromDoc, + ensureTurnById, + findPermissionRequestTurnIndex, + findSystemNotice, + patchHistoryEntry, + readDiffInputsFromView, + resolveActiveAssistantTurnIdFromView, +} from '../src/lib/conversation-view'; + +const sessionId = 'session-selectors' as SessionId; + +const user = (id: string): SessionHistoryInput => ({ + id, + role: 'user', + timestamp: '2026-01-01T00:00:00.000Z', + status: 'seen', + read: true, + finished: true, + fileDiff: [], + items: [{ type: 'text', text: `prompt ${id}` }] as never, + inputConfig: { prompt: `prompt ${id}`, cliType: 'builtin', agentType: 'claude' } as never, +}); + +const assistant = ( + id: string, + overrides: Partial = {} +): SessionHistoryInput => ({ + id, + role: 'assistant', + timestamp: '2026-01-01T00:01:00.000Z', + finished: true, + endedAt: 1_700_000_000_000, + fileDiff: [{ filePath: `src/${id}.ts`, add: 3, del: 1 }] as never, + items: [{ type: 'text', text: `answer ${id}` }] as never, + ...overrides, +}); + +const system = (id: string, name: string): SessionHistoryInput => ({ + id, + role: 'system', + timestamp: '2026-01-01T00:02:00.000Z', + finished: true, + read: true, + fileDiff: [], + items: [{ type: 'system_notice', name, meta: {} }] as never, +}); + +const docOf = (entries: SessionHistoryInput[]): LoroDoc => { + const doc = new LoroDoc(); + for (const entry of entries) appendHistoryEntry(doc, entry); + return doc; +}; + +describe('resolveActiveAssistantTurnIdFromView', () => { + it('matches the shared array rule over index rows, including the open-turn case', () => { + const shapes: SessionHistoryInput[][] = [ + [], + [user('u1')], + [user('u1'), assistant('a1')], + [user('u1'), assistant('a1', { finished: false, endedAt: undefined })], + [user('u1'), assistant('a1', { finished: undefined, endedAt: undefined }), user('u2')], + [user('u1'), assistant('a1', { finished: false, endedAt: 5 })], + ]; + for (const shape of shapes) { + const doc = docOf(shape); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 1 }); + expect(resolveActiveAssistantTurnIdFromView(view)).toBe( + resolveActiveAssistantTurnId(view.readAll()) + ); + view.dispose(); + } + }); + + it('tracks a status flip written through patchHistoryEntry without hydration', () => { + const doc = docOf([user('u1'), assistant('a1', { finished: false, endedAt: undefined })]); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 0 }); + expect(resolveActiveAssistantTurnIdFromView(view)).toBe('a1'); + expect(patchHistoryEntry(doc, 'a1', { finished: true, endedAt: 42 }, 1)).toBe(true); + expect(resolveActiveAssistantTurnIdFromView(view)).toBeUndefined(); + expect(view.isHydrated(1)).toBe(false); + // The patch left the items container alone. + expect(view.readAll()[1]).toMatchObject({ id: 'a1', finished: true, endedAt: 42 }); + expect(view.readAll()[1]?.items).toHaveLength(1); + view.dispose(); + }); +}); + +describe('ensureTurnById', () => { + it('hydrates exactly the requested turn', async () => { + const doc = docOf([user('u1'), assistant('a1'), user('u2'), assistant('a2')]); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 1 }); + expect(view.isHydrated(1)).toBe(false); + expect(await ensureTurnById(view, 'a1')).toMatchObject({ id: 'a1' }); + expect(view.isHydrated(1)).toBe(true); + expect(view.isHydrated(0)).toBe(false); + expect(await ensureTurnById(view, 'missing')).toBeUndefined(); + view.dispose(); + }); +}); + +describe('findPermissionRequestTurnIndex', () => { + it('finds the request on a hydrated tail turn and reports -1 otherwise', () => { + const permission = { + type: 'tool_call', + toolCallId: 'tc1', + title: 'Run', + kind: 'execute', + status: 'pending', + permissionRequest: { requestId: 'req-1', options: [] }, + }; + const doc = docOf([ + user('u1'), + assistant('a1', { items: [permission] as never, finished: false, endedAt: undefined }), + user('u2'), + assistant('a2', { items: [permission] as never }), + ]); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 1 }); + expect(findPermissionRequestTurnIndex(view, 'req-1')).toBe(3); + expect(findPermissionRequestTurnIndex(view, 'req-none')).toBe(-1); + view.dispose(); + }); +}); + +describe('findSystemNotice', () => { + it('skips non-system turns and reports an unhydrated system turn to hydrate', async () => { + const doc = docOf([ + user('u1'), + assistant('a1'), + system('s1', 'session_fork_origin'), + user('u2'), + assistant('a2'), + ]); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 1 }); + expect(findSystemNotice(view, 'session_fork_origin')).toEqual({ + found: false, + unhydratedSystemTurnIndex: 2, + }); + await view.ensureRange(2, 3); + expect(findSystemNotice(view, 'session_fork_origin')).toEqual({ found: true }); + expect(findSystemNotice(view, 'other')).toEqual({ + found: false, + unhydratedSystemTurnIndex: null, + }); + expect(view.isHydrated(0)).toBe(false); + view.dispose(); + }); +}); + +describe('readDiffInputsFromView', () => { + it('reads every turn fileDiff from its own container and follows updates', () => { + const doc = docOf([user('u1'), assistant('a1'), user('u2'), assistant('a2')]); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 1 }); + expect(readDiffInputsFromView(view)).toEqual([ + { id: 'u1', role: 'user', fileDiff: [] }, + { id: 'a1', role: 'assistant', fileDiff: [{ filePath: 'src/a1.ts', add: 3, del: 1 }] }, + { id: 'u2', role: 'user', fileDiff: [] }, + { id: 'a2', role: 'assistant', fileDiff: [{ filePath: 'src/a2.ts', add: 3, del: 1 }] }, + ]); + expect(view.isHydrated(1)).toBe(false); + const first = view.fileDiff(1); + expect(view.fileDiff(1)).toBe(first); + + patchHistoryEntry(doc, 'a1', { fileDiff: [{ filePath: 'src/x.ts', add: 1, del: 0 }] as never }); + expect(view.fileDiff(1)).toEqual([{ filePath: 'src/x.ts', add: 1, del: 0 }]); + expect(view.isHydrated(1)).toBe(false); + view.dispose(); + }); +}); diff --git a/packages/components/tests/conversation-view.test.ts b/packages/components/tests/conversation-view.test.ts new file mode 100644 index 000000000..3a506527e --- /dev/null +++ b/packages/components/tests/conversation-view.test.ts @@ -0,0 +1,202 @@ +import { describe, expect, it } from 'vitest'; +import { LoroDoc } from 'loro-crdt'; +import { Mirror } from 'loro-mirror'; +import { sessionDocSchema, type SessionHistoryInput, type SessionId } from '@lody/shared'; + +import { sessionControlDocSchema } from '../src/lib/conversation-view/control-doc-schema'; +import { createConversationViewFromDoc } from '../src/lib/conversation-view/conversation-view'; +import { + appendHistoryEntry, + replaceHistoryEntry, +} from '../src/lib/conversation-view/history-writer'; + +const sessionId = 'session-1' as SessionId; + +const turn = (index: number): SessionHistoryInput => + index % 2 === 0 + ? { + id: `u${index}`, + role: 'user', + timestamp: `2026-01-01T00:00:${String(index % 60).padStart(2, '0')}.000Z`, + status: 'seen', + read: true, + finished: true, + fileDiff: [], + items: [{ type: 'text', text: `prompt ${index}` }] as never, + inputConfig: { + prompt: `prompt ${index}`, + cliType: 'builtin', + agentType: 'claude', + } as never, + } + : { + id: `a${index}`, + role: 'assistant', + timestamp: `2026-01-01T00:01:${String(index % 60).padStart(2, '0')}.000Z`, + finished: true, + endedAt: 1_700_000_000_000 + index, + fileDiff: [], + items: [ + { type: 'thought', text: `thinking ${index}` }, + { + type: 'tool_call', + toolCallId: `tc${index}`, + title: 'Run ls', + kind: 'execute', + status: 'completed', + content: [{ type: 'terminal_command', command: 'ls', cwd: '/x' }], + }, + { type: 'text', text: `answer ${index}` }, + ] as never, + }; + +function docWithTurns(count: number): LoroDoc { + const doc = new LoroDoc(); + for (let index = 0; index < count; index += 1) appendHistoryEntry(doc, turn(index)); + return doc; +} + +const fullHistory = (doc: LoroDoc): unknown[] => + JSON.parse( + JSON.stringify( + new Mirror({ doc, schema: sessionDocSchema, ignoreUnknownProperties: true }).getState() + .history + ) + ); + +describe('sessionControlDocSchema', () => { + it('never materializes history and keeps control-plane writes working', () => { + const doc = docWithTurns(2_000); + const start = performance.now(); + const mirror = new Mirror({ + doc, + schema: sessionControlDocSchema, + ignoreUnknownProperties: true, + initialState: { session: { id: sessionId } }, + }); + const elapsed = performance.now() - start; + const state = mirror.getState() as Record; + expect(state.history).toBeUndefined(); + // 2,000 turns / ~14k containers: the full-schema Mirror needs hundreds of + // ms here; an ignored root must cost nothing that scales with history. + expect(elapsed).toBeLessThan(150); + + mirror.setState((prev) => ({ + ...prev, + mq: [{ $cid: 'q1', prompt: 'queued', inputConfig: {}, timestamp: 't' } as never], + })); + doc.commit(); + expect(doc.getList('history').length).toBe(2_000); + expect((doc.toJSON() as { mq: unknown[] }).mq).toHaveLength(1); + }); +}); + +describe('ConversationView', () => { + it('indexes every turn eagerly, hydrates the tail, and matches the Mirror state', () => { + const doc = docWithTurns(60); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 5, maxHydrated: 10 }); + expect(view.turnCount).toBe(60); + expect(view.index(0)).toMatchObject({ id: 'u0', role: 'user', itemCount: 1 }); + expect(view.index(1)).toMatchObject({ id: 'a1', role: 'assistant', itemCount: 3 }); + expect(view.indexOf('a59')).toBe(59); + expect(view.isHydrated(59)).toBe(true); + expect(view.isHydrated(0)).toBe(false); + expect(view.turn(0)).toBeUndefined(); + + const expected = fullHistory(doc); + expect(JSON.parse(JSON.stringify(view.turn(59)))).toEqual(expected[59]); + expect(JSON.parse(JSON.stringify(view.readAll()))).toEqual(expected); + view.dispose(); + }); + + it('hydrates ranges on demand and evicts beyond maxHydrated, never the tail', async () => { + const doc = docWithTurns(60); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 5, maxHydrated: 10 }); + await view.ensureRange(0, 8); + expect(view.isHydrated(0)).toBe(true); + expect(view.isHydrated(7)).toBe(true); + await view.ensureRange(20, 40); + // 5 tail + 20 range > 10: the oldest untouched turns are evicted first. + expect(view.isHydrated(0)).toBe(false); + expect(view.isHydrated(59)).toBe(true); + view.dispose(); + }); + + it('follows appends, in-place updates and deletes through doc events', () => { + const doc = docWithTurns(10); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 3 }); + const changes: string[] = []; + view.subscribe((change) => changes.push(change.kind)); + const versionBefore = view.version; + + appendHistoryEntry(doc, turn(10)); + expect(view.turnCount).toBe(11); + expect(view.index(10)?.id).toBe('u10'); + expect(view.isHydrated(10)).toBe(false); + expect(view.version).toBeGreaterThan(versionBefore); + + replaceHistoryEntry(doc, 'a9', { ...turn(9), finished: false, endedAt: undefined }); + expect(view.index(9)).toMatchObject({ id: 'a9', finished: false }); + expect(view.turn(9)).toMatchObject({ id: 'a9', finished: false }); + + doc.getList('history').delete(0, 1); + doc.commit(); + expect(view.turnCount).toBe(10); + expect(view.indexOf('u0')).toBe(-1); + expect(view.indexOf('a9')).toBe(8); + expect(JSON.parse(JSON.stringify(view.readAll()))).toEqual(fullHistory(doc)); + expect(changes).toEqual(expect.arrayContaining(['index', 'tail'])); + view.dispose(); + }); + + it('exposes plan and item counts on index rows without hydrating', () => { + const doc = docWithTurns(4); + replaceHistoryEntry(doc, 'a3', { + ...turn(3), + plan: [{ content: 'step 1', status: 'pending', priority: 'medium' }] as never, + }); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 0 }); + expect(view.index(3)).toMatchObject({ id: 'a3', itemCount: 3, planCount: 1 }); + expect(view.index(1)?.planCount).toBeUndefined(); + expect(view.isHydrated(3)).toBe(false); + view.dispose(); + }); + + it('retain() protects a range from eviction until released', async () => { + const doc = docWithTurns(60); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 5, maxHydrated: 10 }); + const release = view.retain(0, 8); + await view.ensureRange(0, 8); + await view.ensureRange(20, 40); + expect(view.isHydrated(0)).toBe(true); + expect(view.isHydrated(7)).toBe(true); + release(); + await view.ensureRange(40, 50); + expect(view.isHydrated(0)).toBe(false); + view.dispose(); + }); + + it('readAll() hands back the same objects for unchanged turns across changes', () => { + const doc = docWithTurns(30); + const view = createConversationViewFromDoc(doc, { sessionId, tailKeep: 3, maxHydrated: 5 }); + const first = view.readAll(); + expect(view.readAll()).toBe(first); + + appendHistoryEntry(doc, turn(30)); + const second = view.readAll(); + expect(second).not.toBe(first); + expect(second).toHaveLength(31); + for (let i = 0; i < 30; i += 1) expect(second[i]).toBe(first[i]); + + replaceHistoryEntry(doc, 'a11', { ...turn(11), finished: false, endedAt: undefined }); + const third = view.readAll(); + expect(third[11]).not.toBe(second[11]); + expect(third[11]).toMatchObject({ id: 'a11', finished: false }); + expect(third[10]).toBe(second[10]); + expect(third[12]).toBe(second[12]); + expect(JSON.parse(JSON.stringify(third))).toEqual(fullHistory(doc)); + // A full read does not pin the LRU: the window stays bounded. + expect(view.isHydrated(0)).toBe(false); + view.dispose(); + }); +}); diff --git a/packages/components/tests/session-doc-state-source.test.ts b/packages/components/tests/session-doc-state-source.test.ts new file mode 100644 index 000000000..82fad7222 --- /dev/null +++ b/packages/components/tests/session-doc-state-source.test.ts @@ -0,0 +1,224 @@ +import { describe, expect, it } from 'vitest'; +import { LoroDoc, type LoroList, type LoroMap, type LoroText } from 'loro-crdt'; +import { Mirror } from 'loro-mirror'; +import { + sessionDocSchema, + type SessionDocMeta, + type SessionHistoryInput, + type SessionId, +} from '@lody/shared'; +import type { SessionDocState } from '../src/atoms/runtime'; +import { + createSessionDocStateSource, + readSessionDocHistoryRevision, + SessionHistoryWriteThroughMirrorError, +} from '../src/providers/session-doc-state-source'; +import { + appendHistoryEntry, + createSessionControlMirror, + replaceHistoryEntry, +} from '../src/lib/conversation-view'; + +const sessionId = 'session-bridge' as SessionId; + +const turn = (index: number): SessionHistoryInput => + index % 2 === 0 + ? { + id: `u${index}`, + role: 'user', + timestamp: `2026-01-01T00:00:${String(index % 60).padStart(2, '0')}.000Z`, + status: 'seen', + read: true, + finished: true, + fileDiff: [], + items: [{ type: 'text', text: `prompt ${index}` }] as never, + inputConfig: { + prompt: `prompt ${index}`, + cliType: 'builtin', + agentType: 'claude', + } as never, + } + : { + id: `a${index}`, + role: 'assistant', + timestamp: `2026-01-01T00:01:${String(index % 60).padStart(2, '0')}.000Z`, + finished: true, + endedAt: 1_700_000_000_000 + index, + fileDiff: [], + items: [{ type: 'text', text: `answer ${index}` }] as never, + }; + +const docWithTurns = (count: number): LoroDoc => { + const doc = new LoroDoc(); + for (let index = 0; index < count; index += 1) appendHistoryEntry(doc, turn(index)); + return doc; +}; + +const plain = (value: unknown): unknown => JSON.parse(JSON.stringify(value)); + +const mirrorHistory = (doc: LoroDoc): unknown => + plain( + new Mirror({ doc, schema: sessionDocSchema, ignoreUnknownProperties: true }).getState().history + ); + +describe('createSessionDocStateSource (ConversationView enabled)', () => { + it('exposes the view and bridges history lazily with the full-Mirror value', () => { + const doc = docWithTurns(12); + const source = createSessionDocStateSource({ doc, sessionId, conversationViewEnabled: true }); + expect(source.conversationView?.turnCount).toBe(12); + + const state = source.getState(); + expect(Object.keys(state)).toEqual(expect.arrayContaining(['session', 'mq', 'history'])); + expect(plain(state.history)).toEqual(mirrorHistory(doc)); + // Same snapshot while nothing changed, and the same history array on it. + expect(source.getState()).toBe(state); + expect(source.getState().history).toBe(state.history); + source.dispose(); + }); + + it('notifies on history appends and keeps unchanged entries identity-stable', () => { + const doc = docWithTurns(6); + const source = createSessionDocStateSource({ doc, sessionId, conversationViewEnabled: true }); + const before = source.getState(); + const beforeHistory = before.history; + const seen: SessionDocState[] = []; + source.subscribe((state) => seen.push(state)); + + appendHistoryEntry(doc, turn(6)); + + expect(seen).toHaveLength(1); + const after = seen[0]!; + expect(after).not.toBe(before); + expect(after.session).toBe(before.session); + expect(readSessionDocHistoryRevision(after)).not.toBe(readSessionDocHistoryRevision(before)); + expect(after.history).toHaveLength(7); + expect(after.history[0]).toBe(beforeHistory[0]); + expect(after.history[5]).toBe(beforeHistory[5]); + expect(plain(after.history)).toEqual(mirrorHistory(doc)); + source.dispose(); + }); + + it('re-materializes only the changed turn on an in-place update', () => { + const doc = docWithTurns(6); + const source = createSessionDocStateSource({ doc, sessionId, conversationViewEnabled: true }); + const beforeHistory = source.getState().history; + + replaceHistoryEntry(doc, 'a3', { ...turn(3), finished: false, endedAt: undefined }); + + const afterHistory = source.getState().history; + expect(afterHistory[3]).not.toBe(beforeHistory[3]); + expect(afterHistory[3]).toMatchObject({ id: 'a3', finished: false }); + expect(afterHistory[2]).toBe(beforeHistory[2]); + expect(afterHistory[4]).toBe(beforeHistory[4]); + expect(plain(afterHistory)).toEqual(mirrorHistory(doc)); + source.dispose(); + }); + + it('keeps control-plane writes on the Mirror and reports them synchronously', () => { + const doc = docWithTurns(4); + const source = createSessionDocStateSource({ doc, sessionId, conversationViewEnabled: true }); + const seen: SessionDocState[] = []; + source.subscribe((state) => seen.push(state)); + + source.setState((draft: SessionDocMeta) => { + draft.mq = [{ $cid: 'q1', prompt: 'queued', inputConfig: {}, timestamp: 't' } as never]; + }); + + expect(seen).toHaveLength(1); + expect(seen[0]!.mq).toHaveLength(1); + expect((doc.toJSON() as { mq: unknown[] }).mq).toHaveLength(1); + // The history revision did not move: the same lazy array is handed out. + expect(readSessionDocHistoryRevision(seen[0]!)).toBe( + readSessionDocHistoryRevision(source.getState()) + ); + expect(doc.getList('history').length).toBe(4); + source.dispose(); + }); + + it('rejects every setState shape that reaches history', () => { + const doc = docWithTurns(2); + const source = createSessionDocStateSource({ doc, sessionId, conversationViewEnabled: true }); + + expect(() => source.setState({ history: [] } as never)).toThrow( + SessionHistoryWriteThroughMirrorError + ); + expect(() => + source.setState((draft: SessionDocMeta) => { + draft.history.push(turn(2) as never); + }) + ).toThrow(SessionHistoryWriteThroughMirrorError); + expect(() => + source.setState((draft: SessionDocMeta) => { + draft.history = []; + }) + ).toThrow(SessionHistoryWriteThroughMirrorError); + expect(() => + source.setState((prev) => ({ ...(prev as object), history: [] }) as never) + ).toThrow(SessionHistoryWriteThroughMirrorError); + + // Nothing leaked into the doc or the view. + expect(doc.getList('history').length).toBe(2); + expect(source.conversationView?.turnCount).toBe(2); + // And the Mirror still accepts a control write afterwards. + source.setState((draft: SessionDocMeta) => { + draft.session.title = 'still writable'; + }); + expect(source.getState().session.title).toBe('still writable'); + source.dispose(); + }); +}); + +describe('createSessionControlMirror', () => { + it('keeps history out of the control Mirror across appends, replaces and text deltas', () => { + const doc = docWithTurns(4); + const mirror = createSessionControlMirror(doc, sessionId); + const internals = mirror as unknown as { containerRegistry: Map }; + const registeredBefore = internals.containerRegistry.size; + let notifications = 0; + mirror.subscribe(() => { + notifications += 1; + }); + + appendHistoryEntry(doc, turn(4)); + replaceHistoryEntry(doc, 'a1', { ...turn(1), finished: false, endedAt: undefined }); + const tail = doc.getContainerById( + (doc.getList('history').getShallowValue() as string[])[4] as never + ) as LoroMap; + const items = tail.get('items') as LoroList; + const text = (items.get(0) as LoroMap).get('text') as LoroText; + text.insert(text.length, ' streamed'); + doc.commit(); + + const state = mirror.getState() as Record; + expect('history' in state).toBe(false); + expect(internals.containerRegistry.size).toBe(registeredBefore); + expect(notifications).toBeGreaterThan(0); + + mirror.setState((draft: SessionDocMeta) => { + draft.session.title = 'control still writes'; + }); + expect((doc.toJSON() as { session: { title: string } }).session.title).toBe( + 'control still writes' + ); + expect(doc.getList('history').length).toBe(5); + mirror.dispose(); + }); +}); + +describe('createSessionDocStateSource (ConversationView disabled)', () => { + it('is the full Mirror: no view, eager history, setState writes history', () => { + const doc = docWithTurns(3); + const source = createSessionDocStateSource({ doc, sessionId, conversationViewEnabled: false }); + expect(source.conversationView).toBeNull(); + const state = source.getState(); + expect(readSessionDocHistoryRevision(state)).toBe(state.history); + expect(plain(state.history)).toEqual(mirrorHistory(doc)); + + source.setState((draft: SessionDocMeta) => { + draft.history.push(turn(3) as never); + }); + expect(doc.getList('history').length).toBe(4); + expect(source.getState().history).toHaveLength(4); + source.dispose(); + }); +}); diff --git a/packages/components/tests/workspace-writer.test.ts b/packages/components/tests/workspace-writer.test.ts index 40f34042b..e93030e98 100644 --- a/packages/components/tests/workspace-writer.test.ts +++ b/packages/components/tests/workspace-writer.test.ts @@ -6,7 +6,11 @@ import { type MinimalVisualAnnotationAnchor, type PreviewVisualCommentDocInput, } from '@lody/shared'; +import { LoroDoc } from 'loro-crdt'; +import { Mirror } from 'loro-mirror'; +import { sessionDocSchema, type SessionHistoryInput, type SessionId } from '@lody/shared'; import { createDirectWorkspaceWriter } from '../src/providers/workspace-writer-impl'; +import { createSessionDocStateSource } from '../src/providers/session-doc-state-source'; const anchor: MinimalVisualAnnotationAnchor = { version: 1, @@ -110,3 +114,132 @@ describe('createDirectWorkspaceWriter', () => { ); }); }); + +describe('createDirectWorkspaceWriter history writes with a ConversationView', () => { + const sessionId = 'session-writer-view' as SessionId; + const userTurn = (id: string): SessionHistoryInput => ({ + id, + role: 'user', + timestamp: '2026-01-01T00:00:00.000Z', + status: 'pending', + read: false, + finished: true, + fileDiff: [], + items: [{ type: 'text', text: `prompt ${id}` }] as never, + inputConfig: { prompt: `prompt ${id}`, cliType: 'builtin', agentType: 'claude' } as never, + }); + const assistantTurn = (id: string, items: unknown[]): SessionHistoryInput => ({ + id, + role: 'assistant', + timestamp: '2026-01-01T00:01:00.000Z', + finished: false, + fileDiff: [], + items: items as never, + }); + const permissionItem = { + type: 'tool_call', + toolCallId: 'tc1', + title: 'Run', + kind: 'execute', + status: 'pending', + permissionRequest: { requestId: 'req-1', options: [] }, + }; + + const viaMirror = (apply: (mirror: Mirror) => void): unknown => { + const doc = new LoroDoc(); + const mirror = new Mirror({ + doc, + schema: sessionDocSchema, + ignoreUnknownProperties: true, + initialState: { session: { id: sessionId }, history: [] }, + }); + apply(mirror); + doc.commit(); + return JSON.parse(JSON.stringify(mirror.getState().history)); + }; + + const storeWithView = () => { + const doc = new LoroDoc(); + const source = createSessionDocStateSource({ doc, sessionId, conversationViewEnabled: true }); + const setState = vi.fn(source.setState); + const store = { + sessionId, + roomId: `session:${sessionId}`, + doc, + firstSynced: Promise.resolve(), + acquireSync: () => () => {}, + getSyncState: () => 'synced' as const, + subscribeSyncState: () => () => {}, + getState: source.getState, + setState, + subscribe: source.subscribe, + conversationView: source.conversationView, + dispose: source.dispose, + waitUntilSynced: async () => {}, + }; + const writer = createDirectWorkspaceWriter({ + repo: { upsertDocMeta: vi.fn(async () => {}) } as never, + acquireSessionStore: vi.fn(async () => store), + releaseSessionStoreRef: vi.fn(), + acquirePreviewVisualCommentStore: vi.fn(async () => { + throw new Error('not used'); + }), + releasePreviewVisualCommentStoreRef: vi.fn(), + }); + return { + writer, + store, + setState, + history: () => JSON.parse(JSON.stringify(source.getState().history)) as unknown, + }; + }; + + it('appends, replaces and answers permissions through the history writer, never setState', async () => { + const { writer, store, setState, history } = storeWithView(); + + await writer.startSession( + sessionId, + { title: 'x' } as never, + userTurn('u1') as never, + {} as never + ); + await writer.appendSessionTurn( + sessionId, + assistantTurn('a1', [permissionItem]) as never, + {} as never + ); + await writer.appendSessionHistory(sessionId, userTurn('u2') as never); + await writer.updateSessionHistory(sessionId, 'u2', { + ...userTurn('u2'), + status: 'seen', + } as never); + await writer.updateSessionHistory(sessionId, 'missing', userTurn('nope') as never); + await writer.respondSessionPermission(sessionId, 'req-1', { type: 'selected', optionId: 'allow' }); + + expect(setState).not.toHaveBeenCalled(); + expect(store.conversationView?.turnCount).toBe(3); + expect(history()).toEqual( + viaMirror((mirror) => { + mirror.setState((prev) => ({ ...prev, history: [userTurn('u1')] as never })); + mirror.setState((prev) => ({ + ...prev, + history: [...prev.history, assistantTurn('a1', [permissionItem])] as never, + })); + mirror.setState((prev) => ({ ...prev, history: [...prev.history, userTurn('u2')] as never })); + mirror.setState((draft) => { + const entry = draft.history.find((item) => item.id === 'u2'); + if (entry) entry.status = 'seen'; + }); + mirror.setState((draft) => { + const entry = draft.history.find((item) => item.id === 'a1'); + const item = entry?.items?.[0] as + | { permissionRequest?: { outcome?: unknown } } + | undefined; + if (item?.permissionRequest) { + item.permissionRequest.outcome = { type: 'selected', optionId: 'allow' }; + } + }); + }) + ); + }); +});