From cd0b08ec0a414e4636342bf27bf1c564d3ed1911 Mon Sep 17 00:00:00 2001 From: Cyrus Gray <144336577+graycyrus@users.noreply.github.com> Date: Wed, 8 Jul 2026 00:56:05 +0530 Subject: [PATCH] fix(flows): persist + rehydrate the Copilot chat across reload (B11) (#4663) --- .../flows/workflowCopilotThreads.test.ts | 137 ++++++++++++++++ .../flows/workflowCopilotThreads.ts | 66 ++++++-- app/src/hooks/useWorkflowBuilderChat.test.ts | 148 ++++++++++++++++++ app/src/hooks/useWorkflowBuilderChat.ts | 105 +++++++++++-- app/src/store/__tests__/threadSlice.test.ts | 57 +++++++ app/src/store/threadSlice.ts | 35 ++++- 6 files changed, 524 insertions(+), 24 deletions(-) create mode 100644 app/src/components/flows/workflowCopilotThreads.test.ts diff --git a/app/src/components/flows/workflowCopilotThreads.test.ts b/app/src/components/flows/workflowCopilotThreads.test.ts new file mode 100644 index 000000000..a213a6693 --- /dev/null +++ b/app/src/components/flows/workflowCopilotThreads.test.ts @@ -0,0 +1,137 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { copilotThreadKey, getCopilotThreadId, setCopilotThreadId } from './workflowCopilotThreads'; + +// Controllable active-user id so tests can exercise the `${userId}:` scoping +// (#900/#983 convention) without pulling in the real module's boot-priming. +const userScopedState = vi.hoisted(() => ({ activeUserId: null as string | null })); +vi.mock('../../store/userScopedStorage', () => ({ + getActiveUserId: () => userScopedState.activeUserId, +})); + +describe('workflowCopilotThreads', () => { + beforeEach(() => { + window.localStorage.clear(); + userScopedState.activeUserId = null; + }); + + it('returns null for a flow that has never been set', () => { + expect(getCopilotThreadId('flow-1')).toBeNull(); + }); + + it('round-trips a thread id for a persisted flow via localStorage', () => { + setCopilotThreadId('flow-1', 'thread-abc'); + expect(getCopilotThreadId('flow-1')).toBe('thread-abc'); + // Persisted directly in localStorage (not just an in-memory cache) so a + // simulated reload — a fresh read with no prior JS state — still resolves. + expect(window.localStorage.getItem(`copilot-thread:${copilotThreadKey('flow-1')}`)).toBe( + 'thread-abc' + ); + }); + + it('round-trips a thread id for an unsaved draft (null flow id)', () => { + setCopilotThreadId(null, 'thread-draft'); + expect(getCopilotThreadId(null)).toBe('thread-draft'); + expect(copilotThreadKey(null)).toBe('draft'); + }); + + it('survives a simulated reload (fresh read with no prior in-memory state)', () => { + setCopilotThreadId('flow-2', 'thread-xyz'); + + // Simulate a full app reload: nothing survives except localStorage. + expect(getCopilotThreadId('flow-2')).toBe('thread-xyz'); + }); + + it('keeps different flows (and the draft) isolated from each other', () => { + setCopilotThreadId('flow-1', 'thread-1'); + setCopilotThreadId('flow-2', 'thread-2'); + setCopilotThreadId(null, 'thread-draft'); + + expect(getCopilotThreadId('flow-1')).toBe('thread-1'); + expect(getCopilotThreadId('flow-2')).toBe('thread-2'); + expect(getCopilotThreadId(null)).toBe('thread-draft'); + }); + + it('removes the mapping when set to null', () => { + setCopilotThreadId('flow-1', 'thread-abc'); + expect(getCopilotThreadId('flow-1')).toBe('thread-abc'); + + setCopilotThreadId('flow-1', null); + expect(getCopilotThreadId('flow-1')).toBeNull(); + }); + + it('degrades to a no-op instead of throwing when localStorage is unavailable', () => { + const original = window.localStorage.getItem; + // Simulate private-mode / quota errors. + window.localStorage.getItem = () => { + throw new Error('unavailable'); + }; + try { + expect(() => getCopilotThreadId('flow-1')).not.toThrow(); + expect(getCopilotThreadId('flow-1')).toBeNull(); + } finally { + window.localStorage.getItem = original; + } + }); + + it('degrades to a no-op when localStorage.setItem throws (write path)', () => { + const original = window.localStorage.setItem; + // Simulate private-mode / quota errors on the write path. + window.localStorage.setItem = () => { + throw new Error('quota exceeded'); + }; + try { + expect(() => setCopilotThreadId('flow-1', 'thread-abc')).not.toThrow(); + } finally { + window.localStorage.setItem = original; + } + }); + + it('degrades to a no-op when localStorage.removeItem throws (clear path)', () => { + setCopilotThreadId('flow-1', 'thread-abc'); + + const original = window.localStorage.removeItem; + // Simulate private-mode / quota errors on the clear path. + window.localStorage.removeItem = () => { + throw new Error('unavailable'); + }; + try { + expect(() => setCopilotThreadId('flow-1', null)).not.toThrow(); + } finally { + window.localStorage.removeItem = original; + } + }); + + describe('user scoping (#900/#983 convention)', () => { + it('namespaces the storage key with the active user id', () => { + userScopedState.activeUserId = 'user-a'; + setCopilotThreadId('flow-1', 'thread-abc'); + expect( + window.localStorage.getItem(`user-a:copilot-thread:${copilotThreadKey('flow-1')}`) + ).toBe('thread-abc'); + }); + + it("keeps two users' thread ids for the same flow isolated", () => { + userScopedState.activeUserId = 'user-a'; + setCopilotThreadId('flow-1', 'thread-a'); + + userScopedState.activeUserId = 'user-b'; + setCopilotThreadId('flow-1', 'thread-b'); + expect(getCopilotThreadId('flow-1')).toBe('thread-b'); + + // Switching back to user-a must not see user-b's thread id — an + // identity flip (or a "clear my data" scoped to one user's `${userId}:*` + // keys) must never leak or destroy the other user's mapping. + userScopedState.activeUserId = 'user-a'; + expect(getCopilotThreadId('flow-1')).toBe('thread-a'); + }); + + it('falls back to an unscoped key when no user is active yet (pre-login)', () => { + userScopedState.activeUserId = null; + setCopilotThreadId('flow-1', 'thread-abc'); + expect(window.localStorage.getItem(`copilot-thread:${copilotThreadKey('flow-1')}`)).toBe( + 'thread-abc' + ); + }); + }); +}); diff --git a/app/src/components/flows/workflowCopilotThreads.ts b/app/src/components/flows/workflowCopilotThreads.ts index d7400ec9a..989250169 100644 --- a/app/src/components/flows/workflowCopilotThreads.ts +++ b/app/src/components/flows/workflowCopilotThreads.ts @@ -1,29 +1,75 @@ /** - * Session-lived cache of each workflow's copilot chat thread id, keyed by flow + * Persisted cache of each workflow's copilot chat thread id, keyed by flow * (a persisted flow id, or `'draft'` for an unsaved draft). The copilot panel * unmounts when closed and `FlowEditor` remounts when switching flows, so * without this the `workflow_builder` thread — and its transcript — would be * lost on every open/close. Persisting the thread id lets the panel reseed the - * same thread (its messages live in the Redux `messagesByThreadId` store), so + * same thread (its messages live in the core, rehydrated into the Redux + * `messagesByThreadId` store by `useWorkflowBuilderChat`'s mount effect), so * reopening the copilot restores the conversation for that workflow. * - * Module-level (session) scope is deliberate: it survives component remounts - * without coupling to Redux, and a fresh session legitimately starts a new - * authoring conversation. + * Backed by `localStorage` (not a module-level `Map`) so the mapping survives + * a full app reload — the durable half of "Copilot chat not persistent": the + * transcript itself is already durable server-side via `threadApi`, this file + * is what makes the panel know WHICH thread to reload. `localStorage` access + * is wrapped in try/catch — private-mode / quota errors degrade to a no-op + * (the copilot just starts a fresh thread on the next open) rather than + * throwing. + * + * Keys are namespaced by the active user id (`${userId}:copilot-thread:`), + * the same `${userId}:` convention `userScopedStorage`/`clearAllAppData` use for + * every other per-user localStorage blob (#900, #983). Without this an + * identity flip (or a "clear my data" on account B that only purges B's + * `${userId}:*` keys) would leave account A's thread id readable by whoever + * opens the same flow/draft next, pointing them at A's builder thread instead + * of starting fresh. `getActiveUserId()` is synchronous (primed at boot by + * `main.tsx`), so this stays a plain sync read/write unlike the async + * redux-persist storage contract in `userScopedStorage.ts`. */ -const copilotThreadByFlow = new Map(); +import createDebug from 'debug'; + +import { getActiveUserId } from '../../store/userScopedStorage'; + +const log = createDebug('app:flows:copilot-threads'); + +const STORAGE_PREFIX = 'copilot-thread:'; /** Cache key for a flow: its persisted id, or `'draft'` for an unsaved draft. */ export function copilotThreadKey(flowId: string | null): string { return flowId ?? 'draft'; } +function storageKey(flowId: string | null): string { + const userId = getActiveUserId(); + const scope = userId ? `${userId}:` : ''; + return `${scope}${STORAGE_PREFIX}${copilotThreadKey(flowId)}`; +} + export function getCopilotThreadId(flowId: string | null): string | null { - return copilotThreadByFlow.get(copilotThreadKey(flowId)) ?? null; + const flowKey = copilotThreadKey(flowId); + try { + const threadId = window.localStorage.getItem(storageKey(flowId)); + log('get flow=%s -> %s', flowKey, threadId ?? '(none)'); + return threadId; + } catch (err) { + log('get flow=%s failed: %o', flowKey, err); + return null; + } } export function setCopilotThreadId(flowId: string | null, threadId: string | null): void { - const key = copilotThreadKey(flowId); - if (threadId) copilotThreadByFlow.set(key, threadId); - else copilotThreadByFlow.delete(key); + const flowKey = copilotThreadKey(flowId); + try { + if (threadId) { + window.localStorage.setItem(storageKey(flowId), threadId); + log('set flow=%s -> thread=%s', flowKey, threadId); + } else { + window.localStorage.removeItem(storageKey(flowId)); + log('clear flow=%s', flowKey); + } + } catch (err) { + // Private-mode / quota errors are non-fatal — worst case the copilot + // simply starts a fresh thread on the next open instead of resuming. + log('set flow=%s failed: %o', flowKey, err); + } } diff --git a/app/src/hooks/useWorkflowBuilderChat.test.ts b/app/src/hooks/useWorkflowBuilderChat.test.ts index 016d258b6..e0be52332 100644 --- a/app/src/hooks/useWorkflowBuilderChat.test.ts +++ b/app/src/hooks/useWorkflowBuilderChat.test.ts @@ -33,13 +33,21 @@ vi.mock('../store/hooks', () => ({ }), })); +const THREAD_NOT_FOUND_MESSAGE = vi.hoisted(() => 'This thread is no longer available.'); vi.mock('../store/threadSlice', () => ({ createNewThread: (labels: string[]) => ({ type: 'createNewThread', labels }), addMessageLocal: (p: unknown) => ({ type: 'addMessageLocal', p }), + loadThreadMessages: (threadId: string) => ({ type: 'loadThreadMessages', threadId }), + THREAD_NOT_FOUND_MESSAGE, })); vi.mock('../store/chatRuntimeSlice', () => ({ clearWorkflowProposalForThread: (p: unknown) => ({ type: 'clearProposal', p }), setWorkflowProposalForThread: (p: unknown) => ({ type: 'setProposal', p }), + fetchAndHydrateTurnState: (threadId: string) => ({ type: 'fetchAndHydrateTurnState', threadId }), + fetchAndHydrateTurnHistory: (threadId: string) => ({ + type: 'fetchAndHydrateTurnHistory', + threadId, + }), })); // The hook reads the live store directly (not the stale closed-over selector @@ -73,6 +81,10 @@ describe('useWorkflowBuilderChat', () => { if (action.type === 'addMessageLocal') { return { unwrap: () => Promise.resolve(undefined) }; } + if (action.type === 'loadThreadMessages') { + // Default: fetch succeeds (mirrors the real thunk's fulfilled action). + return Promise.resolve({ type: 'loadThreadMessages/fulfilled', payload: { messages: [] } }); + } return undefined; }); }); @@ -302,4 +314,140 @@ describe('useWorkflowBuilderChat', () => { expect(result.current.displayMessages.map(m => m.id)).toEqual(['m1', 'm4']); }); }); + + describe('rehydration on mount with seedThreadId', () => { + it('dispatches loadThreadMessages + turn state/history rehydration for the seed thread', () => { + renderHook(() => useWorkflowBuilderChat('seed-thread-1')); + expect(dispatch).toHaveBeenCalledWith( + expect.objectContaining({ type: 'loadThreadMessages', threadId: 'seed-thread-1' }) + ); + expect(dispatch).toHaveBeenCalledWith( + expect.objectContaining({ type: 'fetchAndHydrateTurnState', threadId: 'seed-thread-1' }) + ); + expect(dispatch).toHaveBeenCalledWith( + expect.objectContaining({ type: 'fetchAndHydrateTurnHistory', threadId: 'seed-thread-1' }) + ); + }); + + it('does not rehydrate when mounted with no seed thread', () => { + renderHook(() => useWorkflowBuilderChat()); + expect(dispatch).not.toHaveBeenCalledWith( + expect.objectContaining({ type: 'loadThreadMessages' }) + ); + }); + + it('does not rehydrate a thread this hook just created when seedThreadId echoes it back', async () => { + // Mirrors the real wiring: `WorkflowCopilotPanel` reports every + // `threadId` change back up via `onThreadIdChange`, and `FlowCanvasPage` + // re-passes that straight back in as `seedThreadId` on the next render. + // A first send() creates a fresh thread; the resulting re-render must + // NOT trigger a rehydrate for it — that would race the in-flight turn + // and `loadThreadMessages.fulfilled` would wipe the just-appended + // message(s) back out of the transcript. + const { result, rerender } = renderHook( + ({ seedThreadId }: { seedThreadId?: string | null }) => + useWorkflowBuilderChat(seedThreadId), + { initialProps: { seedThreadId: undefined as string | null | undefined } } + ); + + await act(async () => { + await result.current.send({ + displayText: 'hi', + request: { mode: 'create', instruction: 'x' }, + }); + }); + expect(result.current.threadId).toBe('builder-1'); + + dispatch.mockClear(); + rerender({ seedThreadId: 'builder-1' }); + + expect(dispatch).not.toHaveBeenCalledWith( + expect.objectContaining({ type: 'loadThreadMessages' }) + ); + expect(dispatch).not.toHaveBeenCalledWith( + expect.objectContaining({ type: 'fetchAndHydrateTurnState' }) + ); + expect(dispatch).not.toHaveBeenCalledWith( + expect.objectContaining({ type: 'fetchAndHydrateTurnHistory' }) + ); + }); + + it('still rehydrates a genuinely pre-existing seed thread this hook did not create', () => { + // A different thread id than any this hook instance created via + // send() — the normal "resume a persisted copilot thread on mount" case + // — must still rehydrate. + renderHook(() => useWorkflowBuilderChat('previously-persisted-thread')); + expect(dispatch).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'loadThreadMessages', + threadId: 'previously-persisted-thread', + }) + ); + }); + + it('clears threadId when the seeded thread no longer exists (stale persisted id)', async () => { + // `loadThreadMessages` rejects with THREAD_NOT_FOUND_MESSAGE when the + // cached/seeded thread was deleted server-side (see threadSlice.ts). + // The hook must null out its `threadId` so a caller's onThreadIdChange + // effect (WorkflowCopilotPanel -> FlowCanvasPage) clears the stale + // `workflowCopilotThreads.ts` cache entry and the next send() creates a + // fresh thread instead of retrying the dead one forever. + dispatch.mockImplementation((action: { type: string }) => { + if (action.type === 'createNewThread') { + return { unwrap: () => Promise.resolve({ id: 'builder-1' }) }; + } + if (action.type === 'addMessageLocal') { + return { unwrap: () => Promise.resolve(undefined) }; + } + if (action.type === 'loadThreadMessages') { + return Promise.resolve({ + type: 'loadThreadMessages/rejected', + payload: THREAD_NOT_FOUND_MESSAGE, + }); + } + return undefined; + }); + + const { result } = renderHook(() => useWorkflowBuilderChat('stale-thread')); + expect(result.current.threadId).toBe('stale-thread'); + + await waitFor(() => expect(result.current.threadId).toBeNull()); + }); + }); + + describe('recovering from a stale thread id during send()', () => { + it('clears threadId when addMessageLocal fails because the seeded thread was deleted', async () => { + // Mirrors a stale `workflowCopilotThreads.ts` seed surviving past mount + // (e.g. the rehydrate GET raced and lost, or the thread was deleted + // between mount and this send). `addMessageLocal` rejects with + // THREAD_NOT_FOUND_MESSAGE (see threadSlice.ts); the hook must recover + // by nulling `threadId` so the NEXT send creates a fresh thread instead + // of erroring forever against the dead one. + dispatch.mockImplementation((action: { type: string }) => { + if (action.type === 'addMessageLocal') { + return { unwrap: () => Promise.reject(THREAD_NOT_FOUND_MESSAGE) }; + } + if (action.type === 'loadThreadMessages') { + return Promise.resolve({ + type: 'loadThreadMessages/fulfilled', + payload: { messages: [] }, + }); + } + return undefined; + }); + + const { result } = renderHook(() => useWorkflowBuilderChat('stale-thread')); + expect(result.current.threadId).toBe('stale-thread'); + + await act(async () => { + await result.current.send({ + displayText: 'hi', + request: { mode: 'create', instruction: 'x' }, + }); + }); + + expect(result.current.error).toBe(THREAD_NOT_FOUND_MESSAGE); + expect(result.current.threadId).toBeNull(); + }); + }); }); diff --git a/app/src/hooks/useWorkflowBuilderChat.ts b/app/src/hooks/useWorkflowBuilderChat.ts index ab31b9cb1..1a290b8ea 100644 --- a/app/src/hooks/useWorkflowBuilderChat.ts +++ b/app/src/hooks/useWorkflowBuilderChat.ts @@ -23,19 +23,26 @@ * a real flow id) may save onto an existing flow. Nothing here enables a flow. */ import createDebug from 'debug'; -import { useCallback, useMemo, useState } from 'react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { type BuilderTurnRequest, buildWorkflow } from '../services/api/flowsApi'; import { store } from '../store'; import { clearWorkflowProposalForThread, + fetchAndHydrateTurnHistory, + fetchAndHydrateTurnState, setWorkflowProposalForThread, type ToolTimelineEntry, type WorkflowProposal, } from '../store/chatRuntimeSlice'; import { useAppDispatch, useAppSelector } from '../store/hooks'; import { selectSocketStatus } from '../store/socketSelectors'; -import { addMessageLocal, createNewThread } from '../store/threadSlice'; +import { + addMessageLocal, + createNewThread, + loadThreadMessages, + THREAD_NOT_FOUND_MESSAGE, +} from '../store/threadSlice'; import type { ThreadMessage } from '../types/thread'; const log = createDebug('app:flows:builder-chat'); @@ -61,10 +68,10 @@ export interface UseWorkflowBuilderChat { proposal: WorkflowProposal | null; /** * The dedicated thread's FULL transcript (user + agent turns, including - * between-tool narration bubbles) so a caller that needs the complete - * history can still get it. Empty until the first send. Sourced from the - * same `messagesByThreadId` store the main chat transcript reads. Most - * callers should render `displayMessages` instead (see below). + * between-tool narration bubbles), so a caller that needs the complete + * history (e.g. persistence/rehydration) can still get it. Empty until the + * first send. Sourced from the same `messagesByThreadId` store the main chat + * transcript reads. */ messages: ThreadMessage[]; /** @@ -110,9 +117,15 @@ const EMPTY_MESSAGES: ThreadMessage[] = []; const EMPTY_TIMELINE: ToolTimelineEntry[] = []; /** - * @param seedThreadId Optional existing thread to bind to instead of creating a - * fresh one — lets a caller reuse a thread across mounts (unused today; the - * prompt bar and copilot each start clean). + * @param seedThreadId Optional existing thread to bind to instead of creating + * a fresh one — lets a caller reuse a thread across mounts. When this + * identifies a genuinely pre-existing thread (i.e. not one `send()` just + * created on this hook instance — see `createdThreadIdRef`), this hook + * rehydrates that thread's messages + turn state/history from the core on + * mount (mirroring `Conversations.tsx`'s thread-switch effect) so a + * persisted copilot thread (`workflowCopilotThreads.ts`) resumes its full + * transcript after a reload instead of starting empty (issue: Copilot chat + * not persistent). */ export function useWorkflowBuilderChat(seedThreadId?: string | null): UseWorkflowBuilderChat { const dispatch = useAppDispatch(); @@ -120,6 +133,18 @@ export function useWorkflowBuilderChat(seedThreadId?: string | null): UseWorkflo const [threadId, setThreadId] = useState(seedThreadId ?? null); const [localSending, setLocalSending] = useState(false); const [error, setError] = useState(null); + // Tracks a thread id this hook created itself via `send()`'s `createNewThread` + // call — as opposed to one that arrived from `seedThreadId` because a caller + // (e.g. `WorkflowCopilotPanel`) reports every `threadId` change back up via + // `onThreadIdChange` and re-passes it in as `seedThreadId` on the next + // render. Without this distinction, the rehydrate effect below would treat + // that echo as "an existing persisted thread just got selected" and refetch + // it from the core mid-turn — a redundant `loadThreadMessages` GET racing + // the in-flight turn's own append. `loadThreadMessages.fulfilled` merges a + // locally-appended message that predates the fetch's snapshot back in + // (rather than wholesale-replacing), so this guard is defense-in-depth + // against the unnecessary refetch itself, not a correctness requirement. + const createdThreadIdRef = useRef(null); const proposalsByThread = useAppSelector( state => state.chatRuntime.pendingWorkflowProposalsByThread @@ -148,7 +173,7 @@ export function useWorkflowBuilderChat(seedThreadId?: string | null): UseWorkflo // via `toolTimeline`/`liveResponse`), keeping every user turn and every // non-interim agent turn (the terminal answer for a round, including a // clarifying question with no `isInterim` tag). `messages` itself stays the - // full set — any future persistence/rehydration needs it intact. + // full set — rehydration (below) and any future persistence need it intact. const displayMessages = useMemo( () => messages.filter(m => m.sender === 'user' || !m.extraMetadata?.isInterim), [messages] @@ -164,6 +189,51 @@ export function useWorkflowBuilderChat(seedThreadId?: string | null): UseWorkflo [threadId, streamingAssistantByThread] ); + // Rehydrate a persisted thread's transcript + turn state/history from the + // core on mount. Messages ARE durable server-side (`threadApi.appendMessage` + // persists every turn) — redux-persist only whitelists `selectedThreadId` + // for the thread slice, so `messagesByThreadId` starts empty on a fresh app + // load and this thread's messages would otherwise never come back. + // + // Skips when `seedThreadId` is one THIS hook created via `send()` (tracked + // by `createdThreadIdRef`): `WorkflowCopilotPanel` reports every `threadId` + // change back up through `onThreadIdChange`, and a caller like + // `FlowCanvasPage` re-passes that straight back in as `seedThreadId` on the + // next render — so `seedThreadId` also changes right after a fresh thread is + // created by a first send, not just when a truly pre-existing persisted + // thread is (re)selected. Rehydrating in that case would be a wasted GET + // racing the in-flight turn's own append; `loadThreadMessages.fulfilled` + // merges rather than wholesale-replaces (see `threadSlice.ts`), so a + // straggler fetch can no longer wipe out a newer local append, but skipping + // it here still avoids the redundant round trip. + useEffect(() => { + if (!seedThreadId) return; + if (createdThreadIdRef.current === seedThreadId) { + log('rehydrate: skipping — this hook created thread=%s locally', seedThreadId); + return; + } + log('rehydrate: loading persisted messages + turn state/history thread=%s', seedThreadId); + // A persisted seed (`workflowCopilotThreads.ts`) can point at a thread + // that no longer exists (e.g. deleted/purged since it was cached). + // `loadThreadMessages` evicts the stale thread from Redux in that case + // and rejects with `THREAD_NOT_FOUND_MESSAGE` — null out this hook's + // `threadId` in response so the effect below that reports it back up + // (`WorkflowCopilotPanel` -> `onThreadIdChange` -> `FlowCanvasPage`) + // clears the stale cached id too, letting the next `send()` create a + // fresh thread instead of retrying the dead one forever. + void dispatch(loadThreadMessages(seedThreadId)).then((action: { payload?: unknown }) => { + if (action?.payload === THREAD_NOT_FOUND_MESSAGE) { + log( + 'rehydrate: thread=%s no longer exists — clearing seed so a future send starts fresh', + seedThreadId + ); + setThreadId(current => (current === seedThreadId ? null : current)); + } + }); + void dispatch(fetchAndHydrateTurnState(seedThreadId)); + void dispatch(fetchAndHydrateTurnHistory(seedThreadId)); + }, [seedThreadId, dispatch]); + // The turn is a single request/response RPC (no streaming runtime), so // "sending" is simply whether that call is in flight. const sending = localSending; @@ -188,6 +258,7 @@ export function useWorkflowBuilderChat(seedThreadId?: string | null): UseWorkflo log('send: creating dedicated builder thread'); const thread = await dispatch(createNewThread(['workflow-builder'])).unwrap(); targetThreadId = thread.id; + createdThreadIdRef.current = targetThreadId; setThreadId(targetThreadId); } // A fresh turn supersedes any prior proposal on this thread. @@ -266,6 +337,20 @@ export function useWorkflowBuilderChat(seedThreadId?: string | null): UseWorkflo const msg = err instanceof Error ? err.message : String(err); log('send: failed err=%o', err); setError(msg); + // The pre-existing/seeded thread this turn targeted no longer exists + // server-side (deleted/purged since it was cached) — `addMessageLocal` + // already evicted it from Redux; clear it here too so the next send + // creates a fresh thread instead of retrying the dead one forever. + // Scoped to `targetThreadId === threadId` (the thread this hook was + // seeded/already bound to) so a failure on a thread just created this + // same call doesn't get misattributed. + if (msg === THREAD_NOT_FOUND_MESSAGE && targetThreadId === threadId) { + log( + 'send: thread=%s no longer exists — clearing cached id so the next send starts fresh', + targetThreadId + ); + setThreadId(null); + } } finally { setLocalSending(false); } diff --git a/app/src/store/__tests__/threadSlice.test.ts b/app/src/store/__tests__/threadSlice.test.ts index d641c36a8..6762ab74e 100644 --- a/app/src/store/__tests__/threadSlice.test.ts +++ b/app/src/store/__tests__/threadSlice.test.ts @@ -282,6 +282,63 @@ describe('threadSlice loadThreadMessages thunk', () => { expect(state.isLoadingMessages).toBe(false); expect(state.messagesError).toBe('boom'); }); + + it('merges a locally-appended message not yet reflected in the fetch instead of wiping it out', async () => { + // Simulates a rehydrate racing a concurrent append on the same thread + // (e.g. useWorkflowBuilderChat's copilot rehydrate vs. an auto-sent + // repair/build turn): the local message already persisted server-side + // (it only lands via a `.fulfilled` reducer) but this particular GET's + // snapshot predates it. + const store = createStore(); + const persisted = makeMessage({ id: 'local-new', content: 'just sent', sender: 'user' }); + mockedThreadApi.appendMessage.mockResolvedValueOnce(persisted); + await store.dispatch(addMessageLocal({ threadId: 't-1', message: persisted })); + expect(store.getState().thread.messagesByThreadId['t-1']).toEqual([persisted]); + + mockedThreadApi.getThreadMessages.mockResolvedValueOnce({ + messages: [makeMessage({ id: 'srv-older', content: 'older, server-known' })], + count: 1, + }); + await store.dispatch(loadThreadMessages('t-1')); + + const merged = store.getState().thread.messagesByThreadId['t-1']; + expect(merged.map(m => m.id)).toEqual(['srv-older', 'local-new']); + }); + + it('does not duplicate a message already present in the fetched snapshot', async () => { + const store = createStore(); + const shared = makeMessage({ id: 'shared' }); + const persisted = makeMessage({ id: 'shared', content: 'server version' }); + // Prime the cache with a local entry sharing the fetched id. + mockedThreadApi.appendMessage.mockResolvedValueOnce(shared); + await store.dispatch(addMessageLocal({ threadId: 't-1', message: shared })); + + mockedThreadApi.getThreadMessages.mockResolvedValueOnce({ messages: [persisted], count: 1 }); + await store.dispatch(loadThreadMessages('t-1')); + + const merged = store.getState().thread.messagesByThreadId['t-1']; + expect(merged).toHaveLength(1); + expect(merged[0].content).toBe('server version'); + }); + + it('clears stale thread state and rejects with THREAD_NOT_FOUND_MESSAGE when the thread was deleted', async () => { + const store = createStore(); + store.dispatch(setSelectedThread('t-1')); + mockedThreadApi.getThreadMessages.mockRejectedValueOnce( + new CoreRpcError('thread t-1 not found', 'thread_not_found', undefined, { + kind: 'ThreadNotFound', + thread_id: 't-1', + }) + ); + mockedThreadApi.getThreads.mockResolvedValueOnce({ threads: [], count: 0 }); + + const result = await store.dispatch(loadThreadMessages('t-1')); + + expect(result.type).toBe('thread/loadThreadMessages/rejected'); + expect(result.payload).toBe(THREAD_NOT_FOUND_MESSAGE); + expect(store.getState().thread.selectedThreadId).toBeNull(); + expect(store.getState().thread.messagesByThreadId['t-1']).toBeUndefined(); + }); }); describe('threadSlice addMessageLocal thunk', () => { diff --git a/app/src/store/threadSlice.ts b/app/src/store/threadSlice.ts index 823bf43c8..7e32c141d 100644 --- a/app/src/store/threadSlice.ts +++ b/app/src/store/threadSlice.ts @@ -112,11 +112,21 @@ export const deleteThread = createAsyncThunk( export const loadThreadMessages = createAsyncThunk( 'thread/loadThreadMessages', - async (threadId: string, { rejectWithValue }) => { + async (threadId: string, { dispatch, rejectWithValue }) => { try { const response = await threadApi.getThreadMessages(threadId); return { threadId, messages: response.messages }; } catch (error) { + // A cached/seeded thread id (e.g. the Flows copilot's persisted + // `workflowCopilotThreads.ts` mapping) can point at a thread that was + // since deleted/purged. Evict it the same way the write thunks below + // do, so callers see a clean rejection instead of permanently retrying + // a dead thread — `useWorkflowBuilderChat`'s rehydrate effect uses this + // to null out its `threadId` and let the next send start a fresh one. + if (isThreadNotFoundCoreRpcError(error, threadId)) { + await evictStaleThread(threadId, dispatch); + return rejectWithValue(THREAD_NOT_FOUND_MESSAGE); + } return rejectWithValue(error instanceof Error ? error.message : 'Failed to load messages'); } } @@ -439,9 +449,26 @@ const threadSlice = createSlice({ }) .addCase(loadThreadMessages.fulfilled, (state, action) => { state.isLoadingMessages = false; - state.messagesByThreadId[action.payload.threadId] = action.payload.messages; - if (action.payload.threadId === state.selectedThreadId) { - state.messages = action.payload.messages; + const { threadId, messages: fetched } = action.payload; + const existing = state.messagesByThreadId[threadId] ?? []; + const fetchedIds = new Set(fetched.map(m => m.id)); + // A message present locally but missing from this fetch already + // persisted server-side (cache entries only ever land via a + // `.fulfilled` append) — it just hasn't shown up in this particular + // GET yet (e.g. a `loadThreadMessages` rehydrate racing a concurrent + // `addMessageLocal`/`addInferenceResponse` append on the same + // thread). Merge it back in instead of wholesale-replacing, or the + // fetch would silently wipe the newer message out of the visible + // transcript. See `useWorkflowBuilderChat`'s copilot-thread rehydrate + // effect, which can run concurrently with an auto-sent turn. + const localOnly = existing.filter(m => !fetchedIds.has(m.id)); + const messages = + localOnly.length > 0 + ? [...fetched, ...localOnly].sort((a, b) => a.createdAt.localeCompare(b.createdAt)) + : fetched; + state.messagesByThreadId[threadId] = messages; + if (threadId === state.selectedThreadId) { + state.messages = messages; } }) .addCase(loadThreadMessages.rejected, (state, action) => {