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

feat(session-status): mobile session quick switch with running/unread indicators (#340)

Browse Source 
* feat(session-status): implement server-authoritative session status tracking

- Add server-side session states and attention tracking with Map storage
- Implement SSE event broadcasting (openchamber:session-status)
- Add REST API endpoints for status/attention queries
- Replace client-side polling with HTTP polling + SSE push
- Track needsAttention based on user message + completion + unviewed
- Add MobileSessionStatusBar for mobile UI
- Handle session view/unview/message-sent state transitions
- Add 24h cleanup for old session states

* feat(session-status): add unread indicator to mobile session status bar

* refactor(session-status): extract SessionStatusHeader component

Extract the session status header into a reusable component and improve
layout structure in CollapsedView and ExpandedView.

* feat(session-status): optimize mobile session status bar UI

* refactor(session-status): remove interval polling, use snapshot sync on reconnect/visibility

- unifies session status/attention sync with existing SSE lifecycle, fixes no-op store churn bug, and drops duplicated client activity state while preserving mobile unread/running UX.

---------

Co-authored-by: Jovines <jovines@qq.com>
Co-authored-by: Bohdan Triapitsyn <artmore@protonmail.com>
This commit is contained in:
Jovines
2026-02-07 03:11:55 +02:00
committed by GitHub
co-authored by Jovines Bohdan Triapitsyn
parent aa85e31420
commit d5a2e49ae8
10 changed files with 1283 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ui/src/hooks/useEventStream.ts
+89 -129
View File
@@ -2,6 +2,7 @@ import React from 'react';
import { opencodeClient, type RoutedOpencodeEvent } from '@/lib/opencode/client';
import { saveSessionCursor } from '@/lib/messageCursorPersistence';
import { useSessionStore } from '@/stores/useSessionStore';
import { useMessageStore } from '@/stores/messageStore';
import { getActiveSessionWindow } from '@/stores/types/sessionTypes';
import { useConfigStore } from '@/stores/useConfigStore';
import { useUIStore, type EventStreamStatus } from '@/stores/useUIStore';
@@ -16,6 +17,7 @@ import { useMcpStore } from '@/stores/useMcpStore';
import { useContextStore } from '@/stores/contextStore';
import { getRegisteredRuntimeAPIs } from '@/contexts/runtimeAPIRegistry';
import { isDesktopLocalOriginActive } from '@/lib/desktop';
import { triggerSessionStatusPoll } from '@/hooks/useServerSessionStatus';
interface EventData {
type: string;
@@ -392,8 +394,7 @@ export const useEventStream = () => {
[bootstrapState]
);
const sessionStatusLastRefreshAtRef = React.useRef<number>(0);
const sessionStatusRefreshInFlightRef = React.useRef<Promise<void> | null>(null);
const currentSessionIdRef = React.useRef<string | null>(currentSessionId);
const previousSessionIdRef = React.useRef<string | null>(null);
const previousSessionDirectoryRef = React.useRef<string | null>(null);
@@ -464,7 +465,6 @@ export const useEventStream = () => {
[applySessionMetadata, getWorktreeMetadata]
);
type SessionStatusPayload = {
type: 'idle' | 'busy' | 'retry';
attempt?: number;
@@ -483,6 +483,9 @@ export const useEventStream = () => {
const prevType = storeStatus?.type ?? 'idle';
const nextType = status?.type ?? 'idle';
// Note: needs_attention logic is now handled by the server
// Server maintains authoritative state based on view tracking and message events
if (prevType !== nextType) {
try {
console.info('[SESSION-STATUS]', {
@@ -550,125 +553,18 @@ export const useEventStream = () => {
const next = new Map(useSessionStore.getState().sessionStatus ?? new Map());
if (nextType === 'idle') {
next.delete(sessionId);
next.set(sessionId, { ...status, confirmedAt: Date.now() });
} else {
next.set(sessionId, status);
const existing = next.get(sessionId);
if (existing?.confirmedAt) {
next.set(sessionId, { ...status, confirmedAt: existing.confirmedAt });
} else {
next.set(sessionId, status);
}
}
useSessionStore.setState({ sessionStatus: next });
}, []);
const refreshSessionStatus = React.useCallback(async () => {
const now = Date.now();
if (sessionStatusRefreshInFlightRef.current) {
return sessionStatusRefreshInFlightRef.current;
}
if (now - sessionStatusLastRefreshAtRef.current < 1500) {
return;
}
sessionStatusLastRefreshAtRef.current = now;
const applyStatusMap = (statusMap: Record<string, { type?: string }>) => {
const observed = new Set<string>();
// Use getState() to avoid sessions dependency which causes cascading updates
const currentSessions = useSessionStore.getState().sessions;
const knownSessionIds = new Set(currentSessions.map((session) => session.id));
for (const [sessionId, raw] of Object.entries(statusMap)) {
if (!sessionId || !raw) continue;
observed.add(sessionId);
const typeRaw = raw.type;
const status: SessionStatusPayload =
typeRaw === 'retry'
? {
type: 'retry',
attempt: (raw as { attempt?: unknown }).attempt as number | undefined,
message: (raw as { message?: unknown }).message as string | undefined,
next: (raw as { next?: unknown }).next as number | undefined,
}
: typeRaw === 'busy' || typeRaw === 'cooldown'
? { type: 'busy' }
: { type: 'idle' };
updateSessionStatus(sessionId, status, 'poll:/session/status');
}
// OpenCode's /session/status may omit idle sessions (returns only busy/retry).
// Treat missing entries as idle to avoid sessions getting stuck "working".
const currentStatuses = useSessionStore.getState().sessionStatus;
if (!currentStatuses) return;
for (const [sessionId, status] of currentStatuses.entries()) {
if (!knownSessionIds.has(sessionId)) continue;
if ((status.type === 'busy' || status.type === 'retry') && !observed.has(sessionId)) {
updateSessionStatus(sessionId, { type: 'idle' }, 'poll:missing->idle');
}
}
};
const task = (async (): Promise<void> => {
try {
// OpenCode global session status (busy/retry only; idle omitted)
const globalStatusMap = await opencodeClient.getGlobalSessionStatus();
if (globalStatusMap && Object.keys(globalStatusMap).length > 0) {
applyStatusMap(globalStatusMap);
return;
}
const directories = new Set<string>();
// Use getState() to avoid sessions dependency which causes cascading updates
const currentSessions = useSessionStore.getState().sessions;
for (const session of currentSessions) {
const directory = resolveSessionDirectoryForStatus(session.id);
if (directory) directories.add(directory);
}
const effective = normalizeDirectory(effectiveDirectory ?? null);
if (effective) directories.add(effective);
const queries = Array.from(directories);
if (queries.length === 0) {
// Fall back to scoped status for whatever the OpenCode client currently tracks.
const scoped = await opencodeClient.getSessionStatus();
if (scoped) {
applyStatusMap(scoped);
}
return;
}
const results = await Promise.allSettled(
queries.map((directory) => opencodeClient.getSessionStatusForDirectory(directory))
);
const merged: Record<string, { type?: string }> = {};
for (const result of results) {
if (result.status !== 'fulfilled' || !result.value) continue;
Object.assign(merged, result.value);
}
if (Object.keys(merged).length === 0) {
const hasActiveStatuses = Array.from(useSessionStore.getState().sessionStatus?.values?.() ?? []).some(
(status) => status?.type === 'busy' || status?.type === 'retry'
);
if (hasActiveStatuses) {
const healthy = await opencodeClient.checkHealth().catch(() => false);
if (!healthy) {
return;
}
}
}
applyStatusMap(merged);
} catch {
// ignored
}
})().finally(() => {
sessionStatusRefreshInFlightRef.current = null;
});
sessionStatusRefreshInFlightRef.current = task;
return task;
}, [effectiveDirectory, normalizeDirectory, resolveSessionDirectoryForStatus, updateSessionStatus]);
React.useEffect(() => {
const nextSessionId = currentSessionId ?? null;
const prevSessionId = previousSessionIdRef.current;
@@ -677,13 +573,13 @@ export const useEventStream = () => {
if (prevSessionId && nextSessionId && prevSessionId !== nextSessionId) {
if (prevDirectory && nextDirectory && prevDirectory !== nextDirectory) {
void refreshSessionStatus();
// Removed: void refreshSessionStatus();
}
}
previousSessionIdRef.current = nextSessionId;
previousSessionDirectoryRef.current = nextDirectory;
}, [currentSessionId, refreshSessionStatus, resolveSessionDirectoryForStatus]);
}, [currentSessionId, resolveSessionDirectoryForStatus]);
const handleEvent = React.useCallback((event: EventData) => {
lastEventTimestampRef.current = Date.now();
@@ -766,6 +662,47 @@ export const useEventStream = () => {
}
break;
case 'openchamber:session-status':
{
const sessionId = typeof props.sessionId === 'string' ? props.sessionId : null;
const status = typeof props.status === 'string' ? props.status : null;
const needsAttention = typeof props.needsAttention === 'boolean' ? props.needsAttention : false;
const timestamp = typeof props.timestamp === 'number' ? props.timestamp : Date.now();
if (sessionId && status) {
// Update session status
if (status === 'busy') {
updateSessionStatus(sessionId, { type: 'busy' }, 'sse:openchamber:session-status');
} else if (status === 'retry') {
const metadata = (typeof props.metadata === 'object' && props.metadata !== null) ? props.metadata as Record<string, unknown> : {};
updateSessionStatus(sessionId, {
type: 'retry',
attempt: typeof metadata.attempt === 'number' ? metadata.attempt : undefined,
message: typeof metadata.message === 'string' ? metadata.message : undefined,
next: typeof metadata.next === 'number' ? metadata.next : undefined,
}, 'sse:openchamber:session-status');
} else {
updateSessionStatus(sessionId, { type: 'idle' }, 'sse:openchamber:session-status');
}
// Update attention state in the same update to ensure atomicity
const currentAttentionStates = useSessionStore.getState().sessionAttentionStates || new Map();
const newAttentionStates = new Map(currentAttentionStates);
const existing = newAttentionStates.get(sessionId);
newAttentionStates.set(sessionId, {
needsAttention,
lastStatusChangeAt: timestamp,
lastUserMessageAt: existing?.lastUserMessageAt ?? null,
status: status as 'idle' | 'busy' | 'retry',
isViewed: existing?.isViewed ?? false,
});
useSessionStore.setState({ sessionAttentionStates: newAttentionStates });
}
}
break;
case 'message.part.updated': {
const part = (typeof props.part === 'object' && props.part !== null) ? (props.part as Part) : null;
if (!part) break;
@@ -938,6 +875,18 @@ export const useEventStream = () => {
trackMessage(messageId, 'message_updated', { role: (messageExt as { role?: unknown }).role });
if ((messageExt as { role?: unknown }).role === 'user') {
// Update lastUserMessageAt in session memory state
const { sessionMemoryState } = useMessageStore.getState();
const currentMemory = sessionMemoryState.get(sessionId);
if (currentMemory) {
const newMemoryState = new Map(sessionMemoryState);
newMemoryState.set(sessionId, {
...currentMemory,
lastUserMessageAt: Date.now(),
});
useMessageStore.setState({ sessionMemoryState: newMemoryState });
}
const serverParts = (props as { parts?: unknown }).parts || (messageExt as { parts?: unknown }).parts;
const partsArray = Array.isArray(serverParts) ? (serverParts as Part[]) : [];
const existingUserMessage = getMessageFromStore(sessionId, messageId);
@@ -1269,6 +1218,8 @@ export const useEventStream = () => {
}
completeStreamingMessage(sessionId, messageId);
updateSessionStatus(sessionId, { type: 'idle' }, 'sse:message.updated.completed');
// Removed: void refreshSessionStatus();
const rawMessageSessionId = (message as { sessionID?: string }).sessionID;
const messageSessionId: string =
@@ -1345,6 +1296,9 @@ export const useEventStream = () => {
? (props.messageID as string)
: null;
if (sessionId) {
updateSessionStatus(sessionId, { type: 'idle' }, 'sse:session.abort');
}
if (sessionId && messageId) {
completeStreamingMessage(sessionId, messageId);
}
@@ -1533,11 +1487,12 @@ export const useEventStream = () => {
applySessionMetadata,
trackMessage,
reportMessage,
updateSessionStatus,
updateSession,
removeSessionFromStore,
bootstrapState,
effectiveDirectory
effectiveDirectory,
updateSessionStatus,
]);
const shouldHoldConnection = React.useCallback(() => {
@@ -1627,10 +1582,11 @@ export const useEventStream = () => {
lastEventTimestampRef.current = Date.now();
publishStatus('connected', null);
checkConnection();
triggerSessionStatusPoll();
// Always refresh session status on connect to detect any
// already-running sessions (e.g., started via CLI before UI opened)
void refreshSessionStatus();
// Removed: void refreshSessionStatus();
if (shouldRefresh) {
void bootstrapState('sse_reconnected');
@@ -1713,7 +1669,7 @@ export const useEventStream = () => {
requestSessionMetadataRefresh,
handleEvent,
effectiveDirectory,
refreshSessionStatus,
debugConnectionState,
bootstrapState
]);
@@ -1800,7 +1756,8 @@ export const useEventStream = () => {
requestSessionMetadataRefresh(sessionId);
}
void refreshSessionStatus();
// Removed: void refreshSessionStatus();
triggerSessionStatusPoll();
publishStatus('connecting', 'Resuming stream');
startStream({ resetAttempts: true });
}
@@ -1825,7 +1782,8 @@ export const useEventStream = () => {
requestSessionMetadataRefresh(sessionId);
scheduleSoftResync(sessionId, 'window_focus', getActiveSessionWindow());
}
void refreshSessionStatus();
// Removed: void refreshSessionStatus();
triggerSessionStatusPoll();
publishStatus('connecting', 'Resuming stream');
startStream({ resetAttempts: true });
@@ -1837,6 +1795,7 @@ export const useEventStream = () => {
onlineStatusRef.current = true;
maybeBootstrapIfStale('network_restored');
if (pendingResumeRef.current || !unsubscribeRef.current) {
triggerSessionStatusPoll();
publishStatus('connecting', 'Network restored');
startStream({ resetAttempts: true });
}
@@ -1865,7 +1824,8 @@ export const useEventStream = () => {
void scheduleSoftResync(sessionId, 'page_show', getActiveSessionWindow());
requestSessionMetadataRefresh(sessionId);
}
void refreshSessionStatus();
// Removed: void refreshSessionStatus();
triggerSessionStatusPoll();
startStream({ resetAttempts: true });
}
};
@@ -1899,7 +1859,7 @@ export const useEventStream = () => {
);
if (hasBusySessions) {
void refreshSessionStatus();
// Removed: void refreshSessionStatus();
}
if (now - lastEventTimestampRef.current > 45000) {
Promise.resolve().then(async () => {
@@ -1974,8 +1934,8 @@ export const useEventStream = () => {
scheduleReconnect,
loadMessages,
requestSessionMetadataRefresh,
updateSessionStatus,
refreshSessionStatus,
shouldHoldConnection,
loadSessions,
maybeBootstrapIfStale,
packages/ui/src/hooks/useServerSessionStatus.ts
+243
View File
@@ -0,0 +1,243 @@
import React from 'react';
import { useSessionStore } from '@/stores/useSessionStore';
interface SessionState {
status: 'idle' | 'busy' | 'retry';
lastUpdateAt: number;
metadata?: {
attempt?: number;
message?: string;
next?: number;
};
}
interface SessionAttentionState {
needsAttention: boolean;
lastUserMessageAt: number | null;
lastStatusChangeAt: number;
status: 'idle' | 'busy' | 'retry';
isViewed: boolean;
}
interface ServerSnapshotResponse {
statusSessions: Record<string, SessionState>;
attentionSessions: Record<string, SessionAttentionState>;
serverTime: number;
}
const IMMEDIATE_POLL_DELAY_MS = 500; // 500ms for immediate poll after notification
// Ref to be accessed from outside (e.g., useEventStream) for triggering immediate poll
let triggerImmediatePollRef: (() => void) | null = null;
// Global function to trigger immediate poll from outside React
export const triggerSessionStatusPoll = () => {
if (triggerImmediatePollRef) {
triggerImmediatePollRef();
}
};
/**
* Hook to synchronize session status and attention state from server.
*
* Architecture: server maintains authoritative state, client applies snapshots.
* SSE remains the primary transport; snapshots repair missed updates.
*/
export function useServerSessionStatus() {
const isSyncingRef = React.useRef(false);
const lastSyncAtRef = React.useRef(0);
const timeoutRef = React.useRef<NodeJS.Timeout | null>(null);
const fetchSessionStatus = React.useCallback(async (immediate = false) => {
const now = Date.now();
if (!immediate && now - lastSyncAtRef.current < 1000) {
return;
}
// Prevent concurrent syncs
if (isSyncingRef.current) {
return;
}
isSyncingRef.current = true;
lastSyncAtRef.current = now;
try {
const snapshotResponse = await fetch('/api/sessions/snapshot', {
method: 'GET',
headers: { Accept: 'application/json' },
});
if (!snapshotResponse.ok) {
console.warn('[useServerSessionStatus] Failed to fetch session snapshot:', snapshotResponse.status);
return;
}
const snapshotData: ServerSnapshotResponse = await snapshotResponse.json();
const statusSessions = snapshotData.statusSessions ?? {};
const attentionSessions = snapshotData.attentionSessions ?? {};
// Update the session store with server state
const currentStatuses = useSessionStore.getState().sessionStatus || new Map();
let newStatuses: Map<string, { type: 'idle' | 'busy' | 'retry'; confirmedAt?: number; attempt?: number; message?: string; next?: number }> | null = null;
const ensureStatusesMap = () => {
if (!newStatuses) {
newStatuses = new Map(currentStatuses);
}
return newStatuses;
};
for (const [sessionId, state] of Object.entries(statusSessions)) {
const existing = currentStatuses.get(sessionId);
const hasChanged =
!existing ||
existing.type !== state.status ||
existing.attempt !== state.metadata?.attempt ||
existing.message !== state.metadata?.message ||
existing.next !== state.metadata?.next ||
existing.confirmedAt !== state.lastUpdateAt;
// Only update if server state is different
if (hasChanged) {
ensureStatusesMap().set(sessionId, {
type: state.status,
confirmedAt: state.lastUpdateAt,
attempt: state.metadata?.attempt,
message: state.metadata?.message,
next: state.metadata?.next,
});
}
}
// Check for sessions that are no longer in server state (treat as idle)
for (const [sessionId, currentStatus] of (newStatuses ?? currentStatuses)) {
if ((currentStatus.type === 'busy' || currentStatus.type === 'retry') &&
!statusSessions[sessionId]) {
// Session was busy but not in server state anymore -> mark as idle
ensureStatusesMap().set(sessionId, {
type: 'idle',
confirmedAt: Date.now(),
});
}
}
// Update attention state from server
const currentAttentionStates = useSessionStore.getState().sessionAttentionStates || new Map();
let newAttentionStates: Map<string, SessionAttentionState> | null = null;
const ensureAttentionMap = () => {
if (!newAttentionStates) {
newAttentionStates = new Map(currentAttentionStates);
}
return newAttentionStates;
};
let attentionStatesChanged = false;
for (const [sessionId, attentionState] of Object.entries(attentionSessions)) {
const existing = currentAttentionStates.get(sessionId);
const serverState = attentionState as SessionAttentionState;
const hasChanged =
!existing ||
existing.needsAttention !== serverState.needsAttention ||
existing.lastUserMessageAt !== serverState.lastUserMessageAt ||
existing.lastStatusChangeAt !== serverState.lastStatusChangeAt ||
existing.status !== serverState.status ||
existing.isViewed !== serverState.isViewed;
if (hasChanged) {
ensureAttentionMap().set(sessionId, serverState);
attentionStatesChanged = true;
}
}
// Remove attention states for sessions that no longer exist
for (const sessionId of (newAttentionStates ?? currentAttentionStates).keys()) {
const inStatus = !!statusSessions[sessionId];
const inAttention = !!attentionSessions[sessionId];
if (!inStatus && !inAttention) {
ensureAttentionMap().delete(sessionId);
attentionStatesChanged = true;
}
}
// Only update store if something actually changed
const statusChanged = newStatuses !== null;
if (statusChanged || attentionStatesChanged) {
useSessionStore.setState({
...(statusChanged && newStatuses ? { sessionStatus: newStatuses } : {}),
...(attentionStatesChanged && newAttentionStates ? { sessionAttentionStates: newAttentionStates } : {}),
});
}
if (process.env.NODE_ENV === 'development') {
console.debug('[useServerSessionStatus] Updated session statuses from server:', {
statusCount: Object.keys(statusSessions).length,
attentionCount: Object.keys(attentionSessions).length,
serverTime: snapshotData.serverTime,
});
}
} catch (error) {
console.warn('[useServerSessionStatus] Error fetching session status:', error);
} finally {
isSyncingRef.current = false;
}
}, []);
// Initial snapshot sync on mount
React.useEffect(() => {
void fetchSessionStatus(true);
return () => {
if (timeoutRef.current) {
clearTimeout(timeoutRef.current);
}
};
}, [fetchSessionStatus]);
// Sync snapshot when tab becomes visible
React.useEffect(() => {
const handleVisibilityChange = () => {
if (document.visibilityState === 'visible') {
// Small delay to let the browser settle
timeoutRef.current = setTimeout(() => {
void fetchSessionStatus(true);
}, 100);
}
};
document.addEventListener('visibilitychange', handleVisibilityChange);
return () => {
document.removeEventListener('visibilitychange', handleVisibilityChange);
};
}, [fetchSessionStatus]);
// Function to trigger immediate snapshot sync from external modules
const triggerImmediatePoll = React.useCallback(() => {
// Clear any pending timeout
if (timeoutRef.current) {
clearTimeout(timeoutRef.current);
}
// Schedule immediate sync with small delay to batch rapid calls
timeoutRef.current = setTimeout(() => {
void fetchSessionStatus(true);
}, IMMEDIATE_POLL_DELAY_MS);
}, [fetchSessionStatus]);
// Update the ref for external access
React.useEffect(() => {
triggerImmediatePollRef = triggerImmediatePoll;
return () => {
triggerImmediatePollRef = null;
};
}, [triggerImmediatePoll]);
return {
fetchSessionStatus,
triggerImmediatePoll,
};
}
// Export ref accessor for external modules
export const getTriggerImmediatePoll = () => triggerImmediatePollRef;
export default useServerSessionStatus;