Skip to content

Commit 29158c6

Browse files
authored
fix(mothership): drop queued tab adds the server already removed (#9050)
* 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. * fix(mothership): type the stale-tab regression test's stream events * 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. * fix(mothership): gate the stale-tab test stream on the client's add, not the clock * fix(mothership): use the shared deferred helper in stale-tab tests * 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. * 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.
1 parent 3ba393d commit 29158c6

8 files changed

Lines changed: 266 additions & 2 deletions

File tree

‎apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -203,6 +203,12 @@ export function handleResourceEvent(ctx: StreamLoopContext, parsed: ResourceEven
203203
notifyWorkflowExternalUpdate(payload.resource.id)
204204
}
205205
if (payload.op === 'refresh') return
206+
if (payload.op === 'remove' && payload.effectId) {
207+
// The server applied this removal before publishing it, so a client add
208+
// still queued for the resource must not restore it, live or on recovery.
209+
const { type, id } = payload.resource
210+
ctx.deps.discardPendingResourceUpsert({ type, id, workspaceId: payload.resource.workspaceId })
211+
}
206212
if (payload.replay || ctx.deps.options.deferFlushes) {
207213
refreshRecoveredChatResources(ctx)
208214
return

‎apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-context.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,10 @@ export interface StreamLoopDeps {
121121
resourceId: string,
122122
workspaceId?: string
123123
) => void
124+
/** Drops a queued client write that would re-add a resource the server already removed. */
125+
discardPendingResourceUpsert: (
126+
resource: Pick<MothershipResource, 'type' | 'id' | 'workspaceId'>
127+
) => void
124128
startClientWorkflowTool: (id: string, name: string, args: Record<string, unknown>) => void
125129
startClientLocalFilesystemTool: (id: string, name: string, args: Record<string, unknown>) => void
126130
startClientComputerTool: (id: string, args: Record<string, unknown>, ts?: string) => void

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ export function makeStreamLoopDeps(overrides: Partial<StreamLoopDeps> = {}): Str
4040
setActiveResourceId: vi.fn(),
4141
addResource: vi.fn(() => true),
4242
removeResource: vi.fn(),
43+
discardPendingResourceUpsert: vi.fn(),
4344
startClientWorkflowTool: vi.fn(),
4445
startClientLocalFilesystemTool: vi.fn(),
4546
startClientComputerTool: vi.fn(),

‎apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
*/
2121

2222
import { type ReactNode, act as reactAct, StrictMode, useEffect, useState } from 'react'
23+
import { createDeferred } from '@sim/testing/helpers/deferred'
2324
import { authClientMock, authClientMockFns } from '@sim/testing/mocks/auth-client.mock'
2425
import { libDesktopMock, libDesktopMockFns } from '@sim/testing/mocks/lib-desktop.mock'
2526
import { nextNavigationMock, nextNavigationMockFns } from '@sim/testing/mocks/next-navigation.mock'
@@ -90,6 +91,7 @@ import {
9091
import { MothershipHandoffStorage } from '@/lib/core/utils/browser-storage'
9192
import { MOTHERSHIP_STREAM_REPLAY_HEADER } from '@/lib/mothership/constants'
9293
import type { MothershipStreamV1EventEnvelope } from '@/lib/mothership/generated/mothership-stream-v1'
94+
import type { PersistedStreamEventEnvelope } from '@/lib/mothership/request/session/contract'
9395
import { getChatResourceSelectionId } from '@/lib/mothership/resources/types'
9496
import { ChatSurfaceProvider } from '@/app/workspace/[workspaceId]/home/components/chat-surface-context'
9597
import { collectCitedMessageSources } from '@/app/workspace/[workspaceId]/home/components/message-content/message-sources'
@@ -749,6 +751,190 @@ describe('useChat remount send recovery', () => {
749751
}
750752
)
751753

754+
describe('a tab the server removed while the client was still adding it', () => {
755+
const chat: MothershipChatHistory = {
756+
id: 'chat-removed-file',
757+
title: 'Files',
758+
messages: [],
759+
activeStreamId: null,
760+
resources: [],
761+
}
762+
763+
/**
764+
* Streams a turn that opens a file tab without an effect (the client
765+
* persists it), then the server's own removal of that file as an effect.
766+
* Later turns only complete. The removal is held until `addSent` resolves,
767+
* so the client's own write for the add has started by then. Returns each
768+
* send's resource attachments.
769+
*/
770+
function streamAddThenServerRemoval(addSent: Promise<void>): unknown[] {
771+
const sentAttachments: unknown[] = []
772+
vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
773+
if (String(input) !== '/api/mothership/chat' || init?.method !== 'POST') {
774+
return fetchStub(input, init)
775+
}
776+
const sent = JSON.parse(String(init.body))
777+
sentAttachments.push(sent.resourceAttachments)
778+
const envelope = { v: 1 as const, ts: '', stream: { streamId: sent.userMessageId } }
779+
const file = { type: 'file' as const, id: 'deleted-file', title: 'report.csv' }
780+
const complete: PersistedStreamEventEnvelope = {
781+
...envelope,
782+
seq: 3,
783+
type: 'complete',
784+
payload: { status: 'complete' },
785+
}
786+
const events: PersistedStreamEventEnvelope[] =
787+
sentAttachments.length === 1
788+
? [
789+
{
790+
...envelope,
791+
seq: 1,
792+
type: 'resource',
793+
payload: { op: 'upsert', resource: file },
794+
},
795+
{
796+
...envelope,
797+
seq: 2,
798+
type: 'resource',
799+
payload: { op: 'remove', resource: file, effectId: 'remove-deleted-file' },
800+
},
801+
complete,
802+
]
803+
: [complete]
804+
const encoder = new TextEncoder()
805+
const body = new ReadableStream<Uint8Array>({
806+
async start(controller) {
807+
const [first, ...rest] = events
808+
controller.enqueue(encoder.encode(`data: ${JSON.stringify(first)}\n\n`))
809+
if (rest.length > 0) {
810+
await addSent
811+
// One macrotask lets a write that already failed settle in the queue.
812+
await sleep(0)
813+
}
814+
for (const event of rest) {
815+
controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`))
816+
}
817+
controller.close()
818+
},
819+
})
820+
return new Response(body, {
821+
headers: { 'Content-Type': 'text/event-stream', 'x-mothership-chat-id': chat.id },
822+
})
823+
})
824+
return sentAttachments
825+
}
826+
827+
/**
828+
* Resource writes the hook sent, as `METHOD id`, and a promise that
829+
* resolves once the first add has been sent.
830+
*/
831+
function recordResourceWrites(persist: () => Promise<unknown>): {
832+
writes: string[]
833+
addSent: Promise<void>
834+
} {
835+
const writes: string[] = []
836+
const addSent = createDeferred<void>()
837+
mockRequestJson.mockImplementation(async (contract: AnyApiRouteContract, input) => {
838+
if (contract.path !== '/api/mothership/chat/resources') return { chats: [] }
839+
const { body } = input as { body: { resource?: { id: string }; resourceId?: string } }
840+
writes.push(`${contract.method} ${body.resource?.id ?? body.resourceId}`)
841+
if (contract.method !== 'POST') return { resources: [] }
842+
addSent.resolve()
843+
return persist()
844+
})
845+
return { writes, addSent: addSent.promise }
846+
}
847+
848+
it('drops a failed add without re-posting it or deleting server-side', async () => {
849+
const { writes, addSent } = recordResourceWrites(async () => {
850+
throw new Error('Persist failed')
851+
})
852+
const sentAttachments = streamAddThenServerRemoval(addSent)
853+
const { getResult } = renderUseChatInChat(chat.id, chat)
854+
855+
await act(async () => {
856+
await getResult().sendMessage('Export the report, then delete it')
857+
})
858+
const writesBeforeRemoval = [...writes]
859+
await act(async () => {
860+
queryClient.setQueryData(mothershipChatKeys.detail(chat.id), { ...chat })
861+
await sleep(0)
862+
})
863+
await act(async () => {
864+
await getResult().sendMessage('What is open?')
865+
})
866+
867+
expect(writesBeforeRemoval).not.toContain('DELETE deleted-file')
868+
expect(writes).toEqual(writesBeforeRemoval)
869+
expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file')
870+
expect(JSON.stringify(sentAttachments.at(-1) ?? [])).not.toContain('deleted-file')
871+
})
872+
873+
it('drops a failed add when the removal arrives in a reconnect replay', async () => {
874+
const { writes } = recordResourceWrites(async () => {
875+
throw new Error('Persist failed')
876+
})
877+
const file = { type: 'file' as const, id: 'deleted-file', title: 'report.csv' }
878+
const sentAttachments: unknown[] = []
879+
let streamId = ''
880+
vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
881+
const url = String(input)
882+
if (url.includes('/api/mothership/chat/stream')) {
883+
const removal: PersistedStreamEventEnvelope = {
884+
v: 1,
885+
ts: '',
886+
stream: { streamId },
887+
seq: 2,
888+
type: 'resource',
889+
payload: { op: 'remove', resource: file, effectId: 'remove-deleted-file' },
890+
}
891+
const opened: PersistedStreamEventEnvelope = {
892+
...removal,
893+
seq: 1,
894+
payload: { op: 'upsert', resource: file },
895+
}
896+
return Response.json({
897+
success: true,
898+
events: [opened, removal].map((event) => ({ eventId: event.seq, streamId, event })),
899+
status: 'complete',
900+
})
901+
}
902+
if (url !== '/api/mothership/chat' || init?.method !== 'POST') {
903+
return fetchStub(input, init)
904+
}
905+
const sent = JSON.parse(String(init.body))
906+
sentAttachments.push(sent.resourceAttachments)
907+
streamId = sent.userMessageId
908+
const envelope = { v: 1 as const, ts: '', stream: { streamId } }
909+
const events: PersistedStreamEventEnvelope[] =
910+
sentAttachments.length === 1
911+
? [{ ...envelope, seq: 1, type: 'resource', payload: { op: 'upsert', resource: file } }]
912+
: [{ ...envelope, seq: 1, type: 'complete', payload: { status: 'complete' } }]
913+
return new Response(events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(''), {
914+
headers: { 'Content-Type': 'text/event-stream', 'x-mothership-chat-id': chat.id },
915+
})
916+
})
917+
const { getResult } = renderUseChatInChat(chat.id, chat)
918+
919+
await act(async () => {
920+
await getResult().sendMessage('Export the report, then delete it')
921+
})
922+
await waitFor(() => !getResult().isSending)
923+
const writesBeforeRemoval = [...writes]
924+
await act(async () => {
925+
queryClient.setQueryData(mothershipChatKeys.detail(chat.id), { ...chat })
926+
await sleep(0)
927+
})
928+
await act(async () => {
929+
await getResult().sendMessage('What is open?')
930+
})
931+
932+
expect(writes).toEqual(writesBeforeRemoval)
933+
expect(getResult().resources.map((resource) => resource.id)).not.toContain('deleted-file')
934+
expect(JSON.stringify(sentAttachments.at(-1) ?? [])).not.toContain('deleted-file')
935+
})
936+
})
937+
752938
it('restores an explicitly selected Search tab while reconnecting an active turn', async () => {
753939
const shape = resolveDeploymentShape()
754940
seedDeploymentShape({ ...shape, features: shape.features })

‎apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1543,6 +1543,21 @@ export function useChat(
15431543
[queryClient, refreshResourceHistory, resourcePersistenceQueue]
15441544
)
15451545

1546+
const discardPendingResourceUpsert = useCallback(
1547+
({ type, id, workspaceId }: Pick<MothershipResource, 'type' | 'id' | 'workspaceId'>) => {
1548+
// An add queued before the chat id resolved stays in the provisional
1549+
// bucket until a flush adopts it, so both scopes may hold it.
1550+
const scopes = new Set([
1551+
chatIdRef.current ?? selectedChatIdRef.current,
1552+
pendingChatKeyRef.current,
1553+
])
1554+
for (const scopeId of scopes) {
1555+
if (scopeId) resourcePersistenceQueue.discardUpsert(type, id, scopeId, workspaceId)
1556+
}
1557+
},
1558+
[resourcePersistenceQueue]
1559+
)
1560+
15461561
/**
15471562
* Drops hydrated workflow tabs whose workflow no longer exists, so an old
15481563
* chat cannot resurrect a deleted workflow. The workspace list is the fast
@@ -2373,6 +2388,7 @@ export function useChat(
23732388
setActiveResourceId,
23742389
addResource,
23752390
removeResource,
2391+
discardPendingResourceUpsert,
23762392
startClientWorkflowTool,
23772393
startClientLocalFilesystemTool: (toolCallId, toolName, toolArgs) =>
23782394
startClientLocalFilesystemTool(toolCallId, toolName, toolArgs, desktopTurn),
@@ -2517,6 +2533,7 @@ export function useChat(
25172533
queryClient,
25182534
addResource,
25192535
removeResource,
2536+
discardPendingResourceUpsert,
25202537
startClientWorkflowTool,
25212538
startClientLocalFilesystemTool,
25222539
startClientTerminalTool,

‎apps/sim/lib/mothership/chat/process-contents.test.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ import {
3838
} from '@sim/testing/mocks/workspace-file-manager.mock'
3939
import { beforeEach, describe, expect, it, type Mock, vi } from 'vitest'
4040
import { admitContent, DelegatedWorkspaceAuthorizationError } from '@/lib/core/application'
41+
import { OrchestrationError } from '@/lib/core/orchestration/types'
4142
import {
4243
MAX_TABLE_SELECTION_PREVIEW_LENGTH,
4344
MAX_TABLE_SELECTION_ROWS,
@@ -971,6 +972,23 @@ describe('file folder context', () => {
971972
resolveFileFolderPath.mockResolvedValueOnce({ path: null })
972973
expect(await resolveActiveResourceContext('filefolder', 'missing', 'ws-1', 'reader')).toBeNull()
973974
})
975+
976+
it.each([
977+
{ tab: 'a deleted file', failure: new OrchestrationError('not_found', 'gone'), level: 'info' },
978+
{
979+
tab: 'a revoked file',
980+
failure: new OrchestrationError('forbidden', 'denied'),
981+
level: 'warn',
982+
},
983+
{ tab: 'a failed lookup', failure: new Error('connection reset'), level: 'error' },
984+
] as const)('drops $tab, logging it at $level', async ({ failure, level }) => {
985+
readWorkspaceFileMetadata.mockRejectedValueOnce(failure)
986+
expect(await resolveActiveResourceContext('file', 'file-1', 'ws-1', 'reader')).toBeNull()
987+
const logger = getMockLogger('ProcessContents')
988+
for (const method of ['info', 'warn', 'error'] as const) {
989+
expect(logger[method]).toHaveBeenCalledTimes(method === level ? 1 : 0)
990+
}
991+
})
974992
})
975993

976994
describe('processContextsServer - file_selection contexts', () => {

‎apps/sim/lib/mothership/chat/process-contents.ts‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import { eq } from 'drizzle-orm'
1010
import type { MothershipTableViewContext } from '@/lib/api/contracts/mothership-resources'
1111
import { observeContentDelivery } from '@/lib/core/application'
1212
import { getAllowedIntegrationsFromEnv } from '@/lib/core/config/env-flags'
13-
import { OrchestrationError } from '@/lib/core/orchestration/types'
13+
import { asOrchestrationError, OrchestrationError } from '@/lib/core/orchestration/types'
1414
import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
1515
import { readWorkspaceDashboard } from '@/lib/dashboards/application/dashboards'
1616
import { isIntegrationDeploymentAvailableForVisibility } from '@/lib/integrations/availability.server'
@@ -925,7 +925,19 @@ export async function resolveActiveResourceContext(
925925
return null
926926
}
927927
} catch (error) {
928-
logger.error('Failed to resolve active resource context', { resourceType, resourceId, error })
928+
// A tab can outlive its resource or the viewer's access to it; the
929+
// request simply goes without that context.
930+
const code = asOrchestrationError(error)?.code
931+
if (code === 'not_found') {
932+
logger.info('Skipped active resource that no longer exists', { resourceType, resourceId })
933+
} else if (code === 'forbidden') {
934+
logger.warn('Skipped active resource the user can no longer access', {
935+
resourceType,
936+
resourceId,
937+
})
938+
} else {
939+
logger.error('Failed to resolve active resource context', { resourceType, resourceId, error })
940+
}
929941
return null
930942
}
931943
}

‎apps/sim/lib/mothership/resources/client-persistence-queue.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,26 @@ export class ResourcePersistenceQueue {
121121
}
122122
}
123123

124+
/**
125+
* Forgets a write that would add back a resource the server has already
126+
* removed on its own, so neither a later flush nor the projection over
127+
* server history resurrects it. A removal the user queued still runs, and a
128+
* write already in flight is left to its own bookkeeping.
129+
*/
130+
discardUpsert(
131+
type: MothershipResource['type'],
132+
id: string,
133+
scopeId: string,
134+
workspaceId?: string
135+
): void {
136+
const key = this.getKey(scopeId, type, id, workspaceId)
137+
if (this.pendingRemovals.has(key) || this.inFlight.has(key)) return
138+
this.desiredUpdates.delete(key)
139+
this.failedKeys.delete(key)
140+
this.pendingKeys.delete(key)
141+
this.persistedKeys.delete(key)
142+
}
143+
124144
/**
125145
* Whether a pending write would change which resources the chat holds — an
126146
* add the server has not accepted yet, or a delete that has not landed.

0 commit comments

Comments
 (0)