diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.ts index 0b895f780f1..63b5c13ce35 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.ts @@ -203,6 +203,12 @@ export function handleResourceEvent(ctx: StreamLoopContext, parsed: ResourceEven notifyWorkflowExternalUpdate(payload.resource.id) } if (payload.op === 'refresh') return + if (payload.op === 'remove' && payload.effectId) { + // The server applied this removal before publishing it, so a client add + // still queued for the resource must not restore it, live or on recovery. + const { type, id } = payload.resource + ctx.deps.discardPendingResourceUpsert({ type, id, workspaceId: payload.resource.workspaceId }) + } if (payload.replay || ctx.deps.options.deferFlushes) { refreshRecoveredChatResources(ctx) return diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-context.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-context.ts index aafb389ad18..270fa35fa2e 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-context.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-context.ts @@ -121,6 +121,10 @@ export interface StreamLoopDeps { resourceId: string, workspaceId?: string ) => void + /** Drops a queued client write that would re-add a resource the server already removed. */ + discardPendingResourceUpsert: ( + resource: Pick + ) => void startClientWorkflowTool: (id: string, name: string, args: Record) => void startClientLocalFilesystemTool: (id: string, name: string, args: Record) => void startClientComputerTool: (id: string, args: Record, ts?: string) => void diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-test-helpers.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-test-helpers.ts index 8b1903f8168..3e573b0e514 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-test-helpers.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-test-helpers.ts @@ -40,6 +40,7 @@ export function makeStreamLoopDeps(overrides: Partial = {}): Str setActiveResourceId: vi.fn(), addResource: vi.fn(() => true), removeResource: vi.fn(), + discardPendingResourceUpsert: vi.fn(), startClientWorkflowTool: vi.fn(), startClientLocalFilesystemTool: vi.fn(), startClientComputerTool: vi.fn(), diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx index adb0af8dbe8..6e4ab2eb4e0 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx @@ -20,6 +20,7 @@ */ import { type ReactNode, act as reactAct, StrictMode, useEffect, useState } from 'react' +import { createDeferred } from '@sim/testing/helpers/deferred' import { authClientMock, authClientMockFns } from '@sim/testing/mocks/auth-client.mock' import { libDesktopMock, libDesktopMockFns } from '@sim/testing/mocks/lib-desktop.mock' import { nextNavigationMock, nextNavigationMockFns } from '@sim/testing/mocks/next-navigation.mock' @@ -90,6 +91,7 @@ import { import { MothershipHandoffStorage } from '@/lib/core/utils/browser-storage' import { MOTHERSHIP_STREAM_REPLAY_HEADER } from '@/lib/mothership/constants' import type { MothershipStreamV1EventEnvelope } from '@/lib/mothership/generated/mothership-stream-v1' +import type { PersistedStreamEventEnvelope } from '@/lib/mothership/request/session/contract' import { getChatResourceSelectionId } from '@/lib/mothership/resources/types' import { ChatSurfaceProvider } from '@/app/workspace/[workspaceId]/home/components/chat-surface-context' import { collectCitedMessageSources } from '@/app/workspace/[workspaceId]/home/components/message-content/message-sources' @@ -749,6 +751,190 @@ describe('useChat remount send recovery', () => { } ) + describe('a tab the server removed while the client was still adding it', () => { + const chat: MothershipChatHistory = { + id: 'chat-removed-file', + title: 'Files', + messages: [], + activeStreamId: null, + resources: [], + } + + /** + * Streams a turn that opens a file tab without an effect (the client + * persists it), then the server's own removal of that file as an effect. + * Later turns only complete. The removal is held until `addSent` resolves, + * so the client's own write for the add has started by then. Returns each + * send's resource attachments. + */ + function streamAddThenServerRemoval(addSent: Promise): unknown[] { + const sentAttachments: unknown[] = [] + vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => { + if (String(input) !== '/api/mothership/chat' || init?.method !== 'POST') { + return fetchStub(input, init) + } + const sent = JSON.parse(String(init.body)) + sentAttachments.push(sent.resourceAttachments) + const envelope = { v: 1 as const, ts: '', stream: { streamId: sent.userMessageId } } + const file = { type: 'file' as const, id: 'deleted-file', title: 'report.csv' } + const complete: PersistedStreamEventEnvelope = { + ...envelope, + seq: 3, + type: 'complete', + payload: { status: 'complete' }, + } + const events: PersistedStreamEventEnvelope[] = + sentAttachments.length === 1 + ? [ + { + ...envelope, + seq: 1, + type: 'resource', + payload: { op: 'upsert', resource: file }, + }, + { + ...envelope, + seq: 2, + type: 'resource', + payload: { op: 'remove', resource: file, effectId: 'remove-deleted-file' }, + }, + complete, + ] + : [complete] + const encoder = new TextEncoder() + const body = new ReadableStream({ + async start(controller) { + const [first, ...rest] = events + controller.enqueue(encoder.encode(`data: ${JSON.stringify(first)}\n\n`)) + if (rest.length > 0) { + await addSent + // One macrotask lets a write that already failed settle in the queue. + await sleep(0) + } + for (const event of rest) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)) + } + controller.close() + }, + }) + return new Response(body, { + headers: { 'Content-Type': 'text/event-stream', 'x-mothership-chat-id': chat.id }, + }) + }) + return sentAttachments + } + + /** + * Resource writes the hook sent, as `METHOD id`, and a promise that + * resolves once the first add has been sent. + */ + function recordResourceWrites(persist: () => Promise): { + writes: string[] + addSent: Promise + } { + const writes: string[] = [] + const addSent = createDeferred() + mockRequestJson.mockImplementation(async (contract: AnyApiRouteContract, input) => { + if (contract.path !== '/api/mothership/chat/resources') return { chats: [] } + const { body } = input as { body: { resource?: { id: string }; resourceId?: string } } + writes.push(`${contract.method} ${body.resource?.id ?? body.resourceId}`) + if (contract.method !== 'POST') return { resources: [] } + addSent.resolve() + return persist() + }) + return { writes, addSent: addSent.promise } + } + + it('drops a failed add without re-posting it or deleting server-side', async () => { + const { writes, addSent } = recordResourceWrites(async () => { + throw new Error('Persist failed') + }) + const sentAttachments = streamAddThenServerRemoval(addSent) + const { getResult } = renderUseChatInChat(chat.id, chat) + + await act(async () => { + await getResult().sendMessage('Export the report, then delete it') + }) + const writesBeforeRemoval = [...writes] + await act(async () => { + queryClient.setQueryData(mothershipChatKeys.detail(chat.id), { ...chat }) + await sleep(0) + }) + await act(async () => { + await getResult().sendMessage('What is open?') + }) + + expect(writesBeforeRemoval).not.toContain('DELETE deleted-file') + expect(writes).toEqual(writesBeforeRemoval) + expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file') + expect(JSON.stringify(sentAttachments.at(-1) ?? [])).not.toContain('deleted-file') + }) + + it('drops a failed add when the removal arrives in a reconnect replay', async () => { + const { writes } = recordResourceWrites(async () => { + throw new Error('Persist failed') + }) + const file = { type: 'file' as const, id: 'deleted-file', title: 'report.csv' } + const sentAttachments: unknown[] = [] + let streamId = '' + vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input) + if (url.includes('/api/mothership/chat/stream')) { + const removal: PersistedStreamEventEnvelope = { + v: 1, + ts: '', + stream: { streamId }, + seq: 2, + type: 'resource', + payload: { op: 'remove', resource: file, effectId: 'remove-deleted-file' }, + } + const opened: PersistedStreamEventEnvelope = { + ...removal, + seq: 1, + payload: { op: 'upsert', resource: file }, + } + return Response.json({ + success: true, + events: [opened, removal].map((event) => ({ eventId: event.seq, streamId, event })), + status: 'complete', + }) + } + if (url !== '/api/mothership/chat' || init?.method !== 'POST') { + return fetchStub(input, init) + } + const sent = JSON.parse(String(init.body)) + sentAttachments.push(sent.resourceAttachments) + streamId = sent.userMessageId + const envelope = { v: 1 as const, ts: '', stream: { streamId } } + const events: PersistedStreamEventEnvelope[] = + sentAttachments.length === 1 + ? [{ ...envelope, seq: 1, type: 'resource', payload: { op: 'upsert', resource: file } }] + : [{ ...envelope, seq: 1, type: 'complete', payload: { status: 'complete' } }] + return new Response(events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(''), { + headers: { 'Content-Type': 'text/event-stream', 'x-mothership-chat-id': chat.id }, + }) + }) + const { getResult } = renderUseChatInChat(chat.id, chat) + + await act(async () => { + await getResult().sendMessage('Export the report, then delete it') + }) + await waitFor(() => !getResult().isSending) + const writesBeforeRemoval = [...writes] + await act(async () => { + queryClient.setQueryData(mothershipChatKeys.detail(chat.id), { ...chat }) + await sleep(0) + }) + await act(async () => { + await getResult().sendMessage('What is open?') + }) + + expect(writes).toEqual(writesBeforeRemoval) + expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file') + expect(JSON.stringify(sentAttachments.at(-1) ?? [])).not.toContain('deleted-file') + }) + }) + it('restores an explicitly selected Search tab while reconnecting an active turn', async () => { const shape = resolveDeploymentShape() seedDeploymentShape({ ...shape, features: shape.features }) diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts index c3ec9381f46..d0cfccfd34c 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -1543,6 +1543,21 @@ export function useChat( [queryClient, refreshResourceHistory, resourcePersistenceQueue] ) + const discardPendingResourceUpsert = useCallback( + ({ type, id, workspaceId }: Pick) => { + // An add queued before the chat id resolved stays in the provisional + // bucket until a flush adopts it, so both scopes may hold it. + const scopes = new Set([ + chatIdRef.current ?? selectedChatIdRef.current, + pendingChatKeyRef.current, + ]) + for (const scopeId of scopes) { + if (scopeId) resourcePersistenceQueue.discardUpsert(type, id, scopeId, workspaceId) + } + }, + [resourcePersistenceQueue] + ) + /** * Drops hydrated workflow tabs whose workflow no longer exists, so an old * chat cannot resurrect a deleted workflow. The workspace list is the fast @@ -2373,6 +2388,7 @@ export function useChat( setActiveResourceId, addResource, removeResource, + discardPendingResourceUpsert, startClientWorkflowTool, startClientLocalFilesystemTool: (toolCallId, toolName, toolArgs) => startClientLocalFilesystemTool(toolCallId, toolName, toolArgs, desktopTurn), @@ -2517,6 +2533,7 @@ export function useChat( queryClient, addResource, removeResource, + discardPendingResourceUpsert, startClientWorkflowTool, startClientLocalFilesystemTool, startClientTerminalTool, diff --git a/apps/sim/lib/mothership/chat/process-contents.test.ts b/apps/sim/lib/mothership/chat/process-contents.test.ts index 9ffa9e61917..249b0cf8e97 100644 --- a/apps/sim/lib/mothership/chat/process-contents.test.ts +++ b/apps/sim/lib/mothership/chat/process-contents.test.ts @@ -38,6 +38,7 @@ import { } from '@sim/testing/mocks/workspace-file-manager.mock' import { beforeEach, describe, expect, it, type Mock, vi } from 'vitest' import { admitContent, DelegatedWorkspaceAuthorizationError } from '@/lib/core/application' +import { OrchestrationError } from '@/lib/core/orchestration/types' import { MAX_TABLE_SELECTION_PREVIEW_LENGTH, MAX_TABLE_SELECTION_ROWS, @@ -971,6 +972,23 @@ describe('file folder context', () => { resolveFileFolderPath.mockResolvedValueOnce({ path: null }) expect(await resolveActiveResourceContext('filefolder', 'missing', 'ws-1', 'reader')).toBeNull() }) + + it.each([ + { tab: 'a deleted file', failure: new OrchestrationError('not_found', 'gone'), level: 'info' }, + { + tab: 'a revoked file', + failure: new OrchestrationError('forbidden', 'denied'), + level: 'warn', + }, + { tab: 'a failed lookup', failure: new Error('connection reset'), level: 'error' }, + ] as const)('drops $tab, logging it at $level', async ({ failure, level }) => { + readWorkspaceFileMetadata.mockRejectedValueOnce(failure) + expect(await resolveActiveResourceContext('file', 'file-1', 'ws-1', 'reader')).toBeNull() + const logger = getMockLogger('ProcessContents') + for (const method of ['info', 'warn', 'error'] as const) { + expect(logger[method]).toHaveBeenCalledTimes(method === level ? 1 : 0) + } + }) }) describe('processContextsServer - file_selection contexts', () => { diff --git a/apps/sim/lib/mothership/chat/process-contents.ts b/apps/sim/lib/mothership/chat/process-contents.ts index 6e4f456b83a..3763159b997 100644 --- a/apps/sim/lib/mothership/chat/process-contents.ts +++ b/apps/sim/lib/mothership/chat/process-contents.ts @@ -10,7 +10,7 @@ import { eq } from 'drizzle-orm' import type { MothershipTableViewContext } from '@/lib/api/contracts/mothership-resources' import { observeContentDelivery } from '@/lib/core/application' import { getAllowedIntegrationsFromEnv } from '@/lib/core/config/env-flags' -import { OrchestrationError } from '@/lib/core/orchestration/types' +import { asOrchestrationError, OrchestrationError } from '@/lib/core/orchestration/types' import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import { readWorkspaceDashboard } from '@/lib/dashboards/application/dashboards' import { isIntegrationDeploymentAvailableForVisibility } from '@/lib/integrations/availability.server' @@ -925,7 +925,19 @@ export async function resolveActiveResourceContext( return null } } catch (error) { - logger.error('Failed to resolve active resource context', { resourceType, resourceId, error }) + // A tab can outlive its resource or the viewer's access to it; the + // request simply goes without that context. + const code = asOrchestrationError(error)?.code + if (code === 'not_found') { + logger.info('Skipped active resource that no longer exists', { resourceType, resourceId }) + } else if (code === 'forbidden') { + logger.warn('Skipped active resource the user can no longer access', { + resourceType, + resourceId, + }) + } else { + logger.error('Failed to resolve active resource context', { resourceType, resourceId, error }) + } return null } } diff --git a/apps/sim/lib/mothership/resources/client-persistence-queue.ts b/apps/sim/lib/mothership/resources/client-persistence-queue.ts index 3f8fe699421..43d0340a485 100644 --- a/apps/sim/lib/mothership/resources/client-persistence-queue.ts +++ b/apps/sim/lib/mothership/resources/client-persistence-queue.ts @@ -121,6 +121,26 @@ export class ResourcePersistenceQueue { } } + /** + * Forgets a write that would add back a resource the server has already + * removed on its own, so neither a later flush nor the projection over + * server history resurrects it. A removal the user queued still runs, and a + * write already in flight is left to its own bookkeeping. + */ + discardUpsert( + type: MothershipResource['type'], + id: string, + scopeId: string, + workspaceId?: string + ): void { + const key = this.getKey(scopeId, type, id, workspaceId) + if (this.pendingRemovals.has(key) || this.inFlight.has(key)) return + this.desiredUpdates.delete(key) + this.failedKeys.delete(key) + this.pendingKeys.delete(key) + this.persistedKeys.delete(key) + } + /** * Whether a pending write would change which resources the chat holds — an * add the server has not accepted yet, or a delete that has not landed.