diff --git a/services/platform/backend/domains/knowledge/service.ts b/services/platform/backend/domains/knowledge/service.ts index 2887e743c5..2326fd41e3 100644 --- a/services/platform/backend/domains/knowledge/service.ts +++ b/services/platform/backend/domains/knowledge/service.ts @@ -63,9 +63,21 @@ interface AccessScopeArg { } /** - * Tier-A retrievable filter over app tables: a file ref passes when its - * metadata row belongs to the org AND (bound document is un-trashed and in - * scope | thread-bound and the thread is in scope | legacy unbound same-org). + * The mandatory re-check over app tables: which corpus refs may this turn + * actually read. + * + * This is the gate `core/knowledge/corpus.ts` delegates to by name. The SQL + * half only ADMITS rows — its own comment says so — so this function is the + * one that decides, and it must FAIL CLOSED. A ref passes only when its + * metadata row belongs to the org, finished indexing, is not trashed, and + * lands in exactly one of the two allow branches: a bound document that is + * live, un-trashed, still the document's current file, and in scope; or a + * thread-bound upload whose thread is in the turn's lineage. + * + * Everything else is denied, including a row bound to nothing. Denying the + * unbound case is what 0.4 did, and it matters because the corpus stamps no + * project and no team for such a row, which the SQL half then reads as an + * org-wide hub row. */ async function filterRetrievableRagFileIds( sql: Sql, @@ -82,7 +94,12 @@ async function filterRetrievableRagFileIds( { storageRef: string; threadId: string | null; + conversationId: string | null; + ragStatus: string | null; + fileTrashed: boolean; documentId: string | null; + docExists: boolean; + docFileRef: string | null; docTrashed: boolean | null; docProjectId: string | null; docTeamId: string | null; @@ -90,7 +107,12 @@ async function filterRetrievableRagFileIds( }[] >` SELECT fm.storage_ref AS "storageRef", fm.thread_id AS "threadId", + fm.conversation_id AS "conversationId", + fm.rag_status AS "ragStatus", + (fm.lifecycle_status = 'trashed') AS "fileTrashed", fm.document_id AS "documentId", + (d.id IS NOT NULL) AS "docExists", + d.file_ref AS "docFileRef", (d.lifecycle_status = 'trashed') AS "docTrashed", d.project_id AS "docProjectId", d.team_id AS "docTeamId", d.team_tags AS "docTeamTags" @@ -111,10 +133,31 @@ async function filterRetrievableRagFileIds( if (!row) { continue; } + // Indexing has to have finished. A row mid-reindex still has its previous + // chunks in the corpus, and answering from them would serve content the + // caller's current scope was never checked against. + if (row.ragStatus !== 'completed') { + continue; + } + if (row.fileTrashed) { + continue; + } if (row.documentId !== null) { + // A document id pointing at a row that is gone is not a hub document — + // without this the LEFT JOIN's NULLs fall through to the hub branch and + // the ref is served to the whole org. + if (!row.docExists) { + continue; + } if (row.docTrashed === true) { continue; } + // A replacement upload moves the document's `file_ref` on and leaves the + // previous corpus row behind. Serving it would answer from a superseded + // version with nothing marking it stale. + if ((row.docFileRef ?? '') !== ref) { + continue; + } if (access === undefined) { retrievable.push(ref); continue; @@ -155,8 +198,18 @@ async function filterRetrievableRagFileIds( } continue; } - // Legacy/unbound: same-org fallback (the 0.4 posture). - retrievable.push(ref); + // An emailed attachment, once something indexes one. The allow branch + // belongs here and is decided by `conversationAssignmentAllows` against + // the conversation's CURRENT assignment, so a reassignment moves the + // attachment with the mail. Until that lands (#3121) a conversation-scoped + // row is denied rather than falling through. + if (row.conversationId !== null) { + continue; + } + // Bound to nothing: denied. The corpus stamps no project and no team for + // such a row, and the SQL half reads that as org-wide, so allowing it here + // would serve one caller's file to every member of the organization. + continue; } return retrievable; } diff --git a/services/platform/backend/integration-check.ts b/services/platform/backend/integration-check.ts index 9d2a55e1f0..e1d21f0725 100644 --- a/services/platform/backend/integration-check.ts +++ b/services/platform/backend/integration-check.ts @@ -3626,6 +3626,99 @@ async function checkProviderCredentials( void credentialId; } +/** + * The mandatory retrieval re-check, which decides which corpus refs a turn may + * read. The SQL half of the two-gate design admits rows and fails open by + * design, so this function is the one that has to fail CLOSED. Needs no object + * storage — it is pure SQL over `app.file_metadata` and `app.documents` — so it + * runs on every invocation, unlike the S3-gated RAG loop below. + */ +async function checkRetrievalGate( + sql: Sql, + ctx: { orgId: string }, +): Promise { + const org = `${ctx.orgId}-gate-fixture`; + const now = Date.now(); + const { knowledgeShimHandlers } = + await import('./domains/knowledge/service.ts'); + + const docRows = await sql<{ id: string }[]>` + INSERT INTO app.documents (org_id, title, file_ref, lifecycle_status, + created_by, created_at_ms, updated_at_ms) + VALUES (${org}, 'Gate fixture', 's3:gate/current.txt', 'active', 'gate-user', + ${now}, ${now}) + RETURNING id + `; + const documentId = docRows[0]?.id ?? ''; + const seedFile = async ( + ref: string, + row: { + ragStatus?: string; + documentId?: string; + conversationId?: string; + lifecycle?: string; + }, + ): Promise => { + await sql` + INSERT INTO app.file_metadata ( + org_id, storage_ref, file_name, content_type, size, rag_status, + document_id, conversation_id, lifecycle_status, created_at_ms + ) VALUES (${org}, ${ref}, 'f.txt', 'text/plain', 10, + ${row.ragStatus ?? 'completed'}, ${row.documentId ?? null}, + ${row.conversationId ?? null}, ${row.lifecycle ?? null}, ${now}) + `; + }; + // The one ref that must pass: the document's CURRENT file, live and un-trashed. + await seedFile('s3:gate/current.txt', { documentId }); + // A replacement moved the document's file_ref on; this row is superseded. + await seedFile('s3:gate/superseded.txt', { documentId }); + // A document id whose row is gone must not read as a hub document. + await seedFile('s3:gate/orphan.txt', { documentId: 'gate-no-such-document' }); + // Bound to nothing: the corpus stamps no project and no team for this shape, + // which the SQL half reads as org-wide. + await seedFile('s3:gate/unbound.txt', {}); + // An emailed attachment, denied until the conversation branch lands. + await seedFile('s3:gate/mail.txt', { conversationId: 'gate-conv' }); + // Mid-reindex: the previous chunks are still in the corpus. + await seedFile('s3:gate/running.txt', { documentId, ragStatus: 'running' }); + await seedFile('s3:gate/trashed.txt', { documentId, lifecycle: 'trashed' }); + + const expectations: [string, boolean][] = [ + ['s3:gate/current.txt', true], + ['s3:gate/superseded.txt', false], + ['s3:gate/orphan.txt', false], + ['s3:gate/unbound.txt', false], + ['s3:gate/mail.txt', false], + ['s3:gate/running.txt', false], + ['s3:gate/trashed.txt', false], + ]; + const handler = + knowledgeShimHandlers(sql)[ + 'documents/internal_queries:filterRetrievableRagFileIds' + ]; + const parsed = z.array(z.string()).safeParse( + await handler?.({ + organizationId: org, + fileIds: expectations.map(([ref]) => ref), + access: { + projectIds: [], + teamIds: [], + includeHub: true, + threadIds: [], + }, + }), + ); + const allowed = new Set(parsed.success ? parsed.data : []); + const wrong = expectations.filter( + ([ref, expect]) => allowed.has(ref) !== expect, + ); + record( + 'retrieval gate fails closed (unbound, superseded, orphan, mid-index)', + parsed.success && wrong.length === 0, + `parsed=${parsed.success}, wrong=${wrong.map(([ref]) => ref).join('|') || 'none'}, allowed=${[...allowed].join('|') || 'none'} (want only s3:gate/current.txt)`, + ); +} + /** * Knowledge (RAG) vertical against the real corpus database and a local * fake embedding endpoint: upload → document bind → rag.index_file job @@ -25902,6 +25995,7 @@ async function main(): Promise { await checkAgents(baseUrl, authCtx); await checkSkills(baseUrl, authCtx); await checkProviderCredentials(sql, baseUrl, authCtx); + await checkRetrievalGate(sql, authCtx); await checkKnowledge(sql, baseUrl, authCtx, `itest-${orgSuffix}`); await checkChat(sql, baseUrl, authCtx, `itest-${orgSuffix}`); await checkTts(sql, baseUrl, authCtx, `itest-${orgSuffix}`);