diff --git a/docs/branch-review-records/5349150e82f255e42866975fa7aa5e9fe76601433ebc427c75226677c5bef854.record.md b/docs/branch-review-records/5349150e82f255e42866975fa7aa5e9fe76601433ebc427c75226677c5bef854.record.md new file mode 100644 index 0000000000..4df086eed5 --- /dev/null +++ b/docs/branch-review-records/5349150e82f255e42866975fa7aa5e9fe76601433ebc427c75226677c5bef854.record.md @@ -0,0 +1 @@ +| 2026-10-17 | claude/rag-ingestion-hardening | 2a9d44e0015358e4e7d197e4cf15f83df0a9ce35 | RAG retrieval/summary/cache generation isolation and ingestion worker/edge hardening | resolved | typecheck:internal, eslint, check:diff-integrity, check:migration-immutability, check:owner-scope, check:rag:fixtures, eval:rag:offline (26 files/979 tests), eval:rag:adversarial:offline (25 tests), vitest (8 suites/254 tests) | diff --git a/docs/branch-review-records/94a7d7908ea7ff58edaecb6ebd9fd41f72d422ad6d345190046cfd5ed2d4372d.record.md b/docs/branch-review-records/94a7d7908ea7ff58edaecb6ebd9fd41f72d422ad6d345190046cfd5ed2d4372d.record.md new file mode 100644 index 0000000000..0e52d145d2 --- /dev/null +++ b/docs/branch-review-records/94a7d7908ea7ff58edaecb6ebd9fd41f72d422ad6d345190046cfd5ed2d4372d.record.md @@ -0,0 +1 @@ +| 2026-10-17 | claude/rag-ingestion-hardening | b8278363b9e55ca218133ac52f6191a2edf60502 | RAG and Ingestion review follow-ups | Resolved PR review findings, repaired bot-applied regressions, and restored rag.ts maintainability budget | npm run typecheck:internal; node scripts/check-maintainability-budgets.mjs; npx eslint; npx vitest run (11 suites) | diff --git a/src/app/api/documents/route.ts b/src/app/api/documents/route.ts index 4e426215e0..44d9c4c202 100644 --- a/src/app/api/documents/route.ts +++ b/src/app/api/documents/route.ts @@ -17,6 +17,7 @@ import { redactNonOwnedDocumentFields, withOwnerReadScope, } from "@/lib/public-api-access"; +import { committedIndexGeneration, isCommittedGenerationMetadata } from "@/lib/reindex-pipeline"; import { parseRequestQuery, queryBoolean, queryInteger } from "@/lib/validation/query"; export const runtime = "nodejs"; @@ -35,6 +36,11 @@ const PUBLIC_DOCUMENT_LIST_COLUMNS = [ "updated_at", ].join(","); +// Public rows are read with `metadata` only so the hidden-label and committed-generation filters can +// run server-side; `projectPublicFields` strips it (and anything else not in the public list) before +// the response is built. +const PUBLIC_DOCUMENT_QUERY_COLUMNS = `${PUBLIC_DOCUMENT_LIST_COLUMNS},metadata`; + const DOCUMENT_LIST_COLUMNS = [ "id", "owner_id", @@ -68,6 +74,8 @@ const PUBLIC_LABEL_LIST_COLUMNS = [ "updated_at", ].join(","); +const PUBLIC_LABEL_QUERY_COLUMNS = `${PUBLIC_LABEL_LIST_COLUMNS},metadata`; + const LABEL_LIST_COLUMNS = [ "id", "document_id", @@ -83,6 +91,8 @@ const LABEL_LIST_COLUMNS = [ const PUBLIC_SUMMARY_LIST_COLUMNS = ["id", "document_id", "summary", "clinical_specifics", "generated_at"].join(","); +const PUBLIC_SUMMARY_QUERY_COLUMNS = `${PUBLIC_SUMMARY_LIST_COLUMNS},metadata`; + const SUMMARY_LIST_COLUMNS = [ "id", "document_id", @@ -180,7 +190,7 @@ export async function GET(request: Request) { } const effectiveIncludeMeta = access.authenticated ? includeMeta : false; - const listColumns = access.authenticated ? DOCUMENT_LIST_COLUMNS : PUBLIC_DOCUMENT_LIST_COLUMNS; + const listColumns = access.authenticated ? DOCUMENT_LIST_COLUMNS : PUBLIC_DOCUMENT_QUERY_COLUMNS; let query = withOwnerReadScope(supabase.from("documents").select(listColumns, { count: "exact" }), access.ownerId) .order("created_at", { ascending: false }) .range(offset, offset + limit - 1); @@ -210,13 +220,16 @@ export async function GET(request: Request) { .filter((document) => !ownedDocumentIds.has(document.id)) .map((document) => document.id); const documents = rawDocuments.map((document) => redactNonOwnedDocumentFields(document, access.ownerId)); + const responseDocuments = documents.map((document) => + ownedDocumentIds.has(document.id) ? document : projectPublicFields(document, PUBLIC_DOCUMENT_LIST_COLUMNS), + ); const documentIds = documents.map((document) => document.id); const indexing = indexingState(documents); const pagination = offsetPagination({ limit, offset, pageLength: documents.length, count }); if (documentIds.length === 0 || !effectiveIncludeMeta) { - return documentsResponse({ documents, pagination }, indexing); + return documentsResponse({ documents: responseDocuments, pagination }, indexing); } const ownedIds = [...ownedDocumentIds]; @@ -234,7 +247,7 @@ export async function GET(request: Request) { publicLabelsPromises.push( supabase .from("document_labels") - .select(PUBLIC_LABEL_LIST_COLUMNS) + .select(PUBLIC_LABEL_QUERY_COLUMNS) .in("document_id", publicDocumentIds.slice(i, i + 100)), ); } @@ -244,7 +257,8 @@ export async function GET(request: Request) { supabase .from("document_summaries") .select(SUMMARY_LIST_COLUMNS) - .in("document_id", ownedIds.slice(i, i + 100)), + .in("document_id", ownedIds.slice(i, i + 100)) + .order("generated_at", { ascending: false }), ); } const publicSummariesPromises = []; @@ -252,8 +266,9 @@ export async function GET(request: Request) { publicSummariesPromises.push( supabase .from("document_summaries") - .select(PUBLIC_SUMMARY_LIST_COLUMNS) - .in("document_id", publicDocumentIds.slice(i, i + 100)), + .select(PUBLIC_SUMMARY_QUERY_COLUMNS) + .in("document_id", publicDocumentIds.slice(i, i + 100)) + .order("generated_at", { ascending: false }), ); } @@ -273,12 +288,35 @@ export async function GET(request: Request) { if (res.error) throw new Error(res.error.message); } + const committedGenerationByDocument = new Map( + rawDocuments.map((doc) => [doc.id, committedIndexGeneration((doc as { metadata?: unknown }).metadata)]), + ); + const isHiddenLabelRow = (row: Record) => { + if (!row.metadata || typeof row.metadata !== "object" || Array.isArray(row.metadata)) return false; + const meta = row.metadata as Record; + return ( + meta.hidden === true || + meta.hidden === "true" || + meta.review_status === "hidden" || + meta.label_review_status === "hidden" + ); + }; + const labelsByDocument = new Map(); const labelRows = parseListRows( [...ownedLabelsResults.flatMap((res) => res.data ?? []), ...publicLabelsResults.flatMap((res) => res.data ?? [])], labelListRowSchema, ); for (const label of labelRows) { + if (isHiddenLabelRow(label)) continue; + if ( + !isCommittedGenerationMetadata({ + rowMetadata: label.metadata, + committedGeneration: committedGenerationByDocument.get(label.document_id) ?? null, + }) + ) { + continue; + } const existing = labelsByDocument.get(label.document_id) ?? []; existing.push( ownedDocumentIds.has(label.document_id) ? label : projectPublicFields(label, PUBLIC_LABEL_LIST_COLUMNS), @@ -292,19 +330,36 @@ export async function GET(request: Request) { ], summaryListRowSchema, ); - const summariesByDocument = new Map( - summaryRows.map((summary) => [ - summary.document_id, - ownedDocumentIds.has(summary.document_id) ? summary : projectPublicFields(summary, PUBLIC_SUMMARY_LIST_COLUMNS), - ]), - ); + const summariesByDocument = new Map(); + for (const summary of summaryRows) { + if ( + !isCommittedGenerationMetadata({ + rowMetadata: summary.metadata, + committedGeneration: committedGenerationByDocument.get(summary.document_id) ?? null, + }) + ) { + continue; + } + const generatedAt = typeof summary.generated_at === "string" ? summary.generated_at : ""; + const existing = summariesByDocument.get(summary.document_id); + if (!existing || (generatedAt && (!existing.generatedAt || generatedAt > existing.generatedAt))) { + summariesByDocument.set(summary.document_id, { + row: ownedDocumentIds.has(summary.document_id) + ? summary + : projectPublicFields(summary, PUBLIC_SUMMARY_LIST_COLUMNS), + generatedAt, + }); + } + } return documentsResponse( { documents: documents.map((document) => ({ - ...document, + ...(ownedDocumentIds.has(document.id) + ? document + : projectPublicFields(document, PUBLIC_DOCUMENT_LIST_COLUMNS)), labels: labelsByDocument.get(document.id) ?? [], - summary: summariesByDocument.get(document.id) ?? null, + summary: summariesByDocument.get(document.id)?.row ?? null, })), pagination, }, diff --git a/src/app/api/ingestion/quality/route.ts b/src/app/api/ingestion/quality/route.ts index 313bb563bf..793ca9253c 100644 --- a/src/app/api/ingestion/quality/route.ts +++ b/src/app/api/ingestion/quality/route.ts @@ -214,13 +214,29 @@ function buildReviewItems(args: { const documentStages = stagesByDocument.get(document.id) ?? []; const documentPages = pagesByDocument.get(document.id) ?? []; const documentImages = imagesByDocument.get(document.id) ?? []; - const failedJob = documentJobs.find((job) => job.status === "failed"); + const latestJob = documentJobs[0]; + const failedJob = + latestJob?.status === "failed" + ? latestJob + : document.status === "failed" + ? documentJobs.find((job) => job.status === "failed") + : undefined; const ingestionJobIds = new Set(documentJobs.map((job) => job.id)); - const failedOcrStage = documentStages.find( - (stage) => - stage.stage_status === "failed" && - includesAny(`${stage.stage_name ?? ""} ${stage.error_message ?? ""}`, ["ocr", "extract"]), - ); + const latestStagesByName = new Map(); + for (const stage of documentStages) { + const key = stage.stage_name ?? stage.id; + if (!latestStagesByName.has(key)) latestStagesByName.set(key, stage); + } + const failedOcrStage = Array.from(latestStagesByName.values()).find((stage) => { + if (stage.stage_status !== "failed") return false; + if (!includesAny(`${stage.stage_name ?? ""} ${stage.error_message ?? ""}`, ["ocr", "extract"])) return false; + if (latestJob?.status !== "completed") return true; + if (stage.job_id && ingestionJobIds.has(stage.job_id)) { + return stage.job_id === latestJob.id; + } + const stageTimestamp = stage.finished_at ?? stage.started_at; + return !latestJob.updated_at || !stageTimestamp || stageTimestamp >= latestJob.updated_at; + }); const warnings = Array.isArray(metadata.extraction_warnings) ? metadata.extraction_warnings.map(String) : []; const ocrWarning = unique([...warnings, ...issues]).find((warning) => includesAny(warning, ["ocr", "scanned"])); const lowTextPages = documentPages.filter((page) => String(page.text ?? "").trim().length < 80); diff --git a/src/lib/document-enrichment.ts b/src/lib/document-enrichment.ts index eed76ce208..3aaceeb61a 100644 --- a/src/lib/document-enrichment.ts +++ b/src/lib/document-enrichment.ts @@ -31,6 +31,7 @@ import type { SupabaseClient } from "@supabase/supabase-js"; import { z } from "zod"; import { classifyDocumentOrganization } from "@/lib/document-organization"; import { env } from "@/lib/env"; +import { committedIndexGeneration, isCommittedGenerationMetadata } from "@/lib/reindex-pipeline"; import { PUBLIC_OWNER_FILTER_SENTINEL, retrievalAccessScopeForArgs, @@ -1074,10 +1075,21 @@ export async function fetchRelatedDocumentMetadata(args: { rpcData = [...merged.values()]; } + const isHiddenLabel = (label: { metadata?: unknown }) => { + if (!label.metadata || typeof label.metadata !== "object" || Array.isArray(label.metadata)) return false; + const meta = label.metadata as Record; + return ( + meta.hidden === true || + meta.hidden === "true" || + meta.review_status === "hidden" || + meta.label_review_status === "hidden" + ); + }; + if (!rpcError) { return ((rpcData ?? []) as RelatedDocumentMetadataRow[]).map((row) => ({ document_id: row.document_id, - labels: row.labels ?? [], + labels: (row.labels ?? []).filter((label) => !isHiddenLabel(label)), summary: row.summary ?? null, })); } @@ -1088,8 +1100,11 @@ export async function fetchRelatedDocumentMetadata(args: { .in("document_id", args.documentIds); let summariesQuery = args.supabase .from("document_summaries") - .select("document_id,owner_id,summary") + .select("document_id,owner_id,summary,generated_at,metadata") .in("document_id", args.documentIds); + if (typeof (summariesQuery as { order?: unknown }).order === "function") { + summariesQuery = summariesQuery.order("generated_at", { ascending: false }); + } if (accessScope.ownerId && accessScope.includePublic) { labelsQuery = labelsQuery.or(`owner_id.eq.${accessScope.ownerId},owner_id.is.null`); @@ -1111,22 +1126,67 @@ export async function fetchRelatedDocumentMetadata(args: { args.signal?.throwIfAborted(); const metadataError = labelsResult.error ?? summariesResult.error; if (metadataError) throw metadataError; - const labels = (labelsResult.data ?? []) as DocumentLabel[]; - const summaries = (summariesResult.data ?? []) as Pick[]; + const rawLabels = (labelsResult.data ?? []) as DocumentLabel[]; + const rawSummaries = (summariesResult.data ?? []) as Array< + Pick & { generated_at?: string | null; metadata?: unknown } + >; + const hasGenerationStampedMetadata = + rawLabels.some((label) => Boolean(committedIndexGeneration(label.metadata))) || + rawSummaries.some((summary) => Boolean(committedIndexGeneration(summary.metadata))); + const committedGenerationByDocument = new Map(); + if (hasGenerationStampedMetadata) { + let documentsQuery = args.supabase.from("documents").select("id,metadata").in("id", args.documentIds); + if (accessScope.ownerId && accessScope.includePublic) { + documentsQuery = documentsQuery.or(`owner_id.eq.${accessScope.ownerId},owner_id.is.null`); + } else if (accessScope.ownerId) { + documentsQuery = documentsQuery.eq("owner_id", accessScope.ownerId); + } else { + documentsQuery = documentsQuery.is("owner_id", null); + } + if (args.signal) { + documentsQuery = documentsQuery.abortSignal(args.signal); + } + const documentsResult = await documentsQuery; + args.signal?.throwIfAborted(); + if (documentsResult?.error) throw documentsResult.error; + for (const doc of (documentsResult?.data ?? []) as Array<{ id: string; metadata?: unknown }>) { + committedGenerationByDocument.set(doc.id, committedIndexGeneration(doc.metadata)); + } + } + const labels = rawLabels.filter( + (label) => + !isHiddenLabel(label) && + isCommittedGenerationMetadata({ + rowMetadata: label.metadata, + committedGeneration: committedGenerationByDocument.get(label.document_id) ?? null, + }), + ); + const summaries = rawSummaries.filter((summary) => + isCommittedGenerationMetadata({ + rowMetadata: summary.metadata, + committedGeneration: committedGenerationByDocument.get(summary.document_id) ?? null, + }), + ); const labelsByDocument = new Map(); - const summariesByDocument = new Map(); + const summariesByDocument = new Map(); for (const label of labels) { const existing = labelsByDocument.get(label.document_id) ?? []; existing.push(label); labelsByDocument.set(label.document_id, existing); } - for (const summary of summaries) summariesByDocument.set(summary.document_id, summary.summary ?? null); + for (const summary of summaries) { + const generatedAt = typeof summary.generated_at === "string" ? summary.generated_at : ""; + const existing = summariesByDocument.get(summary.document_id); + if (!existing || (generatedAt && (!existing.generatedAt || generatedAt > existing.generatedAt))) { + summariesByDocument.set(summary.document_id, { summary: summary.summary ?? null, generatedAt }); + } + } return args.documentIds.map((documentId) => ({ document_id: documentId, labels: labelsByDocument.get(documentId) ?? [], - summary: summariesByDocument.get(documentId) ?? null, + summary: summariesByDocument.get(documentId)?.summary ?? null, })); } diff --git a/src/lib/env.ts b/src/lib/env.ts index 55a3e57816..ee1433c969 100644 --- a/src/lib/env.ts +++ b/src/lib/env.ts @@ -80,11 +80,11 @@ const envSchema = z // Railway deploy webhook -> chat forwarder. Railway only lets you configure a // target URL (no custom headers), so this secret travels as `?token=` in the // configured URL and is compared constant-time. Min 16 chars. - RAILWAY_WEBHOOK_SECRET: z.string().min(16).optional(), + RAILWAY_WEBHOOK_SECRET: z.preprocess(coerceBlankEnv, z.string().min(16).optional()), // Supabase Database Webhook -> ingestion enqueue. Supabase webhooks DO allow // custom headers, so this secret is sent as `Authorization: Bearer` (or the // `x-webhook-secret` header) and compared constant-time. Min 16 chars. - SUPABASE_INGESTION_WEBHOOK_SECRET: z.string().min(16).optional(), + SUPABASE_INGESTION_WEBHOOK_SECRET: z.preprocess(coerceBlankEnv, z.string().min(16).optional()), // Optional outbound chat destinations shared by every /api/webhooks/* forwarder // and the CI-failure GitHub workflow. Set either, both, or neither; a receiver // with no destination configured accepts the event and reports it undelivered. diff --git a/src/lib/ingestion-enqueue.ts b/src/lib/ingestion-enqueue.ts index 846bab58fa..a9bd865d10 100644 --- a/src/lib/ingestion-enqueue.ts +++ b/src/lib/ingestion-enqueue.ts @@ -73,7 +73,10 @@ export async function enqueueDocumentReindexJob(args: { .eq("id", document.id); // Owner scope is the app's single tenancy layer (RLS is bypassed by the // service-role client); scope the write to the owning row when we have it. - const { error: updateError } = await (ownerId ? queueUpdate.eq("owner_id", ownerId) : queueUpdate); + const applyOwnerFilter = T; is?: (col: string, val: null) => T }>( + query: T, + ) => (ownerId ? query.eq("owner_id", ownerId) : typeof query.is === "function" ? query.is("owner_id", null) : query); + const { error: updateError } = await applyOwnerFilter(queueUpdate); if (updateError) throw new Error(updateError.message); const { data: job, error: jobError } = await supabase @@ -102,17 +105,24 @@ export async function enqueueDocumentReindexJob(args: { if (jobError.code === "23505") { const { data: competingJobs, error: competingJobsError } = await supabase .from("ingestion_jobs") - .select("id") + .select("id,status") .eq("document_id", document.id) .in("status", ["pending", "processing"]) .limit(1); - if (!competingJobsError && (competingJobs?.length ?? 0) === 0) { + const hasCompetingProcessingJob = + !atomicReindex && + Boolean(document.status) && + (competingJobs ?? []).some((row) => (row as { status?: unknown }).status === "processing"); + if (!competingJobsError && ((competingJobs?.length ?? 0) === 0 || hasCompetingProcessingJob)) { + const payloadToRestore = hasCompetingProcessingJob + ? { ...rollbackDocumentPayload, status: "processing", error_message: null } + : rollbackDocumentPayload; const rollback = supabase .from("documents") - .update(rollbackDocumentPayload) + .update(payloadToRestore) .eq("id", document.id) .eq("updated_at", rollbackFence); - await (ownerId ? rollback.eq("owner_id", ownerId) : rollback); + await applyOwnerFilter(rollback); } return { outcome: "already_active" }; } diff --git a/src/lib/rag/rag-cache.ts b/src/lib/rag/rag-cache.ts index d5e373651d..5869430730 100644 --- a/src/lib/rag/rag-cache.ts +++ b/src/lib/rag/rag-cache.ts @@ -413,6 +413,7 @@ export function answerCacheLookupAllowedForRequest(args: AnswerCachePolicyArgs, args.reviewedPolicyRequest?.state !== "reviewed" && answerCacheAllowedForSourcePolicyConflicts(args.sourcePolicyConflicts) && answerCacheAllowedForOwner(args.ownerId) && + isRagCacheAccessAllowed(args) && !args.skipCache && env.RAG_ANSWER_CACHE_TTL_MS > 0 ); @@ -682,6 +683,11 @@ export async function getCachedAnswer( return null; } const indexingVersion = options?.indexingVersionAtRequestStart ?? (await cacheIndexingVersion(args)); + if (isUnavailableIndexingStamp(cached.indexingVersion)) { + answerCache.delete(key); + return null; + } + if (isUnavailableIndexingStamp(indexingVersion)) return null; if (cached.indexingVersion !== indexingVersion) { answerCache.delete(key); return null; @@ -776,6 +782,7 @@ export async function setCachedAnswer( throwIfAborted(descriptor.signal); if (invalidationEpochChanged(descriptor.invalidationOwnerId, invalidationEpochAtStart)) return; if (indexingVersionAtRetrievalStart && indexingVersion !== indexingVersionAtRetrievalStart) return; + if (isUnavailableIndexingStamp(indexingVersion)) return; answerCache.set(descriptor.localCacheKey, { expiresAt: Date.now() + env.RAG_ANSWER_CACHE_TTL_MS, answer: capturedAnswer, @@ -975,6 +982,7 @@ export async function getCachedSearch( options?: { indexingVersionAtRequestStart?: string | null }, ): Promise<{ results: SearchResult[]; telemetry: SearchTelemetry } | null> { throwIfAborted(args.signal); + if (args.ragQueryPlanMode === "shadow" && !isIssuedRagShadowControl(args)) return null; if (!isSearchCacheEnabled(args) || !isRagCacheAccessAllowed(args)) return null; const key = scopedSearchCacheKey(args, queryClass, queryVariants); @@ -986,6 +994,11 @@ export async function getCachedSearch( } const indexingVersion = options?.indexingVersionAtRequestStart ?? (await cacheIndexingVersion(args)); throwIfAborted(args.signal); + if (isUnavailableIndexingStamp(cached.indexingVersion)) { + searchCache.delete(key); + return null; + } + if (isUnavailableIndexingStamp(indexingVersion)) return null; if (cached.indexingVersion !== indexingVersion) { searchCache.delete(key); return null; @@ -1040,9 +1053,12 @@ export async function setCachedSearch( const descriptor = createSiteAwareSearchWriteDescriptor(args, capturedTelemetry.query_class, queryVariants); if (!cacheWriteProofAllows(descriptor.args, "search", capturedResults, proof)) return; const indexingVersionAtRetrievalStart = options?.indexingVersionAtRetrievalStart; + const invalidationEpochAtStart = captureInvalidationEpoch(descriptor.invalidationOwnerId); const indexingVersion = await cacheIndexingVersion(descriptor.args, { forceRefresh: true }); throwIfAborted(descriptor.signal); + if (invalidationEpochChanged(descriptor.invalidationOwnerId, invalidationEpochAtStart)) return; if (indexingVersionAtRetrievalStart && indexingVersion !== indexingVersionAtRetrievalStart) return; + if (isUnavailableIndexingStamp(indexingVersion)) return; searchCache.set(descriptor.localCacheKey, { expiresAt: Date.now() + env.RAG_SEARCH_CACHE_TTL_MS, results: capturedResults, @@ -1055,6 +1071,10 @@ export async function setCachedSearch( if (!oldestKey) break; searchCache.delete(oldestKey); } + if (invalidationEpochChanged(descriptor.invalidationOwnerId, invalidationEpochAtStart)) { + searchCache.delete(descriptor.localCacheKey); + return; + } setSharedSiteAwareCachedSearch(descriptor, capturedResults, capturedTelemetry, indexingVersion, proof); return; } @@ -1062,8 +1082,10 @@ export async function setCachedSearch( const cacheTelemetry = normalizeCacheStorageTelemetry(telemetry, isIssuedRagShadowControl(args)); const clonedResults = cloneSearchResults(results); + const invalidationEpochAtStart = captureInvalidationEpoch(args.ownerId); const indexingVersion = await cacheIndexingVersion(args, { forceRefresh: true }); throwIfAborted(args.signal); + if (invalidationEpochChanged(args.ownerId, invalidationEpochAtStart)) return; if (options?.indexingVersionAtRetrievalStart && indexingVersion !== options.indexingVersionAtRetrievalStart) return; if (isUnavailableIndexingStamp(indexingVersion)) return; const key = scopedSearchCacheKey(args, telemetry.query_class, queryVariants); @@ -1080,6 +1102,10 @@ export async function setCachedSearch( if (!oldestKey) break; searchCache.delete(oldestKey); } + if (invalidationEpochChanged(args.ownerId, invalidationEpochAtStart)) { + searchCache.delete(key); + return; + } setSharedCachedSearch( args, clonedResults, @@ -1190,6 +1216,10 @@ export async function cacheIndexingVersion( value = `${ragDeepMemoryVersion}:${INDEXING_STAMP_UNAVAILABLE}`; } throwIfAborted(args.signal); + if (isUnavailableIndexingStamp(value)) { + cacheIndexingVersionCache.delete(cacheKey); + return value; + } writeBoundedExpiringCacheEntry( cacheIndexingVersionCache, cacheKey, @@ -1210,9 +1240,11 @@ export async function getSharedCachedSearch( | null > { throwIfAborted(args.signal); + if (args.ragQueryPlanMode === "shadow" && !isIssuedRagShadowControl(args)) return null; if (args.skipCache || env.RAG_SEARCH_CACHE_TTL_MS <= 0 || !isRagCacheAccessAllowed(args)) return null; const normalizedQuery = retrievalPlanCacheQuery(args, queryClass, queryVariants); const indexingVersion = options?.indexingVersionAtRequestStart ?? (await cacheIndexingVersion(args)); + if (isUnavailableIndexingStamp(indexingVersion)) return { kind: "miss", reason: "cache_lookup_error" }; try { let query = sharedCacheSelector(createAdminClient(), "search", args, indexingVersion, normalizedQuery); if (args.signal) query = query.abortSignal(args.signal); @@ -1328,6 +1360,7 @@ export async function getSharedCachedAnswer( return null; try { const indexingVersion = options?.indexingVersionAtRequestStart ?? (await cacheIndexingVersion(args)); + if (isUnavailableIndexingStamp(indexingVersion)) return null; const { data, error } = await sharedCacheSelector( createAdminClient(), "answer", @@ -1697,7 +1730,7 @@ export function invalidateRagCachesForDocumentMutation( options: { affectsPublicCorpus?: boolean } = {}, ) { invalidateRagCachesForOwner(ownerId); - if (options.affectsPublicCorpus !== false) invalidateAnonymousSharedRagCaches(); + if (options.affectsPublicCorpus !== false && ownerId !== "anonymous") invalidateAnonymousSharedRagCaches(); } export { diff --git a/src/lib/rag/rag-candidate-sources.ts b/src/lib/rag/rag-candidate-sources.ts index 4193e4dca1..3c5c0219da 100644 --- a/src/lib/rag/rag-candidate-sources.ts +++ b/src/lib/rag/rag-candidate-sources.ts @@ -937,6 +937,7 @@ type DocumentLookupChunkRow = { content: string; retrieval_synopsis?: string | null; image_ids: string[] | null; + index_generation_id?: string | null; text_rank?: number | null; }; @@ -1003,6 +1004,7 @@ function documentLookupChunkScore(chunk: DocumentLookupChunkRow, terms: string[] async function fetchBestDocumentLookupChunks(args: { supabase: ReturnType; documentIds: string[]; + committedGenerationByDocument?: ReadonlyMap; query: string; limit: number; ownerId?: string; @@ -1011,6 +1013,12 @@ async function fetchBestDocumentLookupChunks(args: { signal?: AbortSignal; }) { const terms = documentLookupChunkTerms(args.query); + const matchesCommittedGeneration = (chunk: DocumentLookupChunkRow) => { + if (!chunk.index_generation_id || !args.committedGenerationByDocument?.has(chunk.document_id)) { + return true; + } + return chunk.index_generation_id === args.committedGenerationByDocument.get(chunk.document_id); + }; const { data: rpcChunks, error: rpcError } = await callVersionedRetrievalRpc( args.supabase, "match_document_lookup_chunks_text_v2", @@ -1025,6 +1033,7 @@ async function fetchBestDocumentLookupChunks(args: { ); if (!rpcError && rpcChunks?.length) { const ranked = (rpcChunks as DocumentLookupChunkRow[]) + .filter(matchesCommittedGeneration) .map((chunk) => ({ chunk, score: Math.max(Number(chunk.text_rank ?? 0), documentLookupChunkScore(chunk, terms)), @@ -1047,7 +1056,7 @@ async function fetchBestDocumentLookupChunks(args: { const baseQuery = args.supabase .from("document_chunks") .select( - "id,document_id,page_number,chunk_index,section_heading,section_path,heading_level,parent_heading,anchor_id,content,retrieval_synopsis,image_ids", + "id,document_id,page_number,chunk_index,section_heading,section_path,heading_level,parent_heading,anchor_id,content,retrieval_synopsis,image_ids,index_generation_id", ) .in("document_id", args.documentIds) .limit(Math.max(args.limit * 4, 24)); @@ -1058,6 +1067,7 @@ async function fetchBestDocumentLookupChunks(args: { if (!matchedError && matchedChunks?.length) { const ranked = (matchedChunks as DocumentLookupChunkRow[]) + .filter(matchesCommittedGeneration) .map((chunk) => ({ chunk, score: documentLookupChunkScore(chunk, terms) })) .filter((item) => item.score > 0) .sort((a, b) => b.score - a.score || a.chunk.chunk_index - b.chunk.chunk_index) @@ -1068,14 +1078,17 @@ async function fetchBestDocumentLookupChunks(args: { const fallbackQuery = args.supabase .from("document_chunks") .select( - "id,document_id,page_number,chunk_index,section_heading,section_path,heading_level,parent_heading,anchor_id,content,retrieval_synopsis,image_ids", + "id,document_id,page_number,chunk_index,section_heading,section_path,heading_level,parent_heading,anchor_id,content,retrieval_synopsis,image_ids,index_generation_id", ) .in("document_id", args.documentIds) .order("chunk_index", { ascending: true }) - .limit(args.limit); + .limit(Math.max(args.limit * 4, 24)); const { data: fallbackChunks, error: fallbackError } = await fallbackQuery; if (fallbackError || !fallbackChunks?.length) return { chunks: [] as DocumentLookupChunkRow[], terms }; - return { chunks: fallbackChunks as DocumentLookupChunkRow[], terms }; + return { + chunks: (fallbackChunks as DocumentLookupChunkRow[]).filter(matchesCommittedGeneration).slice(0, args.limit), + terms, + }; } /** Fetch document title alias rows. */ @@ -1216,9 +1229,13 @@ export async function searchDocumentLookupFastPath(args: { : args.queryVariants?.length ? args.query : (primaryVariant ?? args.query); + const committedGenerationByDocument = new Map( + rankedDocuments.map((item) => [item.document.id, committedIndexGeneration(item.document.metadata)] as const), + ); const { chunks, terms } = await fetchBestDocumentLookupChunks({ supabase: args.supabase, documentIds: rankedDocuments.map((item) => item.document.id), + committedGenerationByDocument, query: chunkQuery, limit: Math.max(args.matchCount, rankedDocuments.length * 4), ownerId: args.ownerId, diff --git a/src/lib/rag/rag-document-summary-context.ts b/src/lib/rag/rag-document-summary-context.ts index b16213f651..9726bcb6e0 100644 --- a/src/lib/rag/rag-document-summary-context.ts +++ b/src/lib/rag/rag-document-summary-context.ts @@ -1,3 +1,4 @@ +import { committedIndexGeneration } from "@/lib/reindex-pipeline"; import { createAdminClient } from "@/lib/supabase/admin"; function throwIfAborted(signal?: AbortSignal) { @@ -9,28 +10,72 @@ function throwIfAborted(signal?: AbortSignal) { export async function loadDocumentSummaryContext(documentId: string, ownerId?: string, signal?: AbortSignal) { throwIfAborted(signal); const supabase = createAdminClient(); - let documentQuery = supabase.from("documents").select("id,title,file_name,metadata").eq("id", documentId); + let documentQuery = supabase.from("documents").select("id,title,file_name,status,metadata").eq("id", documentId); - documentQuery = ownerId ? documentQuery.eq("owner_id", ownerId) : documentQuery.is("owner_id", null); + if (ownerId) { + documentQuery = + typeof (documentQuery as { or?: unknown }).or === "function" + ? documentQuery.or(`owner_id.eq.${ownerId},and(owner_id.is.null,metadata->>public_corpus.eq.true)`) + : documentQuery.eq("owner_id", ownerId); + } else { + documentQuery = documentQuery.is("owner_id", null); + } if (signal) documentQuery = documentQuery.abortSignal(signal); const { data: document, error: documentError } = await documentQuery.maybeSingle(); throwIfAborted(signal); if (documentError) throw new Error(documentError.message); if (!document) throw new Error("Document not found."); + const documentStatus = (document as { status?: unknown }).status; + if (typeof documentStatus === "string" && documentStatus !== "indexed") { + return { document, chunks: [] }; + } + const committedGeneration = committedIndexGeneration((document as { metadata?: unknown }).metadata); let chunksQuery = supabase .from("document_chunks") .select( "id,document_id,page_number,chunk_index,section_heading,content,retrieval_synopsis,image_ids,index_generation_id", ) - .eq("document_id", documentId) - .order("chunk_index", { ascending: true }) - .limit(40); + .eq("document_id", documentId); + if (committedGeneration) { + chunksQuery = chunksQuery.eq("index_generation_id", committedGeneration); + } else if (typeof (chunksQuery as { is?: unknown }).is === "function") { + chunksQuery = (chunksQuery as typeof chunksQuery & { is: (column: string, value: null) => typeof chunksQuery }).is( + "index_generation_id", + null, + ); + } + chunksQuery = chunksQuery.order("chunk_index", { ascending: true }).limit(40); if (signal) chunksQuery = chunksQuery.abortSignal(signal); const { data: chunks, error } = await chunksQuery; throwIfAborted(signal); if (error) throw new Error(error.message); - return { document, chunks: chunks ?? [] }; + const committedChunks = (chunks ?? []).filter( + (chunk) => + !chunk.index_generation_id || (Boolean(committedGeneration) && chunk.index_generation_id === committedGeneration), + ); + return { document, chunks: committedChunks }; +} + +export function sanitizeUnsupportedSummaryAnswer< + T extends { + grounded?: boolean; + confidence?: string; + bestSource?: unknown; + supportedClaims?: unknown; + evidenceAssessments?: unknown; + smartPanel?: Record | null; + }, +>(answer: T): T { + if (!answer.grounded || answer.confidence === "unsupported") { + answer.bestSource = null; + answer.supportedClaims = undefined; + answer.evidenceAssessments = undefined; + if (answer.smartPanel) { + answer.smartPanel = { ...answer.smartPanel, bestSource: null }; + } + } + return answer; } diff --git a/src/lib/rag/rag-hydration.ts b/src/lib/rag/rag-hydration.ts index 58126a62f5..1b09d600d7 100644 --- a/src/lib/rag/rag-hydration.ts +++ b/src/lib/rag/rag-hydration.ts @@ -295,13 +295,24 @@ async function attachIndexQualityMetadata( .from("document_index_quality") .select("document_id,owner_id,quality_score,extraction_quality,metrics,issues,updated_at") .in("document_id", missingDocumentIds); - if (ownerId) query = query.eq("owner_id", ownerId); + if (ownerId) { + query = + typeof query.or === "function" + ? query.or(`owner_id.eq.${ownerId},owner_id.is.null`) + : query.eq("owner_id", ownerId); + } else if (typeof query.is === "function") { + query = query.is("owner_id", null); + } if (signal) query = query.abortSignal(signal); const { data, error } = await query; signal?.throwIfAborted(); if (error) return results; for (const documentId of missingDocumentIds) cache.indexQuality.set(documentId, null); - for (const row of data ?? []) cache.indexQuality.set(row.document_id, row as SearchResult["indexing_quality"]); + for (const row of data ?? []) { + const { owner_id: rowOwnerId, ...qualityRow } = row; + if (rowOwnerId && rowOwnerId !== ownerId) continue; + cache.indexQuality.set(row.document_id, qualityRow as SearchResult["indexing_quality"]); + } return withCachedIndexQuality(results, cache); } catch { signal?.throwIfAborted(); diff --git a/src/lib/rag/rag.ts b/src/lib/rag/rag.ts index 966c77b8c1..d5f934a1aa 100644 --- a/src/lib/rag/rag.ts +++ b/src/lib/rag/rag.ts @@ -23,7 +23,7 @@ import { withRagProgrammeRollout, answerContractForRollout, adaptiveAnswerRender import { loadReviewedPolicyRequest } from "@/lib/rag/rag-reviewed-policy-input"; import { createGenerationDegradationRecorder } from "@/lib/rag/rag-generation-degradation"; import { createAdminClient } from "@/lib/supabase/admin"; -import { loadDocumentSummaryContext } from "@/lib/rag/rag-document-summary-context"; +import { loadDocumentSummaryContext, sanitizeUnsupportedSummaryAnswer } from "@/lib/rag/rag-document-summary-context"; import { generationFailureDetailToken } from "@/lib/rag/rag-generation-failure-diagnostics"; import { answerLatencyMetadata, answerScopedEvidenceMetadata } from "@/lib/rag/rag-answer-telemetry-metadata"; import { assertRetrievalRows, buildDocumentSummaryResults } from "@/lib/rag/rag-row-contracts"; @@ -335,7 +335,6 @@ import { buildCrossDocumentFusionBrief, buildCrossDocumentSourceGuide } from "@/ import { buildSmartRagApiPlan } from "@/lib/smart-rag-api"; import { clinicalModePrompt, queryClassForClinicalMode, queryForClinicalMode } from "@/lib/clinical-query-mode"; import { annotateSearchResults, buildEvidenceRelevance } from "@/lib/evidence-relevance"; -import { committedIndexGeneration } from "@/lib/reindex-pipeline"; import { applySelectedEvidenceArtifacts, buildSelectedEvidenceArtifacts, @@ -4279,11 +4278,7 @@ export async function summarizeDocument( options?: { signal?: AbortSignal; observationContext?: import("@/lib/rag/rag-contracts").RagObservationContext }, ) { const { document, chunks } = await loadDocumentSummaryContext(documentId, ownerId, options?.signal); - const committedGeneration = committedIndexGeneration((document as { metadata?: unknown }).metadata); - const committedChunks = chunks.filter( - (chunk) => !chunk.index_generation_id || chunk.index_generation_id === committedGeneration, - ); - if (!committedChunks.length) { + if (!chunks.length) { return observeRagAnswer( { answer: "This document has not been indexed yet, so no summary can be generated.", @@ -4296,7 +4291,7 @@ export async function summarizeDocument( ); } - const results = buildDocumentSummaryResults(committedChunks, document); + const results = buildDocumentSummaryResults(chunks, document); const summaryInstructions = `Summarize a clinical document for practical psychiatric use in Perth, Australia. Use only the excerpts provided. Use a layered response: make the answer field a plain high-yield clinical paragraph, @@ -4331,18 +4326,23 @@ ${buildRagSourceBlock(results)}`; answer.documentBreakdown = buildDocumentBreakdown(results, answer.quoteCards); answer.evidenceSummary = buildEvidenceSummary(results, answer.quoteCards); answer.sourceCoverage = buildSourceCoverage(results); - answer.conflictsOrGaps = detectConflictsOrGaps(results); + const detectedConflicts = detectConflictsOrGaps(results); + const existingConflictMessages = new Set((answer.conflictsOrGaps ?? []).map((gap) => gap.message)); + answer.conflictsOrGaps = [ + ...(answer.conflictsOrGaps ?? []), + ...detectedConflicts.filter((gap) => !existingConflictMessages.has(gap.message)), + ]; answer.visualEvidence = buildVisualEvidence(results); answer.bestSource = selectBestSourceRecommendation(results, answer.quoteCards); answer.smartPanel = { ...buildSmartPanel("summary", results), bestSource: answer.bestSource }; answer.modelUsed = env.OPENAI_SUMMARY_MODEL; answer.openAIRequestIds = generated.requestId ? [generated.requestId] : []; answer.openAIUsage = generated.usage; - answer.latencyTimings = { - generation_latency_ms: generated.latencyMs, - total_latency_ms: generated.latencyMs, - }; - return observeRagAnswer(assessAndEnforceClaimSupport(answer), options?.observationContext); + answer.latencyTimings = { generation_latency_ms: generated.latencyMs, total_latency_ms: generated.latencyMs }; + const verifiedAnswer = sanitizeUnsupportedSummaryAnswer( + applyNumericVerification(assessAndEnforceClaimSupport(answer), results), + ); + return observeRagAnswer(verifiedAnswer, options?.observationContext); } export function sanitizePromptUserQuery(query: string): string { diff --git a/supabase/functions/_shared/source-anchored-summary.ts b/supabase/functions/_shared/source-anchored-summary.ts index a518b886fc..af98a9d492 100644 --- a/supabase/functions/_shared/source-anchored-summary.ts +++ b/supabase/functions/_shared/source-anchored-summary.ts @@ -5,6 +5,7 @@ export const SOURCE_ANCHORED_SUMMARY_PROJECTION_COMPLETE_KEY = export const SOURCE_ANCHORED_HEURISTIC_MODEL = "deterministic-source-coverage-heuristic" as const; export const SOURCE_ANCHORED_HEURISTIC_MODEL_VERSION = SOURCE_ANCHORED_SUMMARY_VERSION; export const SOURCE_ANCHORED_STRUCTURED_MODEL_VERSION = "rag-deep-memory-v1" as const; +export const SOURCE_ANCHORED_LOCAL_WORKER_MODEL_VERSION = "rag-deep-memory-v2" as const; export const SOURCE_ANCHORED_SOURCE_PAGE_SIZE = 1000; export const SOURCE_ANCHORED_MAX_CHUNKS = 5000; export const SOURCE_ANCHORED_MAX_IMAGES = 1000; @@ -614,7 +615,13 @@ export function normalizeSourceAnchoredSummary( const labels = validateLabels(raw.labels, profile); if (structured) { - if (model !== authority.expectedStructuredModel || model_version !== authority.expectedStructuredModelVersion) { + const isLocalWorkerStructuredSummary = + authority.expectedStructuredModelVersion === SOURCE_ANCHORED_STRUCTURED_MODEL_VERSION && + model_version === SOURCE_ANCHORED_LOCAL_WORKER_MODEL_VERSION; + const validStructuredModelVersion = + model_version === authority.expectedStructuredModelVersion || isLocalWorkerStructuredSummary; + const validStructuredModel = model === authority.expectedStructuredModel; + if (!validStructuredModel || !validStructuredModelVersion) { fail("source_anchored_summary_structured_model_invalid"); } const projection = projectSourceAnchoredSummary(profile); diff --git a/supabase/functions/indexing-v3-agent/index.ts b/supabase/functions/indexing-v3-agent/index.ts index 7183c7272b..9378cf8710 100644 --- a/supabase/functions/indexing-v3-agent/index.ts +++ b/supabase/functions/indexing-v3-agent/index.ts @@ -531,6 +531,8 @@ async function loadEdgeProviderSourceBinding(job: ClaimedJob) { from public.documents d left join public.document_admission_events a on a.document_id = d.id where d.id = ${job.document_id}::uuid + order by a.created_at desc nulls last, a.id desc + limit 1 `; attempt.check(); return edgeProviderEgressBinding(attempt.fence, job.attempt_count, rows[0]); @@ -1101,7 +1103,7 @@ async function upsertCoreEmbeddingFields(job: ClaimedJob, payload: SourceAnchore assertEmbeddingDim(embeddings[index], `${row.field_type} embedding field`); return { document_id: job.document_id, - source_chunk_id: null, + source_chunk_id: payload.source_chunk_ids[0] ?? null, ...row, embedding: embeddings[index], content_hash: await sha256Hex(row.content), diff --git a/tests/document-enrichment.test.ts b/tests/document-enrichment.test.ts index d704810369..f9bdc0a33b 100644 --- a/tests/document-enrichment.test.ts +++ b/tests/document-enrichment.test.ts @@ -27,7 +27,12 @@ vi.mock("@/lib/openai", () => ({ }), })); -import { generateDocumentEnrichment, ragEnrichmentVersion, upsertDocumentEnrichment } from "@/lib/document-enrichment"; +import { + fetchRelatedDocumentMetadata, + generateDocumentEnrichment, + ragEnrichmentVersion, + upsertDocumentEnrichment, +} from "@/lib/document-enrichment"; type QueryResult = { data: unknown; error: { message: string } | null }; type QueryCall = { @@ -505,4 +510,79 @@ describe("document enrichment", () => { expect.objectContaining({ document: "fallback-protocol.pdf" }), ); }); + + it("filters hidden labels and selects the newest committed summary in fetchRelatedDocumentMetadata", async () => { + const docId = "doc-meta-1"; + const summaryOrderCalls: Array<{ column: string; options?: { ascending?: boolean } }> = []; + const fallbackSupabase = { + rpc: vi.fn(async () => ({ data: null, error: { message: "rpc_unavailable" } })), + from: vi.fn((table: string) => { + const builder = { + select: () => builder, + in: () => builder, + eq: () => builder, + is: () => builder, + or: () => builder, + order: (column: string, options?: { ascending?: boolean }) => { + if (table === "document_summaries") summaryOrderCalls.push({ column, options }); + return builder; + }, + abortSignal: () => builder, + then: (resolve: (value: { data: unknown[]; error: null }) => unknown) => { + if (table === "documents") { + return Promise.resolve({ + data: [{ id: docId, metadata: { index_generation_id: "gen-current" } }], + error: null, + }).then(resolve); + } + if (table === "document_summaries") { + return Promise.resolve({ + data: [ + { + document_id: docId, + summary: "Older committed summary", + generated_at: "2026-07-01T00:00:00.000Z", + }, + { + document_id: docId, + summary: "Newest committed summary", + generated_at: "2026-07-02T00:00:00.000Z", + }, + ], + error: null, + }).then(resolve); + } + return Promise.resolve({ + data: [ + { + document_id: docId, + label: "clozapine", + label_type: "medication", + metadata: { index_generation_id: "gen-current" }, + }, + { + document_id: docId, + label: "suppressed-tag", + label_type: "topic", + metadata: { index_generation_id: "gen-current", review_status: "hidden" }, + }, + ], + error: null, + }).then(resolve); + }, + }; + return builder; + }), + }; + + const fallbackRows = await fetchRelatedDocumentMetadata({ + supabase: fallbackSupabase as never, + documentIds: [docId], + }); + expect(summaryOrderCalls).toEqual([{ column: "generated_at", options: { ascending: false } }]); + expect(fallbackRows[0]?.summary).toBe("Newest committed summary"); + expect(fallbackRows[0]?.labels).toEqual([ + expect.objectContaining({ label: "clozapine", label_type: "medication" }), + ]); + }); }); diff --git a/tests/indexing-v3-agent.test.ts b/tests/indexing-v3-agent.test.ts index 19ef943ad4..419d7442ec 100644 --- a/tests/indexing-v3-agent.test.ts +++ b/tests/indexing-v3-agent.test.ts @@ -427,4 +427,113 @@ describe("indexing-v3-agent claimed-batch request loop (L13)", () => { // stranded the batch; it must not come back. expect(handler).not.toContain("await markJobFailure(job, msg)"); }); + + it("accepts local-worker rag-deep-memory-v2 structured summaries in normalizeSourceAnchoredSummary and persists primary source_chunk_id", async () => { + const { + SOURCE_ANCHORED_STRUCTURED_MODEL_VERSION, + SOURCE_ANCHORED_SUMMARY_VERSION, + normalizeSourceAnchoredSummary, + projectSourceAnchoredSummary, + } = await import("../supabase/functions/_shared/source-anchored-summary"); + + const docId = "00000000-0000-4000-8000-000000000001"; + const genId = "00000000-0000-4000-8000-000000000002"; + const chunkId = "11111111-1111-4111-8111-111111111111"; + const authority = { + documentId: docId, + indexGenerationId: genId, + expectedStructuredModel: "gpt-4.1-mini", + expectedStructuredModelVersion: SOURCE_ANCHORED_STRUCTURED_MODEL_VERSION, + selectedChunks: [{ id: chunkId, page_number: 1, index_generation_id: genId }], + selectedImages: [], + }; + const actionItem = { + text: "Check baseline ANC before starting clozapine.", + source_chunk_ids: [chunkId], + source_image_ids: [], + pages: [1], + evidence_type: "text" as const, + support: "direct" as const, + }; + const baseProfile = { + overview: "placeholder", + applies_to: [], + key_clinical_actions: [actionItem], + medication_dose_monitoring: [], + thresholds_timing: [], + escalation_risk_warnings: [], + required_forms_documentation: [], + not_covered: [], + important_tables_images: [], + best_questions: [], + source_quality_notes: [], + }; + const projectedOverview = projectSourceAnchoredSummary(baseProfile); + const profile = { ...baseProfile, overview: projectedOverview }; + + const normalized = normalizeSourceAnchoredSummary( + { + summary_kind: "structured", + summary_version: SOURCE_ANCHORED_SUMMARY_VERSION, + document_id: docId, + index_generation_id: genId, + model: "gpt-4.1-mini", + model_version: "rag-deep-memory-v2", + generated_at: "2026-07-01T00:00:00.000Z", + source_chunk_ids: [chunkId], + source_image_ids: [], + summary: projectedOverview, + clinical_specifics: { + profile, + actions: [actionItem.text], + thresholds_timing: [], + medication_monitoring: [], + risk_escalation: [], + documentation_forms: [], + exceptions_gaps: [], + }, + labels: [], + heuristic_fallback: false, + }, + authority, + ); + + expect(normalized.model_version).toBe("rag-deep-memory-v2"); + expect(() => + normalizeSourceAnchoredSummary( + { + summary_kind: "structured", + summary_version: SOURCE_ANCHORED_SUMMARY_VERSION, + document_id: docId, + index_generation_id: genId, + model: "unexpected-model", + model_version: "rag-deep-memory-v2", + generated_at: "2026-07-01T00:00:00.000Z", + source_chunk_ids: [chunkId], + source_image_ids: [], + summary: projectedOverview, + clinical_specifics: { + profile, + actions: [actionItem.text], + thresholds_timing: [], + medication_monitoring: [], + risk_escalation: [], + documentation_forms: [], + exceptions_gaps: [], + }, + labels: [], + heuristic_fallback: false, + }, + authority, + ), + ).toThrow("source_anchored_summary_structured_model_invalid"); + + const edgeSource = String( + await import("node:fs/promises").then((fs) => + fs.readFile(new URL("../supabase/functions/indexing-v3-agent/index.ts", import.meta.url), "utf8"), + ), + ); + expect(edgeSource).toContain("order by a.created_at desc nulls last, a.id desc"); + expect(edgeSource).toContain("source_chunk_id: payload.source_chunk_ids[0] ?? null"); + }); }); diff --git a/tests/ingestion-quality-route.test.ts b/tests/ingestion-quality-route.test.ts index 08d4514a51..aa0dc6efec 100644 --- a/tests/ingestion-quality-route.test.ts +++ b/tests/ingestion-quality-route.test.ts @@ -307,4 +307,184 @@ describe("/api/ingestion/quality", () => { expect(payload).toEqual({ items: [], demoMode: true }); expect(client.from).not.toHaveBeenCalled(); }); + + it("ignores historical failed jobs and OCR stages when a newer job completed for the same document", async () => { + const completedJobId = "99999999-9999-4999-8999-999999999999"; + const client = clientWithTables({ + documents: [ + { + id: documentId, + title: "Recovered guideline", + file_name: "recovered.pdf", + status: "indexed", + page_count: 2, + chunk_count: 4, + image_count: 0, + error_message: null, + metadata: { + extraction_quality: "good", + clinical_validation_status: "approved", + document_status: "current", + }, + updated_at: "2026-06-25T00:10:00.000Z", + }, + ], + document_index_quality: [], + // Ordered newest first (updated_at DESC), matching the route query + ingestion_jobs: [ + { + id: completedJobId, + document_id: documentId, + status: "completed", + stage: "completed", + error_message: null, + updated_at: "2026-06-25T00:10:00.000Z", + }, + { + id: jobId, + document_id: documentId, + status: "failed", + stage: "ocr", + error_message: "Historical OCR failure.", + updated_at: "2026-06-25T00:02:00.000Z", + }, + ], + ingestion_job_stages: [ + { + id: "55555555-5555-4555-8555-555555555555", + document_id: documentId, + job_id: "99999999-9999-4999-8999-999999999999", + stage_name: "visual_extract", + stage_status: "completed", + error_message: null, + metadata: {}, + artifact_counts: {}, + started_at: "2026-06-25T00:12:00.000Z", + finished_at: "2026-06-25T00:13:00.000Z", + }, + { + id: "44444444-4444-4444-8444-444444444444", + document_id: documentId, + job_id: "99999999-9999-4999-8999-999999999999", + stage_name: "visual_extract", + stage_status: "failed", + error_message: "Transient extract timeout.", + metadata: {}, + artifact_counts: {}, + started_at: "2026-06-25T00:11:00.000Z", + finished_at: "2026-06-25T00:11:30.000Z", + }, + { + id: "33333333-3333-4333-8333-333333333333", + document_id: documentId, + job_id: jobId, + stage_name: "ocr", + stage_status: "failed", + error_message: "Historical OCR failure.", + metadata: {}, + artifact_counts: {}, + started_at: "2026-06-25T00:02:00.000Z", + finished_at: "2026-06-25T00:03:00.000Z", + }, + ], + document_pages: [ + { + document_id: documentId, + page_number: 1, + text: "Sufficient extracted clinical text after successful retry of the document ingestion job.", + ocr_used: true, + metadata: {}, + }, + ], + document_images: [], + }); + vi.doMock("@/lib/env", () => ({ isDemoMode: () => false })); + vi.doMock("@/lib/supabase/admin", () => ({ createAdminClient: () => client })); + vi.doMock("@/lib/supabase/auth", () => ({ + AuthenticationError: class AuthenticationError extends Error {}, + requireAuthenticatedUser: vi.fn(async () => ({ id: userId })), + unauthorizedResponse: () => Response.json({ error: "Authentication required." }, { status: 401 }), + })); + const { GET } = await import("../src/app/api/ingestion/quality/route"); + + const response = await GET(new Request("http://localhost/api/ingestion/quality")); + const payload = await response.json(); + + expect(response.status).toBe(200); + expect(payload.items.map((item: { type: string }) => item.type)).not.toContain("failed_job"); + expect(payload.items.map((item: { type: string }) => item.type)).not.toContain("failed_ocr"); + }); + + it("rolls back non-atomic document status when enqueueDocumentReindexJob loses a 23505 race to a processing job", async () => { + const documentUpdates: unknown[] = []; + const supabase = { + from: vi.fn((table: string) => { + if (table === "ingestion_jobs") { + let isSelect = false; + const chain = { + select: vi.fn(() => { + isSelect = true; + return chain; + }), + insert: vi.fn(() => chain), + eq: vi.fn(() => chain), + in: vi.fn(() => chain), + order: vi.fn(() => chain), + limit: vi.fn(() => chain), + single: vi.fn(async () => ({ + data: null, + error: { code: "23505", message: "duplicate key value violates unique constraint" }, + })), + then: (resolve: (value: { data: unknown[]; error: null }) => unknown) => + Promise.resolve({ + // First select returns no open jobs; second select (after 23505) returns a processing job + data: isSelect && documentUpdates.length > 0 ? [{ id: jobId, status: "processing" }] : [], + error: null, + }).then(resolve), + }; + return chain; + } + const docChain = { + update: vi.fn((payload: unknown) => { + documentUpdates.push(payload); + return docChain; + }), + eq: vi.fn(() => docChain), + is: vi.fn(() => docChain), + then: (resolve: (value: { data: null; error: null }) => unknown) => + Promise.resolve({ data: null, error: null }).then(resolve), + }; + return docChain; + }), + }; + + vi.doMock("@/lib/env", () => ({ env: { WORKER_MAX_ATTEMPTS: 3 }, isDemoMode: () => false })); + const { enqueueDocumentReindexJob } = await import("../src/lib/ingestion-enqueue"); + const result = await enqueueDocumentReindexJob({ + supabase: supabase as never, + document: { + id: documentId, + owner_id: userId, + status: "failed", + page_count: 5, + chunk_count: 12, + image_count: 2, + error_message: "Prior failure", + }, + }); + + expect(result).toEqual({ + outcome: "already_active", + }); + // First update set status: "queued" and zeroed counts; second update restored prior counts with status: "processing" + expect(documentUpdates).toHaveLength(2); + expect(documentUpdates[0]).toMatchObject({ status: "queued", page_count: 0, chunk_count: 0, image_count: 0 }); + expect(documentUpdates[1]).toMatchObject({ + status: "processing", + page_count: 5, + chunk_count: 12, + image_count: 2, + error_message: null, + }); + }); }); diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts index 0c9d92df4c..a7607786af 100644 --- a/tests/private-access-routes.test.ts +++ b/tests/private-access-routes.test.ts @@ -786,11 +786,19 @@ describe("private document API access", () => { const summaryCalls = client.calls.filter((call) => call.table === "document_summaries"); expect(labelCalls).toHaveLength(2); expect(summaryCalls).toHaveLength(2); + // Public-row queries read `metadata` server-side only so the hidden-label and committed-generation + // filters can run; the response projection below must still strip it. They never read owner-internal columns. + for (const call of [...labelCalls, ...summaryCalls]) { + expect(call.selected).toEqual(expect.stringContaining("metadata")); + } expect(labelCalls.map((call) => call.selected)).toEqual( - expect.arrayContaining([expect.stringContaining("metadata"), expect.not.stringContaining("metadata")]), + expect.arrayContaining([expect.stringContaining("owner_id"), expect.not.stringContaining("owner_id")]), ); expect(summaryCalls.map((call) => call.selected)).toEqual( - expect.arrayContaining([expect.stringContaining("source_chunk_ids"), expect.not.stringContaining("metadata")]), + expect.arrayContaining([ + expect.stringContaining("source_chunk_ids"), + expect.not.stringContaining("source_chunk_ids"), + ]), ); expect((ownedRow.labels as Array>)[0]).toHaveProperty("metadata"); expect(ownedRow.summary).toHaveProperty("source_chunk_ids"); @@ -806,6 +814,67 @@ describe("private document API access", () => { }); }); + it("drops hidden labels and staged-generation labels and summaries on public documents in an authenticated list", async () => { + const shared = { + id: otherDocumentId, + owner_id: null, + title: "Public guideline", + status: "indexed", + metadata: { index_generation_id: "gen-current" }, + }; + const client = createSupabaseMock((call) => { + if (call.table === "documents") return ok([shared]); + if (call.table === "document_labels") { + return ok([ + { + id: "visible", + document_id: otherDocumentId, + label: "Visible", + metadata: { index_generation_id: "gen-current" }, + }, + { id: "hidden", document_id: otherDocumentId, label: "Hidden", metadata: { hidden: true } }, + { + id: "staged", + document_id: otherDocumentId, + label: "Staged", + metadata: { index_generation_id: "gen-staged" }, + }, + ]); + } + if (call.table === "document_summaries") { + return ok([ + { + id: "staged-summary", + document_id: otherDocumentId, + summary: "Staged summary", + generated_at: "2026-02-01T00:00:00Z", + metadata: { index_generation_id: "gen-staged" }, + }, + { + id: "current-summary", + document_id: otherDocumentId, + summary: "Current summary", + generated_at: "2026-01-01T00:00:00Z", + metadata: { index_generation_id: "gen-current" }, + }, + ]); + } + return ok([]); + }); + mockRuntime(client); + const { GET } = await import("../src/app/api/documents/route"); + + const response = await GET(authenticatedRequest("/api/documents?includeMeta=true")); + const body = await payload(response); + const [sharedRow] = body.documents as Array>; + + expect(response.status).toBe(200); + expect((sharedRow.labels as Array>).map((label) => label.id)).toEqual(["visible"]); + expect(sharedRow.summary).toMatchObject({ id: "current-summary", summary: "Current summary" }); + expect(sharedRow).not.toHaveProperty("metadata"); + expect(sharedRow.summary).not.toHaveProperty("metadata"); + }); + it("accepts legacy Supabase auth cookies for private document access", async () => { const documents = [{ id: documentId, owner_id: userId, title: "Owned document" }]; const client = createSupabaseMock((call) => (call.table === "documents" ? ok(documents) : ok([]))); diff --git a/tests/rag-document-summary.test.ts b/tests/rag-document-summary.test.ts index 906a24ca54..13d09efa46 100644 --- a/tests/rag-document-summary.test.ts +++ b/tests/rag-document-summary.test.ts @@ -97,6 +97,7 @@ describe("document summary context", () => { expect(documentQuery.eq).toHaveBeenCalledWith("owner_id", ownerId); expect(documentQuery.abortSignal).toHaveBeenCalledWith(controller.signal); + expect(chunkQuery.eq).toHaveBeenCalledWith("index_generation_id", "generation-current"); expect(chunkQuery.order).toHaveBeenCalledWith("chunk_index", { ascending: true }); expect(chunkQuery.limit).toHaveBeenCalledWith(40); expect(chunkQuery.abortSignal).toHaveBeenCalledWith(controller.signal); @@ -117,6 +118,151 @@ describe("document summary context", () => { ); }); + it("excludes uncommitted staged-generation chunks and enforces numeric verification on document summaries", async () => { + vi.stubEnv("OPENAI_API_KEY", "test-key"); + + const documentId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; + const ownerId = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"; + + const documentQuery = { + select: vi.fn().mockReturnThis(), + eq: vi.fn().mockReturnThis(), + abortSignal: vi.fn().mockReturnThis(), + maybeSingle: vi.fn(async () => ({ + data: { + id: documentId, + title: "Clozapine Initiation Guideline", + file_name: "clozapine-initiation.pdf", + status: "indexed", + metadata: { + index_generation_id: "generation-current", + publisher: "WA Health", + jurisdiction: "Australia/WA", + document_status: "current", + clinical_validation_status: "approved", + extraction_quality: "good", + }, + }, + error: null, + })), + }; + const chunkRows = [ + { + id: "uncommitted-chunk", + document_id: documentId, + page_number: 1, + chunk_index: 0, + section_heading: "Draft Dosing", + content: "Uncommitted draft says start at 999 mg daily.", + retrieval_synopsis: null, + image_ids: [], + index_generation_id: "generation-staged-uncommitted", + }, + { + id: "committed-chunk", + document_id: documentId, + page_number: 1, + chunk_index: 1, + section_heading: "Starting Dose", + content: "Commence clozapine at 12.5 mg once or twice daily on day 1.", + retrieval_synopsis: null, + image_ids: [], + index_generation_id: "generation-current", + }, + ]; + const chunkQuery = { + select: vi.fn().mockReturnThis(), + eq: vi.fn().mockReturnThis(), + order: vi.fn().mockReturnThis(), + limit: vi.fn().mockReturnThis(), + abortSignal: vi.fn().mockReturnThis(), + then: (resolve: (value: { data: typeof chunkRows; error: null }) => unknown) => + Promise.resolve({ data: chunkRows, error: null }).then(resolve), + }; + const from = vi.fn((table: string) => (table === "documents" ? documentQuery : chunkQuery)); + + vi.doMock("@/lib/supabase/admin", () => ({ + createAdminClient: () => ({ from, rpc: vi.fn() }), + })); + + const generateStructuredTextResult = vi.fn(async () => ({ + text: JSON.stringify({ + answer: "Commence clozapine at 500 mg once daily on day 1.", + grounded: true, + confidence: "high", + answerSections: [], + citations: [{ chunk_id: "committed-chunk" }], + quoteCards: [], + conflictsOrGaps: [], + }), + model: "gpt-4.1-mini", + operation: "summary", + latencyMs: 10, + requestId: "req_summary_numeric", + usage: { input_tokens: 100, output_tokens: 40, total_tokens: 140 }, + })); + vi.doMock("@/lib/openai", () => ({ + embedTextWithTelemetry: vi.fn(), + generateStructuredTextResult, + })); + + const { summarizeDocument } = await import("../src/lib/rag/rag"); + const answer = await summarizeDocument(documentId, ownerId); + + expect(documentQuery.select).toHaveBeenCalledWith("id,title,file_name,status,metadata"); + const summaryPrompt = (generateStructuredTextResult.mock.calls[0] as unknown[] | undefined)?.[0] ?? ""; + expect(summaryPrompt).not.toContain("Uncommitted draft"); + expect(summaryPrompt).toContain("Commence clozapine at 12.5 mg"); + // Fabricated 500 mg dose is caught by applyNumericVerification + expect(answer.grounded).toBe(false); + expect(answer.confidence).toBe("unsupported"); + expect(answer.bestSource).toBeNull(); + expect(answer.smartPanel?.bestSource).toBeNull(); + expect(answer.supportedClaims).toBeUndefined(); + expect(answer.evidenceAssessments).toBeUndefined(); + }); + + it("refuses to summarize non-indexed documents without querying chunks or OpenAI", async () => { + vi.stubEnv("OPENAI_API_KEY", "test-key"); + const documentId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; + const documentQuery = { + select: vi.fn().mockReturnThis(), + or: vi.fn().mockReturnThis(), + eq: vi.fn().mockReturnThis(), + maybeSingle: vi.fn(async () => ({ + data: { + id: documentId, + title: "Queued Guideline", + file_name: "queued.pdf", + status: "queued", + metadata: {}, + }, + error: null, + })), + }; + const from = vi.fn(() => documentQuery); + const generateStructuredTextResult = vi.fn(); + + vi.doMock("@/lib/supabase/admin", () => ({ + createAdminClient: () => ({ from, rpc: vi.fn() }), + })); + vi.doMock("@/lib/openai", () => ({ + embedTextWithTelemetry: vi.fn(), + generateStructuredTextResult, + })); + + const { summarizeDocument } = await import("../src/lib/rag/rag"); + const answer = await summarizeDocument(documentId, "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"); + + expect(documentQuery.or).toHaveBeenCalledWith( + "owner_id.eq.bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb,and(owner_id.is.null,metadata->>public_corpus.eq.true)", + ); + expect(from).toHaveBeenCalledTimes(1); + expect(generateStructuredTextResult).not.toHaveBeenCalled(); + expect(answer.grounded).toBe(false); + expect(answer.confidence).toBe("unsupported"); + }); + it("stops before the first database read when the caller already disconnected", async () => { vi.stubEnv("OPENAI_API_KEY", "test-key"); const from = vi.fn(); diff --git a/tests/rag-shared-cache.test.ts b/tests/rag-shared-cache.test.ts index 5c501cd934..b633b66a8e 100644 --- a/tests/rag-shared-cache.test.ts +++ b/tests/rag-shared-cache.test.ts @@ -587,28 +587,26 @@ describe("shared RAG search cache", () => { vi.doMock("@/lib/clinical-search", () => ({ buildClinicalTextSearchQuery: (query: string) => query.trim(), })); + const docBuilder = { + select: () => docBuilder, + eq: () => docBuilder, + is: () => docBuilder, + or: () => docBuilder, + gt: () => docBuilder, + order: () => docBuilder, + limit: () => docBuilder, + then: (resolve: (value: { data: unknown[]; error: null }) => unknown, reject?: (reason: unknown) => unknown) => + Promise.resolve({ + data: [{ id: "doc-1", updated_at: "2026-07-14T00:00:00.000Z", metadata: {} }], + error: null, + }).then(resolve, reject), + maybeSingle: async () => ({ data: null, error: null }), + }; vi.doMock("@/lib/supabase/admin", () => ({ createAdminClient: () => ({ from: (table: string) => { if (table === "rag_response_cache") return shared.builder; - return { - select: () => shared.sharedBuilder, - eq: () => shared.sharedBuilder, - is: () => shared.sharedBuilder, - or: () => shared.sharedBuilder, - gt: () => shared.sharedBuilder, - order: () => shared.sharedBuilder, - limit: () => shared.sharedBuilder, - then: ( - resolve: (value: { data: unknown[]; error: null }) => unknown, - reject?: (reason: unknown) => unknown, - ) => - Promise.resolve({ - data: [{ id: "doc-1", updated_at: "2026-07-14T00:00:00.000Z", metadata: {} }], - error: null, - }).then(resolve, reject), - maybeSingle: async () => ({ data: null, error: null }), - }; + return docBuilder; }, }), })); @@ -627,4 +625,130 @@ describe("shared RAG search cache", () => { expect(result).toBeNull(); expect(shared.deletionCalls).toBe(1); }); + + it("fails closed on local and shared cache reads/writes when the indexing stamp is unavailable", async () => { + vi.resetModules(); + let sharedCacheReads = 0; + let sharedCacheWrites = 0; + + const failingDocBuilder = { + select: () => failingDocBuilder, + eq: () => failingDocBuilder, + is: () => failingDocBuilder, + or: () => failingDocBuilder, + in: () => failingDocBuilder, + order: () => failingDocBuilder, + limit: () => failingDocBuilder, + then: (resolve: (value: { data: unknown; error: { message: string } | null }) => unknown) => + Promise.resolve({ data: null, error: { message: "db_unavailable" } }).then(resolve), + }; + const cacheTableBuilder = { + select: () => cacheTableBuilder, + delete: () => cacheTableBuilder, + insert: () => { + sharedCacheWrites += 1; + return Promise.resolve({ data: null, error: null }); + }, + eq: () => cacheTableBuilder, + is: () => cacheTableBuilder, + gt: () => cacheTableBuilder, + limit: () => cacheTableBuilder, + maybeSingle: async () => { + sharedCacheReads += 1; + return { + data: { + payload: { + results: [], + telemetry: { query_class: "general_clinical_management" }, + }, + }, + error: null, + }; + }, + then: (resolve: (value: { data: null; error: null }) => unknown) => + Promise.resolve({ data: null, error: null }).then(resolve), + }; + + vi.doMock("@/lib/env", () => ({ + env: { + RAG_SEARCH_CACHE_TTL_MS: 60_000, + RAG_SEARCH_CACHE_SIZE: 200, + RAG_ANSWER_CACHE_TTL_MS: 60_000, + RAG_ANSWER_CACHE_SIZE: 200, + RAG_PERSIST_RAW_QUERY_TEXT: false, + RAG_QUERY_HASH_SECRET: "test-query-hash-secret", + }, + isDemoMode: () => false, + isLocalNoAuthMode: () => false, + })); + vi.doMock("@/lib/deep-memory", () => ({ ragDeepMemoryVersion: "test-rag-version" })); + vi.doMock("@/lib/clinical-search", () => ({ + buildClinicalTextSearchQuery: (query: string) => query.trim(), + })); + vi.doMock("@/lib/supabase/admin", () => ({ + createAdminClient: () => ({ + from: (table: string) => (table === "documents" ? failingDocBuilder : cacheTableBuilder), + }), + })); + + const { + cacheIndexingVersion, + getCachedAnswer, + getCachedSearch, + getSharedCachedAnswer, + getSharedCachedSearch, + setCachedAnswer, + setCachedSearch, + } = await import("../src/lib/rag/rag-cache"); + + const args = { query: "lithium toxicity", ownerId }; + await setCachedSearch(args, [], { + search_cache_hit: false, + text_fast_path_latency_ms: 0, + embedding_skipped: true, + embedding_latency_ms: 0, + embedding_cache_hit: false, + supabase_rpc_latency_ms: 0, + rerank_latency_ms: 0, + }); + await setCachedAnswer(args, { + answer: "Grounded answer.", + grounded: true, + confidence: "high", + citations: [], + sources: [], + }); + + expect(await getCachedSearch(args)).toBeNull(); + expect(await getCachedAnswer(args, Date.now())).toBeNull(); + expect(await getSharedCachedSearch(args)).toEqual({ kind: "miss", reason: "cache_lookup_error" }); + expect(await getSharedCachedAnswer(args, Date.now())).toBeNull(); + expect(sharedCacheReads).toBe(0); + expect(sharedCacheWrites).toBe(0); + + // Now test that a valid local entry is NOT evicted when a transient DB stamp failure occurs, + // and that cacheIndexingVersion does not cache the :unavailable stamp for 5s. + let dbUnavailable = false; + failingDocBuilder.then = (resolve: (value: { data: unknown; error: { message: string } | null }) => unknown) => + Promise.resolve( + dbUnavailable + ? { data: null, error: { message: "transient_db_error" } } + : { data: [{ id: "doc-1", updated_at: "2026-08-30T00:00:00.000Z", metadata: {} }], error: null }, + ).then(resolve); + + await setCachedAnswer(args, { + answer: "Recovered grounded answer.", + grounded: true, + confidence: "high", + citations: [], + sources: [], + }); + dbUnavailable = true; + const unavailableStamp = await cacheIndexingVersion(args, { forceRefresh: true }); + expect(unavailableStamp).toContain(":index-stamp-unavailable"); + expect(await getCachedAnswer(args, Date.now())).toBeNull(); + dbUnavailable = false; + const hitAfterRecovery = await getCachedAnswer(args, Date.now()); + expect(hitAfterRecovery?.answer).toBe("Recovered grounded answer."); + }); }); diff --git a/worker/main.ts b/worker/main.ts index 2afb6a7c08..25b9ddc512 100644 --- a/worker/main.ts +++ b/worker/main.ts @@ -271,7 +271,9 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { p_batch_id: job.batch_id, p_stage: stage, }); - invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous"); + invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous", { + affectsPublicCorpus: !job.documents.owner_id, + }); } async function completeStrictEnrichmentJob(job: JobRow) { @@ -1859,7 +1861,9 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { stopHeartbeat: () => clearInterval(heartbeat), rpc: (name, payload) => ingestionRpcClient.rpc(name, payload), }); - invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous"); + invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous", { + affectsPublicCorpus: !job.documents.owner_id, + }); return; } if (originalWitness.status === "retained") { @@ -1979,7 +1983,9 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { stopHeartbeat: () => clearInterval(heartbeat), rpc: (name, payload) => ingestionRpcClient.rpc(name, payload), }); - invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous"); + invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous", { + affectsPublicCorpus: !job.documents.owner_id, + }); return; } } else { @@ -2277,7 +2283,9 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { if (enrichmentStatus === "completed") { const strictCompletion = await completeStrictEnrichmentJob(job); if (strictCompletion.completed) { - invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous"); + invalidateRagCachesForDocumentMutation(job.documents.owner_id ?? "anonymous", { + affectsPublicCorpus: !job.documents.owner_id, + }); await refreshRagTableStats(); return; }