diff --git a/packages/ui/src/components/layout/Header.tsx b/packages/ui/src/components/layout/Header.tsx index 53a0416e..0f5ad4dc 100644 --- a/packages/ui/src/components/layout/Header.tsx +++ b/packages/ui/src/components/layout/Header.tsx @@ -21,7 +21,7 @@ import { DiffIcon } from '@/components/icons/DiffIcon'; import { useUIStore, type MainTab } from '@/stores/useUIStore'; import { useConfigStore } from '@/stores/useConfigStore'; import { useSessionUIStore } from '@/sync/session-ui-store'; -import { useSession, useSessionMessagesResolved } from '@/sync/sync-context'; +import { useAllLiveSessions, useSession, useSessionMessagesResolved } from '@/sync/sync-context'; import { getAllSyncSessions } from '@/sync/sync-refs'; import { useProjectsStore } from '@/stores/useProjectsStore'; import { useQuotaAutoRefresh, useQuotaStore } from '@/stores/useQuotaStore'; @@ -640,6 +640,7 @@ export const Header: React.FC = ({ const currentSessionMessagesResolved = useSessionMessagesResolved(currentSessionId ?? ''); const currentSyncedSession = useSession(currentSessionId ?? null); const globalActiveSessions = useGlobalSessionsStore((state) => state.activeSessions); + const liveSessions = useAllLiveSessions(); const activeProject = useProjectsStore((state) => { if (!state.activeProjectId) { return null; @@ -940,13 +941,12 @@ export const Header: React.FC = ({ const currentSessionLive = React.useMemo(() => { if (!currentSessionId) return null; - // Resolve from the global sessions snapshot first (same source as sidebar). - // Child-store lists are intentionally partial/truncated during bootstrap. - return globalActiveSessions.find((s) => s.id === currentSessionId) + return liveSessions.find((s) => s.id === currentSessionId) + ?? globalActiveSessions.find((s) => s.id === currentSessionId) ?? currentSyncedSession ?? getAllSyncSessions().find((s) => s.id === currentSessionId) ?? null; - }, [currentSessionId, currentSyncedSession, globalActiveSessions]); + }, [currentSessionId, currentSyncedSession, globalActiveSessions, liveSessions]); const lastResolvedSessionRef = React.useRef<{ sessionId: string; diff --git a/packages/ui/src/components/session/SessionSidebar.tsx b/packages/ui/src/components/session/SessionSidebar.tsx index cf3d13ad..604a7ac9 100644 --- a/packages/ui/src/components/session/SessionSidebar.tsx +++ b/packages/ui/src/components/session/SessionSidebar.tsx @@ -8,7 +8,7 @@ import { MobileOverlayPanel } from '@/components/ui/MobileOverlayPanel'; import { sessionEvents } from '@/lib/sessionEvents'; import { formatDirectoryName, cn } from '@/lib/utils'; import { useSessionUIStore } from '@/sync/session-ui-store'; -import { useSidebarSessions, useAllSessionStatuses } from '@/sync/sync-context'; +import { useAllLiveSessions, useAllSessionStatuses } from '@/sync/sync-context'; import { useDirectoryStore } from '@/stores/useDirectoryStore'; import { useSync } from '@/sync/use-sync'; import { useSessionPrefetch } from './sidebar/hooks/useSessionPrefetch'; @@ -57,19 +57,14 @@ import { } from './sidebar/ConfirmDialogs'; import { type SessionGroup, type SessionNode } from './sidebar/types'; import { - type ActiveNowEntry, - addActiveNowSession, - deriveActiveNowSessions, - persistActiveNowEntries, - pruneActiveNowEntries, - readActiveNowEntries, + deriveLiveActiveNowSessions, } from './sidebar/activitySections'; import { compareSessionsByPinnedAndTime, formatProjectLabel, normalizePath, } from './sidebar/utils'; -import { refreshGlobalSessions, resolveGlobalSessionDirectory, useGlobalSessionsStore } from '@/stores/useGlobalSessionsStore'; +import { refreshGlobalSessions, useGlobalSessionsStore } from '@/stores/useGlobalSessionsStore'; import { useRuntimeAPIs } from '@/hooks/useRuntimeAPIs'; import { useGitHubAuthStore } from '@/stores/useGitHubAuthStore'; import { subscribeOpenchamberEvents } from '@/lib/openchamberEvents'; @@ -115,44 +110,6 @@ interface SessionSidebarProps { showOnlyMainWorkspace?: boolean; } -type SessionStatusActivityBridgeProps = { - safeStorage: Storage; - setActiveNowEntries: React.Dispatch>; -}; - -const SessionStatusActivityBridge: React.FC = ({ - safeStorage, - setActiveNowEntries, -}) => { - const globalSessionStatuses = useAllSessionStatuses(); - const sessionStatus = React.useMemo( - () => new Map(Object.entries(globalSessionStatuses)), - [globalSessionStatuses], - ); - - React.useEffect(() => { - const nextStreamingIds = new Set(); - sessionStatus.forEach((status, sessionId) => { - if (status?.type === 'busy' || status?.type === 'retry') { - nextStreamingIds.add(sessionId); - } - }); - - if (nextStreamingIds.size > 0) { - setActiveNowEntries((prev) => { - const next = Array.from(nextStreamingIds).reduce((entries, sessionId) => addActiveNowSession(entries, sessionId), prev); - if (next === prev) { - return prev; - } - persistActiveNowEntries(safeStorage, next); - return next; - }); - } - }, [sessionStatus, safeStorage, setActiveNowEntries]); - - return null; -}; - export const SessionSidebar: React.FC = ({ mobileVariant = false, onSessionSelected, @@ -172,7 +129,6 @@ export const SessionSidebar: React.FC = ({ () => new Map(), ); const safeStorage = React.useMemo(() => getSafeStorage(), []); - const [activeNowEntries, setActiveNowEntries] = React.useState(() => readActiveNowEntries(safeStorage)); const [collapsedProjects, setCollapsedProjects] = React.useState>(new Set()); const [projectRepoStatus, setProjectRepoStatus] = React.useState>(new Map()); @@ -307,10 +263,10 @@ export const SessionSidebar: React.FC = ({ const gitBranches = useGitAllBranches(); const sync = useSync(); - const syncSessions = useSidebarSessions(); + const liveSessions = useAllLiveSessions(); + const liveSessionStatuses = useAllSessionStatuses(); const globalActiveSessions = useGlobalSessionsStore((state) => state.activeSessions); const archivedSessions = useGlobalSessionsStore((state) => state.archivedSessions); - const hasLoadedGlobalSessions = useGlobalSessionsStore((state) => state.hasLoaded); const currentSessionId = useSessionUIStore((state) => state.currentSessionId); const newSessionDraftOpen = useSessionUIStore((state) => Boolean(state.newSessionDraft?.open)); const setCurrentSession = useSessionUIStore((state) => state.setCurrentSession); @@ -324,46 +280,34 @@ export const SessionSidebar: React.FC = ({ const updateStore = useUpdateStore(); const sessions = React.useMemo(() => { - if (!hasLoadedGlobalSessions) { - return syncSessions; - } - - if (syncSessions.length === 0) { - return globalActiveSessions; - } - - const syncedById = new Map(syncSessions.map((session) => [session.id, session])); - const merged = globalActiveSessions.map((session) => syncedById.get(session.id) ?? session); + const liveById = new Map(liveSessions.map((session) => [session.id, session])); + const merged = globalActiveSessions.map((session) => liveById.get(session.id) ?? session); const seenIds = new Set(merged.map((session) => session.id)); - syncSessions.forEach((session) => { + liveSessions.forEach((session) => { if (seenIds.has(session.id)) { return; } - - const sessionDirectory = resolveGlobalSessionDirectory(session); - if (sessionDirectory && sessionDirectory === currentDirectory) { - merged.push(session); - } + merged.push(session); }); return merged; - }, [currentDirectory, globalActiveSessions, hasLoadedGlobalSessions, syncSessions]); + }, [globalActiveSessions, liveSessions]); const syncSessionStructureSignature = React.useMemo( - () => syncSessions + () => liveSessions .map((session) => { const directory = normalizePath((session as Session & { directory?: string | null }).directory ?? null) ?? ''; return `${session.id}:${session.title ?? ''}:${session.time?.archived ? 1 : 0}:${directory}`; }) .join('|'), - [syncSessions], + [liveSessions], ); - const syncSessionsSnapshotRef = React.useRef(syncSessions); + const syncSessionsSnapshotRef = React.useRef(liveSessions); React.useEffect(() => { - syncSessionsSnapshotRef.current = syncSessions; - }, [syncSessionStructureSignature, syncSessions]); + syncSessionsSnapshotRef.current = liveSessions; + }, [syncSessionStructureSignature, liveSessions]); React.useEffect(() => { let cancelled = false; @@ -565,23 +509,6 @@ export const SessionSidebar: React.FC = ({ [sortedSessions], ); - const allKnownSessionsById = React.useMemo(() => { - const next = new Map(); - [...sessions, ...archivedSessions].forEach((session) => { - next.set(session.id, session); - }); - return next; - }, [sessions, archivedSessions]); - - React.useEffect(() => { - const pruned = pruneActiveNowEntries(activeNowEntries, allKnownSessionsById); - if (pruned.length === activeNowEntries.length && pruned.every((entry, index) => entry.sessionId === activeNowEntries[index]?.sessionId)) { - return; - } - setActiveNowEntries(pruned); - persistActiveNowEntries(safeStorage, pruned); - }, [activeNowEntries, allKnownSessionsById, safeStorage]); - const childrenMap = React.useMemo(() => { const map = new Map(); sortedSessions.forEach((session) => { @@ -1084,8 +1011,8 @@ export const SessionSidebar: React.FC = ({ }, [projectSections, homeDirectory]); const activeNowSessions = React.useMemo( - () => deriveActiveNowSessions(activeNowEntries, new Map(sessions.map((session) => [session.id, session]))), - [activeNowEntries, sessions], + () => deriveLiveActiveNowSessions(sessions, liveSessionStatuses), + [liveSessionStatuses, sessions], ); // Prefetch is wired below, after recentSessionIds is computed. @@ -1506,11 +1433,6 @@ export const SessionSidebar: React.FC = ({ ) : null} - - { .join('|'); }; +const treeContainsSessionId = (node: SessionNode, sessionId: string | null): boolean => { + if (!sessionId) { + return false; + } + + if (node.session.id === sessionId) { + return true; + } + + for (const child of node.children) { + if (treeContainsSessionId(child, sessionId)) { + return true; + } + } + + return false; +}; + const areEqual = (prev: Props, next: Props): boolean => { const prevSession = prev.node.session; const nextSession = next.node.session; @@ -121,7 +139,13 @@ const areEqual = (prev: Props, next: Props): boolean => { if (prev.groupDirectory !== next.groupDirectory) return false; if (prev.projectId !== next.projectId) return false; if (prev.archivedBucket !== next.archivedBucket) return false; - if ((prev.currentSessionId === prevSessionId) !== (next.currentSessionId === nextSessionId)) return false; + if (prev.currentSessionId !== next.currentSessionId) { + const prevActiveInTree = treeContainsSessionId(prev.node, prev.currentSessionId); + const nextActiveInTree = treeContainsSessionId(next.node, next.currentSessionId); + if (prevActiveInTree || nextActiveInTree) { + return false; + } + } if (prev.pinnedSessionIds.has(prevSessionId) !== next.pinnedSessionIds.has(nextSessionId)) return false; if (prev.expandedParents.has(prevSessionId) !== next.expandedParents.has(nextSessionId)) return false; if (prev.hasSessionSearchQuery !== next.hasSessionSearchQuery) return false; diff --git a/packages/ui/src/components/session/sidebar/activitySections.ts b/packages/ui/src/components/session/sidebar/activitySections.ts index 0de3683b..ce94cd5a 100644 --- a/packages/ui/src/components/session/sidebar/activitySections.ts +++ b/packages/ui/src/components/session/sidebar/activitySections.ts @@ -1,4 +1,5 @@ import type { Session } from '@opencode-ai/sdk/v2'; +import type { SessionStatus } from '@opencode-ai/sdk/v2/client'; export const ACTIVE_NOW_STORAGE_KEY = 'oc.sessions.activeNow'; export const ACTIVE_NOW_MAX_AGE_MS = 36 * 60 * 60 * 1000; @@ -104,4 +105,20 @@ export const deriveActiveNowSessions = ( return sortSessionsByUpdated(sessions); }; +export const deriveLiveActiveNowSessions = ( + sessions: Session[], + statuses: Record, +): Session[] => { + const activeSessions = sessions.filter((session) => { + if (isArchivedSession(session) || isSubtaskSession(session)) { + return false; + } + + const status = statuses[session.id]; + return status?.type === 'busy' || status?.type === 'retry'; + }); + + return sortSessionsByUpdated(activeSessions); +}; + export const getSessionUpdatedAtMs = getSessionUpdatedAt; diff --git a/packages/ui/src/stores/useGlobalSessionsStore.ts b/packages/ui/src/stores/useGlobalSessionsStore.ts index d78b1bd0..9dbd1d6d 100644 --- a/packages/ui/src/stores/useGlobalSessionsStore.ts +++ b/packages/ui/src/stores/useGlobalSessionsStore.ts @@ -109,6 +109,46 @@ const upsertSessionIntoList = (sessions: Session[], session: Session): Session[] return next; }; +const mergeSessionLists = (existing: Session[], incoming?: Session[]): Session[] => { + if (!incoming || incoming.length === 0) { + return existing; + } + + if (existing.length === 0) { + return incoming; + } + + const byId = new Map(existing.map((session) => [session.id, session])); + incoming.forEach((session) => { + byId.set(session.id, session); + }); + + const ordered: Session[] = []; + const seen = new Set(); + + existing.forEach((session) => { + const next = byId.get(session.id); + if (!next) { + return; + } + ordered.push(next); + seen.add(session.id); + }); + + incoming.forEach((session) => { + if (seen.has(session.id)) { + return; + } + const next = byId.get(session.id); + if (next) { + ordered.push(next); + seen.add(session.id); + } + }); + + return ordered; +}; + const applySnapshot = ( state: GlobalSessionsState, activeSessions: Session[], @@ -172,15 +212,16 @@ export const useGlobalSessionsStore = create((set, get) => listGlobalSessionPages(sdk, { archived: true, pageSize: PAGE_SIZE }), ]); + const fallbackSnapshot = mergeSessionLists(current.activeSessions, fallbackActive); const nextActiveSessions = activeResult.status === 'fulfilled' ? activeResult.value - : (fallbackActive ?? current.activeSessions); + : fallbackSnapshot; const nextArchivedSessions = archivedResult.status === 'fulfilled' ? archivedResult.value : current.archivedSessions; if (activeResult.status === 'rejected') { - console.warn('[GlobalSessions] Failed to load active sessions, using fallback:', activeResult.reason); + console.warn('[GlobalSessions] Failed to load active sessions, preserving existing snapshot with fallback merge:', activeResult.reason); } if (archivedResult.status === 'rejected') { console.warn('[GlobalSessions] Failed to load archived sessions, preserving current snapshot:', archivedResult.reason); @@ -189,7 +230,7 @@ export const useGlobalSessionsStore = create((set, get) => set((state) => applySnapshot(state, nextActiveSessions, nextArchivedSessions, 'ready')); return { activeSessions: nextActiveSessions, archivedSessions: nextArchivedSessions }; } catch (error) { - const nextActiveSessions = fallbackActive ?? current.activeSessions; + const nextActiveSessions = mergeSessionLists(current.activeSessions, fallbackActive); const nextArchivedSessions = current.archivedSessions; console.warn('[GlobalSessions] Failed to load sessions, using fallback snapshot:', error); set((state) => applySnapshot(state, nextActiveSessions, nextArchivedSessions, 'error')); diff --git a/packages/ui/src/sync/DOCUMENTATION.md b/packages/ui/src/sync/DOCUMENTATION.md index cebf8ab2..d270ff1f 100644 --- a/packages/ui/src/sync/DOCUMENTATION.md +++ b/packages/ui/src/sync/DOCUMENTATION.md @@ -20,7 +20,7 @@ There are **two distinct session data scopes** in the UI: - global archived sessions - active sessions indexed by directory -These two scopes are intentionally different. +These two scopes are intentionally different, but they are no longer equal peers for live UI truth. ### Why both exist @@ -34,7 +34,8 @@ The directory-scoped sync stores are **not** a complete global view. So: - Use the **directory sync stores** for per-directory live session/message state -- Use the **global sessions store** for sidebar/retention global session lists +- Use the **global sessions store** for cold/global session coverage (especially archived pages and unopened directories) +- Use **aggregated child-store snapshots** for live session/status truth across already initialized directories ## Ownership map @@ -62,12 +63,22 @@ Examples: ### Global session list -Use `useGlobalSessionsStore` when the UI needs a **shared global session view**. +Use `useGlobalSessionsStore` when the UI needs a **shared global session cache**. + +Current consumers: + +- `useSessionAutoCleanup.ts` + +### Live cross-directory session/status view + +Use the sync hooks backed by aggregated child stores when the UI needs **live truth** for sessions or statuses across all initialized directories. Current consumers: - `SessionSidebar.tsx` -- `useSessionAutoCleanup.ts` +- `SessionNodeItem.tsx` +- `Header.tsx` +- agent/session activity surfaces using `useGlobalSessionStatus()` / `useAllSessionStatuses()` ### Mutation responsibility @@ -83,7 +94,9 @@ Current consumers: - delete - retention cleanup batch archive/delete -This keeps sidebar/retention UI responsive without requiring a refetch after every change. +This keeps cold/global lists responsive without requiring a refetch after every change. + +Live activity/status indicators must not depend on this cache. They must derive from aggregated child-store state. ## Session action rules diff --git a/packages/ui/src/sync/__tests__/event-pipeline.test.js b/packages/ui/src/sync/__tests__/event-pipeline.test.js index c505d645..cbc32e85 100644 --- a/packages/ui/src/sync/__tests__/event-pipeline.test.js +++ b/packages/ui/src/sync/__tests__/event-pipeline.test.js @@ -236,7 +236,7 @@ describe('createEventPipeline', () => { expect(received[0].payload.type).toBe('server.connected'); }); - it('delivers message.part.delta events after a coalesced message.part.updated (no stale-delta skip)', async () => { + it('skips stale message.part.delta events after a newer message.part.updated for the same field', async () => { installDomStubs(); let releaseStream; @@ -246,8 +246,8 @@ describe('createEventPipeline', () => { const received = []; - // Simulate: part.updated arrives first, then delta, then part.updated again (coalesces with first). - // After coalescing, the delta should still be delivered — NOT skipped. + // Simulate: part.updated arrives, then delta, then a newer part.updated for the + // same part. The older queued delta becomes stale and must be skipped. const directory = '/test/dir'; const sdk = createSdkWithEvents([ // T0: message.part.updated for part-A @@ -260,7 +260,7 @@ describe('createEventPipeline', () => { }, }, }, - // T1: message.part.delta for part-A (should flow through even after coalesce) + // T1: message.part.delta for part-A (should be dropped as stale) { payload: { type: 'message.part.delta', @@ -290,7 +290,7 @@ describe('createEventPipeline', () => { sdk, onEvent: (dir, payload) => { received.push({ directory: dir, payload }); - if (received.length === 2) { + if (received.length === 1) { cleanup(); releaseStream(); resolve(); @@ -301,20 +301,39 @@ describe('createEventPipeline', () => { await delivered; - // Coalescing means T0 and T2 merge into one event at T0's queue position. - // The delta is a different event type with no coalesce key, so it gets - // its own queue slot. After coalesce: - // - queue[0] = coalesced part.updated (from T2, replacing T0) - // - queue[1] = part.delta (from T1) - // Total: 2 events delivered - expect(received.length).toBe(2); - - // The first event should be the coalesced message.part.updated + expect(received.length).toBe(1); expect(received[0].payload.type).toBe('message.part.updated'); + }); - // The delta MUST be delivered — it should NOT be skipped - expect(received[1].payload.type).toBe('message.part.delta'); - expect(received[1].payload.properties.delta).toBe(' world'); + it('keeps delta events for other fields on the same part', async () => { + const received = await runPipelineWithEvents([ + { + directory: 'dir-a', + payload: { + type: 'message.part.delta', + properties: { + messageID: 'msg-1', + partID: 'part-1', + field: 'reasoning', + delta: 'before', + }, + }, + }, + { + directory: 'dir-a', + payload: { + type: 'message.part.updated', + properties: { + part: { id: 'part-1', type: 'text', messageID: 'msg-1' }, + }, + }, + }, + ]); + + expect(received).toHaveLength(2); + expect(received[0].payload.type).toBe('message.part.delta'); + expect(received[0].payload.properties.field).toBe('reasoning'); + expect(received[1].payload.type).toBe('message.part.updated'); }); it('coalesces message.part.updated events for the same part', async () => { @@ -367,6 +386,62 @@ describe('createEventPipeline', () => { expect(received.length).toBe(1); expect(received[0].payload.type).toBe('message.part.updated'); }); + + it('routes events before queueing so coalescing happens on the resolved directory', async () => { + installDomStubs(); + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const sdk = createSdkWithEvents([ + { + directory: 'global', + payload: { + type: 'message.part.updated', + properties: { + part: { id: 'part-A', type: 'text', messageID: 'msg-1' }, + }, + }, + }, + { + directory: '/real-dir', + payload: { + type: 'message.part.updated', + properties: { + part: { id: 'part-A', type: 'text', messageID: 'msg-1', text: 'next' }, + }, + }, + }, + ], hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + routeDirectory: (directory, payload) => { + if (payload.type === 'message.part.updated') { + return '/resolved-dir'; + } + return directory; + }, + onEvent: (dir, payload) => { + received.push({ directory: dir, payload }); + cleanup(); + releaseStream(); + resolve(); + }, + }); + }); + + await delivered; + + expect(received).toHaveLength(1); + expect(received[0].directory).toBe('/resolved-dir'); + expect(received[0].payload.type).toBe('message.part.updated'); + expect(received[0].payload.properties.part.text).toBe('next'); + }); }); // --------------------------------------------------------------------------- diff --git a/packages/ui/src/sync/__tests__/event-reducer.test.js b/packages/ui/src/sync/__tests__/event-reducer.test.js index 3f7b5004..8647f455 100644 --- a/packages/ui/src/sync/__tests__/event-reducer.test.js +++ b/packages/ui/src/sync/__tests__/event-reducer.test.js @@ -151,4 +151,50 @@ describe('applyDirectoryEvent', () => { expect(state.part[messageID]?.[0]?.text).toBe('haha') }) + + it('does not let a stale running tool update overwrite a completed tool part', () => { + const state = structuredClone(INITIAL_STATE) + const messageID = 'msg-5' + const partID = 'part-5' + + applyDirectoryEvent(state, { + type: 'message.part.updated', + properties: { + part: { + id: partID, + type: 'tool', + messageID, + tool: 'apply_patch', + state: { + status: 'completed', + time: { + start: 10, + end: 20, + }, + }, + }, + }, + }) + + applyDirectoryEvent(state, { + type: 'message.part.updated', + properties: { + part: { + id: partID, + type: 'tool', + messageID, + tool: 'apply_patch', + state: { + status: 'running', + time: { + start: 10, + }, + }, + }, + }, + }) + + expect(state.part[messageID]?.[0]?.state?.status).toBe('completed') + expect(state.part[messageID]?.[0]?.state?.time?.end).toBe(20) + }) }) diff --git a/packages/ui/src/sync/__tests__/live-aggregate.test.js b/packages/ui/src/sync/__tests__/live-aggregate.test.js new file mode 100644 index 00000000..bfa83daa --- /dev/null +++ b/packages/ui/src/sync/__tests__/live-aggregate.test.js @@ -0,0 +1,78 @@ +import { describe, expect, it } from 'bun:test' + +import { + aggregateLiveSessions, + aggregateLiveSessionStatuses, + findLiveSession, + findLiveSessionStatus, +} from '../live-aggregate.ts' +import { deriveLiveActiveNowSessions } from '../../components/session/sidebar/activitySections.ts' + +const session = (id, directory, updated, extra = {}) => ({ + id, + title: `${id}-title`, + time: { created: updated - 1, updated, archived: undefined }, + directory, + ...extra, +}) + +describe('live aggregate', () => { + it('prefers the freshest live session snapshot across child stores', () => { + const states = [ + { + session: [session('ses-1', '/a', 10, { title: 'old' })], + session_status: {}, + }, + { + session: [session('ses-1', '/a', 25, { title: 'new' }), session('ses-2', '/b', 20)], + session_status: {}, + }, + ] + + const sessions = aggregateLiveSessions(states) + expect(sessions.map((item) => `${item.id}:${item.title}`)).toEqual(['ses-1:new', 'ses-2:ses-2-title']) + expect(findLiveSession(states, 'ses-1')?.title).toBe('new') + }) + + it('prefers busy/retry statuses over stale idle snapshots', () => { + const states = [ + { + session: [], + session_status: { + 'ses-1': { type: 'idle' }, + 'ses-2': { type: 'idle' }, + }, + }, + { + session: [], + session_status: { + 'ses-1': { type: 'busy' }, + 'ses-2': { type: 'retry', message: 'retrying' }, + }, + }, + ] + + const statuses = aggregateLiveSessionStatuses(states) + expect(statuses['ses-1']?.type).toBe('busy') + expect(statuses['ses-2']?.type).toBe('retry') + expect(findLiveSessionStatus(states, 'ses-2')?.type).toBe('retry') + }) + + it('derives active-now sessions from live statuses instead of persisted history', () => { + const sessions = [ + session('ses-1', '/a', 20), + session('ses-2', '/b', 30), + session('ses-3', '/c', 10, { time: { created: 9, updated: 10, archived: 50 } }), + session('ses-4', '/d', 40, { parentID: 'ses-parent' }), + ] + + const activeNow = deriveLiveActiveNowSessions(sessions, { + 'ses-1': { type: 'busy' }, + 'ses-2': { type: 'retry', message: 'retrying' }, + 'ses-3': { type: 'busy' }, + 'ses-4': { type: 'busy' }, + }) + + expect(activeNow.map((item) => item.id)).toEqual(['ses-2', 'ses-1']) + }) +}) diff --git a/packages/ui/src/sync/child-store.ts b/packages/ui/src/sync/child-store.ts index 6e757d65..3d32d49b 100644 --- a/packages/ui/src/sync/child-store.ts +++ b/packages/ui/src/sync/child-store.ts @@ -39,12 +39,19 @@ export class ChildStoreManager { private readonly lifecycle = new Map() private readonly pins = new Map() private readonly disposers = new Map void>() + private readonly registrySubscribers = new Set<() => void>() private onBootstrap?: (directory: string) => void private onDispose?: (directory: string) => void private isBooting?: (directory: string) => boolean private isLoadingSessions?: (directory: string) => boolean + private notifyRegistrySubscribers() { + for (const subscriber of this.registrySubscribers) { + subscriber() + } + } + configure(callbacks: { onBootstrap?: (directory: string) => void onDispose?: (directory: string) => void @@ -91,6 +98,7 @@ export class ChildStoreManager { if (!store) { store = createDirectoryStore(directory) this.children.set(directory, store) + this.notifyRegistrySubscribers() } this.mark(directory) @@ -122,6 +130,7 @@ export class ChildStoreManager { this.lifecycle.delete(directory) this.children.delete(directory) + this.notifyRegistrySubscribers() const dispose = this.disposers.get(directory) if (dispose) { dispose() @@ -165,8 +174,53 @@ export class ChildStoreManager { for (const directory of [...this.children.keys()]) { this.children.delete(directory) } + this.notifyRegistrySubscribers() this.lifecycle.clear() this.pins.clear() this.disposers.clear() } + + subscribeRegistry(listener: () => void): () => void { + this.registrySubscribers.add(listener) + return () => { + this.registrySubscribers.delete(listener) + } + } + + subscribeAll(listener: () => void): () => void { + const storeUnsubscribers = new Map void>() + + const syncStoreSubscriptions = () => { + const activeDirectories = new Set(this.children.keys()) + + for (const [directory, unsubscribe] of storeUnsubscribers.entries()) { + if (activeDirectories.has(directory)) { + continue + } + unsubscribe() + storeUnsubscribers.delete(directory) + } + + for (const [directory, store] of this.children.entries()) { + if (storeUnsubscribers.has(directory)) { + continue + } + storeUnsubscribers.set(directory, store.subscribe(listener)) + } + } + + syncStoreSubscriptions() + const unsubscribeRegistry = this.subscribeRegistry(() => { + syncStoreSubscriptions() + listener() + }) + + return () => { + unsubscribeRegistry() + for (const unsubscribe of storeUnsubscribers.values()) { + unsubscribe() + } + storeUnsubscribers.clear() + } + } } diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index bddf584f..5732ec2b 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -38,6 +38,7 @@ const HEARTBEAT_TIMEOUT_MS = 15_000 export type EventPipelineInput = { sdk: OpencodeClient onEvent: (directory: string, payload: Event) => void + routeDirectory?: (directory: string, payload: Event) => string /** Called after SSE reconnects (visibility restore or heartbeat timeout). */ onReconnect?: () => void } @@ -85,12 +86,13 @@ type DirectoryQueue = { queue: Event[] buffer: Event[] coalesced: Map + staleDeltas: Set timer: ReturnType | undefined last: number } export function createEventPipeline(input: EventPipelineInput) { - const { sdk, onEvent, onReconnect } = input + const { sdk, onEvent, onReconnect, routeDirectory } = input const abort = new AbortController() let hasConnected = false @@ -104,6 +106,7 @@ export function createEventPipeline(input: EventPipelineInput) { queue: [], buffer: [], coalesced: new Map(), + staleDeltas: new Set(), timer: undefined, last: 0, } @@ -136,6 +139,8 @@ export function createEventPipeline(input: EventPipelineInput) { return undefined } + const deltaKey = (messageID: string, partID: string, field: string) => `${messageID}:${partID}:${field}` + // Flush one directory — swap queue, dispatch events. // React 18 auto-batching still collapses the setState calls inside a single // directory's flush into one render pass. @@ -149,14 +154,22 @@ export function createEventPipeline(input: EventPipelineInput) { if (d.queue.length === 0) return const events = d.queue + const staleDeltas = d.staleDeltas.size > 0 ? new Set(d.staleDeltas) : undefined d.queue = d.buffer d.buffer = events d.queue.length = 0 d.coalesced.clear() + d.staleDeltas.clear() d.last = Date.now() syncDebug.pipeline.flush(events.length) for (const payload of events) { + if (staleDeltas && payload.type === "message.part.delta") { + const props = payload.properties as { messageID: string; partID: string; field: string } + if (staleDeltas.has(deltaKey(props.messageID, props.partID, props.field))) { + continue + } + } onEvent(directory, payload) } @@ -241,7 +254,8 @@ export function createEventPipeline(input: EventPipelineInput) { } const normalizedPayload = normalizeEventType(payload) const directory = resolveEventDirectory(event, normalizedPayload) - const d = getOrCreateDir(directory) + const routedDirectory = routeDirectory?.(directory, normalizedPayload) || directory + const d = getOrCreateDir(routedDirectory) const k = key(normalizedPayload) if (k) { const i = d.coalesced.get(k) @@ -261,14 +275,24 @@ export function createEventPipeline(input: EventPipelineInput) { } as unknown as Event } else { d.queue[i] = normalizedPayload + if (normalizedPayload.type === "message.part.updated") { + const part = (normalizedPayload.properties as { part: { messageID: string; id: string } }).part + d.staleDeltas.add(deltaKey(part.messageID, part.id, "text")) + d.staleDeltas.add(deltaKey(part.messageID, part.id, "output")) + } } syncDebug.pipeline.coalesced(normalizedPayload.type, k) continue } d.coalesced.set(k, d.queue.length) } + if (normalizedPayload.type === "message.part.updated") { + const part = (normalizedPayload.properties as { part: { messageID: string; id: string } }).part + d.staleDeltas.add(deltaKey(part.messageID, part.id, "text")) + d.staleDeltas.add(deltaKey(part.messageID, part.id, "output")) + } d.queue.push(normalizedPayload) - scheduleDir(directory) + scheduleDir(routedDirectory) if (Date.now() - yielded < STREAM_YIELD_MS) continue yielded = Date.now() diff --git a/packages/ui/src/sync/event-reducer.ts b/packages/ui/src/sync/event-reducer.ts index 2bdb07ff..a8d71eb8 100644 --- a/packages/ui/src/sync/event-reducer.ts +++ b/packages/ui/src/sync/event-reducer.ts @@ -17,6 +17,7 @@ import { syncDebug } from "./debug" const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) const DELTA_OVERLAP_FIELDS = ["text", "output"] as const +const FINAL_TOOL_STATUSES = new Set(["completed", "error", "aborted", "failed", "timeout", "cancelled"]) type DedupeMetadata = { __dedupeNextDeltaFields?: string[] @@ -50,6 +51,45 @@ function getUpdatedDeltaFields(previous: Part, next: Part) { return dedupeFields } +function getPartEndTime(part: Part): number | undefined { + const stateEnd = (part as { state?: { time?: { end?: unknown } } }).state?.time?.end + if (typeof stateEnd === "number") { + return stateEnd + } + + const timeEnd = (part as { time?: { end?: unknown } }).time?.end + return typeof timeEnd === "number" ? timeEnd : undefined +} + +function getToolStatus(part: Part): string | undefined { + if (part.type !== "tool") { + return undefined + } + + const status = (part as { state?: { status?: unknown } }).state?.status + return typeof status === "string" ? status : undefined +} + +function shouldPreserveExistingPart(previous: Part, next: Part): boolean { + if (previous.type !== "tool" || next.type !== "tool") { + return false + } + + const previousStatus = getToolStatus(previous) + const nextStatus = getToolStatus(next) + if (previousStatus && FINAL_TOOL_STATUSES.has(previousStatus) && (!nextStatus || !FINAL_TOOL_STATUSES.has(nextStatus))) { + return true + } + + const previousEnd = getPartEndTime(previous) + const nextEnd = getPartEndTime(next) + if (typeof previousEnd === "number" && typeof nextEnd !== "number") { + return true + } + + return false +} + // --------------------------------------------------------------------------- // Global events // --------------------------------------------------------------------------- @@ -239,6 +279,9 @@ export function applyDirectoryEvent( const result = Binary.search(next, part.id, (p) => p.id) if (result.found) { const previous = next[result.index] + if (shouldPreserveExistingPart(previous, part)) { + return false + } const dedupeFields = getUpdatedDeltaFields(previous, part) next[result.index] = dedupeFields.length > 0 ? { ...part, __dedupeNextDeltaFields: dedupeFields } as unknown as Part diff --git a/packages/ui/src/sync/live-aggregate.ts b/packages/ui/src/sync/live-aggregate.ts new file mode 100644 index 00000000..bbd65860 --- /dev/null +++ b/packages/ui/src/sync/live-aggregate.ts @@ -0,0 +1,172 @@ +import type { SessionStatus } from '@opencode-ai/sdk/v2/client' +import type { Session } from '@opencode-ai/sdk/v2' +import type { State } from './types' + +type LiveStateSlice = Pick + +const getSessionUpdatedAt = (session: Session): number => { + const updatedAt = session.time?.updated + if (typeof updatedAt === 'number' && Number.isFinite(updatedAt)) { + return updatedAt + } + + const createdAt = session.time?.created + return typeof createdAt === 'number' && Number.isFinite(createdAt) ? createdAt : 0 +} + +const getSessionSignature = (session: Session): string => { + const directory = (session as Session & { directory?: string | null }).directory ?? '' + const parentID = (session as Session & { parentID?: string | null }).parentID ?? '' + return [ + session.id, + session.title ?? '', + session.time?.created ?? 0, + session.time?.updated ?? 0, + session.time?.archived ?? 0, + directory, + parentID, + session.share?.url ?? '', + ].join('|') +} + +const getStatusPriority = (status: SessionStatus | undefined): number => { + switch (status?.type) { + case 'retry': + return 4 + case 'busy': + return 3 + case 'idle': + return 1 + default: + return 0 + } +} + +const getStatusMessage = (status: SessionStatus | undefined): string | null => { + const message = (status as { message?: unknown } | undefined)?.message + return typeof message === 'string' ? message : null +} + +export const areSessionListsEquivalent = (left: Session[], right: Session[]): boolean => { + if (left === right) { + return true + } + if (left.length !== right.length) { + return false + } + + for (let index = 0; index < left.length; index += 1) { + if (getSessionSignature(left[index]) !== getSessionSignature(right[index])) { + return false + } + } + + return true +} + +export const areStatusMapsEquivalent = ( + left: Record, + right: Record, +): 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 (!(key in right)) { + return false + } + const leftStatus = left[key] + const rightStatus = right[key] + if (leftStatus?.type !== rightStatus?.type) { + return false + } + if (getStatusMessage(leftStatus) !== getStatusMessage(rightStatus)) { + return false + } + } + + return true +} + +export function aggregateLiveSessions(states: Iterable): Session[] { + const sessionsById = new Map() + + for (const state of states) { + for (const session of state.session) { + if (!session?.id) { + continue + } + const current = sessionsById.get(session.id) + if (!current || getSessionUpdatedAt(session) >= getSessionUpdatedAt(current)) { + sessionsById.set(session.id, session) + } + } + } + + return Array.from(sessionsById.values()).sort((left, right) => { + return getSessionUpdatedAt(right) - getSessionUpdatedAt(left) + }) +} + +export function aggregateLiveSessionStatuses(states: Iterable): Record { + const statuses: Record = {} + + for (const state of states) { + for (const [sessionId, status] of Object.entries(state.session_status ?? {})) { + const current = statuses[sessionId] + if (!current || getStatusPriority(status) >= getStatusPriority(current)) { + statuses[sessionId] = status + } + } + } + + return statuses +} + +export function findLiveSession(states: Iterable, sessionID?: string | null): Session | undefined { + if (!sessionID) { + return undefined + } + + let match: Session | undefined + for (const state of states) { + const session = state.session.find((candidate) => candidate.id === sessionID) + if (!session) { + continue + } + if (!match || getSessionUpdatedAt(session) >= getSessionUpdatedAt(match)) { + match = session + } + } + + return match +} + +export function findLiveSessionStatus( + states: Iterable, + sessionID?: string | null, +): SessionStatus | undefined { + if (!sessionID) { + return undefined + } + + let match: SessionStatus | undefined + for (const state of states) { + const status = state.session_status?.[sessionID] + if (!status) { + continue + } + if (!match || getStatusPriority(status) >= getStatusPriority(match)) { + match = status + } + } + + return match +} diff --git a/packages/ui/src/sync/session-actions.ts b/packages/ui/src/sync/session-actions.ts index 9002805c..e4a05890 100644 --- a/packages/ui/src/sync/session-actions.ts +++ b/packages/ui/src/sync/session-actions.ts @@ -7,22 +7,21 @@ import type { OpencodeClient, Session, Message, Part } from "@opencode-ai/sdk/v2 import { Binary } from "./binary" import { useSessionUIStore } from "./session-ui-store" import { useInputStore } from "./input-store" -import type { DirectoryStore } from "./child-store" -import type { StoreApi } from "zustand" +import type { ChildStoreManager } from "./child-store" import { opencodeClient } from "@/lib/opencode/client" import { useGlobalSessionsStore } from "@/stores/useGlobalSessionsStore" import { registerSessionDirectory } from "./sync-refs" // Reference set by SyncProvider — allows actions to access SDK and stores let _sdk: OpencodeClient | null = null -let _childStores: { ensureChild: (dir: string) => StoreApi } | null = null +let _childStores: ChildStoreManager | null = null let _getDirectory: () => string = () => "" let _optimisticAdd: ((input: { sessionID: string; message: Message; parts: Part[] }) => void) | null = null let _optimisticRemove: ((input: { sessionID: string; messageID: string }) => void) | null = null export function setActionRefs( sdk: OpencodeClient, - childStores: { ensureChild: (dir: string) => StoreApi }, + childStores: ChildStoreManager, getDirectory: () => string, ) { _sdk = sdk @@ -75,6 +74,59 @@ function getSessionReplyClient(sessionId?: string): OpencodeClient { return sdk() } +function resolveDirectoryForBlockingRequest( + type: "permission" | "question", + sessionId: string, + requestId: string, +): string | null { + const stores = _childStores + if (!stores || !requestId) { + return null + } + + for (const [directory, store] of stores.children) { + const state = store.getState() + const requestMap = type === "permission" ? state.permission : state.question + for (const requests of Object.values(requestMap) as Array | undefined>) { + if (requests?.some((request) => request.id === requestId)) { + return directory + } + } + } + + const sessionDirectory = useSessionUIStore.getState().getDirectoryForSession(sessionId) + if (sessionDirectory) { + return sessionDirectory + } + + for (const [directory, store] of stores.children) { + const state = store.getState() + if ( + state.session.some((session) => session.id === sessionId) + || Object.prototype.hasOwnProperty.call(state.message, sessionId) + || Object.prototype.hasOwnProperty.call(state.session_status ?? {}, sessionId) + || Object.prototype.hasOwnProperty.call(state.permission ?? {}, sessionId) + || Object.prototype.hasOwnProperty.call(state.question ?? {}, sessionId) + ) { + return directory + } + } + + return null +} + +function getRequestReplyClient( + type: "permission" | "question", + sessionId: string, + requestId: string, +): OpencodeClient { + const requestDirectory = resolveDirectoryForBlockingRequest(type, sessionId, requestId) + if (requestDirectory) { + return opencodeClient.getScopedSdkClient(requestDirectory) + } + return getSessionReplyClient(sessionId) +} + // --------------------------------------------------------------------------- // Session CRUD // --------------------------------------------------------------------------- @@ -356,7 +408,7 @@ export async function respondToPermission( requestId: string, response: "once" | "always" | "reject", ): Promise { - const result = await getSessionReplyClient(sessionId).permission.reply({ + const result = await getRequestReplyClient("permission", sessionId, requestId).permission.reply({ requestID: requestId, reply: response, }) @@ -369,7 +421,7 @@ export async function dismissPermission( sessionId: string, requestId: string, ): Promise { - const result = await getSessionReplyClient(sessionId).permission.reply({ + const result = await getRequestReplyClient("permission", sessionId, requestId).permission.reply({ requestID: requestId, reply: "reject", }) @@ -387,7 +439,7 @@ export async function respondToQuestion( requestId: string, answers: string[] | string[][], ): Promise { - const result = await getSessionReplyClient(sessionId).question.reply({ + const result = await getRequestReplyClient("question", sessionId, requestId).question.reply({ requestID: requestId, answers: answers as Array>, }) @@ -400,7 +452,7 @@ export async function rejectQuestion( sessionId: string, requestId: string, ): Promise { - const result = await getSessionReplyClient(sessionId).question.reject({ + const result = await getRequestReplyClient("question", sessionId, requestId).question.reject({ requestID: requestId, }) if (!result.data) { diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index 36d20a7d..c07198ee 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -9,6 +9,14 @@ import { createEventPipeline } from "./event-pipeline" import { reduceGlobalEvent, applyGlobalProject, applyDirectoryEvent } from "./event-reducer" import { useGlobalSyncStore, type GlobalSyncStore } from "./global-sync-store" import { ChildStoreManager, type DirectoryStore } from "./child-store" +import { + aggregateLiveSessions, + aggregateLiveSessionStatuses, + areSessionListsEquivalent, + areStatusMapsEquivalent, + findLiveSession, + findLiveSessionStatus, +} from "./live-aggregate" import { bootstrapGlobal, bootstrapDirectory } from "./bootstrap" import { retry } from "./retry" import { updateStreamingState } from "./streaming" @@ -24,7 +32,6 @@ import type { State } from "./types" import type { SessionStatus } from "@opencode-ai/sdk/v2/client" import type { PermissionRequest } from "@/types/permission" import type { QuestionRequest } from "@/types/question" -import { create } from "zustand" import * as sessionActions from "./session-actions" // --------------------------------------------------------------------------- @@ -52,44 +59,59 @@ function useSyncSystem() { return ctx } +function getLiveStates(childStores: ChildStoreManager): State[] { + return Array.from(childStores.children.values(), (store) => store.getState()) +} + +function useLiveSyncSelector(selector: (states: State[]) => T, isEqual: (left: T, right: T) => boolean = Object.is): T { + const { childStores } = useSyncSystem() + const cacheRef = useRef(undefined) + const initializedRef = useRef(false) + + const getSnapshot = useCallback(() => { + const next = selector(getLiveStates(childStores)) + if (initializedRef.current && isEqual(cacheRef.current as T, next)) { + return cacheRef.current as T + } + + cacheRef.current = next + initializedRef.current = true + return next + }, [childStores, isEqual, selector]) + + return React.useSyncExternalStore( + useCallback((notify) => childStores.subscribeAll(notify), [childStores]), + getSnapshot, + getSnapshot, + ) +} + // --------------------------------------------------------------------------- // Event handler — applies one SSE event at a time to the live store. // Each event reads live state, creates a shallow draft, applies, writes back. // React 18 batches synchronous setState calls automatically. // --------------------------------------------------------------------------- -// --------------------------------------------------------------------------- -// Global session status store — cross-directory status tracking. -// -// OpenCode isolates sessions behind project navrails, so per-directory -// session_status is sufficient. OpenChamber shows all sessions in one sidebar, -// so we need a global view. Updated from handleEvent on every session.status. -// --------------------------------------------------------------------------- - -interface GlobalSessionStatusStore { - statuses: Record -} - -const useGlobalSessionStatusStore = create(() => ({ - statuses: {}, -})) - -function setGlobalSessionStatus(sessionId: string, status: SessionStatus) { - const current = useGlobalSessionStatusStore.getState().statuses - if (current[sessionId] === status) return - useGlobalSessionStatusStore.setState({ - statuses: { ...current, [sessionId]: status }, - }) -} - /** Read status for a session across all directories */ export function useGlobalSessionStatus(sessionId: string): SessionStatus | undefined { - return useGlobalSessionStatusStore((s) => s.statuses[sessionId]) + return useLiveSyncSelector( + useCallback((states) => findLiveSessionStatus(states, sessionId), [sessionId]), + ) } /** Read all session statuses (for sidebar) */ export function useAllSessionStatuses(): Record { - return useGlobalSessionStatusStore((s) => s.statuses) + return useLiveSyncSelector( + useCallback((states) => aggregateLiveSessionStatuses(states), []), + areStatusMapsEquivalent, + ) +} + +export function useAllLiveSessions(): Session[] { + return useLiveSyncSelector( + useCallback((states) => aggregateLiveSessions(states), []), + areSessionListsEquivalent, + ) } // Boot debounce — suppresses redundant refresh/re-bootstrap events during startup. @@ -108,6 +130,34 @@ const requestSignature = (items: Array<{ id: string }> | undefined): string => { const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) +const partRepairSignature = (part: Part): string => JSON.stringify(part) + +function haveEquivalentPartSnapshots(left: Part[] | undefined, right: Part[]): boolean { + if (!left) { + return right.length === 0 + } + + if (left.length !== right.length) { + return false + } + + for (let index = 0; index < left.length; index += 1) { + const leftPart = left[index] + const rightPart = right[index] + if (!leftPart || !rightPart) { + return false + } + if (leftPart.id !== rightPart.id) { + return false + } + if (partRepairSignature(leftPart) !== partRepairSignature(rightPart)) { + return false + } + } + + return true +} + // --------------------------------------------------------------------------- // Parts-gap recovery — when SSE events arrive but parts are missing, // trigger a targeted re-fetch for the affected sessions. @@ -172,8 +222,8 @@ async function repairSessionParts( .sort((a: Part, b: Part) => cmp(a.id, b.id)) const existing = nextPartState[messageId] - // Only patch if parts were missing or fewer than server has - if (!existing || existing.length < newParts.length) { + // Repair when parts are missing, truncated, or stale-but-same-length. + if (!haveEquivalentPartSnapshots(existing, newParts)) { nextPartState[messageId] = newParts } } @@ -201,22 +251,6 @@ function isRecentBoot() { return bootingRoot || Date.now() - bootedAt < BOOT_DEBOUNCE_MS } -function setGlobalSessionStatuses(nextStatuses: Record) { - const current = useGlobalSessionStatusStore.getState().statuses - let changed = false - const merged = { ...current } - - for (const [sessionId, status] of Object.entries(nextStatuses)) { - if (!status || merged[sessionId] === status) continue - merged[sessionId] = status - changed = true - } - - if (changed) { - useGlobalSessionStatusStore.setState({ statuses: merged }) - } -} - function getReconnectCandidateSessionIds(state: State) { const ids = new Set() @@ -662,7 +696,6 @@ async function resyncDirectoryAfterReconnect( store.setState((state: DirectoryStore) => ({ session_status: { ...state.session_status, ...relevantStatuses }, })) - setGlobalSessionStatuses(relevantStatuses) } const scopedClient = opencodeClient.getScopedSdkClient(directory) @@ -733,6 +766,13 @@ async function resyncDirectoryAfterReconnect( // If SSE changed a session while the request was in-flight, keep that data. try { const before = store.getState() + const knownSessionIds = new Set([ + ...before.session.map((session) => session.id), + ...Object.keys(before.message ?? {}), + ...Object.keys(before.session_status ?? {}), + ...Object.keys(before.question ?? {}), + ...Object.keys(before.permission ?? {}), + ]) const beforeSignatures = new Map( candidateSessionIds.map((sessionId) => [sessionId, requestSignature(before.question[sessionId])]), ) @@ -740,6 +780,7 @@ async function resyncDirectoryAfterReconnect( const grouped: Record = {} for (const q of pendingQuestions) { if (!q?.id || !q.sessionID) continue + if (!knownSessionIds.has(q.sessionID)) continue const list = grouped[q.sessionID] if (list) list.push(q) else grouped[q.sessionID] = [q] @@ -956,17 +997,6 @@ function handleEvent( updateRoutingIndexFromEvent(routingIndex, resolvedDirectory, payload) - // Update global session status for cross-directory sidebar visibility - if (payload.type === "session.status") { - const props = payload.properties as { sessionID: string; status: SessionStatus } - setGlobalSessionStatus(props.sessionID, props.status) - } - - if (payload.type === "session.idle" || payload.type === "session.error") { - const props = payload.properties as { sessionID: string } - setGlobalSessionStatus(props.sessionID, { type: "idle" }) - } - if (payload.type === "permission.asked") { const nd = normalizeDirectory(resolvedDirectory) if (!nd) { @@ -1030,11 +1060,6 @@ export function SyncProvider(props: { if (patch.session || patch.message) { ingestDirectoryStateIntoRoutingIndex(routingIndex, directory, store.getState()) } - if (patch.session_status) { - const current = useGlobalSessionStatusStore.getState().statuses - const merged = { ...current, ...patch.session_status } - useGlobalSessionStatusStore.setState({ statuses: merged }) - } }, global: { config: globalState.config, @@ -1103,6 +1128,9 @@ export function SyncProvider(props: { const { cleanup } = createEventPipeline({ sdk: props.sdk, + routeDirectory: (directory, payload) => { + return resolveDirectoryFromRoutingIndex(routingIndex, directory, payload, childStores) + }, onEvent: (directory, payload) => { handleEvent(directory, payload, childStores, routingIndex) }, @@ -1381,31 +1409,28 @@ export function useSidebarSessions(directory?: string): Session[] { /** Get one session by id for a directory */ export function useSession(sessionID?: string | null, directory?: string) { - return useDirectorySync( - useCallback( - (state: State) => { - if (!sessionID) return undefined - return state.session.find((session) => session.id === sessionID) - }, - [sessionID], - ), - directory, - ) + const { childStores } = useSyncSystem() + const getSnapshot = useCallback(() => { + if (directory) { + return childStores.getChild(directory)?.getState().session.find((session) => session.id === sessionID) + } + return findLiveSession(getLiveStates(childStores), sessionID) + }, [childStores, directory, sessionID]) + + const subscribe = useCallback((notify: () => void) => { + if (directory) { + return childStores.ensureChild(directory).subscribe(notify) + } + return childStores.subscribeAll(notify) + }, [childStores, directory]) + + return React.useSyncExternalStore(subscribe, getSnapshot, getSnapshot) } /** Get one session directory by id for a directory */ export function useSessionDirectory(sessionID?: string | null, directory?: string): string | undefined { - return useDirectorySync( - useCallback( - (state: State) => { - if (!sessionID) return undefined - const session = state.session.find((candidate) => candidate.id === sessionID) - return (session as (typeof session & { directory?: string | null }) | undefined)?.directory ?? undefined - }, - [sessionID], - ), - directory, - ) + const session = useSession(sessionID, directory) + return (session as (typeof session & { directory?: string | null }) | undefined)?.directory ?? undefined } /** Get the SDK client */ @@ -1451,42 +1476,84 @@ const getFirstTextFromParts = (parts: Part[]): string => { return "" } -function usePartsSnapshotForMessageIds(messageIds: string[], directory?: string, suspendUpdates = false) { - const store = useDirectoryStore(directory) - const prevPartsRef = useRef>({}) - const [partsSnapshot, setPartsSnapshot] = React.useState>({}) +type SessionMessageRecord = { info: Message; parts: Part[] } - React.useEffect(() => { - const flush = () => { - const state = store.getState() - const prev = prevPartsRef.current - let changed = false - const next: Record = {} - for (const id of messageIds) { - const parts = state.part[id] ?? EMPTY_PARTS - next[id] = prev[id] === parts ? prev[id] : parts - if (next[id] !== prev[id]) changed = true - } - if (changed || Object.keys(prev).length !== messageIds.length) { - prevPartsRef.current = next - setPartsSnapshot(next) - } +type SessionMessageRecordsSnapshot = { + sessionID: string + sourceMessages: Message[] + visibleMessages: Message[] + revertMessageID?: string + list: SessionMessageRecord[] + byId: Map +} + +function getVisibleMessagesForSession(state: State, sessionID: string, previous?: SessionMessageRecordsSnapshot): { + sourceMessages: Message[] + visibleMessages: Message[] + revertMessageID?: string +} { + const sourceMessages = state.message[sessionID] ?? EMPTY_MESSAGES + const session = state.session.find((candidate) => candidate.id === sessionID) + const revertMessageID = (session as { revert?: { messageID?: string } } | undefined)?.revert?.messageID + + if ( + previous + && previous.sourceMessages === sourceMessages + && previous.revertMessageID === revertMessageID + ) { + return { + sourceMessages, + visibleMessages: previous.visibleMessages, + revertMessageID, } + } - flush() + return { + sourceMessages, + visibleMessages: revertMessageID ? sourceMessages.filter((message) => message.id < revertMessageID) : sourceMessages, + revertMessageID, + } +} - if (suspendUpdates) { - return - } +function buildSessionMessageRecordsSnapshot( + state: State, + sessionID: string, + previous?: SessionMessageRecordsSnapshot, + suspendPartUpdates = false, +): SessionMessageRecordsSnapshot { + const { sourceMessages, visibleMessages, revertMessageID } = getVisibleMessagesForSession(state, sessionID, previous) + const nextById = new Map() + const nextList = visibleMessages.map((message) => { + const previousRecord = previous?.byId.get(message.id) + const parts = suspendPartUpdates && previousRecord + ? previousRecord.parts + : (state.part[message.id] ?? EMPTY_PARTS) - const unsub = store.subscribe(flush) + const nextRecord = previousRecord && previousRecord.info === message && previousRecord.parts === parts + ? previousRecord + : { info: message, parts } - return () => { - unsub() - } - }, [messageIds, store, suspendUpdates]) + nextById.set(message.id, nextRecord) + return nextRecord + }) - return partsSnapshot + const unchanged = Boolean(previous) + && previous?.visibleMessages === visibleMessages + && previous.list.length === nextList.length + && previous.list.every((record, index) => record === nextList[index]) + + if (unchanged && previous) { + return previous + } + + return { + sessionID, + sourceMessages, + visibleMessages, + revertMessageID, + list: nextList, + byId: nextById, + } } export function useSessionMessageCount(sessionID: string, directory?: string): number { @@ -1500,40 +1567,33 @@ export function useSessionMessageCount(sessionID: string, directory?: string): n } export function useSessionTextMessages(sessionID: string, directory?: string): SessionTextMessage[] { - const messages = useVisibleSessionMessages(sessionID, directory) - const messageIds = useMemo(() => messages.map((message) => message.id), [messages]) - const partsSnapshot = usePartsSnapshotForMessageIds(messageIds, directory) + const records = useSessionMessageRecords(sessionID, directory) return useMemo( - () => messages.map((message) => ({ - id: message.id, - role: typeof message.role === "string" ? message.role : null, - text: getConcatenatedTextFromParts(partsSnapshot[message.id] ?? EMPTY_PARTS), + () => records.map((record) => ({ + id: record.info.id, + role: typeof record.info.role === "string" ? record.info.role : null, + text: getConcatenatedTextFromParts(record.parts), })), - [messages, partsSnapshot], + [records], ) } export function useUserMessageHistory(sessionID: string, directory?: string): string[] { - const messages = useVisibleSessionMessages(sessionID, directory) - const userMessages = useMemo( - () => messages.filter((message) => message.role === "user"), - [messages], - ) - const userMessageIds = useMemo(() => userMessages.map((message) => message.id), [userMessages]) - const partsSnapshot = usePartsSnapshotForMessageIds(userMessageIds, directory) + const records = useSessionMessageRecords(sessionID, directory) + const userMessages = useMemo(() => records.filter((record) => record.info.role === 'user'), [records]) return useMemo(() => { const history: string[] = [] for (let index = userMessages.length - 1; index >= 0; index -= 1) { const message = userMessages[index] - const text = getFirstTextFromParts(partsSnapshot[message.id] ?? EMPTY_PARTS) + const text = getFirstTextFromParts(message.parts) if (text.length > 0) { history.push(text) } } return history - }, [partsSnapshot, userMessages]) + }, [userMessages]) } /** @@ -1548,44 +1608,28 @@ export function useSessionMessageRecords( directory?: string, options?: { suspendPartUpdates?: boolean }, ) { - const messages = useVisibleSessionMessages(sessionID, directory) - const messageIds = useMemo(() => messages.map((message) => message.id), [messages]) - const partsSnapshot = usePartsSnapshotForMessageIds(messageIds, directory, Boolean(options?.suspendPartUpdates)) - const previousRecordsRef = useRef<{ - list: Array<{ info: (typeof messages)[number]; parts: Part[] }> - byId: Map - }>({ + const store = useDirectoryStore(directory) + const snapshotRef = useRef({ + sessionID, + sourceMessages: EMPTY_MESSAGES, + visibleMessages: EMPTY_MESSAGES, + revertMessageID: undefined, list: [], byId: new Map(), }) - return useMemo(() => { - const previous = previousRecordsRef.current - const nextById = new Map() - const nextList = messages.map((message) => { - const parts = partsSnapshot[message.id] ?? EMPTY_PARTS - const previousRecord = previous.byId.get(message.id) - const record = previousRecord && previousRecord.info === message && previousRecord.parts === parts - ? previousRecord - : { info: message, parts } - nextById.set(message.id, record) - return record - }) + const getSnapshot = useCallback(() => { + const nextSnapshot = buildSessionMessageRecordsSnapshot( + store.getState(), + sessionID, + snapshotRef.current.sessionID === sessionID ? snapshotRef.current : undefined, + Boolean(options?.suspendPartUpdates), + ) + snapshotRef.current = nextSnapshot + return nextSnapshot.list + }, [options?.suspendPartUpdates, sessionID, store]) - const unchanged = previous.list.length === nextList.length - && previous.list.every((record, index) => record === nextList[index]) - - if (unchanged) { - return previous.list - } - - previousRecordsRef.current = { - list: nextList, - byId: nextById, - } - - return nextList - }, [messages, partsSnapshot]) + return React.useSyncExternalStore(store.subscribe, getSnapshot, getSnapshot) } /**