diff --git a/packages/plugins/semantic-search/src/api/group.ts b/packages/plugins/semantic-search/src/api/group.ts index 9a36dadc42..78e61ebc43 100644 --- a/packages/plugins/semantic-search/src/api/group.ts +++ b/packages/plugins/semantic-search/src/api/group.ts @@ -24,6 +24,7 @@ export const ReindexResponse = Schema.Struct({ namespace: Schema.String, total: Schema.Number, indexed: Schema.Number, + failed: Schema.optional(Schema.Number), skipped: Schema.Number, removed: Schema.Number, offset: Schema.optional(Schema.Number), diff --git a/packages/plugins/semantic-search/src/sdk/ai-search.test.ts b/packages/plugins/semantic-search/src/sdk/ai-search.test.ts index 097a0be814..d9013247e1 100644 --- a/packages/plugins/semantic-search/src/sdk/ai-search.test.ts +++ b/packages/plugins/semantic-search/src/sdk/ai-search.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from "@effect/vitest"; import type { AiSearchInstance } from "@cloudflare/workers-types"; -import { type PluginStorageCollectionFacade, type PluginStorageEntry } from "@executor-js/sdk/core"; +import { + StorageError, + type PluginStorageCollectionFacade, + type PluginStorageEntry, +} from "@executor-js/sdk/core"; import { Effect } from "effect"; import { @@ -10,6 +14,7 @@ import { statusAiSearch, } from "./ai-search"; import { type aiSearchItems, type AiSearchItemRow } from "./collections"; +import { toolItemKey } from "./documents"; import { cyrb53 } from "./fingerprint"; type ItemsCollection = PluginStorageCollectionFacade; @@ -149,11 +154,10 @@ describe("makeAiSearchToolDiscoveryProvider", () => { Effect.gen(function* () { const provider = makeAiSearchToolDiscoveryProvider({ aiSearch: makeAiSearch(), - items: undefined, }); const page = yield* provider!.searchTools({ - executor: undefined as never, + executor: { tools: { manifest: () => Effect.succeed([]) } } as never, query: "create repo", limit: 10, offset: 0, @@ -173,7 +177,6 @@ describe("makeAiSearchToolDiscoveryProvider", () => { Effect.gen(function* () { const provider = makeAiSearchToolDiscoveryProvider({ aiSearch: makeAiSearch(), - items: undefined, }); const unfiltered = yield* provider!.searchTools({ @@ -206,7 +209,6 @@ describe("makeAiSearchToolDiscoveryProvider", () => { return makeAiSearch().search(input); }, }, - items: undefined, }); yield* provider!.searchTools({ @@ -250,14 +252,6 @@ describe("makeAiSearchToolDiscoveryProvider", () => { ], }), }, - items: makeItemsCollection({ - getMany: ({ keys }) => - Effect.succeed( - new Map( - keys.flatMap((key) => (key === githubRow.key ? [[key, githubRow] as const] : [])), - ), - ), - }), }); const page = yield* provider!.searchTools({ @@ -277,44 +271,110 @@ describe("makeAiSearchToolDiscoveryProvider", () => { }), ); - it.effect("ignores AI Search chunks whose paths are not current locally", () => + it.effect("pushes an integration namespace into AI Search retrieval", () => + Effect.gen(function* () { + const requests: Parameters[0][] = []; + const provider = makeAiSearchToolDiscoveryProvider({ + aiSearch: { + ...makeAiSearch(), + search: async (input) => { + requests.push(input); + return makeAiSearch().search(input); + }, + }, + }); + + yield* provider!.searchTools({ + executor: { tools: { manifest: () => Effect.succeed([]) } } as never, + query: "authenticated user", + limit: 5, + offset: 0, + }); + yield* provider!.searchTools({ + executor: { tools: { manifest: () => Effect.succeed([]) } } as never, + query: "authenticated user", + namespace: "github_api", + limit: 5, + offset: 0, + }); + + expect(requests[0]?.ai_search_options?.retrieval).toMatchObject({ + retrieval_type: "hybrid", + match_threshold: 0.1, + max_num_results: 50, + return_on_failure: true, + }); + expect(requests[0]?.ai_search_options?.reranking).toEqual({ + enabled: true, + match_threshold: 0.1, + }); + expect(requests[0]?.ai_search_options?.query_rewrite).toBeUndefined(); + expect(requests[1]?.ai_search_options?.retrieval).toMatchObject({ + retrieval_type: "hybrid", + match_threshold: 0.1, + max_num_results: 50, + filters: { integration: { $eq: "github_api" } }, + return_on_failure: true, + }); + expect(requests[1]?.ai_search_options?.reranking).toEqual({ + enabled: true, + match_threshold: 0.1, + }); + }), + ); + + it.effect("returns AI Search chunks while the local indexing ledger lags", () => Effect.gen(function* () { const provider = makeAiSearchToolDiscoveryProvider({ aiSearch: makeAiSearch(), - items: makeItemsCollection({ - getMany: ({ keys }) => - Effect.succeed( - new Map( - keys.flatMap((key) => (key === githubRow.key ? [[key, githubRow] as const] : [])), - ), - ), - }), }); const page = yield* provider!.searchTools({ - executor: undefined as never, + executor: { tools: { manifest: () => Effect.succeed([]) } } as never, query: "tool", limit: 10, offset: 0, }); - expect(page.items.map((item) => item.path)).toEqual(["github.default.main.repos.create"]); - expect(page.total).toBe(1); + expect(page.items.map((item) => item.path)).toEqual([ + "github.default.main.repos.create", + "slack.default.main.messages.send", + ]); + expect(page.total).toBe(2); }), ); - it.effect("returns an empty page when no returned paths are current locally", () => + it.effect("returns an empty page when AI Search finds no chunks", () => Effect.gen(function* () { const provider = makeAiSearchToolDiscoveryProvider({ - aiSearch: makeAiSearch(), - items: makeItemsCollection({ - getMany: () => Effect.succeed(new Map()), - }), + aiSearch: { + ...makeAiSearch(), + search: async () => ({ search_query: "stripe list balance", chunks: [] }), + }, }); const page = yield* provider!.searchTools({ - executor: undefined as never, - query: "tool", + executor: { + tools: { + manifest: () => + Effect.succeed([ + { + path: "stripe_api.org.main.balance.getBalance", + name: "balance.getBalance", + description: "Retrieve the current account balance.", + integration: "stripe_api", + }, + { + path: "stripe_api.org.main.customers.list", + name: "customers.list", + description: "List customers.", + integration: "stripe_api", + }, + ]), + }, + } as never, + query: "stripe list balance", + namespace: "stripe_api", limit: 10, offset: 0, }); @@ -327,6 +387,46 @@ describe("makeAiSearchToolDiscoveryProvider", () => { }); }), ); + + it.effect("surfaces AI Search failures instead of substituting a local ranking", () => + Effect.gen(function* () { + const provider = makeAiSearchToolDiscoveryProvider({ + aiSearch: { + ...makeAiSearch(), + search: () => { + const deferred = + Promise.withResolvers>>(); + deferred.reject("AI Search unavailable"); + return deferred.promise; + }, + }, + }); + + const error = yield* Effect.flip( + provider!.searchTools({ + executor: { + tools: { + manifest: () => + Effect.succeed([ + { + path: "github_api.org.main.repos.listForAuthenticatedUser", + name: "repos.listForAuthenticatedUser", + description: "List repositories for the authenticated user.", + integration: "github_api", + }, + ]), + }, + } as never, + query: "list", + namespace: "github_api", + limit: 10, + offset: 0, + }), + ); + + expect(error).toMatchObject({ message: "AI Search tool search failed." }); + }), + ); }); describe("reindexAiSearch", () => { @@ -337,6 +437,16 @@ describe("reindexAiSearch", () => { const result = yield* reindexAiSearch({ executor: { + integrations: { + list: () => + Effect.succeed([ + { + slug: "github", + name: "GitHub", + description: "Repositories, issues, pull requests, actions, and users.", + }, + ]), + }, tools: { manifest: () => Effect.succeed([ @@ -375,9 +485,65 @@ describe("reindexAiSearch", () => { expect(result).toMatchObject({ indexed: 1, skipped: 0, removed: 0 }); expect(uploadedContent).toContain("# github.default.main.repos.create"); + expect(uploadedContent).toContain("Integration name: GitHub"); + expect(uploadedContent).toContain( + "Integration purpose: Repositories, issues, pull requests, actions, and users.", + ); expect(uploadedContent).toContain("Description: Create a repository"); expect(uploadedContent).not.toContain("Input schema"); - expect(stored[0]?.fingerprint).toBe("github.default.main.repos.create:v1:fingerprint:"); + expect(stored[0]?.fingerprint).toContain("ai-search-tool-document/v2:"); + expect(stored[0]?.fingerprint).toContain(":GitHub:Repositories, issues"); + }), + ); + + it.effect("continues indexing when integration context is unavailable", () => + Effect.gen(function* () { + let uploadedContent = ""; + const stored: AiSearchItemRow[] = []; + + const result = yield* reindexAiSearch({ + executor: { + integrations: { list: () => Effect.fail("integration list unavailable") }, + tools: { + manifest: () => + Effect.succeed([ + { + path: "github.default.main.repos.create", + name: "repos.create", + description: "Create a repository", + integration: "github", + fingerprintVersion: "v1", + indexFingerprint: "fingerprint", + }, + ]), + schema: () => Effect.fail("schema unavailable"), + }, + } as never, + aiSearch: { + ...makeAiSearch(), + items: { + ...makeAiSearchItems(), + upload: async (name, content) => { + uploadedContent = String(content); + return { id: `item:${name}`, key: name, status: "queued" }; + }, + }, + }, + items: makeItemsCollection({ + list: () => Effect.succeed([]), + putMany: ({ entries }) => + Effect.sync(() => { + stored.push(...entries.map((entry) => entry.data)); + }), + }), + owner: "org", + namespace: "org", + }); + + expect(result).toMatchObject({ indexed: 1, skipped: 0, removed: 0 }); + expect(uploadedContent).toContain("# github.default.main.repos.create"); + expect(uploadedContent).not.toContain("Integration name:"); + expect(stored).toHaveLength(1); }), ); @@ -386,6 +552,7 @@ describe("reindexAiSearch", () => { const removed: string[] = []; const result = yield* reindexAiSearch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed([]), }, @@ -422,6 +589,7 @@ describe("reindexAiSearch", () => { const stored: AiSearchItemRow[] = []; const result = yield* reindexAiSearch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed([ @@ -441,11 +609,7 @@ describe("reindexAiSearch", () => { ...makeAiSearch(), items: { ...makeAiSearchItems(), - upload: async (name) => ({ - id: `new:${name}`, - key: name, - status: "queued", - }), + upload: async (name) => ({ id: `new:${name}`, key: name, status: "completed" }), delete: async (id) => { deleted.push(id); }, @@ -466,10 +630,68 @@ describe("reindexAiSearch", () => { expect(result).toMatchObject({ indexed: 1, skipped: 0, removed: 0 }); expect(stored[0]?.itemId).toMatch(/^new:tool-[a-z0-9]+\.md$/); expect(stored[0]?.key).toBe(stored[0]?.itemId.replace(/^new:/, "")); + expect(stored[0]?.pendingDeleteItemIds).toEqual([githubRow.data.itemId]); expect(deleted).toEqual(["item:github.repos.create.md"]); }), ); + it.effect("keeps the previous remote item when local row persistence fails", () => + Effect.gen(function* () { + const deleted: string[] = []; + const error = yield* Effect.flip( + reindexAiSearch({ + executor: { + integrations: { list: () => Effect.succeed([]) }, + tools: { + manifest: () => + Effect.succeed([ + { + path: "github.default.main.repos.create", + name: "repos.create", + description: "Create a repository", + integration: "github", + fingerprintVersion: "v1", + indexFingerprint: "new-fingerprint", + }, + ]), + schema: () => Effect.fail("schema unavailable"), + }, + } as never, + aiSearch: { + ...makeAiSearch(), + items: { + ...makeAiSearchItems(), + get: () => ({ + info: async () => ({ + id: githubRow.data.itemId, + key: githubRow.data.key, + status: "error" as const, + }), + download: async () => expect.unreachable("Unexpected AI Search item download"), + }), + upload: async (name) => ({ id: `new:${name}`, key: name, status: "completed" }), + delete: async (id) => { + deleted.push(id); + }, + }, + }, + items: makeItemsCollection({ + getManyForOwner: () => Effect.succeed(new Map([[githubRow.key, githubRow]])), + putMany: () => + Effect.fail(new StorageError({ message: "row persistence failed", cause: "test" })), + }), + owner: "org", + namespace: "org", + }), + ); + + expect(error).toMatchObject({ message: "Failed to record AI Search item rows." }); + expect(deleted).toHaveLength(1); + expect(deleted[0]).not.toBe(githubRow.data.itemId); + expect(deleted[0]).toMatch(/^new:tool-[a-z0-9]+\.md$/); + }), + ); + it.effect("records uploaded rows in bounded batches", () => Effect.gen(function* () { const putManySizes: number[] = []; @@ -484,6 +706,7 @@ describe("reindexAiSearch", () => { const result = yield* reindexAiSearch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed(manifests), schema: () => Effect.fail("schema unavailable"), @@ -526,7 +749,7 @@ describe("reindexAiSearch", () => { fingerprintVersion: "v1", indexFingerprint: "fingerprint", }; - const fingerprint = "github.default.main.repos.create:v1:fingerprint:"; + const fingerprint = toolItemKey(manifest); const itemName = `tool-${cyrb53(`${manifest.path}\u0000${fingerprint}`).toString(36)}.md`; const existing = { ...githubRow, @@ -540,6 +763,7 @@ describe("reindexAiSearch", () => { const result = yield* reindexAiSearchBatch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed([manifest]), schema: () => Effect.fail("schema unavailable"), @@ -588,6 +812,7 @@ describe("reindexAiSearch", () => { const result = yield* reindexAiSearchBatch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed(manifests), schema: () => Effect.fail("schema unavailable"), @@ -645,12 +870,13 @@ describe("reindexAiSearch", () => { ...githubRow, data: { ...githubRow.data, - fingerprint: "github.default.main.repos.create:v1:fingerprint:", + fingerprint: toolItemKey(manifest), }, }; const result = yield* reindexAiSearch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed([manifest]), schema: () => Effect.fail("schema unavailable"), @@ -688,6 +914,161 @@ describe("reindexAiSearch", () => { }), ); + it.effect("replaces remote items that report an outdated AI Search status", () => + Effect.gen(function* () { + const deleted: string[] = []; + const stored: AiSearchItemRow[] = []; + const manifest = { + path: "github.default.main.repos.create", + name: "repos.create", + description: "Create a repository", + integration: "github", + fingerprintVersion: "v1", + indexFingerprint: "fingerprint", + }; + const fingerprint = toolItemKey(manifest); + const itemName = `tool-${cyrb53(`${manifest.path}\u0000${fingerprint}`).toString(36)}.md`; + const existing = { + ...githubRow, + data: { + ...githubRow.data, + key: itemName, + itemId: "stale:item", + fingerprint, + pendingDeleteItemIds: ["previous:item"], + }, + }; + + const result = yield* reindexAiSearchBatch({ + executor: { + integrations: { list: () => Effect.succeed([]) }, + tools: { + manifest: () => Effect.succeed([manifest]), + schema: () => Effect.fail("schema unavailable"), + }, + } as never, + aiSearch: { + ...makeAiSearch(), + items: { + ...makeAiSearchItems(), + get: () => ({ + info: async () => ({ + id: existing.data.itemId, + key: existing.data.key, + status: "outdated" as never, + }), + download: async () => expect.unreachable("Unexpected AI Search item download"), + }), + upload: async (name) => ({ + id: `replacement:${name}`, + key: name, + status: "completed", + }), + delete: async (id) => { + deleted.push(id); + }, + }, + }, + items: makeItemsCollection({ + getManyForOwner: () => Effect.succeed(new Map([[manifest.path, existing]])), + list: () => Effect.succeed([existing]), + putMany: ({ entries }) => + Effect.sync(() => { + stored.push(...entries.map((entry) => entry.data)); + }), + }), + owner: "org", + namespace: "org", + offset: 0, + pageSize: 1, + }); + + expect(result).toMatchObject({ indexed: 1, skipped: 0 }); + expect(deleted).toEqual(expect.arrayContaining(["stale:item", "previous:item"])); + expect(stored[0]?.itemId).toBe("replacement:" + itemName); + expect(stored.at(-1)?.pendingDeleteItemIds).toBeUndefined(); + }), + ); + + it.effect("retains every failed replacement deletion for retry", () => + Effect.gen(function* () { + const deleted: string[] = []; + const stored: AiSearchItemRow[] = []; + const manifest = { + path: "github.default.main.repos.create", + name: "repos.create", + description: "Create a repository", + integration: "github", + fingerprintVersion: "v1", + indexFingerprint: "fingerprint", + }; + const fingerprint = toolItemKey(manifest); + const itemName = `tool-${cyrb53(`${manifest.path}\u0000${fingerprint}`).toString(36)}.md`; + const existing = { + ...githubRow, + data: { + ...githubRow.data, + key: itemName, + itemId: "stale:item", + fingerprint, + pendingDeleteItemIds: ["previous:item"], + }, + }; + + const result = yield* reindexAiSearchBatch({ + executor: { + integrations: { list: () => Effect.succeed([]) }, + tools: { + manifest: () => Effect.succeed([manifest]), + schema: () => Effect.fail("schema unavailable"), + }, + } as never, + aiSearch: { + ...makeAiSearch(), + items: { + ...makeAiSearchItems(), + get: () => ({ + info: async () => ({ + id: existing.data.itemId, + key: existing.data.key, + status: "outdated" as never, + }), + download: async () => expect.unreachable("Unexpected AI Search item download"), + }), + upload: async (name) => ({ + id: `replacement:${name}`, + key: name, + status: "completed", + }), + delete: async (id) => { + deleted.push(id); + if (id === "previous:item") { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: test double for rejected AI Search delete promise + throw new Error("previous deletion failed"); + } + }, + }, + }, + items: makeItemsCollection({ + getManyForOwner: () => Effect.succeed(new Map([[manifest.path, existing]])), + list: () => Effect.succeed([existing]), + putMany: ({ entries }) => + Effect.sync(() => { + stored.push(...entries.map((entry) => entry.data)); + }), + }), + owner: "org", + namespace: "org", + offset: 0, + pageSize: 1, + }); + + expect(result).toMatchObject({ indexed: 1, skipped: 0 }); + expect(deleted).toEqual(expect.arrayContaining(["stale:item", "previous:item"])); + expect(stored.at(-1)?.pendingDeleteItemIds).toEqual(["previous:item"]); + }), + ); + it.effect("reads status from instance statistics without listing every remote item", () => Effect.gen(function* () { const status = yield* statusAiSearch({ @@ -736,6 +1117,7 @@ describe("reindexAiSearch", () => { yield* reindexAiSearch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed([ diff --git a/packages/plugins/semantic-search/src/sdk/ai-search.ts b/packages/plugins/semantic-search/src/sdk/ai-search.ts index d90c443016..3c152dc080 100644 --- a/packages/plugins/semantic-search/src/sdk/ai-search.ts +++ b/packages/plugins/semantic-search/src/sdk/ai-search.ts @@ -11,7 +11,7 @@ import { type ToolDiscoveryProvider, type ToolDiscoveryResult, } from "@executor-js/sdk/core"; -import { Effect } from "effect"; +import { Effect, Predicate, Result } from "effect"; import { type AiSearchItemRow, aiSearchItems, type AiSearchItemStatus } from "./collections"; import { @@ -47,6 +47,11 @@ export interface AiSearchToolSearchBackendStorage { const DEFAULT_SEARCH_LIMIT = 20; const AI_SEARCH_UPLOAD_CONCURRENCY = 2; const AI_SEARCH_UPLOAD_BATCH_SIZE = 25; +// Cloudflare's default reranking threshold (0.4) drops short, valid catalog +// queries such as "search web" before their hybrid matches reach the caller. +// Keep AI Search authoritative while lowering only its native retrieval/rerank +// cutoff for this sparse tool catalog. +const AI_SEARCH_MATCH_THRESHOLD = 0.1; const nowIso = (): string => new Date().toISOString(); @@ -63,6 +68,13 @@ const toStatus = (status: string | undefined): AiSearchItemStatus => const toItemName = (document: ToolSearchDocument): string => `tool-${cyrb53(`${document.path}\u0000${document.fingerprint}`).toString(36)}.md`; +const uniqueItemIds = (ids: readonly (string | undefined)[]): string[] => [ + ...new Set(ids.filter(Predicate.isNotUndefined)), +]; + +const pendingDeleteItemIdsForRow = (row: AiSearchItemRow): string[] => + uniqueItemIds([...(row.pendingDeleteItemIds ?? []), row.pendingDeleteItemId]); + const normalizeBatchInput = ( input: SemanticSearchReindexBatchInput, ): SemanticSearchReindexBatchInput => ({ @@ -137,6 +149,7 @@ const getAiSearchItem = ( const toIndexedItemRow = ( document: ToolSearchDocument, uploaded: AiSearchItemInfo, + pendingDeleteItemIds?: readonly string[], ): AiSearchItemRow => ({ path: document.path, key: uploaded.key, @@ -149,14 +162,32 @@ const toIndexedItemRow = ( fingerprint: document.fingerprint, status: toStatus(uploaded.status), updatedAt: nowIso(), + ...(pendingDeleteItemIds === undefined || pendingDeleteItemIds.length === 0 + ? {} + : { pendingDeleteItemIds: uniqueItemIds(pendingDeleteItemIds) }), }); +const withPendingDeleteItemIds = ( + row: AiSearchItemRow, + pendingDeleteItemIds: readonly string[], +): AiSearchItemRow => { + const { + pendingDeleteItemId: _pendingDeleteItemId, + pendingDeleteItemIds: _previousPendingDeleteItemIds, + ...base + } = row; + const uniquePendingDeleteItemIds = uniqueItemIds(pendingDeleteItemIds); + return uniquePendingDeleteItemIds.length === 0 + ? base + : { ...base, pendingDeleteItemIds: uniquePendingDeleteItemIds }; +}; + interface UploadedDocument { readonly deleteOnStorageFailure: boolean; - readonly previousItemId?: string; readonly uploadedItemId: string; readonly key: string; readonly row: AiSearchItemRow; + readonly deferredDeleteItemIds: readonly string[]; } const uploadDocument = ( @@ -167,19 +198,35 @@ const uploadDocument = ( ): Effect.Effect => Effect.gen(function* () { const itemName = toItemName(document); + const previousPendingDeleteItemIds = + previous === undefined ? [] : pendingDeleteItemIdsForRow(previous); if (remote !== undefined && isReusableRemoteStatus(remote.status)) { + if (previousPendingDeleteItemIds.length > 0 && remote.status === "completed") { + const deletions = yield* Effect.forEach( + previousPendingDeleteItemIds, + (itemId) => deleteItem(aiSearch, itemId).pipe(Effect.result), + { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY }, + ); + const failedPendingDeleteItemIds = previousPendingDeleteItemIds.filter((_, index) => + Result.isFailure(deletions[index]), + ); + return { + deleteOnStorageFailure: false, + uploadedItemId: remote.id, + key: document.path, + row: toIndexedItemRow(document, remote, failedPendingDeleteItemIds), + deferredDeleteItemIds: [], + }; + } return { deleteOnStorageFailure: false, uploadedItemId: remote.id, key: document.path, - row: toIndexedItemRow(document, remote), + row: toIndexedItemRow(document, remote, previousPendingDeleteItemIds), + deferredDeleteItemIds: [], }; } - if (remote !== undefined) { - yield* deleteItemBestEffort(aiSearch, remote.id); - } - const uploaded = yield* Effect.tryPromise({ try: () => aiSearch.items.upload(itemName, document.content, { @@ -188,14 +235,20 @@ const uploadDocument = ( catch: mapUploadError(document), }); + const deferredDeleteItemIds = uniqueItemIds( + remote !== undefined && !isReusableRemoteStatus(remote.status) + ? [remote.id, ...previousPendingDeleteItemIds] + : remote === undefined && previous !== undefined && previous.key !== itemName + ? [previous.itemId, ...previousPendingDeleteItemIds] + : previousPendingDeleteItemIds, + ); + return { deleteOnStorageFailure: true, - ...(previous !== undefined && previous.key !== itemName - ? { previousItemId: previous.itemId } - : {}), uploadedItemId: uploaded.id, key: document.path, - row: toIndexedItemRow(document, uploaded), + row: toIndexedItemRow(document, uploaded, deferredDeleteItemIds), + deferredDeleteItemIds: toStatus(uploaded.status) === "completed" ? deferredDeleteItemIds : [], }; }); @@ -213,9 +266,28 @@ export const reindexAiSearchBatch = (input: { const aiSearch = input.aiSearch; return Effect.gen(function* () { const batch = normalizeBatchInput(input); - const manifests = yield* listToolManifests(input.executor, { - maxTools: batch.maxTools, - }); + const [manifests, integrations] = yield* Effect.all( + [ + listToolManifests(input.executor, { maxTools: batch.maxTools }), + input.executor.integrations.list().pipe( + Effect.catch((cause) => + Effect.sync(() => { + console.warn( + JSON.stringify({ + event: "tool_search_index_integration_context_failed", + cause, + }), + ); + return []; + }), + ), + ), + ] as const, + { concurrency: 2 }, + ); + const integrationBySlug = new Map( + integrations.map((integration) => [String(integration.slug), integration] as const), + ); const page = manifests.slice(batch.offset, batch.offset + batch.pageSize); const nextOffset = batch.offset + page.length < manifests.length ? batch.offset + page.length : null; @@ -232,53 +304,97 @@ export const reindexAiSearchBatch = (input: { ); const prepared = page.map((manifest) => ({ manifest, - fingerprint: toolItemKey(manifest), + integration: integrationBySlug.get(manifest.integration), + fingerprint: toolItemKey(manifest, integrationBySlug.get(manifest.integration)), previous: existingByPath.get(manifest.path), })); - const remoteByKey = new Map( - yield* Effect.forEach( - prepared.flatMap(({ previous, fingerprint }) => - previous?.fingerprint === fingerprint ? [previous] : [], + const remoteCandidates = prepared.flatMap(({ previous, fingerprint }) => + previous?.fingerprint === fingerprint ? [previous] : [], + ); + const remoteLookupResults = yield* Effect.forEach( + remoteCandidates, + (previous) => + getAiSearchItem(aiSearch, previous.itemId).pipe( + Effect.map((item) => [previous, item] as const), + Effect.result, ), - (previous) => - getAiSearchItem(aiSearch, previous.itemId).pipe( - Effect.map((item) => [previous.key, item] as const), - ), - { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY }, - ), + { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY }, ); + const remoteByKey = new Map(); + const remoteLookupFailures = new Set(); + for (const [index, result] of remoteLookupResults.entries()) { + const previous = remoteCandidates[index]; + if (Result.isSuccess(result)) { + remoteByKey.set(result.success[0].key, result.success[1]); + continue; + } + if (previous !== undefined) remoteLookupFailures.add(previous.key); + console.warn( + JSON.stringify({ + event: "tool_search_index_remote_lookup_failed", + key: previous?.key, + itemId: previous?.itemId, + cause: result.failure, + }), + ); + } let skipped = 0; + const failedPaths: string[] = []; const changed: { readonly manifest: (typeof manifests)[number]; + readonly integration?: (typeof integrations)[number]; readonly previous?: AiSearchItemRow; }[] = []; - for (const { manifest, fingerprint, previous } of prepared) { + for (const { manifest, integration, fingerprint, previous } of prepared) { const remote = previous === undefined ? undefined : remoteByKey.get(previous.key); + if (previous !== undefined && remoteLookupFailures.has(previous.key)) { + failedPaths.push(manifest.path); + continue; + } if ( previous?.fingerprint === fingerprint && remote !== undefined && - isReusableRemoteStatus(remote.status) + isReusableRemoteStatus(remote.status) && + pendingDeleteItemIdsForRow(previous).length === 0 ) { skipped += 1; continue; } changed.push({ manifest, + ...(integration === undefined ? {} : { integration }), ...(previous === undefined ? {} : { previous }), }); } - const uploaded = yield* Effect.forEach( + const uploadResults = yield* Effect.forEach( changed, - ({ manifest, previous }) => - collectToolSearchDocument(input.executor, manifest).pipe( + ({ manifest, integration, previous }) => + collectToolSearchDocument(input.executor, manifest, integration).pipe( Effect.flatMap((document) => uploadDocument(aiSearch, document, previous, remoteByKey.get(toItemName(document))), ), + Effect.result, ), { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY }, ); + const uploaded: UploadedDocument[] = []; + for (const [index, result] of uploadResults.entries()) { + if (Result.isSuccess(result)) { + uploaded.push(result.success); + } else { + const path = changed[index]?.manifest.path; + if (path !== undefined) failedPaths.push(path); + console.warn( + JSON.stringify({ + event: "tool_search_index_item_failed", + path, + cause: result.failure, + }), + ); + } + } if (uploaded.length > 0) { yield* input.items @@ -303,14 +419,69 @@ export const reindexAiSearchBatch = (input: { Effect.mapError(mapStorageError("Failed to record AI Search item rows.")), ); - yield* Effect.forEach( - uploaded, + const replacements = uploaded.filter( (entry) => - entry.previousItemId === undefined - ? Effect.void - : deleteItemBestEffort(aiSearch, entry.previousItemId), - { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY, discard: true }, + entry.row.status === "completed" && + (entry.deferredDeleteItemIds.length > 0 || + pendingDeleteItemIdsForRow(entry.row).length > 0), ); + if (replacements.length > 0) { + const cleanupResults = yield* Effect.forEach( + replacements, + (entry) => { + const itemIds = + entry.deferredDeleteItemIds.length > 0 + ? entry.deferredDeleteItemIds + : pendingDeleteItemIdsForRow(entry.row); + return Effect.forEach( + itemIds, + (itemId) => deleteItem(aiSearch, itemId).pipe(Effect.result), + { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY }, + ).pipe(Effect.map((results) => ({ entry, itemIds, results }))); + }, + { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY }, + ); + for (const { itemIds, results } of cleanupResults) { + for (const [index, result] of results.entries()) { + if (Result.isFailure(result)) { + console.warn( + JSON.stringify({ + event: "tool_search_index_previous_item_delete_failed", + itemId: itemIds[index], + cause: result.failure, + }), + ); + } + } + } + if (cleanupResults.length > 0) { + yield* input.items + .putMany({ + owner: input.owner, + entries: cleanupResults.map(({ entry, itemIds, results }) => ({ + key: entry.key, + data: withPendingDeleteItemIds( + entry.row, + itemIds.filter((_, index) => Result.isFailure(results[index])), + ), + })), + }) + .pipe( + Effect.mapError(mapStorageError("Failed to finalize AI Search item rows.")), + Effect.tapError((cause) => + Effect.sync(() => { + console.warn( + JSON.stringify({ + event: "tool_search_index_previous_item_cleanup_persist_failed", + cause, + }), + ); + }), + ), + Effect.catch(() => Effect.void), + ); + } + } } const removedEntries = shouldRemoveStale @@ -338,6 +509,7 @@ export const reindexAiSearchBatch = (input: { namespace: input.namespace, total: manifests.length, indexed: uploaded.length, + ...(failedPaths.length === 0 ? {} : { failed: failedPaths.length }), skipped, removed: removedEntries.length, offset: batch.offset, @@ -359,6 +531,7 @@ export const reindexAiSearch = (input: { let nextOffset: number | null = 0; let total = 0; let indexed = 0; + let failed = 0; let skipped = 0; let removed = 0; @@ -370,6 +543,7 @@ export const reindexAiSearch = (input: { }); total = result.total; indexed += result.indexed; + failed += result.failed ?? 0; skipped += result.skipped; removed += result.removed; nextOffset = result.nextOffset; @@ -379,6 +553,7 @@ export const reindexAiSearch = (input: { namespace: input.namespace, total, indexed, + ...(failed === 0 ? {} : { failed }), skipped, removed, }; @@ -448,7 +623,6 @@ const chunkToResult = ( export const makeAiSearchToolDiscoveryProvider = (deps: { readonly aiSearch: Pick | undefined; - readonly items: ItemsCollection | undefined; }): ToolDiscoveryProvider | undefined => { if (!deps.aiSearch) return undefined; const aiSearch = deps.aiSearch; @@ -467,14 +641,12 @@ export const makeAiSearchToolDiscoveryProvider = (deps: { ai_search_options: { retrieval: { retrieval_type: "hybrid", - // Retrieve a broad candidate set before deduplication and paging. Asking - // AI Search for only the caller's page size makes plausible tools vanish - // when several chunks belong to one tool or beat the desired integration. + match_threshold: AI_SEARCH_MATCH_THRESHOLD, max_num_results: 50, ...(integration ? { filters: { integration: { $eq: integration } } } : {}), return_on_failure: true, }, - reranking: { enabled: true }, + reranking: { enabled: true, match_threshold: AI_SEARCH_MATCH_THRESHOLD }, }, }), catch: (cause) => @@ -484,38 +656,18 @@ export const makeAiSearchToolDiscoveryProvider = (deps: { }), }); - // AI Search carries the canonical tool metadata on every chunk. Validate its - // path against the current catalog, rather than reconciling its opaque item key - // with the local upload ledger. That key is provider-owned and may be rewritten - // during an upload, while the catalog path is the stable identity clients use. + // AI Search carries the canonical tool metadata on every chunk. Its provider-owned + // item key may be rewritten during upload, but `path` is the stable identity that + // clients use. Do not gate a successful AI Search result through the local upload + // ledger: Cloudflare can finish indexing before that status projection advances. + // The ledger remains the indexer's recovery/status record, not query authority. const chunkResults = (response.chunks ?? []).flatMap((chunk) => { const result = chunkToResult(chunk); return result === null ? [] : [result]; }); - const visiblePaths = - deps.items === undefined - ? undefined - : new Set( - (yield* deps.items - .getMany({ keys: chunkResults.map((result) => result.path) }) - .pipe( - Effect.mapError( - (cause) => - new ExecutionToolError({ - message: "AI Search tool search failed.", - cause, - }), - ), - )).keys(), - ); - const bestByPath = new Map(); for (const result of chunkResults) { - if ( - (visiblePaths !== undefined && !visiblePaths.has(result.path)) || - !matchesNamespace(result.path, input.namespace) - ) - continue; + if (!matchesNamespace(result.path, input.namespace)) continue; const previous = bestByPath.get(result.path); if (!previous || result.score > previous.score) bestByPath.set(result.path, result); } @@ -550,7 +702,6 @@ export const makeAiSearchToolSearchBackend = ( build: ({ storage }) => { const provider = makeAiSearchToolDiscoveryProvider({ aiSearch: options.aiSearch, - items: storage.aiSearchItems, }); return { namespace, diff --git a/packages/plugins/semantic-search/src/sdk/collections.ts b/packages/plugins/semantic-search/src/sdk/collections.ts index 66457df39e..45625c684f 100644 --- a/packages/plugins/semantic-search/src/sdk/collections.ts +++ b/packages/plugins/semantic-search/src/sdk/collections.ts @@ -18,6 +18,17 @@ export const AiSearchItemRow = Schema.Struct({ status: AiSearchItemStatus, updatedAt: Schema.String, error: Schema.optional(Schema.String), + /** + * Provider items that stay live until their replacement is completed. This + * prevents an eventual-consistency gap while AI Search processes the newly + * uploaded document and allows partial cleanup failures to be retried. + */ + pendingDeleteItemIds: Schema.optional(Schema.Array(Schema.String)), + /** + * Legacy single-item recovery field. Rows written before the list field was + * introduced remain decodable and are normalized by the reindexer. + */ + pendingDeleteItemId: Schema.optional(Schema.String), }); export type AiSearchItemRow = typeof AiSearchItemRow.Type; diff --git a/packages/plugins/semantic-search/src/sdk/documents.ts b/packages/plugins/semantic-search/src/sdk/documents.ts index b3ccf3df55..32bec02287 100644 --- a/packages/plugins/semantic-search/src/sdk/documents.ts +++ b/packages/plugins/semantic-search/src/sdk/documents.ts @@ -1,4 +1,4 @@ -import type { Executor, Tool, ToolSchemaManifest } from "@executor-js/sdk/core"; +import type { Executor, Integration, Tool, ToolSchemaManifest } from "@executor-js/sdk/core"; import { Effect } from "effect"; import type { ToolDocumentInput } from "./chunker"; @@ -7,6 +7,7 @@ import { cyrb53 } from "./fingerprint"; const ADDRESS_PREFIX = "tools."; const MAX_AI_SEARCH_FILE_BYTES = 3_500_000; +const TOOL_SEARCH_DOCUMENT_VERSION = "ai-search-tool-document/v2"; const textEncoder = new TextEncoder(); const textDecoder = new TextDecoder(); @@ -171,12 +172,25 @@ const truncateToAiSearchLimit = (document: string): string => { return textDecoder.decode(bytes.subarray(0, end)); }; -export const toolItemKey = (manifest: ToolSchemaManifest): string => +type IntegrationSearchContext = Pick; +type ToolItemKeyManifest = Pick< + ToolSchemaManifest, + "path" | "fingerprintVersion" | "indexFingerprint" | "sourceRevision" +>; + +export const toolItemKey = ( + manifest: ToolItemKeyManifest, + integration?: IntegrationSearchContext, +): string => [ + TOOL_SEARCH_DOCUMENT_VERSION, manifest.path, manifest.fingerprintVersion, manifest.indexFingerprint, manifest.sourceRevision ?? "", + integration?.name ?? "", + integration?.description ?? "", + integration?.displayUrl ?? "", ].join(":"); export interface ToolSearchDocument { @@ -299,11 +313,12 @@ export const collectDocForTool = ( export const collectToolSearchDocument = ( executor: Executor, manifest: ToolSchemaManifest, + integration?: IntegrationSearchContext, ): Effect.Effect => { const path = manifest.path; const name = manifest.name; const description = stripHtml(manifest.description ?? ""); - const fingerprint = toolItemKey(manifest); + const fingerprint = toolItemKey(manifest, integration); return executor.tools.schema(`${ADDRESS_PREFIX}${path}` as Tool["address"]).pipe( Effect.catch(() => Effect.succeed(null)), Effect.map((view) => { @@ -311,6 +326,11 @@ export const collectToolSearchDocument = ( `# ${path}`, `Name: ${name}`, `Integration: ${manifest.integration}`, + integration ? `Integration name: ${stripHtml(integration.name)}` : undefined, + integration?.description + ? `Integration purpose: ${stripHtml(integration.description)}` + : undefined, + integration?.displayUrl ? `Integration URL: ${integration.displayUrl}` : undefined, manifest.connection ? `Connection: ${manifest.connection}` : undefined, manifest.pluginId ? `Plugin: ${manifest.pluginId}` : undefined, description ? `Description: ${description}` : undefined, diff --git a/packages/plugins/semantic-search/src/sdk/tool-search-backend.ts b/packages/plugins/semantic-search/src/sdk/tool-search-backend.ts index a179c8da1f..bf2c08d9eb 100644 --- a/packages/plugins/semantic-search/src/sdk/tool-search-backend.ts +++ b/packages/plugins/semantic-search/src/sdk/tool-search-backend.ts @@ -47,6 +47,8 @@ export interface SemanticSearchRefreshResult { readonly namespace: string; readonly total: number; readonly indexed: number; + /** Items whose upload failed while the rest of the batch continued. */ + readonly failed?: number; readonly skipped: number; readonly removed: number; }