BuzzbeeSCD/openchamber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(sync): unify live session truth across chat and sidebar

Browse Source
This commit is contained in:
Bohdan Triapitsyn
2026-04-16 19:16:32 +03:00
parent 4c4f24a404
commit ddc1039d1c
15 changed files with 904 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ui/src/components/layout/Header.tsx
+5 -5
View File
@@ -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<HeaderProps> = ({
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<HeaderProps> = ({
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;
packages/ui/src/components/session/SessionSidebar.tsx
+17 -95
View File
@@ -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<React.SetStateAction<ActiveNowEntry[]>>;
};
const SessionStatusActivityBridge: React.FC<SessionStatusActivityBridgeProps> = ({
safeStorage,
setActiveNowEntries,
}) => {
const globalSessionStatuses = useAllSessionStatuses();
const sessionStatus = React.useMemo(
() => new Map(Object.entries(globalSessionStatuses)),
[globalSessionStatuses],
);
React.useEffect(() => {
const nextStreamingIds = new Set<string>();
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<SessionSidebarProps> = ({
mobileVariant = false,
onSessionSelected,
@@ -172,7 +129,6 @@ export const SessionSidebar: React.FC<SessionSidebarProps> = ({
() => new Map(),
);
const safeStorage = React.useMemo(() => getSafeStorage(), []);
const [activeNowEntries, setActiveNowEntries] = React.useState(() => readActiveNowEntries(safeStorage));
const [collapsedProjects, setCollapsedProjects] = React.useState<Set<string>>(new Set());
const [projectRepoStatus, setProjectRepoStatus] = React.useState<Map<string, boolean | null>>(new Map());
@@ -307,10 +263,10 @@ export const SessionSidebar: React.FC<SessionSidebarProps> = ({
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<SessionSidebarProps> = ({
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<Session[]>(syncSessions);
const syncSessionsSnapshotRef = React.useRef<Session[]>(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<SessionSidebarProps> = ({
[sortedSessions],
);
const allKnownSessionsById = React.useMemo(() => {
const next = new Map<string, Session>();
[...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<string, Session[]>();
sortedSessions.forEach((session) => {
@@ -1084,8 +1011,8 @@ export const SessionSidebar: React.FC<SessionSidebarProps> = ({
}, [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<SessionSidebarProps> = ({
</div>
) : null}
<SessionStatusActivityBridge
safeStorage={safeStorage}
setActiveNowEntries={setActiveNowEntries}
/>
<SidebarHeader
hideDirectoryControls={hideDirectoryControls}
handleOpenDirectoryDialog={handleOpenDirectoryDialog}
packages/ui/src/components/session/sidebar/SessionNodeItem.tsx
+25 -1
View File
@@ -108,6 +108,24 @@ const getNodeChildSignature = (node: SessionNode): string => {
.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;
packages/ui/src/components/session/sidebar/activitySections.ts
+17
View File
@@ -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<string, SessionStatus>,
): 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;
packages/ui/src/stores/useGlobalSessionsStore.ts
+44 -3
View File
@@ -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<string>();
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<GlobalSessionsState>((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<GlobalSessionsState>((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'));
packages/ui/src/sync/DOCUMENTATION.md
+18 -5
View File
@@ -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
packages/ui/src/sync/__tests__/event-pipeline.test.js
+92 -17
View File
@@ -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');
});
});
// ---------------------------------------------------------------------------
packages/ui/src/sync/__tests__/event-reducer.test.js
+46
View File
@@ -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)
})
})
packages/ui/src/sync/__tests__/live-aggregate.test.js
+78
View File
@@ -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'])
})
})
packages/ui/src/sync/child-store.ts
+54
View File
@@ -39,12 +39,19 @@ export class ChildStoreManager {
private readonly lifecycle = new Map<string, DirState>()
private readonly pins = new Map<string, number>()
private readonly disposers = new Map<string, () => 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<string, () => 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()
}
}
}
packages/ui/src/sync/event-pipeline.ts
+27 -3
View File
@@ -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<string, number>
staleDeltas: Set<string>
timer: ReturnType<typeof setTimeout> | 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()
packages/ui/src/sync/event-reducer.ts
+43
View File
@@ -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
packages/ui/src/sync/live-aggregate.ts
+172
View File
@@ -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<State, 'session' | 'session_status'>
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<string, SessionStatus>,
right: Record<string, SessionStatus>,
): 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<LiveStateSlice>): Session[] {
const sessionsById = new Map<string, Session>()
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<LiveStateSlice>): Record<string, SessionStatus> {
const statuses: Record<string, SessionStatus> = {}
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<LiveStateSlice>, 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<LiveStateSlice>,
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
}
packages/ui/src/sync/session-actions.ts
+60 -8
View File
@@ -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<DirectoryStore> } | 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<DirectoryStore> },
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<Array<{ id: string }> | 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<void> {
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<void> {
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<void> {
const result = await getSessionReplyClient(sessionId).question.reply({
const result = await getRequestReplyClient("question", sessionId, requestId).question.reply({
requestID: requestId,
answers: answers as Array<Array<string>>,
})
@@ -400,7 +452,7 @@ export async function rejectQuestion(
sessionId: string,
requestId: string,
): Promise<void> {
const result = await getSessionReplyClient(sessionId).question.reject({
const result = await getRequestReplyClient("question", sessionId, requestId).question.reject({
requestID: requestId,
})
if (!result.data) {
packages/ui/src/sync/sync-context.tsx
+206 -162
View File
@@ -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<T>(selector: (states: State[]) => T, isEqual: (left: T, right: T) => boolean = Object.is): T {
const { childStores } = useSyncSystem()
const cacheRef = useRef<T | undefined>(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<string, SessionStatus>
}
const useGlobalSessionStatusStore = create<GlobalSessionStatusStore>(() => ({
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<string, SessionStatus> {
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<string, SessionStatus>) {
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<string>()
@@ -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<string>([
...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<string, QuestionRequest[]> = {}
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<Record<string, Part[]>>({})
const [partsSnapshot, setPartsSnapshot] = React.useState<Record<string, Part[]>>({})
type SessionMessageRecord = { info: Message; parts: Part[] }
React.useEffect(() => {
const flush = () => {
const state = store.getState()
const prev = prevPartsRef.current
let changed = false
const next: Record<string, Part[]> = {}
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<string, SessionMessageRecord>
}
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<string, SessionMessageRecord>()
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<string, { info: (typeof messages)[number]; parts: Part[] }>
}>({
const store = useDirectoryStore(directory)
const snapshotRef = useRef<SessionMessageRecordsSnapshot>({
sessionID,
sourceMessages: EMPTY_MESSAGES,
visibleMessages: EMPTY_MESSAGES,
revertMessageID: undefined,
list: [],
byId: new Map(),
})
return useMemo(() => {
const previous = previousRecordsRef.current
const nextById = new Map<string, { info: (typeof messages)[number]; parts: Part[] }>()
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)
}
/**