fix desktop notifications and chat scroll stability
This commit is contained in:
@@ -2068,16 +2068,23 @@ async fn spawn_local_server(app: &tauri::AppHandle) -> Result<String> {
|
||||
let app_handle = app.clone();
|
||||
tauri::async_runtime::spawn(async move {
|
||||
let mut rx = rx;
|
||||
let mut stdout_buffer = String::new();
|
||||
while let Some(event) = rx.recv().await {
|
||||
match event {
|
||||
CommandEvent::Stdout(bytes) => {
|
||||
let line = String::from_utf8_lossy(&bytes);
|
||||
if let Some(rest) = line.strip_prefix(SIDECAR_NOTIFY_PREFIX) {
|
||||
if let Ok(parsed) =
|
||||
serde_json::from_str::<SidecarNotifyPayload>(rest.trim())
|
||||
{
|
||||
maybe_show_sidecar_notification(&app_handle, parsed);
|
||||
stdout_buffer.push_str(&String::from_utf8_lossy(&bytes));
|
||||
|
||||
while let Some(newline_index) = stdout_buffer.find('\n') {
|
||||
let line = stdout_buffer[..newline_index].trim_end_matches('\r');
|
||||
if let Some(rest) = line.strip_prefix(SIDECAR_NOTIFY_PREFIX) {
|
||||
if let Ok(parsed) =
|
||||
serde_json::from_str::<SidecarNotifyPayload>(rest.trim())
|
||||
{
|
||||
maybe_show_sidecar_notification(&app_handle, parsed);
|
||||
}
|
||||
}
|
||||
|
||||
stdout_buffer.drain(..=newline_index);
|
||||
}
|
||||
}
|
||||
CommandEvent::Error(error) => {
|
||||
|
||||
@@ -175,6 +175,7 @@ function App({ apis }: AppProps) {
|
||||
const appReadyDispatchedRef = React.useRef(false);
|
||||
const embeddedSessionChat = React.useMemo<EmbeddedSessionChatConfig | null>(() => readEmbeddedSessionChatConfig(), []);
|
||||
const embeddedBackgroundWorkEnabled = !embeddedSessionChat || isEmbeddedVisible;
|
||||
const recentDesktopNotificationTagsRef = React.useRef<Map<string, number>>(new Map());
|
||||
|
||||
React.useEffect(() => {
|
||||
setStreamPerfEnabled(showMemoryDebug);
|
||||
@@ -371,6 +372,63 @@ function App({ apis }: AppProps) {
|
||||
};
|
||||
}, [embeddedSessionChat]);
|
||||
|
||||
React.useEffect(() => {
|
||||
if (embeddedSessionChat || !isDesktopRuntime || typeof window === 'undefined' || typeof EventSource === 'undefined') {
|
||||
return;
|
||||
}
|
||||
|
||||
const source = new EventSource('/api/notifications/stream');
|
||||
|
||||
const handleMessage = (event: MessageEvent<string>) => {
|
||||
type DesktopNotificationEvent = {
|
||||
type?: string;
|
||||
properties?: {
|
||||
title?: string;
|
||||
body?: string;
|
||||
tag?: string;
|
||||
};
|
||||
};
|
||||
|
||||
let payload: DesktopNotificationEvent;
|
||||
|
||||
try {
|
||||
payload = JSON.parse(event.data) as DesktopNotificationEvent;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
if (payload?.type !== 'openchamber:notification') {
|
||||
return;
|
||||
}
|
||||
|
||||
const tag = typeof payload.properties?.tag === 'string' ? payload.properties.tag : '';
|
||||
if (tag) {
|
||||
const now = Date.now();
|
||||
const lastSeenAt = recentDesktopNotificationTagsRef.current.get(tag) ?? 0;
|
||||
if (now - lastSeenAt < 5000) {
|
||||
return;
|
||||
}
|
||||
recentDesktopNotificationTagsRef.current.set(tag, now);
|
||||
}
|
||||
|
||||
void apis.notifications.notifyAgentCompletion({
|
||||
title: payload.properties?.title,
|
||||
body: payload.properties?.body,
|
||||
tag: tag || undefined,
|
||||
});
|
||||
};
|
||||
|
||||
source.addEventListener('message', handleMessage as EventListener);
|
||||
source.onerror = () => {
|
||||
// Let EventSource reconnect automatically.
|
||||
};
|
||||
|
||||
return () => {
|
||||
source.removeEventListener('message', handleMessage as EventListener);
|
||||
source.close();
|
||||
};
|
||||
}, [apis.notifications, embeddedSessionChat, isDesktopRuntime]);
|
||||
|
||||
React.useEffect(() => {
|
||||
if (!embeddedSessionChat?.directory || isVSCodeRuntime) {
|
||||
return;
|
||||
|
||||
@@ -631,6 +631,7 @@ export const ChatContainer: React.FC = () => {
|
||||
}, [currentSessionId, isDesktopExpandedInput, scrollRef]);
|
||||
|
||||
const hasHistoryMetadata = Boolean(historyMeta);
|
||||
const lastHydratedSessionRef = React.useRef<string | null>(null);
|
||||
|
||||
const isSessionHydrating =
|
||||
Boolean(currentSessionId)
|
||||
@@ -640,12 +641,15 @@ export const ChatContainer: React.FC = () => {
|
||||
if (!currentSessionId) return;
|
||||
if (hasSessionMessagesEntry && hasHistoryMetadata) return;
|
||||
|
||||
const isSessionSwitch = lastHydratedSessionRef.current !== currentSessionId;
|
||||
lastHydratedSessionRef.current = currentSessionId;
|
||||
|
||||
const load = async () => {
|
||||
await loadMessages(currentSessionId).finally(() => {
|
||||
const statusType = sessionStatusForCurrent.type ?? 'idle';
|
||||
const isActivePhase = statusType === 'busy' || statusType === 'retry';
|
||||
const hasHashTarget = typeof window !== 'undefined' && window.location.hash.length > 0;
|
||||
const shouldSkipScroll = (isActivePhase && isPinned) || hasHashTarget;
|
||||
const shouldSkipScroll = hasHashTarget || (isActivePhase && isPinned && !isSessionSwitch);
|
||||
|
||||
if (!shouldSkipScroll) {
|
||||
if (typeof window === 'undefined') {
|
||||
|
||||
@@ -1692,17 +1692,10 @@ const MessageList = React.forwardRef<MessageListHandle, MessageListProps>(({
|
||||
return false;
|
||||
}
|
||||
|
||||
const container = resolveScrollContainer();
|
||||
if (!container) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const virtualizerBehavior = behavior === 'smooth' ? 'smooth' : 'auto';
|
||||
historyVirtualizer.scrollToIndex(index, { align: 'start', behavior: virtualizerBehavior });
|
||||
const targetTop = Math.max(0, container.scrollTop - 50);
|
||||
container.scrollTo({ top: targetTop, behavior });
|
||||
return true;
|
||||
}, [historyEntries.length, historyVirtualizer, resolveScrollContainer, shouldVirtualizeHistory]);
|
||||
}, [historyEntries.length, historyVirtualizer, shouldVirtualizeHistory]);
|
||||
|
||||
const scrollMessageElementIntoView = React.useCallback((messageId: string, behavior: ScrollBehavior = 'auto') => {
|
||||
const container = resolveScrollContainer();
|
||||
|
||||
@@ -14,19 +14,17 @@ import {
|
||||
import type { TurnHistorySignals } from '../lib/turns/historySignals';
|
||||
import { getMemoryLimits, type SessionHistoryMeta } from '@/stores/types/sessionTypes';
|
||||
|
||||
const waitForFrames = async (count = 1): Promise<void> => {
|
||||
if (typeof window === 'undefined') {
|
||||
return;
|
||||
}
|
||||
for (let index = 0; index < count; index += 1) {
|
||||
await new Promise<void>((resolve) => {
|
||||
window.requestAnimationFrame(() => resolve());
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
type ViewportAnchor = { messageId: string; offsetTop: number };
|
||||
|
||||
type PendingScrollRequest = {
|
||||
sessionId: string;
|
||||
kind: 'turn' | 'message';
|
||||
id: string;
|
||||
behavior: ScrollBehavior;
|
||||
turnId: string | null;
|
||||
resolve: (value: boolean) => void;
|
||||
};
|
||||
|
||||
interface UseChatTimelineControllerOptions {
|
||||
sessionId: string | null;
|
||||
messages: ChatMessageEntry[];
|
||||
@@ -100,6 +98,8 @@ export const useChatTimelineController = ({
|
||||
const historyMetaRef = React.useRef<SessionHistoryMeta | null>(historyMeta);
|
||||
const previousTurnCountRef = React.useRef(turnWindowModel.turnCount);
|
||||
const initializedSessionRef = React.useRef<string | null>(null);
|
||||
const pendingRenderResolversRef = React.useRef<Array<() => void>>([]);
|
||||
const pendingScrollRequestRef = React.useRef<PendingScrollRequest | null>(null);
|
||||
|
||||
const historySignals = React.useMemo(() => {
|
||||
const defaultLimit = getMemoryLimits().HISTORICAL_MESSAGES;
|
||||
@@ -189,10 +189,78 @@ export const useChatTimelineController = ({
|
||||
previousTurnCountRef.current = nextTurnCount;
|
||||
}, [turnWindowModel.turnCount]);
|
||||
|
||||
const resolvePendingRenderWaiters = React.useCallback(() => {
|
||||
const resolvers = pendingRenderResolversRef.current;
|
||||
if (resolvers.length === 0) {
|
||||
return;
|
||||
}
|
||||
pendingRenderResolversRef.current = [];
|
||||
resolvers.forEach((resolve) => resolve());
|
||||
}, []);
|
||||
|
||||
const waitForNextRenderCommit = React.useCallback((): Promise<void> => {
|
||||
return new Promise<void>((resolve) => {
|
||||
pendingRenderResolversRef.current.push(resolve);
|
||||
});
|
||||
}, []);
|
||||
|
||||
const resolvePendingScrollRequest = React.useCallback((value: boolean) => {
|
||||
const pending = pendingScrollRequestRef.current;
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
pendingScrollRequestRef.current = null;
|
||||
pending.resolve(value);
|
||||
}, []);
|
||||
|
||||
const attemptPendingScrollRequest = React.useCallback(() => {
|
||||
const pending = pendingScrollRequestRef.current;
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (pending.sessionId !== sessionIdRef.current) {
|
||||
resolvePendingScrollRequest(false);
|
||||
return;
|
||||
}
|
||||
|
||||
const didScroll = pending.kind === 'turn'
|
||||
? (messageListRef.current?.scrollToTurnId(pending.id, { behavior: pending.behavior }) ?? false)
|
||||
: (messageListRef.current?.scrollToMessageId(pending.id, { behavior: pending.behavior }) ?? false);
|
||||
|
||||
if (didScroll) {
|
||||
if (pending.turnId) {
|
||||
setActiveTurnId(pending.turnId);
|
||||
}
|
||||
resolvePendingScrollRequest(true);
|
||||
return;
|
||||
}
|
||||
|
||||
const targetIndex = pending.kind === 'turn'
|
||||
? turnModelRef.current.turnIndexById.get(pending.id)
|
||||
: turnModelRef.current.messageToTurnIndex.get(pending.id);
|
||||
|
||||
if (typeof targetIndex === 'number' && targetIndex >= turnStartRef.current) {
|
||||
resolvePendingScrollRequest(false);
|
||||
}
|
||||
}, [messageListRef, resolvePendingScrollRequest]);
|
||||
|
||||
React.useEffect(() => {
|
||||
return () => {
|
||||
resolvePendingRenderWaiters();
|
||||
resolvePendingScrollRequest(false);
|
||||
};
|
||||
}, [resolvePendingRenderWaiters, resolvePendingScrollRequest]);
|
||||
|
||||
const renderedMessages = React.useMemo(() => {
|
||||
return windowMessagesByTurn(messages, turnWindowModel, turnStart);
|
||||
}, [messages, turnStart, turnWindowModel]);
|
||||
|
||||
React.useLayoutEffect(() => {
|
||||
resolvePendingRenderWaiters();
|
||||
attemptPendingScrollRequest();
|
||||
}, [attemptPendingScrollRequest, renderedMessages, resolvePendingRenderWaiters, turnStart]);
|
||||
|
||||
// --- Synchronous scroll compensation for load-more / reveal ---
|
||||
// fetchOlderHistory and revealBufferedTurns store a snapshot here
|
||||
// before triggering the state change. useLayoutEffect consumes it
|
||||
@@ -257,10 +325,10 @@ export const useChatTimelineController = ({
|
||||
return next > 0 ? next : 0;
|
||||
});
|
||||
|
||||
await waitForFrames(1);
|
||||
await waitForNextRenderCommit();
|
||||
setPendingRevealWork(false);
|
||||
return true;
|
||||
}, [captureViewportAnchor, scrollRef]);
|
||||
}, [captureViewportAnchor, scrollRef, waitForNextRenderCommit]);
|
||||
|
||||
const fetchOlderHistory = React.useCallback(async (input: {
|
||||
preserveViewport: boolean;
|
||||
@@ -344,26 +412,29 @@ export const useChatTimelineController = ({
|
||||
|
||||
if (turnIndex < turnStartRef.current) {
|
||||
setTurnStart(turnIndex);
|
||||
await waitForFrames(2);
|
||||
}
|
||||
|
||||
const didScroll = messageListRef.current?.scrollToTurnId(turnId, {
|
||||
behavior: options?.behavior,
|
||||
}) ?? false;
|
||||
const result = await new Promise<boolean>((resolve) => {
|
||||
pendingScrollRequestRef.current = {
|
||||
sessionId: sessionIdRef.current ?? sessionId ?? '',
|
||||
kind: 'turn',
|
||||
id: turnId,
|
||||
behavior: options?.behavior ?? 'auto',
|
||||
turnId,
|
||||
resolve,
|
||||
};
|
||||
attemptPendingScrollRequest();
|
||||
});
|
||||
|
||||
if (didScroll) {
|
||||
setActiveTurnId(turnId);
|
||||
if (result) {
|
||||
return true;
|
||||
}
|
||||
|
||||
await waitForFrames(2);
|
||||
return messageListRef.current?.scrollToTurnId(turnId, {
|
||||
behavior: options?.behavior,
|
||||
}) ?? false;
|
||||
return false;
|
||||
} finally {
|
||||
setPendingRevealWork(false);
|
||||
}
|
||||
}, [messageListRef, sessionId]);
|
||||
}, [attemptPendingScrollRequest, sessionId]);
|
||||
|
||||
const scrollToMessage = React.useCallback(async (
|
||||
messageId: string,
|
||||
@@ -389,28 +460,29 @@ export const useChatTimelineController = ({
|
||||
|
||||
if (turnIndex < turnStartRef.current) {
|
||||
setTurnStart(turnIndex);
|
||||
await waitForFrames(2);
|
||||
}
|
||||
|
||||
const didScroll = messageListRef.current?.scrollToMessageId(messageId, {
|
||||
behavior: options?.behavior,
|
||||
}) ?? false;
|
||||
const result = await new Promise<boolean>((resolve) => {
|
||||
pendingScrollRequestRef.current = {
|
||||
sessionId: sessionIdRef.current ?? sessionId ?? '',
|
||||
kind: 'message',
|
||||
id: messageId,
|
||||
behavior: options?.behavior ?? 'auto',
|
||||
turnId: turnId ?? null,
|
||||
resolve,
|
||||
};
|
||||
attemptPendingScrollRequest();
|
||||
});
|
||||
|
||||
if (didScroll) {
|
||||
if (turnId) {
|
||||
setActiveTurnId(turnId);
|
||||
}
|
||||
if (result) {
|
||||
return true;
|
||||
}
|
||||
|
||||
await waitForFrames(2);
|
||||
return messageListRef.current?.scrollToMessageId(messageId, {
|
||||
behavior: options?.behavior,
|
||||
}) ?? false;
|
||||
return false;
|
||||
} finally {
|
||||
setPendingRevealWork(false);
|
||||
}
|
||||
}, [messageListRef, sessionId]);
|
||||
}, [attemptPendingScrollRequest, sessionId]);
|
||||
|
||||
const resumeToBottom = React.useCallback(() => {
|
||||
const nextStart = getInitialTurnStart(turnModelRef.current.turnCount);
|
||||
|
||||
@@ -114,6 +114,7 @@ export const useChatScrollManager = ({
|
||||
const pinnedSyncRafRef = React.useRef<number | null>(null);
|
||||
const preferInstantPinRef = React.useRef(false);
|
||||
const autoFollowDuringWorkRef = React.useRef(false);
|
||||
const pendingSessionSwitchSnapRef = React.useRef(false);
|
||||
const viewportAnchorTimerRef = React.useRef<ReturnType<typeof setTimeout> | null>(null);
|
||||
const pendingViewportAnchorRef = React.useRef<{ sessionId: string; anchor: number } | null>(null);
|
||||
const lastViewportAnchorRef = React.useRef<{ sessionId: string; anchor: number } | null>(null);
|
||||
@@ -206,21 +207,35 @@ export const useChatScrollManager = ({
|
||||
pinnedSyncRafRef.current = null;
|
||||
updateScrollButtonVisibility();
|
||||
if (!isPinnedRef.current) {
|
||||
pendingSessionSwitchSnapRef.current = false;
|
||||
return;
|
||||
}
|
||||
|
||||
const distanceFromBottom = getDistanceFromBottom();
|
||||
if (sessionIsWorking) {
|
||||
if (distanceFromBottom <= 0.5) {
|
||||
if (pendingSessionSwitchSnapRef.current && distanceFromBottom > 0.5) {
|
||||
autoFollowDuringWorkRef.current = false;
|
||||
scrollToBottomInternal({ instant: true });
|
||||
pendingSessionSwitchSnapRef.current = false;
|
||||
preferInstantPinRef.current = false;
|
||||
return;
|
||||
}
|
||||
|
||||
if (distanceFromBottom <= 0.5) {
|
||||
if (!pendingSessionSwitchSnapRef.current) {
|
||||
preferInstantPinRef.current = false;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
scrollPinnedToBottom(distanceFromBottom);
|
||||
preferInstantPinRef.current = false;
|
||||
pendingSessionSwitchSnapRef.current = false;
|
||||
return;
|
||||
}
|
||||
|
||||
pendingSessionSwitchSnapRef.current = false;
|
||||
|
||||
if (distanceFromBottom <= getAutoFollowThreshold()) {
|
||||
preferInstantPinRef.current = false;
|
||||
return;
|
||||
@@ -480,6 +495,7 @@ export const useChatScrollManager = ({
|
||||
flushViewportAnchor();
|
||||
pendingViewportAnchorRef.current = null;
|
||||
autoFollowDuringWorkRef.current = false;
|
||||
pendingSessionSwitchSnapRef.current = true;
|
||||
|
||||
// Always start pinned at bottom on session switch
|
||||
preferInstantPinRef.current = true;
|
||||
@@ -497,6 +513,7 @@ export const useChatScrollManager = ({
|
||||
React.useEffect(() => {
|
||||
if (!sessionIsWorking) {
|
||||
autoFollowDuringWorkRef.current = false;
|
||||
pendingSessionSwitchSnapRef.current = false;
|
||||
}
|
||||
}, [sessionIsWorking]);
|
||||
|
||||
|
||||
@@ -407,7 +407,19 @@ const {
|
||||
|
||||
const ENV_SKIP_OPENCODE_START = process.env.OPENCODE_SKIP_START === 'true' ||
|
||||
process.env.OPENCHAMBER_SKIP_OPENCODE_START === 'true';
|
||||
const ENV_DESKTOP_NOTIFY = process.env.OPENCHAMBER_DESKTOP_NOTIFY === 'true';
|
||||
const ENV_DESKTOP_NOTIFY = (() => {
|
||||
if (process.env.OPENCHAMBER_DESKTOP_NOTIFY === 'true') {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (process.env.OPENCHAMBER_RUNTIME === 'desktop') {
|
||||
return true;
|
||||
}
|
||||
|
||||
const argv0 = typeof process.argv?.[0] === 'string' ? process.argv[0] : '';
|
||||
const argv1 = typeof process.argv?.[1] === 'string' ? process.argv[1] : '';
|
||||
return /openchamber-server/i.test(argv0) || /openchamber-server/i.test(argv1);
|
||||
})();
|
||||
const ENV_CONFIGURED_OPENCODE_WSL_DISTRO =
|
||||
typeof process.env.OPENCODE_WSL_DISTRO === 'string' && process.env.OPENCODE_WSL_DISTRO.trim().length > 0
|
||||
? process.env.OPENCODE_WSL_DISTRO.trim()
|
||||
@@ -758,6 +770,11 @@ const bootstrapOpenCodeAtStartup = async (...args) => {
|
||||
if (openCodeLifecycleState.openCodeProcess && !openCodeLifecycleState.isExternalOpenCode) {
|
||||
startHealthMonitoring();
|
||||
}
|
||||
if (ENV_DESKTOP_NOTIFY) {
|
||||
void ensureGlobalWatcherStarted().catch((error) => {
|
||||
console.warn(`Global event watcher startup failed: ${error?.message || error}`);
|
||||
});
|
||||
}
|
||||
};
|
||||
const killProcessOnPort = (...args) => openCodeLifecycleRuntime.killProcessOnPort(...args);
|
||||
|
||||
@@ -874,6 +891,7 @@ async function main(options = {}) {
|
||||
opencodeWslDistro: resolvedWslDistro || null,
|
||||
nodeBinaryResolved: resolvedNodeBinary || null,
|
||||
bunBinaryResolved: resolvedBunBinary || null,
|
||||
desktopNotifyEnabled: ENV_DESKTOP_NOTIFY,
|
||||
planModeExperimentalEnabled: PLAN_MODE_EXPERIMENT_ENABLED,
|
||||
}),
|
||||
uiPassword,
|
||||
@@ -891,6 +909,8 @@ async function main(options = {}) {
|
||||
removePushSubscription,
|
||||
updateUiVisibility,
|
||||
isUiVisible,
|
||||
getUiNotificationClients: () => uiNotificationClients,
|
||||
writeSseEvent,
|
||||
sessionRuntime,
|
||||
setPushInitialized,
|
||||
fs,
|
||||
|
||||
@@ -35,6 +35,8 @@ export const registerNotificationRoutes = (app, dependencies) => {
|
||||
removePushSubscription,
|
||||
updateUiVisibility,
|
||||
isUiVisible,
|
||||
getUiNotificationClients,
|
||||
writeSseEvent,
|
||||
getSessionActivitySnapshot,
|
||||
getSessionStateSnapshot,
|
||||
getSessionAttentionSnapshot,
|
||||
@@ -158,6 +160,35 @@ export const registerNotificationRoutes = (app, dependencies) => {
|
||||
});
|
||||
});
|
||||
|
||||
app.get('/api/notifications/stream', async (req, res) => {
|
||||
const uiToken = uiAuthController?.ensureSessionToken
|
||||
? await uiAuthController.ensureSessionToken(req, res)
|
||||
: getUiSessionTokenFromRequest(req);
|
||||
if (!uiToken) {
|
||||
return;
|
||||
}
|
||||
|
||||
res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
|
||||
res.setHeader('Cache-Control', 'no-cache, no-transform');
|
||||
res.setHeader('Connection', 'keep-alive');
|
||||
res.flushHeaders?.();
|
||||
|
||||
const clients = getUiNotificationClients();
|
||||
clients.add(res);
|
||||
|
||||
try {
|
||||
writeSseEvent(res, {
|
||||
type: 'openchamber:notification-stream-ready',
|
||||
properties: { uiToken },
|
||||
});
|
||||
} catch {
|
||||
}
|
||||
|
||||
req.on('close', () => {
|
||||
clients.delete(res);
|
||||
});
|
||||
});
|
||||
|
||||
app.get('/api/session-activity', (_req, res) => {
|
||||
void ensureSessionWatcher();
|
||||
res.json(getSessionActivitySnapshot());
|
||||
|
||||
@@ -33,6 +33,8 @@ export const createBootstrapRuntime = (dependencies) => {
|
||||
removePushSubscription,
|
||||
updateUiVisibility,
|
||||
isUiVisible,
|
||||
getUiNotificationClients,
|
||||
writeSseEvent,
|
||||
sessionRuntime,
|
||||
setPushInitialized,
|
||||
fs,
|
||||
@@ -84,6 +86,8 @@ export const createBootstrapRuntime = (dependencies) => {
|
||||
removePushSubscription,
|
||||
updateUiVisibility,
|
||||
isUiVisible,
|
||||
getUiNotificationClients,
|
||||
writeSseEvent,
|
||||
getSessionActivitySnapshot: sessionRuntime.getSessionActivitySnapshot,
|
||||
getSessionStateSnapshot: sessionRuntime.getSessionStateSnapshot,
|
||||
getSessionAttentionSnapshot: sessionRuntime.getSessionAttentionSnapshot,
|
||||
|
||||
@@ -1,14 +1,27 @@
|
||||
import { createOpencodeClient } from '@opencode-ai/sdk/v2';
|
||||
|
||||
export const createOpenCodeWatcherRuntime = (deps) => {
|
||||
const {
|
||||
waitForOpenCodePort,
|
||||
buildOpenCodeUrl,
|
||||
getOpenCodeAuthHeaders,
|
||||
parseSseDataPayload,
|
||||
onPayload,
|
||||
} = deps;
|
||||
|
||||
let abortController = null;
|
||||
|
||||
const unwrapGlobalEventPayload = (eventData) => {
|
||||
if (!eventData || typeof eventData !== 'object') {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (eventData.payload && typeof eventData.payload === 'object') {
|
||||
return eventData.payload;
|
||||
}
|
||||
|
||||
return eventData;
|
||||
};
|
||||
|
||||
const start = async () => {
|
||||
if (abortController) {
|
||||
return;
|
||||
@@ -23,62 +36,38 @@ export const createOpenCodeWatcherRuntime = (deps) => {
|
||||
const run = async () => {
|
||||
while (!signal.aborted) {
|
||||
attempt += 1;
|
||||
let upstream;
|
||||
let reader;
|
||||
try {
|
||||
const url = buildOpenCodeUrl('/global/event', '');
|
||||
upstream = await fetch(url, {
|
||||
headers: {
|
||||
Accept: 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
Connection: 'keep-alive',
|
||||
...getOpenCodeAuthHeaders(),
|
||||
},
|
||||
signal,
|
||||
const baseUrl = buildOpenCodeUrl('/', '').replace(/\/$/, '');
|
||||
const client = createOpencodeClient({
|
||||
baseUrl,
|
||||
headers: getOpenCodeAuthHeaders(),
|
||||
});
|
||||
|
||||
if (!upstream.ok || !upstream.body) {
|
||||
throw new Error(`bad status ${upstream.status}`);
|
||||
}
|
||||
const result = await client.global.event({
|
||||
signal,
|
||||
sseMaxRetryAttempts: 0,
|
||||
onSseEvent: (event) => {
|
||||
const payload = unwrapGlobalEventPayload(event.data);
|
||||
if (!payload || typeof payload !== 'object') {
|
||||
return;
|
||||
}
|
||||
onPayload(payload);
|
||||
},
|
||||
});
|
||||
|
||||
console.log('[PushWatcher] connected');
|
||||
|
||||
const decoder = new TextDecoder();
|
||||
reader = upstream.body.getReader();
|
||||
let buffer = '';
|
||||
|
||||
while (!signal.aborted) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done) {
|
||||
for await (const _ of result.stream) {
|
||||
void _;
|
||||
if (signal.aborted) {
|
||||
break;
|
||||
}
|
||||
|
||||
buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n');
|
||||
|
||||
let separatorIndex = buffer.indexOf('\n\n');
|
||||
while (separatorIndex !== -1) {
|
||||
const block = buffer.slice(0, separatorIndex);
|
||||
buffer = buffer.slice(separatorIndex + 2);
|
||||
separatorIndex = buffer.indexOf('\n\n');
|
||||
const payload = parseSseDataPayload(block);
|
||||
onPayload(payload);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
if (signal.aborted) {
|
||||
return;
|
||||
}
|
||||
console.warn('[PushWatcher] disconnected', error?.message ?? error);
|
||||
} finally {
|
||||
try {
|
||||
if (reader) {
|
||||
await reader.cancel();
|
||||
reader.releaseLock();
|
||||
} else if (upstream?.body && !upstream.body.locked) {
|
||||
await upstream.body.cancel();
|
||||
}
|
||||
} catch {
|
||||
}
|
||||
}
|
||||
|
||||
const backoffMs = Math.min(1000 * Math.pow(2, Math.min(attempt, 5)), 30000);
|
||||
|
||||
Reference in New Issue
Block a user