From b71073d6cb9b49dc34e791ca49beca9ea8c51356 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Sun, 5 Jul 2026 22:01:25 +0800 Subject: [PATCH] fix: harden chat assets on demand --- .../sources/[sourceId]/chunks/route.test.ts | 2 - src/components/workspace-citation-focus.ts | 3 + .../workspace-selected-chunks.test.ts | 52 +++ src/components/workspace-selected-chunks.ts | 15 +- src/components/workspace-shell.tsx | 1 + src/components/workspace-source-state.test.ts | 21 ++ src/components/workspace-source-state.ts | 23 ++ src/components/workspace-source-workflow.ts | 6 + src/domains/chat/contracts.ts | 4 +- src/domains/chat/index.test.ts | 60 ++-- src/domains/chat/index.ts | 4 +- .../chat/media-asset-hardening.test.ts | 32 +- src/domains/chat/media-asset-hardening.ts | 123 +++++--- src/domains/chat/media-assets.test.ts | 50 ++- src/domains/chat/media-assets.ts | 295 +++++++----------- src/domains/chat/page-citation-assets.test.ts | 33 +- src/domains/chat/page-citation-assets.ts | 56 +--- src/domains/chat/route-answer.ts | 148 +++++++-- src/domains/chat/route-service.test.ts | 103 +++--- src/domains/chat/service.ts | 4 +- src/domains/chunks/index.ts | 4 +- src/domains/chunks/read.test.ts | 4 +- src/domains/chunks/read.ts | 60 +--- .../sources/source-reconcile-workflow.test.ts | 64 +++- .../sources/source-reconcile-workflow.ts | 53 +++- 25 files changed, 707 insertions(+), 513 deletions(-) diff --git a/src/app/api/sources/[sourceId]/chunks/route.test.ts b/src/app/api/sources/[sourceId]/chunks/route.test.ts index 85fb1bd..2f0851c 100644 --- a/src/app/api/sources/[sourceId]/chunks/route.test.ts +++ b/src/app/api/sources/[sourceId]/chunks/route.test.ts @@ -498,7 +498,6 @@ describe("GET /api/sources/[sourceId]/chunks", () => { revisionKey: "job_1", page: 1, pageSize: 1, - assetUrlPolicy: "durable", }) }) @@ -660,7 +659,6 @@ describe("GET /api/sources/[sourceId]/chunks", () => { revisionKey: "job_result_1", page: 1, pageSize: 1, - assetUrlPolicy: "durable", }) }) }) diff --git a/src/components/workspace-citation-focus.ts b/src/components/workspace-citation-focus.ts index 3485dd2..357cd4e 100644 --- a/src/components/workspace-citation-focus.ts +++ b/src/components/workspace-citation-focus.ts @@ -21,6 +21,7 @@ type PrefetchedChunksUpdater = ( type WorkspaceCitationFocusInput = { readonly fetchChunks: (sourceId: string) => Promise readonly initialPrefetchedChunksBySourceId?: PrefetchedChunksBySourceId + readonly onRemoteSourceChunksLoaded?: (sourceId: string) => void readonly onSelectSource: (sourceId: string | null) => void readonly selectedSourceId: string | null readonly sources: readonly SourceView[] @@ -51,6 +52,7 @@ type WorkspaceCitationFocus = { export function useWorkspaceCitationFocus({ fetchChunks, initialPrefetchedChunksBySourceId = {}, + onRemoteSourceChunksLoaded, onSelectSource, selectedSourceId, sources, @@ -88,6 +90,7 @@ export function useWorkspaceCitationFocus({ selectedSourceId, sources, prefetchedChunksBySourceId, + onRemoteSourceChunksLoaded, }) const requestChunkFocus = useCallback( diff --git a/src/components/workspace-selected-chunks.test.ts b/src/components/workspace-selected-chunks.test.ts index fa08df5..45db25a 100644 --- a/src/components/workspace-selected-chunks.test.ts +++ b/src/components/workspace-selected-chunks.test.ts @@ -161,6 +161,58 @@ describe("useWorkspaceSelectedChunks", () => { expect(result.current.selectedChunks).toEqual([]); }); + it("requests a source refresh after loading an unlocalized remote source", async () => { + const onRemoteSourceChunksLoaded = vi.fn(); + const remoteSource: SourceView = { + ...readySource, + id: "knowhere-doc:default:doc_remote", + kind: "remote", + documentId: "doc_remote", + excludedFromQuery: false, + }; + fetchChunkPageMock.mockResolvedValue({ + chunks: [ + { + chunkId: "chunk_1", + type: "text", + content: "Remote content", + sourceTitle: "Remote.pdf", + }, + ], + pagination: { + page: 1, + pageSize: 50, + total: 1, + totalPages: 1, + }, + }); + + const { rerender } = renderHook( + (input: { + readonly onRemoteSourceChunksLoaded: (sourceId: string) => void; + }) => + useWorkspaceSelectedChunks({ + selectedSourceId: "knowhere-doc:default:doc_remote", + sources: [remoteSource], + prefetchedChunksBySourceId: {}, + onRemoteSourceChunksLoaded: input.onRemoteSourceChunksLoaded, + }), + { + initialProps: { onRemoteSourceChunksLoaded }, + wrapper: createSWRWrapper, + }, + ); + + await waitFor(() => + expect(onRemoteSourceChunksLoaded).toHaveBeenCalledWith( + "knowhere-doc:default:doc_remote", + ), + ); + + rerender({ onRemoteSourceChunksLoaded }); + expect(onRemoteSourceChunksLoaded).toHaveBeenCalledTimes(1); + }); + it("returns an empty chunk list when no source is selected", () => { const { result } = renderHook( () => diff --git a/src/components/workspace-selected-chunks.ts b/src/components/workspace-selected-chunks.ts index 6d79175..f00ac68 100644 --- a/src/components/workspace-selected-chunks.ts +++ b/src/components/workspace-selected-chunks.ts @@ -1,6 +1,6 @@ "use client" -import { useMemo } from "react" +import { useEffect, useMemo, useRef } from "react" import useSWRInfinite from "swr/infinite" import { workspaceClient } from "@/domains/workspace/client" @@ -17,6 +17,7 @@ type WorkspaceSelectedChunksInput = { readonly selectedSourceId: string | null readonly sources: readonly SourceView[] readonly prefetchedChunksBySourceId: Readonly> + readonly onRemoteSourceChunksLoaded?: (sourceId: string) => void } type WorkspaceSelectedChunks = { @@ -33,8 +34,10 @@ export function useWorkspaceSelectedChunks({ selectedSourceId, sources, prefetchedChunksBySourceId, + onRemoteSourceChunksLoaded, }: WorkspaceSelectedChunksInput): WorkspaceSelectedChunks { const selectedSource = sources.find((source) => source.id === selectedSourceId) + const remoteSourceRefreshRequestedIdsRef = useRef>(new Set()) const prefetchedSelectedChunks = selectedSourceId ? prefetchedChunksBySourceId[selectedSourceId] : undefined @@ -100,6 +103,16 @@ export function useWorkspaceSelectedChunks({ !selectedChunkPages && isChunksLoading) + useEffect(() => { + const sourceId = selectedSource?.id + if (!sourceId || selectedSource.kind !== "remote") return + if (!selectedChunkPages || selectedChunkPages.length === 0) return + if (remoteSourceRefreshRequestedIdsRef.current.has(sourceId)) return + + remoteSourceRefreshRequestedIdsRef.current.add(sourceId) + onRemoteSourceChunksLoaded?.(sourceId) + }, [onRemoteSourceChunksLoaded, selectedChunkPages, selectedSource]) + function handleLoadMoreChunks(): void { if (!hasMoreSelectedChunks || isSelectedChunksLoadingMore) return void setSelectedChunkPageCount(selectedChunkPageCount + 1) diff --git a/src/components/workspace-shell.tsx b/src/components/workspace-shell.tsx index 4dd8b9c..64663a0 100644 --- a/src/components/workspace-shell.tsx +++ b/src/components/workspace-shell.tsx @@ -114,6 +114,7 @@ function WorkspaceShellContent({ fetchChunks: workspaceClient.fetchChunks, initialPrefetchedChunksBySourceId: initialPrefetchedChunksBySourceId ?? undefined, + onRemoteSourceChunksLoaded: sourceWorkflow.handleSourcesRefresh, onSelectSource: handleCitationSourceSelected, selectedSourceId: sourceWorkflow.selectedSourceId, sources: sourceWorkflow.sources, diff --git a/src/components/workspace-source-state.test.ts b/src/components/workspace-source-state.test.ts index f1069d5..19e6cb1 100644 --- a/src/components/workspace-source-state.test.ts +++ b/src/components/workspace-source-state.test.ts @@ -83,6 +83,27 @@ describe("workspaceSourceState", () => { ).toBe("source_target"); }); + it("keeps a localized remote document selected after source refresh", () => { + const sources: readonly SourceView[] = [ + { + id: "source_localized", + kind: "workspace", + title: "remote.pdf", + status: "ready", + mimeType: "application/pdf", + documentId: "doc_remote", + excludedFromQuery: false, + }, + ]; + + expect( + workspaceSourceState.getResolvedSelectedSourceId( + sources, + "knowhere-doc:default:doc_remote", + ), + ).toBe("source_localized"); + }); + it("applies source query exclusions without mutating the source list", () => { const sources: readonly SourceView[] = [ { diff --git a/src/components/workspace-source-state.ts b/src/components/workspace-source-state.ts index 676e020..0373983 100644 --- a/src/components/workspace-source-state.ts +++ b/src/components/workspace-source-state.ts @@ -74,6 +74,15 @@ function getResolvedSelectedSourceId( return selectedSource.id } + const selectedDocumentId = getRemoteSourceDocumentId(selectedSourceId) + if (selectedDocumentId) { + const localizedSource = sources.find( + (source) => + source.documentId === selectedDocumentId && isReadySource(source), + ) + if (localizedSource) return localizedSource.id + } + return getFirstReadySourceId(sources) } @@ -134,6 +143,20 @@ function removeRecordKey( return remaining } +function getRemoteSourceDocumentId(sourceId: string | null): string | null { + if (!sourceId) return null + + const parts = sourceId.split(":") + if (parts.length !== 3 || parts[0] !== "knowhere-doc") return null + + try { + const documentId = decodeURIComponent(parts[2] ?? "") + return documentId.length > 0 ? documentId : null + } catch { + return null + } +} + export const workspaceSourceState: WorkspaceSourceStateModule = { getFirstReadySourceId, getInitialSelectedSourceId, diff --git a/src/components/workspace-source-workflow.ts b/src/components/workspace-source-workflow.ts index 9b9f1f8..7ad2871 100644 --- a/src/components/workspace-source-workflow.ts +++ b/src/components/workspace-source-workflow.ts @@ -22,6 +22,7 @@ type WorkspaceSourceWorkflow = { readonly handleRetrySource: (sourceId: string) => Promise readonly handleOfficialLibrarySourceAdd: (demoSourceId: string) => Promise readonly handleSelectedSourceChange: (sourceId: string | null) => void + readonly handleSourcesRefresh: () => void readonly handleSourcesMaterialized: ( demoSourceIds: readonly string[], materializedSources: readonly SourceView[], @@ -153,6 +154,10 @@ export function useWorkspaceSourceWorkflow({ setSelectedSourceId(sourceId) } + function handleSourcesRefresh(): void { + void mutateSources() + } + async function handleArchiveSource(sourceId: string): Promise { setArchivingSourceIds((current) => workspaceSourceState.addPendingId(current, sourceId), @@ -237,6 +242,7 @@ export function useWorkspaceSourceWorkflow({ handleRetrySource, handleOfficialLibrarySourceAdd, handleSelectedSourceChange, + handleSourcesRefresh, handleSourcesMaterialized, handleSourceUploaded, handleToggleIncluded, diff --git a/src/domains/chat/contracts.ts b/src/domains/chat/contracts.ts index 4c8d224..4752b86 100644 --- a/src/domains/chat/contracts.ts +++ b/src/domains/chat/contracts.ts @@ -10,7 +10,7 @@ import type { ChatCitationView, } from "@/domains/chat/types" import type { HardenMediaAssetUrls } from "./media-asset-hardening" -import type { LoadSourceAssetUrls } from "./media-assets" +import type { HardenChatAssetUrl } from "./media-assets" export type RetrievalClient = { query(params: RetrievalQueryParams): Promise @@ -67,7 +67,7 @@ export type AnswerQuestionInput = { excludedSourceIds: readonly string[] retrieval: RetrievalClient generateAnswer: GenerateAnswer - loadSourceAssetUrls?: LoadSourceAssetUrls + hardenChatAssetUrl?: HardenChatAssetUrl hardenMediaAssetUrls?: HardenMediaAssetUrls messages: readonly ChatHistoryMessage[] } diff --git a/src/domains/chat/index.test.ts b/src/domains/chat/index.test.ts index 59899b0..0be6b78 100644 --- a/src/domains/chat/index.test.ts +++ b/src/domains/chat/index.test.ts @@ -452,10 +452,11 @@ describe("answerQuestionWithRetrieval", () => { }); return makeHarnessRunResult(`Use this launch photo. ${upstreamAssetUrl}`); }); - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "images/image-9-Night Rocket Launch.jpg": + const hardenChatAssetUrl = vi + .fn() + .mockResolvedValue( "https://blob.example/images/image-9-Night%20Rocket%20Launch.jpg", - }); + ); const answer = await Effect.runPromise( answerQuestionWithRetrieval({ @@ -471,14 +472,16 @@ describe("answerQuestionWithRetrieval", () => { excludedSourceIds: [], retrieval, generateAnswer, - loadSourceAssetUrls, + hardenChatAssetUrl, messages: [], }), ); - expect(loadSourceAssetUrls).toHaveBeenCalledWith( - expect.objectContaining({ id: "source_spacex" }), - ); + expect(hardenChatAssetUrl).toHaveBeenCalledWith({ + source: expect.objectContaining({ id: "source_spacex" }), + sourcePath: "images/image-9-Night Rocket Launch.jpg", + assetUrl: upstreamAssetUrl, + }); expect(retrieval.query).toHaveBeenCalledWith({ namespace: "notebook-workspace", query: "SpaceX rocket photos", @@ -731,6 +734,7 @@ describe("answerQuestionWithRetrieval", () => { ...(artifacts ? { artifacts: [...artifacts] } : {}), }), ); + const hardenChatAssetUrl = vi.fn().mockResolvedValue(storedPageAssetUrl); const answer = await Effect.runPromise( answerQuestionWithRetrieval({ @@ -747,13 +751,17 @@ describe("answerQuestionWithRetrieval", () => { retrieval, generateAnswer, hardenMediaAssetUrls, - loadSourceAssetUrls: vi.fn(async () => ({ - "page_citation_assets/page-4.png": storedPageAssetUrl, - })), + hardenChatAssetUrl, messages: [], }), ); + expect(hardenChatAssetUrl).toHaveBeenCalledWith({ + source: expect.objectContaining({ id: "source_pages" }), + sourcePath: "page_citation_assets/page-4.png", + assetUrl: rawPageAssetUrl, + contentType: "image/png", + }); expect(hardenMediaAssetUrls).toHaveBeenCalledWith({ results: [ expect.objectContaining({ @@ -844,6 +852,7 @@ describe("answerQuestionWithRetrieval", () => { ...(artifacts ? { artifacts: [...artifacts] } : {}), }), ); + const hardenChatAssetUrl = vi.fn().mockResolvedValue(storedPageAssetUrl); const answer = await Effect.runPromise( answerQuestionWithRetrieval({ @@ -860,13 +869,17 @@ describe("answerQuestionWithRetrieval", () => { retrieval, generateAnswer, hardenMediaAssetUrls, - loadSourceAssetUrls: vi.fn(async () => ({ - "page_citation_assets/page-6.png": storedPageAssetUrl, - })), + hardenChatAssetUrl, messages: [], }), ); + expect(hardenChatAssetUrl).toHaveBeenCalledWith({ + source: expect.objectContaining({ id: "source_pages" }), + sourcePath: "page_citation_assets/page-6.png", + assetUrl: rawPageAssetUrl, + contentType: "image/png", + }); expect(hardenMediaAssetUrls).toHaveBeenCalledWith({ results: [ expect.objectContaining({ @@ -1406,7 +1419,7 @@ describe("answerQuestionWithRetrieval", () => { ); }); - it("turns retrieved evidence image filenames into image citations", async () => { + it("does not turn evidence-only image filenames into image citations", async () => { const result = makeRetrievalResult({ content: "This section contains identity proof attachments.", source: { @@ -1434,12 +1447,7 @@ describe("answerQuestionWithRetrieval", () => { }); return makeHarnessRunResult("这里是相关身份证明图片。"); }); - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "images/image-6-中华人民共和国居民身份证.jpg": - "https://blob.example/images/image-6-id-front.jpg", - "images/image-7-中国居民身份证.jpg": - "https://blob.example/images/image-7-id-back.jpg", - }); + const hardenChatAssetUrl = vi.fn().mockResolvedValue(null); const sources = [ makeSource({ id: "source_identity", @@ -1456,7 +1464,7 @@ describe("answerQuestionWithRetrieval", () => { excludedSourceIds: [], retrieval, generateAnswer, - loadSourceAssetUrls, + hardenChatAssetUrl, messages: [], }), ); @@ -1478,14 +1486,8 @@ describe("answerQuestionWithRetrieval", () => { const imageCitations = answer.citations.filter( (citation) => citation.assetUrl, ) - expect(imageCitations.map((citation) => citation.assetUrl)).toEqual([ - "https://blob.example/images/image-6-id-front.jpg", - "https://blob.example/images/image-7-id-back.jpg", - ]); - expect(imageCitations.map((citation) => citation.chunkType)).toEqual([ - "image", - "image", - ]); + expect(imageCitations).toEqual([]); + expect(hardenChatAssetUrl).not.toHaveBeenCalled(); }); it("returns the agent answer without citations when retrieval has no results", async () => { diff --git a/src/domains/chat/index.ts b/src/domains/chat/index.ts index bbf95b9..dc10453 100644 --- a/src/domains/chat/index.ts +++ b/src/domains/chat/index.ts @@ -242,7 +242,7 @@ export const answerQuestionWithRetrieval = ( enrichRetrievalResultsWithAssetUrls({ results: useNotebookSourceTitles(rawResults, input.sources), sources: input.sources, - loadSourceAssetUrls: input.loadSourceAssetUrls, + hardenChatAssetUrl: input.hardenChatAssetUrl, evidenceText: formatRetrievalEvidenceText(retrievalResponses), }), ) @@ -250,7 +250,7 @@ export const answerQuestionWithRetrieval = ( enrichRetrievalResultsWithPageCitationAssetUrls({ results: enrichedResults, sources: input.sources, - loadSourceAssetUrls: input.loadSourceAssetUrls, + hardenChatAssetUrl: input.hardenChatAssetUrl, }), ) const artifacts = toChatArtifactViewsFromHarness(generatedAnswer, input.sources) diff --git a/src/domains/chat/media-asset-hardening.test.ts b/src/domains/chat/media-asset-hardening.test.ts index d3c87b2..7c106c9 100644 --- a/src/domains/chat/media-asset-hardening.test.ts +++ b/src/domains/chat/media-asset-hardening.test.ts @@ -22,10 +22,10 @@ afterEach(() => { }) describe("hardenChatMediaAssetUrls", () => { - it("keeps an already Notebook-owned asset URL without loading the asset map", async () => { + it("keeps an already Notebook-owned asset URL without calling the hardener", async () => { const ownedUrl = "https://fake.public.blob.vercel-storage.com/workspaces/workspace_1/parsed-documents/doc_1/rev_1/assets/images/a.png" - const loadSourceAssetUrls = vi.fn(async () => ({})) + const hardenChatAssetUrl = vi.fn(async () => null) const result = await hardenChatMediaAssetUrls({ workspaceId: "workspace_1", @@ -41,11 +41,11 @@ describe("hardenChatMediaAssetUrls", () => { }, }), ], - loadSourceAssetUrls, + hardenChatAssetUrl, }) expect(result.results[0]?.assetUrl).toBe(ownedUrl) - expect(loadSourceAssetUrls).not.toHaveBeenCalled() + expect(hardenChatAssetUrl).not.toHaveBeenCalled() }) it("resolves a raw asset URL to the durable parsed asset URL", async () => { @@ -53,9 +53,7 @@ describe("hardenChatMediaAssetUrls", () => { "https://knowhere-storage.example/results/job_1/images/id-front.jpg?AWSAccessKeyId=test" const durableUrl = "https://fake.public.blob.vercel-storage.com/workspaces/workspace_1/parsed-documents/doc_identity/rev_1/assets/images/id-front.jpg" - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "images/id-front.jpg": durableUrl, - }) + const hardenChatAssetUrl = vi.fn().mockResolvedValue(durableUrl) const result = await hardenChatMediaAssetUrls({ workspaceId: "workspace_1", @@ -76,19 +74,21 @@ describe("hardenChatMediaAssetUrls", () => { }, }), ], - loadSourceAssetUrls, + hardenChatAssetUrl, }) - expect(loadSourceAssetUrls).toHaveBeenCalledWith( - expect.objectContaining({ id: "source_identity" }), - ) + expect(hardenChatAssetUrl).toHaveBeenCalledWith({ + source: expect.objectContaining({ id: "source_identity" }), + sourcePath: "images/id-front.jpg", + assetUrl: rawAssetUrl, + }) expect(result.results[0]?.assetUrl).toBe(durableUrl) }) it("omits an asset URL that cannot be resolved to a durable URL", async () => { const rawAssetUrl = "https://knowhere-storage.example/results/job_1/tables/table-1.html?AWSAccessKeyId=test" - const loadSourceAssetUrls = vi.fn().mockResolvedValue({}) + const hardenChatAssetUrl = vi.fn().mockResolvedValue(null) const result = await hardenChatMediaAssetUrls({ workspaceId: "workspace_1", @@ -106,7 +106,7 @@ describe("hardenChatMediaAssetUrls", () => { }, }), ], - loadSourceAssetUrls, + hardenChatAssetUrl, }) expect(result.results[0]?.assetUrl).toBeUndefined() @@ -117,9 +117,7 @@ describe("hardenChatMediaAssetUrls", () => { "https://knowhere-storage.example/results/job_1/images/front.jpg?AWSAccessKeyId=test" const durableUrl = "https://fake.public.blob.vercel-storage.com/workspaces/workspace_1/parsed-documents/doc_identity/rev_1/assets/images/front.jpg" - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "images/front.jpg": durableUrl, - }) + const hardenChatAssetUrl = vi.fn().mockResolvedValue(durableUrl) const result = await hardenChatMediaAssetUrls({ workspaceId: "workspace_1", @@ -148,7 +146,7 @@ describe("hardenChatMediaAssetUrls", () => { }, }, ], - loadSourceAssetUrls, + hardenChatAssetUrl, }) const [artifact] = result.artifacts ?? [] diff --git a/src/domains/chat/media-asset-hardening.ts b/src/domains/chat/media-asset-hardening.ts index f768515..ba88bb1 100644 --- a/src/domains/chat/media-asset-hardening.ts +++ b/src/domains/chat/media-asset-hardening.ts @@ -6,8 +6,7 @@ import type { } from "@/domains/chat/types" import type { Source } from "@/infrastructure/db/schema" import { logger } from "@/lib/logger" -import type { LoadSourceAssetUrls } from "./media-assets" -import { resolveAssetUrlFromReferenceText } from "./media-assets" +import type { HardenChatAssetUrl } from "./media-assets" export type HardenableRetrievalResult = RetrievalResult & { readonly pageCitationAssetUrl?: string @@ -31,7 +30,7 @@ export type HardenChatMediaAssetUrlsForWorkspaceInput = HardenMediaAssetUrlsInput & { readonly workspaceId: string readonly sources: readonly Source[] - readonly loadSourceAssetUrls: LoadSourceAssetUrls + readonly hardenChatAssetUrl: HardenChatAssetUrl } type AssetReferenceSource = ChatCitationView["source"] @@ -44,11 +43,7 @@ type AssetUrlReference = { type HardeningContext = { readonly sourcesByDocumentId: ReadonlyMap - readonly loadSourceAssetUrls: LoadSourceAssetUrls - readonly assetUrlsBySourceId: Map< - string, - Promise>> - > + readonly hardenChatAssetUrl: HardenChatAssetUrl } const parsedResultDirectoryName = "parsed-result" @@ -56,9 +51,9 @@ const chatAssetsDirectoryName = "chat-assets" /** * Resolve chat citation/media asset URLs to durable Notebook Blob URLs. The - * single hardening path is the SDK's `assetUrlPolicy: "durable"` read that - * `loadSourceAssetUrls` performs; here we only map a retrieval result's - * reference text to the durable URL that read produced. + * hardening path is one asset at a time: derive the source path already present + * on the returned chat/citation asset, then let the route-level hardener check + * Blob and write only that missing asset. * * An asset that is already Notebook-owned is kept as-is. An asset that cannot * be resolved to a durable URL is omitted rather than exposing a presigned @@ -68,12 +63,11 @@ export async function hardenChatMediaAssetUrls({ results, artifacts, sources, - loadSourceAssetUrls, + hardenChatAssetUrl, }: HardenChatMediaAssetUrlsForWorkspaceInput): Promise { const context: HardeningContext = { sourcesByDocumentId: createSourcesByDocumentId(sources), - loadSourceAssetUrls, - assetUrlsBySourceId: new Map(), + hardenChatAssetUrl, } const hardenedResults = await Promise.all( @@ -195,7 +189,7 @@ async function hardenCitation( /** * Return a durable Notebook-owned URL for a reference: keep already-owned URLs, - * otherwise resolve against the source's durable parsed asset map. Returns + * otherwise harden only the referenced asset path. Returns * `undefined` when no durable URL is available so callers omit the URL rather * than leak a presigned Knowhere URL. */ @@ -210,15 +204,24 @@ async function resolveDurableAssetUrl( const source = resolveSourceForReference(reference, context) if (!source) return undefined - const assetUrlsByFilePath = await getCachedSourceAssetUrls(source, context) - const durableUrl = resolveAssetUrlFromReferenceText({ - values: [ - reference.source?.sectionPath, - reference.content, - getAssetUrlPathname(reference.assetUrl), - ], - assetUrlsByFilePath, - }) + const sourcePath = resolveSourcePathForReference(reference) + if (!sourcePath) return undefined + + const durableUrl = await context + .hardenChatAssetUrl({ + source, + sourcePath, + assetUrl: reference.assetUrl, + }) + .catch((error: unknown) => { + logger.warn("chat: failed to harden referenced asset", { + sourceId: source.id, + sourcePath, + error: formatUnknownError(error), + }) + return null + }) + return durableUrl ?? undefined } @@ -261,26 +264,6 @@ function applyAssetUrls< return next as T } -async function getCachedSourceAssetUrls( - source: Source, - context: HardeningContext, -): Promise>> { - const cached = context.assetUrlsBySourceId.get(source.id) - if (cached) return cached - - const loaded = context - .loadSourceAssetUrls(source) - .catch((error: unknown) => { - logger.warn("chat: failed to load durable parsed asset map", { - sourceId: source.id, - error: formatUnknownError(error), - }) - return {} - }) - context.assetUrlsBySourceId.set(source.id, loaded) - return loaded -} - export function isNotebookOwnedAssetUrl(assetUrl: string): boolean { const pathname = getAssetUrlPathname(assetUrl).toLowerCase() if ( @@ -313,6 +296,58 @@ function parseAbsoluteHttpUrl(assetUrl: string): URL | null { } } +function resolveSourcePathForReference( + reference: AssetUrlReference, +): string | null { + const candidates = [ + reference.source?.sectionPath, + reference.content, + getAssetUrlPathname(reference.assetUrl), + ] + + for (const candidate of candidates) { + const sourcePath = getSupportedAssetPath(candidate) + if (sourcePath) return sourcePath + } + + return null +} + +function getSupportedAssetPath(value: string | null | undefined): string | null { + const normalizedText = normalizeSourcePathCandidate(value) + if (!normalizedText) return null + + const match = + /(?:^|\/)((?:images|tables|pages|page_citation_assets)\/[^?#]+)/i.exec( + normalizedText, + ) + const matchedPath = match?.[1] + return matchedPath ? matchedPath.trim() : null +} + +function normalizeSourcePathCandidate( + value: string | null | undefined, +): string | null { + const trimmedValue = getTrimmedString(value) + if (!trimmedValue) return null + + const normalized = decodeUrlText(trimmedValue) + .replaceAll("\\", "/") + .replace(/\s*\/\s*/g, "/") + .replace(/\s+/g, " ") + .trim() + + return normalized.length > 0 ? normalized : null +} + +function decodeUrlText(value: string): string { + try { + return decodeURIComponent(value) + } catch { + return value + } +} + function resolveSourceForReference( reference: AssetUrlReference, context: HardeningContext, diff --git a/src/domains/chat/media-assets.test.ts b/src/domains/chat/media-assets.test.ts index 2fc84cb..470255d 100644 --- a/src/domains/chat/media-assets.test.ts +++ b/src/domains/chat/media-assets.test.ts @@ -11,10 +11,11 @@ import type { Source } from "@/infrastructure/db/schema" describe("chat media assets", () => { it("enriches retrieved image chunks from Notebook parsed asset URLs", async () => { - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "images/image-9-Night Rocket Launch.jpg": + const hardenChatAssetUrl = vi + .fn() + .mockResolvedValue( "https://blob.example/images/image-9-Night%20Rocket%20Launch.jpg", - }) + ) const [result] = await enrichRetrievalResultsWithAssetUrls({ results: [ @@ -33,20 +34,25 @@ describe("chat media assets", () => { knowhereDocumentId: "doc_spacex", }), ], - loadSourceAssetUrls, + hardenChatAssetUrl, }) - expect(loadSourceAssetUrls).toHaveBeenCalledTimes(1) + expect(hardenChatAssetUrl).toHaveBeenCalledWith({ + source: expect.objectContaining({ id: "source_spacex" }), + sourcePath: "images/image-9-Night Rocket Launch.jpg", + assetUrl: null, + }) expect(result?.assetUrl).toBe( "https://blob.example/images/image-9-Night%20Rocket%20Launch.jpg", ) }) it("prefers Notebook parsed asset URLs over existing upstream asset URLs", async () => { - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "images/image-6-情感分类模型.jpg": + const hardenChatAssetUrl = vi + .fn() + .mockResolvedValue( "https://blob.example/workspaces/workspace_1/sources/source_doc/parsed-result/images/image-6-model.jpg", - }) + ) const [result] = await enrichRetrievalResultsWithAssetUrls({ results: [ @@ -67,7 +73,7 @@ describe("chat media assets", () => { knowhereDocumentId: "doc_model", }), ], - loadSourceAssetUrls, + hardenChatAssetUrl, }) expect(result?.assetUrl).toBe( @@ -75,13 +81,8 @@ describe("chat media assets", () => { ) }) - it("adds image citation results for asset filenames that only appear in evidence text", async () => { - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "images/image-6-中华人民共和国居民身份证.jpg": - "https://blob.example/images/image-6-id-front.jpg", - "images/image-7-中国居民身份证.jpg": - "https://blob.example/images/image-7-id-back.jpg", - }) + it("does not scan a source for asset filenames that only appear in evidence text", async () => { + const hardenChatAssetUrl = vi.fn().mockResolvedValue(null) const results = await enrichRetrievalResultsWithAssetUrls({ results: [ @@ -101,25 +102,14 @@ describe("chat media assets", () => { knowhereDocumentId: "doc_identity", }), ], - loadSourceAssetUrls, + hardenChatAssetUrl, evidenceText: "[image-6-中华人民共和国居民身份证.jpg]\n[image-7-中国居民身份证.jpg]", }) - expect(results).toHaveLength(3) + expect(results).toHaveLength(1) expect(results[0]?.assetUrl).toBeUndefined() - expect(results.slice(1).map((result) => result.assetUrl)).toEqual([ - "https://blob.example/images/image-6-id-front.jpg", - "https://blob.example/images/image-7-id-back.jpg", - ]) - expect(results.slice(1).map((result) => result.chunkType)).toEqual([ - "image", - "image", - ]) - expect(results.slice(1).map((result) => result.source.sectionPath)).toEqual([ - "images/image-6-中华人民共和国居民身份证.jpg", - "images/image-7-中国居民身份证.jpg", - ]) + expect(hardenChatAssetUrl).not.toHaveBeenCalled() }) it("deduplicates media citation assets globally by asset URL", async () => { diff --git a/src/domains/chat/media-assets.ts b/src/domains/chat/media-assets.ts index 85984cc..bafaae8 100644 --- a/src/domains/chat/media-assets.ts +++ b/src/domains/chat/media-assets.ts @@ -12,24 +12,30 @@ const internalMetadataKeys = new Set([ "chunk_id", ]) -export type LoadSourceAssetUrls = ( - source: Source, -) => Promise>> +export type HardenChatAssetUrlInput = { + readonly source: Source + readonly sourcePath: string + readonly assetUrl?: string | null + readonly contentType?: string | null +} + +export type HardenChatAssetUrl = ( + input: HardenChatAssetUrlInput, +) => Promise export type RetrievalResultAssetInput = { readonly results: readonly RetrievalResult[] readonly sources: readonly Source[] - readonly loadSourceAssetUrls?: LoadSourceAssetUrls + readonly hardenChatAssetUrl?: HardenChatAssetUrl readonly evidenceText?: string } export async function enrichRetrievalResultsWithAssetUrls({ results, sources, - loadSourceAssetUrls, - evidenceText, + hardenChatAssetUrl, }: RetrievalResultAssetInput): Promise { - if (!loadSourceAssetUrls || results.length === 0) { + if (!hardenChatAssetUrl || results.length === 0) { return dedupeMediaCitationResults(results) } @@ -38,23 +44,13 @@ export async function enrichRetrievalResultsWithAssetUrls({ source.knowhereDocumentId ? [[source.knowhereDocumentId, source]] : [], ), ) - const assetUrlsBySourceId = new Map< - string, - Promise>> - >() - const enrichedResults = await Promise.all( results.map(async (result): Promise => { const documentId = getTrimmedString(result.source.documentId) const source = documentId ? sourcesByDocumentId.get(documentId) : undefined if (!source) return [result] - const assetUrls = await getCachedSourceAssetUrls( - source, - loadSourceAssetUrls, - assetUrlsBySourceId, - ) - return addAssetCitationResults(result, assetUrls, evidenceText) + return addAssetCitationResults(result, source, hardenChatAssetUrl) }), ) @@ -234,56 +230,33 @@ function isNotebookParsedAssetUrl(assetUrl: string): boolean { ) === true } -async function getCachedSourceAssetUrls( +async function addAssetCitationResults( + result: RetrievalResult, source: Source, - loadSourceAssetUrls: LoadSourceAssetUrls, - cache: Map>>>, -): Promise>> { - let cached = cache.get(source.id) - if (!cached) { - cached = loadSourceAssetUrls(source).catch(() => ({})) - cache.set(source.id, cached) - } - return cached -} + hardenChatAssetUrl: HardenChatAssetUrl, +): Promise { + if (result.chunkType.toLowerCase() === "page") return [result] -function addAssetCitationResults( - result: RetrievalResult, - assetUrlsByFilePath: Readonly>, - evidenceText: string | undefined, -): readonly RetrievalResult[] { const existingAssetUrl = getTrimmedString(result.assetUrl) - const resultMatches = resolveAssetReferenceMatches(result, assetUrlsByFilePath) - const evidenceMatches = resolveAssetReferenceMatchesFromText( - evidenceText, - assetUrlsByFilePath, - ) - const seenAssetUrls = new Set() - const output: RetrievalResult[] = [] - - if (resultMatches.length > 0) { - const [firstMatch, ...remainingMatches] = resultMatches - seenAssetUrls.add(firstMatch.assetUrl) - output.push(toAssetResult(result, firstMatch)) - for (const match of remainingMatches) { - if (seenAssetUrls.has(match.assetUrl)) continue - seenAssetUrls.add(match.assetUrl) - output.push(toAssetResult(result, match)) - } - } else if (existingAssetUrl) { - seenAssetUrls.add(existingAssetUrl) - output.push(result) - } else { - output.push(result) - } - - for (const match of evidenceMatches) { - if (seenAssetUrls.has(match.assetUrl)) continue - seenAssetUrls.add(match.assetUrl) - output.push(toAssetResult(result, match)) - } - - return output + if (existingAssetUrl && isNotebookOwnedAssetUrl(existingAssetUrl)) return [result] + + const sourcePath = getAssetSourcePathFromResult(result, existingAssetUrl) + if (!sourcePath) return [result] + + const assetUrl = await hardenChatAssetUrl({ + source, + sourcePath, + assetUrl: existingAssetUrl, + }).catch(() => null) + if (!assetUrl) return [result] + + return [ + toAssetResult(result, { + assetPath: sourcePath, + assetUrl, + index: 0, + }), + ] } function toAssetResult( @@ -346,128 +319,12 @@ function isAssetFilePath(value: string): boolean { return normalizedPath ? /^(images|tables)\//.test(normalizedPath) : false } -function resolveAssetReferenceMatches( - result: RetrievalResult, - assetUrlsByFilePath: Readonly>, -): readonly AssetReferenceMatch[] { - const normalizedHaystacks = [ - result.source.sectionPath, - result.content, - ].flatMap((value): string[] => { - const normalized = normalizeAssetLookupText(value) - return normalized ? [normalized] : [] - }) - if (normalizedHaystacks.length === 0) return [] - - return resolveAssetReferenceMatchesFromHaystacks( - normalizedHaystacks, - assetUrlsByFilePath, - ) -} - -function resolveAssetReferenceMatchesFromText( - value: string | null | undefined, - assetUrlsByFilePath: Readonly>, -): readonly AssetReferenceMatch[] { - const normalized = normalizeAssetLookupText(value) - if (!normalized) return [] - - return resolveAssetReferenceMatchesFromHaystacks( - [normalized], - assetUrlsByFilePath, - ) -} - -export function resolveAssetUrlFromReferenceText(input: { - readonly values: readonly (string | null | undefined)[] - readonly assetUrlsByFilePath: Readonly> -}): string | null { - const normalizedHaystacks = input.values.flatMap((value): string[] => { - const normalized = normalizeAssetLookupText(value) - return normalized ? [normalized] : [] - }) - if (normalizedHaystacks.length === 0) return null - - const [match] = resolveAssetReferenceMatchesFromHaystacks( - normalizedHaystacks, - input.assetUrlsByFilePath, - ) - return match?.assetUrl ?? null -} - -function resolveAssetReferenceMatchesFromHaystacks( - normalizedHaystacks: readonly string[], - assetUrlsByFilePath: Readonly>, -): readonly AssetReferenceMatch[] { - const basenameCounts = getNormalizedBasenameCounts(assetUrlsByFilePath) - return Object.entries(assetUrlsByFilePath) - .flatMap(([assetPath, assetUrl]): readonly AssetReferenceMatch[] => { - const trimmedUrl = getTrimmedString(assetUrl) - if (!trimmedUrl || !isSupportedAssetPath(assetPath)) return [] - - const index = getAssetReferenceIndex( - normalizedHaystacks, - assetPath, - basenameCounts, - ) - return index === null ? [] : [{ assetPath, assetUrl: trimmedUrl, index }] - }) - .sort(compareAssetReferenceMatches) -} - type AssetReferenceMatch = { readonly assetPath: string readonly assetUrl: string readonly index: number } -function compareAssetReferenceMatches( - left: AssetReferenceMatch, - right: AssetReferenceMatch, -): number { - return left.index - right.index || left.assetPath.localeCompare(right.assetPath) -} - -function getAssetReferenceIndex( - normalizedHaystacks: readonly string[], - assetPath: string, - basenameCounts: ReadonlyMap, -): number | null { - const normalizedPath = normalizeAssetLookupText(assetPath) - if (!normalizedPath) return null - - const directIndex = getFirstIndex(normalizedHaystacks, normalizedPath) - if (directIndex !== null) return directIndex - - const basename = getNormalizedBasename(assetPath) - if (!basename || basenameCounts.get(basename) !== 1) return null - - return getFirstIndex(normalizedHaystacks, basename) -} - -function getFirstIndex( - normalizedHaystacks: readonly string[], - needle: string, -): number | null { - const indexes = normalizedHaystacks - .map((haystack): number => haystack.indexOf(needle)) - .filter((index): index is number => index >= 0) - - return indexes.length > 0 ? Math.min(...indexes) : null -} - -function getNormalizedBasenameCounts( - assetUrlsByFilePath: Readonly>, -): ReadonlyMap { - const counts = new Map() - for (const assetPath of Object.keys(assetUrlsByFilePath)) { - const basename = getNormalizedBasename(assetPath) - if (!basename) continue - counts.set(basename, (counts.get(basename) ?? 0) + 1) - } - return counts -} - function getNormalizedBasename(assetPath: string): string | null { const basename = assetPath.replaceAll("\\", "/").split("/").pop() return normalizeAssetLookupText(basename) @@ -495,14 +352,84 @@ function decodeUrlText(value: string): string { } } +function getAssetSourcePathFromResult( + result: RetrievalResult, + assetUrl: string | null, +): string | null { + const candidates = [ + result.filePath, + result.sourceChunkPath, + result.source.sectionPath, + assetUrl ? getUrlPathname(assetUrl) : null, + result.content, + ] + + for (const candidate of candidates) { + const sourcePath = getSupportedAssetPath(candidate) + if (sourcePath) return sourcePath + } + + return null +} + +function getSupportedAssetPath(value: string | null | undefined): string | null { + const normalizedText = normalizeSourcePathCandidate(value) + if (!normalizedText) return null + + const match = + /(?:^|\/)((?:images|tables|pages|page_citation_assets)\/[^?#]+)/i.exec( + normalizedText, + ) + const matchedPath = match?.[1] + return matchedPath ? matchedPath.trim() : null +} + +function normalizeSourcePathCandidate( + value: string | null | undefined, +): string | null { + const trimmedValue = getTrimmedString(value) + if (!trimmedValue) return null + + const normalized = decodeUrlText(trimmedValue) + .replaceAll("\\", "/") + .replace(/\s*\/\s*/g, "/") + .replace(/\s+/g, " ") + .trim() + + return normalized.length > 0 ? normalized : null +} + function isSupportedAssetPath(assetPath: string): boolean { - const normalizedPath = normalizeAssetLookupText(assetPath) + return getSupportedAssetPath(assetPath) !== null +} + +function isNotebookOwnedAssetUrl(assetUrl: string): boolean { + const pathname = getUrlPathname(assetUrl).toLowerCase() + if ( + pathname.includes("/parsed-result/") || + pathname.includes("/chat-assets/") || + pathname.includes("/parsed-documents/") + ) { + return true + } + + const absoluteUrl = parseAbsoluteHttpUrl(assetUrl) return ( - normalizedPath?.startsWith("images/") === true || - normalizedPath?.startsWith("tables/") === true + absoluteUrl?.hostname + .toLowerCase() + .endsWith(".blob.vercel-storage.com") === true ) } +function parseAbsoluteHttpUrl(assetUrl: string): URL | null { + try { + const url = new URL(assetUrl) + return url.protocol === "http:" || url.protocol === "https:" ? url : null + } catch { + return null + } +} + function isRenderableMediaAsset( result: RetrievalResult, assetUrl: string, diff --git a/src/domains/chat/page-citation-assets.test.ts b/src/domains/chat/page-citation-assets.test.ts index d99dbcb..4553d5d 100644 --- a/src/domains/chat/page-citation-assets.test.ts +++ b/src/domains/chat/page-citation-assets.test.ts @@ -29,10 +29,9 @@ describe("enrichRetrievalResultsWithPageCitationAssetUrls", () => { }) it("uses stored Blob URLs for page citation assets", async () => { - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "page_citation_assets/page-2.png": - "https://blob.example/page_citation_assets/page-2.png", - }) + const hardenChatAssetUrl = vi + .fn() + .mockResolvedValue("https://blob.example/page_citation_assets/page-2.png") const [result] = await enrichRetrievalResultsWithPageCitationAssetUrls({ results: [ @@ -51,22 +50,24 @@ describe("enrichRetrievalResultsWithPageCitationAssetUrls", () => { }), ], sources: [makeSource()], - loadSourceAssetUrls, + hardenChatAssetUrl, }) - expect(loadSourceAssetUrls).toHaveBeenCalledWith( - expect.objectContaining({ id: "source_1" }), - ) + expect(hardenChatAssetUrl).toHaveBeenCalledWith({ + source: expect.objectContaining({ id: "source_1" }), + sourcePath: "page_citation_assets/page-2.png", + assetUrl: "https://assets.example/pages/page-2.png", + contentType: undefined, + }) expect(result?.pageCitationAssetUrl).toBe( "https://blob.example/page_citation_assets/page-2.png", ) }) it("chooses the stored asset matching the citation page metadata", async () => { - const loadSourceAssetUrls = vi.fn().mockResolvedValue({ - "page_citation_assets/page-4.png": - "https://blob.example/page_citation_assets/page-4.png", - }) + const hardenChatAssetUrl = vi + .fn() + .mockResolvedValue("https://blob.example/page_citation_assets/page-4.png") const [result] = await enrichRetrievalResultsWithPageCitationAssetUrls({ results: [ @@ -90,9 +91,15 @@ describe("enrichRetrievalResultsWithPageCitationAssetUrls", () => { }), ], sources: [makeSource()], - loadSourceAssetUrls, + hardenChatAssetUrl, }) + expect(hardenChatAssetUrl).toHaveBeenCalledWith({ + source: expect.objectContaining({ id: "source_1" }), + sourcePath: "page_citation_assets/page-4.png", + assetUrl: "https://assets.example/pages/page-4.png", + contentType: undefined, + }) expect(result?.pageCitationAssetUrl).toBe( "https://blob.example/page_citation_assets/page-4.png", ) diff --git a/src/domains/chat/page-citation-assets.ts b/src/domains/chat/page-citation-assets.ts index 77b8d53..566cead 100644 --- a/src/domains/chat/page-citation-assets.ts +++ b/src/domains/chat/page-citation-assets.ts @@ -3,7 +3,7 @@ import "server-only" import type { RetrievalResult } from "@ontos-ai/knowhere-sdk" import type { Source } from "@/infrastructure/db/schema" -import type { LoadSourceAssetUrls } from "./media-assets" +import type { HardenChatAssetUrl } from "./media-assets" export type PageCitationAssetRetrievalResult = RetrievalResult & { readonly pageCitationAssetUrl?: string @@ -12,19 +12,20 @@ export type PageCitationAssetRetrievalResult = RetrievalResult & { type EnrichRetrievalResultsWithPageCitationAssetUrlsInput = { readonly results: readonly RetrievalResult[] readonly sources: readonly Source[] - readonly loadSourceAssetUrls?: LoadSourceAssetUrls + readonly hardenChatAssetUrl?: HardenChatAssetUrl } type PageCitationAssetCandidate = { readonly pageNum: number readonly artifactRef?: string readonly assetUrl?: string + readonly contentType?: string } export async function enrichRetrievalResultsWithPageCitationAssetUrls({ results, sources, - loadSourceAssetUrls, + hardenChatAssetUrl, }: EnrichRetrievalResultsWithPageCitationAssetUrlsInput): Promise< PageCitationAssetRetrievalResult[] > { @@ -35,18 +36,12 @@ export async function enrichRetrievalResultsWithPageCitationAssetUrls({ source.knowhereDocumentId ? [[source.knowhereDocumentId, source]] : [], ), ) - const assetUrlsBySourceId = new Map< - string, - Promise>> - >() - return Promise.all( results.map((result) => enrichRetrievalResultWithPageCitationAssetUrl({ result, sourcesByDocumentId, - loadSourceAssetUrls, - assetUrlsBySourceId, + hardenChatAssetUrl, }), ), ) @@ -55,11 +50,7 @@ export async function enrichRetrievalResultsWithPageCitationAssetUrls({ async function enrichRetrievalResultWithPageCitationAssetUrl(input: { readonly result: RetrievalResult readonly sourcesByDocumentId: ReadonlyMap - readonly loadSourceAssetUrls?: LoadSourceAssetUrls - readonly assetUrlsBySourceId: Map< - string, - Promise>> - > + readonly hardenChatAssetUrl?: HardenChatAssetUrl }): Promise { if (!isPageResult(input.result)) return input.result @@ -69,8 +60,7 @@ async function enrichRetrievalResultWithPageCitationAssetUrl(input: { result: input.result, directAsset, sourcesByDocumentId: input.sourcesByDocumentId, - loadSourceAssetUrls: input.loadSourceAssetUrls, - assetUrlsBySourceId: input.assetUrlsBySourceId, + hardenChatAssetUrl: input.hardenChatAssetUrl, }) if (sourceAssetUrl) { return { @@ -86,25 +76,21 @@ async function getStoredPageCitationAssetUrl(input: { readonly result: RetrievalResult readonly directAsset: PageCitationAssetCandidate | null readonly sourcesByDocumentId: ReadonlyMap - readonly loadSourceAssetUrls?: LoadSourceAssetUrls - readonly assetUrlsBySourceId: Map< - string, - Promise>> - > + readonly hardenChatAssetUrl?: HardenChatAssetUrl }): Promise { const artifactRef = getTrimmedString(input.directAsset?.artifactRef) const documentId = getTrimmedString(input.result.source.documentId) - if (!artifactRef || !documentId || !input.loadSourceAssetUrls) return null + if (!artifactRef || !documentId || !input.hardenChatAssetUrl) return null const source = input.sourcesByDocumentId.get(documentId) if (!source) return null - const assetUrls = await getCachedSourceAssetUrls( + return input.hardenChatAssetUrl({ source, - input.loadSourceAssetUrls, - input.assetUrlsBySourceId, - ) - return getTrimmedString(assetUrls[artifactRef]) + sourcePath: artifactRef, + assetUrl: input.directAsset?.assetUrl, + contentType: input.directAsset?.contentType, + }).catch(() => null) } function isPageResult(result: RetrievalResult): boolean { @@ -142,24 +128,12 @@ function parsePageCitationAssetCandidates( pageNum, artifactRef: getTrimmedString(item.artifactRef) ?? undefined, assetUrl: getTrimmedString(item.assetUrl) ?? undefined, + contentType: getTrimmedString(item.contentType) ?? undefined, }, ] }) } -async function getCachedSourceAssetUrls( - source: Source, - loadSourceAssetUrls: LoadSourceAssetUrls, - cache: Map>>>, -): Promise>> { - let cached = cache.get(source.id) - if (!cached) { - cached = loadSourceAssetUrls(source).catch(() => ({})) - cache.set(source.id, cached) - } - return cached -} - function getPageNumbers( metadata: Readonly> | undefined, ): readonly number[] { diff --git a/src/domains/chat/route-answer.ts b/src/domains/chat/route-answer.ts index 9582b77..03c26ca 100644 --- a/src/domains/chat/route-answer.ts +++ b/src/domains/chat/route-answer.ts @@ -11,12 +11,12 @@ import { type ChatTurnValue, } from "@/domains/chat/service" import { chatTurnPersistence } from "@/domains/chat/chat-turn-persistence" -import { readSourceAssetUrls } from "@/domains/chunks/read" import { startBackgroundReconciliation } from "@/domains/sources/background-reconcile" +import { BlobParsedDocumentStorage } from "@/domains/sources/parsed-document-blob-storage" import { sourceWorkflowRuntime } from "@/domains/sources/workflow-runtime" -import { makeKnowhereClientWithParsedStorage } from "@/integrations/knowhere" import { notebookRequestContext } from "@/domains/workspace/request-context" import type { Source } from "@/infrastructure/db/schema" +import type { HardenChatAssetUrl } from "./media-assets" import { isAuthError } from "@/integrations/dashboard/api-key-service" import { summarizeUnknownError } from "@/lib/format-log-value" import { logger } from "@/lib/logger" @@ -69,30 +69,23 @@ const answerChatEffect = (input: AnswerChatInput) => apiKey, }), ) - const { knowledge } = makeKnowhereClientWithParsedStorage(apiKey, { + const parsedStorage = new BlobParsedDocumentStorage({ workspaceId: workspace.id, }) - const loadSourceAssetUrls = async ( - source: (typeof sources)[number], - ): Promise>> => { - if (source.status !== "ready" || !source.knowhereDocumentId) return {} - - try { - return await readSourceAssetUrls({ - knowledge, - documentId: source.knowhereDocumentId, - revisionKey: source.knowhereJobId, - }) - } catch (error) { - logger.warn("chat: durable parsed asset read failed", { - workspaceId: workspace.id, - sourceId: source.id, - documentId: source.knowhereDocumentId, - error: summarizeUnknownError(error), - }) - return {} - } - } + const hardenChatAssetUrl: HardenChatAssetUrl = async ({ + source, + sourcePath, + assetUrl, + contentType, + }): Promise => + hardenSingleChatAsset({ + workspaceId: workspace.id, + parsedStorage, + source, + sourcePath, + assetUrl, + contentType, + }) const result: Either.Either = yield* Effect.tryPromise(() => @@ -104,14 +97,14 @@ const answerChatEffect = (input: AnswerChatInput) => excludedSourceIds: body.value.excludedSourceIds, retrieval: client.retrieval, generateAnswer: generateAgenticOutputManifest, - loadSourceAssetUrls, + hardenChatAssetUrl, hardenMediaAssetUrls: ({ results, artifacts }) => hardenChatMediaAssetUrls({ workspaceId: workspace.id, sources, results, artifacts, - loadSourceAssetUrls, + hardenChatAssetUrl, }), repository: chatTurnPersistence.createRepository(), }), @@ -167,6 +160,104 @@ export const chatAnswerRouteService: ChatAnswerRouteService = { answerChat, } +async function hardenSingleChatAsset(input: { + readonly workspaceId: string + readonly parsedStorage: BlobParsedDocumentStorage + readonly source: Source + readonly sourcePath: string + readonly assetUrl?: string | null + readonly contentType?: string | null +}): Promise { + if ( + input.source.status !== "ready" || + !input.source.knowhereDocumentId || + !input.source.knowhereJobId + ) { + return null + } + + try { + const existingUrl = await input.parsedStorage.getAssetUrl({ + documentId: input.source.knowhereDocumentId, + revisionKey: input.source.knowhereJobId, + sourcePath: input.sourcePath, + }) + if (existingUrl) return existingUrl + + const sourceUrl = getTrimmedString(input.assetUrl) + if (!sourceUrl) return null + + const fetchedAsset = await fetchChatAsset({ + assetUrl: sourceUrl, + fallbackContentType: + getTrimmedString(input.contentType) ?? + inferContentTypeFromPath(input.sourcePath), + }) + const writtenAsset = await input.parsedStorage.writeAsset({ + documentId: input.source.knowhereDocumentId, + revisionKey: input.source.knowhereJobId, + sourcePath: input.sourcePath, + body: fetchedAsset.body, + contentType: fetchedAsset.contentType, + }) + + return ( + writtenAsset.url ?? + (await input.parsedStorage.getAssetUrl({ + documentId: input.source.knowhereDocumentId, + revisionKey: input.source.knowhereJobId, + sourcePath: input.sourcePath, + })) + ) + } catch (error) { + logger.warn("chat: single parsed asset hardening failed", { + workspaceId: input.workspaceId, + sourceId: input.source.id, + documentId: input.source.knowhereDocumentId, + sourcePath: input.sourcePath, + error: summarizeUnknownError(error), + }) + return null + } +} + +async function fetchChatAsset(input: { + readonly assetUrl: string + readonly fallbackContentType: string +}): Promise<{ + readonly body: Uint8Array + readonly contentType: string +}> { + const response = await fetch(input.assetUrl) + if (!response.ok) { + throw new Error( + `Failed to fetch chat asset (${response.status} ${response.statusText})`, + ) + } + + return { + body: new Uint8Array(await response.arrayBuffer()), + contentType: + getTrimmedString(response.headers.get("content-type")) ?? + input.fallbackContentType, + } +} + +function inferContentTypeFromPath(sourcePath: string): string { + const pathname = sourcePath.toLowerCase().split("?")[0] ?? sourcePath + if (pathname.endsWith(".png")) return "image/png" + if (pathname.endsWith(".jpg") || pathname.endsWith(".jpeg")) { + return "image/jpeg" + } + if (pathname.endsWith(".gif")) return "image/gif" + if (pathname.endsWith(".webp")) return "image/webp" + if (pathname.endsWith(".svg")) return "image/svg+xml" + if (pathname.endsWith(".html") || pathname.endsWith(".htm")) { + return "text/html; charset=utf-8" + } + return "application/octet-stream" +} + function triggerBackgroundReconciliationForParsingSources(input: { readonly workspaceId: string readonly sources: readonly Source[] @@ -246,3 +337,8 @@ function isMeaningfulSummary(value: string): boolean { normalized !== "An unknown error occurred in Effect.tryPromise" ) } + +function getTrimmedString(value: string | null | undefined): string | null { + const trimmedValue = value?.trim() ?? "" + return trimmedValue.length > 0 ? trimmedValue : null +} diff --git a/src/domains/chat/route-service.test.ts b/src/domains/chat/route-service.test.ts index f4df92a..6365edd 100644 --- a/src/domains/chat/route-service.test.ts +++ b/src/domains/chat/route-service.test.ts @@ -1,5 +1,5 @@ import { Either } from "effect" -import { beforeEach, describe, expect, it, vi } from "vitest" +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" import type { ChatMessage, ChatThread, Source, Workspace } from "@/infrastructure/db/schema" @@ -18,8 +18,8 @@ const mocks = vi.hoisted(() => ({ loggerInfo: vi.fn(), loggerWarn: vi.fn(), listSourcesForWorkspace: vi.fn(), - makeKnowhereClientWithParsedStorage: vi.fn(), - readChunks: vi.fn(), + parsedStorageGetAssetUrl: vi.fn(), + parsedStorageWriteAsset: vi.fn(), softDeleteChatThread: vi.fn(), startBackgroundReconciliation: vi.fn(), })) @@ -53,9 +53,13 @@ vi.mock("@/domains/workspace/request-context", () => ({ }, })) -vi.mock("@/integrations/knowhere", () => ({ - makeKnowhereClientWithParsedStorage: - mocks.makeKnowhereClientWithParsedStorage, +vi.mock("@/domains/sources/parsed-document-blob-storage", () => ({ + BlobParsedDocumentStorage: vi.fn().mockImplementation(function () { + return { + getAssetUrl: mocks.parsedStorageGetAssetUrl, + writeAsset: mocks.parsedStorageWriteAsset, + } + }), })) vi.mock("@/domains/chat/thread-service", () => ({ @@ -84,18 +88,18 @@ import { chatThreadRouteService } from "./route-threads" describe("chat route services", () => { beforeEach(() => { vi.clearAllMocks() - mocks.makeKnowhereClientWithParsedStorage.mockReturnValue({ - client: { documents: { listChunks: vi.fn() } }, - knowledge: { readChunks: mocks.readChunks }, - }) - mocks.readChunks.mockResolvedValue({ - document: { localDocumentId: "doc" }, - chunks: [], - page: 1, - pageSize: 200, - totalChunks: 0, - totalPages: 1, - }) + mocks.parsedStorageGetAssetUrl.mockResolvedValue(null) + mocks.parsedStorageWriteAsset.mockResolvedValue({ url: null }) + vi.stubGlobal( + "fetch", + vi.fn(async () => new Response(new Uint8Array([1, 2, 3]), { + headers: { "content-type": "image/png" }, + })), + ) + }) + + afterEach(() => { + vi.unstubAllGlobals() }) it("orchestrates a chat turn from request body to response body", async () => { @@ -148,7 +152,7 @@ describe("chat route services", () => { excludedSourceIds: ["source_skipped"], retrieval: client.retrieval, generateAnswer: mocks.generateAgenticOutputManifest, - loadSourceAssetUrls: expect.any(Function), + hardenChatAssetUrl: expect.any(Function), repository: expect.objectContaining({ appendMessageToThread: expect.any(Function), ensureDefaultChatThread: expect.any(Function), @@ -159,7 +163,7 @@ describe("chat route services", () => { ) }) - it("builds durable citation asset URLs from SDK reads for a ready source", async () => { + it("hardens one chat asset through Notebook Blob for a ready source", async () => { const workspace = makeWorkspace() const client = { retrieval: { query: vi.fn() } } const readySource = makeSource({ @@ -167,29 +171,10 @@ describe("chat route services", () => { knowhereDocumentId: "doc_legacy", knowhereJobId: "job_1", }) + const rawUrl = "https://knowhere-storage.example/results/job_1/pages/page-1.png" const durableUrl = "https://fake.public.blob.vercel-storage.com/workspaces/workspace_1/parsed-documents/doc_legacy/job_1/assets/pages/page-1.png" - mocks.readChunks.mockResolvedValue({ - document: { localDocumentId: "doc_legacy" }, - chunks: [ - { - position: 1, - chunkId: "c1", - chunkType: "page", - content: "Page", - readableContent: "Page", - sectionPath: "Page 1", - sourceChunkPath: "Page 1", - filePath: "pages/page-1.png", - assetUrl: durableUrl, - metadata: {}, - }, - ], - page: 1, - pageSize: 200, - totalChunks: 1, - totalPages: 1, - }) + mocks.parsedStorageWriteAsset.mockResolvedValue({ url: durableUrl }) mocks.getAuthenticatedWithClient.mockResolvedValue({ user: { id: "user_1" }, workspace, @@ -199,12 +184,20 @@ describe("chat route services", () => { mocks.listSourcesForWorkspace.mockResolvedValue([readySource]) mocks.handleChatTurn.mockImplementation( async (input: { - readonly loadSourceAssetUrls?: ( - source: Source, - ) => Promise>> + readonly hardenChatAssetUrl?: (assetInput: { + readonly source: Source + readonly sourcePath: string + readonly assetUrl?: string | null + readonly contentType?: string | null + }) => Promise }) => { - const assetUrls = await input.loadSourceAssetUrls?.(readySource) - expect(assetUrls).toEqual({ "pages/page-1.png": durableUrl }) + const assetUrl = await input.hardenChatAssetUrl?.({ + source: readySource, + sourcePath: "pages/page-1.png", + assetUrl: rawUrl, + contentType: "image/png", + }) + expect(assetUrl).toBe(durableUrl) return Either.right({ threadId: "thread_1", messages: [ @@ -220,16 +213,18 @@ describe("chat route services", () => { }) expect(result.status).toBe(200) - expect(mocks.makeKnowhereClientWithParsedStorage).toHaveBeenCalledWith( - "jwt_123", - { workspaceId: workspace.id }, - ) - expect(mocks.readChunks).toHaveBeenCalledWith({ + expect(mocks.parsedStorageGetAssetUrl).toHaveBeenCalledWith({ + documentId: "doc_legacy", + revisionKey: "job_1", + sourcePath: "pages/page-1.png", + }) + expect(fetch).toHaveBeenCalledWith(rawUrl) + expect(mocks.parsedStorageWriteAsset).toHaveBeenCalledWith({ documentId: "doc_legacy", revisionKey: "job_1", - page: 1, - pageSize: 200, - assetUrlPolicy: "durable", + sourcePath: "pages/page-1.png", + body: new Uint8Array([1, 2, 3]), + contentType: "image/png", }) }) diff --git a/src/domains/chat/service.ts b/src/domains/chat/service.ts index 5361d5b..8b67e7c 100644 --- a/src/domains/chat/service.ts +++ b/src/domains/chat/service.ts @@ -67,7 +67,7 @@ type ChatTurnInput = { excludedSourceIds: readonly string[] retrieval: RetrievalClient generateAnswer: GenerateAnswer - loadSourceAssetUrls?: AnswerQuestionInput["loadSourceAssetUrls"] + hardenChatAssetUrl?: AnswerQuestionInput["hardenChatAssetUrl"] hardenMediaAssetUrls?: AnswerQuestionInput["hardenMediaAssetUrls"] repository: ChatRepository } @@ -126,7 +126,7 @@ export const handleChatTurnEffect = (input: ChatTurnInput) => excludedSourceIds: input.excludedSourceIds, retrieval: input.retrieval, generateAnswer: input.generateAnswer, - loadSourceAssetUrls: input.loadSourceAssetUrls, + hardenChatAssetUrl: input.hardenChatAssetUrl, hardenMediaAssetUrls: input.hardenMediaAssetUrls, messages: chatHistoryMessages, }).pipe(Effect.catchAllCause(Effect.die)) diff --git a/src/domains/chunks/index.ts b/src/domains/chunks/index.ts index 5e1b25a..56c2644 100644 --- a/src/domains/chunks/index.ts +++ b/src/domains/chunks/index.ts @@ -181,8 +181,8 @@ export function toParsedChunkView( /** * Map an SDK `KnowledgeReadChunk` (from `knowledge.readChunks`) to the view - * model. Asset URLs on durable reads are already hardened SDK-side, so no - * `assetUrlsByFilePath` remap is needed here. + * model. Display reads do not request durable asset URL hardening; chat + * hardens the specific assets it returns separately. */ export function toParsedChunkViewFromReadChunk( chunk: KnowledgeReadChunk, diff --git a/src/domains/chunks/read.test.ts b/src/domains/chunks/read.test.ts index 1d0786d..b6a3979 100644 --- a/src/domains/chunks/read.test.ts +++ b/src/domains/chunks/read.test.ts @@ -20,7 +20,7 @@ function makeReadChunk(overrides: Record = {}) { } describe("readSourceChunkPage", () => { - it("reads a durable page and maps chunks to the view model", async () => { + it("reads a page without durable asset hardening and maps chunks to the view model", async () => { const readChunks = vi.fn(async () => ({ document: { localDocumentId: "doc_1" }, chunks: [ @@ -48,7 +48,6 @@ describe("readSourceChunkPage", () => { revisionKey: "rev_1", page: 2, pageSize: 50, - assetUrlPolicy: "durable", }) expect(result.pagination).toEqual({ page: 2, @@ -85,7 +84,6 @@ describe("readSourceChunkPage", () => { documentId: "doc_1", page: 1, pageSize: 50, - assetUrlPolicy: "durable", }) }) }) diff --git a/src/domains/chunks/read.ts b/src/domains/chunks/read.ts index f375d48..320d559 100644 --- a/src/domains/chunks/read.ts +++ b/src/domains/chunks/read.ts @@ -16,8 +16,8 @@ type ReadableSource = { /** * Read a single display page of parsed chunks through the SDK. The SDK serves * from configured Blob storage when fresh and falls back to Knowhere remote - * otherwise, hardening visible asset URLs into durable Blob URLs - * (`assetUrlPolicy: "durable"`) and scheduling a background sync on a miss. + * otherwise. Display reads intentionally do not request durable asset URLs: + * chat hardens only the specific assets it sends back to the user. */ export async function readSourceChunkPage(input: { readonly knowledge: Knowledge @@ -29,7 +29,6 @@ export async function readSourceChunkPage(input: { ...(input.source.revisionKey ? { revisionKey: input.source.revisionKey } : {}), page: input.params.page, pageSize: input.params.pageSize, - assetUrlPolicy: "durable", }) const chunks = response.chunks.map((chunk) => @@ -69,7 +68,6 @@ export async function readAllSourceChunks(input: { : {}), page, pageSize: loadAllPageSize, - assetUrlPolicy: "durable", }) for (const chunk of response.chunks) { chunks.push( @@ -86,57 +84,3 @@ export async function readAllSourceChunks(input: { return chunks } - -/** - * Build a `filePath -> durable Blob URL` map for a source by paging durable - * reads to exhaustion. This is the single asset-hardening path for chat: the - * SDK writes any missing asset into Blob during the durable read and returns - * the durable URL, which we index by both the chunk file path and any - * `metadata.pageAssets[].artifactRef`. - */ -export async function readSourceAssetUrls(input: { - readonly knowledge: Knowledge - readonly documentId: string - readonly revisionKey?: string | null -}): Promise>> { - const assetUrlsByFilePath: Record = {} - let page = 1 - let totalPages = 1 - - do { - const response = await input.knowledge.readChunks({ - documentId: input.documentId, - ...(input.revisionKey ? { revisionKey: input.revisionKey } : {}), - page, - pageSize: loadAllPageSize, - assetUrlPolicy: "durable", - }) - for (const chunk of response.chunks) { - if (chunk.filePath && chunk.assetUrl) { - assetUrlsByFilePath[chunk.filePath] = chunk.assetUrl - } - collectPageAssetUrls(chunk.metadata, assetUrlsByFilePath) - } - totalPages = Math.max(1, response.totalPages ?? 1) - page += 1 - } while (page <= totalPages) - - return assetUrlsByFilePath -} - -function collectPageAssetUrls( - metadata: Record, - target: Record, -): void { - const pageAssets = metadata["pageAssets"] - if (!Array.isArray(pageAssets)) return - for (const pageAsset of pageAssets) { - if (typeof pageAsset !== "object" || pageAsset === null) continue - const record = pageAsset as Record - const artifactRef = record["artifactRef"] - const assetUrl = record["assetUrl"] - if (typeof artifactRef === "string" && typeof assetUrl === "string") { - target[artifactRef] = assetUrl - } - } -} diff --git a/src/domains/sources/source-reconcile-workflow.test.ts b/src/domains/sources/source-reconcile-workflow.test.ts index 5757da8..68769b7 100644 --- a/src/domains/sources/source-reconcile-workflow.test.ts +++ b/src/domains/sources/source-reconcile-workflow.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it, vi } from "vitest" -import type { JobResult } from "@ontos-ai/knowhere-sdk" +import { NotFoundError, type JobResult } from "@ontos-ai/knowhere-sdk" import type { Source, Workspace } from "@/infrastructure/db/schema" import { @@ -140,6 +140,68 @@ describe("pollSourceReconciliation", () => { ) }) + it("marks parsing sources failed when the Knowhere job no longer exists", async () => { + const source = makeSource({ + stagedBlobPathname: "source-uploads/upload_1/document.pdf", + }) + const repository = createRepository(source) + const blobStore = { + deleteStagedSourceBlob: vi.fn(async () => undefined), + } + const client = { + jobs: { + get: vi.fn(async () => { + throw new NotFoundError("Job not found") + }), + }, + } + + const result = await pollSourceReconciliation({ + workspaceId: workspace.id, + sourceId: "source_1", + client, + repository, + blobStore, + }) + + expect(result).toEqual({ kind: "resolved", status: "failed" }) + expect(repository.markFailed).toHaveBeenCalledWith( + workspace.id, + "source_1", + "Knowhere job job_1 was not found during reconciliation.", + "parsing", + ) + expect(blobStore.deleteStagedSourceBlob).toHaveBeenCalledWith( + "source-uploads/upload_1/document.pdf", + ) + expect(repository.clearStagedBlob).toHaveBeenCalledWith( + workspace.id, + "source_1", + ) + }) + + it("rethrows non-not-found job lookup errors for workflow retries", async () => { + const repository = createRepository() + const client = { + jobs: { + get: vi.fn(async () => { + throw new Error("Knowhere timeout") + }), + }, + } + + await expect( + pollSourceReconciliation({ + workspaceId: workspace.id, + sourceId: "source_1", + client, + repository, + }), + ).rejects.toThrow("Knowhere timeout") + + expect(repository.markFailed).not.toHaveBeenCalled() + }) + it("accepts done jobs without a result URL when a document id is published", async () => { const repository = createRepository() const jobWithoutResultUrl = makeDoneJob() diff --git a/src/domains/sources/source-reconcile-workflow.ts b/src/domains/sources/source-reconcile-workflow.ts index 4571038..7556c76 100644 --- a/src/domains/sources/source-reconcile-workflow.ts +++ b/src/domains/sources/source-reconcile-workflow.ts @@ -1,7 +1,7 @@ import "server-only" import { del } from "@vercel/blob" -import type { JobResult } from "@ontos-ai/knowhere-sdk" +import { NotFoundError, type JobResult } from "@ontos-ai/knowhere-sdk" import type { Source } from "@/infrastructure/db/schema" import { logger } from "@/lib/logger" @@ -13,6 +13,12 @@ type SourceReconcileWorkflowClient = { } } +type ApiErrorLike = { + readonly name?: string + readonly statusCode?: number + readonly message?: string +} + type SourceReconcileWorkflowRepository = { readonly findInWorkspace: ( workspaceId: string, @@ -112,7 +118,15 @@ export async function pollSourceReconciliation({ return { kind: "resolved", status: "failed" } } - const job = await client.jobs.get(jobId) + const job = await getJobOrFailMissingJob({ + workspaceId, + source, + jobId, + client, + repository, + blobStore, + }) + if (!job) return { kind: "resolved", status: "failed" } if (isFailedJob(job)) { await failSourceAndCleanup({ workspaceId, @@ -150,6 +164,30 @@ export async function pollSourceReconciliation({ } } +async function getJobOrFailMissingJob(input: { + readonly workspaceId: string + readonly source: Source + readonly jobId: string + readonly client: SourceReconcileWorkflowClient + readonly repository: SourceReconcileWorkflowRepository + readonly blobStore: SourceReconcileWorkflowBlobStore +}): Promise { + try { + return await input.client.jobs.get(input.jobId) + } catch (error) { + if (!isKnowhereJobNotFoundError(error)) throw error + + await failSourceAndCleanup({ + workspaceId: input.workspaceId, + source: input.source, + reason: `Knowhere job ${input.jobId} was not found during reconciliation.`, + repository: input.repository, + blobStore: input.blobStore, + }) + return null + } +} + export async function markSourceReadyAfterReconciliation({ workspaceId, sourceId, @@ -245,6 +283,17 @@ function isFailedJob(job: JobResult): boolean { return job.isFailed || job.status === "failed" } +function isKnowhereJobNotFoundError(error: unknown): boolean { + if (error instanceof NotFoundError) return true + if (!isApiErrorLike(error)) return false + + return error.name === "NotFoundError" || error.statusCode === 404 +} + +function isApiErrorLike(error: unknown): error is ApiErrorLike { + return error !== null && typeof error === "object" +} + const vercelBlobStore: SourceReconcileWorkflowBlobStore = { deleteStagedSourceBlob: del, }