fix(desktop): sync sessions across remote clients

apps/desktop/src/app/contrib/active-transcript-refresh.test.ts

@@ -0,0 +1,130 @@
+import type { MutableRefObject } from 'react'
+import { describe, expect, it, vi } from 'vitest'
+
+import type { SessionMessage, SessionMessagesResponse } from '@/hermes'
+import type { SessionInfo } from '@/types/hermes'
+
+import type { ClientSessionState } from '../types'
+
+import { advanceTranscriptScope, refreshActiveTranscript } from './active-transcript-refresh'
+
+const ref = <T>(current: T): MutableRefObject<T> => ({ current })
+
+const idleState = (storedSessionId = 'stored-a'): ClientSessionState =>
+  ({
+    storedSessionId,
+    messages: [],
+    busy: false,
+    awaitingResponse: false,
+    streamId: null
+  }) as unknown as ClientSessionState
+
+const message = (content: string): SessionMessage => ({ content, role: 'user', timestamp: 1 })
+const response = (messages: SessionMessage[]): SessionMessagesResponse => ({ messages, session_id: 'stored-a' })
+const stored = (id = 'stored-a'): SessionInfo => ({ id, profile: 'san' }) as SessionInfo
+
+function harness() {
+  const activeSessionIdRef = ref<null | string>('runtime-a')
+  const busyRef = ref(false)
+  const generationRef = ref(0)
+  const selectedStoredSessionIdRef = ref<null | string>('stored-a')
+  const sessionStateByRuntimeIdRef = ref(new Map([['runtime-a', idleState()]]))
+  const signatureRef = ref(new Map<string, string>())
+  const findStoredSession = vi.fn(() => stored())
+
+  const updateSessionState = vi.fn(
+    (runtimeSessionId: string, updater: (state: ClientSessionState) => ClientSessionState) => {
+      const previous = sessionStateByRuntimeIdRef.current.get(runtimeSessionId)!
+      const next = updater(previous)
+      sessionStateByRuntimeIdRef.current.set(runtimeSessionId, next)
+
+      return next
+    }
+  )
+
+  return {
+    activeSessionIdRef,
+    busyRef,
+    findStoredSession,
+    generationRef,
+    selectedStoredSessionIdRef,
+    sessionStateByRuntimeIdRef,
+    signatureRef,
+    updateSessionState
+  }
+}
+
+describe('active transcript refresh', () => {
+  it('publishes a changed persisted transcript into its idle owning runtime', async () => {
+    const h = harness()
+
+    await refreshActiveTranscript({
+      ...h,
+      loadMessages: vi.fn().mockResolvedValue(response([message('remote')]))
+    })
+
+    expect(h.updateSessionState).toHaveBeenCalledTimes(1)
+    expect(h.sessionStateByRuntimeIdRef.current.get('runtime-a')?.messages).toHaveLength(1)
+    expect(h.signatureRef.current.size).toBe(1)
+  })
+
+  it('rejects an old A response after an A to B to A scope transition', async () => {
+    const h = harness()
+    const scopeKeyRef = ref('remote-a:san:stored-a:runtime-a')
+    let resolve: ((messages: SessionMessagesResponse) => void) | undefined
+
+    const loadMessages = vi.fn(
+      () =>
+        new Promise<SessionMessagesResponse>(res => {
+          resolve = res
+        })
+    )
+
+    const pending = refreshActiveTranscript({ ...h, loadMessages })
+
+    advanceTranscriptScope(
+      { generationRef: h.generationRef, scopeKeyRef },
+      'remote-a:san:stored-b:runtime-b'
+    )
+    advanceTranscriptScope(
+      { generationRef: h.generationRef, scopeKeyRef },
+      'remote-a:san:stored-a:runtime-a'
+    )
+    resolve?.(response([message('stale')]))
+    await pending
+
+    expect(h.updateSessionState).not.toHaveBeenCalled()
+    expect(h.signatureRef.current.size).toBe(0)
+  })
+
+  it('rejects a response when a local turn starts while the request is pending', async () => {
+    const h = harness()
+    let resolve: ((messages: SessionMessagesResponse) => void) | undefined
+
+    const loadMessages = vi.fn(
+      () =>
+        new Promise<SessionMessagesResponse>(res => {
+          resolve = res
+        })
+    )
+
+    const pending = refreshActiveTranscript({ ...h, loadMessages })
+    h.busyRef.current = true
+    h.sessionStateByRuntimeIdRef.current.set('runtime-a', { ...idleState(), busy: true })
+    resolve?.(response([message('stale')]))
+    await pending
+
+    expect(h.updateSessionState).not.toHaveBeenCalled()
+  })
+
+  it('keeps the runtime untouched when the persisted signature did not change', async () => {
+    const h = harness()
+    const latest = [message('same')]
+
+    await refreshActiveTranscript({ ...h, loadMessages: vi.fn().mockResolvedValue(response(latest)) })
+    h.updateSessionState.mockClear()
+    await refreshActiveTranscript({ ...h, loadMessages: vi.fn().mockResolvedValue(response(latest)) })
+
+    expect(h.updateSessionState).not.toHaveBeenCalled()
+  })
+})

apps/desktop/src/app/contrib/active-transcript-refresh.ts

@@ -0,0 +1,142 @@
+import type { MutableRefObject } from 'react'
+
+import { getSessionMessages } from '@/hermes'
+import { preserveLocalAssistantErrors, toChatMessages } from '@/lib/chat-messages'
+import { sessionMessagesSignature } from '@/lib/session-signatures'
+import type { SessionInfo } from '@/types/hermes'
+
+import type { ClientSessionState } from '../types'
+
+interface ActiveTranscriptRefreshRefs {
+  activeSessionIdRef: MutableRefObject<null | string>
+  busyRef: MutableRefObject<boolean>
+  generationRef: MutableRefObject<number>
+  selectedStoredSessionIdRef: MutableRefObject<null | string>
+  sessionStateByRuntimeIdRef: MutableRefObject<Map<string, ClientSessionState>>
+  signatureRef: MutableRefObject<Map<string, string>>
+}
+
+interface RefreshActiveTranscriptOptions extends ActiveTranscriptRefreshRefs {
+  findStoredSession: (storedSessionId: string) => SessionInfo | undefined
+  loadMessages?: typeof getSessionMessages
+  updateSessionState: (
+    runtimeSessionId: string,
+    updater: (state: ClientSessionState) => ClientSessionState,
+    storedSessionId?: null | string
+  ) => ClientSessionState
+}
+
+interface TranscriptScopeRefs {
+  generationRef: MutableRefObject<number>
+  scopeKeyRef: MutableRefObject<string>
+}
+
+export function advanceTranscriptScope({ generationRef, scopeKeyRef }: TranscriptScopeRefs, scopeKey: string): void {
+  if (scopeKeyRef.current === scopeKey) {
+    return
+  }
+
+  scopeKeyRef.current = scopeKey
+  generationRef.current += 1
+}
+
+function isIdleOwnedState(
+  state: ClientSessionState | undefined,
+  storedSessionId: string
+): state is ClientSessionState {
+  return Boolean(
+    state &&
+      state.storedSessionId === storedSessionId &&
+      !state.busy &&
+      !state.awaitingResponse &&
+      state.streamId == null
+  )
+}
+
+export async function refreshActiveTranscript({
+  activeSessionIdRef,
+  busyRef,
+  findStoredSession,
+  generationRef,
+  loadMessages = getSessionMessages,
+  selectedStoredSessionIdRef,
+  sessionStateByRuntimeIdRef,
+  signatureRef,
+  updateSessionState
+}: RefreshActiveTranscriptOptions): Promise<void> {
+  const storedSessionId = selectedStoredSessionIdRef.current
+  const runtimeSessionId = activeSessionIdRef.current
+  const generation = generationRef.current
+
+  if (!storedSessionId || !runtimeSessionId || busyRef.current) {
+    return
+  }
+
+  const initialState = sessionStateByRuntimeIdRef.current.get(runtimeSessionId)
+
+  if (!isIdleOwnedState(initialState, storedSessionId)) {
+    return
+  }
+
+  const stored = findStoredSession(storedSessionId)
+
+  if (!stored) {
+    return
+  }
+
+  try {
+    const latest = await loadMessages(storedSessionId, stored.profile)
+
+    if (
+      generationRef.current !== generation ||
+      selectedStoredSessionIdRef.current !== storedSessionId ||
+      activeSessionIdRef.current !== runtimeSessionId ||
+      busyRef.current
+    ) {
+      return
+    }
+
+    const currentState = sessionStateByRuntimeIdRef.current.get(runtimeSessionId)
+
+    if (!isIdleOwnedState(currentState, storedSessionId)) {
+      return
+    }
+
+    const signatureKey = `${stored.profile ?? 'default'}:${storedSessionId}`
+    const signature = sessionMessagesSignature(latest.messages)
+
+    if (signatureRef.current.get(signatureKey) === signature) {
+      return
+    }
+
+    const remoteMessages = toChatMessages(latest.messages)
+    const messages = preserveLocalAssistantErrors(remoteMessages, currentState.messages)
+    let applied = false
+
+    updateSessionState(
+      runtimeSessionId,
+      state => {
+        if (
+          generationRef.current !== generation ||
+          selectedStoredSessionIdRef.current !== storedSessionId ||
+          activeSessionIdRef.current !== runtimeSessionId ||
+          busyRef.current ||
+          !isIdleOwnedState(state, storedSessionId)
+        ) {
+          return state
+        }
+
+        applied = true
+
+        return { ...state, messages }
+      },
+      storedSessionId
+    )
+
+    if (applied) {
+      signatureRef.current.set(signatureKey, signature)
+    }
+  } catch {
+    // Non-fatal: the next poll or a manual resume can hydrate the transcript.
+  }
+}