From 266342a64ea176f7afd3d22f583ad4505954f772 Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Thu, 28 May 2026 15:19:06 +0400 Subject: [PATCH] Studio: skip re-indexing an already-indexed document (content-hash dedup) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-uploading the same file into the same scope (KB or thread) used to parse, chunk, caption and embed it all over again, creating a duplicate set of chunks. Dedup by content hash instead: - schema: add rag_documents.content_hash (sha256 of the bytes) via the standard PRAGMA/ALTER migration, plus (scope, content_hash) indexes. - upload: _save_upload now streams the bytes through sha256 and returns the digest alongside path/name/size. - _start_ingestion: before inserting, look for a COMPLETED row in the same scope with the same hash. If found, delete the redundant upload from disk and return the existing document_id with already_indexed= true and an empty job_id — no ingestion job is started. Only 'completed' rows dedup, so a failed/in-flight prior attempt can still retry. Scope-local: the same file in two KBs is indexed in each. - frontend: UploadResponse.already_indexed flows through the rag-store (skips job subscription) into both upload paths, which mark the chip ready immediately and toast ' is already indexed'. Pre-existing rows have NULL content_hash and won't dedup until re-uploaded once under the new path. Not build/UI-verified here (no bun in this env); needs typecheck + browser check. --- studio/backend/routes/rag.py | 54 ++++++++++++++++--- studio/backend/storage/studio_db.py | 15 ++++++ .../chat/hooks/use-thread-doc-uploads.ts | 23 +++++++- .../src/features/chat/shared-composer.tsx | 22 +++++++- .../frontend/src/features/rag/api/rag-api.ts | 3 ++ .../src/features/rag/stores/rag-store.ts | 31 +++++++---- 6 files changed, 131 insertions(+), 17 deletions(-) diff --git a/studio/backend/routes/rag.py b/studio/backend/routes/rag.py index c018cac60c..35d4a43f16 100644 --- a/studio/backend/routes/rag.py +++ b/studio/backend/routes/rag.py @@ -6,6 +6,7 @@ from __future__ import annotations import asyncio +import hashlib import json import os import queue as queue_module @@ -121,6 +122,9 @@ class UploadResponse(BaseModel): document_id: str job_id: str filename: str + # True when an identical file (same content hash) was already indexed + # in this scope, so no new ingestion job was started. job_id is "". + already_indexed: bool = False class SearchRequest(BaseModel): @@ -248,7 +252,7 @@ def _document_or_404(document_id: str) -> Any: return row -async def _save_upload(file: UploadFile) -> tuple[Path, str, int]: +async def _save_upload(file: UploadFile) -> tuple[Path, str, int, str]: import anyio filename = _sanitize_filename(file.filename or "document") @@ -264,6 +268,9 @@ async def _save_upload(file: UploadFile) -> tuple[Path, str, int]: stored_path = upload_dir / stored_name max_bytes = RAG_MAX_UPLOAD_MB * 1024 * 1024 written = 0 + # Hash the bytes as they stream so we can dedup identical re-uploads + # within a scope without re-reading the file. + hasher = hashlib.sha256() # Route writes through anyio worker thread so the event loop stays free. # Outer try/except cleans up partial files after async-with closes the fd # (Windows refuses unlink on an open fd). @@ -279,6 +286,7 @@ async def _save_upload(file: UploadFile) -> tuple[Path, str, int]: status_code = 413, detail = f"File exceeds {RAG_MAX_UPLOAD_MB} MB limit", ) + hasher.update(chunk) await f.write(chunk) except HTTPException: stored_path.unlink(missing_ok = True) @@ -286,7 +294,7 @@ async def _save_upload(file: UploadFile) -> tuple[Path, str, int]: if written == 0: stored_path.unlink(missing_ok = True) raise HTTPException(status_code = 400, detail = "Empty upload payload") - return stored_path, filename, written + return stored_path, filename, written, hasher.hexdigest() def _start_ingestion( @@ -300,15 +308,46 @@ def _start_ingestion( embedding_model: str, chunking_strategy: str = "standard", mode: str = "text", + content_hash: str | None = None, ) -> UploadResponse: document_id = str(uuid4()) with get_connection() as conn: + # Dedup: if an identical file (same content hash) is already + # indexed in this scope, skip re-ingestion. Only a 'completed' + # row counts — a failed/in-flight prior attempt should be allowed + # to retry. Scope is the same kb_id or thread_id the upload + # targets (a file shared across two KBs is indexed in each). + if content_hash: + if kb_id is not None: + existing = conn.execute( + "SELECT id, filename FROM rag_documents " + "WHERE kb_id = ? AND content_hash = ? AND status = 'completed' " + "LIMIT 1", + (kb_id, content_hash), + ).fetchone() + else: + existing = conn.execute( + "SELECT id, filename FROM rag_documents " + "WHERE thread_id = ? AND content_hash = ? AND status = 'completed' " + "LIMIT 1", + (thread_id, content_hash), + ).fetchone() + if existing is not None: + # Drop the redundant upload we just wrote to disk; the + # already-indexed copy stays the source of truth. + _unlink_if_under_uploads(stored_path) + return UploadResponse( + document_id = existing["id"], + job_id = "", + filename = existing["filename"], + already_indexed = True, + ) conn.execute( """ INSERT INTO rag_documents (id, kb_id, thread_id, filename, content_type, stored_path, - status, num_chunks, byte_size, created_at) - VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?) + status, num_chunks, byte_size, content_hash, created_at) + VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?) """, ( document_id, @@ -318,6 +357,7 @@ def _start_ingestion( content_type, str(stored_path), byte_size, + content_hash, _now_ms(), ), ) @@ -806,7 +846,7 @@ async def upload_kb_document( current_subject: str = Depends(get_current_subject), ) -> UploadResponse: kb_row = _kb_or_404(kb_id) - stored_path, filename, byte_size = await _save_upload(file) + stored_path, filename, byte_size, content_hash = await _save_upload(file) # Tolerate pre-Phase-3 rows missing chunking_strategy/mode. kb_keys = kb_row.keys() if hasattr(kb_row, "keys") else () chunking_strategy = ( @@ -823,6 +863,7 @@ async def upload_kb_document( embedding_model = kb_row["embedding_model"], chunking_strategy = chunking_strategy, mode = mode, + content_hash = content_hash, ) @@ -835,7 +876,7 @@ async def upload_thread_document( from utils.rag.config import resolve_embedder # No chat_threads check — fresh threads aren't persisted until first run. - stored_path, filename, byte_size = await _save_upload(file) + stored_path, filename, byte_size, content_hash = await _save_upload(file) settings = _load_thread_settings(thread_id) embedder = settings.embedding_model or resolve_embedder( settings.mode, @@ -851,6 +892,7 @@ async def upload_thread_document( embedding_model = embedder, chunking_strategy = settings.chunking_strategy, mode = settings.mode, + content_hash = content_hash, ) diff --git a/studio/backend/storage/studio_db.py b/studio/backend/storage/studio_db.py index bb893a81de..8ee36fd151 100644 --- a/studio/backend/storage/studio_db.py +++ b/studio/backend/storage/studio_db.py @@ -262,6 +262,21 @@ def _ensure_schema(conn: sqlite3.Connection) -> None: conn.execute( "CREATE INDEX IF NOT EXISTS idx_rag_documents_thread_id ON rag_documents(thread_id)" ) + # content_hash: sha256 of the uploaded bytes, used to skip re-indexing a + # file that already exists in the same scope (kb_id / thread_id). + rag_documents_columns = { + row[1] for row in conn.execute("PRAGMA table_info(rag_documents)").fetchall() + } + if "content_hash" not in rag_documents_columns: + conn.execute("ALTER TABLE rag_documents ADD COLUMN content_hash TEXT") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_rag_documents_kb_hash " + "ON rag_documents(kb_id, content_hash)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_rag_documents_thread_hash " + "ON rag_documents(thread_id, content_hash)" + ) # kind: text|image|caption. linked_chunk_id pairs image↔caption (both null for text). conn.execute( """ diff --git a/studio/frontend/src/features/chat/hooks/use-thread-doc-uploads.ts b/studio/frontend/src/features/chat/hooks/use-thread-doc-uploads.ts index a9b184e801..aa3dc9b71f 100644 --- a/studio/frontend/src/features/chat/hooks/use-thread-doc-uploads.ts +++ b/studio/frontend/src/features/chat/hooks/use-thread-doc-uploads.ts @@ -123,7 +123,28 @@ export function useThreadDocUploads(): UseThreadDocUploadsResult { setChipScopeKeys((m) => ({ ...m, [localChipId]: scopeKey })); const uploadDocument = useRagStore.getState().uploadDocument; try { - const { documentId, jobId } = await uploadDocument(scope, file); + const { documentId, jobId, alreadyIndexed } = await uploadDocument( + scope, + file, + ); + if (alreadyIndexed) { + // Identical file already in this scope — no re-index. + setPendingDocs((prev) => + prev.map((d) => + d.id === localChipId + ? { ...d, status: "ready", documentId } + : d, + ), + ); + toast.info(`${file.name} is already indexed`); + if ( + scope?.kind === "thread" && + useChatRuntimeStore.getState().ragSource.kind === "off" + ) { + useChatRuntimeStore.getState().setRagSource({ kind: "thread" }); + } + return; + } setPendingDocs((prev) => prev.map((d) => d.id === localChipId diff --git a/studio/frontend/src/features/chat/shared-composer.tsx b/studio/frontend/src/features/chat/shared-composer.tsx index c27b40da7b..c44f312ae2 100644 --- a/studio/frontend/src/features/chat/shared-composer.tsx +++ b/studio/frontend/src/features/chat/shared-composer.tsx @@ -612,7 +612,27 @@ export function SharedComposer({ } const uploadDocument = useRagStore.getState().uploadDocument; try { - const { documentId, jobId } = await uploadDocument(scope, file); + const { documentId, jobId, alreadyIndexed } = await uploadDocument( + scope, + file, + ); + if (alreadyIndexed) { + setPendingDocs((prev) => + prev.map((d) => + d.id === localChipId + ? { ...d, status: "ready", documentId } + : d, + ), + ); + toast.info(`${file.name} is already indexed`); + if ( + scope?.kind === "thread" && + useChatRuntimeStore.getState().ragSource.kind === "off" + ) { + useChatRuntimeStore.getState().setRagSource({ kind: "thread" }); + } + return; + } setPendingDocs((prev) => prev.map((d) => d.id === localChipId diff --git a/studio/frontend/src/features/rag/api/rag-api.ts b/studio/frontend/src/features/rag/api/rag-api.ts index d22c9d2bf6..47f082453f 100644 --- a/studio/frontend/src/features/rag/api/rag-api.ts +++ b/studio/frontend/src/features/rag/api/rag-api.ts @@ -36,6 +36,9 @@ export interface UploadResponse { document_id: string; job_id: string; filename: string; + /** True when an identical file was already indexed in this scope; no + * new ingestion job was started and job_id is "". */ + already_indexed?: boolean; } export interface SearchHit { diff --git a/studio/frontend/src/features/rag/stores/rag-store.ts b/studio/frontend/src/features/rag/stores/rag-store.ts index 13f7b36923..f20ba22656 100644 --- a/studio/frontend/src/features/rag/stores/rag-store.ts +++ b/studio/frontend/src/features/rag/stores/rag-store.ts @@ -56,7 +56,11 @@ interface RagStoreState { uploadDocument: ( scope: { kind: "kb"; kbId: string } | { kind: "thread"; threadId: string }, file: File, - ) => Promise<{ documentId: string; jobId: string }>; + ) => Promise<{ + documentId: string; + jobId: string; + alreadyIndexed: boolean; + }>; deleteDocument: (documentId: string, scopeKey: string) => Promise; loadThreadIndexes: () => Promise; @@ -199,16 +203,25 @@ export const useRagStore = create((set, get) => ({ } else { void get().loadThreadDocuments(scope.threadId); } - get().subscribeJob(result.job_id, () => { - if (scope.kind === "kb") { - void get().loadKBDocuments(scope.kbId); - } else { - void get().loadThreadDocuments(scope.threadId); - } - }); - return { documentId: result.document_id, jobId: result.job_id, scopeKey } as { + // Identical file already indexed in this scope: no job to track. + if (!result.already_indexed && result.job_id) { + get().subscribeJob(result.job_id, () => { + if (scope.kind === "kb") { + void get().loadKBDocuments(scope.kbId); + } else { + void get().loadThreadDocuments(scope.threadId); + } + }); + } + return { + documentId: result.document_id, + jobId: result.job_id, + alreadyIndexed: result.already_indexed ?? false, + scopeKey, + } as { documentId: string; jobId: string; + alreadyIndexed: boolean; }; },