From 1cf828497873424c010fd2e5067e03bb23305e9d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 16:07:22 -0700 Subject: [PATCH 1/7] fix(mothership): drop queued tab adds the server already removed A server-applied remove effect cleared the tab locally but left the client's queued add for the same resource in the persistence queue, so the next flush re-posted it and hydration re-projected the tab. The deleted resource then went out as an open tab on every later message. The remove effect now discards that pending add (the server already removed it, so no client delete is sent). Active-tab resolution now logs a missing resource at info and a revoked one at warn instead of error; other failures still log at error. --- .../hooks/stream/handle-resource-event.ts | 1 + .../home/hooks/stream/stream-context.ts | 4 ++ .../home/hooks/stream/stream-test-helpers.ts | 1 + .../home/hooks/use-chat.dom.test.tsx | 64 +++++++++++++++++++ .../[workspaceId]/home/hooks/use-chat.ts | 17 +++++ .../mothership/chat/process-contents.test.ts | 18 ++++++ .../lib/mothership/chat/process-contents.ts | 16 ++++- .../resources/client-persistence-queue.ts | 19 ++++++ 8 files changed, 138 insertions(+), 2 deletions(-) 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..294df957373 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 @@ -250,6 +250,7 @@ export function handleResourceEvent(ctx: StreamLoopContext, parsed: ResourceEven if (payload.op === MothershipStreamV1ResourceOp.remove) { const resourceType = resource.type if (payload.effectId) { + ctx.deps.discardPendingResourceUpsert(resource) applyResourceEffect(ctx, (current) => current.filter((item) => getChatResourceKey(item) !== getChatResourceKey(resource)) ) 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..48c193f6853 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 @@ -749,6 +749,70 @@ describe('useChat remount send recovery', () => { } ) + it('does not re-add a tab whose queued add the server removal outran', async () => { + const history: MothershipChatHistory = { + id: 'chat-removed-file', + title: 'Files', + messages: [], + activeStreamId: null, + resources: [], + } + const persistedIds: string[] = [] + mockRequestJson.mockImplementation(async (contract: AnyApiRouteContract, input) => { + if (contract.path === '/api/mothership/chat/resources' && contract.method === 'POST') { + const { body } = input as { body: { resource: { id: string } } } + persistedIds.push(body.resource.id) + throw new Error('Persist failed') + } + return { chats: [] } + }) + let turn = 0 + 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)) + state.postBodies.push(sent) + turn += 1 + const envelope = { v: 1 as const, ts: '', stream: { streamId: sent.userMessageId } } + const file = { type: 'file' as const, id: 'deleted-file', title: 'report.csv' } + const events: MothershipStreamV1EventEnvelope[] = + turn === 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' }, + }, + { ...envelope, seq: 3, type: 'complete', payload: { status: 'complete' } }, + ] + : [{ ...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': history.id }, + }) + }) + const { getResult } = renderUseChatInChat(history.id, history) + + await act(async () => { + await getResult().sendMessage('Export the report, then delete it') + }) + await act(async () => { + queryClient.setQueryData(mothershipChatKeys.detail(history.id), { ...history }) + await sleep(0) + }) + await act(async () => { + await getResult().sendMessage('What is open?') + }) + + expect(persistedIds).toEqual(['deleted-file']) + expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file') + expect(JSON.stringify(state.postBodies.at(-1)?.resourceAttachments ?? [])).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..da9f4a04cea 100644 --- a/apps/sim/lib/mothership/resources/client-persistence-queue.ts +++ b/apps/sim/lib/mothership/resources/client-persistence-queue.ts @@ -121,6 +121,25 @@ 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. + */ + discardUpsert( + type: MothershipResource['type'], + id: string, + scopeId: string, + workspaceId?: string + ): void { + const key = this.getKey(scopeId, type, id, workspaceId) + if (this.pendingRemovals.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. From 7e3ce08b83972d730a22350d7d6feb4328ef282a Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 16:11:29 -0700 Subject: [PATCH 2/7] fix(mothership): type the stale-tab regression test's stream events --- .../[workspaceId]/home/hooks/use-chat.dom.test.tsx | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) 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 48c193f6853..4b07a70a6f0 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 @@ -90,6 +90,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' @@ -766,17 +767,17 @@ describe('useChat remount send recovery', () => { } return { chats: [] } }) - let turn = 0 + 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)) - state.postBodies.push(sent) - turn += 1 + sentAttachments.push(sent.resourceAttachments) + const turn = sentAttachments.length const envelope = { v: 1 as const, ts: '', stream: { streamId: sent.userMessageId } } const file = { type: 'file' as const, id: 'deleted-file', title: 'report.csv' } - const events: MothershipStreamV1EventEnvelope[] = + const events: PersistedStreamEventEnvelope[] = turn === 1 ? [ { ...envelope, seq: 1, type: 'resource', payload: { op: 'upsert', resource: file } }, @@ -808,9 +809,7 @@ describe('useChat remount send recovery', () => { expect(persistedIds).toEqual(['deleted-file']) expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file') - expect(JSON.stringify(state.postBodies.at(-1)?.resourceAttachments ?? [])).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 () => { From b99ed4468fb131614ff67838e5d463f9d2b89844 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 16:18:13 -0700 Subject: [PATCH 3/7] fix(mothership): close in-flight and recovery gaps for server-removed tabs An add already in flight when the server's removal arrives can commit after it, so the queue now schedules a compensating delete ordered after that write settles. Queued or failed adds are still dropped with no delete. Recovered remove effects (replay or deferred batch) now discard the queued add too; previously the recovery path returned before the discard ran. --- .../hooks/stream/handle-resource-event.ts | 7 +- .../home/hooks/use-chat.dom.test.tsx | 222 ++++++++++++++---- .../[workspaceId]/home/hooks/use-chat.ts | 58 +++-- .../resources/client-persistence-queue.ts | 8 +- 4 files changed, 225 insertions(+), 70 deletions(-) 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 294df957373..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 @@ -250,7 +256,6 @@ export function handleResourceEvent(ctx: StreamLoopContext, parsed: ResourceEven if (payload.op === MothershipStreamV1ResourceOp.remove) { const resourceType = resource.type if (payload.effectId) { - ctx.deps.discardPendingResourceUpsert(resource) applyResourceEffect(ctx, (current) => current.filter((item) => getChatResourceKey(item) !== getChatResourceKey(resource)) ) 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 4b07a70a6f0..43ee00a77b9 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 @@ -750,66 +750,192 @@ describe('useChat remount send recovery', () => { } ) - it('does not re-add a tab whose queued add the server removal outran', async () => { - const history: MothershipChatHistory = { + 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: [], } - const persistedIds: string[] = [] - mockRequestJson.mockImplementation(async (contract: AnyApiRouteContract, input) => { - if (contract.path === '/api/mothership/chat/resources' && contract.method === 'POST') { - const { body } = input as { body: { resource: { id: string } } } - persistedIds.push(body.resource.id) + + /** + * 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. Returns each send's resource attachments. + */ + function streamAddThenServerRemoval(): 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) { + for (const event of events) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)) + // Lets the client's own write for the add settle before the next event. + await sleep(10) + } + 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`. */ + function recordResourceWrites(persist: () => Promise): string[] { + const writes: string[] = [] + 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}`) + return contract.method === 'POST' ? persist() : { resources: [] } + }) + return writes + } + + it('drops a failed add without re-posting it or deleting server-side', async () => { + const writes = recordResourceWrites(async () => { throw new Error('Persist failed') - } - return { chats: [] } - }) - 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 turn = sentAttachments.length - const envelope = { v: 1 as const, ts: '', stream: { streamId: sent.userMessageId } } - const file = { type: 'file' as const, id: 'deleted-file', title: 'report.csv' } - const events: PersistedStreamEventEnvelope[] = - turn === 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' }, - }, - { ...envelope, seq: 3, type: 'complete', payload: { status: 'complete' } }, - ] - : [{ ...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': history.id }, }) - }) - const { getResult } = renderUseChatInChat(history.id, history) + const sentAttachments = streamAddThenServerRemoval() + const { getResult } = renderUseChatInChat(chat.id, chat) - await act(async () => { - await getResult().sendMessage('Export the report, then delete it') - }) - await act(async () => { - queryClient.setQueryData(mothershipChatKeys.detail(history.id), { ...history }) - await sleep(0) + 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') }) - await act(async () => { - await getResult().sendMessage('What is open?') + + it('deletes an add still in flight once it lands, after the server removal', async () => { + const add = Promise.withResolvers() + const writes = recordResourceWrites(() => add.promise) + streamAddThenServerRemoval() + const { getResult } = renderUseChatInChat(chat.id, chat) + + await act(async () => { + await getResult().sendMessage('Export the report, then delete it') + }) + expect(writes).toEqual(['POST deleted-file']) + + await act(async () => { + add.resolve({ resources: [] }) + await sleep(0) + }) + + expect(writes).toEqual(['POST deleted-file', 'DELETE deleted-file']) + expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file') }) - expect(persistedIds).toEqual(['deleted-file']) - 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 () => { 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 d0cfccfd34c..c8892ab88dc 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -263,6 +263,9 @@ interface WithdrawnSendResult { */ type StartSendMessageResult = boolean | WithdrawnSendResult +/** The fields that address a stored chat resource. */ +type ResourceIdentity = Pick + interface StartSendMessageOptions { /** Awaited before dispatch. Defaults to the hook's in-flight stop, if any. */ pendingStop?: Promise | null @@ -1479,6 +1482,18 @@ export function useChat( [queryClient, resourcePersistenceQueue] ) + const deleteStoredResource = useCallback( + (chatId: string, { type, id, workspaceId }: ResourceIdentity) => + async () => { + const result = await requestJson(removeMothershipChatResourceContract, { + body: { chatId, resourceType: type, resourceId: id, workspaceId }, + }) + await refreshResourceHistory(chatId) + return result + }, + [refreshResourceHistory] + ) + const removeResource = useCallback( (resourceType: MothershipResourceType, resourceId: string, resourceWorkspaceId?: string) => { const matches = (resource: MothershipResource) => @@ -1527,35 +1542,38 @@ export function useChat( if (wasPending && !inFlightAdd && !wasPersisted) return if (!persistChatId) return - scheduleDelete(persistChatId, async () => { - const result = await requestJson(removeMothershipChatResourceContract, { - body: { - chatId: persistChatId, - resourceType, - resourceId, - workspaceId: resourceWorkspaceId, - }, + scheduleDelete( + persistChatId, + deleteStoredResource(persistChatId, { + type: resourceType, + id: resourceId, + workspaceId: resourceWorkspaceId, }) - await refreshResourceHistory(persistChatId) - return result - }) + ) }, - [queryClient, refreshResourceHistory, resourcePersistenceQueue] + [queryClient, deleteStoredResource, resourcePersistenceQueue] ) const discardPendingResourceUpsert = useCallback( - ({ type, id, workspaceId }: Pick) => { + (resource: ResourceIdentity) => { + const { type, id, workspaceId } = resource + const persistChatId = chatIdRef.current ?? selectedChatIdRef.current // 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) + for (const scopeId of new Set([persistChatId, pendingChatKeyRef.current])) { + if (!scopeId) continue + const scheduleDelete = resourcePersistenceQueue.discardUpsert( + type, + id, + scopeId, + workspaceId + ) + if (scheduleDelete && persistChatId) { + scheduleDelete(persistChatId, deleteStoredResource(persistChatId, resource)) + } } }, - [resourcePersistenceQueue] + [deleteStoredResource, resourcePersistenceQueue] ) /** diff --git a/apps/sim/lib/mothership/resources/client-persistence-queue.ts b/apps/sim/lib/mothership/resources/client-persistence-queue.ts index da9f4a04cea..d0f78ebb6df 100644 --- a/apps/sim/lib/mothership/resources/client-persistence-queue.ts +++ b/apps/sim/lib/mothership/resources/client-persistence-queue.ts @@ -125,15 +125,21 @@ 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. + * + * An add already in flight may commit after that server removal, so for it + * this returns a scheduler for a compensating delete, ordered after the + * write settles; a queued or failed add is simply dropped. */ discardUpsert( type: MothershipResource['type'], id: string, scopeId: string, workspaceId?: string - ): void { + ): RemovedResourcePersistence['scheduleDelete'] | undefined { const key = this.getKey(scopeId, type, id, workspaceId) if (this.pendingRemovals.has(key)) return + if (this.inFlight.has(key)) + return this.remove(type, id, scopeId, false, workspaceId).scheduleDelete this.desiredUpdates.delete(key) this.failedKeys.delete(key) this.pendingKeys.delete(key) From e7f4fdd4766e2bd7755859583cc6a2987225db1f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 16:24:03 -0700 Subject: [PATCH 4/7] fix(mothership): gate the stale-tab test stream on the client's add, not the clock --- .../home/hooks/use-chat.dom.test.tsx | 44 +++++++++++++------ 1 file changed, 30 insertions(+), 14 deletions(-) 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 43ee00a77b9..7cdd3b4e3a4 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 @@ -762,9 +762,11 @@ describe('useChat remount send recovery', () => { /** * 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. Returns each send's resource attachments. + * 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(): unknown[] { + 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') { @@ -801,10 +803,15 @@ describe('useChat remount send recovery', () => { const encoder = new TextEncoder() const body = new ReadableStream({ async start(controller) { - for (const event of events) { + 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`)) - // Lets the client's own write for the add settle before the next event. - await sleep(10) } controller.close() }, @@ -816,23 +823,32 @@ describe('useChat remount send recovery', () => { return sentAttachments } - /** Resource writes the hook sent, as `METHOD id`. */ - function recordResourceWrites(persist: () => Promise): string[] { + /** + * 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 = Promise.withResolvers() 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}`) - return contract.method === 'POST' ? persist() : { resources: [] } + if (contract.method !== 'POST') return { resources: [] } + addSent.resolve() + return persist() }) - return writes + return { writes, addSent: addSent.promise } } it('drops a failed add without re-posting it or deleting server-side', async () => { - const writes = recordResourceWrites(async () => { + const { writes, addSent } = recordResourceWrites(async () => { throw new Error('Persist failed') }) - const sentAttachments = streamAddThenServerRemoval() + const sentAttachments = streamAddThenServerRemoval(addSent) const { getResult } = renderUseChatInChat(chat.id, chat) await act(async () => { @@ -855,8 +871,8 @@ describe('useChat remount send recovery', () => { it('deletes an add still in flight once it lands, after the server removal', async () => { const add = Promise.withResolvers() - const writes = recordResourceWrites(() => add.promise) - streamAddThenServerRemoval() + const { writes, addSent } = recordResourceWrites(() => add.promise) + streamAddThenServerRemoval(addSent) const { getResult } = renderUseChatInChat(chat.id, chat) await act(async () => { @@ -874,7 +890,7 @@ describe('useChat remount send recovery', () => { }) it('drops a failed add when the removal arrives in a reconnect replay', async () => { - const writes = recordResourceWrites(async () => { + const { writes } = recordResourceWrites(async () => { throw new Error('Persist failed') }) const file = { type: 'file' as const, id: 'deleted-file', title: 'report.csv' } From be11be5dc349f10a3a7f5e0229088c0306670ec5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 16:28:57 -0700 Subject: [PATCH 5/7] fix(mothership): use the shared deferred helper in stale-tab tests --- .../workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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 7cdd3b4e3a4..27728e9dcec 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' @@ -832,7 +833,7 @@ describe('useChat remount send recovery', () => { addSent: Promise } { const writes: string[] = [] - const addSent = Promise.withResolvers() + 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 } } @@ -870,7 +871,7 @@ describe('useChat remount send recovery', () => { }) it('deletes an add still in flight once it lands, after the server removal', async () => { - const add = Promise.withResolvers() + const add = createDeferred() const { writes, addSent } = recordResourceWrites(() => add.promise) streamAddThenServerRemoval(addSent) const { getResult } = renderUseChatInChat(chat.id, chat) From 65b42f029c506a105c63f4c41eb2c2fefe42d3cf Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 16:49:42 -0700 Subject: [PATCH 6/7] fix(mothership): a server restore cancels a pending compensating delete An upsert effect means the server stored the resource again, so a client delete still waiting behind an in-flight add must not run. The queue drops that removal the same way a newer add does; nothing is written since the server already holds the resource. --- .../hooks/stream/handle-resource-event.ts | 11 +++-- .../home/hooks/stream/stream-context.ts | 4 ++ .../home/hooks/stream/stream-test-helpers.ts | 1 + .../home/hooks/use-chat.dom.test.tsx | 41 +++++++++++++++++-- .../[workspaceId]/home/hooks/use-chat.ts | 14 +++++++ .../resources/client-persistence-queue.ts | 18 ++++++++ 6 files changed, 81 insertions(+), 8 deletions(-) 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 63b5c13ce35..f5eb0f87e04 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,11 +203,14 @@ 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. + if (payload.effectId && (payload.op === 'remove' || payload.op === 'upsert')) { + // The server applied this effect before publishing it, live or on + // recovery: a queued client add must not undo a removal, and a queued + // client delete must not undo a restore. const { type, id } = payload.resource - ctx.deps.discardPendingResourceUpsert({ type, id, workspaceId: payload.resource.workspaceId }) + const identity = { type, id, workspaceId: payload.resource.workspaceId } + if (payload.op === 'remove') ctx.deps.discardPendingResourceUpsert(identity) + else ctx.deps.cancelPendingResourceRemoval(identity) } if (payload.replay || ctx.deps.options.deferFlushes) { refreshRecoveredChatResources(ctx) 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 270fa35fa2e..7fad31ca26b 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 @@ -125,6 +125,10 @@ export interface StreamLoopDeps { discardPendingResourceUpsert: ( resource: Pick ) => void + /** Drops a queued client delete for a resource the server has stored again. */ + cancelPendingResourceRemoval: ( + 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 3e573b0e514..e230a1556c7 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 @@ -41,6 +41,7 @@ export function makeStreamLoopDeps(overrides: Partial = {}): Str addResource: vi.fn(() => true), removeResource: vi.fn(), discardPendingResourceUpsert: vi.fn(), + cancelPendingResourceRemoval: 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 27728e9dcec..e07dd1aad8d 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 @@ -764,10 +764,11 @@ describe('useChat remount send recovery', () => { * 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. + * so the client's own write for the add has started by then. With + * `restore`, the server then stores the file again as an effect. Returns + * each send's resource attachments. */ - function streamAddThenServerRemoval(addSent: Promise): unknown[] { + function streamAddThenServerRemoval(addSent: Promise, restore = false): unknown[] { const sentAttachments: unknown[] = [] vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => { if (String(input) !== '/api/mothership/chat' || init?.method !== 'POST') { @@ -798,7 +799,21 @@ describe('useChat remount send recovery', () => { type: 'resource', payload: { op: 'remove', resource: file, effectId: 'remove-deleted-file' }, }, - complete, + ...(restore + ? [ + { + ...envelope, + seq: 3, + type: 'resource' as const, + payload: { + op: 'upsert' as const, + resource: file, + effectId: 'restore-file', + }, + }, + ] + : []), + { ...complete, seq: 4 }, ] : [complete] const encoder = new TextEncoder() @@ -890,6 +905,24 @@ describe('useChat remount send recovery', () => { expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file') }) + it('keeps a tab the server restored after removing it while its add was in flight', async () => { + const add = createDeferred() + const { writes, addSent } = recordResourceWrites(() => add.promise) + streamAddThenServerRemoval(addSent, true) + const { getResult } = renderUseChatInChat(chat.id, chat) + + await act(async () => { + await getResult().sendMessage('Export the report, delete it, then restore it') + }) + await act(async () => { + add.resolve({ resources: [] }) + await sleep(0) + }) + + expect(writes).toEqual(['POST deleted-file']) + expect(getResult().resources.map((resource) => resource.id)).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') 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 c8892ab88dc..e44079d91cf 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -1576,6 +1576,18 @@ export function useChat( [deleteStoredResource, resourcePersistenceQueue] ) + const cancelPendingResourceRemoval = useCallback( + ({ type, id, workspaceId }: ResourceIdentity) => { + for (const scopeId of new Set([ + chatIdRef.current ?? selectedChatIdRef.current, + pendingChatKeyRef.current, + ])) { + if (scopeId) resourcePersistenceQueue.cancelRemoval(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 @@ -2407,6 +2419,7 @@ export function useChat( addResource, removeResource, discardPendingResourceUpsert, + cancelPendingResourceRemoval, startClientWorkflowTool, startClientLocalFilesystemTool: (toolCallId, toolName, toolArgs) => startClientLocalFilesystemTool(toolCallId, toolName, toolArgs, desktopTurn), @@ -2552,6 +2565,7 @@ export function useChat( addResource, removeResource, discardPendingResourceUpsert, + cancelPendingResourceRemoval, startClientWorkflowTool, startClientLocalFilesystemTool, startClientTerminalTool, diff --git a/apps/sim/lib/mothership/resources/client-persistence-queue.ts b/apps/sim/lib/mothership/resources/client-persistence-queue.ts index d0f78ebb6df..2a9f50cff90 100644 --- a/apps/sim/lib/mothership/resources/client-persistence-queue.ts +++ b/apps/sim/lib/mothership/resources/client-persistence-queue.ts @@ -146,6 +146,24 @@ export class ResourcePersistenceQueue { this.persistedKeys.delete(key) } + /** + * Cancels a delete still waiting to run for a resource the server has since + * stored again on its own, as a newer add would. The server already holds + * the resource, so nothing is written. + */ + cancelRemoval( + type: MothershipResource['type'], + id: string, + scopeId: string, + workspaceId?: string + ): void { + const key = this.getKey(scopeId, type, id, workspaceId) + if (!this.removalTokens.delete(key)) return + this.pendingRemovals.delete(key) + this.failedKeys.delete(key) + if (!this.desiredUpdates.has(key)) this.pendingKeys.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. From 8bd8c38c1c9a5f0d87d8547f38c499dcd219188a Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 17:08:32 -0700 Subject: [PATCH 7/7] fix(mothership): scope stale-tab fix to queued and failed adds Drops the in-flight compensating delete and the restore cancel path: each added a new client/server ordering hazard. A server remove effect, live or on recovery, now only discards a queued or failed client add for that resource; a write already in flight is left untouched. --- .../hooks/stream/handle-resource-event.ts | 11 ++-- .../home/hooks/stream/stream-context.ts | 4 -- .../home/hooks/stream/stream-test-helpers.ts | 1 - .../home/hooks/use-chat.dom.test.tsx | 61 ++--------------- .../[workspaceId]/home/hooks/use-chat.ts | 66 +++++-------------- .../resources/client-persistence-queue.ts | 31 ++------- 6 files changed, 29 insertions(+), 145 deletions(-) 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 f5eb0f87e04..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,14 +203,11 @@ export function handleResourceEvent(ctx: StreamLoopContext, parsed: ResourceEven notifyWorkflowExternalUpdate(payload.resource.id) } if (payload.op === 'refresh') return - if (payload.effectId && (payload.op === 'remove' || payload.op === 'upsert')) { - // The server applied this effect before publishing it, live or on - // recovery: a queued client add must not undo a removal, and a queued - // client delete must not undo a restore. + 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 - const identity = { type, id, workspaceId: payload.resource.workspaceId } - if (payload.op === 'remove') ctx.deps.discardPendingResourceUpsert(identity) - else ctx.deps.cancelPendingResourceRemoval(identity) + ctx.deps.discardPendingResourceUpsert({ type, id, workspaceId: payload.resource.workspaceId }) } if (payload.replay || ctx.deps.options.deferFlushes) { refreshRecoveredChatResources(ctx) 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 7fad31ca26b..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 @@ -125,10 +125,6 @@ export interface StreamLoopDeps { discardPendingResourceUpsert: ( resource: Pick ) => void - /** Drops a queued client delete for a resource the server has stored again. */ - cancelPendingResourceRemoval: ( - 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 e230a1556c7..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 @@ -41,7 +41,6 @@ export function makeStreamLoopDeps(overrides: Partial = {}): Str addResource: vi.fn(() => true), removeResource: vi.fn(), discardPendingResourceUpsert: vi.fn(), - cancelPendingResourceRemoval: 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 e07dd1aad8d..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 @@ -764,11 +764,10 @@ describe('useChat remount send recovery', () => { * 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. With - * `restore`, the server then stores the file again as an effect. Returns - * each send's resource attachments. + * so the client's own write for the add has started by then. Returns each + * send's resource attachments. */ - function streamAddThenServerRemoval(addSent: Promise, restore = false): unknown[] { + 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') { @@ -799,21 +798,7 @@ describe('useChat remount send recovery', () => { type: 'resource', payload: { op: 'remove', resource: file, effectId: 'remove-deleted-file' }, }, - ...(restore - ? [ - { - ...envelope, - seq: 3, - type: 'resource' as const, - payload: { - op: 'upsert' as const, - resource: file, - effectId: 'restore-file', - }, - }, - ] - : []), - { ...complete, seq: 4 }, + complete, ] : [complete] const encoder = new TextEncoder() @@ -885,44 +870,6 @@ describe('useChat remount send recovery', () => { expect(JSON.stringify(sentAttachments.at(-1) ?? [])).not.toContain('deleted-file') }) - it('deletes an add still in flight once it lands, after the server removal', async () => { - const add = createDeferred() - const { writes, addSent } = recordResourceWrites(() => add.promise) - streamAddThenServerRemoval(addSent) - const { getResult } = renderUseChatInChat(chat.id, chat) - - await act(async () => { - await getResult().sendMessage('Export the report, then delete it') - }) - expect(writes).toEqual(['POST deleted-file']) - - await act(async () => { - add.resolve({ resources: [] }) - await sleep(0) - }) - - expect(writes).toEqual(['POST deleted-file', 'DELETE deleted-file']) - expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file') - }) - - it('keeps a tab the server restored after removing it while its add was in flight', async () => { - const add = createDeferred() - const { writes, addSent } = recordResourceWrites(() => add.promise) - streamAddThenServerRemoval(addSent, true) - const { getResult } = renderUseChatInChat(chat.id, chat) - - await act(async () => { - await getResult().sendMessage('Export the report, delete it, then restore it') - }) - await act(async () => { - add.resolve({ resources: [] }) - await sleep(0) - }) - - expect(writes).toEqual(['POST deleted-file']) - expect(getResult().resources.map((resource) => resource.id)).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') 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 e44079d91cf..d0cfccfd34c 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -263,9 +263,6 @@ interface WithdrawnSendResult { */ type StartSendMessageResult = boolean | WithdrawnSendResult -/** The fields that address a stored chat resource. */ -type ResourceIdentity = Pick - interface StartSendMessageOptions { /** Awaited before dispatch. Defaults to the hook's in-flight stop, if any. */ pendingStop?: Promise | null @@ -1482,18 +1479,6 @@ export function useChat( [queryClient, resourcePersistenceQueue] ) - const deleteStoredResource = useCallback( - (chatId: string, { type, id, workspaceId }: ResourceIdentity) => - async () => { - const result = await requestJson(removeMothershipChatResourceContract, { - body: { chatId, resourceType: type, resourceId: id, workspaceId }, - }) - await refreshResourceHistory(chatId) - return result - }, - [refreshResourceHistory] - ) - const removeResource = useCallback( (resourceType: MothershipResourceType, resourceId: string, resourceWorkspaceId?: string) => { const matches = (resource: MothershipResource) => @@ -1542,47 +1527,32 @@ export function useChat( if (wasPending && !inFlightAdd && !wasPersisted) return if (!persistChatId) return - scheduleDelete( - persistChatId, - deleteStoredResource(persistChatId, { - type: resourceType, - id: resourceId, - workspaceId: resourceWorkspaceId, + scheduleDelete(persistChatId, async () => { + const result = await requestJson(removeMothershipChatResourceContract, { + body: { + chatId: persistChatId, + resourceType, + resourceId, + workspaceId: resourceWorkspaceId, + }, }) - ) + await refreshResourceHistory(persistChatId) + return result + }) }, - [queryClient, deleteStoredResource, resourcePersistenceQueue] + [queryClient, refreshResourceHistory, resourcePersistenceQueue] ) const discardPendingResourceUpsert = useCallback( - (resource: ResourceIdentity) => { - const { type, id, workspaceId } = resource - const persistChatId = chatIdRef.current ?? selectedChatIdRef.current + ({ 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. - for (const scopeId of new Set([persistChatId, pendingChatKeyRef.current])) { - if (!scopeId) continue - const scheduleDelete = resourcePersistenceQueue.discardUpsert( - type, - id, - scopeId, - workspaceId - ) - if (scheduleDelete && persistChatId) { - scheduleDelete(persistChatId, deleteStoredResource(persistChatId, resource)) - } - } - }, - [deleteStoredResource, resourcePersistenceQueue] - ) - - const cancelPendingResourceRemoval = useCallback( - ({ type, id, workspaceId }: ResourceIdentity) => { - for (const scopeId of new Set([ + const scopes = new Set([ chatIdRef.current ?? selectedChatIdRef.current, pendingChatKeyRef.current, - ])) { - if (scopeId) resourcePersistenceQueue.cancelRemoval(type, id, scopeId, workspaceId) + ]) + for (const scopeId of scopes) { + if (scopeId) resourcePersistenceQueue.discardUpsert(type, id, scopeId, workspaceId) } }, [resourcePersistenceQueue] @@ -2419,7 +2389,6 @@ export function useChat( addResource, removeResource, discardPendingResourceUpsert, - cancelPendingResourceRemoval, startClientWorkflowTool, startClientLocalFilesystemTool: (toolCallId, toolName, toolArgs) => startClientLocalFilesystemTool(toolCallId, toolName, toolArgs, desktopTurn), @@ -2565,7 +2534,6 @@ export function useChat( addResource, removeResource, discardPendingResourceUpsert, - cancelPendingResourceRemoval, startClientWorkflowTool, startClientLocalFilesystemTool, startClientTerminalTool, diff --git a/apps/sim/lib/mothership/resources/client-persistence-queue.ts b/apps/sim/lib/mothership/resources/client-persistence-queue.ts index 2a9f50cff90..43d0340a485 100644 --- a/apps/sim/lib/mothership/resources/client-persistence-queue.ts +++ b/apps/sim/lib/mothership/resources/client-persistence-queue.ts @@ -124,46 +124,23 @@ 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. - * - * An add already in flight may commit after that server removal, so for it - * this returns a scheduler for a compensating delete, ordered after the - * write settles; a queued or failed add is simply dropped. + * 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 - ): RemovedResourcePersistence['scheduleDelete'] | undefined { + ): void { const key = this.getKey(scopeId, type, id, workspaceId) - if (this.pendingRemovals.has(key)) return - if (this.inFlight.has(key)) - return this.remove(type, id, scopeId, false, workspaceId).scheduleDelete + 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) } - /** - * Cancels a delete still waiting to run for a resource the server has since - * stored again on its own, as a newer add would. The server already holds - * the resource, so nothing is written. - */ - cancelRemoval( - type: MothershipResource['type'], - id: string, - scopeId: string, - workspaceId?: string - ): void { - const key = this.getKey(scopeId, type, id, workspaceId) - if (!this.removalTokens.delete(key)) return - this.pendingRemovals.delete(key) - this.failedKeys.delete(key) - if (!this.desiredUpdates.has(key)) this.pendingKeys.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.