fix(mothership): drop queued tab adds the server already removed

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

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<MothershipResource, 'type' | 'id' | 'workspaceId'>
+  ) => void
   startClientWorkflowTool: (id: string, name: string, args: Record<string, unknown>) => void
   startClientLocalFilesystemTool: (id: string, name: string, args: Record<string, unknown>) => void
   startClientComputerTool: (id: string, args: Record<string, unknown>, ts?: string) => void

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-test-helpers.ts

@@ -40,6 +40,7 @@ export function makeStreamLoopDeps(overrides: Partial<StreamLoopDeps> = {}): Str
     setActiveResourceId: vi.fn(),
     addResource: vi.fn(() => true),
     removeResource: vi.fn(),
+    discardPendingResourceUpsert: vi.fn(),
     startClientWorkflowTool: vi.fn(),
     startClientLocalFilesystemTool: vi.fn(),
     startClientComputerTool: vi.fn(),

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<void>): 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<Uint8Array>({
+          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<unknown>): {
+      writes: string[]
+      addSent: Promise<void>
+    } {
+      const writes: string[] = []
+      const addSent = createDeferred<void>()
+      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 })

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<MothershipResource, 'type' | 'id' | 'workspaceId'>) => {
+      // 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,

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', () => {

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
   }
 }

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.