diff --git a/apps/mobile/src/app/(app)/_layout.tsx b/apps/mobile/src/app/(app)/_layout.tsx index dcd2aab7d4..715eef1c8c 100644 --- a/apps/mobile/src/app/(app)/_layout.tsx +++ b/apps/mobile/src/app/(app)/_layout.tsx @@ -3,6 +3,7 @@ import { Stack } from 'expo-router'; import { UserWebConnectionProvider } from '@/components/agents/user-web-connection-provider'; import { KiloChatPresenceMount } from '@/components/kilo-chat/kilo-chat-presence-mount'; import { KiloChatProvider } from '@/components/kilo-chat/kilo-chat-provider'; +import { ActiveSessionsLiveSyncMount } from '@/lib/active-sessions-live-sync-mount'; import { useFormSheetDetents } from '@/lib/form-sheet'; import { useThemeColors } from '@/lib/hooks/use-theme-colors'; import { StoreKiloPassPurchaseProvider } from '@/lib/kilo-pass/use-store-kilo-pass-purchase'; @@ -13,6 +14,7 @@ export default function AppLayout() { return ( + diff --git a/apps/mobile/src/components/agents/active-now-section.test.ts b/apps/mobile/src/components/agents/active-now-section.test.ts new file mode 100644 index 0000000000..54f5e0d734 --- /dev/null +++ b/apps/mobile/src/components/agents/active-now-section.test.ts @@ -0,0 +1,263 @@ +import * as React from 'react'; +import { describe, expect, it, vi } from 'vitest'; + +import { type ActiveSession } from '@/lib/hooks/use-agent-sessions'; + +import { ACTIVE_NOW_TRAY_CAP, selectTrayWindow } from './active-now-window'; + +// Feature B (tray overflow) is a pure client-side presentation of already +// loaded rows. There is no fetch and therefore no retryable / non-retryable +// failure mode of its own — that rationale is recorded here per the +// feature-state matrix. Errors still come from the underlying active-session +// query and surface via the screen's existing inline "Couldn't refresh" line +// (unchanged by this slice). + +// Feature B (tray overflow) is a pure client-side presentation of already +// loaded rows. There is no fetch and therefore no retryable / non-retryable +// failure mode of its own — that rationale is recorded here per the +// feature-state matrix. Errors still come from the underlying active-session +// query and surface via the screen's existing inline "Couldn't refresh" line +// (unchanged by this slice). + +function makeActive(over: Partial = {}): ActiveSession { + return { + id: 'a1', + status: 'running', + title: 'test', + connectionId: 'c1', + ...over, + }; +} + +const PINNED_IDS = ['a', 'b', 'c', 'd', 'e']; + +function pinned(): ActiveSession[] { + return PINNED_IDS.map(id => makeActive({ id })); +} + +describe('ACTIVE_NOW_TRAY_CAP', () => { + it('is 3 (matches the design cap for the collapsed tray)', () => { + // Regression guard — the cap is referenced from the section component, + // the test, and the design doc; if it changes intentionally the design + // doc must be updated too. + expect(ACTIVE_NOW_TRAY_CAP).toBe(3); + }); +}); + +describe('selectTrayWindow — edge/empty (≤ cap)', () => { + it('returns everything visible with hiddenCount 0 when at the cap', () => { + const sessions = pinned().slice(0, 3); + const result = selectTrayWindow(sessions, false, ACTIVE_NOW_TRAY_CAP); + expect(result.visible).toEqual(sessions); + expect(result.hiddenCount).toBe(0); + }); + + it('returns everything visible when below the cap, regardless of expanded', () => { + const sessions = pinned().slice(0, 2); + const collapsed = selectTrayWindow(sessions, false, ACTIVE_NOW_TRAY_CAP); + const expanded = selectTrayWindow(sessions, true, ACTIVE_NOW_TRAY_CAP); + expect(collapsed.visible).toEqual(sessions); + expect(collapsed.hiddenCount).toBe(0); + expect(expanded.visible).toEqual(sessions); + expect(expanded.hiddenCount).toBe(0); + }); + + it('returns an empty window for an empty pinned set', () => { + const result = selectTrayWindow([], false, ACTIVE_NOW_TRAY_CAP); + expect(result.visible).toEqual([]); + expect(result.hiddenCount).toBe(0); + }); + + it('expanded-then-shrunk to ≤ cap still returns all visible with no hidden', () => { + // The "expanded but count drops to ≤ cap" case from the spec: + // the section stops rendering any button; the state may stay `true` + // harmlessly. The helper therefore must NOT clamp visible based on + // `expanded` once the set fits under the cap. + const shrunk = pinned().slice(0, 2); + const result = selectTrayWindow(shrunk, true, ACTIVE_NOW_TRAY_CAP); + expect(result.visible).toEqual(shrunk); + expect(result.hiddenCount).toBe(0); + }); +}); + +describe('selectTrayWindow — happy (> cap, collapsed)', () => { + it('shows the first cap rows in caller order when collapsed', () => { + const sessions = pinned(); + const result = selectTrayWindow(sessions, false, ACTIVE_NOW_TRAY_CAP); + expect(result.visible.map(s => s.id)).toEqual(['a', 'b', 'c']); + // Caller order is preserved — the helper does not re-sort, it just slices. + }); + + it('reports the correct hiddenCount for the `+N more` label', () => { + // 5 sessions total, cap 3 → 2 hidden + const sessions = pinned(); + const result = selectTrayWindow(sessions, false, ACTIVE_NOW_TRAY_CAP); + expect(result.hiddenCount).toBe(2); + }); + + it('uses the caller-supplied cap, not a hardcoded value', () => { + // 5 sessions, cap 4 → 1 hidden + const sessions = pinned(); + const result = selectTrayWindow(sessions, false, 4); + expect(result.visible.map(s => s.id)).toEqual(['a', 'b', 'c', 'd']); + expect(result.hiddenCount).toBe(1); + }); +}); + +describe('selectTrayWindow — happy (> cap, expanded)', () => { + it('returns all rows with hiddenCount > 0 when expanded (drives `Show less`)', () => { + // 5 sessions, cap 3 → 2 hidden → drives the `Show less` button + const sessions = pinned(); + const result = selectTrayWindow(sessions, true, ACTIVE_NOW_TRAY_CAP); + expect(result.visible).toEqual(sessions); + expect(result.hiddenCount).toBe(2); + }); + + it('preserves caller order when expanded', () => { + const shuffled = ['e', 'a', 'c', 'b', 'd'].map(id => makeActive({ id })); + const result = selectTrayWindow(shuffled, true, ACTIVE_NOW_TRAY_CAP); + expect(result.visible.map(s => s.id)).toEqual(['e', 'a', 'c', 'b', 'd']); + }); +}); + +describe('expander button presence/absence (driven by selectTrayWindow)', () => { + // The ActiveNowSection renders the expander button iff `hiddenCount > 0`. + // The cases below are the per-state matrix entries the section must + // produce, derived directly from the helper's contract. + + it('renders NO expander when at the cap', () => { + const result = selectTrayWindow(pinned().slice(0, 3), false, ACTIVE_NOW_TRAY_CAP); + expect(result.hiddenCount > 0).toBe(false); + }); + + it('renders NO expander when below the cap, even if expanded state lingers', () => { + const result = selectTrayWindow(pinned().slice(0, 1), true, ACTIVE_NOW_TRAY_CAP); + expect(result.hiddenCount > 0).toBe(false); + }); + + it('renders the `+N more` button when collapsed above the cap', () => { + const result = selectTrayWindow(pinned(), false, ACTIVE_NOW_TRAY_CAP); + expect(result.hiddenCount > 0).toBe(true); + expect(result.hiddenCount).toBe(2); + }); + + it('renders the `Show less` button when expanded above the cap', () => { + const result = selectTrayWindow(pinned(), true, ACTIVE_NOW_TRAY_CAP); + expect(result.hiddenCount > 0).toBe(true); + expect(result.hiddenCount).toBe(2); + }); + + it('renders NO expander when expanded-then-shrunk to ≤ cap', () => { + const result = selectTrayWindow(pinned().slice(0, 2), true, ACTIVE_NOW_TRAY_CAP); + expect(result.hiddenCount > 0).toBe(false); + }); +}); + +// Smoke test for the reduced-motion path: the component must import and +// structurally reference the module under a `useReducedMotion() => true` +// mock without throwing, so the non-animated branch is reachable from +// tests. The deep assertion of behavior (no FadeIn/FadeOut applied) is +// covered on-device via the accessibility hierarchy; here we only verify +// the module is testable end-to-end with a mocked reanimated. +vi.mock('react-native', () => ({ + Pressable: 'Pressable', + View: 'View', +})); + +vi.mock('react', async () => { + const actual = (await vi.importActual('react')) as typeof React; + return { + ...actual, + useState: vi.fn((initial: T) => [initial, () => undefined] as [T, () => void]), + }; +}); +vi.mock('react-native-reanimated', () => ({ + __esModule: true, + default: { View: 'Animated.View' }, + FadeIn: { duration: () => ({}) }, + FadeOut: { duration: () => ({}) }, + LinearTransition: { duration: () => ({}) }, + useReducedMotion: () => true, +})); +vi.mock('@/components/agents/session-row', () => ({ + RemoteSessionRow: 'RemoteSessionRow', +})); +vi.mock('@/components/agents/session-list-section-header', () => ({ + SessionListSectionHeader: 'SessionListSectionHeader', +})); +vi.mock('@/components/ui/text', () => ({ + Text: 'Text', +})); + +describe('ActiveNowSection import smoke (reduced-motion path)', () => { + it('loads the module under useReducedMotion() => true without throwing', async () => { + // Importing lazily after the mock is registered. We only need the + // module to be importable and the exported function to be defined. + const mod = await import('./active-now-section'); + expect(typeof mod.ActiveNowSection).toBe('function'); + }); + + it('suppresses LinearTransition on tray and expander under reduced motion', async () => { + // The module is mocked with useReducedMotion() => true. The tray + // container and expander wrapper must pass layout={undefined} so + // no layout transition animates under reduced motion. + const mod = await import('./active-now-section'); + const { ActiveNowSection } = mod; + + // Call the component as a function (not JSX) so we can inspect the + // returned tree under the useState mock without a renderer. + type SectionProps = { + pinned: ActiveSession[]; + organizationIdBySessionId: Map; + onSessionPress: () => void; + }; + const renderSection = ActiveNowSection as (props: SectionProps) => React.ReactElement; + const element = renderSection({ + pinned: pinned(), + organizationIdBySessionId: new Map(), + onSessionPress: () => undefined, + }); + + // Collect every element that carries a `layout` prop key. Under + // reduced motion every such layout must be undefined (the gate is + // `layout={reducedMotion ? undefined : LinearTransition}`). + const layoutElements = findElementsWithLayoutProp(element); + expect(layoutElements.length).toBeGreaterThan(0); + for (const view of layoutElements) { + const props = view.props as { layout?: unknown }; + expect(props.layout).toBeUndefined(); + } + }); +}); + +type ElementProps = { layout?: unknown; children?: unknown }; + +function findElementsWithLayoutProp(node: unknown): React.ReactElement[] { + const results: React.ReactElement[] = []; + function visit(current: unknown) { + if ( + current === null || + current === undefined || + typeof current === 'string' || + typeof current === 'number' || + typeof current === 'boolean' + ) { + return; + } + if (Array.isArray(current)) { + for (const child of current) { + visit(child); + } + return; + } + if (React.isValidElement(current)) { + const props = current.props as ElementProps; + if (Object.hasOwn(props, 'layout')) { + results.push(current); + } + visit(props.children); + } + } + visit(node); + return results; +} diff --git a/apps/mobile/src/components/agents/active-now-section.tsx b/apps/mobile/src/components/agents/active-now-section.tsx index 1ceedfa37d..0a3c399ced 100644 --- a/apps/mobile/src/components/agents/active-now-section.tsx +++ b/apps/mobile/src/components/agents/active-now-section.tsx @@ -1,7 +1,16 @@ -import Animated, { FadeIn, FadeOut } from 'react-native-reanimated'; +import { useState } from 'react'; +import { Pressable, View } from 'react-native'; +import Animated, { + FadeIn, + FadeOut, + LinearTransition, + useReducedMotion, +} from 'react-native-reanimated'; import { RemoteSessionRow } from '@/components/agents/session-row'; import { SessionListSectionHeader } from '@/components/agents/session-list-section-header'; +import { ACTIVE_NOW_TRAY_CAP, selectTrayWindow } from '@/components/agents/active-now-window'; +import { Text } from '@/components/ui/text'; import { type ActiveSession } from '@/lib/hooks/use-agent-sessions'; type ActiveNowSectionProps = { @@ -22,32 +31,96 @@ type ActiveNowSectionProps = { * screen's non-scrolling vertical layout so it animates in/out with * Reanimated `FadeIn`/`FadeOut` while the screen's `LinearTransition` * wrappers absorb the layout change without jumping the history list. + * + * The tray caps the visible rows at `ACTIVE_NOW_TRAY_CAP` while collapsed + * and exposes a `+N more` expander when more sessions are pinned. Expansion + * is local `useState` (no persistence — resets on unmount). */ export function ActiveNowSection({ pinned, organizationIdBySessionId, onSessionPress, }: Readonly) { + const [expanded, setExpanded] = useState(false); + const reducedMotion = useReducedMotion(); + if (pinned.length === 0) { return null; } + const { visible, hiddenCount } = selectTrayWindow(pinned, expanded, ACTIVE_NOW_TRAY_CAP); + return ( - {pinned.map(session => ( - ( + { onSessionPress(session.id, organizationIdBySessionId.get(session.id)); }} /> ))} + {hiddenCount > 0 && ( + + { + setExpanded(prev => !prev); + }} + accessibilityRole="button" + accessibilityLabel={ + expanded ? 'Show fewer active sessions' : `${hiddenCount} more active sessions` + } + hitSlop={8} + className="min-h-[44px] items-center justify-center px-[22px] py-2 active:opacity-70" + > + + {expanded ? 'Show less' : `+${hiddenCount} more`} + + + + )} + + ); +} + +/** + * Per-row wrapper that fades individual rows in/out while the tray is + * animating expansion. Under `useReducedMotion()` the entering/exiting + * row animations are suppressed AND the tray/expander layout transition + * is omitted so rows appear instantly; the expand/collapse still works + * functionally. + */ +function AnimatedRow({ + reducedMotion, + session, + onPress, +}: Readonly<{ + reducedMotion: boolean | null; + session: ActiveSession; + onPress: () => void; +}>) { + if (reducedMotion) { + return ( + + + + ); + } + return ( + + ); } diff --git a/apps/mobile/src/components/agents/active-now-window.ts b/apps/mobile/src/components/agents/active-now-window.ts new file mode 100644 index 0000000000..77e8d52b6d --- /dev/null +++ b/apps/mobile/src/components/agents/active-now-window.ts @@ -0,0 +1,39 @@ +/** + * Cap on the number of pinned "Active now" rows shown while the tray is + * collapsed. Three keeps the tray (~260-280pt) visible on the smallest + * supported iPhone class (~667pt viewport) without pushing the history + * sections off-screen. + */ +export const ACTIVE_NOW_TRAY_CAP = 3; + +type TrayWindow = { + visible: T[]; + hiddenCount: number; +}; + +/** + * Pure selection helper for the pinned "Active now" tray window. + * + * Rules: + * - `pinned.length <= cap` → all rows visible, `hiddenCount` is 0, no + * expander button should be rendered. + * - collapsed → first `cap` rows (preserving caller order, which is the + * existing `selectPinnedActiveSessions` order), `hiddenCount` is + * `pinned.length - cap` and the expander button is `+N more`. + * - expanded → all rows visible, `hiddenCount` is `pinned.length - cap` + * (the caller uses that to render `Show less` whenever some rows are + * hidden). When the count drops to `≤ cap` the caller stops rendering + * any button; the `expanded` state may stay true harmlessly. + * + * No animation, no React, no data fetch — just slice/len arithmetic. + */ +export function selectTrayWindow(pinned: T[], expanded: boolean, cap: number): TrayWindow { + const hiddenCount = Math.max(0, pinned.length - cap); + if (pinned.length <= cap) { + return { visible: pinned, hiddenCount }; + } + if (expanded) { + return { visible: pinned, hiddenCount }; + } + return { visible: pinned.slice(0, cap), hiddenCount }; +} diff --git a/apps/mobile/src/components/agents/session-row-accessibility-label.test.ts b/apps/mobile/src/components/agents/session-row-accessibility-label.test.ts new file mode 100644 index 0000000000..c45f295d14 --- /dev/null +++ b/apps/mobile/src/components/agents/session-row-accessibility-label.test.ts @@ -0,0 +1,239 @@ +import { describe, expect, it } from 'vitest'; + +import { + formatSpokenTimeAgo, + sessionRowAccessibilityLabel, +} from './session-row-accessibility-label'; + +/** + * Feature C — a11y labels mirroring visible content. + * + * The three exclusive variants below cover the matrix in plan-k-tray-followups + * Item 3 / Feature C, matched one-to-one with `selectSessionRowEyebrowRight`. + * + * Other feature states (loading, empty, error, retryable/non-retryable + * failures, CTAs) are structurally n/a for these helpers: they are pure + * presentation functions for rows that already exist, and rows only exist + * when their data exists. Coverage for those states lives on the consuming + * screens (session-list-screen, session-list-content) and is unchanged by + * this slice. + */ + +describe('formatSpokenTimeAgo', () => { + it('expands "1m ago" to "1 minute ago" (singular minutes)', () => { + const t = new Date(Date.now() - 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('1 minute ago'); + }); + + it('expands "5m ago" to "5 minutes ago" (plural minutes)', () => { + const t = new Date(Date.now() - 5 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('5 minutes ago'); + }); + + it('expands "1h ago" to "1 hour ago" (singular hours)', () => { + const t = new Date(Date.now() - 60 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('1 hour ago'); + }); + + it('expands "3h ago" to "3 hours ago" (plural hours)', () => { + const t = new Date(Date.now() - 3 * 60 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('3 hours ago'); + }); + + it('expands "1d ago" to "1 day ago" (singular days)', () => { + const t = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('1 day ago'); + }); + + it('expands "3d ago" to "3 days ago" (plural days)', () => { + const t = new Date(Date.now() - 3 * 24 * 60 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('3 days ago'); + }); + + it('expands "1mo ago" to "1 month ago" (singular months)', () => { + // 35 days pushes past the 30-day boundary into the months branch + const t = new Date(Date.now() - 35 * 24 * 60 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('1 month ago'); + }); + + it('expands "2mo ago" to "2 months ago" (plural months)', () => { + const t = new Date(Date.now() - 65 * 24 * 60 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('2 months ago'); + }); + + it('expands "1y ago" to "1 year ago" (singular years)', () => { + // 370 days pushes past the 12-month boundary into the years branch + const t = new Date(Date.now() - 370 * 24 * 60 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('1 year ago'); + }); + + it('expands "2y ago" to "2 years ago" (plural years)', () => { + const t = new Date(Date.now() - 2 * 370 * 24 * 60 * 60 * 1000).toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('2 years ago'); + }); + + it('leaves "just now" unchanged', () => { + const t = new Date().toISOString(); + expect(formatSpokenTimeAgo(t)).toBe('just now'); + }); +}); + +describe('sessionRowAccessibilityLabel', () => { + describe('needs-input variant — StoredSessionRow (meta omitted)', () => { + it('produces "title, needs input, badge" with meta=null', () => { + // Stored row, needs-input eyebrow wins: meta is NOT rendered. + // The caller passes meta=null; the composer skips empty parts. + expect( + sessionRowAccessibilityLabel({ + title: 'Fix login bug', + needsInput: true, + badge: 'CLI', + meta: null, + }) + ).toBe('Fix login bug, needs input, CLI'); + }); + + it('includes meta when the caller passes it (caller is responsible for omitting)', () => { + // The composer is a simple joiner: it includes meta when passed. + // The row wires meta=null for needs-input rows (see session-row.tsx); + // if a future caller forgets, the composer does not silently strip + // meta. This test pins that contract so a "helpful" change here + // is a deliberate decision, not an accident. + expect( + sessionRowAccessibilityLabel({ + title: 'Fix login bug', + needsInput: true, + badge: 'VSCODE', + meta: '5 minutes ago', + }) + ).toBe('Fix login bug, needs input, VSCODE, 5 minutes ago'); + }); + }); + + describe('visible-meta variant — StoredSessionRow ("meta" eyebrow)', () => { + it('produces "title, badge, meta" with a spoken time', () => { + // Stored row, meta eyebrow renders the relative time. + expect( + sessionRowAccessibilityLabel({ + title: 'Fix login bug', + needsInput: false, + badge: 'CLI', + meta: '5 minutes ago', + }) + ).toBe('Fix login bug, CLI, 5 minutes ago'); + }); + + it('produces "title, badge, meta" with "just now"', () => { + expect( + sessionRowAccessibilityLabel({ + title: 'New session', + needsInput: false, + badge: 'CLOUD AGENT', + meta: 'just now', + }) + ).toBe('New session, CLOUD AGENT, just now'); + }); + + it('produces "title, badge, meta" with a days-ago spoken form', () => { + expect( + sessionRowAccessibilityLabel({ + title: 'Old session', + needsInput: false, + badge: 'SLACK', + meta: '3 days ago', + }) + ).toBe('Old session, SLACK, 3 days ago'); + }); + }); + + describe('needs-input variant — RemoteSessionRow (meta omitted)', () => { + it('produces "title, needs input, badge" with meta=null', () => { + // Remote row, needs-input eyebrow wins: live dot is hidden, meta is + // hidden. Badge is the left-eyebrow platform label or LIVE fallback. + expect( + sessionRowAccessibilityLabel({ + title: 'Live session', + needsInput: true, + badge: 'CLI', + meta: null, + }) + ).toBe('Live session, needs input, CLI'); + }); + + it('works with the LIVE fallback badge', () => { + expect( + sessionRowAccessibilityLabel({ + title: 'Live session', + needsInput: true, + badge: 'LIVE', + meta: null, + }) + ).toBe('Live session, needs input, LIVE'); + }); + }); + + describe('visible-meta variant — RemoteSessionRow ("live-and-meta" eyebrow)', () => { + it('produces "title, badge, meta" with a spoken time', () => { + // Remote row with updatedAt: live dot + meta render side-by-side. + expect( + sessionRowAccessibilityLabel({ + title: 'Live session', + needsInput: false, + badge: 'CLI', + meta: '2 hours ago', + }) + ).toBe('Live session, CLI, 2 hours ago'); + }); + }); + + describe('bare-live variant — RemoteSessionRow ("live" eyebrow, no meta)', () => { + it('produces "title, badge" with no meta', () => { + // Remote row: live dot only (no updatedAt to render as meta). + expect( + sessionRowAccessibilityLabel({ + title: 'Live session', + needsInput: false, + badge: 'LIVE', + meta: null, + }) + ).toBe('Live session, LIVE'); + }); + }); + + describe('order invariant (title → needs input → badge → meta)', () => { + it('preserves fixed order with all parts present', () => { + expect( + sessionRowAccessibilityLabel({ + title: 'A', + needsInput: true, + badge: 'B', + meta: 'C', + }) + ).toBe('A, needs input, B, C'); + }); + + it('preserves fixed order with needsInput false and no meta', () => { + expect( + sessionRowAccessibilityLabel({ + title: 'A', + needsInput: false, + badge: 'B', + meta: null, + }) + ).toBe('A, B'); + }); + + it('preserves fixed order with title and meta only (no badge, no needsInput)', () => { + // Defensive: badge is always visible per the brief, but if the caller + // passes an empty string the composer skips it. + expect( + sessionRowAccessibilityLabel({ + title: 'Orphan', + needsInput: false, + badge: '', + meta: '1 day ago', + }) + ).toBe('Orphan, 1 day ago'); + }); + }); +}); diff --git a/apps/mobile/src/components/agents/session-row-accessibility-label.ts b/apps/mobile/src/components/agents/session-row-accessibility-label.ts new file mode 100644 index 0000000000..55ac23d6b7 --- /dev/null +++ b/apps/mobile/src/components/agents/session-row-accessibility-label.ts @@ -0,0 +1,99 @@ +import { parseTimestamp, timeAgo } from '@/lib/utils'; + +/** + * Speech-friendly relative-time formatter. + * + * `timeAgo` (`@/lib/utils`) produces abbreviated strings like `"5m ago"`, + * `"1h ago"`, `"3d ago"`, `"1mo ago"`, `"2y ago"`, or `"just now"`. When + * the row renders those through `formatMeta` they are uppercased + * (`"5M AGO"`), which VoiceOver reads letter-by-letter. This helper + * expands every unit `timeAgo` emits into a form VoiceOver reads as words, + * with singular/plural handled, and leaves `"just now"` unchanged. + * + * Inputs that don't match a known unit pass through unchanged so a future + * `timeAgo` unit added without updating this helper doesn't get silently + * mangled — the worst case is the same letter-by-letter reading the + * uppercased form already has. + */ +export function formatSpokenTimeAgo(timestamp: string): string { + const raw = timeAgo(parseTimestamp(timestamp)); + const match = /^(\d+)([a-z]+)\s+ago$/.exec(raw); + if (!match) { + // "just now" or any future-unrecognized form + return raw; + } + const n = Number(match[1]); + const unit = match[2]; + const singular: Record = { + m: 'minute', + h: 'hour', + d: 'day', + mo: 'month', + y: 'year', + }; + const word = unit ? singular[unit] : undefined; + if (!word) { + // Unrecognized unit — pass through so a future `timeAgo` unit added + // without updating this helper doesn't get silently mangled. + return raw; + } + return `${n} ${n === 1 ? word : `${word}s`} ago`; +} + +type SessionRowAccessibilityLabelInputs = { + /** Row title, always present (e.g. "Untitled session" fallback). */ + title: string; + /** True when the row's right eyebrow renders the `NEEDS INPUT` state. */ + needsInput: boolean; + /** + * Left-eyebrow badge text, always visible (e.g. "CLI", "VSCODE", "LIVE", + * "CLOUD AGENT"). Pass an empty string only as a defensive fallback — + * the row always supplies a non-empty badge today. + */ + badge: string; + /** + * Right-eyebrow meta text in spoken form (typically + * `formatSpokenTimeAgo` output). Pass `null`/`undefined` when the row + * does NOT render meta (needs-input eyebrow, or bare-live eyebrow with + * no meta). + */ + meta?: string | null; +}; + +/** + * Compose the screen-reader label for a `SessionRow`, mirroring its visible + * content in the order the row renders parts: title, then `needs input` + * (only when the needs-input eyebrow is shown), then the always-visible + * left-eyebrow badge, then the meta text (only when the row visibly + * renders meta). Empty parts are skipped; the order is fixed. + * + * Three exclusive variants, aligned with `selectSessionRowEyebrowRight`: + * - **needs-input variant** (`needs-input` eyebrow): + * `", needs input, <badge>"` — meta is NOT rendered, so it is + * omitted. The left-eyebrow badge is always included. + * - **visible-meta variant** (`live-and-meta` / `meta` eyebrow): + * `"<title>, <badge>, <meta>"` — meta is included in spoken form. + * - **bare-live variant** (`live` eyebrow): + * `"<title>, <badge>"` — no meta text. + * + * Extends the `, needs input` pattern from PR #4605 (commit 635eaddc6) + * without replacing it. + */ +export function sessionRowAccessibilityLabel({ + title, + needsInput, + badge, + meta, +}: SessionRowAccessibilityLabelInputs): string { + const parts: string[] = [title]; + if (needsInput) { + parts.push('needs input'); + } + if (badge) { + parts.push(badge); + } + if (meta) { + parts.push(meta); + } + return parts.join(', '); +} diff --git a/apps/mobile/src/components/agents/session-row.tsx b/apps/mobile/src/components/agents/session-row.tsx index 0ed5e85f87..dbf09336a7 100644 --- a/apps/mobile/src/components/agents/session-row.tsx +++ b/apps/mobile/src/components/agents/session-row.tsx @@ -14,6 +14,10 @@ import { useSessionAttentionRevision, } from '@/lib/session-attention'; import { formatMeta, platformLabel, remoteAgentLabel, remoteMeta } from './session-list-helpers'; +import { + formatSpokenTimeAgo, + sessionRowAccessibilityLabel, +} from './session-row-accessibility-label'; type StoredSessionRowProps = { session: { @@ -85,6 +89,7 @@ export function StoredSessionRow({ const [renameVisible, setRenameVisible] = useState(false); const renameTextRef = useRef(title); const agentLabel = platformLabel(session.created_on_platform); + const timestamp = getAgentSessionTimestamp(session, sortBy); const revision = useSessionAttentionRevision(); const raiseId = session.status_updated_at ?? session.status ?? null; @@ -144,19 +149,30 @@ export function StoredSessionRow({ } }; + // Spoken meta mirrors the visible meta for the same inputs the row + // already uses to render `formatMeta(timestamp)`. When `needsInput` + // wins, the right eyebrow shows `NEEDS INPUT` and meta is NOT rendered, + // so the label omits it. + const spokenMeta = needsInput ? null : formatSpokenTimeAgo(timestamp); + return ( <> <Pressable onPress={onPress} onLongPress={handleLongPress} - accessibilityLabel={needsInput ? `${title}, needs input` : title} + accessibilityLabel={sessionRowAccessibilityLabel({ + title, + needsInput, + badge: agentLabel, + meta: spokenMeta, + })} className="active:opacity-70" > <SessionRow agentLabel={agentLabel} title={title} subtitle={session.git_branch} - meta={formatMeta(getAgentSessionTimestamp(session, sortBy))} + meta={formatMeta(timestamp)} needsInput={needsInput} stripMode="inline" className="pl-[22px] pr-[22px]" @@ -229,10 +245,29 @@ export function RemoteSessionRow({ session, onPress }: Readonly<RemoteSessionRow reconcileSessionAttention(session.id, session.status, null); }, [session.id, session.status, revision]); + // Spoken meta mirrors the visible meta the row renders. When `needsInput` + // wins, the right eyebrow shows `NEEDS INPUT` and meta is NOT rendered, + // so the label omits it. The remote row's `live` eyebrow (`live-and-meta`) + // renders the timestamp when `updatedAt` is present, otherwise the + // uppercased status. For speech we expand the timestamp via + // `formatSpokenTimeAgo` and lowercase/underscore-strip the status so + // VoiceOver doesn't read it letter-by-letter. + let spokenMeta: string | null = null; + if (!needsInput) { + spokenMeta = session.updatedAt + ? formatSpokenTimeAgo(session.updatedAt) + : session.status.toLowerCase().replaceAll('_', ' '); + } + return ( <Pressable onPress={onPress} - accessibilityLabel={needsInput ? `${title}, needs input` : title} + accessibilityLabel={sessionRowAccessibilityLabel({ + title, + needsInput, + badge: remoteAgentLabel(session.createdOnPlatform), + meta: spokenMeta, + })} className="active:opacity-70" > <SessionRow diff --git a/apps/mobile/src/lib/active-sessions-live-sync-mount.tsx b/apps/mobile/src/lib/active-sessions-live-sync-mount.tsx new file mode 100644 index 0000000000..d2b4239e24 --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync-mount.tsx @@ -0,0 +1,47 @@ +import { useEffect, useMemo, useRef } from 'react'; +import { type QueryFunction, useQueryClient } from '@tanstack/react-query'; + +import { useUserWebConnection } from '@/components/agents/user-web-connection-provider'; +import { ActiveSessionsLiveSync } from '@/lib/active-sessions-live-sync'; +import { type CachedActiveSessionsData } from '@/lib/active-sessions-live'; +import { useTRPC } from '@/lib/trpc'; + +/** + * React entry point for the active-sessions live-sync owner. Mounts an + * `ActiveSessionsLiveSync` instance exactly once per provider lifetime. + */ +function useActiveSessionsLiveSync(): void { + const connection = useUserWebConnection(); + const queryClient = useQueryClient(); + const trpc = useTRPC(); + const queryKey = useMemo(() => trpc.activeSessions.list.queryKey(), [trpc]); + // `trpc.activeSessions.list.queryOptions()` returns a fresh object on + // every call; we want a stable queryFn per provider lifetime so the + // live-sync owner can call it via fetchQuery. + const queryFn = useMemo( + () => + trpc.activeSessions.list.queryOptions().queryFn as QueryFunction<CachedActiveSessionsData>, + [trpc] + ); + + const syncRef = useRef<ActiveSessionsLiveSync | null>(null); + syncRef.current ??= new ActiveSessionsLiveSync({ + connection, + queryClient, + queryKey, + queryFn, + }); + + useEffect(() => { + const sync = syncRef.current; + return sync ? sync.attach() : undefined; + }, []); +} + +/** + * Component form for the layout wiring. Renders `null`. + */ +export function ActiveSessionsLiveSyncMount(): null { + useActiveSessionsLiveSync(); + return null; +} diff --git a/apps/mobile/src/lib/active-sessions-live-sync.departure.test.ts b/apps/mobile/src/lib/active-sessions-live-sync.departure.test.ts new file mode 100644 index 0000000000..6035a359dd --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync.departure.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from 'vitest'; + +import { + ActiveSessionsLiveSync, + makeCached, + makeConnection, + makeFakeQueryClient, + makeQueryFn, + QUERY_KEY, + setupTimers, + type SystemEvent, +} from '@/lib/active-sessions-live-sync.test-helpers'; + +setupTimers(); + +describe('ActiveSessionsLiveSync — departure refetch hook still works', () => { + // The departure-triggered stored refetch lives in use-agent-sessions + // and keys off the active-id set. This test simulates the data flow + // that hook observes: a heartbeat omits a previously-known id, the + // active-id set shrinks, and a refetch on the stored query is fired. + // (We use a side-effect callback to model the refetch; the live-sync + // owner is not responsible for triggering the stored refetch — it + // only owns the active cache.) + it('a heartbeat that omits an id shrinks the active set, enabling departure detection', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ + sessions: [ + makeCached({ id: 'a', connectionId: 'c1' }), + makeCached({ id: 'b', connectionId: 'c1' }), + ], + }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn: makeQueryFn(), + }); + sync.attach(); + const activeIdsBefore = new Set((qc.__getCached()?.sessions ?? []).map(s => s.id)); + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + const activeIdsAfter = new Set((qc.__getCached()?.sessions ?? []).map(s => s.id)); + // The use-agent-sessions departure check (id present before, absent now) + // would see `b` departed and fire stored.refetch(). We assert the + // contract the hook relies on: the set is strictly smaller. + expect(activeIdsBefore.has('b')).toBe(true); + expect(activeIdsAfter.has('b')).toBe(false); + }); +}); diff --git a/apps/mobile/src/lib/active-sessions-live-sync.enrichment.test.ts b/apps/mobile/src/lib/active-sessions-live-sync.enrichment.test.ts new file mode 100644 index 0000000000..0f23e78683 --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync.enrichment.test.ts @@ -0,0 +1,293 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + ActiveSessionsLiveSync, + makeCached, + makeConnection, + makeFakeQueryClient, + makeQueryFn, + QUERY_KEY, + setupNow, + setupTimers, + type SystemEvent, +} from '@/lib/active-sessions-live-sync.test-helpers'; + +setupTimers(); + +describe('ActiveSessionsLiveSync — enrichment retry policy', () => { + it('schedules enrichment when the cache has an unenriched live id', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a' })] }); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(sync.getPendingReasons().has('enrichment')).toBe(true); + expect(queryFn).toHaveBeenCalledTimes(1); + }); + + it('a failed enrichment retry is dropped once the id enriches or leaves', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a' })] }); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(sync.getPendingReasons().has('enrichment')).toBe(true); + // The enrichment fetch fails. The reason stays pending for now. + qc.__triggerFetchReject(new Error('network down')); + await sync.getFetchCompletion(); + expect(sync.getPendingReasons().has('enrichment')).toBe(true); + expect(queryFn).toHaveBeenCalledTimes(1); + // A later heartbeat no longer reports the unenriched id, so the + // stale enrichment reason is cleared and cannot drive retries. + const emptyHeartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { connectionId: 'c1', sessions: [] }, + }; + conn.__fireSystem(emptyHeartbeat); + await sync.getWriteQueue(); + expect(sync.getPendingReasons().has('enrichment')).toBe(false); + expect(queryFn).toHaveBeenCalledTimes(1); + }); + + it('does NOT schedule enrichment when every row is already enriched', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a', createdOnPlatform: 'cli' })] }); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(sync.getPendingReasons().has('enrichment')).toBe(false); + expect(queryFn).toHaveBeenCalledTimes(0); + }); + + it('rate-limits enrichment retries ≥10s apart', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a' })] }); + const queryFn = makeQueryFn(); + const { now, advance } = setupNow(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + now, + }); + sync.attach(); + // First enrichment attempt. + const heartbeat1: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat1); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + qc.__triggerFetchResolve({ sessions: [makeCached({ id: 'a' })] }); + await sync.getFetchCompletion(); + // Second heartbeat within < 10s of last completion → no new fetch. + const heartbeat2: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat2); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(1); + // Advance past 10s and try again → second fetch. + advance(10_001); + const heartbeat3: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat3); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(2); + }); + + it('rate-limits enrichment retries after a failed fetch', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a' })] }); + const queryFn = makeQueryFn(); + const { now, advance } = setupNow(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + now, + }); + sync.attach(); + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + qc.__triggerFetchReject(new Error('network down')); + await sync.getFetchCompletion(); + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await Promise.resolve(); + expect(queryFn).toHaveBeenCalledTimes(1); + + advance(10_000); + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(2); + }); + + it('DB row appears only after the first attempt: cache ends enriched', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a' })] }); + const queryFn = vi.fn(); + queryFn.mockResolvedValueOnce({ sessions: [makeCached({ id: 'a' })] }); + queryFn.mockResolvedValueOnce({ + sessions: [makeCached({ id: 'a', createdOnPlatform: 'cli' })], + }); + const { now, advance } = setupNow(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + now, + }); + sync.attach(); + const heartbeat1: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat1); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + qc.__triggerFetchResolve({ sessions: [makeCached({ id: 'a' })] }); + await sync.getFetchCompletion(); + expect(qc.__getCached()?.sessions[0]?.createdOnPlatform).toBeUndefined(); + // Advance past 10s; second heartbeat triggers a second enrichment. + advance(10_001); + const heartbeat2: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat2); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + qc.__triggerFetchResolve({ + sessions: [makeCached({ id: 'a', createdOnPlatform: 'cli' })], + }); + await sync.getFetchCompletion(); + expect(qc.__getCached()?.sessions[0]?.createdOnPlatform).toBe('cli'); + }); + + it('heartbeat-after-empty-snapshot repopulates the cache (enrichment preserved)', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ + sessions: [makeCached({ id: 'a', createdOnPlatform: 'cli' })], + }); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + // Empty snapshot arrives (degraded ingest path). + const emptySnapshot: SystemEvent = { + event: 'sessions.list', + data: { sessions: [] }, + }; + conn.__fireSystem(emptySnapshot); + await sync.getWriteQueue(); + expect(qc.__getCached()?.sessions).toEqual([]); + // Next heartbeat repopulates. + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + qc.__triggerFetchResolve({ + sessions: [makeCached({ id: 'a', createdOnPlatform: 'cli' })], + }); + await sync.getFetchCompletion(); + const cached = qc.__getCached(); + expect(cached?.sessions.map(s => s.id)).toEqual(['a']); + expect(cached?.sessions[0]?.createdOnPlatform).toBe('cli'); + }); +}); diff --git a/apps/mobile/src/lib/active-sessions-live-sync.pending.test.ts b/apps/mobile/src/lib/active-sessions-live-sync.pending.test.ts new file mode 100644 index 0000000000..21c24fb3dd --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync.pending.test.ts @@ -0,0 +1,226 @@ +import { describe, expect, it } from 'vitest'; + +import { + ActiveSessionsLiveSync, + makeCached, + makeConnection, + makeFakeQueryClient, + makeQueryFn, + QUERY_KEY, + setupNow, + setupTimers, + type SystemEvent, +} from '@/lib/active-sessions-live-sync.test-helpers'; + +setupTimers(); + +describe('ActiveSessionsLiveSync — pending-reason semantics', () => { + it('enrichment clears only when its own fetch completes', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + sync.scheduleRefresh('enrichment'); + await sync.getFetchQueue(); + expect(qc.__hasPendingFetch()).toBe(true); + qc.__triggerFetchResolve({ sessions: [] }); + await sync.getFetchCompletion(); + expect(sync.getPendingReasons().has('enrichment')).toBe(false); + }); + + it('cli-disconnected survives cancellation by a heartbeat', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + const disconnected: SystemEvent = { + event: 'cli.disconnected', + data: { connectionId: 'c1' }, + }; + conn.__fireSystem(disconnected); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(qc.__hasPendingFetch()).toBe(true); + expect(queryFn).toHaveBeenCalledTimes(1); + // A heartbeat arrives → cancelQueries cancels the in-flight fetch. + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c2', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + // The fetch was canceled; cli-disconnected's reason is still + // pending and a fresh fetch was kicked. + expect(sync.getPendingReasons().has('cli-disconnected')).toBe(true); + expect(qc.__hasPendingFetch()).toBe(true); + expect(queryFn).toHaveBeenCalledTimes(2); + // Resolve it; the reason clears. + qc.__triggerFetchResolve({ sessions: [] }); + await sync.getFetchCompletion(); + expect(sync.getPendingReasons().has('cli-disconnected')).toBe(false); + }); + + it('a fetch that errors does not clear its reason (transient failure)', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + sync.scheduleRefresh('cli-connected'); + await sync.getFetchQueue(); + expect(qc.__hasPendingFetch()).toBe(true); + qc.__triggerFetchReject(new Error('network down')); + await sync.getFetchCompletion(); + // The reason stays pending for the next scheduled trigger to retry. + expect(sync.getPendingReasons().has('cli-connected')).toBe(true); + // Genuine failures must NOT tight-loop; only a cancellation or a new + // reason can trigger an immediate re-kick. + expect(queryFn).toHaveBeenCalledTimes(1); + }); + + it('does not immediately retry a failed enrichment queued by a successful fetch', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const { now, advance } = setupNow(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + now, + }); + sync.attach(); + + sync.scheduleRefresh('cli-connected'); + await sync.getFetchQueue(); + qc.__triggerFetchResolve({ sessions: [makeCached({ id: 'a' })] }); + await sync.getFetchCompletion(); + await sync.getFetchQueue(); + + expect(sync.getPendingReasons()).toEqual(new Set(['enrichment'])); + expect(queryFn).toHaveBeenCalledTimes(2); + + qc.__triggerFetchReject(new Error('network down')); + await sync.getFetchCompletion(); + await Promise.resolve(); + await Promise.resolve(); + expect(queryFn).toHaveBeenCalledTimes(2); + + advance(10_001); + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(3); + }); + + it('does not schedule enrichment after a fully enriched cli-connected fetch', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + + sync.scheduleRefresh('cli-connected'); + await sync.getFetchQueue(); + qc.__triggerFetchResolve({ + sessions: [makeCached({ id: 'a', createdOnPlatform: 'cli' })], + }); + await sync.getFetchCompletion(); + + expect(sync.getPendingReasons()).toEqual(new Set()); + expect(queryFn).toHaveBeenCalledTimes(1); + }); + + it('a genuine fetch failure does not trigger an immediate retry (no tight loop)', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + sync.scheduleRefresh('reconnect'); + await sync.getFetchQueue(); + expect(qc.__hasPendingFetch()).toBe(true); + // Genuine failure (not a cancellation by newer work). + qc.__triggerFetchReject(new Error('network down')); + await sync.getFetchCompletion(); + expect(sync.getPendingReasons().has('reconnect')).toBe(true); + // No immediate re-fetch on genuine failure. + expect(queryFn).toHaveBeenCalledTimes(1); + // A later scheduled trigger retries the still-pending reason. + sync.scheduleRefresh('reconnect'); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(2); + }); + + it('reconnect survives a heartbeat-driven cancellation', async () => { + const conn = makeConnection(); + conn.__setConnected(false); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + conn.__fireConnection(true); + await sync.getFetchQueue(); + expect(qc.__hasPendingFetch()).toBe(true); + // Heartbeat cancels the reconnect fetch. + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + await sync.getFetchQueue(); + expect(sync.getPendingReasons().has('reconnect')).toBe(true); + expect(queryFn).toHaveBeenCalledTimes(2); + qc.__triggerFetchResolve({ sessions: [] }); + await sync.getFetchCompletion(); + expect(sync.getPendingReasons().has('reconnect')).toBe(false); + }); +}); diff --git a/apps/mobile/src/lib/active-sessions-live-sync.race.test.ts b/apps/mobile/src/lib/active-sessions-live-sync.race.test.ts new file mode 100644 index 0000000000..15cb867e52 --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync.race.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + ActiveSessionsLiveSync, + type CachedActiveSessionsData, + deferred, + makeCached, + makeConnection, + makeFakeQueryClient, + makeQueryFn, + QUERY_KEY, + setupTimers, + type SystemEvent, +} from '@/lib/active-sessions-live-sync.test-helpers'; + +setupTimers(); + +describe('ActiveSessionsLiveSync — race tests', () => { + it('heartbeat wins over an in-flight fetch (cache reflects heartbeat)', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a' })] }); + const queryFn = makeQueryFn({ + sessions: [makeCached({ id: 'a', title: 'from-network' })], + }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + // Start a refresh (e.g. reconnect) — fetch in flight. + sync.scheduleRefresh('reconnect'); + await sync.getFetchQueue(); + expect(qc.__hasPendingFetch()).toBe(true); + // Heartbeat arrives → cancelQueries cancels the in-flight fetch, + // then setQueryData writes the heartbeat data. + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'from-heartbeat' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + expect(qc.__getCached()?.sessions[0]?.title).toBe('from-heartbeat'); + // The original fetch was canceled; reconnect's reason is still + // pending (canceled, not completed) — a fresh fetch was kicked. + expect(sync.getPendingReasons().has('reconnect')).toBe(true); + }); + + it('removal vs late fetch: cli.disconnected wins, the late fetch is canceled', async () => { + const conn = makeConnection(); + conn.__setConnected(false); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a', connectionId: 'c1' })] }); + const queryFn = makeQueryFn({ + // Late result would have re-added the row, but it must not win. + sessions: [makeCached({ id: 'a', connectionId: 'c1', title: 'late' })], + }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + // Reconnect triggers a fetch. + conn.__fireConnection(true); + await sync.getFetchQueue(); + expect(qc.__hasPendingFetch()).toBe(true); + expect(queryFn).toHaveBeenCalledTimes(1); + // cli.disconnected for c1 — write removes c1 rows + schedules refresh. + const disconnected: SystemEvent = { + event: 'cli.disconnected', + data: { connectionId: 'c1' }, + }; + conn.__fireSystem(disconnected); + await sync.getWriteQueue(); + // Wait for the replacement fetch to start before asserting it exists. + await sync.getFetchQueue(); + // The write's cancelQueries canceled the original fetch; the + // scheduled refresh started a new one. The cache no longer has + // c1 rows. + expect(qc.__getCached()?.sessions).toEqual([]); + expect(qc.__hasPendingFetch()).toBe(true); + expect(queryFn).toHaveBeenCalledTimes(2); + }); + + it('stalled fetch never blocks a later WS write', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [] }); + // The queryFn returns a promise we never resolve. The cancelQueries + // call inside the write pipeline must reject it. + let stalled: ReturnType<typeof deferred<CachedActiveSessionsData>> | null = null; + const queryFn = vi.fn(async () => { + stalled = deferred<CachedActiveSessionsData>(); + const result = await stalled.promise; + return result; + }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + sync.scheduleRefresh('reconnect'); + await sync.getFetchQueue(); + expect(stalled).not.toBeNull(); + // A heartbeat arrives — its write must complete even though the + // fetch is stalled. + const heartbeat: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(heartbeat); + await sync.getWriteQueue(); + expect(qc.__getCached()?.sessions[0]?.title).toBe('A'); + }); +}); diff --git a/apps/mobile/src/lib/active-sessions-live-sync.reconnect.test.ts b/apps/mobile/src/lib/active-sessions-live-sync.reconnect.test.ts new file mode 100644 index 0000000000..6b46a36510 --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync.reconnect.test.ts @@ -0,0 +1,96 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + ActiveSessionsLiveSync, + makeConnection, + makeFakeQueryClient, + makeQueryFn, + QUERY_KEY, + setupTimers, +} from '@/lib/active-sessions-live-sync.test-helpers'; + +setupTimers(); + +describe('ActiveSessionsLiveSync — reconnect (onConnectionChange rising edge)', () => { + it('triggers exactly one refresh per false → true transition', async () => { + const conn = makeConnection(); + conn.__setConnected(false); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + conn.__fireConnection(true); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(1); + // Repeated true → still no new refresh. + conn.__fireConnection(true); + expect(queryFn).toHaveBeenCalledTimes(1); + // false transition does not trigger. + conn.__fireConnection(false); + expect(queryFn).toHaveBeenCalledTimes(1); + // Second rising edge → second refresh. + conn.__fireConnection(true); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(2); + }); + + it('release-then-retain still produces exactly one refresh per rising edge', async () => { + // release-then-retain via re-attach: the plan covers the case where + // a previous retain is fully released and a new retain starts the + // connection back up. The rising-edge detector must not fire on + // mount-with-already-connected state. + const release = vi.fn(); + const retain = vi.fn(() => release); + const conn = makeConnection({ retain }); + conn.__setConnected(false); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + const detach1 = sync.attach(); + detach1(); + expect(release).toHaveBeenCalledTimes(1); + // Re-attach while already connected → no initial refresh. + conn.__setConnected(true); + sync.attach(); + conn.__fireConnection(true); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(0); + // Cycle: false → true (rising edge) → exactly one refresh. + conn.__setConnected(false); + conn.__fireConnection(false); + conn.__fireConnection(true); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(1); + }); +}); + +describe('ActiveSessionsLiveSync — scheduled fetch invokes the queryFn', () => { + it('queryFn is invoked when scheduleRefresh runs', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn({ sessions: [] }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + sync.scheduleRefresh('reconnect'); + await sync.getFetchQueue(); + expect(queryFn).toHaveBeenCalledTimes(1); + qc.__triggerFetchResolve({ sessions: [] }); + await sync.getFetchCompletion(); + expect(queryFn).toHaveBeenCalledTimes(1); + }); +}); diff --git a/apps/mobile/src/lib/active-sessions-live-sync.test-helpers.ts b/apps/mobile/src/lib/active-sessions-live-sync.test-helpers.ts new file mode 100644 index 0000000000..742773d94b --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync.test-helpers.ts @@ -0,0 +1,236 @@ +import { afterEach, beforeEach, vi } from 'vitest'; +import { type QueryFunction, type QueryKey } from '@tanstack/react-query'; + +import { + ActiveSessionsLiveSync, + type LiveSyncConnection, + type LiveSyncQueryClient, +} from '@/lib/active-sessions-live-sync'; +import { + type CachedActiveSession, + type CachedActiveSessionsData, +} from '@/lib/active-sessions-live'; +import { type UserWebSystemEvent } from 'cloud-agent-sdk'; + +export type { CachedActiveSessionsData }; + +type SystemEvent = UserWebSystemEvent; +export type { SystemEvent }; + +type FakeConnection = LiveSyncConnection & { + __setConnected: (value: boolean) => void; + __fireSystem: (event: SystemEvent) => void; + __fireConnection: (value: boolean) => void; +}; + +export function makeConnection(over: Partial<LiveSyncConnection> = {}): FakeConnection { + const systemListeners = new Set<(event: SystemEvent) => void>(); + const connectionListeners = new Set<(connected: boolean) => void>(); + let connected = false; + const base: LiveSyncConnection = { + retain: vi.fn(() => () => undefined), + isConnected: vi.fn(() => connected), + onSystemEvent: vi.fn((listener: (event: SystemEvent) => void) => { + systemListeners.add(listener); + return () => { + systemListeners.delete(listener); + }; + }), + onConnectionChange: vi.fn((listener: (connected: boolean) => void) => { + connectionListeners.add(listener); + return () => { + connectionListeners.delete(listener); + }; + }), + ...over, + }; + return Object.assign(base, { + __setConnected(value: boolean) { + connected = value; + }, + __fireSystem(event: SystemEvent) { + for (const l of systemListeners) { + l(event); + } + }, + __fireConnection(value: boolean) { + connected = value; + for (const l of connectionListeners) { + l(value); + } + }, + }); +} + +type Deferred<T> = { + promise: Promise<T>; + resolve: (value: T) => void; + reject: (reason: Error) => void; +}; + +export function deferred<T>(): Deferred<T> { + // Standard deferred promise: the callbacks are supplied by the Promise + // executor and then exposed. Placeholders are initialized so TypeScript + // and the linter never see an uninitialized variable; they are replaced + // before any caller can invoke them. + const callbacks: { + resolve: (value: T) => void; + reject: (reason: Error) => void; + } = { + resolve: () => undefined, + reject: () => undefined, + }; + const promise = new Promise<T>((resolve, reject) => { + Object.assign(callbacks, { resolve, reject }); + }); + return { + promise, + resolve: value => { + callbacks.resolve(value); + }, + reject: reason => { + callbacks.reject(reason); + }, + }; +} + +type FakeQueryClient = LiveSyncQueryClient & { + fetchQueryCalls: number; + cancelQueriesCalls: number; + __setCached: (data: CachedActiveSessionsData | undefined) => void; + __triggerFetchResolve: (data: CachedActiveSessionsData) => void; + __triggerFetchReject: (error: Error) => void; + __hasPendingFetch: () => boolean; + __getCached: () => CachedActiveSessionsData | undefined; +}; + +const emptySessionsData = (): CachedActiveSessionsData => ({ sessions: [] }); + +export function makeFakeQueryClient( + initial: CachedActiveSessionsData = emptySessionsData() +): FakeQueryClient { + let cache: CachedActiveSessionsData | undefined = initial; + let pendingFetch: Deferred<CachedActiveSessionsData> | null = null; + let fetchQueryCalls = 0; + let cancelQueriesCalls = 0; + const qc = { + cancelQueries: vi.fn(async () => { + cancelQueriesCalls += 1; + if (pendingFetch) { + const d = pendingFetch; + pendingFetch = null; + d.reject(new Error('canceled')); + } + await Promise.resolve(); + }), + setQueryData: vi.fn((_key: QueryKey, updater: unknown): unknown => { + const next = + typeof updater === 'function' + ? (updater as (old: CachedActiveSessionsData | undefined) => unknown)(cache) + : updater; + cache = next as CachedActiveSessionsData; + return next; + }), + getQueryData: vi.fn((_key: QueryKey) => cache as unknown), + fetchQuery: vi.fn( + async (opts: { + queryKey: QueryKey; + queryFn: QueryFunction<CachedActiveSessionsData>; + staleTime?: number; + }): Promise<CachedActiveSessionsData> => { + fetchQueryCalls += 1; + // Drive the supplied queryFn so tests can assert it was invoked, + // but let the test control the resolved value via __triggerFetchResolve. + // QueryFunction requires a context arg; the fake only needs to observe the call. + await (opts.queryFn as unknown as () => Promise<unknown>)(); + const d = deferred<CachedActiveSessionsData>(); + pendingFetch = d; + try { + const result = await d.promise; + return result; + } finally { + if (pendingFetch === d) { + pendingFetch = null; + } + } + } + ), + fetchQueryCalls: 0, + cancelQueriesCalls: 0, + __setCached(data: CachedActiveSessionsData | undefined) { + cache = data; + }, + __triggerFetchResolve(data: CachedActiveSessionsData) { + if (pendingFetch) { + const d = pendingFetch; + pendingFetch = null; + cache = data; + d.resolve(data); + } + }, + __triggerFetchReject(error: Error) { + if (pendingFetch) { + const d = pendingFetch; + pendingFetch = null; + d.reject(error); + } + }, + __hasPendingFetch() { + return pendingFetch !== null; + }, + __getCached() { + return cache; + }, + }; + Object.defineProperty(qc, 'fetchQueryCalls', { + get() { + return fetchQueryCalls; + }, + }); + Object.defineProperty(qc, 'cancelQueriesCalls', { + get() { + return cancelQueriesCalls; + }, + }); + return qc as unknown as FakeQueryClient; +} + +export const QUERY_KEY = ['activeSessions', 'list'] as const; + +export function makeQueryFn(response: CachedActiveSessionsData = emptySessionsData()) { + return vi.fn(async () => { + const resolved = await Promise.resolve(response); + return resolved; + }); +} + +export function makeCached(over: Partial<CachedActiveSession> = {}): CachedActiveSession { + return { + id: 'a1', + status: 'running', + title: 'test', + connectionId: 'c1', + ...over, + }; +} + +export function setupNow() { + let now = 1_000_000; + return { + now: () => now, + advance: (ms: number) => { + now += ms; + }, + }; +} + +export function setupTimers(): void { + beforeEach(() => { + vi.useRealTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); +} + +export { ActiveSessionsLiveSync }; diff --git a/apps/mobile/src/lib/active-sessions-live-sync.test.ts b/apps/mobile/src/lib/active-sessions-live-sync.test.ts new file mode 100644 index 0000000000..b56def1e9c --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync.test.ts @@ -0,0 +1,310 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + ActiveSessionsLiveSync, + deferred, + makeCached, + makeConnection, + makeFakeQueryClient, + makeQueryFn, + QUERY_KEY, + setupTimers, + type SystemEvent, +} from '@/lib/active-sessions-live-sync.test-helpers'; + +setupTimers(); + +describe('ActiveSessionsLiveSync — attach / detach', () => { + it('retains the connection on attach and releases on detach', () => { + const release = vi.fn(); + const conn = makeConnection({ retain: vi.fn(() => release) }); + const qc = makeFakeQueryClient(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn: makeQueryFn(), + }); + const detach = sync.attach(); + expect(conn.retain).toHaveBeenCalledTimes(1); + expect(conn.onSystemEvent).toHaveBeenCalledTimes(1); + expect(conn.onConnectionChange).toHaveBeenCalledTimes(1); + detach(); + expect(release).toHaveBeenCalledTimes(1); + }); + + it('does not publish a queued write after detach', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn: makeQueryFn(), + }); + sync.attach(); + const blockedWrite = deferred<undefined>(); + vi.mocked(qc.cancelQueries).mockImplementationOnce(async () => { + await blockedWrite.promise; + }); + conn.__fireSystem({ + event: 'sessions.list', + data: { sessions: [{ id: 'first', status: 'running', title: 'First' }] }, + }); + conn.__fireSystem({ + event: 'sessions.list', + data: { sessions: [{ id: 'queued', status: 'running', title: 'Queued' }] }, + }); + sync.detach(); + blockedWrite.resolve(undefined); + await sync.getWriteQueue(); + expect(qc.setQueryData).not.toHaveBeenCalled(); + }); + + it('keeps queued work fenced after detach and re-attach', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ + sessions: [ + makeCached({ + id: 'current', + connectionId: 'c2', + createdAt: '2026-07-20T00:00:00.000Z', + }), + ], + }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn: makeQueryFn(), + }); + const detach = sync.attach(); + const blockedWrite = deferred<undefined>(); + vi.mocked(qc.cancelQueries).mockImplementationOnce(async () => { + await blockedWrite.promise; + }); + conn.__fireSystem({ + event: 'sessions.list', + data: { + sessions: [ + { + id: 'stale', + status: 'running', + title: 'Stale', + connectionId: 'c1', + createdAt: '2026-07-20T00:00:00.000Z', + }, + ], + }, + }); + await new Promise<void>(resolve => { + setTimeout(resolve, 0); + }); + expect(qc.cancelQueries).toHaveBeenCalledTimes(1); + + detach(); + sync.attach(); + conn.__fireSystem({ + event: 'sessions.heartbeat', + data: { + connectionId: 'c2', + sessions: [ + { + id: 'current', + status: 'running', + title: 'Current', + }, + ], + }, + }); + blockedWrite.resolve(undefined); + await sync.getWriteQueue(); + + expect(qc.setQueryData).toHaveBeenCalledTimes(1); + expect(qc.__getCached()?.sessions.map(session => session.id)).toEqual(['current']); + expect(sync.getPendingReasons()).toEqual(new Set()); + }); + + it('does not publish or re-kick an in-flight fetch after detach', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn: makeQueryFn(), + }); + sync.attach(); + sync.scheduleRefresh('reconnect'); + await sync.getFetchQueue(); + sync.scheduleRefresh('cli-connected'); + sync.detach(); + await sync.getFetchCompletion(); + await Promise.resolve(); + expect(sync.getPendingReasons()).toEqual(new Set()); + expect(qc.fetchQueryCalls).toBe(1); + expect(qc.setQueryData).not.toHaveBeenCalled(); + }); +}); + +describe('ActiveSessionsLiveSync — sessions.list', () => { + it('replaces the cache with the snapshot (root filter applied)', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'old' })] }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn: makeQueryFn(), + }); + sync.attach(); + const event: SystemEvent = { + event: 'sessions.list', + data: { + sessions: [ + { id: 'a', status: 'running', title: 'A', connectionId: 'c1' }, + { + id: 'child', + status: 'running', + title: 'Child', + connectionId: 'c1', + parentSessionId: 'a', + }, + ], + }, + }; + conn.__fireSystem(event); + await sync.getWriteQueue(); + expect(qc.__getCached()?.sessions.map(s => s.id)).toEqual(['a']); + }); +}); + +describe('ActiveSessionsLiveSync — sessions.heartbeat', () => { + it('merges a heartbeat into the cache', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'x', connectionId: 'c2' })] }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn: makeQueryFn(), + }); + sync.attach(); + const event: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(event); + await sync.getWriteQueue(); + expect(qc.__getCached()?.sessions.map(s => `${s.id}/${s.connectionId}`)).toEqual([ + 'a/c1', + 'x/c2', + ]); + }); + + it('a heartbeat that omits a previously-reported session drops it', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ + sessions: [makeCached({ id: 'a' }), makeCached({ id: 'b' })], + }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn: makeQueryFn(), + }); + sync.attach(); + const event: SystemEvent = { + event: 'sessions.heartbeat', + data: { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }, + }; + conn.__fireSystem(event); + await sync.getWriteQueue(); + expect(qc.__getCached()?.sessions.map(s => s.id)).toEqual(['a']); + }); +}); + +describe('ActiveSessionsLiveSync — cli.disconnected', () => { + it('drops the connection rows and schedules a refresh (queryFn actually invoked)', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ + sessions: [ + makeCached({ id: 'a', connectionId: 'c1' }), + makeCached({ id: 'b', connectionId: 'c2' }), + ], + }); + const queryFn = makeQueryFn({ sessions: [] }); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + const event: SystemEvent = { + event: 'cli.disconnected', + data: { connectionId: 'c1' }, + }; + conn.__fireSystem(event); + await sync.getWriteQueue(); + expect(qc.__getCached()?.sessions.map(s => s.id)).toEqual(['b']); + await sync.getFetchQueue(); + expect(qc.__hasPendingFetch()).toBe(true); + expect(queryFn).toHaveBeenCalledTimes(1); + }); +}); + +describe('ActiveSessionsLiveSync — cli.connected', () => { + it('schedules a refresh without a cache write', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + qc.__setCached({ sessions: [makeCached({ id: 'a' })] }); + const setQueryDataCalls = qc.setQueryData as ReturnType<typeof vi.fn>; + setQueryDataCalls.mockClear(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + const event: SystemEvent = { + event: 'cli.connected', + data: { connectionId: 'c1' }, + }; + conn.__fireSystem(event); + await sync.getFetchQueue(); + expect(setQueryDataCalls).not.toHaveBeenCalled(); + expect(queryFn).toHaveBeenCalledTimes(1); + }); + + it('ignores a malformed payload', async () => { + const conn = makeConnection(); + const qc = makeFakeQueryClient(); + const queryFn = makeQueryFn(); + const sync = new ActiveSessionsLiveSync({ + connection: conn, + queryClient: qc, + queryKey: QUERY_KEY, + queryFn, + }); + sync.attach(); + const event: SystemEvent = { event: 'cli.connected', data: { connectionId: 42 } }; + conn.__fireSystem(event); + await Promise.resolve(); + expect(sync.getPendingReasons()).toEqual(new Set()); + expect(queryFn).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/mobile/src/lib/active-sessions-live-sync.ts b/apps/mobile/src/lib/active-sessions-live-sync.ts new file mode 100644 index 0000000000..d235d40ca6 --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live-sync.ts @@ -0,0 +1,408 @@ +/** + * App-level owner for the active-sessions live-sync. The owner: + * + * - retains the shared `UserWebConnection` while mounted; + * - subscribes to `onSystemEvent` and applies the WS payloads to the + * shared `trpc.activeSessions.list` cache through ONE serialized + * pipeline (`cancelQueries` + `setQueryData`); the pipeline never + * awaits a network fetch, so a stalled tRPC refetch cannot block + * later heartbeats; + * - requests explicit refreshes (`cli.connected`, `cli.disconnected`, + * reconnect, enrichment) through a coalescing `scheduleRefresh` with + * durable per-reason pending state. Refreshes call + * `queryClient.fetchQuery({ queryKey, queryFn, staleTime: 0 })` so + * the network call is forced even after a preceding `setQueryData`. + * - observes the connection-state rising edge and triggers exactly + * one reconnect refresh per disconnect → connect transition. + * + * This module is framework-agnostic: it does not import React or the + * `UserWebConnectionProvider`. The thin React glue that wires it into the + * provider lives in `active-sessions-live-sync-mount.tsx`. + */ + +import { type QueryClient, type QueryFunction, type QueryKey } from '@tanstack/react-query'; + +import { + type CachedActiveSession, + type CachedActiveSessionsData, + hasUnenrichedLiveId, + mergeHeartbeatForActiveSessions, + mergeSnapshotForActiveSessions, + parseCliConnectionPayload, + parseHeartbeatPayload, + parseSessionsListPayload, + removeActiveSessionsForConnection, + selectRootWsSessions, +} from './active-sessions-live'; + +import { type UserWebConnection, type UserWebSystemEvent } from 'cloud-agent-sdk'; + +const ENRICHMENT_RETRY_MIN_INTERVAL_MS = 10_000; + +type RefreshReason = 'enrichment' | 'cli-connected' | 'cli-disconnected' | 'reconnect'; + +type SystemEvent = UserWebSystemEvent; + +type WriteUpdater = (current: CachedActiveSession[]) => CachedActiveSession[]; + +/** + * Minimal contract this owner needs from the SDK. Mirrors the public + * surface of `UserWebConnection`; the test double in + * `mobile-session-manager.test.ts` already conforms (see S2). + */ +export type LiveSyncConnection = Pick< + UserWebConnection, + 'retain' | 'isConnected' | 'onConnectionChange' | 'onSystemEvent' +>; + +export type LiveSyncQueryClient = Pick< + QueryClient, + 'cancelQueries' | 'setQueryData' | 'getQueryData' | 'fetchQuery' +>; + +type CreateLiveSyncOptions = { + connection: LiveSyncConnection; + queryClient: LiveSyncQueryClient; + queryKey: QueryKey; + queryFn: QueryFunction<CachedActiveSessionsData>; + now?: () => number; +}; + +/** + * The owner as a plain class. Exposed so the test suite can exercise + * the serialized pipeline, pending-reason state, and reconnect + * detection without a React renderer. + */ +export class ActiveSessionsLiveSync { + private readonly connection: LiveSyncConnection; + private readonly queryClient: LiveSyncQueryClient; + private readonly queryKey: QueryKey; + private readonly queryFn: QueryFunction<CachedActiveSessionsData>; + private readonly now: () => number; + + // eslint-disable-next-line promise/prefer-await-to-then + private writeQueue: Promise<void> = Promise.resolve(); + // eslint-disable-next-line promise/prefer-await-to-then + private fetchQueue: Promise<void> = Promise.resolve(); + + private fetchStartCount = 0; + private fetchStartWaiters: (() => void)[] = []; + private lastGetFetchQueueCount = 0; + private fetchCompletionWaiters: (() => void)[] = []; + + private readonly pendingReasons = new Set<RefreshReason>(); + private inFlightReasons: Set<RefreshReason> | null = null; + private isFetchInFlight = false; + private inFlightFetchCanceled = false; + private lastEnrichmentAttemptAt: number | null = null; + private lastConnectedState: boolean; + private attachmentEpoch = 0; + + private releaseRetain: (() => void) | null = null; + private systemListenerUnsubscribe: (() => void) | null = null; + private connectionListenerUnsubscribe: (() => void) | null = null; + + constructor(options: CreateLiveSyncOptions) { + this.connection = options.connection; + this.queryClient = options.queryClient; + this.queryKey = options.queryKey; + this.queryFn = options.queryFn; + this.now = options.now ?? (() => Date.now()); + this.lastConnectedState = this.connection.isConnected(); + } + + /** + * Subscribes to WS events, retains the connection, and tracks the + * initial connection state for the reconnect-rising-edge detector. + * Returns a detach function that releases all listeners and the + * retain. + */ + attach(): () => void { + if (this.releaseRetain) { + throw new Error('ActiveSessionsLiveSync already attached'); + } + // Re-attach while connected must not look like a rising edge. + this.attachmentEpoch += 1; + this.lastConnectedState = this.connection.isConnected(); + this.releaseRetain = this.connection.retain(); + this.systemListenerUnsubscribe = this.connection.onSystemEvent(event => { + this.handleSystemEvent(event); + }); + this.connectionListenerUnsubscribe = this.connection.onConnectionChange(connected => { + this.handleConnectionChange(connected); + }); + return () => { + this.detach(); + }; + } + + detach(): void { + this.attachmentEpoch += 1; + this.pendingReasons.clear(); + this.systemListenerUnsubscribe?.(); + this.connectionListenerUnsubscribe?.(); + this.releaseRetain?.(); + this.systemListenerUnsubscribe = null; + this.connectionListenerUnsubscribe = null; + this.releaseRetain = null; + void this.queryClient.cancelQueries({ queryKey: this.queryKey }); + } + + scheduleRefresh(reason: RefreshReason): void { + if (this.releaseRetain === null) { + return; + } + this.pendingReasons.add(reason); + this.kickFetch(); + } + + async getWriteQueue(): Promise<void> { + await this.writeQueue; + } + + async getFetchQueue(): Promise<void> { + if (this.pendingReasons.size === 0) { + return; + } + if (this.fetchStartCount > this.lastGetFetchQueueCount) { + this.lastGetFetchQueueCount = this.fetchStartCount; + return; + } + await new Promise<void>(resolve => { + this.fetchStartWaiters.push(resolve); + }); + this.lastGetFetchQueueCount = this.fetchStartCount; + } + + async getFetchCompletion(): Promise<void> { + if (!this.isFetchInFlight) { + return; + } + await new Promise<void>(resolve => { + this.fetchCompletionWaiters.push(resolve); + }); + } + + getPendingReasons(): Set<RefreshReason> { + return new Set(this.pendingReasons); + } + + private handleSystemEvent(event: SystemEvent): void { + if (event.event === 'sessions.list') { + const sessions = parseSessionsListPayload(event.data); + if (sessions) { + const roots = selectRootWsSessions(sessions); + this.enqueueWrite(current => mergeSnapshotForActiveSessions(current, roots)); + } + return; + } + if (event.event === 'sessions.heartbeat') { + const payload = parseHeartbeatPayload(event.data); + if (payload) { + const roots = selectRootWsSessions(payload.sessions); + this.enqueueWrite(current => + mergeHeartbeatForActiveSessions(current, { + connectionId: payload.connectionId, + sessions: roots, + }) + ); + } + return; + } + if (event.event === 'cli.disconnected') { + const payload = parseCliConnectionPayload(event.data); + if (payload) { + this.enqueueWrite(current => + removeActiveSessionsForConnection(current, payload.connectionId) + ); + this.scheduleRefresh('cli-disconnected'); + } + return; + } + if (event.event === 'cli.connected' && parseCliConnectionPayload(event.data)) { + this.scheduleRefresh('cli-connected'); + } + } + + private handleConnectionChange(connected: boolean): void { + // Rising-edge detector: a false → true transition triggers exactly + // one reconnect refresh. A true → false transition does not + // schedule a refresh (the conditional `refetchInterval` covers + // the WS-down window; re-scheduling here would either duplicate + // the work or be absorbed by the next refresh regardless). + if (!this.lastConnectedState && connected) { + this.scheduleRefresh('reconnect'); + } + this.lastConnectedState = connected; + } + + private enqueueWrite(updater: WriteUpdater): void { + if (this.releaseRetain === null) { + return; + } + const attachmentEpoch = this.attachmentEpoch; + // Serialize ALL cache writes (cancel + setQueryData) on one queue. + // The pipeline never awaits a network fetch — that is the + // cancel-based fencing model. + this.writeQueue = (async () => { + await this.writeQueue; + if (attachmentEpoch !== this.attachmentEpoch) { + return; + } + // A write always cancels the in-flight fetch so the new cache + // state can never be overwritten by a stale result. Record that + // this cancellation was intentional, so the fetch queue can retry + // immediately rather than waiting for the next external trigger. + if (this.isFetchInFlight) { + this.inFlightFetchCanceled = true; + } + await this.queryClient.cancelQueries({ queryKey: this.queryKey }); + if (attachmentEpoch !== this.attachmentEpoch) { + return; + } + this.queryClient.setQueryData<CachedActiveSessionsData>(this.queryKey, current => { + const existing = current?.sessions ?? []; + return { sessions: updater(existing) }; + }); + this.maybeScheduleEnrichmentRefresh(); + })(); + } + + private maybeScheduleEnrichmentRefresh(): void { + this.updateEnrichmentReason(); + if (this.pendingReasons.has('enrichment') && !this.inFlightReasons?.has('enrichment')) { + this.kickFetch(); + } + } + + private updateEnrichmentReason(): void { + const cachedSessions = + this.queryClient.getQueryData<CachedActiveSessionsData>(this.queryKey)?.sessions ?? []; + if (!hasUnenrichedLiveId(cachedSessions)) { + this.pendingReasons.delete('enrichment'); + return; + } + + if (this.inFlightReasons?.has('enrichment')) { + return; + } + + if ( + this.lastEnrichmentAttemptAt === null || + this.now() - this.lastEnrichmentAttemptAt >= ENRICHMENT_RETRY_MIN_INTERVAL_MS + ) { + this.pendingReasons.add('enrichment'); + } else { + this.pendingReasons.delete('enrichment'); + } + } + + private notifyFetchStart(): void { + this.fetchStartCount += 1; + const waiters = this.fetchStartWaiters; + this.fetchStartWaiters = []; + for (const resolve of waiters) { + resolve(); + } + } + + private notifyFetchCompletion(): void { + const waiters = this.fetchCompletionWaiters; + this.fetchCompletionWaiters = []; + for (const resolve of waiters) { + resolve(); + } + } + + private kickFetch(): void { + if (this.releaseRetain === null) { + return; + } + const attachmentEpoch = this.attachmentEpoch; + // Cancel any in-flight fetch so a newly scheduled refresh can start + // immediately instead of being queued behind a stale one. + if (this.isFetchInFlight) { + // Record that this cancellation was intentional, so the fetch + // queue can retry immediately rather than waiting for the next + // external trigger. + this.inFlightFetchCanceled = true; + void this.queryClient.cancelQueries({ queryKey: this.queryKey }); + } + this.fetchQueue = (async () => { + await this.fetchQueue; + await this.processFetchQueue(attachmentEpoch); + })(); + } + + private async processFetchQueue(attachmentEpoch: number): Promise<void> { + if (attachmentEpoch !== this.attachmentEpoch || this.pendingReasons.size === 0) { + return; + } + if (this.isFetchInFlight) { + return; + } + this.isFetchInFlight = true; + // Reset the cancellation flag for THIS fetch instance. Any cancel + // that targets this fetch will set it back to true before the catch. + this.inFlightFetchCanceled = false; + const inFlightReasons = new Set(this.pendingReasons); + this.inFlightReasons = inFlightReasons; + let success = false; + try { + await this.queryClient.cancelQueries({ queryKey: this.queryKey }); + if (attachmentEpoch === this.attachmentEpoch) { + // fetchQuery with staleTime: 0 forces a network call regardless + // of any preceding setQueryData. + const fetchPromise = this.queryClient.fetchQuery({ + queryKey: this.queryKey, + queryFn: this.queryFn, + staleTime: 0, + }); + // The fetch is now in flight: let tests waiting on getFetchQueue() + // observe the pending state and/or issue cancellations. + this.notifyFetchStart(); + await fetchPromise; + success = true; + } + } catch { + // Either the query was canceled by a later WS write, or the + // network call itself failed. In either case, the reasons stay + // pending so a future scheduleRefresh (or the trailing + // re-kick) re-attempts. + } finally { + this.isFetchInFlight = false; + } + if (attachmentEpoch !== this.attachmentEpoch) { + this.inFlightReasons = null; + this.notifyFetchCompletion(); + return; + } + if (inFlightReasons.has('enrichment')) { + this.lastEnrichmentAttemptAt = this.now(); + } + if (success) { + for (const r of inFlightReasons) { + this.pendingReasons.delete(r); + } + this.updateEnrichmentReason(); + } + const hasNewReasons = [...this.pendingReasons].some(reason => !inFlightReasons.has(reason)); + // Read via helper so control-flow analysis does not treat the field as + // stuck at the `false` written above — other methods flip it during await. + const wasCanceled = this.readInFlightFetchCanceled(); + this.inFlightReasons = null; + this.notifyFetchCompletion(); + // Re-kick immediately only when a replacement fetch is genuinely + // warranted: either the in-flight fetch was intentionally canceled by + // newer work, or new reasons were raised while it was in flight. On a + // genuine failure without either condition, stay quiet and let the + // next scheduled trigger (WS event / fallback poll) retry. + if (this.pendingReasons.size > 0 && (wasCanceled || hasNewReasons)) { + this.kickFetch(); + } + } + + private readInFlightFetchCanceled(): boolean { + return this.inFlightFetchCanceled; + } +} diff --git a/apps/mobile/src/lib/active-sessions-live.merge.test.ts b/apps/mobile/src/lib/active-sessions-live.merge.test.ts new file mode 100644 index 0000000000..7d6f60bdfe --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live.merge.test.ts @@ -0,0 +1,260 @@ +import { describe, expect, it } from 'vitest'; + +import { + type CachedActiveSession, + hasUnenrichedLiveId, + isEnriched, + mergeHeartbeatForActiveSessions, + mergeSnapshotForActiveSessions, + removeActiveSessionsForConnection, +} from '@/lib/active-sessions-live'; + +function makeCached(over: Partial<CachedActiveSession> = {}): CachedActiveSession { + return { + id: 'a1', + status: 'running', + title: 'test', + connectionId: 'c1', + ...over, + }; +} + +// ── Snapshot merge ─────────────────────────────────────────────────── + +describe('mergeSnapshotForActiveSessions', () => { + it('replaces the cache wholesale', () => { + const current = [makeCached({ id: 'a' }), makeCached({ id: 'b' })]; + const snapshot = [ + { id: 'a', status: 'running', title: 'A2', connectionId: 'c1' }, + { id: 'c', status: 'running', title: 'C', connectionId: 'c1' }, + ]; + const result = mergeSnapshotForActiveSessions(current, snapshot); + expect(result.map(r => r.id)).toEqual(['a', 'c']); + expect(result[0]?.title).toBe('A2'); + expect(result[1]?.title).toBe('C'); + }); + + it('preserves ONLY the three enrichment fields for known ids', () => { + const current: CachedActiveSession[] = [ + makeCached({ + id: 'a', + title: 'cached-title', + createdOnPlatform: 'cli', + createdAt: '2024-01-01T00:00:00Z', + updatedAt: '2024-01-02T00:00:00Z', + gitUrl: 'git@cached', + }), + ]; + const snapshot = [ + { + id: 'a', + status: 'running', + title: 'ws-title', + connectionId: 'c1', + gitUrl: 'git@ws', + }, + ]; + const result = mergeSnapshotForActiveSessions(current, snapshot); + expect(result).toHaveLength(1); + expect(result[0]?.title).toBe('ws-title'); + expect(result[0]?.connectionId).toBe('c1'); + expect(result[0]?.gitUrl).toBe('git@ws'); + expect(result[0]?.createdOnPlatform).toBe('cli'); + expect(result[0]?.createdAt).toBe('2024-01-01T00:00:00Z'); + expect(result[0]?.updatedAt).toBe('2024-01-02T00:00:00Z'); + }); + + it('drops rows absent from the snapshot', () => { + const current = [makeCached({ id: 'a' }), makeCached({ id: 'b' })]; + const snapshot = [{ id: 'a', status: 'running', title: 'A', connectionId: 'c1' }]; + expect(mergeSnapshotForActiveSessions(current, snapshot).map(r => r.id)).toEqual(['a']); + }); + + it('a WS-new id enters unenriched (no enrichment fields filled in)', () => { + const current: CachedActiveSession[] = []; + const snapshot = [{ id: 'new', status: 'running', title: 'New', connectionId: 'c1' }]; + const result = mergeSnapshotForActiveSessions(current, snapshot); + expect(result[0]).toMatchObject({ + id: 'new', + title: 'New', + connectionId: 'c1', + createdOnPlatform: undefined, + createdAt: undefined, + updatedAt: undefined, + }); + expect(result[0]).toBeDefined(); + expect(isEnriched(result[0] ?? makeCached())).toBe(false); + }); + + it('a snapshot must never wipe enrichment for known ids', () => { + const current: CachedActiveSession[] = [ + makeCached({ + id: 'a', + createdOnPlatform: 'cli', + createdAt: '2024-01-01T00:00:00Z', + updatedAt: '2024-01-02T00:00:00Z', + }), + ]; + const snapshot = [ + { id: 'a', status: 'running', title: 'A', connectionId: 'c1' }, + { id: 'b', status: 'running', title: 'B', connectionId: 'c1' }, + ]; + const result = mergeSnapshotForActiveSessions(current, snapshot); + const found = result.find(r => r.id === 'a'); + expect(found).toBeDefined(); + expect(found?.createdOnPlatform).toBe('cli'); + expect(found?.createdAt).toBe('2024-01-01T00:00:00Z'); + expect(found?.updatedAt).toBe('2024-01-02T00:00:00Z'); + }); +}); + +// ── Heartbeat merge ────────────────────────────────────────────────── + +describe('mergeHeartbeatForActiveSessions', () => { + it('replaces rows for the heartbeat connectionId and keeps other connections', () => { + const current = [ + makeCached({ id: 'a', connectionId: 'c1' }), + makeCached({ id: 'b', connectionId: 'c1' }), + makeCached({ id: 'x', connectionId: 'c2' }), + ]; + const payload = { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A2' }], + }; + const result = mergeHeartbeatForActiveSessions(current, payload); + expect(result.map(r => `${r.id}/${r.connectionId}`)).toEqual(['a/c1', 'x/c2']); + const found = result.find(r => r.id === 'a'); + expect(found).toBeDefined(); + expect(found?.title).toBe('A2'); + }); + + it('preserves enrichment for ids present in both cache and payload', () => { + const current: CachedActiveSession[] = [ + makeCached({ + id: 'a', + connectionId: 'c1', + createdOnPlatform: 'cli', + createdAt: '2024-01-01T00:00:00Z', + updatedAt: '2024-01-02T00:00:00Z', + }), + ]; + const payload = { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A2' }], + }; + const result = mergeHeartbeatForActiveSessions(current, payload); + expect(result[0]).toBeDefined(); + expect(result[0]?.title).toBe('A2'); + expect(result[0]?.createdOnPlatform).toBe('cli'); + expect(result[0]?.createdAt).toBe('2024-01-01T00:00:00Z'); + expect(result[0]?.updatedAt).toBe('2024-01-02T00:00:00Z'); + }); + + it('drops a cached row whose id appears in the payload under a different connectionId (ownership transfer)', () => { + const current = [makeCached({ id: 'a', connectionId: 'c1' })]; + const payload = { + connectionId: 'c2', + sessions: [{ id: 'a', status: 'running', title: 'A on c2' }], + }; + const result = mergeHeartbeatForActiveSessions(current, payload); + expect(result).toHaveLength(1); + expect(result[0]).toMatchObject({ id: 'a', connectionId: 'c2' }); + }); + + it('does not duplicate when both connections stay live with the same session id', () => { + const current = [makeCached({ id: 'a', connectionId: 'c1', title: 'old' })]; + const payloadFromC2 = { + connectionId: 'c2', + sessions: [{ id: 'a', status: 'running', title: 'A on c2' }], + }; + const mergedAfterTransfer = mergeHeartbeatForActiveSessions(current, payloadFromC2); + const payloadFromC1 = { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A on c1 again' }], + }; + const final = mergeHeartbeatForActiveSessions(mergedAfterTransfer, payloadFromC1); + expect(final).toHaveLength(1); + expect(final[0]).toMatchObject({ id: 'a', connectionId: 'c1', title: 'A on c1 again' }); + }); + + it('preserves enrichment across ownership transfer', () => { + const current: CachedActiveSession[] = [ + makeCached({ + id: 'a', + connectionId: 'c1', + createdOnPlatform: 'cli', + createdAt: '2024-01-01T00:00:00Z', + updatedAt: '2024-01-02T00:00:00Z', + }), + ]; + const payload = { + connectionId: 'c2', + sessions: [{ id: 'a', status: 'running', title: 'A on c2' }], + }; + const result = mergeHeartbeatForActiveSessions(current, payload); + expect(result).toHaveLength(1); + expect(result[0]).toMatchObject({ + id: 'a', + connectionId: 'c2', + createdOnPlatform: 'cli', + createdAt: '2024-01-01T00:00:00Z', + updatedAt: '2024-01-02T00:00:00Z', + }); + }); + + it('omitted session id leaves the row out (heartbeat omits prev id)', () => { + const current = [ + makeCached({ id: 'a', connectionId: 'c1' }), + makeCached({ id: 'b', connectionId: 'c1' }), + ]; + const payload = { + connectionId: 'c1', + sessions: [{ id: 'a', status: 'running', title: 'A' }], + }; + const result = mergeHeartbeatForActiveSessions(current, payload); + expect(result.map(r => r.id)).toEqual(['a']); + }); +}); + +// ── Connection removal ─────────────────────────────────────────────── + +describe('removeActiveSessionsForConnection', () => { + it('drops rows for the given connectionId', () => { + const current = [ + makeCached({ id: 'a', connectionId: 'c1' }), + makeCached({ id: 'b', connectionId: 'c2' }), + ]; + expect(removeActiveSessionsForConnection(current, 'c1').map(r => r.id)).toEqual(['b']); + }); + + it('owner transfer followed by old owner cli.disconnected removes nothing', () => { + const current: CachedActiveSession[] = [makeCached({ id: 'a', connectionId: 'c2' })]; + expect(removeActiveSessionsForConnection(current, 'c1').map(r => r.id)).toEqual(['a']); + }); +}); + +// ── Enrichment helpers ─────────────────────────────────────────────── + +describe('isEnriched / hasUnenrichedLiveId', () => { + it('an unenriched row is detected', () => { + expect(isEnriched(makeCached())).toBe(false); + expect(hasUnenrichedLiveId([makeCached()])).toBe(true); + }); + it('a row with createdOnPlatform is enriched', () => { + expect(isEnriched(makeCached({ createdOnPlatform: 'cli' }))).toBe(true); + }); + it('a row with only createdAt is enriched', () => { + expect(isEnriched(makeCached({ createdAt: '2024-01-01T00:00:00Z' }))).toBe(true); + }); + it('a row with only updatedAt is enriched', () => { + expect(isEnriched(makeCached({ updatedAt: '2024-01-02T00:00:00Z' }))).toBe(true); + }); + it('hasUnenrichedLiveId is false when every row is enriched', () => { + expect( + hasUnenrichedLiveId([ + makeCached({ createdOnPlatform: 'cli' }), + makeCached({ createdAt: '2024-01-01T00:00:00Z' }), + ]) + ).toBe(false); + }); +}); diff --git a/apps/mobile/src/lib/active-sessions-live.test.ts b/apps/mobile/src/lib/active-sessions-live.test.ts new file mode 100644 index 0000000000..00f847c9f4 --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live.test.ts @@ -0,0 +1,93 @@ +import { describe, expect, it } from 'vitest'; + +import { + parseCliConnectionPayload, + parseHeartbeatPayload, + parseSessionsListPayload, + selectRootWsSessions, +} from '@/lib/active-sessions-live'; + +// ── Root filter ────────────────────────────────────────────────────── + +describe('selectRootWsSessions', () => { + it('drops rows with a parentSessionId', () => { + const rows = [ + { id: 'root1', status: 'running', title: 'r1', connectionId: 'c1' }, + { + id: 'child1', + status: 'running', + title: 'c1', + connectionId: 'c1', + parentSessionId: 'root1', + }, + { id: 'root2', status: 'running', title: 'r2', connectionId: 'c1' }, + ]; + expect(selectRootWsSessions(rows).map(r => r.id)).toEqual(['root1', 'root2']); + }); +}); + +// ── Payload parsing ────────────────────────────────────────────────── + +describe('parseHeartbeatPayload', () => { + it('parses a valid heartbeat', () => { + expect( + parseHeartbeatPayload({ + connectionId: 'c1', + sessions: [{ id: 's1', status: 'running', title: 't' }], + }) + ).toEqual({ + connectionId: 'c1', + sessions: [{ id: 's1', status: 'running', title: 't' }], + }); + }); + + it('strips the extra protocolVersion field (default Zod behavior on the parent)', () => { + // The SDK schema only requires connectionId+sessions; extra fields + // are accepted and dropped, so this should still parse to the same + // shape. + const parsed = parseHeartbeatPayload({ + connectionId: 'c1', + protocolVersion: 2, + sessions: [{ id: 's1', status: 'running', title: 't' }], + }); + expect(parsed).toEqual({ + connectionId: 'c1', + sessions: [{ id: 's1', status: 'running', title: 't' }], + }); + }); + + it('rejects when connectionId is missing', () => { + expect(parseHeartbeatPayload({ sessions: [] })).toBeNull(); + }); + + it('rejects when sessions is not an array', () => { + expect(parseHeartbeatPayload({ connectionId: 'c1', sessions: 'nope' })).toBeNull(); + }); +}); + +describe('parseSessionsListPayload', () => { + it('parses a valid sessions list', () => { + expect( + parseSessionsListPayload({ + sessions: [{ id: 's1', status: 'running', title: 't', connectionId: 'c1' }], + }) + ).toEqual([{ id: 's1', status: 'running', title: 't', connectionId: 'c1' }]); + }); + + it('rejects when connectionId is missing on a row', () => { + expect( + parseSessionsListPayload({ + sessions: [{ id: 's1', status: 'running', title: 't' }], + }) + ).toBeNull(); + }); +}); + +describe('parseCliConnectionPayload', () => { + it('parses a cli connection payload', () => { + expect(parseCliConnectionPayload({ connectionId: 'c1' })).toEqual({ connectionId: 'c1' }); + }); + it('rejects unknown shape', () => { + expect(parseCliConnectionPayload({})).toBeNull(); + }); +}); diff --git a/apps/mobile/src/lib/active-sessions-live.ts b/apps/mobile/src/lib/active-sessions-live.ts new file mode 100644 index 0000000000..043eda556e --- /dev/null +++ b/apps/mobile/src/lib/active-sessions-live.ts @@ -0,0 +1,220 @@ +/** + * Pure helpers for the app-level active-sessions live-sync owner. + * + * WS payloads lack enrichment fields (`createdOnPlatform`/`createdAt`/ + * `updatedAt`); the merge helpers preserve those fields for ids already in + * the cache while letting every other field (including `connectionId`) + * come from the latest WS payload, so session ownership can transfer + * between CLI connections. The functions here never touch React, the + * network, or a QueryClient — they are pure and exhaustively unit-tested + * alongside this file. + */ + +import { + type CliConnectionData, + cliConnectionDataSchema, + type HeartbeatData, + heartbeatDataSchema, + type SessionsListData, + sessionsListDataSchema, +} from 'cloud-agent-sdk/schemas'; + +import { type ActiveSession } from '@/lib/hooks/use-agent-sessions'; + +/** + * Incoming WS session row (per the SDK schemas). Carries + * `parentSessionId` so the root filter can drop subagent sessions; the + * cached `ActiveSession` (from the tRPC router) does not. + */ +type IncomingWsSession = { + id: string; + status: string; + title: string; + gitUrl?: string; + gitBranch?: string; + parentSessionId?: string; + connectionId?: string; +}; + +/** + * Cached active session: `ActiveSession` (tRPC output) plus the + * enrichment fields `createdOnPlatform` / `createdAt` / `updatedAt` that + * the live-sync owner preserves across WS updates. + */ +export type CachedActiveSession = ActiveSession; + +export type CachedActiveSessionsData = { + sessions: CachedActiveSession[]; +}; + +/** + * The three enrichment fields that the live-sync owner preserves across + * WS updates. Every other field comes from the latest WS payload (this + * includes `connectionId`, so ownership transfer between CLIs lands + * correctly on the next heartbeat/snapshot). + */ +const ENRICHMENT_FIELDS = ['createdOnPlatform', 'createdAt', 'updatedAt'] as const; +type EnrichmentField = (typeof ENRICHMENT_FIELDS)[number]; + +function isRootWsSession(session: IncomingWsSession): boolean { + return !session.parentSessionId; +} + +export function selectRootWsSessions<T extends IncomingWsSession>(sessions: readonly T[]): T[] { + return sessions.filter(session => isRootWsSession(session)); +} + +// ── Payload parsing (WS trust boundary) ────────────────────────────── + +type HeartbeatPayload = { + connectionId: string; + sessions: HeartbeatData['sessions']; +}; + +export function parseHeartbeatPayload(value: unknown): HeartbeatPayload | null { + const parsed = heartbeatDataSchema.safeParse(value); + if (!parsed.success) { + return null; + } + return { + connectionId: parsed.data.connectionId, + sessions: parsed.data.sessions, + }; +} + +export function parseSessionsListPayload(value: unknown): SessionsListData['sessions'] | null { + const parsed = sessionsListDataSchema.safeParse(value); + if (!parsed.success) { + return null; + } + return parsed.data.sessions; +} + +export function parseCliConnectionPayload(value: unknown): CliConnectionData | null { + const parsed = cliConnectionDataSchema.safeParse(value); + if (!parsed.success) { + return null; + } + return parsed.data; +} + +// ── Enrichment-preserving merge helpers ────────────────────────────── + +function readEnrichment( + current: CachedActiveSession | undefined +): Record<EnrichmentField, string | undefined> { + return { + createdOnPlatform: + typeof current?.createdOnPlatform === 'string' ? current.createdOnPlatform : undefined, + createdAt: typeof current?.createdAt === 'string' ? current.createdAt : undefined, + updatedAt: typeof current?.updatedAt === 'string' ? current.updatedAt : undefined, + }; +} + +type WithoutConnectionId<T> = Omit<T, 'connectionId'>; + +function withEnrichmentAndConnectionId( + row: WithoutConnectionId<IncomingWsSession>, + current: CachedActiveSession | undefined, + connectionId: string +): CachedActiveSession { + const enrichment = readEnrichment(current); + return { + id: row.id, + status: row.status, + title: row.title, + gitUrl: row.gitUrl, + gitBranch: row.gitBranch, + connectionId, + ...enrichment, + }; +} + +/** + * Replace the entire cache with the snapshot. Rows whose id is in both + * the snapshot and the cache keep ONLY the three enrichment fields from + * the cache; every other field (including `connectionId`) comes from the + * snapshot. Rows absent from the snapshot are dropped. + */ +export function mergeSnapshotForActiveSessions( + current: readonly CachedActiveSession[], + snapshot: SessionsListData['sessions'] +): CachedActiveSession[] { + const currentById = new Map<string, CachedActiveSession>(); + for (const row of current) { + currentById.set(row.id, row); + } + const snapshotIds = new Set<string>(); + const result: CachedActiveSession[] = []; + for (const row of snapshot) { + snapshotIds.add(row.id); + const enriched = withEnrichmentAndConnectionId(row, currentById.get(row.id), row.connectionId); + result.push(enriched); + } + return result.filter(row => snapshotIds.has(row.id)); +} + +/** + * Heartbeat merge: id-unique with latest-payload-wins. + * + * Stronger than web's `applyActiveSessionsHeartbeat`: in addition to + * dropping cached rows whose `connectionId` matches the payload's + * connectionId, this also drops cached rows whose session id appears in + * the payload under a DIFFERENT connectionId — so ownership transfer + * between CLIs (same session id, new owner) reflects the new owner on + * the next heartbeat without leaving a stale copy under the old one. + */ +export function mergeHeartbeatForActiveSessions( + current: readonly CachedActiveSession[], + payload: HeartbeatPayload +): CachedActiveSession[] { + const currentById = new Map<string, CachedActiveSession>(); + for (const row of current) { + currentById.set(row.id, row); + } + const payloadIds = new Set<string>(); + for (const row of payload.sessions) { + payloadIds.add(row.id); + } + + const result: CachedActiveSession[] = []; + for (const row of payload.sessions) { + const enriched = withEnrichmentAndConnectionId( + row, + currentById.get(row.id), + payload.connectionId + ); + result.push(enriched); + } + for (const row of current) { + if (row.connectionId !== payload.connectionId && !payloadIds.has(row.id)) { + result.push(row); + } + } + return result; +} + +export function removeActiveSessionsForConnection( + current: readonly CachedActiveSession[], + connectionId: string +): CachedActiveSession[] { + return current.filter(row => row.connectionId !== connectionId); +} + +/** + * Heuristic: a row counts as "enriched" when at least one enrichment + * field is set. Empty `createdOnPlatform` (e.g. `'unknown'`) is still a + * real value from the tRPC router and counts as enriched; the trpc + * pipeline is the source of truth for "the DB row has been joined in". + */ +export function isEnriched(row: CachedActiveSession): boolean { + return ( + typeof row.createdOnPlatform === 'string' || + typeof row.createdAt === 'string' || + typeof row.updatedAt === 'string' + ); +} + +export function hasUnenrichedLiveId(rows: readonly CachedActiveSession[]): boolean { + return rows.some(row => !isEnriched(row)); +} diff --git a/apps/mobile/src/lib/hooks/use-agent-sessions.ts b/apps/mobile/src/lib/hooks/use-agent-sessions.ts index fb8e14ef70..41714c52cf 100644 --- a/apps/mobile/src/lib/hooks/use-agent-sessions.ts +++ b/apps/mobile/src/lib/hooks/use-agent-sessions.ts @@ -13,6 +13,7 @@ import { parseAgentSessionSortBy, } from '@/lib/agent-session-sort'; import { useTRPC } from '@/lib/trpc'; +import { useUserWebConnectionState } from '@/lib/hooks/use-user-web-connection-state'; // ── Types ──────────────────────────────────────────────────────────── @@ -84,9 +85,15 @@ function useStoredSessions(options?: UseAgentSessionsOptions) { function useActiveSessions(options?: UseAgentSessionsOptions) { const trpc = useTRPC(); + // While the shared WS is connected, the app-level `ActiveSessionsLiveSync` + // owner pushes tray updates through `setQueryData` and triggers refreshes + // on connect/enrichment/etc. — the 10s poll would only mask the WS as + // the source of truth. When the socket is down, fall back to the 10s + // interval so a transient outage still updates the tray. + const wsConnected = useUserWebConnectionState(); return useQuery( trpc.activeSessions.list.queryOptions(undefined, { - refetchInterval: 10_000, + refetchInterval: wsConnected ? false : 10_000, staleTime: 5000, enabled: options?.enabled, }) diff --git a/apps/mobile/src/lib/hooks/use-user-web-connection-state.ts b/apps/mobile/src/lib/hooks/use-user-web-connection-state.ts new file mode 100644 index 0000000000..7865958d6b --- /dev/null +++ b/apps/mobile/src/lib/hooks/use-user-web-connection-state.ts @@ -0,0 +1,19 @@ +import { useSyncExternalStore } from 'react'; + +import { useUserWebConnection } from '@/components/agents/user-web-connection-provider'; + +/** + * Reactive binding to the shared user-web-connection transport + * readiness. Used by `useActiveSessions` to flip its `refetchInterval` + * between 10s (offline) and disabled (connected). The subscription is + * `useSyncExternalStore` over the wrapper-owned connection-state API + * added in S2 (`isConnected` + `onConnectionChange`), so concurrent + * consumers share one subscription per connection. + */ +export function useUserWebConnectionState(): boolean { + const connection = useUserWebConnection(); + return useSyncExternalStore( + listener => connection.onConnectionChange(listener), + () => connection.isConnected() + ); +} diff --git a/apps/mobile/vitest.config.ts b/apps/mobile/vitest.config.ts index 7f3e3dba83..956954bcc5 100644 --- a/apps/mobile/vitest.config.ts +++ b/apps/mobile/vitest.config.ts @@ -39,6 +39,9 @@ export default defineConfig({ 'cloud-agent-sdk/preparation-attempts': fileURLToPath( new URL('../../apps/web/src/lib/cloud-agent-sdk/preparation-attempts.ts', import.meta.url) ), + 'cloud-agent-sdk/schemas': fileURLToPath( + new URL('../../apps/web/src/lib/cloud-agent-sdk/schemas.ts', import.meta.url) + ), 'cloud-agent-sdk': fileURLToPath( new URL('../../apps/web/src/lib/cloud-agent-sdk/index.ts', import.meta.url) ), diff --git a/apps/web/src/lib/cloud-agent-sdk/session-routing.test.ts b/apps/web/src/lib/cloud-agent-sdk/session-routing.test.ts index b35e9ecb86..64e035f1c0 100644 --- a/apps/web/src/lib/cloud-agent-sdk/session-routing.test.ts +++ b/apps/web/src/lib/cloud-agent-sdk/session-routing.test.ts @@ -123,9 +123,12 @@ describe('session transport routing', () => { ); const release = jest.fn(); const userWebConnection = { + retain: jest.fn(() => jest.fn()), connect: jest.fn(), disconnect: jest.fn(), destroy: jest.fn(), + isConnected: jest.fn(() => false), + onConnectionChange: jest.fn(() => jest.fn()), subscribeToCliSession: jest.fn(() => release), sendCommand: jest.fn(() => Promise.resolve()), sendCommandToConnection: jest.fn(() => Promise.resolve()), @@ -214,6 +217,8 @@ describe('session transport routing', () => { connect: jest.fn(), disconnect: jest.fn(), destroy: jest.fn(), + isConnected: jest.fn(() => false), + onConnectionChange: jest.fn(() => jest.fn()), subscribeToCliSession: jest.fn(() => subscribeRelease), sendCommand: jest.fn(() => Promise.resolve()), sendCommandToConnection: jest.fn(() => Promise.resolve()), diff --git a/apps/web/src/lib/cloud-agent-sdk/session-transport.test.ts b/apps/web/src/lib/cloud-agent-sdk/session-transport.test.ts index df57f4651c..936f1838a2 100644 --- a/apps/web/src/lib/cloud-agent-sdk/session-transport.test.ts +++ b/apps/web/src/lib/cloud-agent-sdk/session-transport.test.ts @@ -94,9 +94,12 @@ async function connectSession(session: CloudAgentSession): Promise<void> { function createUserWebConnection() { let systemListener: ((event: UserWebSystemEvent) => void) | undefined; return { + retain: jest.fn(() => jest.fn()), connect: jest.fn(), disconnect: jest.fn(), destroy: jest.fn(), + isConnected: jest.fn(() => false), + onConnectionChange: jest.fn(() => jest.fn()), subscribeToCliSession: jest.fn(() => jest.fn()), sendCommand: jest.fn((_sessionId: string, command: string) => Promise.resolve( diff --git a/apps/web/src/lib/cloud-agent-sdk/session.test.ts b/apps/web/src/lib/cloud-agent-sdk/session.test.ts index 54ad04a8b8..2ca61225a7 100644 --- a/apps/web/src/lib/cloud-agent-sdk/session.test.ts +++ b/apps/web/src/lib/cloud-agent-sdk/session.test.ts @@ -369,9 +369,12 @@ describe('remote session transport state', () => { const onRemoteModelStateChange = jest.fn<void, [RemoteModelState]>(); const onTransportCapabilityChange = jest.fn(); const userWebConnection = { + retain: jest.fn(() => jest.fn()), connect: jest.fn(), disconnect: jest.fn(), destroy: jest.fn(), + isConnected: jest.fn(() => false), + onConnectionChange: jest.fn(() => jest.fn()), subscribeToCliSession: jest.fn(() => jest.fn()), sendCommand: jest.fn(() => Promise.resolve({ @@ -439,9 +442,12 @@ describe('remote session create and retry commands', () => { function createRemoteSessionFixture() { let systemListener: ((event: UserWebSystemEvent) => void) | undefined; const userWebConnection = { + retain: jest.fn(() => jest.fn()), connect: jest.fn(), disconnect: jest.fn(), destroy: jest.fn(), + isConnected: jest.fn(() => false), + onConnectionChange: jest.fn(() => jest.fn()), subscribeToCliSession: jest.fn(() => jest.fn()), sendCommand: jest.fn<Promise<unknown>, [string, string, unknown, string?]>(() => Promise.resolve({ diff --git a/apps/web/src/lib/cloud-agent-sdk/user-web-connection.test.ts b/apps/web/src/lib/cloud-agent-sdk/user-web-connection.test.ts index 9a12a44ce2..88abcad83a 100644 --- a/apps/web/src/lib/cloud-agent-sdk/user-web-connection.test.ts +++ b/apps/web/src/lib/cloud-agent-sdk/user-web-connection.test.ts @@ -1706,3 +1706,193 @@ describe('createUserWebConnection sendCommandToConnection', () => { await expect(promise).rejects.not.toBeInstanceOf(UserWebCommandError); }); }); + +describe('createUserWebConnection connection-state API', () => { + it('reports false until the first server message flips it true and true until the final release flips it false', () => { + const client = createUserWebConnection({ websocketUrl: WS_URL, getAuthToken: () => 'token' }); + const listener = jest.fn(); + const unsubscribe = client.onConnectionChange(listener); + + expect(client.isConnected()).toBe(false); + + const release = client.retain(); + open(); + // Open alone does not mark connected — only the first inbound message does + // (mirroring the base-connection's `onConnected` semantics). + expect(client.isConnected()).toBe(false); + + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + expect(client.isConnected()).toBe(true); + expect(listener).toHaveBeenLastCalledWith(true); + + release(); + expect(client.isConnected()).toBe(false); + expect(listener).toHaveBeenLastCalledWith(false); + + unsubscribe(); + client.destroy(); + }); + + it('covers a release-then-retain transition (true → release-all → false → retain → true)', () => { + const client = createUserWebConnection({ websocketUrl: WS_URL, getAuthToken: () => 'token' }); + const listener = jest.fn(); + client.onConnectionChange(listener); + + const releaseA = client.retain(); + const releaseB = client.retain(); + open(); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + expect(client.isConnected()).toBe(true); + const transitionCountAfterFirstConnect = listener.mock.calls.length; + + releaseA(); + expect(client.isConnected()).toBe(true); + releaseB(); + expect(client.isConnected()).toBe(false); + + const releaseC = client.retain(); + open(sockets[1]); + expect(client.isConnected()).toBe(false); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }, sockets[1]); + expect(client.isConnected()).toBe(true); + + const transitions = listener.mock.calls.map(call => call[0]); + expect(transitions).toEqual([true, false, true]); + expect(listener.mock.calls.length).toBe(transitionCountAfterFirstConnect + 2); + + releaseC(); + client.destroy(); + }); + + it('reports false on destroy() even though base-connection.destroy() emits no callback', () => { + const client = createUserWebConnection({ websocketUrl: WS_URL, getAuthToken: () => 'token' }); + const listener = jest.fn(); + client.onConnectionChange(listener); + + client.retain(); + open(); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + expect(client.isConnected()).toBe(true); + + client.destroy(); + + expect(client.isConnected()).toBe(false); + expect(listener.mock.calls.map(call => call[0])).toEqual([true, false]); + }); + + it('does not emit state changes after destroy()', () => { + const client = createUserWebConnection({ websocketUrl: WS_URL, getAuthToken: () => 'token' }); + const listener = jest.fn(); + client.onConnectionChange(listener); + + client.retain(); + open(); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + expect(listener.mock.calls.map(call => call[0])).toEqual([true]); + + client.destroy(); + // destroy() must drive the final false transition; the listener set is + // cleared as part of teardown so no further notifications can fire. + expect(listener.mock.calls.map(call => call[0])).toEqual([true, false]); + + // A late inbound on the now-destroyed client must not emit another + // transition. + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + expect(listener.mock.calls.map(call => call[0])).toEqual([true, false]); + }); + + it('ignores connection-state listeners registered after destroy()', () => { + const client = createUserWebConnection({ websocketUrl: WS_URL, getAuthToken: () => 'token' }); + client.destroy(); + const listener = jest.fn(); + + const addSpy = jest.spyOn(Set.prototype, 'add'); + try { + const unsubscribe = client.onConnectionChange(listener); + expect(addSpy).not.toHaveBeenCalled(); + unsubscribe(); + } finally { + addSpy.mockRestore(); + } + + expect(listener).not.toHaveBeenCalled(); + expect(client.isConnected()).toBe(false); + }); + + it('flips disconnected → connected across a pong-timeout-triggered reconnect', async () => { + jest.useFakeTimers(); + try { + const getAuthToken = jest + .fn() + .mockReturnValueOnce('old-token') + .mockResolvedValue('recovered-token'); + const client = createUserWebConnection({ websocketUrl: WS_URL, getAuthToken }); + const listener = jest.fn(); + client.onConnectionChange(listener); + const release = client.retain(); + open(); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + expect(client.isConnected()).toBe(true); + const callsAfterFirstConnect = listener.mock.calls.length; + + // Drive the existing ping/pong liveness machinery: the ping fires on + // the viewer-ping interval, the pong is wrong, then the pong-timeout + // triggers `reconnectWithRefreshedAuth` — which fires `onDisconnected` + // before the new socket's first message flips state back to true. + jest.advanceTimersByTime(VIEWER_PING_INTERVAL_MS); + inbound({ type: 'pong', nonce: 'wrong-nonce' }); + jest.advanceTimersByTime(VIEWER_PONG_TIMEOUT_MS); + await Promise.resolve(); + await Promise.resolve(); + + expect(sockets[0].close).toHaveBeenCalledTimes(1); + expect(sockets[1]).toBeDefined(); + expect(client.isConnected()).toBe(false); + + open(sockets[1]); + expect(client.isConnected()).toBe(false); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }, sockets[1]); + expect(client.isConnected()).toBe(true); + + const transitions = listener.mock.calls.slice(callsAfterFirstConnect).map(call => call[0]); + expect(transitions).toEqual([false, true]); + + release(); + client.destroy(); + } finally { + jest.useRealTimers(); + } + }); + + it('coalesces repeated identical transitions and the unsubscribe is a no-op thereafter', () => { + const client = createUserWebConnection({ websocketUrl: WS_URL, getAuthToken: () => 'token' }); + const listener = jest.fn(); + const unsubscribe = client.onConnectionChange(listener); + + const releaseA = client.retain(); + const releaseB = client.retain(); + open(); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }); + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenLastCalledWith(true); + + releaseA(); + // Release of a non-final retain does not change state. + expect(listener).toHaveBeenCalledTimes(1); + + releaseB(); + expect(listener).toHaveBeenCalledTimes(2); + expect(listener).toHaveBeenLastCalledWith(false); + + unsubscribe(); + const releaseC = client.retain(); + open(sockets[1]); + inbound({ type: 'system', event: 'sessions.list', data: { sessions: [] } }, sockets[1]); + expect(listener).toHaveBeenCalledTimes(2); + + releaseC(); + client.destroy(); + }); +}); diff --git a/apps/web/src/lib/cloud-agent-sdk/user-web-connection.ts b/apps/web/src/lib/cloud-agent-sdk/user-web-connection.ts index 5c45b98c5e..5b81fbda2e 100644 --- a/apps/web/src/lib/cloud-agent-sdk/user-web-connection.ts +++ b/apps/web/src/lib/cloud-agent-sdk/user-web-connection.ts @@ -69,13 +69,26 @@ type SendCommandToConnectionInput = { }; type UserWebConnection = { - /** New connection owners use this lease; optional on injected legacy clients until they migrate. */ - retain?: () => () => void; + retain: () => () => void; /** @deprecated Retain the connection explicitly with retain() instead. */ connect: () => void; /** @deprecated Release the function returned by retain() instead. */ disconnect: () => void; destroy: () => void; + /** + * Returns the current high-level transport readiness. `true` once the base + * connection has reported its first successful message after open (or + * re-open); `false` otherwise — including during reconnect attempts and + * after a final release or `destroy()`. + */ + isConnected: () => boolean; + /** + * Subscribes to transport-readiness transitions. The listener fires only + * when the boolean value actually changes. The returned function + * unsubscribes. Safe to call on a destroyed connection: the listener is + * never invoked and the unsubscribe is a no-op. + */ + onConnectionChange: (listener: (connected: boolean) => void) => () => void; subscribeToCliSession: (sessionId: string) => () => void; sendCommand: ( sessionId: string, @@ -159,6 +172,17 @@ function createUserWebConnection( reject: (reason: Error) => void; timer: ReturnType<typeof setTimeout>; }>(); + // Wrapper-owned transport-readiness state. The base connection deliberately + // emits no callback on `destroy()`, so transitions around full release / + // destroy are driven here rather than from base-connection callbacks. + let connected = false; + const connectionChangeListeners = new Set<(connected: boolean) => void>(); + + function setConnected(value: boolean): void { + if (connected === value) return; + connected = value; + for (const listener of connectionChangeListeners) listener(value); + } function hasLifetime(): boolean { return retainCount > 0; @@ -417,8 +441,11 @@ function createUserWebConnection( for (const sessionId of subscriptionCounts.keys()) sendSubscribe(sessionId); startLiveness(); }, - onConnected: () => {}, + onConnected: () => { + setConnected(true); + }, onReconnected: () => { + setConnected(true); config.onReconnect?.(); for (const listener of reconnectListeners) listener(); }, @@ -428,9 +455,11 @@ function createUserWebConnection( onDisconnected: () => { currentWs = null; clearLiveness(); + setConnected(false); }, onUnexpectedDisconnect: () => { rejectPending('Connection lost during reconnect'); + setConnected(false); }, onError: config.onError, isAuthFailure: event => event.code === 4001 || event.code === 1008, @@ -491,6 +520,10 @@ function createUserWebConnection( rejectPending(message); baseConnection?.destroy(); baseConnection = null; + // `base-connection.destroy()` deliberately emits no `onDisconnected` + // callback, so the wrapper must drive the disconnected transition itself + // before the next `startConnection()` reopens from a false baseline. + setConnected(false); } function connect(): void { @@ -590,6 +623,15 @@ function createUserWebConnection( systemListeners.clear(); reconnectListeners.clear(); sessionListeners.clear(); + connectionChangeListeners.clear(); + }, + isConnected: () => connected, + onConnectionChange(listener) { + if (destroyed) return () => {}; + connectionChangeListeners.add(listener); + return () => { + connectionChangeListeners.delete(listener); + }; }, subscribeToCliSession(sessionId) { if (destroyed) return () => {}; diff --git a/services/session-ingest/src/dos/UserConnectionDO.test.ts b/services/session-ingest/src/dos/UserConnectionDO.test.ts index b067243f54..2928918522 100644 --- a/services/session-ingest/src/dos/UserConnectionDO.test.ts +++ b/services/session-ingest/src/dos/UserConnectionDO.test.ts @@ -450,6 +450,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, firstOwner, [makeSession('s1')]); sendHeartbeat(doInstance, nextOwner, []); firstOwner.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'list_models', @@ -461,7 +462,10 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, nextOwner, [makeSession('s1')]); - expect(parseSent(webWs)).toEqual({ + // The owner-change heartbeat broadcasts sessions.heartbeat and also fires + // the SESSION_OWNER_CHANGED error response for the in-flight command. The + // test cares about the latter; find it by type+id. + expect(allSent(webWs).find(m => m.type === 'response' && m.id === 'cmd-1')).toEqual({ type: 'response', id: 'cmd-1', error: { @@ -471,7 +475,11 @@ describe('UserConnectionDO', () => { }, }); sendCliResponse(doInstance, firstOwner, { id: correlationId, result: 'late' }); - expect(webWs.send).toHaveBeenCalledTimes(1); + // The late sendCliResponse is filtered (the pending entry was already + // removed by the owner-change path) so webWs sees exactly the two + // messages produced by the owner-change heartbeat itself: the broadcast + // sessions.heartbeat and the SESSION_OWNER_CHANGED error response. + expect(webWs.send).toHaveBeenCalledTimes(2); }); it('replays existing web subscriptions when a session gets a new CLI owner', () => { @@ -498,7 +506,7 @@ describe('UserConnectionDO', () => { expect(cli2Msgs).toContainEqual({ type: 'subscribe', sessionId: 's1' }); }); - it('sends heartbeat only to web clients subscribed to sessions from this connection', () => { + it('broadcasts heartbeat to every web socket regardless of subscription', () => { const { doInstance, mockCtx } = setup(); const cli1 = addCliSocket(mockCtx, 'cli-1'); const cli2 = addCliSocket(mockCtx, 'cli-2'); @@ -509,13 +517,14 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cli1, [makeSession('s1')]); sendHeartbeat(doInstance, cli2, [makeSession('s2')]); - // subWeb subscribes to s1, otherWeb subscribes to s2 + // subWeb subscribes to s1, otherWeb subscribes to s2 (subscriptions are + // irrelevant to broadcast delivery — both should still receive cli1's heartbeat) sendSubscribe(doInstance, subWeb, 's1'); sendSubscribe(doInstance, otherWeb, 's2'); subWeb.send.mockClear(); otherWeb.send.mockClear(); - // cli1 sends heartbeat — only subWeb (watching s1) should receive it + // cli1 sends heartbeat — both viewers must receive it sendHeartbeat(doInstance, cli1, [makeSession('s1')]); expect(subWeb.send).toHaveBeenCalledTimes(1); @@ -524,10 +533,45 @@ describe('UserConnectionDO', () => { event: 'sessions.heartbeat', data: { connectionId: 'cli-1' }, }); - expect(otherWeb.send).not.toHaveBeenCalled(); + expect(otherWeb.send).toHaveBeenCalledTimes(1); + expect(parseSent(otherWeb)).toMatchObject({ + type: 'system', + event: 'sessions.heartbeat', + data: { connectionId: 'cli-1' }, + }); + }); + + it('delivers one heartbeat per web socket (delivery count equals ws count)', () => { + const { doInstance, mockCtx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + const web1 = addWebSocket(mockCtx, 'web-1'); + const web2 = addWebSocket(mockCtx, 'web-2'); + const web3 = addWebSocket(mockCtx, 'web-3'); + + // No subscriptions — the broadcast must still hit every web socket. + web1.send.mockClear(); + web2.send.mockClear(); + web3.send.mockClear(); + + sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); + + // One heartbeat frame per active web socket. + expect(web1.send).toHaveBeenCalledTimes(1); + expect(web2.send).toHaveBeenCalledTimes(1); + expect(web3.send).toHaveBeenCalledTimes(1); + + // The delivered payload is the same shape (including the connectionId). + for (const ws of [web1, web2, web3]) { + const sent = parseSent(ws) as { data: { connectionId: string; sessions: unknown[] } }; + expect(sent).toMatchObject({ + type: 'system', + event: 'sessions.heartbeat', + data: { connectionId: 'cli-1', sessions: [{ id: 's1' }] }, + }); + } }); - it('forwards the CLI-reported protocolVersion to subscribed web clients', () => { + it('forwards the CLI-reported protocolVersion to every web socket', () => { const { doInstance, mockCtx } = setup(); const cliWs = addCliSocket(mockCtx, 'cli-1'); const webWs = addWebSocket(mockCtx, 'web-1'); @@ -560,36 +604,50 @@ describe('UserConnectionDO', () => { expect(sent.data).not.toHaveProperty('protocolVersion'); }); - it('sends heartbeat to subscribers of removed sessions', () => { + it('broadcasts removed-session information to every web socket (no subscriber special-case)', () => { const { doInstance, mockCtx } = setup(); const cliWs = addCliSocket(mockCtx, 'cli-1'); - const webWs = addWebSocket(mockCtx, 'web-1'); + // subWeb subscribed to s1, otherWeb is unrelated — both must learn s1 is gone. + const subWeb = addWebSocket(mockCtx, 'web-sub'); + const otherWeb = addWebSocket(mockCtx, 'web-other'); // cli1 owns s1 sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); - sendSubscribe(doInstance, webWs, 's1'); - webWs.send.mockClear(); + sendSubscribe(doInstance, subWeb, 's1'); + subWeb.send.mockClear(); + otherWeb.send.mockClear(); - // s1 disappears from heartbeat — subscriber should still get the heartbeat + // s1 disappears from heartbeat — every web socket gets a heartbeat with sessions:[]. sendHeartbeat(doInstance, cliWs, []); - expect(webWs.send).toHaveBeenCalledTimes(1); - expect(parseSent(webWs)).toMatchObject({ + expect(subWeb.send).toHaveBeenCalledTimes(1); + expect(parseSent(subWeb)).toMatchObject({ + type: 'system', + event: 'sessions.heartbeat', + data: { connectionId: 'cli-1', sessions: [] }, + }); + expect(otherWeb.send).toHaveBeenCalledTimes(1); + expect(parseSent(otherWeb)).toMatchObject({ type: 'system', event: 'sessions.heartbeat', data: { connectionId: 'cli-1', sessions: [] }, }); }); - it('does not send heartbeat to unsubscribed web clients', () => { + it('delivers heartbeat to web sockets that are not subscribed to anything', () => { const { doInstance, mockCtx } = setup(); const cliWs = addCliSocket(mockCtx, 'cli-1'); const webWs = addWebSocket(mockCtx, 'web-1'); - // webWs is not subscribed to anything + // webWs has no subscriptions sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); - expect(webWs.send).not.toHaveBeenCalled(); + expect(webWs.send).toHaveBeenCalledTimes(1); + expect(parseSent(webWs)).toMatchObject({ + type: 'system', + event: 'sessions.heartbeat', + data: { connectionId: 'cli-1', sessions: [{ id: 's1' }] }, + }); }); it('schedules stale alarm on heartbeat', () => { @@ -880,6 +938,7 @@ describe('UserConnectionDO', () => { const webWs = addWebSocket(mockCtx, 'web-1'); sendHeartbeat(doInstance, cliWs, [makeSession('s1', 'busy', 'Fix bug')]); + webWs.send.mockClear(); sendSubscribe(doInstance, webWs, 's1'); expect(parseSent(webWs)).toEqual({ @@ -1559,6 +1618,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'list_models', @@ -1593,6 +1653,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'list_models', @@ -1676,6 +1737,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); for (let index = 0; index < 128; index++) { sendCommand(doInstance, webWs, { id: `cmd-${index}`, @@ -1816,6 +1878,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, staleOwner, []); currentOwner.send.mockClear(); staleOwner.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', @@ -1951,6 +2014,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'eval', sessionId: 's1' }); @@ -1973,6 +2037,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', @@ -2023,6 +2088,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'eval', sessionId: 's1' }); const sent = parseSent(webWs) as Record<string, unknown>; @@ -2047,6 +2113,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, staleOwner, []); currentOwner.send.mockClear(); staleOwner.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', @@ -2081,6 +2148,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'list_commands', @@ -2476,6 +2544,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'send_command', @@ -2516,6 +2585,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, []); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'create_session', @@ -2622,6 +2692,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'exit_cli', @@ -2681,6 +2752,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, staleOwner, []); currentOwner.send.mockClear(); staleOwner.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', @@ -2728,6 +2800,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'exit_cli', @@ -2756,6 +2829,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('s1')]); cliWs.send.mockClear(); + webWs.send.mockClear(); sendCommand(doInstance, webWs, { id: 'cmd-1', command: 'exit_cli', @@ -2968,15 +3042,13 @@ describe('UserConnectionDO', () => { // ------------------------------------------------------------------------- describe('broadcast resilience', () => { - it('one closed socket does not abort send to other subscribers', () => { + it('one closed socket does not abort send to other web sockets', () => { const { doInstance, mockCtx } = setup(); const cliWs = addCliSocket(mockCtx, 'cli-1'); const failWeb = addWebSocket(mockCtx, 'web-fail'); const okWeb = addWebSocket(mockCtx, 'web-ok'); - // Both subscribe to s1 so they receive heartbeats - sendSubscribe(doInstance, failWeb, 's1'); - sendSubscribe(doInstance, okWeb, 's1'); + // Both web sockets receive heartbeats via broadcast (no subscription needed). failWeb.send.mockClear(); okWeb.send.mockClear(); @@ -3338,6 +3410,43 @@ describe('UserConnectionDO', () => { }); }); + // ------------------------------------------------------------------------- + // Owner-unique active sessions (W-followup) + // ------------------------------------------------------------------------- + + describe('owner-unique active sessions', () => { + it('emits owner-unique rows: ownership transfer with both CLIs live yields exactly one row under the new owner', () => { + const { doInstance, ctx, mockCtx } = setup(); + const oldOwner = addCliSocket(mockCtx, 'cli-old'); + const newOwner = addCliSocket(mockCtx, 'cli-new'); + + // cli-old claims the session + sendHeartbeat(doInstance, oldOwner, [makeSession('ses_transfer', 'busy', 'Transfer me')]); + + // cli-new also claims the same session id while cli-old is still connected. + // The DO routes the session to the new owner (sessionOwners.get === 'cli-new'). + sendHeartbeat(doInstance, newOwner, [makeSession('ses_transfer', 'busy', 'Transfer me')]); + + // Both CLIs are still live sockets — the snapshot should see them both. + expect(ctx.getWebSockets('cli').map(ws => ws.deserializeAttachment())).toEqual([ + expect.objectContaining({ role: 'cli', connectionId: 'cli-old' }), + expect.objectContaining({ role: 'cli', connectionId: 'cli-new' }), + ]); + + const result = doInstance.getActiveSessions(); + + // Exactly one row for the transferred session id, under the new owner. + expect(result).toEqual([ + { + id: 'ses_transfer', + status: 'busy', + title: 'Transfer me', + connectionId: 'cli-new', + }, + ]); + }); + }); + // ------------------------------------------------------------------------- // Edge cases // ------------------------------------------------------------------------- diff --git a/services/session-ingest/src/dos/UserConnectionDO.ts b/services/session-ingest/src/dos/UserConnectionDO.ts index 05133d423c..fe24f15a4c 100644 --- a/services/session-ingest/src/dos/UserConnectionDO.ts +++ b/services/session-ingest/src/dos/UserConnectionDO.ts @@ -487,29 +487,14 @@ export class UserConnectionDO extends DurableObject<Env> { }; ws.serializeAttachment(updatedAttachment); - // Send heartbeat only to web clients subscribed to sessions from this connection. - // Include subscribers for just-removed sessions so they learn the session is gone. - const subscribers = new Set<WebSocket>(); - for (const session of sessions) { - const subs = this.webSubscriptions.get(session.id); - if (subs) for (const ws2 of subs) subscribers.add(ws2); - } - for (const prev of previousSessions) { - if (!currentIds.has(prev.id)) { - const subs = this.webSubscriptions.get(prev.id); - if (subs) for (const ws2 of subs) subscribers.add(ws2); - } - } - if (subscribers.size > 0) { - const msg: WebInboundMessage = { - type: 'system', - event: 'sessions.heartbeat', - data: { connectionId, protocolVersion, capabilities, sessions }, - }; - for (const ws2 of subscribers) { - this.sendToWeb(ws2, msg); - } - } + // Broadcast the heartbeat to every one of the user's web sockets. Subscribers + // and non-subscribers both receive it: a removed session id is detectable + // from its absence in the payload, so no subscriber special-case is needed. + this.broadcastToWeb({ + type: 'system', + event: 'sessions.heartbeat', + data: { connectionId, protocolVersion, capabilities, sessions }, + }); this.sendToCli(ws, { type: 'heartbeat_ack' }); } @@ -1175,6 +1160,10 @@ export class UserConnectionDO extends DurableObject<Env> { const capabilities = this.connectionCapabilities.get(connectionId); for (const session of sessions) { if (session.parentSessionId) continue; + // Owner-unique: only emit a row for a session id under its current owner, + // so a session that has transferred owners while both CLIs are still + // connected does not appear twice in the snapshot. + if (this.sessionOwners.get(session.id) !== connectionId) continue; result.push({ ...session, connectionId,