diff --git a/apps/sim/app/api/knowledge/search/route.ts b/apps/sim/app/api/knowledge/search/route.ts index 610507ea53e..8a3e2ccab3a 100644 --- a/apps/sim/app/api/knowledge/search/route.ts +++ b/apps/sim/app/api/knowledge/search/route.ts @@ -17,7 +17,11 @@ export const POST = defineInternalJsonRoute({ 'Provider limits apply independently to each user grant; requests have bounded fanout and deadlines', }), errorPolicy: internalKnowledgeErrorPolicies.search, - mapInput: ({ body }, { request }) => ({ ...body, signal: request.signal }), + mapInput: ({ body }, { request }) => ({ + ...body, + signal: request.signal, + surface: 'dashboard' as const, + }), useCase: searchLiveKnowledge, present: (data) => ({ success: true as const, data }), }) diff --git a/apps/sim/lib/knowledge/application/search-diagnostics.ts b/apps/sim/lib/knowledge/application/search-diagnostics.ts index da404865994..a1e444d726c 100644 --- a/apps/sim/lib/knowledge/application/search-diagnostics.ts +++ b/apps/sim/lib/knowledge/application/search-diagnostics.ts @@ -33,7 +33,12 @@ export function instrumentSearchUseCase< topK: args.input.topK, documentFilterCount: args.input.filters?.documentIds?.length ?? 0, hasSourceFilter: Boolean(args.input.filters?.source), - hasDateFilter: Boolean(args.input.filters?.modifiedAfter), + hasDateFilter: Boolean( + args.input.filters?.modifiedAfter || + args.input.filters?.modifiedBefore || + args.input.filters?.startDate || + args.input.filters?.endDate + ), tagFilterCount: args.input.tagFilters?.length ?? 0, hasProvenance: Boolean( args.input.resultSecretRegistry || args.input.prepareModelInputProvenance diff --git a/apps/sim/lib/knowledge/mcp/server.ts b/apps/sim/lib/knowledge/mcp/server.ts index 56d058ba17a..e37d68aaf09 100644 --- a/apps/sim/lib/knowledge/mcp/server.ts +++ b/apps/sim/lib/knowledge/mcp/server.ts @@ -139,6 +139,7 @@ export function createKnowledgeMcpServer(context: KnowledgeMcpContext): McpServe const result = await searchLiveKnowledge.execute({ principal, input: { + surface: 'mcp', organizationId, query, topK: topK ?? liveSearchMcpDefaultTopK(parsed), diff --git a/apps/sim/lib/knowledge/search/diagnostics.test.ts b/apps/sim/lib/knowledge/search/diagnostics.test.ts index 3d9a261e727..59fe303e715 100644 --- a/apps/sim/lib/knowledge/search/diagnostics.test.ts +++ b/apps/sim/lib/knowledge/search/diagnostics.test.ts @@ -1,6 +1,13 @@ +import { createDeferred } from '@sim/testing/helpers' import { getMockLogger } from '@sim/testing/mocks/logger.mock' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { measureSearchStage, withSearchDiagnostics } from '@/lib/knowledge/search/diagnostics' +import { + measureSearchStage, + recordSearchProviderResult, + withSearchDiagnostics, + withSearchProviderDiagnostics, + withSearchRequestDiagnostics, +} from '@/lib/knowledge/search/diagnostics' const logs = getMockLogger('KnowledgeSearchDiagnostics') @@ -9,21 +16,13 @@ beforeEach(() => { }) afterEach(() => vi.useRealTimers()) -function _deferred() { - let resolve!: () => void - const promise = new Promise((done) => { - resolve = done - }) - return { promise, resolve } -} - describe('search pipeline diagnostics', () => { it('preserves failures, records repeated stage errors, and does not log error content', async () => { const error = new Error('private provider response') await expect( withSearchDiagnostics({ surface: 'copilot' }, async () => { - await measureSearchStage('vector.authorization', () => 'allowed') - return measureSearchStage('vector.authorization', () => { + await measureSearchStage('live.resolve', () => 'allowed') + return measureSearchStage('live.resolve', () => { throw error }) }) @@ -33,11 +32,143 @@ describe('search pipeline diagnostics', () => { expect.objectContaining({ outcome: 'error', stages: { - 'vector.authorization': { count: 2, totalMs: 0, maxMs: 0, errors: 1 }, + 'live.resolve': { count: 2, totalMs: 0, maxMs: 0, errors: 1 }, }, }) ) expect(JSON.stringify(logs.info.mock.calls)).not.toContain(error.message) expect(vi.getTimerCount()).toBe(0) }) + + it('isolates concurrent provider failures and retains errors caught by a search adapter', async () => { + const error = Object.assign(new Error('private query, provider response, and credential'), { + code: 'private credential', + httpStatus: 403, + diagnosticCode: 'provider_access_denied', + }) + const result = await withSearchDiagnostics({ surface: 'dashboard' }, () => + Promise.all([ + withSearchProviderDiagnostics({ provider: 'gmail', accountIndex: 0 }, async () => { + await measureSearchStage('live.search', () => + withSearchRequestDiagnostics({ transport: 'native_http', method: 'GET' }, async () => { + await Promise.resolve() + throw error + }).catch(() => undefined) + ) + recordSearchProviderResult({ status: 'partial', resultCount: 1 }) + return ['available result'] + }), + withSearchProviderDiagnostics({ provider: 'github', accountIndex: 1 }, async () => { + await measureSearchStage('live.verify', () => + withSearchRequestDiagnostics( + { transport: 'native_http', method: 'GET' }, + async (record) => { + record({ httpStatus: 200 }) + } + ) + ) + recordSearchProviderResult({ status: 'ok', resultCount: 2 }) + return ['other result'] + }), + ]) + ) + expect(result).toEqual([['available result'], ['other result']]) + const summaries = logs.info.mock.calls + .filter(([message]) => message === 'Live search provider completed') + .map(([, data]) => data) + expect(summaries).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + provider: 'gmail', + requests: [ + expect.objectContaining({ + stage: 'live.search', + outcome: 'error', + failure: { + reason: 'provider_access_denied', + httpStatus: 403, + }, + }), + ], + results: [{ status: 'partial', resultCount: 1 }], + }), + expect.objectContaining({ + provider: 'github', + requests: [ + expect.objectContaining({ stage: 'live.verify', outcome: 'success', httpStatus: 200 }), + ], + results: [{ status: 'ok', resultCount: 2 }], + }), + ]) + ) + const serialized = JSON.stringify(logs.info.mock.calls) + expect(serialized).not.toContain('private') + expect(serialized).not.toContain('available result') + expect(serialized).not.toContain('other result') + expect(new Set(summaries.map((summary) => summary.searchId)).size).toBe(1) + }) + + it('preserves cancellation and bounds request diagnostics without losing failure counts', async () => { + const aborted = new DOMException('private cancellation reason', 'AbortError') + await expect( + withSearchDiagnostics({}, () => + withSearchProviderDiagnostics({ provider: 'gmail', accountIndex: 0 }, async () => { + for (let index = 0; index < 140; index++) { + await withSearchRequestDiagnostics( + { transport: 'native_http', method: 'GET' }, + async () => { + throw Object.assign(new Error('private failure'), { code: 'ECONNRESET' }) + } + ).catch(() => undefined) + } + return withSearchRequestDiagnostics( + { transport: 'native_http', method: 'GET' }, + async () => { + throw aborted + } + ) + }) + ) + ).rejects.toBe(aborted) + expect(logs.info).toHaveBeenCalledWith( + 'Live search provider completed', + expect.objectContaining({ + requestCount: 141, + failedRequestCount: 141, + omittedRequestCount: 13, + outcome: 'error', + failure: { reason: 'aborted' }, + }) + ) + expect(JSON.stringify(logs.info.mock.calls)).not.toContain('private') + expect(vi.getTimerCount()).toBe(0) + }) + + it('records preview failures that settle after the search has returned', async () => { + const pending = createDeferred() + let request: Promise | undefined + await withSearchDiagnostics({}, () => + withSearchProviderDiagnostics({ provider: 'gmail', accountIndex: 0 }, async () => { + request = withSearchRequestDiagnostics( + { transport: 'native_http', method: 'GET' }, + () => pending.promise + ).catch(() => undefined) + }) + ) + expect(logs.info).toHaveBeenCalledWith( + 'Live search provider completed', + expect.objectContaining({ pendingRequestCount: 1 }) + ) + pending.reject(new DOMException('private preview content', 'AbortError')) + await request + expect(logs.info).toHaveBeenLastCalledWith( + 'Live search request completed after provider', + expect.objectContaining({ + requestIndex: 0, + outcome: 'error', + failure: { reason: 'aborted' }, + }) + ) + expect(JSON.stringify(logs.info.mock.calls)).not.toContain('private preview content') + }) }) diff --git a/apps/sim/lib/knowledge/search/diagnostics.ts b/apps/sim/lib/knowledge/search/diagnostics.ts index 81a13db8188..6b603bd3abe 100644 --- a/apps/sim/lib/knowledge/search/diagnostics.ts +++ b/apps/sim/lib/knowledge/search/diagnostics.ts @@ -1,6 +1,8 @@ import { AsyncLocalStorage } from 'node:async_hooks' import { createLogger } from '@sim/logger' import { generateId } from '@sim/utils/id' +import { toRecord } from '@sim/utils/object' +import type { LiveSearchProvider } from '@/lib/api/contracts/mothership-assistant-tools' const logger = createLogger('KnowledgeSearchDiagnostics', { logLevel: 'INFO' }) const PROGRESS_INTERVAL_MS = 5000 @@ -104,6 +106,9 @@ export interface SearchDiagnosticMetadata { accessBatchCount?: number /** Connector identities sent for live proof, summed over every batch after the first. */ liveProofConnectorCount?: number + accountCount?: number + searchedAccountCount?: number + nativeQueryCount?: number } interface StageTiming { @@ -118,12 +123,264 @@ interface SearchTrace { startedAt: number metadata: SearchDiagnosticMetadata stages: Partial> - active: Map + active: Map< + symbol, + { stage: SearchStage; startedAt: number; provider?: LiveSearchProvider; accountIndex?: number } + > } const traces = new AsyncLocalStorage() +const providers = new AsyncLocalStorage() +const currentStages = new AsyncLocalStorage() +const currentQueries = new AsyncLocalStorage() const roundMs = (value: number) => Math.round(value * 100) / 100 +const MAX_PROVIDER_REQUEST_DIAGNOSTICS = 128 +const FAILURE_REASONS = [ + 'provider_validation', + 'provider_access_denied', + 'provider_rate_limited', + 'provider_http_error', + 'request_budget_exceeded', + 'mcp_tool_unavailable', + 'mcp_schema_incompatible', + 'mcp_tool_error', + 'mcp_payload_too_large', + 'mcp_payload_unreadable', + 'mcp_account_required', +] as const +export type SearchFailureReason = (typeof FAILURE_REASONS)[number] + +const ERROR_CODES = new Set([ + 'ECONNRESET', + 'ECONNREFUSED', + 'ETIMEDOUT', + 'ENOTFOUND', + 'EAI_AGAIN', + 'EPIPE', + 'UND_ERR_CONNECT_TIMEOUT', + 'UND_ERR_HEADERS_TIMEOUT', + 'UND_ERR_BODY_TIMEOUT', + 'UND_ERR_SOCKET', + 'UND_ERR_ABORTED', + 'ABORT_ERR', + 'ERR_STREAM_PREMATURE_CLOSE', + 'MANAGED_CREDENTIAL_NOT_FOUND', + 'MANAGED_CREDENTIAL_UNAVAILABLE', + 'MANAGED_CREDENTIAL_PROVIDER_MISMATCH', + 'MANAGED_CREDENTIAL_REVOKED', + 'MANAGED_CREDENTIAL_NEEDS_REAUTH', + 'MANAGED_CREDENTIAL_INSUFFICIENT_SCOPE', + 'MANAGED_CREDENTIAL_INVALID_TOKEN_SET', + 'MANAGED_CREDENTIAL_CONFIGURATION_UNAVAILABLE', + 'MANAGED_CREDENTIAL_IDENTITY_UNAVAILABLE', + 'MANAGED_CREDENTIAL_REFRESH_FAILED', +]) + +interface ProviderMetadata { + provider: LiveSearchProvider + accountIndex: number +} + +interface ProviderResult { + queryIndex?: number + status: 'ok' | 'partial' | 'reconnect' | 'rate_limited' | 'unavailable' | 'timeout' + resultCount: number + coverage?: { + unverified: boolean + verificationRateLimited: boolean + servicePartial: boolean + providerPartial: boolean + undatedExcluded: boolean + undatedUnsorted: boolean + moreUnsorted: boolean + moreUnreachable: boolean + emptyContinuable: boolean + } +} + +interface RequestMetadata { + transport: 'native_http' | 'mcp' + method?: 'GET' | 'POST' + /** Only code-allowlisted tool names; never remote descriptions or arguments. */ + tool?: string +} + +interface RequestOutcome { + httpStatus?: number + /** Provider reason codes already filtered through the provider's fixed allowlist. */ + providerReasons?: string[] +} + +interface RequestDiagnostic extends RequestMetadata, RequestOutcome { + stage?: SearchStage + queryIndex?: number + startedAfterMs: number + elapsedMs?: number + outcome?: 'success' | 'error' + failure?: ReturnType +} + +interface ProviderTrace { + searchId: string + closed: boolean + pendingRequestCount: number + metadata: ProviderMetadata + startedAt: number + stages: Partial> + results: Array }> + requests: RequestDiagnostic[] + requestCount: number + failedRequestCount: number +} + +/** Error messages, stacks, response bodies, and arbitrary provider codes never enter logs. */ +function describeFailure(error: unknown) { + const record = toRecord(error) + const code = (value: unknown) => + typeof value === 'string' && ERROR_CODES.has(value) + ? value + : typeof value === 'number' && + [-32700, -32600, -32601, -32602, -32603, -32000, -32001].includes(value) + ? value + : undefined + const reason = + FAILURE_REASONS.find((candidate) => candidate === record.diagnosticCode) ?? + (record.name === 'AbortError' + ? 'aborted' + : record.name === 'PayloadSizeLimitError' + ? 'response_size_limit' + : record.name === 'SyntaxError' + ? 'invalid_json' + : record.name === 'TimeoutError' + ? 'timeout' + : 'unexpected_error') + const httpStatus = + typeof record.httpStatus === 'number' && + Number.isInteger(record.httpStatus) && + record.httpStatus >= 100 && + record.httpStatus <= 599 + ? record.httpStatus + : undefined + const errorCode = code(record.code) + const causeCode = code(toRecord(record.cause).code) + return { + reason, + ...(httpStatus === undefined ? {} : { httpStatus }), + ...(errorCode === undefined ? {} : { code: errorCode }), + ...(causeCode === undefined ? {} : { causeCode }), + } +} + +/** A provider keeps its own waterfall while sharing the request's search ID and stage totals. */ +export async function withSearchProviderDiagnostics( + metadata: ProviderMetadata, + run: () => Promise +): Promise { + const root = traces.getStore() + if (!root) return run() + const provider: ProviderTrace = { + searchId: root.searchId, + closed: false, + pendingRequestCount: 0, + metadata, + startedAt: performance.now(), + stages: {}, + results: [], + requests: [], + requestCount: 0, + failedRequestCount: 0, + } + return providers.run(provider, async () => { + let failure: ReturnType | undefined + try { + return await run() + } catch (error) { + failure = describeFailure(error) + throw error + } finally { + provider.closed = true + logger.info('Live search provider completed', { + searchId: root.searchId, + ...metadata, + startedAfterMs: roundMs(provider.startedAt - root.startedAt), + elapsedMs: roundMs(performance.now() - provider.startedAt), + outcome: failure + ? 'error' + : provider.results.some((result) => result.status !== 'ok') + ? 'partial' + : 'success', + ...(failure ? { failure } : {}), + stages: provider.stages, + results: provider.results, + requests: structuredClone(provider.requests), + requestCount: provider.requestCount, + failedRequestCount: provider.failedRequestCount, + pendingRequestCount: provider.pendingRequestCount, + omittedRequestCount: provider.requestCount - provider.requests.length, + }) + } + }) +} + +/** Captures query-level failures even when the application returns other providers' results. */ +export function recordSearchProviderResult(result: ProviderResult, error?: unknown): void { + providers.getStore()?.results.push({ + ...result, + ...(error === undefined ? {} : { failure: describeFailure(error) }), + }) +} + +/** Native query ordinals distinguish parallel requests without retaining their query text. */ +export function withSearchQueryDiagnostics( + queryIndex: number | undefined, + run: () => Promise +): Promise { + return currentQueries.run(queryIndex, run) +} + +/** Bounded request records retain failures swallowed by fallbacks and verification filters. */ +export async function withSearchRequestDiagnostics( + metadata: RequestMetadata, + run: (record: (outcome: RequestOutcome) => void) => Promise +): Promise { + const provider = providers.getStore() + if (!provider) return run(() => undefined) + const startedAt = performance.now() + const diagnostic: RequestDiagnostic = { + ...metadata, + stage: currentStages.getStore(), + queryIndex: currentQueries.getStore(), + startedAfterMs: roundMs(startedAt - provider.startedAt), + } + provider.requestCount++ + const requestIndex = provider.requestCount - 1 + provider.pendingRequestCount++ + const retained = provider.requests.length < MAX_PROVIDER_REQUEST_DIAGNOSTICS + if (retained) provider.requests.push(diagnostic) + try { + const result = await run((outcome) => Object.assign(diagnostic, outcome)) + diagnostic.outcome = 'success' + return result + } catch (error) { + provider.failedRequestCount++ + diagnostic.outcome = 'error' + diagnostic.failure = describeFailure(error) + throw error + } finally { + diagnostic.elapsedMs = roundMs(performance.now() - startedAt) + provider.pendingRequestCount-- + if (provider.closed && retained) { + logger.info('Live search request completed after provider', { + searchId: provider.searchId, + ...provider.metadata, + requestIndex, + ...diagnostic, + }) + } + } +} + export function annotateSearchDiagnostics(metadata: SearchDiagnosticMetadata): void { const trace = traces.getStore() if (trace) Object.assign(trace.metadata, metadata) @@ -136,6 +393,13 @@ export function recordSearchStageDuration(stage: SearchStage, milliseconds: numb timing.count++ timing.totalMs = roundMs(timing.totalMs + milliseconds) timing.maxMs = roundMs(Math.max(timing.maxMs, milliseconds)) + const provider = providers.getStore() + if (provider) { + const local = (provider.stages[stage] ??= { count: 0, totalMs: 0, maxMs: 0, errors: 0 }) + local.count++ + local.totalMs = roundMs(local.totalMs + milliseconds) + local.maxMs = roundMs(Math.max(local.maxMs, milliseconds)) + } } /** Timings include waiting on the dependency; nested and parallel stages must not be summed. */ @@ -147,12 +411,17 @@ export async function measureSearchStage( if (!trace) return run() const span = Symbol(stage) const startedAt = performance.now() - trace.active.set(span, { stage, startedAt }) + trace.active.set(span, { stage, startedAt, ...providers.getStore()?.metadata }) try { - return await run() + return await currentStages.run(stage, run) } catch (error) { const timing = (trace.stages[stage] ??= { count: 0, totalMs: 0, maxMs: 0, errors: 0 }) timing.errors++ + const provider = providers.getStore() + if (provider) { + const local = (provider.stages[stage] ??= { count: 0, totalMs: 0, maxMs: 0, errors: 0 }) + local.errors++ + } throw error } finally { trace.active.delete(span) @@ -182,8 +451,8 @@ export async function withSearchDiagnostics( ...trace.metadata, elapsedMs: roundMs(performance.now() - trace.startedAt), stages: structuredClone(trace.stages), - activeStages: [...trace.active.values()].map(({ stage, startedAt }) => ({ - stage, + activeStages: [...trace.active.values()].map(({ startedAt, ...active }) => ({ + ...active, elapsedMs: roundMs(performance.now() - startedAt), })), }) diff --git a/apps/sim/lib/sim-search/live/README.md b/apps/sim/lib/sim-search/live/README.md index 2a81b679331..6df8082a441 100644 --- a/apps/sim/lib/sim-search/live/README.md +++ b/apps/sim/lib/sim-search/live/README.md @@ -161,6 +161,16 @@ Zoom Search defaults off for organization-scoped rollout. Enable selected organi Only the canonical organization ID participates in this rollout check. For local or self-hosted deployments, `ZOOM_SEARCH=true` enables Zoom Search for all otherwise eligible organization-owned scopes; personal workspaces without an organization cannot use managed connected accounts. Leave that boolean fallback off for an organization-targeted rollout. Setup, enrollment and retrieval enforce the flag. The dedicated Zoom MCP Search connector is gated wherever it is invoked, including generic MCP tools; the standard workflow Zoom OAuth/tools remain available. Disabling the flag preserves saved grants and conversations while denying subsequent Search use; existing approvals can still be removed and connected accounts disconnected. Other providers retain the shared Search and credential-group availability policies without a separate provider rollout gate. +### Search diagnostics + +Dashboard, Copilot, and MCP searches share a `searchId` across the application summary and each provider account. The route logger also carries its request ID. `Live search provider completed` records the account's ordinal within that search, elapsed time, stage timings, query outcomes, coverage limitations, and up to 128 outbound requests. Each request includes its stage, native-query ordinal when present, transport, status, start offset, elapsed time including body decoding, and safe failure codes. Stage times can overlap; do not sum them to calculate request latency. + +The coverage flags distinguish failed verification, verification rate limits, partial service-account coverage, provider-reported partial results, missing dates, and pagination limits. A partial result can occur even when every HTTP request succeeds. Failures caught by verification or preview fallbacks remain visible in the request records. + +`pendingRequestCount` identifies requests still running when a provider returns. Retained requests finishing later emit `Live search request completed after provider`, correlated by `searchId`, provider, account ordinal, and zero-based `requestIndex`. Search never waits for telemetry or late previews. `omittedRequestCount` reports requests beyond the diagnostic cap; their individual records, including late completions, are intentionally omitted. + +In CloudWatch Logs Insights, start with the request's time window and filter `@message` for `Knowledge search completed` or `Live search provider completed`, then follow its `searchId` across all records. Authentication and session setup, search, verification, and previews have separate stages. Only fixed failure classifications, HTTP statuses, allowlisted network/credential codes, and allowlisted provider reason codes are retained. Unknown errors remain `unexpected_error`; query text, URLs, document content, account identifiers, credentials, provider error messages, and stacks are not logged by these diagnostics. + ### Shared invariants - Keep provider parsing isolated from authorization. The application operation owns current membership, policy loading, active account resolution, scoped references, and result projection. diff --git a/apps/sim/lib/sim-search/live/application.ts b/apps/sim/lib/sim-search/live/application.ts index 98d58f20334..a068e0c9abc 100644 --- a/apps/sim/lib/sim-search/live/application.ts +++ b/apps/sim/lib/sim-search/live/application.ts @@ -33,7 +33,15 @@ import { requireOrganizationSearchAvailable } from '@/lib/knowledge/access/avail import { defineAuthorizedKnowledgeUseCase } from '@/lib/knowledge/application/authorized-knowledge-use-case' import { resolveKnowledgeOwnerContext } from '@/lib/knowledge/application/contexts' import { knowledgeOperations } from '@/lib/knowledge/application/operations' -import { measureSearchStage } from '@/lib/knowledge/search/diagnostics' +import type { SearchKnowledgeInput } from '@/lib/knowledge/application/search' +import { instrumentSearchUseCase } from '@/lib/knowledge/application/search-diagnostics' +import { + annotateSearchDiagnostics, + measureSearchStage, + recordSearchProviderResult, + withSearchProviderDiagnostics, + withSearchQueryDiagnostics, +} from '@/lib/knowledge/search/diagnostics' import { RRF_K } from '@/lib/knowledge/search/recency' import { matchPassage } from '@/lib/knowledge/search/snippet' import { isKnowledgeSourceUrl } from '@/lib/knowledge/search/source-url' @@ -163,6 +171,7 @@ export function decodeLiveReference(value: string): Reference { } interface LiveSearchOptions { + surface?: SearchKnowledgeInput['surface'] query: string topK: number filters?: WorkspaceSearchFilters @@ -389,7 +398,7 @@ function compareAccounts(left: LiveAccount, right: LiveAccount): number { ) } -export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ +const liveKnowledgeSearch = defineAuthorizedKnowledgeUseCase({ operation: knowledgeOperations.search, resolveContext: ({ input }: { input: LiveSearchInput }) => resolveKnowledgeOwnerContext(input), async execute({ principal, input }): Promise { @@ -452,6 +461,12 @@ export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ ) .sort(compareAccounts) const selected = eligible.slice(0, MAX_ACCOUNTS) + annotateSearchDiagnostics({ + operation: 'search_workspace', + accountCount: eligible.length, + searchedAccountCount: selected.length, + nativeQueryCount: queries?.length ?? 0, + }) const direction = dateSortDirection(filters) const dateSorted = Boolean(direction) const pool = createPinnedConnectionPool() @@ -573,7 +588,7 @@ export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ moreUnsorted || moreUnreachable || emptyContinuable - return { + const searched: SearchedQuery = { status: { ...status, status: degraded ? 'partial' : 'ok', @@ -636,6 +651,27 @@ export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ } }), } + recordSearchProviderResult({ + queryIndex: status.queryIndex, + status: searched.status.status, + resultCount: searched.results.length, + ...(degraded + ? { + coverage: { + unverified, + verificationRateLimited: rateLimited, + servicePartial: Boolean(session.servicePartial), + providerPartial: Boolean(page.partial), + undatedExcluded, + undatedUnsorted, + moreUnsorted, + moreUnreachable, + emptyContinuable, + }, + } + : {}), + }) + return searched } /** One session per account serves each of its native queries, each reported on its own. */ const searchAccount = async (account: LiveAccount): Promise => { @@ -663,6 +699,10 @@ export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ 'unavailable', 'This account could not be searched. Check the connection and try again.' ) + recordSearchProviderResult( + { queryIndex: target.queryIndex, status: failure.status, resultCount: 0 }, + error + ) return { status: { ...statusFor(target), @@ -693,9 +733,9 @@ export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ const currentSession = session return await Promise.all( natives.map((target) => - searchQuery(account, resolved, currentSession, target.native, statusFor(target)).catch( - (error) => failed(error, target) - ) + withSearchQueryDiagnostics(target.queryIndex, () => + searchQuery(account, resolved, currentSession, target.native, statusFor(target)) + ).catch((error) => failed(error, target)) ) ) } catch (error) { @@ -707,7 +747,13 @@ export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ } let searched: SearchedQuery[] try { - searched = (await mapWithConcurrency(selected, ACCOUNT_CONCURRENCY, searchAccount)).flat() + searched = ( + await mapWithConcurrency(selected, ACCOUNT_CONCURRENCY, (account, accountIndex) => + withSearchProviderDiagnostics({ provider: account.provider, accountIndex }, () => + searchAccount(account) + ) + ) + ).flat() } finally { pool.destroy() } @@ -782,14 +828,19 @@ export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ message: 'No connection with this provider is configured and approved in this scope.', }) } + const retrievalStatus = + accounts.some((account) => account.status !== 'ok') || eligible.length > selected.length + ? 'partial' + : 'complete' + annotateSearchDiagnostics({ + resultCount: Math.min(ranked.length, input.topK), + retrievalStatus, + }) return { query: input.query, results: ranked.slice(0, input.topK).map(({ result }) => result), retrieval: { - status: - accounts.some((account) => account.status !== 'ok') || eligible.length > selected.length - ? 'partial' - : 'complete', + status: retrievalStatus, timedOutLegs: [], }, live: { @@ -804,6 +855,10 @@ export const searchLiveKnowledge = defineAuthorizedKnowledgeUseCase({ } }, }) +export const searchLiveKnowledge = instrumentSearchUseCase( + 'knowledge_application', + liveKnowledgeSearch +) export type LiveReadInput = ResourceOwner & { documentId: string diff --git a/apps/sim/lib/sim-search/live/http.ts b/apps/sim/lib/sim-search/live/http.ts index d2b7ea5ddb6..f24d814d1f7 100644 --- a/apps/sim/lib/sim-search/live/http.ts +++ b/apps/sim/lib/sim-search/live/http.ts @@ -2,8 +2,15 @@ import { type PinnedConnectionPool, secureFetchWithValidation, } from '@/lib/core/security/input-validation.server' +import { + type SearchFailureReason, + withSearchRequestDiagnostics, +} from '@/lib/knowledge/search/diagnostics' import type { NativeClient } from '@/lib/sim-search/live/types' -import { isGoogleQuotaReason } from '@/connectors/google-workspace/api-errors' +import { + isGoogleQuotaReason, + safeGoogleErrorReasons, +} from '@/connectors/google-workspace/api-errors' export class NativeSearchError extends Error { constructor( @@ -11,7 +18,8 @@ export class NativeSearchError extends Error { message: string, readonly retryAfterSeconds?: number, /** The provider's HTTP status, when the failure is a plain non-success response. */ - readonly httpStatus?: number + readonly httpStatus?: number, + readonly diagnosticCode: SearchFailureReason = 'provider_validation' ) { super(message) } @@ -33,104 +41,137 @@ export function createNativeClient(input: { requestBudget?: number }): NativeClient { let requests = 0 - async function request( + async function request( path: string, - options?: { - query?: Record - body?: unknown - googleService?: 'sheets' - signal?: AbortSignal - } + options: + | { + query?: Record + body?: unknown + googleService?: 'sheets' + signal?: AbortSignal + } + | undefined, + read: (response: Awaited>) => Promise ) { const signal = options?.signal ? AbortSignal.any([input.signal, options.signal]) : input.signal - signal.throwIfAborted() - if (++requests > (input.requestBudget ?? NATIVE_SEARCH_REQUEST_BUDGET)) - throw new NativeSearchError('unavailable', 'Request budget reached. Narrow the query.') - const url = new URL(input.origin) - if (options?.googleService === 'sheets') { - if (url.origin !== 'https://www.googleapis.com') throw new Error('Invalid Google service') - url.hostname = 'sheets.googleapis.com' - } - if (!path.startsWith('/') || path.startsWith('//') || path.includes('\\')) - throw new Error('Invalid provider path') - url.pathname = path - for (const [key, value] of Object.entries(options?.query ?? {})) { - for (const item of Array.isArray(value) ? value : [value]) url.searchParams.append(key, item) - } - const response = await secureFetchWithValidation(url.toString(), { - profile: 'configuredEndpoint', - headers: { - Authorization: `Bearer ${input.accessToken}`, - Accept: - url.hostname === 'api.github.com' - ? 'application/vnd.github.text-match+json' - : 'application/json', - ...(url.hostname === 'api.github.com' ? { 'X-GitHub-Api-Version': '2026-03-10' } : {}), - ...(options?.body ? { 'Content-Type': 'application/json' } : {}), - }, - method: options?.body ? 'POST' : 'GET', - ...(options?.body ? { body: JSON.stringify(options.body) } : {}), - signal, - timeout: 10_000, - maxResponseBytes: NATIVE_RESPONSE_MAX_BYTES, - maxRedirects: 0, - acceptCompressed: true, - connectionPool: input.pool, - }) - if (!response.ok) { - /** Reading the small error body also returns a pooled connection for reuse. */ - let quotaExceeded = false - if (response.status === 403 && url.hostname.endsWith('.googleapis.com')) - quotaExceeded = isGoogleQuotaError(await response.json().catch(() => null)) - else await response.text().catch(() => '') - if ( - response.status === 429 || - quotaExceeded || - (response.status === 403 && - (response.headers.get('retry-after') !== null || - response.headers.get('x-ratelimit-remaining') === '0')) - ) { - const retry = Number(response.headers.get('retry-after')) - const reset = Number(response.headers.get('x-ratelimit-reset')) - const seconds = retry > 0 ? retry : reset > 0 ? Math.ceil(reset - Date.now() / 1000) : 0 - throw new NativeSearchError( - 'rate_limited', - 'Provider rate limit reached. Try again later.', - Number.isFinite(seconds) && seconds > 0 ? seconds : undefined - ) + return withSearchRequestDiagnostics( + { transport: 'native_http', method: options?.body ? 'POST' : 'GET' }, + async (record) => { + signal.throwIfAborted() + if (++requests > (input.requestBudget ?? NATIVE_SEARCH_REQUEST_BUDGET)) + throw new NativeSearchError( + 'unavailable', + 'Request budget reached. Narrow the query.', + undefined, + undefined, + 'request_budget_exceeded' + ) + const url = new URL(input.origin) + if (options?.googleService === 'sheets') { + if (url.origin !== 'https://www.googleapis.com') throw new Error('Invalid Google service') + url.hostname = 'sheets.googleapis.com' + } + if (!path.startsWith('/') || path.startsWith('//') || path.includes('\\')) + throw new Error('Invalid provider path') + url.pathname = path + for (const [key, value] of Object.entries(options?.query ?? {})) { + for (const item of Array.isArray(value) ? value : [value]) + url.searchParams.append(key, item) + } + const response = await secureFetchWithValidation(url.toString(), { + profile: 'configuredEndpoint', + headers: { + Authorization: `Bearer ${input.accessToken}`, + Accept: + url.hostname === 'api.github.com' + ? 'application/vnd.github.text-match+json' + : 'application/json', + ...(url.hostname === 'api.github.com' ? { 'X-GitHub-Api-Version': '2026-03-10' } : {}), + ...(options?.body ? { 'Content-Type': 'application/json' } : {}), + }, + method: options?.body ? 'POST' : 'GET', + ...(options?.body ? { body: JSON.stringify(options.body) } : {}), + signal, + timeout: 10_000, + maxResponseBytes: NATIVE_RESPONSE_MAX_BYTES, + maxRedirects: 0, + acceptCompressed: true, + connectionPool: input.pool, + }) + record({ httpStatus: response.status }) + if (!response.ok) { + /** Reading the small error body also returns a pooled connection for reuse. */ + let quotaExceeded = false + if (response.status === 403 && url.hostname.endsWith('.googleapis.com')) { + const body = await response.json().catch(() => null) + quotaExceeded = isGoogleQuotaError(body) + const error = object(object(body).error) + record({ + providerReasons: safeGoogleErrorReasons( + [...array(error.errors), ...array(error.details)].map((entry) => + string(entry.reason) + ) + ), + }) + } else await response.text().catch(() => '') + if ( + response.status === 429 || + quotaExceeded || + (response.status === 403 && + (response.headers.get('retry-after') !== null || + response.headers.get('x-ratelimit-remaining') === '0')) + ) { + const retry = Number(response.headers.get('retry-after')) + const reset = Number(response.headers.get('x-ratelimit-reset')) + const seconds = retry > 0 ? retry : reset > 0 ? Math.ceil(reset - Date.now() / 1000) : 0 + throw new NativeSearchError( + 'rate_limited', + 'Provider rate limit reached. Try again later.', + Number.isFinite(seconds) && seconds > 0 ? seconds : undefined, + response.status, + 'provider_rate_limited' + ) + } + if (response.status === 401 || response.status === 403) + throw new NativeSearchError( + 'reconnect', + url.hostname === 'api.github.com' + ? 'GitHub denied access. Reconnect your account and check that the GitHub App has access to this repository, with read permissions for Contents, Issues, and Pull requests.' + : 'The provider denied access. Reconnect your account with search and read permissions.', + undefined, + response.status, + 'provider_access_denied' + ) + throw new NativeSearchError( + 'unavailable', + response.status === 400 || response.status === 422 + ? `The provider rejected this query (${response.status}). Check its native query syntax and supported search scope.` + : `Provider request failed (${response.status}).`, + undefined, + response.status, + 'provider_http_error' + ) + } + return read(response) } - if (response.status === 401 || response.status === 403) - throw new NativeSearchError( - 'reconnect', - url.hostname === 'api.github.com' - ? 'GitHub denied access. Reconnect your account and check that the GitHub App has access to this repository, with read permissions for Contents, Issues, and Pull requests.' - : 'The provider denied access. Reconnect your account with search and read permissions.' - ) - throw new NativeSearchError( - 'unavailable', - response.status === 400 || response.status === 422 - ? `The provider rejected this query (${response.status}). Check its native query syntax and supported search scope.` - : `Provider request failed (${response.status}).`, - undefined, - response.status - ) - } - return response + ) } return withJsonMemo({ async json(path, options) { - return (await request(path, options)).json() + return request(path, options, (response) => response.json()) }, async bytes(path, query) { try { - return Buffer.from(await (await request(path, { query })).arrayBuffer()) + return await request(path, { query }, async (response) => + Buffer.from(await response.arrayBuffer()) + ) } catch (error) { input.signal.throwIfAborted() throw error } }, async text(path, query) { - return (await request(path, { query })).text() + return request(path, { query }, (response) => response.text()) }, }) } diff --git a/apps/sim/lib/sim-search/live/managed-mcp-payload.ts b/apps/sim/lib/sim-search/live/managed-mcp-payload.ts index e97d98f97d6..b20140adac7 100644 --- a/apps/sim/lib/sim-search/live/managed-mcp-payload.ts +++ b/apps/sim/lib/sim-search/live/managed-mcp-payload.ts @@ -10,7 +10,10 @@ export function managedMcpPayload(result: McpToolResult, label: string): unknown if (!isJsonWithinByteLimit(result, MAX_SEARCH_MCP_PAYLOAD_BYTES)) throw new NativeSearchError( 'unavailable', - `${label} response exceeded the search size limit. Narrow the query.` + `${label} response exceeded the search size limit. Narrow the query.`, + undefined, + undefined, + 'mcp_payload_too_large' ) if (result.isError) { if ( @@ -23,7 +26,10 @@ export function managedMcpPayload(result: McpToolResult, label: string): unknown ) throw new NativeSearchError( 'reconnect', - 'Reconnect using an existing Granola account. Check the account email in the Granola app.' + 'Reconnect using an existing Granola account. Check the account email in the Granola app.', + undefined, + undefined, + 'mcp_account_required' ) const quota = result.content?.some( (block) => @@ -34,7 +40,10 @@ export function managedMcpPayload(result: McpToolResult, label: string): unknown quota ? 'rate_limited' : 'unavailable', quota ? `${label} MCP request limit reached. Try again when it resets.` - : `${label} could not complete this read. Check the query and your access.` + : `${label} could not complete this read. Check the query and your access.`, + undefined, + undefined, + quota ? 'provider_rate_limited' : 'mcp_tool_error' ) } const unwrap = (value: unknown) => @@ -51,7 +60,14 @@ export function managedMcpPayload(result: McpToolResult, label: string): unknown .filter((block) => block.type === 'text') .map((block) => block.text ?? '') .join('\n') - if (!text) throw new NativeSearchError('unavailable', `${label} returned no readable content.`) + if (!text) + throw new NativeSearchError( + 'unavailable', + `${label} returned no readable content.`, + undefined, + undefined, + 'mcp_payload_unreadable' + ) try { return unwrap(JSON.parse(text)) } catch { diff --git a/apps/sim/lib/sim-search/live/managed-mcp.ts b/apps/sim/lib/sim-search/live/managed-mcp.ts index ee56fa6000e..3bcfc9fbbce 100644 --- a/apps/sim/lib/sim-search/live/managed-mcp.ts +++ b/apps/sim/lib/sim-search/live/managed-mcp.ts @@ -1,6 +1,7 @@ import { isRecordLike, toRecord } from '@sim/utils/object' import type { ResourceOwner } from '@/lib/core/resource-scope' import { MANAGED_MCP_CONNECTORS } from '@/lib/credential-groups/managed-mcp-connectors' +import { withSearchRequestDiagnostics } from '@/lib/knowledge/search/diagnostics' import { createManagedMcpAuthProvider } from '@/lib/mcp/application/managed-auth-provider' import { mcpService } from '@/lib/mcp/service' import { compileMcpToolSchema } from '@/lib/mcp/tool-schema' @@ -74,32 +75,46 @@ export async function createManagedSearchMcpClient( return true }, async call(name, args) { - signal.throwIfAborted() - if (!allowed.includes(name)) - throw new NativeSearchError('unavailable', `${label} Search permits read-only tools.`) - if (++requests > budget) - throw new NativeSearchError( - 'unavailable', - `${label} read request limit reached. Narrow the query.` - ) - const tool = byName.get(name) - if (!tool) - throw new NativeSearchError( - 'unavailable', - `${label} no longer advertises ${name}. Reconnect or update the connector.` - ) - if (!compileMcpToolSchema(tool.inputSchema)(args)) - throw new NativeSearchError( - 'unavailable', - `${label} rejected these search arguments. Its current tool schema is incompatible with this query.` - ) - await loadCurrent() - const result = await session.callTool( - { name, arguments: args }, - { signal, timeoutMs: 10_000 } + return withSearchRequestDiagnostics( + { transport: 'mcp', tool: allowed.includes(name) ? name : undefined }, + async () => { + signal.throwIfAborted() + if (!allowed.includes(name)) + throw new NativeSearchError('unavailable', `${label} Search permits read-only tools.`) + if (++requests > budget) + throw new NativeSearchError( + 'unavailable', + `${label} read request limit reached. Narrow the query.`, + undefined, + undefined, + 'request_budget_exceeded' + ) + const tool = byName.get(name) + if (!tool) + throw new NativeSearchError( + 'unavailable', + `${label} no longer advertises ${name}. Reconnect or update the connector.`, + undefined, + undefined, + 'mcp_tool_unavailable' + ) + if (!compileMcpToolSchema(tool.inputSchema)(args)) + throw new NativeSearchError( + 'unavailable', + `${label} rejected these search arguments. Its current tool schema is incompatible with this query.`, + undefined, + undefined, + 'mcp_schema_incompatible' + ) + await loadCurrent() + const result = await session.callTool( + { name, arguments: args }, + { signal, timeoutMs: 10_000 } + ) + await loadCurrent() + return managedMcpPayload(result, label) + } ) - await loadCurrent() - return managedMcpPayload(result, label) }, } } diff --git a/scripts/check-test-types.baseline.json b/scripts/check-test-types.baseline.json index 9b8fb20b0dc..23e8a675324 100644 --- a/scripts/check-test-types.baseline.json +++ b/scripts/check-test-types.baseline.json @@ -1719,9 +1719,6 @@ "apps/sim/lib/knowledge/reranker.test.ts": { "TS2540": 2 }, - "apps/sim/lib/knowledge/search/diagnostics.test.ts": { - "TS2345": 2 - }, "apps/sim/lib/knowledge/search/prewarm.test.ts": { "TS2345": 4 },