From d0f9d5625eca66e5f7abfa76ca9ed5d408e41f64 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 21 Aug 2026 00:44:37 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=A7=B5=20fix:=20Close=20Child-Thread=20Re?= =?UTF-8?q?ad=20and=20Search-Cleanup=20Gaps=20(#15055)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: close child thread read and cleanup gaps * fix: preserve child search cleanup invariants * test: complete mocked update result * perf: parallelize scoped message reads * fix: close child thread compatibility gaps * test: expect preserved cleanup failure * fix: reconcile legacy Meili cleanup markers --- api/db/indexSync.js | 93 +++++---- api/db/indexSync.spec.js | 58 +++++- .../__tests__/validateMessageReq.spec.js | 42 ++++ api/server/middleware/index.js | 7 +- api/server/routes/__tests__/convos.spec.js | 31 ++- .../messages-get-real-validation.spec.js | 26 +++ .../routes/__tests__/messages-get.spec.js | 122 ++++++++++- api/server/routes/convos.js | 2 +- api/server/routes/messages.js | 42 +++- .../api/src/middleware/messageValidation.ts | 19 +- .../src/methods/conversation.spec.ts | 26 +++ .../data-schemas/src/methods/conversation.ts | 10 +- .../src/models/plugins/mongoMeili.spec.ts | 178 +++++++++++++++- .../src/models/plugins/mongoMeili.ts | 191 +++++++++++++----- 14 files changed, 732 insertions(+), 115 deletions(-) diff --git a/api/db/indexSync.js b/api/db/indexSync.js index 9a24edc4de..54b2861723 100644 --- a/api/db/indexSync.js +++ b/api/db/indexSync.js @@ -230,44 +230,54 @@ async function performSync(flowManager, flowId, flowType) { await batchResetMeiliFlags(Conversation.collection); } - // Check if we need to sync messages - logger.info('[indexSync] Requesting message sync progress...'); - const messageProgress = await Message.getSyncProgress(); - if (!messageProgress.isComplete || settingsUpdated) { - logger.info( - `[indexSync] Messages need syncing: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments} indexed`, - ); - - const messageCount = messageProgress.totalDocuments; - const messagesIndexed = messageProgress.totalProcessed; - const unindexedMessages = messageCount - messagesIndexed; - const messagesPendingCleanup = messageProgress.pendingCleanup ?? 0; - const noneIndexed = messagesIndexed === 0 && unindexedMessages > 0; - - if ( - settingsUpdated || - noneIndexed || - unindexedMessages > syncThreshold || - messagesPendingCleanup > 0 - ) { - if (noneIndexed && !settingsUpdated) { - logger.info('[indexSync] No messages marked as indexed, forcing full sync'); - } + let messageSyncError; + try { + // Check if we need to sync messages + logger.info('[indexSync] Requesting message sync progress...'); + const messageProgress = await Message.getSyncProgress(); + if (!messageProgress.isComplete || settingsUpdated) { logger.info( - messagesPendingCleanup > 0 - ? `[indexSync] Starting message sync (${unindexedMessages} unindexed, ${messagesPendingCleanup} pending cleanup)` - : `[indexSync] Starting message sync (${unindexedMessages} unindexed)`, + `[indexSync] Messages need syncing: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments} indexed`, ); - await Message.syncWithMeili(); - messagesSync = true; - } else if (unindexedMessages > 0) { + + const messageCount = messageProgress.totalDocuments; + const messagesIndexed = messageProgress.totalProcessed; + const unindexedMessages = messageCount - messagesIndexed; + const messagesPendingCleanup = messageProgress.pendingCleanup ?? 0; + const noneIndexed = messagesIndexed === 0 && unindexedMessages > 0; + + if (settingsUpdated || noneIndexed || unindexedMessages > syncThreshold) { + if (noneIndexed && !settingsUpdated) { + logger.info('[indexSync] No messages marked as indexed, forcing full sync'); + } + logger.info( + messagesPendingCleanup > 0 + ? `[indexSync] Starting message sync (${unindexedMessages} unindexed, ${messagesPendingCleanup} pending cleanup)` + : `[indexSync] Starting message sync (${unindexedMessages} unindexed)`, + ); + await Message.syncWithMeili(); + messagesSync = true; + } else if (messagesPendingCleanup > 0) { + logger.info( + `[indexSync] Cleaning ${messagesPendingCleanup} excluded messages from search`, + ); + await Message.cleanupExcludedMeiliIndex(); + messagesSync = true; + } else if (unindexedMessages > 0) { + logger.info( + `[indexSync] ${unindexedMessages} messages unindexed (below threshold: ${syncThreshold}, skipping)`, + ); + } + } else { logger.info( - `[indexSync] ${unindexedMessages} messages unindexed (below threshold: ${syncThreshold}, skipping)`, + `[indexSync] Messages are fully synced: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments}`, ); } - } else { - logger.info( - `[indexSync] Messages are fully synced: ${messageProgress.totalProcessed}/${messageProgress.totalDocuments}`, + } catch (error) { + messageSyncError = error; + logger.error( + '[indexSync] Message reconciliation failed; continuing with conversations:', + error, ); } @@ -284,12 +294,7 @@ async function performSync(flowManager, flowId, flowType) { const convosPendingCleanup = convoProgress.pendingCleanup ?? 0; const noneConvosIndexed = convosIndexed === 0 && unindexedConvos > 0; - if ( - settingsUpdated || - noneConvosIndexed || - unindexedConvos > syncThreshold || - convosPendingCleanup > 0 - ) { + if (settingsUpdated || noneConvosIndexed || unindexedConvos > syncThreshold) { if (noneConvosIndexed && !settingsUpdated) { logger.info('[indexSync] No conversations marked as indexed, forcing full sync'); } @@ -300,6 +305,12 @@ async function performSync(flowManager, flowId, flowType) { ); await Conversation.syncWithMeili(); convosSync = true; + } else if (convosPendingCleanup > 0) { + logger.info( + `[indexSync] Cleaning ${convosPendingCleanup} excluded conversations from search`, + ); + await Conversation.cleanupExcludedMeiliIndex(); + convosSync = true; } else if (unindexedConvos > 0) { logger.info( `[indexSync] ${unindexedConvos} convos unindexed (below threshold: ${syncThreshold}, skipping)`, @@ -311,6 +322,10 @@ async function performSync(flowManager, flowId, flowType) { ); } + if (messageSyncError) { + throw messageSyncError; + } + return { messagesSync, convosSync }; } finally { if (indexingDisabled === true) { diff --git a/api/db/indexSync.spec.js b/api/db/indexSync.spec.js index 9191714dce..a1a23db64d 100644 --- a/api/db/indexSync.spec.js +++ b/api/db/indexSync.spec.js @@ -25,6 +25,7 @@ const createMockModel = (collectionName) => ({ collection: { name: collectionName }, getSyncProgress: jest.fn(), syncWithMeili: jest.fn(), + cleanupExcludedMeiliIndex: jest.fn(), countDocuments: jest.fn(), }); @@ -528,7 +529,7 @@ describe('performSync() - syncThreshold logic', () => { ); }); - test('forces sync when search contains documents that are now excluded', async () => { + test('runs bounded cleanup when search contains documents that are now excluded', async () => { Message.getSyncProgress.mockResolvedValue({ totalProcessed: 100, totalDocuments: 100, @@ -545,10 +546,61 @@ describe('performSync() - syncThreshold logic', () => { const indexSync = require('./indexSync'); await indexSync(); - expect(Message.syncWithMeili).toHaveBeenCalledTimes(1); + expect(Message.syncWithMeili).not.toHaveBeenCalled(); + expect(Message.cleanupExcludedMeiliIndex).toHaveBeenCalledTimes(1); expect(Conversation.syncWithMeili).not.toHaveBeenCalled(); expect(mockLogger.info).toHaveBeenCalledWith( - '[indexSync] Starting message sync (0 unindexed, 1 pending cleanup)', + '[indexSync] Cleaning 1 excluded messages from search', + ); + }); + + test('does not start cleanup for excluded documents that were never indexed', async () => { + Message.getSyncProgress.mockResolvedValue({ + totalProcessed: 100, + totalDocuments: 100, + pendingCleanup: 0, + isComplete: true, + }); + Conversation.getSyncProgress.mockResolvedValue({ + totalProcessed: 50, + totalDocuments: 50, + pendingCleanup: 0, + isComplete: true, + }); + + const indexSync = require('./indexSync'); + await indexSync(); + + expect(Message.syncWithMeili).not.toHaveBeenCalled(); + expect(Message.cleanupExcludedMeiliIndex).not.toHaveBeenCalled(); + expect(Conversation.syncWithMeili).not.toHaveBeenCalled(); + expect(Conversation.cleanupExcludedMeiliIndex).not.toHaveBeenCalled(); + }); + + test('continues conversation cleanup when message cleanup fails transiently', async () => { + const cleanupError = new Error('message cleanup timed out'); + Message.getSyncProgress.mockResolvedValue({ + totalProcessed: 100, + totalDocuments: 100, + pendingCleanup: 1, + isComplete: false, + }); + Message.cleanupExcludedMeiliIndex.mockRejectedValue(cleanupError); + Conversation.getSyncProgress.mockResolvedValue({ + totalProcessed: 50, + totalDocuments: 50, + pendingCleanup: 1, + isComplete: false, + }); + + const indexSync = require('./indexSync'); + await expect(indexSync()).rejects.toThrow(cleanupError); + + expect(Message.cleanupExcludedMeiliIndex).toHaveBeenCalledTimes(1); + expect(Conversation.cleanupExcludedMeiliIndex).toHaveBeenCalledTimes(1); + expect(mockLogger.error).toHaveBeenCalledWith( + '[indexSync] Message reconciliation failed; continuing with conversations:', + cleanupError, ); }); }); diff --git a/api/server/middleware/__tests__/validateMessageReq.spec.js b/api/server/middleware/__tests__/validateMessageReq.spec.js index 997286dfe7..9d32611144 100644 --- a/api/server/middleware/__tests__/validateMessageReq.spec.js +++ b/api/server/middleware/__tests__/validateMessageReq.spec.js @@ -89,6 +89,48 @@ describe('validateMessageReq', () => { expect(next).toHaveBeenCalledTimes(1); }); + it('returns not found for a direct child-thread message read', async () => { + const req = { + method: 'GET', + params: { conversationId: 'child-convo', messageId: 'child-message' }, + body: {}, + user: { id: userId }, + }; + const res = createResponse(); + const next = jest.fn(); + getConvoOwnership.mockResolvedValue({ + user: userId, + subagentThread: { parentConversationId: 'parent-convo' }, + }); + + await validateMessageReq(req, res, next); + + expect(res.status).toHaveBeenCalledWith(404); + expect(res.json).toHaveBeenCalledWith({ error: 'Conversation not found' }); + expect(next).not.toHaveBeenCalled(); + }); + + it('returns not found for a HEAD request to a child thread', async () => { + const req = { + method: 'HEAD', + params: { conversationId: 'child-convo' }, + body: {}, + user: { id: userId }, + }; + const res = createResponse(); + const next = jest.fn(); + getConvoOwnership.mockResolvedValue({ + user: userId, + subagentThread: { parentConversationId: 'parent-convo' }, + }); + + await validateMessageReq(req, res, next); + + expect(res.status).toHaveBeenCalledWith(404); + expect(res.json).toHaveBeenCalledWith({ error: 'Conversation not found' }); + expect(next).not.toHaveBeenCalled(); + }); + it('should allow message reads for an owned active generation job before the conversation is saved', async () => { const req = { method: 'GET', diff --git a/api/server/middleware/index.js b/api/server/middleware/index.js index 42fdb9b1d7..775d140e24 100644 --- a/api/server/middleware/index.js +++ b/api/server/middleware/index.js @@ -4,7 +4,11 @@ const validateRegistration = require('./validateRegistration'); const buildEndpointOption = require('./buildEndpointOption'); const validateEmailLogin = require('./validateEmailLogin'); const validateMessageReq = require('./validateMessageReq'); -const { prepareMessageRequestValidation, sendValidationResponse } = require('./messageValidation'); +const { + canReadActiveJobConversation, + prepareMessageRequestValidation, + sendValidationResponse, +} = require('./messageValidation'); const checkDomainAllowed = require('./checkDomainAllowed'); const requireLocalAuth = require('./requireLocalAuth'); const canDeleteAccount = require('./canDeleteAccount'); @@ -49,6 +53,7 @@ module.exports = { configMiddleware, checkDomainAllowed, validateMessageReq, + canReadActiveJobConversation, sendValidationResponse, prepareMessageRequestValidation, buildEndpointOption, diff --git a/api/server/routes/__tests__/convos.spec.js b/api/server/routes/__tests__/convos.spec.js index c09a8ac075..c5c7ee03be 100644 --- a/api/server/routes/__tests__/convos.spec.js +++ b/api/server/routes/__tests__/convos.spec.js @@ -25,7 +25,7 @@ jest.mock('~/server/services/Endpoints/agents/subagentThreadStore', () => describe('Convos Routes', () => { let app; let convosRouter; - const { deleteToolCalls, deleteConvos, saveConvo } = require('~/models'); + const { deleteToolCalls, deleteConvos, getConvo, saveConvo } = require('~/models'); const { deleteAgentCheckpoints, deleteAllSharedLinksWithCleanup, @@ -52,6 +52,35 @@ describe('Convos Routes', () => { jest.clearAllMocks(); }); + describe('GET /:conversationId', () => { + it('returns an ordinary owned conversation', async () => { + getConvo.mockResolvedValue({ conversationId: 'ordinary', title: 'Ordinary' }); + + const response = await request(app).get('/api/convos/ordinary'); + + expect(response.status).toBe(200); + expect(response.body).toEqual({ conversationId: 'ordinary', title: 'Ordinary' }); + expect(getConvo).toHaveBeenCalledWith('test-user-123', 'ordinary'); + }); + + it('returns the same not-found response for an owned child thread', async () => { + getConvo.mockResolvedValue({ + conversationId: 'child', + subagentThread: { parentConversationId: 'parent' }, + }); + + const childResponse = await request(app).get('/api/convos/child'); + getConvo.mockResolvedValue(null); + const missingResponse = await request(app).get('/api/convos/missing'); + + expect(childResponse.status).toBe(404); + expect(childResponse.text).toBe(''); + expect(childResponse.status).toBe(missingResponse.status); + expect(childResponse.text).toBe(missingResponse.text); + expect(getConvo).toHaveBeenNthCalledWith(1, 'test-user-123', 'child'); + }); + }); + describe('DELETE /all', () => { it('prunes the deleted conversations’ agent checkpoints (bulk, ids from deleteConvos)', async () => { // HITL: a paused conversation's durable checkpoint must not outlive the conversation. diff --git a/api/server/routes/__tests__/messages-get-real-validation.spec.js b/api/server/routes/__tests__/messages-get-real-validation.spec.js index cc4c3956cb..79ca934736 100644 --- a/api/server/routes/__tests__/messages-get-real-validation.spec.js +++ b/api/server/routes/__tests__/messages-get-real-validation.spec.js @@ -169,4 +169,30 @@ describe('GET /api/messages/:conversationId with real validation middleware', () expect(response.status).toBe(200); expect(response.body).toEqual([{ messageId: 'message-1', conversationId: 'convo-1' }]); }); + + it('does not return messages for a directly addressed child thread', async () => { + getConvoOwnership.mockResolvedValue({ + conversationId: 'child-convo', + user: authenticatedUserId, + subagentThread: { parentConversationId: 'parent-convo' }, + }); + getMessages.mockResolvedValue([{ messageId: 'child-message', conversationId: 'child-convo' }]); + + const response = await request(app).get('/api/messages/child-convo'); + + expect(response.status).toBe(404); + expect(response.body).toEqual({ error: 'Conversation not found' }); + }); + + it('does not expose a directly addressed child thread through HEAD', async () => { + getConvoOwnership.mockResolvedValue({ + conversationId: 'child-convo', + user: authenticatedUserId, + subagentThread: { parentConversationId: 'parent-convo' }, + }); + + const response = await request(app).head('/api/messages/child-convo'); + + expect(response.status).toBe(404); + }); }); diff --git a/api/server/routes/__tests__/messages-get.spec.js b/api/server/routes/__tests__/messages-get.spec.js index 2e26f20ee3..0c8ca9749a 100644 --- a/api/server/routes/__tests__/messages-get.spec.js +++ b/api/server/routes/__tests__/messages-get.spec.js @@ -36,6 +36,7 @@ jest.mock('~/models', () => ({ getMessage: jest.fn(), saveMessage: jest.fn(), getMessages: jest.fn(), + getConvoOwnership: jest.fn(), updateMessage: jest.fn(), deleteMessages: jest.fn(), getConvosQueried: jest.fn(), @@ -52,6 +53,7 @@ jest.mock('~/server/middleware/requireJwtAuth', () => (req, res, next) => next() jest.mock('~/server/middleware', () => { const validateMessageReq = jest.fn((req, res, next) => next()); + const canReadActiveJobConversation = jest.fn().mockResolvedValue(false); const prepareMessageRequestValidation = jest.fn((req, res, next) => { req.messageRequestValidation = { conversationId: 'convo-1', @@ -70,6 +72,7 @@ jest.mock('~/server/middleware', () => { return { requireJwtAuth: (req, res, next) => next(), validateMessageReq, + canReadActiveJobConversation, sendValidationResponse, prepareMessageRequestValidation, configMiddleware: (req, res, next) => next(), @@ -86,8 +89,17 @@ jest.mock('~/db/models', () => ({ describe('message route conversation ownership filters', () => { let app; - const { getMessages, saveConvo, saveMessage } = require('~/models'); - const { prepareMessageRequestValidation } = require('~/server/middleware'); + const { + getConvoOwnership, + getMessages, + getMessagesByCursor, + saveConvo, + saveMessage, + } = require('~/models'); + const { + canReadActiveJobConversation, + prepareMessageRequestValidation, + } = require('~/server/middleware'); const authenticatedUserId = 'user-owner-123'; @@ -105,6 +117,7 @@ describe('message route conversation ownership filters', () => { beforeEach(() => { jest.clearAllMocks(); + canReadActiveJobConversation.mockResolvedValue(false); prepareMessageRequestValidation.mockImplementation((req, res, next) => { req.messageRequestValidation = { conversationId: 'convo-1', @@ -180,6 +193,111 @@ describe('message route conversation ownership filters', () => { ); }); + it('returns indistinguishable not-found responses for child and missing query reads', async () => { + getConvoOwnership.mockResolvedValueOnce({ + user: authenticatedUserId, + subagentThread: { parentConversationId: 'parent-convo' }, + }); + + const childResponse = await request(app).get( + '/api/messages?conversationId=child-convo&messageId=child-message', + ); + getConvoOwnership.mockResolvedValueOnce(null); + const missingResponse = await request(app).get( + '/api/messages?conversationId=missing-convo&messageId=missing-message', + ); + + expect(childResponse.status).toBe(404); + expect(childResponse.body).toEqual({ error: 'Conversation not found' }); + expect(childResponse.status).toBe(missingResponse.status); + expect(childResponse.body).toEqual(missingResponse.body); + expect(getMessages).toHaveBeenCalledTimes(2); + }); + + it.each([ + { + name: 'single-message', + path: '/api/messages?conversationId=convo-1&messageId=message-1', + readMock: getMessages, + readResult: [{ messageId: 'message-1', conversationId: 'convo-1' }], + }, + { + name: 'cursor', + path: '/api/messages?conversationId=convo-1', + readMock: getMessagesByCursor, + readResult: { messages: [], nextCursor: null }, + }, + ])( + 'starts the $name query read before ownership validation resolves', + async ({ path, readMock, readResult }) => { + const events = []; + let resolveOwnership; + const ownershipPromise = new Promise((resolve) => { + resolveOwnership = resolve; + }); + getConvoOwnership.mockImplementationOnce(() => { + events.push('ownership-started'); + return ownershipPromise; + }); + + let resolveReadStarted; + const readStartedPromise = new Promise((resolve) => { + resolveReadStarted = resolve; + }); + readMock.mockImplementationOnce(() => { + events.push('messages-started'); + resolveReadStarted(); + return Promise.resolve(readResult); + }); + + const responsePromise = new Promise((resolve, reject) => { + request(app) + .get(path) + .end((error, response) => (error ? reject(error) : resolve(response))); + }); + + await Promise.race([readStartedPromise, new Promise((resolve) => setTimeout(resolve, 100))]); + const eventsBeforeValidation = [...events]; + resolveOwnership({ user: authenticatedUserId }); + const response = await responsePromise; + + expect(eventsBeforeValidation).toEqual(['ownership-started', 'messages-started']); + expect(response.status).toBe(200); + }, + ); + + it('allows an ordinary owned conversation query read', async () => { + getConvoOwnership.mockResolvedValue({ user: authenticatedUserId }); + getMessages.mockResolvedValue([{ messageId: 'message-1', conversationId: 'convo-1' }]); + + const response = await request(app).get( + '/api/messages?conversationId=convo-1&messageId=message-1', + ); + + expect(response.status).toBe(200); + expect(response.body.messages).toEqual([{ messageId: 'message-1', conversationId: 'convo-1' }]); + }); + + it('allows an owner-scoped active generation before its conversation row exists', async () => { + getConvoOwnership.mockResolvedValue(null); + canReadActiveJobConversation.mockResolvedValue(true); + getMessagesByCursor.mockResolvedValue({ + messages: [{ messageId: 'prompt-1', conversationId: 'active-convo' }], + nextCursor: null, + }); + + const response = await request(app).get('/api/messages?conversationId=active-convo'); + + expect(response.status).toBe(200); + expect(response.body.messages).toEqual([ + { messageId: 'prompt-1', conversationId: 'active-convo' }, + ]); + expect(canReadActiveJobConversation).toHaveBeenCalledWith( + expect.objectContaining({ user: { id: authenticatedUserId } }), + 'active-convo', + ); + }); + it('should start conversation message reads before validation resolves', async () => { const events = []; let resolveValidation; diff --git a/api/server/routes/convos.js b/api/server/routes/convos.js index fcba4f9a83..e7ad4e26c8 100644 --- a/api/server/routes/convos.js +++ b/api/server/routes/convos.js @@ -83,7 +83,7 @@ router.get('/:conversationId', async (req, res) => { const { conversationId } = req.params; const convo = await db.getConvo(req.user.id, conversationId); - if (convo) { + if (convo && convo.subagentThread == null) { res.status(200).json(convo); } else { res.status(404).end(); diff --git a/api/server/routes/messages.js b/api/server/routes/messages.js index eb87e7a05c..22d499bd41 100644 --- a/api/server/routes/messages.js +++ b/api/server/routes/messages.js @@ -23,6 +23,7 @@ const { validateMessageReq, configMiddleware, sendValidationResponse, + canReadActiveJobConversation, prepareMessageRequestValidation, } = require('~/server/middleware'); const db = require('~/models'); @@ -68,14 +69,45 @@ router.get('/', async (req, res) => { : 'createdAt'; const sortOrder = sortDirection === 'asc' ? 1 : -1; + let scopedMessageRead; + if (typeof conversationId === 'string') { + const ownershipRead = db.getConvoOwnership(user, conversationId); + const messageRead = messageId + ? db.getMessages({ conversationId, messageId, user }) + : db.getMessagesByCursor( + { conversationId, user }, + { sortField, sortOrder, limit: pageSize, cursor }, + ); + scopedMessageRead = Promise.resolve(messageRead).then( + (value) => ({ ok: true, value }), + (error) => ({ ok: false, error }), + ); + + const conversation = await ownershipRead; + const canReadActiveJob = + conversation == null && + !messageId && + (await canReadActiveJobConversation(req, conversationId)); + if ((!conversation && !canReadActiveJob) || conversation?.subagentThread != null) { + return res.status(404).json({ error: 'Conversation not found' }); + } + } else if (conversationId) { + return res.status(404).json({ error: 'Conversation not found' }); + } + if (conversationId && messageId) { - const messages = await db.getMessages({ conversationId, messageId, user }); + const messageResult = await scopedMessageRead; + if (!messageResult.ok) { + throw messageResult.error; + } + const messages = messageResult.value; response = { messages: messages?.length ? [messages[0]] : [], nextCursor: null }; } else if (conversationId) { - response = await db.getMessagesByCursor( - { conversationId, user }, - { sortField, sortOrder, limit: pageSize, cursor }, - ); + const messageResult = await scopedMessageRead; + if (!messageResult.ok) { + throw messageResult.error; + } + response = messageResult.value; } else if (search) { const searchResults = await db.searchMessages(search, { filter: `user = "${user}"` }, true); diff --git a/packages/api/src/middleware/messageValidation.ts b/packages/api/src/middleware/messageValidation.ts index fd003c5d5b..ba39849755 100644 --- a/packages/api/src/middleware/messageValidation.ts +++ b/packages/api/src/middleware/messageValidation.ts @@ -27,6 +27,7 @@ export type MessageValidationRequest = { type ConversationRecord = { user?: string; + subagentThread?: unknown; } | null; type PendingActionRecord = unknown; @@ -71,6 +72,10 @@ export type MessageValidationDeps = { }; export type MessageRequestMiddleware = { + canReadActiveJobConversation: ( + req: MessageValidationRequest, + conversationId?: string, + ) => Promise; createMessageRequestValidation: (req: MessageValidationRequest) => MessageRequestValidation; prepareMessageRequestValidation: ( req: MessageValidationRequest, @@ -90,6 +95,10 @@ function hasTenantMismatch(job: GenerationJobRecord, user: MessageValidationUser return job?.metadata?.tenantId != null && job.metadata.tenantId !== user.tenantId; } +function isPublicReadMethod(method?: string): boolean { + return method === 'GET' || method === 'HEAD'; +} + export function createMessageRequestMiddleware( deps: MessageValidationDeps, ): MessageRequestMiddleware { @@ -97,7 +106,7 @@ export function createMessageRequestMiddleware( req: MessageValidationRequest, conversationId?: string, ): Promise { - if (req.method !== 'GET' || req.params?.messageId) { + if (!isPublicReadMethod(req.method) || req.params?.messageId) { return false; } @@ -153,6 +162,13 @@ export function createMessageRequestMiddleware( }; } + // Child threads are internal execution records, not standalone public + // conversations. Keep the same response as a missing conversation so the + // read boundary does not disclose whether a supplied child id exists. + if (isPublicReadMethod(req.method) && conversation.subagentThread != null) { + return { ok: false, status: 404, body: { error: 'Conversation not found' } }; + } + return { ok: true }; } @@ -225,6 +241,7 @@ export function createMessageRequestMiddleware( } return { + canReadActiveJobConversation, createMessageRequestValidation: createMessageRequestValidation, prepareMessageRequestValidation: prepareMessageRequestValidation, sendValidationResponse: sendValidationResponse, diff --git a/packages/data-schemas/src/methods/conversation.spec.ts b/packages/data-schemas/src/methods/conversation.spec.ts index fc38719eba..7323c12a06 100644 --- a/packages/data-schemas/src/methods/conversation.spec.ts +++ b/packages/data-schemas/src/methods/conversation.spec.ts @@ -1518,6 +1518,32 @@ describe('Conversation Operations', () => { ).toBeNull(); expect(await methods.getConvoOwnership('user123', 'non-existent-id')).toBeNull(); }); + + it('includes child-thread identity without materializing conversation content', async () => { + await Conversation.create({ + conversationId: 'child-conversation', + user: 'user123', + title: 'Internal child', + endpoint: EModelEndpoint.agents, + subagentThread: { + rootConversationId: 'root-conversation', + parentConversationId: 'parent-conversation', + parentMessageId: 'parent-message', + parentToolCallId: 'parent-tool-call', + subagentType: 'child-agent', + subagentKind: 'agent', + depth: 1, + }, + }); + + const result = await methods.getConvoOwnership('user123', 'child-conversation'); + + expect(result).toMatchObject({ + user: 'user123', + subagentThread: { parentConversationId: 'parent-conversation' }, + }); + expect(result).not.toHaveProperty('title'); + }); }); describe('getConvoRetention', () => { diff --git a/packages/data-schemas/src/methods/conversation.ts b/packages/data-schemas/src/methods/conversation.ts index c9c7e60318..e44dabb10a 100644 --- a/packages/data-schemas/src/methods/conversation.ts +++ b/packages/data-schemas/src/methods/conversation.ts @@ -192,7 +192,7 @@ export interface ConversationMethods { getConvoOwnership( user: string, conversationId: string, - ): Promise | null>; + ): Promise | null>; getConvoRetention( user: string, conversationId: string, @@ -432,15 +432,15 @@ export function createConversationMethods( } /** - * Ownership probe for request validation: resolves only the owning user id - * instead of materializing the full conversation document (preset spread + + * Public-read probe: resolves ownership plus the child-thread discriminator + * without materializing the full conversation document (preset spread + * message ObjectId array). */ async function getConvoOwnership(user: string, conversationId: string) { try { const Conversation = mongoose.models.Conversation as Model; - return await Conversation.findOne({ user, conversationId }, 'user').lean< - Pick + return await Conversation.findOne({ user, conversationId }, 'user subagentThread').lean< + Pick >(); } catch (error) { logger.error('[getConvoOwnership] Error checking conversation ownership', error); diff --git a/packages/data-schemas/src/models/plugins/mongoMeili.spec.ts b/packages/data-schemas/src/models/plugins/mongoMeili.spec.ts index 70835a5c4a..302e92876f 100644 --- a/packages/data-schemas/src/models/plugins/mongoMeili.spec.ts +++ b/packages/data-schemas/src/models/plugins/mongoMeili.spec.ts @@ -12,6 +12,8 @@ interface DynamicMeiliDocument extends mongoose.Document { isTemporary?: boolean; expiredAt?: Date | null; _meiliIndex?: boolean; + _meiliIndexAttempted?: boolean; + _meiliCleanupVersion?: number; } type DynamicMeiliModel = mongoose.Model & SchemaWithMeiliMethods; @@ -105,7 +107,7 @@ describe('Meilisearch Mongoose plugin', () => { mockAddDocuments.mockClear(); mockAddDocumentsInBatches.mockClear(); mockUpdateDocuments.mockClear(); - mockDeleteDocument.mockClear(); + mockDeleteDocument.mockReset().mockResolvedValue({ taskUid: 2 }); mockDeleteDocuments.mockReset().mockResolvedValue({ taskUid: 1 }); mockGetDocument.mockClear(); mockGetDocuments.mockReset().mockResolvedValue({ results: [] }); @@ -574,6 +576,54 @@ describe('Meilisearch Mongoose plugin', () => { expect(storedDoc?._meiliIndex).toBe(true); }); + test('does not clear an indexed marker until an update-hook deletion succeeds', async () => { + const conversationModel = createConversationModel( + mongoose, + ) as unknown as SchemaWithMeiliMethods; + await conversationModel.deleteMany({}); + const conversationId = new mongoose.Types.ObjectId().toString(); + + await conversationModel.create({ + conversationId, + user: new mongoose.Types.ObjectId().toString(), + title: 'Initially searchable conversation', + endpoint: EModelEndpoint.agents, + }); + const conversation = await conversationModel + .findOne({ conversationId }) + .select('+_meiliIndex +_meiliIndexAttempted'); + expect(conversation?._meiliIndex).toBe(true); + expect(conversation?._meiliIndexAttempted).toBe(true); + + conversation!.subagentThread = { + rootConversationId: 'root-conversation', + parentConversationId: 'parent-conversation', + parentMessageId: 'parent-message', + parentToolCallId: 'parent-tool-call', + subagentType: 'agent-child', + subagentKind: 'agent', + depth: 1, + }; + mockWaitForTask.mockResolvedValueOnce({ status: 'failed' }); + await conversation!.save(); + expect((await conversationModel.collection.findOne({ conversationId }))?._meiliIndex).toBe( + true, + ); + + conversation!.title = 'Retry deletion'; + mockWaitForTask.mockResolvedValueOnce({ status: 'succeeded' }); + await conversation!.save(); + const storedDoc = await conversationModel.collection.findOne({ conversationId }); + + expect(mockDeleteDocument).toHaveBeenCalledTimes(2); + expect(mockWaitForTask).toHaveBeenCalledWith(2, { + timeOutMs: 10000, + intervalMs: 100, + }); + expect(storedDoc?._meiliIndex).toBeUndefined(); + expect(storedDoc?._meiliIndexAttempted).toBeUndefined(); + }); + test('retries cleanup when Meili deletion succeeds before the Mongo flag update fails', async () => { const conversationModel = createConversationModel( mongoose, @@ -705,7 +755,7 @@ describe('Meilisearch Mongoose plugin', () => { expect(storedDoc?._meiliIndex).toBe(false); }); - test('sync removes an existing child message even when its index flag is false', async () => { + test('does not schedule cleanup for a child message that was never indexed', async () => { const messageModel = createMessageModel(mongoose) as unknown as SchemaWithMeiliMethods; await messageModel.deleteMany({}); const messageId = new mongoose.Types.ObjectId().toString(); @@ -721,18 +771,81 @@ describe('Meilisearch Mongoose plugin', () => { status: 'completed', }, _meiliIndex: false, + _meiliCleanupVersion: 1, + createdAt: new Date(), + updatedAt: new Date(), + }); + const progress = await messageModel.getSyncProgress(); + await messageModel.cleanupExcludedMeiliIndex(); + const storedDoc = await messageModel.collection.findOne({ messageId }); + + expect(progress).toMatchObject({ pendingCleanup: 0, isComplete: true }); + expect(mockDeleteDocuments).not.toHaveBeenCalled(); + expect(mockGetDocuments).not.toHaveBeenCalled(); + expect(storedDoc?._meiliIndex).toBe(false); + }); + + test('reconciles a legacy excluded false marker exactly once', async () => { + const messageModel = createMessageModel(mongoose) as unknown as SchemaWithMeiliMethods; + await messageModel.deleteMany({}); + const messageId = new mongoose.Types.ObjectId().toString(); + + await messageModel.collection.insertOne({ + messageId, + conversationId: new mongoose.Types.ObjectId().toString(), + user: new mongoose.Types.ObjectId().toString(), + isCreatedByUser: true, + text: 'Legacy child transcript with an ambiguous deletion result', + subagentTask: { + attemptKey: 'legacy-attempt-key', + status: 'completed', + }, + _meiliIndex: false, + createdAt: new Date(), + updatedAt: new Date(), + }); + + const progressBefore = await messageModel.getSyncProgress(); + await messageModel.cleanupExcludedMeiliIndex(); + const progressAfter = await messageModel.getSyncProgress(); + const storedDoc = await messageModel.collection.findOne({ messageId }); + + expect(progressBefore).toMatchObject({ pendingCleanup: 1, isComplete: false }); + expect(progressAfter).toMatchObject({ pendingCleanup: 0, isComplete: true }); + expect(mockDeleteDocuments).toHaveBeenCalledWith([messageId]); + expect(storedDoc?._meiliIndex).toBeUndefined(); + expect(storedDoc?._meiliCleanupVersion).toBe(1); + }); + + test('cleans an excluded child when an earlier Meili add was attempted but not acknowledged', async () => { + const messageModel = createMessageModel(mongoose) as unknown as SchemaWithMeiliMethods; + await messageModel.deleteMany({}); + const messageId = new mongoose.Types.ObjectId().toString(); + + await messageModel.collection.insertOne({ + messageId, + conversationId: new mongoose.Types.ObjectId().toString(), + user: new mongoose.Types.ObjectId().toString(), + isCreatedByUser: true, + text: 'Possibly indexed child transcript', + subagentTask: { + attemptKey: 'attempt-key', + status: 'completed', + }, + _meiliIndex: false, + _meiliIndexAttempted: true, createdAt: new Date(), updatedAt: new Date(), }); - mockGetDocuments.mockResolvedValueOnce({ results: [{ messageId }] }); const progress = await messageModel.getSyncProgress(); - await messageModel.syncWithMeili(); + await messageModel.cleanupExcludedMeiliIndex(); const storedDoc = await messageModel.collection.findOne({ messageId }); expect(progress).toMatchObject({ pendingCleanup: 1, isComplete: false }); expect(mockDeleteDocuments).toHaveBeenCalledWith([messageId]); expect(storedDoc?._meiliIndex).toBeUndefined(); + expect(storedDoc?._meiliIndexAttempted).toBeUndefined(); }); test('defines partial indexes for pending excluded-document cleanup', () => { @@ -742,20 +855,62 @@ describe('Meilisearch Mongoose plugin', () => { expect(conversationIndexes).toContainEqual([ { _meiliIndex: 1, conversationId: 1 }, expect.objectContaining({ - name: 'meili_excluded_cleanup', + name: 'meili_excluded_indexed_cleanup_v3', partialFilterExpression: { subagentThread: { $exists: true }, - _meiliIndex: { $exists: true }, + _meiliIndex: { $eq: true }, + }, + }), + ]); + expect(conversationIndexes).toContainEqual([ + { _meiliIndexAttempted: 1, conversationId: 1 }, + expect.objectContaining({ + name: 'meili_excluded_attempted_cleanup_v3', + partialFilterExpression: { + subagentThread: { $exists: true }, + _meiliIndexAttempted: { $eq: true }, + }, + }), + ]); + expect(conversationIndexes).toContainEqual([ + { _meiliIndex: 1, _meiliCleanupVersion: 1, conversationId: 1 }, + expect.objectContaining({ + name: 'meili_excluded_legacy_cleanup_v3', + partialFilterExpression: { + subagentThread: { $exists: true }, + _meiliIndex: { $eq: false }, + _meiliCleanupVersion: { $exists: false }, }, }), ]); expect(messageIndexes).toContainEqual([ { _meiliIndex: 1, messageId: 1 }, expect.objectContaining({ - name: 'meili_excluded_cleanup', + name: 'meili_excluded_indexed_cleanup_v3', partialFilterExpression: { subagentTask: { $exists: true }, - _meiliIndex: { $exists: true }, + _meiliIndex: { $eq: true }, + }, + }), + ]); + expect(messageIndexes).toContainEqual([ + { _meiliIndexAttempted: 1, messageId: 1 }, + expect.objectContaining({ + name: 'meili_excluded_attempted_cleanup_v3', + partialFilterExpression: { + subagentTask: { $exists: true }, + _meiliIndexAttempted: { $eq: true }, + }, + }), + ]); + expect(messageIndexes).toContainEqual([ + { _meiliIndex: 1, _meiliCleanupVersion: 1, messageId: 1 }, + expect.objectContaining({ + name: 'meili_excluded_legacy_cleanup_v3', + partialFilterExpression: { + subagentTask: { $exists: true }, + _meiliIndex: { $eq: false }, + _meiliCleanupVersion: { $exists: false }, }, }), ]); @@ -1196,6 +1351,13 @@ describe('Meilisearch Mongoose plugin', () => { // Spy on updateMany and make it fail const updateManySpy = jest .spyOn(conversationModel, 'updateMany') + .mockResolvedValueOnce({ + acknowledged: true, + matchedCount: 1, + modifiedCount: 1, + upsertedCount: 0, + upsertedId: null, + }) .mockRejectedValueOnce(new Error('Database connection error')); // Sync should throw the error diff --git a/packages/data-schemas/src/models/plugins/mongoMeili.ts b/packages/data-schemas/src/models/plugins/mongoMeili.ts index 06467a5ae2..c04f0ddfc6 100644 --- a/packages/data-schemas/src/models/plugins/mongoMeili.ts +++ b/packages/data-schemas/src/models/plugins/mongoMeili.ts @@ -30,6 +30,8 @@ interface MongoMeiliOptions { interface MeiliIndexable { [key: string]: unknown; _meiliIndex?: boolean; + _meiliIndexAttempted?: boolean; + _meiliCleanupVersion?: number; } interface SyncProgress { @@ -42,6 +44,8 @@ interface SyncProgress { interface _DocumentWithMeiliIndex extends Document { _meiliIndex?: boolean; + _meiliIndexAttempted?: boolean; + _meiliCleanupVersion?: number; isTemporary?: boolean; expiredAt?: Date | null; preprocessObjectForIndex?: () => Record; @@ -62,6 +66,7 @@ export interface SchemaWithMeiliMethods extends Model { index: Index, documents: Array>, ): Promise; + cleanupExcludedMeiliIndex(): Promise; cleanupMeiliIndex( index: Index, primaryKey: string, @@ -100,6 +105,7 @@ const hasSchemaPath = (schema: Schema, path: string): boolean => Object.prototype.hasOwnProperty.call(schema.obj, path); const explicitTemporaryFlagKey = 'meiliExplicitTemporaryFlag'; +const meiliCleanupVersion = 1; const buildRetentionIndexableQuery = (schema: Schema): FilterQuery => { if (hasSchemaPath(schema, 'isTemporary')) { @@ -133,7 +139,11 @@ const buildExcludedIndexedQuery = (excludeFromIndexPath?: string): FilterQuery doc._id); + await this.updateMany( + { _id: { $in: docsIds } }, + { $set: { _meiliIndexAttempted: true } }, + { timestamps: false }, + ); + // Add documents to MeiliSearch await index.addDocumentsInBatches(formattedDocs, undefined, { primaryKey }); // Update MongoDB to mark documents as indexed. // { timestamps: false } prevents Mongoose from touching updatedAt, preserving // original conversation/message timestamps (fixes sidebar chronological sort). - const docsIds = documents.map((doc) => doc._id); await this.updateMany( { _id: { $in: docsIds } }, - { $set: { _meiliIndex: true } }, + { $set: { _meiliIndex: true, _meiliCleanupVersion: meiliCleanupVersion } }, { timestamps: false }, ); } catch (error) { @@ -374,6 +390,58 @@ const createMeiliMongooseModel = ({ } } + /** + * Remove documents that are intentionally excluded from search without + * scanning the complete Meili index. The Mongo marker is cleared only + * after Meili confirms deletion, so interrupted cleanup is retried. + */ + static async cleanupExcludedMeiliIndex(this: SchemaWithMeiliMethods): Promise { + const excludedIndexedQuery = getExcludedIndexedQuery(); + if (excludedIndexedQuery == null) { + return; + } + + const { batchSize, delayMs } = syncConfig; + while (true) { + const pendingExcludedDocuments = await this.find(excludedIndexedQuery) + .select(primaryKey) + .limit(batchSize) + .lean(); + if (pendingExcludedDocuments.length === 0) { + break; + } + + const pendingIds = pendingExcludedDocuments.map( + (doc: Record) => doc[primaryKey], + ); + const deletion = await index.deleteDocuments(pendingIds.map(String)); + const deletionTask = await client.waitForTask(deletion.taskUid, { + timeOutMs: 10000, + intervalMs: 100, + }); + if (deletionTask.status !== 'succeeded') { + throw new Error( + `Meili cleanup task ${deletion.taskUid} ended with ${deletionTask.status}`, + ); + } + await this.updateMany( + { ...excludedIndexedQuery, [primaryKey]: { $in: pendingIds } }, + { + $set: { _meiliCleanupVersion: meiliCleanupVersion }, + $unset: { _meiliIndex: '', _meiliIndexAttempted: '' }, + }, + { timestamps: false }, + ); + + if (pendingExcludedDocuments.length < batchSize) { + break; + } + if (delayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, delayMs)); + } + } + } + /** * Clean up documents in MeiliSearch that no longer exist in MongoDB */ @@ -385,46 +453,7 @@ const createMeiliMongooseModel = ({ delayMs: number, ): Promise { try { - const excludedIndexedQuery = getExcludedIndexedQuery(); - if (excludedIndexedQuery != null) { - let hasPendingExcludedDocuments = true; - while (hasPendingExcludedDocuments) { - const pendingExcludedDocuments = await this.find(excludedIndexedQuery) - .select(primaryKey) - .limit(batchSize) - .lean(); - if (pendingExcludedDocuments.length === 0) { - hasPendingExcludedDocuments = false; - break; - } - - const pendingIds = pendingExcludedDocuments.map( - (doc: Record) => doc[primaryKey], - ); - const deletion = await index.deleteDocuments(pendingIds.map(String)); - const deletionTask = await client.waitForTask(deletion.taskUid, { - timeOutMs: 10000, - intervalMs: 100, - }); - if (deletionTask.status !== 'succeeded') { - throw new Error( - `Meili cleanup task ${deletion.taskUid} ended with ${deletionTask.status}`, - ); - } - await this.updateMany( - { ...excludedIndexedQuery, [primaryKey]: { $in: pendingIds } }, - { $unset: { _meiliIndex: '' } }, - { timestamps: false }, - ); - - if (pendingExcludedDocuments.length < batchSize) { - break; - } - if (delayMs > 0) { - await new Promise((resolve) => setTimeout(resolve, delayMs)); - } - } - } + await this.cleanupExcludedMeiliIndex(); let offset = 0; let moreDocuments = true; @@ -463,7 +492,10 @@ const createMeiliMongooseModel = ({ } await this.updateMany( { [primaryKey]: { $in: toDelete } }, - { $unset: { _meiliIndex: '' } }, + { + $set: { _meiliCleanupVersion: meiliCleanupVersion }, + $unset: { _meiliIndex: '', _meiliIndexAttempted: '' }, + }, { timestamps: false }, ); logger.debug(`[cleanupMeiliIndex] Deleted ${toDelete.length} orphaned documents`); @@ -578,6 +610,21 @@ const createMeiliMongooseModel = ({ const maxRetries = 3; let retryCount = 0; + try { + // Mark the possible Meili presence before enqueueing the add. If the + // later Mongo acknowledgement fails, excluded-record cleanup can still + // distinguish this row from a child that was never submitted to Meili. + const model = this.constructor as Model; + await model.updateOne( + { _id: this._id as Types.ObjectId }, + { $set: { _meiliIndexAttempted: true } }, + { timestamps: false }, + ); + } catch (error) { + logger.error('[addObjectToMeili] Error marking Meili indexing attempt:', error); + return next(); + } + while (retryCount < maxRetries) { try { await index.addDocuments([object], { primaryKey }); @@ -597,7 +644,7 @@ const createMeiliMongooseModel = ({ // eslint-disable-next-line no-restricted-syntax -- _meiliIndex is an internal bookkeeping flag, not tenant-scoped data await this.collection.updateOne( { _id: this._id as Types.ObjectId }, - { $set: { _meiliIndex: true } }, + { $set: { _meiliIndex: true, _meiliCleanupVersion: meiliCleanupVersion } }, ); } catch (error) { logger.error('[addObjectToMeili] Error updating _meiliIndex field:', error); @@ -616,11 +663,25 @@ const createMeiliMongooseModel = ({ ): Promise { try { if (!isIndexableDocument(this, excludeFromIndexPath)) { - await index.deleteDocument(String(this[primaryKey as keyof DocumentWithMeiliIndex])); + const deletion = await index.deleteDocument( + String(this[primaryKey as keyof DocumentWithMeiliIndex]), + ); + const deletionTask = await client.waitForTask(deletion.taskUid, { + timeOutMs: 10000, + intervalMs: 100, + }); + if (deletionTask.status !== 'succeeded') { + throw new Error( + `Meili cleanup task ${deletion.taskUid} ended with ${deletionTask.status}`, + ); + } const model = this.constructor as Model; await model.updateOne( { _id: this._id as Types.ObjectId }, - { $set: { _meiliIndex: false } }, + { + $set: { _meiliCleanupVersion: meiliCleanupVersion }, + $unset: { _meiliIndex: '', _meiliIndexAttempted: '' }, + }, ); return next(); } @@ -736,16 +797,48 @@ export default function mongoMeili(schema: Schema, options: MongoMeiliOptions): select: false, default: false, }, + _meiliIndexAttempted: { + type: Boolean, + required: false, + select: false, + }, + _meiliCleanupVersion: { + type: Number, + required: false, + select: false, + default: meiliCleanupVersion, + }, }); if (options.excludeFromIndexPath != null) { schema.index( { _meiliIndex: 1, [options.primaryKey]: 1 }, { - name: 'meili_excluded_cleanup', + name: 'meili_excluded_indexed_cleanup_v3', partialFilterExpression: { [options.excludeFromIndexPath]: { $exists: true }, - _meiliIndex: { $exists: true }, + _meiliIndex: { $eq: true }, + }, + }, + ); + schema.index( + { _meiliIndexAttempted: 1, [options.primaryKey]: 1 }, + { + name: 'meili_excluded_attempted_cleanup_v3', + partialFilterExpression: { + [options.excludeFromIndexPath]: { $exists: true }, + _meiliIndexAttempted: { $eq: true }, + }, + }, + ); + schema.index( + { _meiliIndex: 1, _meiliCleanupVersion: 1, [options.primaryKey]: 1 }, + { + name: 'meili_excluded_legacy_cleanup_v3', + partialFilterExpression: { + [options.excludeFromIndexPath]: { $exists: true }, + _meiliIndex: { $eq: false }, + _meiliCleanupVersion: { $exists: false }, }, }, );