import { useRef, ReactNode, useCallback, useEffect } from "react"; import { Buffer } from "buffer"; import { AppState } from "react-native"; import { observeOpenWorkspaceAgentIds } from "@/stores/workspace-layout-store"; import { useQueryClient } from "@tanstack/react-query"; import { useTranslation } from "react-i18next"; import { useClientActivity } from "@/hooks/use-client-activity"; import { useAppVisible } from "@/hooks/use-app-visible"; import { startPushNotifications } from "@/push-notifications"; import { createSetAgentInitializing, refreshAgentInitializationTimeout, } from "@/hooks/use-agent-initialization"; import type { StreamItem } from "@/types/stream"; import { deriveAgentStreamTurnLiveness } from "@/timeline/session-stream-reducers"; import { planTimelineTailFetch } from "@/timeline/timeline-sync-plan"; import { requestTimelineReplacement } from "@/timeline/timeline-replacement"; import { type ViewedTimelineOwner } from "@/timeline/viewed-timeline-sync"; import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { parseServerInfoStatusPayload } from "@getpaseo/protocol/messages"; import { buildAgentAttentionNotificationPayload, type AgentAttentionReason, type AgentAttentionNotificationPayload, type NotificationPermissionRequest, } from "@getpaseo/protocol/agent-attention-notification"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import type { AgentPermissionResponse } from "@getpaseo/protocol/agent-types"; import { getHostRuntimeStore, useHostRuntimeIsConnected } from "@/runtime/host-runtime"; import { useVoiceAudioEngineOptional, useVoiceRuntimeOptional } from "@/contexts/voice-context"; import type { AudioPlaybackSource } from "@/audio"; import { selectAgentTimelineState, toDaemonServerInfo, useSessionStore, type SessionState, } from "@/stores/session-store"; import { sendOsNotification } from "@/utils/os-notifications"; import { getIsAppActivelyVisible, getIsAppVisible } from "@/utils/app-visibility"; import { getInitKey, getInitDeferred, createInitDeferred, rejectInitDeferred, } from "@/utils/agent-initialization"; import { derivePendingPermissionKey } from "@/utils/agent-snapshots"; import { useToast } from "@/contexts/toast-context"; import { toErrorMessage } from "@/utils/error-messages"; import { showProviderNoticeToast } from "@/utils/provider-notice-toast"; import { applyCheckoutStatusUpdateFromEvent } from "@/git/checkout-status-cache"; import { resyncProviderSubagents, useProviderSubagentStore } from "@/subagents/provider-store"; // Re-export types from session-store and draft-store for backward compatibility export type { DraftInput } from "@/stores/draft-store"; export type { Agent, ExplorerEntry, ExplorerFile, ExplorerEntryKind, ExplorerFileKind, ExplorerEncoding, AgentFileExplorerState, } from "@/stores/session-store"; type AudioOutputPayload = Extract["payload"]; interface BufferedAudioChunk { chunkIndex: number; audio: string; format: string; id: string; } function decodeBase64Chunk(base64: string): Uint8Array { return Buffer.from(base64, "base64"); } function buildAudioPlaybackSource(chunks: BufferedAudioChunk[]): AudioPlaybackSource { const decodedChunks = chunks.map((chunk) => decodeBase64Chunk(chunk.audio)); const totalSize = decodedChunks.reduce((sum, chunk) => sum + chunk.length, 0); const output = new Uint8Array(totalSize); let offset = 0; for (const chunk of decodedChunks) { output.set(chunk, offset); offset += chunk.length; } const format = chunks[0]?.format ?? "pcm"; let mimeType: string; if (format === "pcm") mimeType = "audio/pcm;rate=24000;bits=16"; else if (format === "mp3") mimeType = "audio/mpeg"; else mimeType = `audio/${format}`; const bytes = output.slice(); return { size: bytes.byteLength, type: mimeType, async arrayBuffer() { return bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength); }, }; } const findLatestAssistantMessageText = (items: StreamItem[]): string | null => { for (let i = items.length - 1; i >= 0; i -= 1) { const item = items[i]; if (item.kind === "assistant_message") { return item.text; } } return null; }; const getLatestPermissionRequest = ( session: SessionState | undefined, agentId: string, ): NotificationPermissionRequest | null => { if (!session) { return null; } let latest: NotificationPermissionRequest | null = null; for (const pending of session.pendingPermissions.values()) { if (pending.agentId === agentId) { latest = pending.request; } } if (latest) { return latest; } const agentPending = session.agents.get(agentId)?.pendingPermissions; if (agentPending && agentPending.length > 0) { return agentPending[agentPending.length - 1] as NotificationPermissionRequest; } return null; }; interface AgentAttentionNotificationInput { notification?: AgentAttentionNotificationPayload; reason: AgentAttentionReason; serverId: string; workspaceId: string | undefined; agentId: string; assistantMessage: string | null; permissionRequest: NotificationPermissionRequest | null; } function resolveAgentAttentionNotification( input: AgentAttentionNotificationInput, ): AgentAttentionNotificationPayload | null { if (input.notification) { return input.notification.data.workspaceId ? input.notification : null; } if (!input.workspaceId) { return null; } return buildAgentAttentionNotificationPayload({ reason: input.reason, serverId: input.serverId, workspaceId: input.workspaceId, agentId: input.agentId, assistantMessage: input.reason === "finished" ? input.assistantMessage : null, permissionRequest: input.reason === "permission" ? input.permissionRequest : null, }); } function notifyVoiceAbortFailure( data: Extract["payload"], notifyError: (message: string) => void, ): void { if (data.type === "error" && data.metadata?.voiceAbortFailed === true) { notifyError(data.content); } } interface SessionProviderSharedProps { children: ReactNode; serverId: string; } interface SessionProviderClientProps extends SessionProviderSharedProps { client: DaemonClient; } export type SessionProviderProps = SessionProviderClientProps; function SessionProviderWithClient({ children, serverId, client }: SessionProviderClientProps) { return ( {children} ); } // SessionProvider: Daemon client message handler that updates Zustand store export function SessionProvider(props: SessionProviderProps) { return ; } function SessionProviderInternal({ children, serverId, client }: SessionProviderClientProps) { const { t } = useTranslation(); const voiceRuntime = useVoiceRuntimeOptional(); const voiceAudioEngine = useVoiceAudioEngineOptional(); const queryClient = useQueryClient(); const isConnected = useHostRuntimeIsConnected(serverId); const toast = useToast(); // Zustand store actions const setIsPlayingAudio = useSessionStore((state) => state.setIsPlayingAudio); const setAgentStreamTail = useSessionStore((state) => state.setAgentStreamTail); const setAgentStreamHead = useSessionStore((state) => state.setAgentStreamHead); const clearAgentStreamHead = useSessionStore((state) => state.clearAgentStreamHead); const setInitializingAgents = useSessionStore((state) => state.setInitializingAgents); const setAgents = useSessionStore((state) => state.setAgents); const flushAgentLastActivity = useSessionStore((state) => state.flushAgentLastActivity); const setPendingPermissions = useSessionStore((state) => state.setPendingPermissions); const updateSessionServerInfo = useSessionStore((state) => state.updateSessionServerInfo); const setViewedTimelineSync = useSessionStore((state) => state.setViewedTimelineSync); // Track focused agent for heartbeat const focusedAgentId = useSessionStore( (state) => state.sessions[serverId]?.focusedAgentId ?? null, ); const focusedTerminalId = useSessionStore( (state) => state.sessions[serverId]?.focusedTerminalId ?? null, ); const _sessionStateTimeoutRef = useRef | null>(null); const attentionNotifiedRef = useRef>(new Map()); const appStateRef = useRef(AppState.currentState); const viewedTimelineSyncRef = useRef(null); const audioOutputBuffersRef = useRef>(new Map()); const activeAudioGroupsRef = useRef>(new Set()); const isAppVisible = useAppVisible(); useEffect(() => { const subscription = AppState.addEventListener("change", (nextState) => { appStateRef.current = nextState; }); return () => { subscription.remove(); }; }, []); useEffect(() => { viewedTimelineSyncRef.current?.setActive(isAppVisible); }, [isAppVisible]); // Client activity tracking (heartbeat, push token registration) useClientActivity({ client, focusedAgentId, focusedTerminalId, }); useEffect(() => startPushNotifications({ client, serverId }), [client, serverId]); const notifyAgentAttention = useCallback( (params: { agentId: string; reason: "finished" | "error" | "permission"; timestamp: string; notification?: AgentAttentionNotificationPayload; }) => { const appState = appStateRef.current; const session = useSessionStore.getState().sessions[serverId]; const attentionFocusedAgentId = session?.focusedAgentId ?? null; if (params.reason === "error") { return; } const isActivelyVisible = getIsAppActivelyVisible(appState); const isAwayFromAgent = !isActivelyVisible || attentionFocusedAgentId !== params.agentId; if (!isAwayFromAgent) { return; } const timestampMs = new Date(params.timestamp).getTime(); const lastNotified = attentionNotifiedRef.current.get(params.agentId); if (lastNotified && lastNotified >= timestampMs) { return; } attentionNotifiedRef.current.set(params.agentId, timestampMs); const head = session?.agentStreamHead.get(params.agentId) ?? []; const tail = session?.agentStreamTail.get(params.agentId) ?? []; const assistantMessage = findLatestAssistantMessageText(head) ?? findLatestAssistantMessageText(tail); const permissionRequest = getLatestPermissionRequest(session, params.agentId); const workspaceId = session?.agents?.get(params.agentId)?.workspaceId; const notification = resolveAgentAttentionNotification({ notification: params.notification, reason: params.reason, serverId, workspaceId, agentId: params.agentId, assistantMessage, permissionRequest, }); if (!notification) { return; } void sendOsNotification({ title: notification.title, body: notification.body, data: notification.data, }); }, [serverId], ); useEffect(() => { const unregister = voiceRuntime?.registerSession({ serverId, setVoiceMode: async (enabled, agentId) => { if (!client) { throw new Error(t("common.errors.daemonUnavailable")); } await client.setVoiceMode(enabled, agentId); }, sendVoiceAudioChunk: async (audioData, mimeType) => { if (!client) { throw new Error(t("common.errors.daemonUnavailable")); } await client.sendVoiceAudioChunk(audioData, mimeType); }, audioPlayed: async (chunkId) => { if (!client) { throw new Error(t("common.errors.daemonUnavailable")); } await client.audioPlayed(chunkId); }, abortRequest: async () => { if (!client) { throw new Error(t("common.errors.daemonUnavailable")); } await client.abortRequest(); }, setAssistantAudioPlaying: (isPlaying) => { setIsPlayingAudio(serverId, isPlaying); }, }); return () => unregister?.(); }, [client, serverId, setIsPlayingAudio, t, voiceRuntime]); useEffect(() => { voiceRuntime?.updateSessionConnection(serverId, isConnected); }, [isConnected, serverId, voiceRuntime]); // If the client drops mid-initialization, clear pending flags useEffect(() => { if (!isConnected) { flushAgentLastActivity(); setInitializingAgents(serverId, new Map()); } }, [flushAgentLastActivity, serverId, isConnected, setInitializingAgents]); useEffect(() => { const setAgentInitializing = createSetAgentInitializing(serverId, setInitializingAgents); const onStream = (message: SessionOutboundMessage) => { if (message.type !== "agent_stream") return; const { agentId, event, timestamp, seq, epoch } = message.payload; const parsedTimestamp = new Date(timestamp); const streamEvent = event; if ( event.type === "turn_started" || event.type === "turn_completed" || event.type === "turn_failed" || event.type === "turn_canceled" ) { voiceRuntime?.onTurnEvent(serverId, agentId, event.type); } const turnLiveness = deriveAgentStreamTurnLiveness([ { event: streamEvent, seq, epoch, timestamp: parsedTimestamp }, ]); if (turnLiveness.length > 0) { getHostRuntimeStore().applyAgentTurnLiveness(serverId, agentId, turnLiveness); } sync.enqueueStreamEvent(agentId, { event: streamEvent, seq, epoch, timestamp: parsedTimestamp, }); // NOTE: We don't update lastActivityAt on every stream event to prevent // cascading rerenders. The agent_update handler updates agent.lastActivityAt // on status changes, which is sufficient for sorting and display purposes. }; const onReplacement = (message: SessionOutboundMessage) => { if (message.type !== "agent.timeline.replacement") return; void requestTimelineReplacement( { fetchAgentTimeline: (agentId, request) => getHostRuntimeStore().fetchAgentTimeline(serverId, agentId, request), }, message.payload.agentId, ).catch((error: unknown) => { console.warn("[Session] timeline replacement refresh failed", { serverId, error }); }); }; const sync = getHostRuntimeStore().createViewedTimelineOwner(serverId, { observe: (agentIds) => { const observation = client.observeTimeline(agentIds); observation.subscribe({ snapshot: () => {}, update: (message) => { if (message.type === "agent_stream") onStream(message); else if (message.type === "agent.timeline.replacement") onReplacement(message); }, }); return observation; }, readCursor: (agentId) => { const timeline = selectAgentTimelineState( useSessionStore.getState().sessions[serverId], agentId, ); return timeline.status === "synced" && timeline.range ? { epoch: timeline.range.epoch, endSeq: timeline.range.endSeq } : undefined; }, fetchPage: async (agentId, request) => { const session = useSessionStore.getState().sessions[serverId]; const initKey = getInitKey(serverId, agentId); const shouldInitialize = selectAgentTimelineState(session, agentId).status !== "synced"; if (shouldInitialize) { if (!getInitDeferred(initKey)) { const deferred = createInitDeferred(initKey, request.direction ?? "tail"); void deferred.promise.catch(() => undefined); } refreshAgentInitializationTimeout({ key: initKey, agentId, setAgentInitializing, }); setAgentInitializing(agentId, true); } try { const page = await getHostRuntimeStore().fetchAgentTimeline(serverId, agentId, request); if (shouldInitialize && getInitDeferred(initKey)) { refreshAgentInitializationTimeout({ key: initKey, agentId, setAgentInitializing }); } return page; } catch (error) { if (shouldInitialize) { setAgentInitializing(agentId, false); rejectInitDeferred(initKey, error instanceof Error ? error : new Error(String(error))); } throw error; } }, fetchLatestTail: (agentId) => getHostRuntimeStore().fetchAgentTimeline(serverId, agentId, planTimelineTailFetch()), reportError: (error) => { console.warn("[Session] viewed timeline synchronization failed", { serverId, error }); }, schedule: (task, delayMs) => { const timeout = setTimeout(task, delayMs); return () => clearTimeout(timeout); }, }); viewedTimelineSyncRef.current = sync; setViewedTimelineSync(serverId, sync); sync.setActive(getIsAppVisible(appStateRef.current)); const stopObservingOpenChats = observeOpenWorkspaceAgentIds(serverId, (agentIds) => sync.replaceOpenTabAgentIds(agentIds), ); return () => { stopObservingOpenChats(); if (viewedTimelineSyncRef.current === sync) { viewedTimelineSyncRef.current = null; } setViewedTimelineSync(serverId, null); sync.dispose(); }; }, [client, serverId, setInitializingAgents, setViewedTimelineSync, voiceRuntime]); useEffect(() => { viewedTimelineSyncRef.current?.setConnected(isConnected); }, [isConnected]); // Daemon message handlers - directly update Zustand store useEffect(() => { const feeds = client.observeEvents( [ "agent_attention_required", "terminal_attention_required", "agent_permission_request", "agent_permission_resolved", "agent.provider_subagents.update", "checkout_status_update", "status.server_info", ], { notifications: true }, ); const onFeed = ( type: SessionOutboundMessage["type"], handler: (message: SessionOutboundMessage) => void, ) => feeds.subscribe({ snapshot: () => {}, update: (message) => { if (message.type === type) handler(message); }, }); const unsubAgentAttention = onFeed("agent_attention_required", (message) => { if (message.type !== "agent_attention_required") return; const notification = message.payload; if (notification.shouldNotify) { notifyAgentAttention(notification); } }); const unsubProviderSubagentUpdate = feeds.subscribe({ snapshot: () => { void resyncProviderSubagents(client, serverId); }, update: (message) => { if (message.type !== "agent.provider_subagents.update") return; useProviderSubagentStore.getState().applyUpdate(serverId, message.payload); }, }); const unsubCheckoutStatusUpdate = onFeed("checkout_status_update", (message) => { if (message.type !== "checkout_status_update") return; applyCheckoutStatusUpdateFromEvent({ queryClient, serverId, message }); }); const unsubStatus = onFeed("status", (message) => { if (message.type !== "status") return; const serverInfo = parseServerInfoStatusPayload(message.payload); if (serverInfo) { updateSessionServerInfo(serverId, toDaemonServerInfo(serverInfo)); return; } }); const unsubPermissionRequest = onFeed("agent_permission_request", (message) => { if (message.type !== "agent_permission_request") return; const { agentId, request } = message.payload; setPendingPermissions(serverId, (prev) => { const next = new Map(prev); const key = derivePendingPermissionKey(agentId, request); next.set(key, { key, agentId, request }); return next; }); }); const unsubPermissionResolved = onFeed("agent_permission_resolved", (message) => { if (message.type !== "agent_permission_resolved") return; const { requestId, agentId } = message.payload; setPendingPermissions(serverId, (prev) => { const next = new Map(prev); const derivedKey = `${agentId}:${requestId}`; if (!next.delete(derivedKey)) { for (const [key, pending] of next.entries()) { if (pending.agentId === agentId && pending.request.id === requestId) { next.delete(key); break; } } } return next; }); }); const unsubAudioOutput = client.on("audio_output", async (message) => { if (message.type !== "audio_output") return; if (!voiceAudioEngine) { return; } const payload: AudioOutputPayload = message.payload; if (payload.isVoiceMode && voiceRuntime) { voiceRuntime.handleAudioOutput(serverId, payload); return; } const playbackGroupId = payload.groupId ?? payload.id; const chunkIndex = payload.chunkIndex ?? 0; const isFinalChunk = payload.isLastChunk ?? true; if (!audioOutputBuffersRef.current.has(playbackGroupId)) { audioOutputBuffersRef.current.set(playbackGroupId, []); } const bufferedChunks = audioOutputBuffersRef.current.get(playbackGroupId)!; bufferedChunks.push({ chunkIndex, audio: payload.audio, format: payload.format, id: payload.id, }); activeAudioGroupsRef.current.add(playbackGroupId); setIsPlayingAudio(serverId, true); if (!isFinalChunk) { return; } bufferedChunks.sort((left, right) => left.chunkIndex - right.chunkIndex); const chunkIds = bufferedChunks.map((chunk) => chunk.id); const shouldPlay = !payload.isVoiceMode || (voiceRuntime?.shouldPlayVoiceAudio(serverId) ?? false); const audioBlob = buildAudioPlaybackSource(bufferedChunks); function logAudioPlayedError(error: unknown): void { console.warn("[Session] Failed to confirm audio playback:", error); } const confirmAudioPlayed = async () => { await Promise.all( chunkIds.map((chunkId) => client.audioPlayed(chunkId).catch(logAudioPlayedError)), ); }; let startedVoicePlayback = false; try { if (shouldPlay) { if (payload.isVoiceMode) { startedVoicePlayback = true; voiceRuntime?.onAssistantAudioStarted(serverId); } await voiceAudioEngine.play(audioBlob); } await confirmAudioPlayed(); } catch (error) { console.error("[Session] Audio playback error:", error); await confirmAudioPlayed(); } finally { audioOutputBuffersRef.current.delete(playbackGroupId); activeAudioGroupsRef.current.delete(playbackGroupId); setIsPlayingAudio(serverId, activeAudioGroupsRef.current.size > 0); if (startedVoicePlayback) { voiceRuntime?.onAssistantAudioFinished(serverId); } } }); const unsubActivity = client.on("activity_log", (message) => { if (message.type !== "activity_log") return; notifyVoiceAbortFailure(message.payload, toast.error); }); const unsubTranscription = client.on("transcription_result", (message) => { if (message.type !== "transcription_result") return; const transcriptText = message.payload.text.trim(); voiceRuntime?.onTranscriptionResult(serverId, transcriptText); }); const unsubVoiceInputState = client.on("voice_input_state", (message) => { if (message.type !== "voice_input_state") return; voiceRuntime?.onServerSpeechStateChanged(serverId, message.payload.isSpeaking); }); const unsubTerminalAttention = onFeed("terminal_attention_required", (message) => { if (message.type !== "terminal_attention_required") { return; } if (!message.payload.shouldNotify) { return; } void sendOsNotification({ title: message.payload.title, body: message.payload.body, // serverId + workspaceId + terminalId route a tap to the terminal tab; cwd is // carried as a fallback identifier when the daemon resolved no workspace. data: { serverId: message.payload.serverId ?? serverId, terminalId: message.payload.terminalId, cwd: message.payload.cwd, ...(message.payload.workspaceId ? { workspaceId: message.payload.workspaceId } : {}), }, }); }); return () => { void feeds .release() .catch((error) => console.warn("[Session] Failed to release feeds", error)); unsubProviderSubagentUpdate(); unsubAgentAttention(); unsubCheckoutStatusUpdate(); unsubStatus(); unsubPermissionRequest(); unsubPermissionResolved(); unsubAudioOutput(); unsubActivity(); unsubTranscription(); unsubVoiceInputState(); unsubTerminalAttention(); }; }, [ client, queryClient, serverId, setIsPlayingAudio, setAgentStreamTail, setAgentStreamHead, clearAgentStreamHead, setInitializingAgents, setAgents, setPendingPermissions, notifyAgentAttention, updateSessionServerInfo, toast, voiceRuntime, voiceAudioEngine, ]); const _cancelAgentRun = useCallback( (agentId: string) => { if (!client) { console.warn("[Session] cancelAgent skipped: daemon unavailable"); return; } void client.cancelAgent(agentId).catch((error) => { console.error("[Session] Failed to cancel agent:", error); }); }, [client], ); const _deleteAgent = useCallback( (agentId: string) => { if (!client) { console.warn("[Session] deleteAgent skipped: daemon unavailable"); return; } void client.deleteAgent(agentId).catch((error) => { console.error("[Session] Failed to delete agent:", error); }); }, [client], ); const _archiveAgent = useCallback( (agentId: string) => { if (!client) { console.warn("[Session] archiveAgent skipped: daemon unavailable"); return; } void client.archiveAgent(agentId).catch((error) => { console.error("[Session] Failed to archive agent:", error); }); }, [client], ); const _restartServer = useCallback( (reason?: string) => { if (!client) { console.warn("[Session] restartServer skipped: daemon unavailable"); return; } void client.restartServer(reason).catch((error) => { console.error("[Session] Failed to restart server:", error); }); }, [client], ); const _setAgentMode = useCallback( (agentId: string, modeId: string) => { if (!client) { console.warn("[Session] setAgentMode skipped: daemon unavailable"); return; } void client .setAgentMode(agentId, modeId) .then((notice) => showProviderNoticeToast(toast, notice)) .catch((error) => { console.error("[Session] Failed to set agent mode:", error); toast.error(toErrorMessage(error)); }); }, [client, toast], ); const _setAgentModel = useCallback( (agentId: string, modelId: string | null) => { if (!client) { console.warn("[Session] setAgentModel skipped: daemon unavailable"); return; } void client.setAgentModel(agentId, modelId).catch((error) => { console.error("[Session] Failed to set agent model:", error); toast.error(toErrorMessage(error)); }); }, [client, toast], ); const _setAgentThinkingOption = useCallback( (agentId: string, thinkingOptionId: string | null) => { if (!client) { console.warn("[Session] setAgentThinkingOption skipped: daemon unavailable"); return; } void client .setAgentThinkingOption(agentId, thinkingOptionId) .then((notice) => showProviderNoticeToast(toast, notice)) .catch((error) => { console.error("[Session] Failed to set agent thinking option:", error); toast.error(toErrorMessage(error)); }); }, [client, toast], ); const _respondToPermission = useCallback( (agentId: string, requestId: string, response: AgentPermissionResponse) => { if (!client) { console.warn("[Session] respondToPermission skipped: daemon unavailable"); return; } void client.respondToPermission(agentId, requestId, response).catch((error) => { console.error("[Session] Failed to respond to permission:", error); }); }, [client], ); return children; }