From 0dbeacfb3f00b5bd6fc251aaefca47eb212df766 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sun, 23 Aug 2026 07:00:44 +0530 Subject: [PATCH 1/4] fix(semantic-search): align AI Search catalog retrieval Keep native hybrid retrieval authoritative for tool search while enriching indexed documents with integration context. Treat the local upload ledger as recovery state instead of query visibility, and isolate per-document reindex failures so one tool cannot abort a batch. --- .../plugins/semantic-search/src/api/group.ts | 1 + .../semantic-search/src/sdk/ai-search.test.ts | 268 +++++++++++++++--- .../semantic-search/src/sdk/ai-search.ts | 189 +++++++----- .../semantic-search/src/sdk/collections.ts | 6 + .../semantic-search/src/sdk/documents.ts | 26 +- .../src/sdk/tool-search-backend.ts | 2 + 6 files changed, 383 insertions(+), 109 deletions(-) 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..196024e7d3 100644 --- a/packages/plugins/semantic-search/src/sdk/ai-search.test.ts +++ b/packages/plugins/semantic-search/src/sdk/ai-search.test.ts @@ -10,6 +10,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 +150,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 +173,6 @@ describe("makeAiSearchToolDiscoveryProvider", () => { Effect.gen(function* () { const provider = makeAiSearchToolDiscoveryProvider({ aiSearch: makeAiSearch(), - items: undefined, }); const unfiltered = yield* provider!.searchTools({ @@ -206,7 +205,6 @@ describe("makeAiSearchToolDiscoveryProvider", () => { return makeAiSearch().search(input); }, }, - items: undefined, }); yield* provider!.searchTools({ @@ -250,14 +248,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 +267,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 +383,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 +433,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 +481,14 @@ 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"); }), ); @@ -386,6 +497,7 @@ describe("reindexAiSearch", () => { const removed: string[] = []; const result = yield* reindexAiSearch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed([]), }, @@ -422,6 +534,7 @@ describe("reindexAiSearch", () => { const stored: AiSearchItemRow[] = []; const result = yield* reindexAiSearch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed([ @@ -441,11 +554,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); }, @@ -484,6 +593,7 @@ describe("reindexAiSearch", () => { const result = yield* reindexAiSearch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed(manifests), schema: () => Effect.fail("schema unavailable"), @@ -526,7 +636,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 +650,7 @@ describe("reindexAiSearch", () => { const result = yield* reindexAiSearchBatch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed([manifest]), schema: () => Effect.fail("schema unavailable"), @@ -588,6 +699,7 @@ describe("reindexAiSearch", () => { const result = yield* reindexAiSearchBatch({ executor: { + integrations: { list: () => Effect.succeed([]) }, tools: { manifest: () => Effect.succeed(manifests), schema: () => Effect.fail("schema unavailable"), @@ -645,12 +757,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 +801,82 @@ 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, + pendingDeleteItemId: "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(["stale:item", "previous:item"]); + expect(stored[0]?.itemId).toBe("replacement:" + itemName); + expect(stored[0]?.pendingDeleteItemId).toBeUndefined(); + }), + ); + it.effect("reads status from instance statistics without listing every remote item", () => Effect.gen(function* () { const status = yield* statusAiSearch({ @@ -736,6 +925,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..c79204d232 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, 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(); @@ -137,6 +142,7 @@ const getAiSearchItem = ( const toIndexedItemRow = ( document: ToolSearchDocument, uploaded: AiSearchItemInfo, + pendingDeleteItemId?: string, ): AiSearchItemRow => ({ path: document.path, key: uploaded.key, @@ -149,11 +155,11 @@ const toIndexedItemRow = ( fingerprint: document.fingerprint, status: toStatus(uploaded.status), updatedAt: nowIso(), + ...(pendingDeleteItemId === undefined ? {} : { pendingDeleteItemId }), }); interface UploadedDocument { readonly deleteOnStorageFailure: boolean; - readonly previousItemId?: string; readonly uploadedItemId: string; readonly key: string; readonly row: AiSearchItemRow; @@ -168,11 +174,21 @@ const uploadDocument = ( Effect.gen(function* () { const itemName = toItemName(document); if (remote !== undefined && isReusableRemoteStatus(remote.status)) { + const pendingDeleteItemId = previous?.pendingDeleteItemId; + if (pendingDeleteItemId !== undefined && remote.status === "completed") { + yield* deleteItemBestEffort(aiSearch, pendingDeleteItemId); + return { + deleteOnStorageFailure: false, + uploadedItemId: remote.id, + key: document.path, + row: toIndexedItemRow(document, remote), + }; + } return { deleteOnStorageFailure: false, uploadedItemId: remote.id, key: document.path, - row: toIndexedItemRow(document, remote), + row: toIndexedItemRow(document, remote, pendingDeleteItemId), }; } @@ -188,14 +204,23 @@ const uploadDocument = ( catch: mapUploadError(document), }); + const pendingDeleteItemId = + previous !== undefined && previous.key !== itemName + ? previous.itemId + : previous?.pendingDeleteItemId; + if (pendingDeleteItemId !== undefined && uploaded.status === "completed") { + yield* deleteItemBestEffort(aiSearch, pendingDeleteItemId); + } + 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, + uploaded.status === "completed" ? undefined : pendingDeleteItemId, + ), }; }); @@ -213,9 +238,24 @@ 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.mapError( + (cause) => + new SemanticSearchError({ + message: "Failed to list integration context for AI Search indexing.", + cause, + }), + ), + ), + ] 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 +272,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) && + previous.pendingDeleteItemId === undefined ) { 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 @@ -302,15 +386,6 @@ export const reindexAiSearchBatch = (input: { ), Effect.mapError(mapStorageError("Failed to record AI Search item rows.")), ); - - yield* Effect.forEach( - uploaded, - (entry) => - entry.previousItemId === undefined - ? Effect.void - : deleteItemBestEffort(aiSearch, entry.previousItemId), - { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY, discard: true }, - ); } const removedEntries = shouldRemoveStale @@ -338,6 +413,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 +435,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 +447,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 +457,7 @@ export const reindexAiSearch = (input: { namespace: input.namespace, total, indexed, + ...(failed === 0 ? {} : { failed }), skipped, removed, }; @@ -448,7 +527,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 +545,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 +560,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 +606,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..0061755e91 100644 --- a/packages/plugins/semantic-search/src/sdk/collections.ts +++ b/packages/plugins/semantic-search/src/sdk/collections.ts @@ -18,6 +18,12 @@ export const AiSearchItemRow = Schema.Struct({ status: AiSearchItemStatus, updatedAt: Schema.String, error: Schema.optional(Schema.String), + /** + * The previous provider item stays live until its replacement is completed. + * This prevents an eventual-consistency gap while AI Search processes the + * newly uploaded document. + */ + 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; } From 6aac8589f9ae95d94cd9e178ffd04192cab08ac8 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sun, 23 Aug 2026 09:12:04 +0530 Subject: [PATCH 2/4] fix(semantic-search): harden AI Search reindex recovery (greptile) Persist replacement rows before retiring previous AI Search items, retry deferred cleanup safely, and continue indexing when optional integration context is unavailable. --- .../semantic-search/src/sdk/ai-search.test.ts | 109 +++++++++++++++++- .../semantic-search/src/sdk/ai-search.ts | 89 +++++++++++--- 2 files changed, 180 insertions(+), 18 deletions(-) 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 196024e7d3..582fb64ad3 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 { @@ -492,6 +496,57 @@ describe("reindexAiSearch", () => { }), ); + 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); + }), + ); + it.effect("removes stale rows even when deleting the remote AI Search item fails", () => Effect.gen(function* () { const removed: string[] = []; @@ -575,10 +630,60 @@ 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]?.pendingDeleteItemId).toBe(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(), + 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[] = []; @@ -873,7 +978,7 @@ describe("reindexAiSearch", () => { expect(result).toMatchObject({ indexed: 1, skipped: 0 }); expect(deleted).toEqual(["stale:item", "previous:item"]); expect(stored[0]?.itemId).toBe("replacement:" + itemName); - expect(stored[0]?.pendingDeleteItemId).toBeUndefined(); + expect(stored.at(-1)?.pendingDeleteItemId).toBeUndefined(); }), ); diff --git a/packages/plugins/semantic-search/src/sdk/ai-search.ts b/packages/plugins/semantic-search/src/sdk/ai-search.ts index c79204d232..8b6d777708 100644 --- a/packages/plugins/semantic-search/src/sdk/ai-search.ts +++ b/packages/plugins/semantic-search/src/sdk/ai-search.ts @@ -158,6 +158,11 @@ const toIndexedItemRow = ( ...(pendingDeleteItemId === undefined ? {} : { pendingDeleteItemId }), }); +const withoutPendingDeleteItemId = ({ + pendingDeleteItemId: _pendingDeleteItemId, + ...row +}: AiSearchItemRow): AiSearchItemRow => row; + interface UploadedDocument { readonly deleteOnStorageFailure: boolean; readonly uploadedItemId: string; @@ -176,12 +181,16 @@ const uploadDocument = ( if (remote !== undefined && isReusableRemoteStatus(remote.status)) { const pendingDeleteItemId = previous?.pendingDeleteItemId; if (pendingDeleteItemId !== undefined && remote.status === "completed") { - yield* deleteItemBestEffort(aiSearch, pendingDeleteItemId); + const deletion = yield* deleteItem(aiSearch, pendingDeleteItemId).pipe(Effect.result); return { deleteOnStorageFailure: false, uploadedItemId: remote.id, key: document.path, - row: toIndexedItemRow(document, remote), + row: toIndexedItemRow( + document, + remote, + Result.isSuccess(deletion) ? undefined : pendingDeleteItemId, + ), }; } return { @@ -208,19 +217,12 @@ const uploadDocument = ( previous !== undefined && previous.key !== itemName ? previous.itemId : previous?.pendingDeleteItemId; - if (pendingDeleteItemId !== undefined && uploaded.status === "completed") { - yield* deleteItemBestEffort(aiSearch, pendingDeleteItemId); - } return { deleteOnStorageFailure: true, uploadedItemId: uploaded.id, key: document.path, - row: toIndexedItemRow( - document, - uploaded, - uploaded.status === "completed" ? undefined : pendingDeleteItemId, - ), + row: toIndexedItemRow(document, uploaded, pendingDeleteItemId), }; }); @@ -242,12 +244,16 @@ export const reindexAiSearchBatch = (input: { [ listToolManifests(input.executor, { maxTools: batch.maxTools }), input.executor.integrations.list().pipe( - Effect.mapError( - (cause) => - new SemanticSearchError({ - message: "Failed to list integration context for AI Search indexing.", - cause, - }), + Effect.catch((cause) => + Effect.sync(() => { + console.warn( + JSON.stringify({ + event: "tool_search_index_integration_context_failed", + cause, + }), + ); + return []; + }), ), ), ] as const, @@ -386,6 +392,57 @@ export const reindexAiSearchBatch = (input: { ), Effect.mapError(mapStorageError("Failed to record AI Search item rows.")), ); + + const replacements = uploaded.filter( + (entry) => entry.row.status === "completed" && entry.row.pendingDeleteItemId !== undefined, + ); + if (replacements.length > 0) { + const cleanupResults = yield* Effect.forEach( + replacements, + (entry) => + deleteItem(aiSearch, entry.row.pendingDeleteItemId!).pipe( + Effect.map(() => entry), + Effect.result, + ), + { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY }, + ); + const cleaned = cleanupResults.flatMap((result) => + Result.isSuccess(result) ? [result.success] : [], + ); + const cleanupFailures = cleanupResults.filter((result) => Result.isFailure(result)); + for (const failure of cleanupFailures) { + console.warn( + JSON.stringify({ + event: "tool_search_index_previous_item_delete_failed", + cause: Result.isFailure(failure) ? failure.failure : undefined, + }), + ); + } + if (cleaned.length > 0) { + yield* input.items + .putMany({ + owner: input.owner, + entries: cleaned.map((entry) => ({ + key: entry.key, + data: withoutPendingDeleteItemId(entry.row), + })), + }) + .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 From 1faeab60131a75b874ee89bbdd61071e33f6ed4e Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sun, 23 Aug 2026 09:20:53 +0530 Subject: [PATCH 3/4] fix(semantic-search): defer failed-item replacement cleanup (greptile) Keep errored and outdated AI Search items recoverable until the replacement row is persisted, then retire all superseded remote items. --- .../semantic-search/src/sdk/ai-search.test.ts | 10 ++- .../semantic-search/src/sdk/ai-search.ts | 72 ++++++++++++------- 2 files changed, 56 insertions(+), 26 deletions(-) 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 582fb64ad3..cdf3a91265 100644 --- a/packages/plugins/semantic-search/src/sdk/ai-search.test.ts +++ b/packages/plugins/semantic-search/src/sdk/ai-search.test.ts @@ -661,6 +661,14 @@ describe("reindexAiSearch", () => { ...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); @@ -976,7 +984,7 @@ describe("reindexAiSearch", () => { }); expect(result).toMatchObject({ indexed: 1, skipped: 0 }); - expect(deleted).toEqual(["stale:item", "previous:item"]); + expect(deleted).toEqual(expect.arrayContaining(["stale:item", "previous:item"])); expect(stored[0]?.itemId).toBe("replacement:" + itemName); expect(stored.at(-1)?.pendingDeleteItemId).toBeUndefined(); }), diff --git a/packages/plugins/semantic-search/src/sdk/ai-search.ts b/packages/plugins/semantic-search/src/sdk/ai-search.ts index 8b6d777708..8ca364f68b 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, Result } from "effect"; +import { Effect, Predicate, Result } from "effect"; import { type AiSearchItemRow, aiSearchItems, type AiSearchItemStatus } from "./collections"; import { @@ -168,8 +168,13 @@ interface UploadedDocument { readonly uploadedItemId: string; readonly key: string; readonly row: AiSearchItemRow; + readonly deferredDeleteItemIds: readonly string[]; } +const uniqueItemIds = (ids: readonly (string | undefined)[]): string[] => [ + ...new Set(ids.filter(Predicate.isNotUndefined)), +]; + const uploadDocument = ( aiSearch: Pick, document: ToolSearchDocument, @@ -191,6 +196,7 @@ const uploadDocument = ( remote, Result.isSuccess(deletion) ? undefined : pendingDeleteItemId, ), + deferredDeleteItemIds: [], }; } return { @@ -198,13 +204,10 @@ const uploadDocument = ( uploadedItemId: remote.id, key: document.path, row: toIndexedItemRow(document, remote, pendingDeleteItemId), + deferredDeleteItemIds: [], }; } - if (remote !== undefined) { - yield* deleteItemBestEffort(aiSearch, remote.id); - } - const uploaded = yield* Effect.tryPromise({ try: () => aiSearch.items.upload(itemName, document.content, { @@ -213,16 +216,21 @@ const uploadDocument = ( catch: mapUploadError(document), }); - const pendingDeleteItemId = - previous !== undefined && previous.key !== itemName - ? previous.itemId - : previous?.pendingDeleteItemId; + const deferredDeleteItemIds = uniqueItemIds( + remote !== undefined && !isReusableRemoteStatus(remote.status) + ? [remote.id, previous?.pendingDeleteItemId] + : remote === undefined && previous !== undefined && previous.key !== itemName + ? [previous.itemId, previous.pendingDeleteItemId] + : [previous?.pendingDeleteItemId], + ); + const pendingDeleteItemId = deferredDeleteItemIds[0]; return { deleteOnStorageFailure: true, uploadedItemId: uploaded.id, key: document.path, row: toIndexedItemRow(document, uploaded, pendingDeleteItemId), + deferredDeleteItemIds: toStatus(uploaded.status) === "completed" ? deferredDeleteItemIds : [], }; }); @@ -394,29 +402,43 @@ export const reindexAiSearchBatch = (input: { ); const replacements = uploaded.filter( - (entry) => entry.row.status === "completed" && entry.row.pendingDeleteItemId !== undefined, + (entry) => + entry.row.status === "completed" && + (entry.deferredDeleteItemIds.length > 0 || entry.row.pendingDeleteItemId !== undefined), ); if (replacements.length > 0) { const cleanupResults = yield* Effect.forEach( replacements, - (entry) => - deleteItem(aiSearch, entry.row.pendingDeleteItemId!).pipe( - Effect.map(() => entry), - Effect.result, - ), + (entry) => { + const itemIds = + entry.deferredDeleteItemIds.length > 0 + ? entry.deferredDeleteItemIds + : entry.row.pendingDeleteItemId === undefined + ? [] + : [entry.row.pendingDeleteItemId]; + 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 }, ); - const cleaned = cleanupResults.flatMap((result) => - Result.isSuccess(result) ? [result.success] : [], + const cleaned = cleanupResults.flatMap(({ entry, results }) => + results.every((result) => Result.isSuccess(result)) ? [entry] : [], ); - const cleanupFailures = cleanupResults.filter((result) => Result.isFailure(result)); - for (const failure of cleanupFailures) { - console.warn( - JSON.stringify({ - event: "tool_search_index_previous_item_delete_failed", - cause: Result.isFailure(failure) ? failure.failure : undefined, - }), - ); + 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 (cleaned.length > 0) { yield* input.items From 4887b1024f2ea7cb970a7c3a871d4f82cd410169 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sun, 23 Aug 2026 09:30:10 +0530 Subject: [PATCH 4/4] fix(semantic-search): retain all deferred deletions (greptile) --- .../semantic-search/src/sdk/ai-search.test.ts | 85 +++++++++++++++++- .../semantic-search/src/sdk/ai-search.ts | 87 +++++++++++-------- .../semantic-search/src/sdk/collections.ts | 11 ++- 3 files changed, 142 insertions(+), 41 deletions(-) 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 cdf3a91265..d9013247e1 100644 --- a/packages/plugins/semantic-search/src/sdk/ai-search.test.ts +++ b/packages/plugins/semantic-search/src/sdk/ai-search.test.ts @@ -630,7 +630,7 @@ 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]?.pendingDeleteItemId).toBe(githubRow.data.itemId); + expect(stored[0]?.pendingDeleteItemIds).toEqual([githubRow.data.itemId]); expect(deleted).toEqual(["item:github.repos.create.md"]); }), ); @@ -935,7 +935,7 @@ describe("reindexAiSearch", () => { key: itemName, itemId: "stale:item", fingerprint, - pendingDeleteItemId: "previous:item", + pendingDeleteItemIds: ["previous:item"], }, }; @@ -986,7 +986,86 @@ describe("reindexAiSearch", () => { 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)?.pendingDeleteItemId).toBeUndefined(); + 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"]); }), ); diff --git a/packages/plugins/semantic-search/src/sdk/ai-search.ts b/packages/plugins/semantic-search/src/sdk/ai-search.ts index 8ca364f68b..3c152dc080 100644 --- a/packages/plugins/semantic-search/src/sdk/ai-search.ts +++ b/packages/plugins/semantic-search/src/sdk/ai-search.ts @@ -68,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 => ({ @@ -142,7 +149,7 @@ const getAiSearchItem = ( const toIndexedItemRow = ( document: ToolSearchDocument, uploaded: AiSearchItemInfo, - pendingDeleteItemId?: string, + pendingDeleteItemIds?: readonly string[], ): AiSearchItemRow => ({ path: document.path, key: uploaded.key, @@ -155,13 +162,25 @@ const toIndexedItemRow = ( fingerprint: document.fingerprint, status: toStatus(uploaded.status), updatedAt: nowIso(), - ...(pendingDeleteItemId === undefined ? {} : { pendingDeleteItemId }), + ...(pendingDeleteItemIds === undefined || pendingDeleteItemIds.length === 0 + ? {} + : { pendingDeleteItemIds: uniqueItemIds(pendingDeleteItemIds) }), }); -const withoutPendingDeleteItemId = ({ - pendingDeleteItemId: _pendingDeleteItemId, - ...row -}: AiSearchItemRow): AiSearchItemRow => row; +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; @@ -171,10 +190,6 @@ interface UploadedDocument { readonly deferredDeleteItemIds: readonly string[]; } -const uniqueItemIds = (ids: readonly (string | undefined)[]): string[] => [ - ...new Set(ids.filter(Predicate.isNotUndefined)), -]; - const uploadDocument = ( aiSearch: Pick, document: ToolSearchDocument, @@ -183,19 +198,23 @@ const uploadDocument = ( ): Effect.Effect => Effect.gen(function* () { const itemName = toItemName(document); + const previousPendingDeleteItemIds = + previous === undefined ? [] : pendingDeleteItemIdsForRow(previous); if (remote !== undefined && isReusableRemoteStatus(remote.status)) { - const pendingDeleteItemId = previous?.pendingDeleteItemId; - if (pendingDeleteItemId !== undefined && remote.status === "completed") { - const deletion = yield* deleteItem(aiSearch, pendingDeleteItemId).pipe(Effect.result); + 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, - Result.isSuccess(deletion) ? undefined : pendingDeleteItemId, - ), + row: toIndexedItemRow(document, remote, failedPendingDeleteItemIds), deferredDeleteItemIds: [], }; } @@ -203,7 +222,7 @@ const uploadDocument = ( deleteOnStorageFailure: false, uploadedItemId: remote.id, key: document.path, - row: toIndexedItemRow(document, remote, pendingDeleteItemId), + row: toIndexedItemRow(document, remote, previousPendingDeleteItemIds), deferredDeleteItemIds: [], }; } @@ -218,18 +237,17 @@ const uploadDocument = ( const deferredDeleteItemIds = uniqueItemIds( remote !== undefined && !isReusableRemoteStatus(remote.status) - ? [remote.id, previous?.pendingDeleteItemId] + ? [remote.id, ...previousPendingDeleteItemIds] : remote === undefined && previous !== undefined && previous.key !== itemName - ? [previous.itemId, previous.pendingDeleteItemId] - : [previous?.pendingDeleteItemId], + ? [previous.itemId, ...previousPendingDeleteItemIds] + : previousPendingDeleteItemIds, ); - const pendingDeleteItemId = deferredDeleteItemIds[0]; return { deleteOnStorageFailure: true, uploadedItemId: uploaded.id, key: document.path, - row: toIndexedItemRow(document, uploaded, pendingDeleteItemId), + row: toIndexedItemRow(document, uploaded, deferredDeleteItemIds), deferredDeleteItemIds: toStatus(uploaded.status) === "completed" ? deferredDeleteItemIds : [], }; }); @@ -338,7 +356,7 @@ export const reindexAiSearchBatch = (input: { previous?.fingerprint === fingerprint && remote !== undefined && isReusableRemoteStatus(remote.status) && - previous.pendingDeleteItemId === undefined + pendingDeleteItemIdsForRow(previous).length === 0 ) { skipped += 1; continue; @@ -404,7 +422,8 @@ export const reindexAiSearchBatch = (input: { const replacements = uploaded.filter( (entry) => entry.row.status === "completed" && - (entry.deferredDeleteItemIds.length > 0 || entry.row.pendingDeleteItemId !== undefined), + (entry.deferredDeleteItemIds.length > 0 || + pendingDeleteItemIdsForRow(entry.row).length > 0), ); if (replacements.length > 0) { const cleanupResults = yield* Effect.forEach( @@ -413,9 +432,7 @@ export const reindexAiSearchBatch = (input: { const itemIds = entry.deferredDeleteItemIds.length > 0 ? entry.deferredDeleteItemIds - : entry.row.pendingDeleteItemId === undefined - ? [] - : [entry.row.pendingDeleteItemId]; + : pendingDeleteItemIdsForRow(entry.row); return Effect.forEach( itemIds, (itemId) => deleteItem(aiSearch, itemId).pipe(Effect.result), @@ -424,9 +441,6 @@ export const reindexAiSearchBatch = (input: { }, { concurrency: AI_SEARCH_UPLOAD_CONCURRENCY }, ); - const cleaned = cleanupResults.flatMap(({ entry, results }) => - results.every((result) => Result.isSuccess(result)) ? [entry] : [], - ); for (const { itemIds, results } of cleanupResults) { for (const [index, result] of results.entries()) { if (Result.isFailure(result)) { @@ -440,13 +454,16 @@ export const reindexAiSearchBatch = (input: { } } } - if (cleaned.length > 0) { + if (cleanupResults.length > 0) { yield* input.items .putMany({ owner: input.owner, - entries: cleaned.map((entry) => ({ + entries: cleanupResults.map(({ entry, itemIds, results }) => ({ key: entry.key, - data: withoutPendingDeleteItemId(entry.row), + data: withPendingDeleteItemIds( + entry.row, + itemIds.filter((_, index) => Result.isFailure(results[index])), + ), })), }) .pipe( diff --git a/packages/plugins/semantic-search/src/sdk/collections.ts b/packages/plugins/semantic-search/src/sdk/collections.ts index 0061755e91..45625c684f 100644 --- a/packages/plugins/semantic-search/src/sdk/collections.ts +++ b/packages/plugins/semantic-search/src/sdk/collections.ts @@ -19,9 +19,14 @@ export const AiSearchItemRow = Schema.Struct({ updatedAt: Schema.String, error: Schema.optional(Schema.String), /** - * The previous provider item stays live until its replacement is completed. - * This prevents an eventual-consistency gap while AI Search processes the - * newly uploaded document. + * 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), });