diff --git a/app/src/components/skills/SkillsExplorerTab.tsx b/app/src/components/skills/SkillsExplorerTab.tsx index a489f21b1..216b3c5da 100644 --- a/app/src/components/skills/SkillsExplorerTab.tsx +++ b/app/src/components/skills/SkillsExplorerTab.tsx @@ -529,6 +529,9 @@ export default function SkillsExplorerTab({ onToast }: SkillsExplorerTabProps) { const [catalogEntries, setCatalogEntries] = useState([]); const [catalogTotal, setCatalogTotal] = useState(0); + // How many catalog entries are currently revealed. We fetch the whole list + // up front, then page through it client-side via the "Show more" control. + const [visibleCount, setVisibleCount] = useState(CATALOG_PAGE_SIZE); const [catalogLoading, setCatalogLoading] = useState(false); const [catalogError, setCatalogError] = useState(null); const [catalogInitialized, setCatalogInitialized] = useState(false); @@ -610,7 +613,10 @@ export default function SkillsExplorerTab({ onToast }: SkillsExplorerTabProps) { } log('fetchCatalog: total=%d', entries.length); setCatalogTotal(entries.length); - setCatalogEntries(entries.slice(0, CATALOG_PAGE_SIZE)); + // Keep the full list so "Show more" can page through it without another + // RPC; only a window of it is rendered (see displayedCatalog). + setCatalogEntries(entries); + setVisibleCount(CATALOG_PAGE_SIZE); setCatalogInitialized(true); } catch (err) { const msg = err instanceof Error ? err.message : String(err); @@ -678,7 +684,7 @@ export default function SkillsExplorerTab({ onToast }: SkillsExplorerTabProps) { // When multiple sources are active (but not all), do client-side filtering // on the already-fetched results since the RPC only supports single source filter. - const displayedCatalog = useMemo(() => { + const filteredCatalog = useMemo(() => { if ( activeSources.size === 0 || activeSources.size >= sources.length || @@ -689,6 +695,13 @@ export default function SkillsExplorerTab({ onToast }: SkillsExplorerTabProps) { return catalogEntries.filter(e => activeSources.has(e.source)); }, [catalogEntries, activeSources, sources.length]); + // Client-side pagination window: we already hold the full fetched list, so + // "Show more" reveals the next page instantly with no extra RPC. + const displayedCatalog = useMemo( + () => filteredCatalog.slice(0, visibleCount), + [filteredCatalog, visibleCount] + ); + const handleInstalled = useCallback( (result: InstallWorkflowFromUrlResult) => { log('handleInstalled: newSkills=%d', result.newWorkflows.length); @@ -966,7 +979,7 @@ export default function SkillsExplorerTab({ onToast }: SkillsExplorerTabProps) { {/* ── Registry view ── */} {view === 'registry' && !loading && !error && ( <> - {catalogInitialized && catalogEntries.length === 0 && ( + {catalogInitialized && filteredCatalog.length === 0 && ( ))} - {catalogTotal > displayedCatalog.length && ( -

- {t('skills.explorer.showingOf') - .replace('{shown}', String(displayedCatalog.length)) - .replace('{total}', catalogTotal.toLocaleString()) || - `Showing ${displayedCatalog.length} of ${catalogTotal.toLocaleString()} results. Refine your search to see more.`} -

+ {filteredCatalog.length > displayedCatalog.length && ( +
+ +

+ {displayedCatalog.length.toLocaleString()} /{' '} + {filteredCatalog.length.toLocaleString()} +

+
)} )} diff --git a/app/src/components/skills/__tests__/SkillsExplorerTab.test.tsx b/app/src/components/skills/__tests__/SkillsExplorerTab.test.tsx index a0ac3ee27..82905c035 100644 --- a/app/src/components/skills/__tests__/SkillsExplorerTab.test.tsx +++ b/app/src/components/skills/__tests__/SkillsExplorerTab.test.tsx @@ -134,6 +134,42 @@ describe('SkillsExplorerTab', () => { expect(screen.getByText('built-in')).toBeInTheDocument(); }); + it('paginates the registry catalog via the Show more control', async () => { + const { workflowsApi } = await import('../../../services/api/workflowsApi'); + const { skillRegistryApi } = await import('../../../services/api/skillRegistryApi'); + vi.mocked(workflowsApi.listWorkflows).mockResolvedValue([]); + const entries: CatalogEntry[] = Array.from({ length: 130 }, (_, i) => ({ + ...MOCK_CATALOG_ENTRY, + id: `paged-skill-${i}`, + name: `Paged Skill ${i}`, + })); + vi.mocked(skillRegistryApi.browse).mockResolvedValue(entries); + + const { container } = render(); + + await waitFor(() => { + expect(screen.getByText('Paged Skill 0')).toBeInTheDocument(); + }); + + const tileCount = () => + container.querySelectorAll('[data-testid^="registry-tile-"]').length; + + // First page only: 60 of 130 revealed. + expect(tileCount()).toBe(60); + + await act(async () => { + fireEvent.click(screen.getByTestId('registry-show-more')); + }); + expect(tileCount()).toBe(120); + + await act(async () => { + fireEvent.click(screen.getByTestId('registry-show-more')); + }); + // All 130 revealed → the control disappears. + expect(tileCount()).toBe(130); + expect(screen.queryByTestId('registry-show-more')).toBeNull(); + }); + it('searches catalog via RPC when typing in search box', async () => { const { workflowsApi } = await import('../../../services/api/workflowsApi'); const { skillRegistryApi } = await import('../../../services/api/skillRegistryApi'); diff --git a/app/src/lib/ai/skillToolChainLatency.test.ts b/app/src/lib/ai/skillToolChainLatency.test.ts new file mode 100644 index 000000000..dd96534df --- /dev/null +++ b/app/src/lib/ai/skillToolChainLatency.test.ts @@ -0,0 +1,98 @@ +import { describe, expect, it } from 'vitest'; + +import { + createSkillToolChainLatencyTracker, + SKILL_TOOL_CHAIN_TARGET_MS, +} from './skillToolChainLatency'; + +/** A clock whose value we advance by hand, so latency is deterministic. */ +function fakeClock(start = 0) { + let t = start; + return { + now: () => t, + advance: (ms: number) => { + t += ms; + }, + }; +} + +describe('skillToolChainLatency', () => { + it('returns null when a turn finished without any tool calls', () => { + const tracker = createSkillToolChainLatencyTracker(() => 0); + expect(tracker.finishChain('t1')).toBeNull(); + }); + + it('measures elapsed time from the first tool call to completion', () => { + const clock = fakeClock(); + const tracker = createSkillToolChainLatencyTracker(clock.now); + + tracker.noteToolCall('t1'); // chain starts at 0 + clock.advance(1_500); + tracker.noteToolCall('t1'); // second tool, still same chain + clock.advance(500); + + const m = tracker.finishChain('t1'); + expect(m).not.toBeNull(); + expect(m!.elapsedMs).toBe(2_000); + expect(m!.toolCount).toBe(2); + expect(m!.withinTarget).toBe(true); + expect(m!.ok).toBe(true); + }); + + it('flags a chain that overruns the 60s target', () => { + const clock = fakeClock(); + const tracker = createSkillToolChainLatencyTracker(clock.now); + + tracker.noteToolCall('t1'); + clock.advance(SKILL_TOOL_CHAIN_TARGET_MS + 1); + + const m = tracker.finishChain('t1'); + expect(m!.withinTarget).toBe(false); + expect(m!.elapsedMs).toBe(SKILL_TOOL_CHAIN_TARGET_MS + 1); + }); + + it('treats exactly the target boundary as within target', () => { + const clock = fakeClock(); + const tracker = createSkillToolChainLatencyTracker(clock.now); + tracker.noteToolCall('t1'); + clock.advance(SKILL_TOOL_CHAIN_TARGET_MS); + expect(tracker.finishChain('t1')!.withinTarget).toBe(true); + }); + + it('marks a chain ended by chat_error as not ok', () => { + const tracker = createSkillToolChainLatencyTracker(() => 0); + tracker.noteToolCall('t1'); + const m = tracker.finishChain('t1', { ok: false }); + expect(m!.ok).toBe(false); + }); + + it('tracks chains on separate threads independently', () => { + const clock = fakeClock(); + const tracker = createSkillToolChainLatencyTracker(clock.now); + + tracker.noteToolCall('a'); // a starts at 0 + clock.advance(1_000); + tracker.noteToolCall('b'); // b starts at 1000 + clock.advance(1_000); + + const a = tracker.finishChain('a'); + const b = tracker.finishChain('b'); + expect(a!.elapsedMs).toBe(2_000); + expect(b!.elapsedMs).toBe(1_000); + }); + + it('does not double-count after a chain is finished', () => { + const tracker = createSkillToolChainLatencyTracker(() => 0); + tracker.noteToolCall('t1'); + expect(tracker.finishChain('t1')!.toolCount).toBe(1); + // Same thread, fresh chain — must not resurrect the old state. + expect(tracker.finishChain('t1')).toBeNull(); + }); + + it('reset drops all in-flight chains', () => { + const tracker = createSkillToolChainLatencyTracker(() => 0); + tracker.noteToolCall('t1'); + tracker.reset(); + expect(tracker.finishChain('t1')).toBeNull(); + }); +}); diff --git a/app/src/lib/ai/skillToolChainLatency.ts b/app/src/lib/ai/skillToolChainLatency.ts new file mode 100644 index 000000000..c0a8c0564 --- /dev/null +++ b/app/src/lib/ai/skillToolChainLatency.ts @@ -0,0 +1,94 @@ +/** + * Skill tool-chain latency observability (issue #4273, AC3). + * + * The acceptance criterion is that a complex skill tool chain "completes within + * 60 seconds". We deliberately do NOT enforce that as a hard cap — clamping the + * existing 120s tool timeout down to 60s would be a breaking behaviour change + * that could cut off legitimately long chains. Instead we *measure* the wall + * clock from the first tool call of a turn to the turn's completion and flag any + * chain that overruns the target, so the budget is observable in logs/telemetry + * and regressions surface without changing runtime behaviour. + * + * This module is pure and side-effect free (it takes an injectable clock) so it + * can be unit-tested without sockets. `ChatRuntimeProvider` owns the single + * instance and feeds it the canonical `tool_call` / `chat_done` / `chat_error` + * events it already subscribes to. + */ + +/** The AC3 budget: complex skill tool chains should finish within this. */ +export const SKILL_TOOL_CHAIN_TARGET_MS = 60_000; + +export interface ToolChainMeasurement { + /** Thread the chain ran on. */ + threadId: string; + /** Wall-clock from the first tool call to turn completion, in ms. */ + elapsedMs: number; + /** Number of tool calls observed during the chain — counts every call, not + * unique tool names (the tracker is keyed only by chain id, not by tool). */ + toolCount: number; + /** Whether the chain finished within {@link SKILL_TOOL_CHAIN_TARGET_MS}. */ + withinTarget: boolean; + /** False when the turn ended via `chat_error` rather than `chat_done`. */ + ok: boolean; +} + +export interface SkillToolChainLatencyTracker { + /** + * Record a tool call on a thread. The first call starts the chain timer; each + * subsequent call increments the tool count. No-op shape until the next + * `finishChain` resets the thread's state. + */ + noteToolCall(threadId: string): void; + /** + * Close out a thread's chain and return its measurement, or `null` if no tool + * call was ever seen for the thread (a pure text turn is not a "tool chain"). + */ + finishChain(threadId: string, opts?: { ok?: boolean }): ToolChainMeasurement | null; + /** Drop all in-flight chain state (e.g. on provider unmount). */ + reset(): void; +} + +interface ChainState { + startedAt: number; + toolCount: number; +} + +/** + * Create a tracker. `now` and `targetMs` are injectable for tests; production + * uses `Date.now()` and {@link SKILL_TOOL_CHAIN_TARGET_MS}. + */ +export function createSkillToolChainLatencyTracker( + now: () => number = () => Date.now(), + targetMs: number = SKILL_TOOL_CHAIN_TARGET_MS +): SkillToolChainLatencyTracker { + const chains = new Map(); + + return { + noteToolCall(threadId: string): void { + const existing = chains.get(threadId); + if (existing) { + existing.toolCount += 1; + return; + } + chains.set(threadId, { startedAt: now(), toolCount: 1 }); + }, + + finishChain(threadId: string, opts?: { ok?: boolean }): ToolChainMeasurement | null { + const state = chains.get(threadId); + if (!state) return null; + chains.delete(threadId); + const elapsedMs = Math.max(0, now() - state.startedAt); + return { + threadId, + elapsedMs, + toolCount: state.toolCount, + withinTarget: elapsedMs <= targetMs, + ok: opts?.ok ?? true, + }; + }, + + reset(): void { + chains.clear(); + }, + }; +} diff --git a/app/src/lib/composio/connectionCache.test.ts b/app/src/lib/composio/connectionCache.test.ts new file mode 100644 index 000000000..d983a872e --- /dev/null +++ b/app/src/lib/composio/connectionCache.test.ts @@ -0,0 +1,153 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { clearConnectionCache, readConnectionCache, writeConnectionCache } from './connectionCache'; +import type { ComposioConnection, ComposioToolkitCatalogEntry } from './types'; + +const ACTIVE_USER_KEY = 'OPENHUMAN_ACTIVE_USER_ID'; + +const conn = (toolkit: string, status = 'ACTIVE'): ComposioConnection => ({ + id: `c-${toolkit}`, + toolkit, + status, +}); +const cat = (slug: string): ComposioToolkitCatalogEntry => ({ slug, name: slug }); + +describe('connectionCache', () => { + beforeEach(() => { + window.localStorage.clear(); + // The module keeps an in-memory mirror; clear it between tests by evicting + // for whatever user id a prior test may have left active. + clearConnectionCache(); + window.localStorage.setItem(ACTIVE_USER_KEY, 'user-a'); + clearConnectionCache(); + }); + + afterEach(() => { + vi.useRealTimers(); + window.localStorage.clear(); + }); + + it('returns null when nothing is cached', () => { + expect(readConnectionCache()).toBeNull(); + }); + + it('round-trips a written snapshot', () => { + writeConnectionCache({ + connections: [conn('gmail'), conn('notion')], + toolkits: ['gmail', 'notion'], + catalog: [cat('gmail'), cat('notion')], + }); + + const got = readConnectionCache(); + expect(got?.connections.map(c => c.toolkit)).toEqual(['gmail', 'notion']); + expect(got?.toolkits).toEqual(['gmail', 'notion']); + expect(got?.catalog.map(c => c.slug)).toEqual(['gmail', 'notion']); + }); + + it('merges a connections-only patch without clobbering toolkit/catalog', () => { + writeConnectionCache({ + connections: [conn('gmail')], + toolkits: ['gmail', 'github'], + catalog: [cat('gmail'), cat('github')], + }); + + // Poll-style refresh: only connections supplied. + writeConnectionCache({ connections: [conn('gmail'), conn('github')] }); + + const got = readConnectionCache(); + expect(got?.connections.map(c => c.toolkit)).toEqual(['gmail', 'github']); + // Toolkit allowlist + catalog survive the partial write. + expect(got?.toolkits).toEqual(['gmail', 'github']); + expect(got?.catalog.map(c => c.slug)).toEqual(['gmail', 'github']); + }); + + it('clears the cached snapshot', () => { + writeConnectionCache({ connections: [conn('gmail')] }); + expect(readConnectionCache()).not.toBeNull(); + + clearConnectionCache(); + expect(readConnectionCache()).toBeNull(); + }); + + it('expires snapshots older than the TTL', () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-01-01T00:00:00Z')); + writeConnectionCache({ connections: [conn('gmail')] }); + expect(readConnectionCache()).not.toBeNull(); + + // Advance just past the 24h TTL; the in-memory mirror must also respect it. + vi.setSystemTime(new Date('2026-01-02T00:00:01Z')); + expect(readConnectionCache()).toBeNull(); + }); + + it('namespaces the cache per active user', () => { + writeConnectionCache({ connections: [conn('gmail')] }); + expect(readConnectionCache()?.connections).toHaveLength(1); + + // Identity flip to user-b — must not see user-a's connections. + window.localStorage.setItem(ACTIVE_USER_KEY, 'user-b'); + expect(readConnectionCache()).toBeNull(); + + writeConnectionCache({ connections: [conn('slack'), conn('github')] }); + expect(readConnectionCache()?.connections.map(c => c.toolkit)).toEqual(['slack', 'github']); + + // Flipping back to user-a restores their original snapshot from disk. + window.localStorage.setItem(ACTIVE_USER_KEY, 'user-a'); + expect(readConnectionCache()?.connections.map(c => c.toolkit)).toEqual(['gmail']); + }); + + it('returns null on malformed persisted JSON', () => { + window.localStorage.setItem('user-a:composio:connections:v1', '{ not json'); + expect(readConnectionCache()).toBeNull(); + }); + + it('returns null when the persisted shape is invalid', () => { + window.localStorage.setItem( + 'user-a:composio:connections:v1', + JSON.stringify({ fetchedAt: Date.now(), connections: 'nope' }) + ); + expect(readConnectionCache()).toBeNull(); + }); + + it('namespaces the cache under the user prefix so clearAllAppData purges it', () => { + writeConnectionCache({ connections: [conn('gmail')] }); + // clearAllAppData removes every `${userId}:` key — the cache must live there. + expect(window.localStorage.getItem('user-a:composio:connections:v1')).not.toBeNull(); + }); + + it('does not revive expired toolkit/catalog via a later connections-only write', () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-01-01T00:00:00Z')); + writeConnectionCache({ + connections: [conn('gmail')], + toolkits: ['gmail'], + catalog: [cat('gmail')], + }); + + // Past the TTL: the snapshot is stale. A connections-only poll write must + // start from an empty base, not resurrect the expired toolkits/catalog. + vi.setSystemTime(new Date('2026-01-02T00:00:01Z')); + writeConnectionCache({ connections: [conn('gmail'), conn('notion')] }); + + const got = readConnectionCache(); + expect(got?.connections.map(c => c.toolkit)).toEqual(['gmail', 'notion']); + expect(got?.toolkits).toEqual([]); + expect(got?.catalog).toEqual([]); + }); + + it('survives a simulated process restart (warm localStorage, cold module memory)', async () => { + writeConnectionCache({ + connections: [conn('gmail')], + toolkits: ['gmail'], + catalog: [cat('gmail')], + }); + + // Drop the module-level in-memory mirror but keep the durable localStorage + // blob, exactly like a cold app restart, then re-import the module fresh. + vi.resetModules(); + const fresh = await import('./connectionCache'); + const got = fresh.readConnectionCache(); + expect(got?.connections.map(c => c.toolkit)).toEqual(['gmail']); + expect(got?.toolkits).toEqual(['gmail']); + }); +}); diff --git a/app/src/lib/composio/connectionCache.ts b/app/src/lib/composio/connectionCache.ts new file mode 100644 index 000000000..32e5e8b39 --- /dev/null +++ b/app/src/lib/composio/connectionCache.ts @@ -0,0 +1,172 @@ +/** + * Client-side cache for the Composio connection ("activation") state. + * + * Composio connections are backend-managed: `useComposioIntegrations` fetches + * them fresh on every mount and reconciles via a 5s poll. That makes the + * source of truth durable, but it also means a cold app restart shows an empty + * loading skeleton until the first round-trip lands — even though the user + * already authorised those toolkits. To an activation flow that is supposed to + * "survive an app restart" (issue #4273, AC1) that reads as a regression: the + * Skills page momentarily looks disconnected. + * + * This module is the durable half of that flow. It mirrors the sibling + * `catalogCache.ts` (synchronous `localStorage` + in-memory mirror) so the hook + * can seed its initial state instantly on mount and render the last-known + * connected toolkits with no flash, then reconcile against the live backend + * fetch. It is intentionally NOT the source of truth — a stale cache is always + * corrected by the very next fetch/poll a few seconds later. + * + * The cache key is namespaced by the active user id (the same id + * `userScopedStorage` uses) so user A's connections never bleed into user B's + * session after an identity flip (#900). + * + * `clearConnectionCache()` drops both tiers — call it when the Composio client + * identity changes (backend ↔ direct mode, BYO API key) or whenever the live + * state is known to be "no connections", exactly like the existing + * `composio:config-changed` refresh path does for the catalog. + */ +import type { ComposioConnection, ComposioToolkitCatalogEntry } from './types'; + +const CACHE_KEY_SUFFIX = 'composio:connections:v1'; +/** + * Bound how stale a seeded snapshot may be. The value only gates the *initial + * paint* — the hook always issues a live fetch on mount regardless — so this is + * a safety net against rendering very old state if the backend is unreachable, + * not a freshness contract. + */ +const TTL_MS = 24 * 60 * 60 * 1000; + +const ACTIVE_USER_KEY = 'OPENHUMAN_ACTIVE_USER_ID'; + +export interface CachedConnectionState { + fetchedAt: number; + connections: ComposioConnection[]; + toolkits: string[]; + catalog: ComposioToolkitCatalogEntry[]; +} + +/** In-memory mirror so repeated reads on the hot path skip a `JSON.parse`. */ +let memory: CachedConnectionState | null = null; +/** The user id `memory` belongs to, so a mid-session flip re-reads from disk. */ +let memoryUserId: string | null = null; + +/** + * The active user id, or `null` when no user is identified yet (pre-login / + * mid identity-flip). When `null` the cache is fully inert — reads return + * `null` and writes are no-ops — mirroring `userScopedStorage`'s signed-out + * behaviour so two unidentified sessions can never share a blob and leak one + * user's Composio connections into another's shell. + */ +function activeUserId(): string | null { + try { + const id = window.localStorage.getItem(ACTIVE_USER_KEY); + return id && id.length > 0 ? id : null; + } catch { + return null; + } +} + +function cacheKey(userId: string): string { + // User id FIRST so the key matches userScopedStorage's `${userId}:...` shape. + // clearAllAppData's purge removes every `${userId}:` key, so this namespacing + // ensures "clear my data" also drops the connection cache instead of leaving + // it to re-seed stale connected toolkits on the next sign-in (PR #4288). + return `${userId}:${CACHE_KEY_SUFFIX}`; +} + +/** A cached snapshot is usable only while still inside the TTL window. */ +function isFresh(entry: CachedConnectionState): boolean { + return Date.now() - entry.fetchedAt < TTL_MS; +} + +function isValid(parsed: unknown): parsed is CachedConnectionState { + if (!parsed || typeof parsed !== 'object') return false; + const entry = parsed as Partial; + return ( + typeof entry.fetchedAt === 'number' && + Array.isArray(entry.connections) && + Array.isArray(entry.toolkits) && + Array.isArray(entry.catalog) + ); +} + +/** + * Read the last-known connection snapshot for the active user, or `null` when + * absent / malformed / expired. Cheap on repeat calls thanks to the in-memory + * mirror; the mirror is dropped automatically when the active user changes. + */ +export function readConnectionCache(): CachedConnectionState | null { + const userId = activeUserId(); + // No identified user → cache is inert so two unidentified sessions can never + // share a blob and leak one user's connections into another's shell. + if (userId === null) return null; + if (memory && memoryUserId === userId) { + return isFresh(memory) ? memory : null; + } + try { + const raw = window.localStorage.getItem(cacheKey(userId)); + if (!raw) return null; + const parsed: unknown = JSON.parse(raw); + if (!isValid(parsed)) return null; + // Never promote an expired snapshot into the in-memory mirror: a later + // partial write (e.g. a connections-only poll after a failed toolkit fetch) + // would otherwise merge against — and revive — stale toolkit/catalog data + // the TTL was meant to suppress (PR #4288). + if (!isFresh(parsed)) return null; + memory = parsed; + memoryUserId = userId; + return parsed; + } catch { + return null; + } +} + +/** + * Persist (merge) the latest known connection state. Any field omitted from + * `patch` keeps its previously cached value, so the 5s poll can refresh just + * `connections` without clobbering the toolkit allowlist / catalog. `fetchedAt` + * is bumped on every write so the freshest poll wins the next cold paint. + */ +export function writeConnectionCache(patch: { + connections?: ComposioConnection[]; + toolkits?: string[]; + catalog?: ComposioToolkitCatalogEntry[]; +}): void { + const userId = activeUserId(); + // Inert until a user is identified — mirrors userScopedStorage's signed-out + // no-op so we never write a user-shaped blob to a shared key. + if (userId === null) return; + const base = + memory && memoryUserId === userId && isFresh(memory) + ? memory + : { fetchedAt: 0, connections: [], toolkits: [], catalog: [] }; + const entry: CachedConnectionState = { + fetchedAt: Date.now(), + connections: patch.connections ?? base.connections, + toolkits: patch.toolkits ?? base.toolkits, + catalog: patch.catalog ?? base.catalog, + }; + memory = entry; + memoryUserId = userId; + try { + window.localStorage.setItem(cacheKey(userId), JSON.stringify(entry)); + } catch { + // Private-mode / quota errors are non-fatal — the in-memory mirror still + // serves this session. + } +} + +/** Drop both cache tiers for the active user so the next read re-fetches. */ +export function clearConnectionCache(): void { + const userId = activeUserId(); + // Always drop the in-memory mirror; only the keyed localStorage entry needs a + // user id (nothing to remove when none is set). + memory = null; + memoryUserId = null; + if (userId === null) return; + try { + window.localStorage.removeItem(cacheKey(userId)); + } catch { + // ignore + } +} diff --git a/app/src/lib/composio/hooks.test.ts b/app/src/lib/composio/hooks.test.ts index 1db118a52..9104d6dde 100644 --- a/app/src/lib/composio/hooks.test.ts +++ b/app/src/lib/composio/hooks.test.ts @@ -98,6 +98,60 @@ describe('useComposioIntegrations', () => { expect(result.current.catalogByToolkit.size).toBe(0); }); + it('seeds initial state from the durable connection cache for an instant cold-start paint', async () => { + // Pre-populate the durable cache as if a prior session had persisted it, + // under the active user's namespace (the cache is inert without a user id). + window.localStorage.setItem('OPENHUMAN_ACTIVE_USER_ID', 'test-user'); + window.localStorage.setItem( + 'test-user:composio:connections:v1', + JSON.stringify({ + fetchedAt: Date.now(), + connections: [{ id: 'seed', toolkit: 'gmail', status: 'ACTIVE' }], + toolkits: ['gmail'], + catalog: [{ slug: 'gmail', name: 'Gmail' }], + }) + ); + + // The live fetch hangs so we can observe the seeded state in isolation. + mockListToolkits.mockReturnValue(new Promise(() => {})); + mockListConnections.mockReturnValue(new Promise(() => {})); + + const { useComposioIntegrations } = await import('./hooks'); + const { result } = renderHook(() => useComposioIntegrations(0)); + + // No loading skeleton — the cached snapshot paints immediately. + expect(result.current.loading).toBe(false); + expect(result.current.connectionByToolkit.get('gmail')?.status).toBe('ACTIVE'); + expect(result.current.toolkits).toEqual(['gmail']); + expect(result.current.catalogByToolkit.get('gmail')?.name).toBe('Gmail'); + }); + + it('reconciles a stale cached connection away once the live fetch lands', async () => { + window.localStorage.setItem('OPENHUMAN_ACTIVE_USER_ID', 'test-user'); + window.localStorage.setItem( + 'test-user:composio:connections:v1', + JSON.stringify({ + fetchedAt: Date.now(), + connections: [{ id: 'seed', toolkit: 'gmail', status: 'ACTIVE' }], + toolkits: ['gmail'], + catalog: [], + }) + ); + // Backend now reports the toolkit disconnected. + mockListToolkits.mockResolvedValue({ toolkits: ['gmail'] }); + mockListConnections.mockResolvedValue({ connections: [] }); + + const { useComposioIntegrations } = await import('./hooks'); + const { result } = renderHook(() => useComposioIntegrations(0)); + + // Seeded immediately from the cache… + expect(result.current.connectionByToolkit.get('gmail')?.status).toBe('ACTIVE'); + // …then the live fetch reconciles the stale connection away. + await waitFor(() => { + expect(result.current.connectionByToolkit.size).toBe(0); + }); + }); + it('groups connections by toolkit, sorts by status then createdAt', async () => { const { useComposioIntegrations } = await import('./hooks'); diff --git a/app/src/lib/composio/hooks.ts b/app/src/lib/composio/hooks.ts index f0c2ea3bf..cb0b36eba 100644 --- a/app/src/lib/composio/hooks.ts +++ b/app/src/lib/composio/hooks.ts @@ -5,6 +5,7 @@ import { openhumanComposioGetMode } from '../../utils/tauriCommands'; import { getCoreStateSnapshot } from '../coreState/store'; import { getToolkitCatalog, invalidateToolkitCatalogCache } from './catalogCache'; import { COMPOSIO_FETCH_TIMEOUT_MS, listAgentReadyToolkits, listConnections } from './composioApi'; +import { clearConnectionCache, readConnectionCache, writeConnectionCache } from './connectionCache'; import { canonicalizeComposioToolkitSlug } from './toolkitSlug'; import type { ComposioConnection, ComposioToolkitCatalogEntry } from './types'; @@ -47,15 +48,31 @@ export interface UseComposioIntegrationsResult { */ export function useComposioIntegrations(pollIntervalMs = 5_000): UseComposioIntegrationsResult { const isLocalSession = isLocalSessionToken(getCoreStateSnapshot().snapshot.sessionToken); - const [toolkits, setToolkits] = useState([]); - const [catalog, setCatalog] = useState([]); - const [connections, setConnections] = useState([]); - const [loading, setLoading] = useState(true); + // Seed from the durable connection cache so a cold restart re-paints the + // last-known connected toolkits instantly instead of flashing an empty + // loading skeleton (#4273, AC1). The live fetch below still runs on mount and + // reconciles within a few seconds, so a stale seed is self-correcting. + const [toolkits, setToolkits] = useState(() => readConnectionCache()?.toolkits ?? []); + const [catalog, setCatalog] = useState( + () => readConnectionCache()?.catalog ?? [] + ); + const [connections, setConnections] = useState( + () => readConnectionCache()?.connections ?? [] + ); + // No skeleton when we already have a cached snapshot to show. + const [loading, setLoading] = useState(() => readConnectionCache() == null); const [error, setError] = useState(null); const [fetchEnabled, setFetchEnabled] = useState(() => isLocalSession ? null : true ); const mountedRef = useRef(true); + // Bumped whenever the Composio client identity changes (backend ↔ direct / + // BYO key) via the config-changed handler. Any refresh/poll started under an + // older generation must not commit its result — otherwise an in-flight fetch + // can repopulate the cache the handler just cleared with the previous + // tenant's connections, painting phantom activations on the next restart + // (PR #4288). + const configGenerationRef = useRef(0); useEffect(() => { mountedRef.current = true; @@ -85,8 +102,13 @@ export function useComposioIntegrations(pollIntervalMs = 5_000): UseComposioInte }, [isLocalSession]); const refresh = useCallback(async () => { + const generation = configGenerationRef.current; const enabled = fetchEnabled ?? (await resolveFetchEnabled()); if (!enabled) { + // Direct mode with no API key configured: there can be no connections, so + // drop any cached snapshot too — otherwise a removed key would still + // re-paint phantom "connected" toolkits on the next restart. + clearConnectionCache(); if (mountedRef.current) { setToolkits([]); setCatalog([]); @@ -106,7 +128,9 @@ export function useComposioIntegrations(pollIntervalMs = 5_000): UseComposioInte getToolkitCatalog({ timeoutMs: COMPOSIO_FETCH_TIMEOUT_MS }), listConnections({ timeoutMs: COMPOSIO_FETCH_TIMEOUT_MS }), ]); - if (!mountedRef.current) return; + // Drop results from a client that has since been swapped out — committing + // them would revive the previous tenant's state post-invalidation. + if (!mountedRef.current || generation !== configGenerationRef.current) return; if (toolkitsResult.status === 'fulfilled') { setToolkits(toolkitsResult.value.toolkits ?? []); @@ -121,7 +145,23 @@ export function useComposioIntegrations(pollIntervalMs = 5_000): UseComposioInte } if (connectionsResult.status === 'fulfilled') { - setConnections(connectionsResult.value.connections ?? []); + const freshConnections = connectionsResult.value.connections ?? []; + setConnections(freshConnections); + // Persist the latest activation snapshot for instant cold-start paint. + // Pair it with this round's toolkit data when that fetch also + // succeeded; otherwise leave the cached toolkit/catalog fields intact + // (writeConnectionCache merges, so `undefined` keeps the prior value). + writeConnectionCache({ + connections: freshConnections, + toolkits: + toolkitsResult.status === 'fulfilled' + ? (toolkitsResult.value.toolkits ?? []) + : undefined, + catalog: + toolkitsResult.status === 'fulfilled' + ? (toolkitsResult.value.catalog ?? []) + : undefined, + }); } else { const message = connectionsResult.reason instanceof Error @@ -142,10 +182,15 @@ export function useComposioIntegrations(pollIntervalMs = 5_000): UseComposioInte void refresh(); if (pollIntervalMs <= 0 || fetchEnabled !== true) return; const id = window.setInterval(() => { + const generation = configGenerationRef.current; void listConnections({ timeoutMs: COMPOSIO_FETCH_TIMEOUT_MS }) .then(resp => { - if (!mountedRef.current) return; - setConnections(resp.connections ?? []); + if (!mountedRef.current || generation !== configGenerationRef.current) return; + const freshConnections = resp.connections ?? []; + setConnections(freshConnections); + // Keep the durable cache current with each poll so a restart paints + // the freshest activation state (merge preserves toolkit/catalog). + writeConnectionCache({ connections: freshConnections }); }) .catch(err => { console.warn( @@ -169,10 +214,15 @@ export function useComposioIntegrations(pollIntervalMs = 5_000): UseComposioInte useEffect(() => { const onConfigChanged = () => { console.debug('[composio-cache] window:composio:config-changed → refresh()'); + // Invalidate any refresh/poll already in flight under the previous client + // so it can't write its result back after the caches below are cleared. + configGenerationRef.current += 1; // The Composio client identity changed (backend ↔ direct / BYO key), - // so the cached catalog belongs to the previous tenant. Drop it before - // refetching, mirroring the core-side ComposioConfigChanged eviction. + // so the cached catalog AND connections belong to the previous tenant. + // Drop both before refetching, mirroring the core-side + // ComposioConfigChanged eviction. invalidateToolkitCatalogCache(); + clearConnectionCache(); if (isLocalSession) { void resolveFetchEnabled().then(enabled => { if (enabled) { diff --git a/app/src/providers/ChatRuntimeProvider.tsx b/app/src/providers/ChatRuntimeProvider.tsx index 0d8675479..2950eca9e 100644 --- a/app/src/providers/ChatRuntimeProvider.tsx +++ b/app/src/providers/ChatRuntimeProvider.tsx @@ -3,6 +3,10 @@ import { useCallback, useEffect, useRef } from 'react'; import { requestUsageRefresh } from '../hooks/usageRefresh'; import { useRefetchSnapshotOnTurnEnd } from '../hooks/useRefetchSnapshotOnTurnEnd'; +import { + createSkillToolChainLatencyTracker, + SKILL_TOOL_CHAIN_TARGET_MS, +} from '../lib/ai/skillToolChainLatency'; import { ingestRuntimeErrorSignal } from '../lib/userErrors/report'; import { type ChatApprovalRequestEvent, @@ -284,6 +288,10 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { const toolTimelineRef = useRef(toolTimelineByThread); const inferenceStatusRef = useRef(inferenceStatusByThread); const streamingAssistantRef = useRef(streamingAssistantByThread); + // Measures wall-clock of each turn's tool chain against the 60s target + // (#4273, AC3). Single instance for the provider's lifetime; observability + // only — it never gates or cancels a turn. + const skillLatencyRef = useRef(createSkillToolChainLatencyTracker()); useEffect(() => { toolTimelineRef.current = toolTimelineByThread; @@ -510,6 +518,11 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { ) return; + // Start (or extend) the tool-chain latency window for this turn (#4273). + // Key by thread+request (same scheme as segment delivery) so parallel / + // forked turns that share a thread_id keep independent chains (#4288). + skillLatencyRef.current.noteToolCall(segmentDeliveryKey(event.thread_id, event.request_id)); + const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? []; const existingIdx = event.tool_call_id ? existing.findIndex(entry => entry.id === event.tool_call_id) @@ -1121,6 +1134,28 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { output_tokens: event.total_output_tokens, }); + // Close the tool-chain latency window and surface overruns of the 60s + // target (#4273, AC3). Observability only — never blocks the turn. + const latency = skillLatencyRef.current.finishChain( + segmentDeliveryKey(event.thread_id, event.request_id), + { ok: true } + ); + if (latency) { + rtLog('skill_tool_chain_latency', { + thread: event.thread_id, + request: event.request_id, + elapsed_ms: latency.elapsedMs, + tools: latency.toolCount, + within_target: latency.withinTarget ? 'true' : 'false', + }); + if (!latency.withinTarget) { + console.warn( + `[skill-latency] tool chain on thread ${event.thread_id} took ${latency.elapsedMs}ms ` + + `across ${latency.toolCount} tool(s) — exceeds the ${SKILL_TOOL_CHAIN_TARGET_MS}ms target` + ); + } + } + // Parallel (forked) turn: resolve only its own lane. The primary turn's // stream / status / lifecycle / active marker may still be running, so // we must NOT clear them here. Segmented parallel turns already @@ -1263,6 +1298,23 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { err: event.error_type, }); + // A failed turn still closes its latency window so the chain timer never + // leaks into a later turn on the same thread (#4273, AC3). + const errLatency = skillLatencyRef.current.finishChain( + segmentDeliveryKey(event.thread_id, event.request_id), + { ok: false } + ); + if (errLatency) { + rtLog('skill_tool_chain_latency', { + thread: event.thread_id, + request: event.request_id, + elapsed_ms: errLatency.elapsedMs, + tools: errLatency.toolCount, + within_target: errLatency.withinTarget ? 'true' : 'false', + ok: 'false', + }); + } + // #3931: surface expected, user-actionable provider/billing states // (insufficient BYO credits, managed-budget exhaustion) in the shell's // dedicated error panel — in ADDITION to the inline chat message below. @@ -1377,6 +1429,10 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { const threadIds = Object.keys(lifecycles); const activeThreadIds = Object.keys(state.thread.activeThreadIds); if (threadIds.length === 0 && activeThreadIds.length === 0) return; + // Abandon any in-flight tool-chain latency windows: a disconnect tears down + // these turns without an onDone/onError, so without this the next tool call + // on a reused thread would attribute stale elapsed/tool counts (#4288). + skillLatencyRef.current.reset(); rtLog('socket_disconnect_reconcile', { socket: socketStatus, inFlight: threadIds.length, diff --git a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx index baf3699de..ee62fd02c 100644 --- a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx +++ b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx @@ -1,7 +1,7 @@ import { render, waitFor } from '@testing-library/react'; import { act } from 'react'; import { Provider } from 'react-redux'; -import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import * as chatService from '../../services/chatService'; import { threadApi } from '../../services/api/threadApi'; @@ -1418,3 +1418,113 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria }); }); }); + +describe('ChatRuntimeProvider — skill tool-chain latency (#4273 AC3)', () => { + beforeEach(() => { + vi.clearAllMocks(); + resetRuntimeState(); + vi.mocked(threadApi.appendMessage).mockImplementation(async (_tid, msg) => msg); + vi.mocked(threadApi.getThreads).mockResolvedValue({ threads: [], count: 0 }); + vi.mocked(threadApi.generateTitleIfNeeded).mockResolvedValue({ + id: 'tid', + title: 'new', + } as never); + }); + + afterEach(() => { + // Restore the console.warn spy + real timers in shared cleanup so a failing + // assertion can't leave a mocked console for the next test (PR #4288). + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + const toolCall = (thread: string): chatService.ChatToolCallEvent => ({ + thread_id: thread, + request_id: 'r1', + round: 0, + tool_name: 'composio_execute', + skill_id: 'gmail', + args: {}, + tool_call_id: `${thread}-call-1`, + }); + + const done = (thread: string): chatService.ChatDoneEvent => + ({ + thread_id: thread, + request_id: 'r1', + full_response: '', + rounds_used: 1, + total_input_tokens: 0, + total_output_tokens: 0, + }) as chatService.ChatDoneEvent; + + it('warns when a tool chain overruns the 60s target', () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const listeners = renderProvider(); + + vi.useFakeTimers(); + try { + vi.setSystemTime(new Date('2026-06-29T00:00:00.000Z')); + act(() => { + listeners.onToolCall?.(toolCall('t-slow')); + }); + // 61s later — past the 60s budget. + vi.setSystemTime(new Date('2026-06-29T00:01:01.000Z')); + act(() => { + listeners.onDone?.(done('t-slow')); + }); + } finally { + vi.useRealTimers(); + } + + expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining('[skill-latency]')); + expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining('exceeds the 60000ms target')); + }); + + it('does not warn for a chain that completes within the target', () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const listeners = renderProvider(); + + vi.useFakeTimers(); + try { + vi.setSystemTime(new Date('2026-06-29T00:00:00.000Z')); + act(() => { + listeners.onToolCall?.(toolCall('t-fast')); + }); + vi.setSystemTime(new Date('2026-06-29T00:00:02.000Z')); // 2s + act(() => { + listeners.onDone?.(done('t-fast')); + }); + } finally { + vi.useRealTimers(); + } + + expect(warnSpy.mock.calls.some(args => String(args[0]).includes('[skill-latency]'))).toBe( + false + ); + }); + + it('closes the latency window on chat_error without warning', () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const listeners = renderProvider(); + + act(() => { + listeners.onToolCall?.(toolCall('t-err')); + }); + expect(() => { + act(() => { + listeners.onError?.({ + thread_id: 't-err', + request_id: 'r1', + error_type: 'provider_error', + message: 'boom', + } as chatService.ChatErrorEvent); + }); + }).not.toThrow(); + + // Error path logs structured latency but never emits the overrun warning. + expect(warnSpy.mock.calls.some(args => String(args[0]).includes('[skill-latency]'))).toBe( + false + ); + }); +}); diff --git a/app/src/services/api/skillRegistryApi.test.ts b/app/src/services/api/skillRegistryApi.test.ts index 9091537ec..30c9bd7c0 100644 --- a/app/src/services/api/skillRegistryApi.test.ts +++ b/app/src/services/api/skillRegistryApi.test.ts @@ -1,6 +1,6 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; -import { skillRegistryApi } from './skillRegistryApi'; +import { invalidateSkillBrowseCache, skillRegistryApi } from './skillRegistryApi'; const mockCallCoreRpc = vi.fn(); vi.mock('../coreRpcClient', () => ({ callCoreRpc: (...a: unknown[]) => mockCallCoreRpc(...a) })); @@ -8,6 +8,8 @@ vi.mock('../coreRpcClient', () => ({ callCoreRpc: (...a: unknown[]) => mockCallC describe('skillRegistryApi', () => { beforeEach(() => { mockCallCoreRpc.mockReset(); + // The browse cache is module-level; clear it so each test starts cold. + invalidateSkillBrowseCache(); }); it('normalizes install new_skills to newSkills', async () => { @@ -163,4 +165,58 @@ describe('skillRegistryApi', () => { timeoutMs: 120_000, }); }); + + it('browse serves the second call from the in-memory cache (one RPC)', async () => { + mockCallCoreRpc.mockResolvedValue({ entries: [{ id: 'a', name: 'A' }] }); + + const first = await skillRegistryApi.browse(); + const second = await skillRegistryApi.browse(); + + expect(mockCallCoreRpc).toHaveBeenCalledTimes(1); + expect(second).toBe(first); // same cached array reference + expect(second[0].id).toBe('a'); + }); + + it('browse de-dupes concurrent callers into a single in-flight RPC', async () => { + let resolveRpc: (v: { entries: { id: string }[] }) => void = () => {}; + mockCallCoreRpc.mockReturnValue( + new Promise(res => { + resolveRpc = res; + }) + ); + + const a = skillRegistryApi.browse(); + const b = skillRegistryApi.browse(); + resolveRpc({ entries: [{ id: 'x' }] }); + const [ra, rb] = await Promise.all([a, b]); + + expect(mockCallCoreRpc).toHaveBeenCalledTimes(1); + expect(ra).toBe(rb); + }); + + it('browse(forceRefresh=true) bypasses the cache and re-fetches', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ entries: [{ id: 'old' }] }); + await skillRegistryApi.browse(); // populates cache + mockCallCoreRpc.mockResolvedValueOnce({ entries: [{ id: 'new' }] }); + + const refreshed = await skillRegistryApi.browse(true); + + expect(mockCallCoreRpc).toHaveBeenCalledTimes(2); + expect(refreshed[0].id).toBe('new'); + + // Subsequent default call now serves the refreshed value from cache. + const cached = await skillRegistryApi.browse(); + expect(mockCallCoreRpc).toHaveBeenCalledTimes(2); + expect(cached[0].id).toBe('new'); + }); + + it('invalidateSkillBrowseCache forces the next browse to re-fetch', async () => { + mockCallCoreRpc.mockResolvedValue({ entries: [{ id: 'a' }] }); + await skillRegistryApi.browse(); + expect(mockCallCoreRpc).toHaveBeenCalledTimes(1); + + invalidateSkillBrowseCache(); + await skillRegistryApi.browse(); + expect(mockCallCoreRpc).toHaveBeenCalledTimes(2); + }); }); diff --git a/app/src/services/api/skillRegistryApi.ts b/app/src/services/api/skillRegistryApi.ts index 26b5ca569..346055574 100644 --- a/app/src/services/api/skillRegistryApi.ts +++ b/app/src/services/api/skillRegistryApi.ts @@ -16,6 +16,31 @@ const log = debug('skillRegistryApi'); */ const CATALOG_RPC_TIMEOUT_MS = 120_000; +/** + * In-memory, session-scoped cache for the unfiltered `browse()` catalog. + * + * The backend already single-flights the upstream ~90k-entry fetch, so warm + * backend reads are fast — but the FRONTEND still re-pulls and re-parses that + * whole payload (tens of MB) over RPC on every Skills-page mount, which is the + * recurring slowness users feel when clicking around. Holding the parsed result + * here makes repeat visits within a session instant and de-dupes concurrent + * mounts via a shared in-flight promise. + * + * Deliberately NOT persisted to localStorage: the payload far exceeds the ~5MB + * quota, so a cold restart re-fetches (and the backend serves it warm). A TTL + * bounds staleness for very long-lived sessions; `force_refresh` and + * `invalidateSkillBrowseCache()` both drop it on demand. + */ +const BROWSE_CACHE_TTL_MS = 30 * 60 * 1000; +let browseCache: { fetchedAt: number; entries: CatalogEntry[] } | null = null; +let browseInflight: Promise | null = null; + +/** Drop the in-memory browse cache (e.g. after an install changes the list). */ +export function invalidateSkillBrowseCache(): void { + browseCache = null; + browseInflight = null; +} + export interface CatalogEntry { id: string; name: string; @@ -82,16 +107,43 @@ function unwrap(response: Envelope | T): T { export const skillRegistryApi = { browse: async (forceRefresh = false): Promise => { log('browse: forceRefresh=%s', forceRefresh); - const response = await callCoreRpc< - Envelope<{ entries: CatalogEntry[] }> | { entries: CatalogEntry[] } - >({ - method: 'openhuman.skill_registry_browse', - params: { force_refresh: forceRefresh }, - timeoutMs: CATALOG_RPC_TIMEOUT_MS, - }); - const result = unwrap(response); - log('browse: count=%d', result.entries.length); - return result.entries; + if (forceRefresh) { + // Explicit refresh: drop any cache / in-flight join so we hit the backend. + invalidateSkillBrowseCache(); + } else { + if (browseCache && Date.now() - browseCache.fetchedAt < BROWSE_CACHE_TTL_MS) { + log('browse: served from in-memory cache count=%d', browseCache.entries.length); + return browseCache.entries; + } + // Concurrent mounts share a single fetch instead of each firing the RPC. + if (browseInflight) { + log('browse: joining in-flight request'); + return browseInflight; + } + } + + const fetchPromise = (async () => { + const response = await callCoreRpc< + Envelope<{ entries: CatalogEntry[] }> | { entries: CatalogEntry[] } + >({ + method: 'openhuman.skill_registry_browse', + params: { force_refresh: forceRefresh }, + timeoutMs: CATALOG_RPC_TIMEOUT_MS, + }); + const result = unwrap(response); + log('browse: count=%d', result.entries.length); + browseCache = { fetchedAt: Date.now(), entries: result.entries }; + return result.entries; + })(); + + browseInflight = fetchPromise; + try { + return await fetchPromise; + } finally { + // Only clear the slot if it's still ours (a concurrent invalidate may have + // already reset it). + if (browseInflight === fetchPromise) browseInflight = null; + } }, search: async (query: string, source?: string, category?: string): Promise => { diff --git a/app/test/e2e/specs/skill-activate-invoke-chat.spec.ts b/app/test/e2e/specs/skill-activate-invoke-chat.spec.ts new file mode 100644 index 000000000..47e92872c --- /dev/null +++ b/app/test/e2e/specs/skill-activate-invoke-chat.spec.ts @@ -0,0 +1,145 @@ +/** + * E2E: activate a skill → invoke it from chat → see the result (issue #4273, AC5). + * + * This is the headline lifecycle the Skills page promises but that no spec + * previously covered end to end: a user activates a skill (here a Composio + * toolkit connection), then asks the agent to do something in chat, the agent + * routes through the skills/Composio tool surface, and the result renders in + * the conversation. + * + * Strategy: + * - Seed an ACTIVE Composio connection so the skill is "activated". + * - Force two LLM turns via the mock: turn 1 emits a `composio_execute` + * tool call, turn 2 returns the final answer once the tool result is fed + * back. This mirrors `chat-tool-call-flow.spec.ts` but exercises the skill + * (Composio) tool path instead of `web_fetch`. + * - Assert the tool call reached the Composio execute endpoint AND the final + * answer renders. + * - Assert the whole chain completes within the 60s latency target (AC3). + */ +import { waitForApp } from '../helpers/app-helpers'; +import { + chatMounted, + clickByTitle, + clickSend, + getSelectedThreadId, + typeIntoComposer, + waitForAssistantReplyContaining, + waitForSocketConnected, + waitForToolCallInMockLog, +} from '../helpers/chat-harness'; +import { seedComposioConnection, seedComposioToolkits } from '../helpers/composio-helpers'; +import { resetApp } from '../helpers/reset-app'; +import { navigateViaHash } from '../helpers/shared-flows'; +import { + clearRequestLog, + resetMockBehavior, + setMockBehavior, + startMockServer, + stopMockServer, +} from '../mock-server'; + +const LOG_PREFIX = '[skill-activate-invoke-chat]'; +const USER_ID = 'e2e-skill-activate-invoke-chat'; +const TOOLKIT_SLUG = 'gmail'; +const COMPOSIO_ACTION = 'GMAIL_FETCH_EMAILS'; +const PROMPT = 'Use my connected Gmail to fetch my latest emails.'; +const CANARY_FINAL = 'canary-skill-invoked-e7f1a9'; + +// Turn 1: the agent calls the Composio skill tool. Turn 2: the final answer +// after the core feeds the tool result back to the LLM. +const FORCED_RESPONSES = [ + { + content: '', + toolCalls: [ + { + id: 'call_composio_execute_1', + name: 'composio_execute', + arguments: JSON.stringify({ tool: COMPOSIO_ACTION, arguments: {} }), + }, + ], + }, + { content: `Here are your latest emails: ${CANARY_FINAL}` }, +]; + +describe('Skill activate → invoke from chat', () => { + before(async function beforeSuite() { + this.timeout(90_000); + await startMockServer(); + // Activate the skill: an ACTIVE Composio connection the agent can route to. + seedComposioToolkits([TOOLKIT_SLUG]); + seedComposioConnection(TOOLKIT_SLUG, 'ACTIVE', 'c-gmail-1'); + await waitForApp(); + await resetApp(USER_ID); + + setMockBehavior('llmForcedResponses', JSON.stringify(FORCED_RESPONSES)); + setMockBehavior('llmStreamChunkDelayMs', '10'); + clearRequestLog(); + console.log(`${LOG_PREFIX} setup complete — skill activated, forced responses configured`); + }); + + after(async () => { + setMockBehavior('llmForcedResponses', ''); + setMockBehavior('llmStreamChunkDelayMs', ''); + resetMockBehavior(); + await stopMockServer(); + }); + + it('routes a chat request through the activated skill and renders the result within 60s', async function () { + this.timeout(90_000); + + await navigateViaHash('/chat'); + await browser.waitUntil(async () => await chatMounted(), { + timeout: 15_000, + timeoutMsg: 'chat did not mount', + }); + expect(await clickByTitle('New thread', 8_000)).toBe(true); + + const threadId = (await browser.waitUntil(async () => await getSelectedThreadId(), { + timeout: 8_000, + timeoutMsg: 'thread.selectedThreadId never populated', + })) as string; + expect(typeof threadId).toBe('string'); + + await typeIntoComposer(PROMPT); + const socketReady = await waitForSocketConnected(30_000); + if (!socketReady) { + console.warn(`${LOG_PREFIX} socket did not connect within 30s — send may fail`); + } + + // Start the latency clock at send so we can assert the AC3 budget on the + // full activate→invoke→result chain. + const startedAt = Date.now(); + expect( + await browser.waitUntil(async () => await clickSend(), { + timeout: 5_000, + timeoutMsg: 'Send button never enabled', + }) + ).toBe(true); + + // The agent should route to the Composio skill execute endpoint. This is a + // soft signal (the final answer below is the hard oracle) so a faster-than- + // poll turn doesn't flake the test. + const toolHit = await waitForToolCallInMockLog(COMPOSIO_ACTION, { + source: 'composio', + timeoutMs: 45_000, + logPrefix: LOG_PREFIX, + }); + // Hard assertion: the whole point of this spec is the activate→invoke + // routing guarantee, so the agent MUST hit the Composio execute endpoint — + // a missing call is a failure, not a warning (PR #4288 review). + expect(toolHit).toBeDefined(); + + // Hard oracle: the skill's result renders in the conversation. + const replied = await waitForAssistantReplyContaining(CANARY_FINAL, { + timeoutMs: 50_000, + logPrefix: LOG_PREFIX, + }); + expect(replied).toBe(true); + + const elapsedMs = Date.now() - startedAt; + console.log(`${LOG_PREFIX} activate→invoke→result completed in ${elapsedMs}ms`); + // AC3: complex skill tool chains complete within 60 seconds. + expect(elapsedMs).toBeLessThan(60_000); + }); +}); diff --git a/app/test/e2e/specs/skill-activation-persistence.spec.ts b/app/test/e2e/specs/skill-activation-persistence.spec.ts new file mode 100644 index 000000000..840edd7a1 --- /dev/null +++ b/app/test/e2e/specs/skill-activation-persistence.spec.ts @@ -0,0 +1,128 @@ +/** + * E2E: skill activation persists across a (simulated) app restart (issue #4273, AC1). + * + * The Skills page fetches Composio connections fresh on every mount, so a cold + * start used to flash an empty/disconnected grid until the first backend + * round-trip landed. The durable connection cache (`connectionCache.ts`) fixes + * that by seeding the last-known activation state instantly on mount. + * + * This spec proves the activated state survives a restart-equivalent re-mount + * EVEN WHEN the backend is unreachable at that moment — which is exactly the + * cold-start window the cache exists to cover. We: + * 1. Seed an ACTIVE Composio connection and load the Skills page once so the + * durable cache is populated. + * 2. Inject a Composio backend fault so any *fresh* fetch would fail. + * 3. Re-mount the page (navigate away + back — the restart-equivalent the + * rewards-persistence spec also uses, since tauri-driver has no cheap real + * restart) and assert the activated skill still renders, served from the + * cache rather than a successful new fetch. + * + * If the card only ever came from a live fetch, step 3 would show it missing + * once the fault is injected — so a passing assertion is specifically + * attributable to the persisted cache. + */ +import { waitForApp } from '../helpers/app-helpers'; +import { + assertConnectorCardVisible, + assertSessionNotNuked, + injectComposioFault, + seedComposioConnection, + seedComposioToolkits, +} from '../helpers/composio-helpers'; +import { triggerAuthDeepLinkBypass } from '../helpers/deep-link-helpers'; +import { + textExists, + waitForText, + waitForWebView, + waitForWindowVisible, +} from '../helpers/element-helpers'; +import { + completeOnboardingIfVisible, + navigateToConnections, + navigateViaHash, +} from '../helpers/shared-flows'; +import { + clearRequestLog, + resetMockBehavior, + startMockServer, + stopMockServer, +} from '../mock-server'; + +const LOG = '[skill-activation-persistence]'; +const CONNECTOR_NAME = 'Gmail'; +const TOOLKIT_SLUG = 'gmail'; +const AUTH_TOKEN = 'e2e-skill-activation-persistence-token'; + +/** + * Restart-equivalent: navigate away so the Skills page unmounts, then back so + * it re-mounts and re-runs its on-mount fetch — the same approach + * `rewards-progression-persistence.spec.ts` uses in lieu of a real process + * restart (which tauri-driver does not support cheaply). + */ +async function simulateRestart(): Promise { + await navigateViaHash('/home'); + await browser.pause(1_000); + await navigateToConnections(); + await browser.pause(1_000); +} + +describe('Skill activation persistence across restart', () => { + before(async function () { + this.timeout(90_000); + await startMockServer(); + seedComposioToolkits([TOOLKIT_SLUG]); + seedComposioConnection(TOOLKIT_SLUG, 'ACTIVE', 'c-gmail-1'); + await waitForApp(); + clearRequestLog(); + await triggerAuthDeepLinkBypass(AUTH_TOKEN); + await waitForWindowVisible(25_000); + await waitForWebView(15_000); + await completeOnboardingIfVisible(LOG); + }); + + after(async () => { + resetMockBehavior(); + await stopMockServer(); + }); + + it('shows the activated skill on first load and writes it to the durable cache', async function () { + this.timeout(60_000); + await assertConnectorCardVisible(CONNECTOR_NAME); + + // Durable-write proof (PR #4288 review): the connection must be persisted to + // localStorage under the user-scoped `${userId}:composio:connections:v1` + // key — not merely held in the module's in-memory mirror. Asserting the + // durable blob here makes the persistence path load-bearing for this spec, + // so a broken write fails the test instead of being masked by the in-memory + // hydrate on the re-mount below. (Cold-restart read-back — fresh module + // memory, warm localStorage — is covered by the connectionCache unit test, + // which tauri-driver cannot cheaply reproduce with a real relaunch.) + const persisted = await browser.execute(() => { + // Resolve the active user id and read that exact user-scoped key, rather + // than suffix-matching any cache entry — otherwise a stale blob from a + // different user could satisfy the assertion (PR #4288 review). + const userId = window.localStorage.getItem('OPENHUMAN_ACTIVE_USER_ID'); + return userId ? window.localStorage.getItem(`${userId}:composio:connections:v1`) : null; + }); + expect(persisted).toBeTruthy(); + expect(String(persisted).toLowerCase()).toContain(TOOLKIT_SLUG); + console.log(`${LOG} PASS: activated skill visible + persisted to durable cache`); + }); + + it('still shows the activated skill after a restart when the backend is unreachable', async function () { + this.timeout(60_000); + + // From here on, any fresh Composio fetch fails — so a card that appears + // after the re-mount came from the seeded/persisted state, not a new + // backend fetch. + injectComposioFault(500); + + await simulateRestart(); + + await waitForText(CONNECTOR_NAME, 15_000); + expect(await textExists(CONNECTOR_NAME)).toBe(true); + // The unreachable backend must not blank the page or tear down the session. + await assertSessionNotNuked(); + console.log(`${LOG} PASS: activation survived restart via the durable cache`); + }); +});