diff --git a/apps/api/alembic/env.py b/apps/api/alembic/env.py index 7b9e5022c..8e3a5d5a7 100644 --- a/apps/api/alembic/env.py +++ b/apps/api/alembic/env.py @@ -101,6 +101,10 @@ def run_migrations_offline() -> None: include_object=include_object, literal_binds=True, dialect_opts={"paramstyle": "named"}, + # Keep each migration in its own transaction so migrations that + # require an autocommit block (for example CREATE INDEX + # CONCURRENTLY) can safely commit only their own predecessor. + transaction_per_migration=True, # Pass through configured SSL connect args. connect_args=ssl_connect_args, ) @@ -119,13 +123,25 @@ def run_migrations_online() -> None: configured_connection = config.attributes.get("connection") def run_with_connection(connection: Connection) -> None: + caller_owned_transaction = connection.in_transaction() if settings.API_STANDALONE_MODE_ENABLED: ensure_better_auth_user_table(connection) + # The standalone bootstrap query starts SQLAlchemy's implicit + # transaction before Alembic begins tracking migration + # transactions. End only that transaction; never commit a + # transaction supplied by the caller. + if not caller_owned_transaction: + connection.commit() context.configure( connection=connection, target_metadata=target_metadata, include_object=include_object, + # Required for migrations that use autocommit_block(). + transaction_per_migration=True, + # Concurrent DDL cannot run inside a transaction owned by the + # caller. Migrations use regular DDL for that compatibility path. + knowhere_external_transaction=caller_owned_transaction, ) with context.begin_transaction(): @@ -136,7 +152,7 @@ def run_with_connection(connection: Connection) -> None: return if isinstance(configured_connection, Engine): - with configured_connection.begin() as connection: + with configured_connection.connect() as connection: run_with_connection(connection) return @@ -149,7 +165,7 @@ def run_with_connection(connection: Connection) -> None: connect_args=ssl_connect_args, ) - with connectable.begin() as connection: + with connectable.connect() as connection: run_with_connection(connection) diff --git a/apps/api/alembic/versions/fbf0c1d2e3f4_add_chunk_snapshot_order_index.py b/apps/api/alembic/versions/fbf0c1d2e3f4_add_chunk_snapshot_order_index.py new file mode 100644 index 000000000..56b6f4e3f --- /dev/null +++ b/apps/api/alembic/versions/fbf0c1d2e3f4_add_chunk_snapshot_order_index.py @@ -0,0 +1,110 @@ +"""Add the chunk snapshot pagination order index. + +Revision ID: fbf0c1d2e3f4 +Revises: f0d85d209e68, fbe1c2d3e4f5 +Create Date: 2026-08-27 00:00:00.000000 +""" + +from __future__ import annotations + +from typing import Sequence, Union + +from alembic import op +from sqlalchemy import text + + +revision: str = "fbf0c1d2e3f4" +down_revision: Union[str, Sequence[str], None] = ( + "f0d85d209e68", + "fbe1c2d3e4f5", +) +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +_INDEX_NAME = "idx_document_chunks_revision_snapshot_order" +_INDEX_COLUMNS = "(document_id, job_result_id, sort_order, chunk_id, id)" +_INDEX_METHOD_AND_COLUMNS = f"using btree {_INDEX_COLUMNS}" + + +def _read_index_definition() -> tuple[str, bool] | None: + row = op.get_bind().execute( + text( + """ + SELECT indexdef, index_metadata.indisvalid + FROM pg_indexes AS indexes + JOIN pg_class AS index_class + ON index_class.relname = indexes.indexname + JOIN pg_namespace AS index_namespace + ON index_namespace.oid = index_class.relnamespace + AND index_namespace.nspname = indexes.schemaname + JOIN pg_index AS index_metadata + ON index_metadata.indexrelid = index_class.oid + WHERE indexes.schemaname = current_schema() + AND indexes.tablename = 'document_chunks' + AND indexes.indexname = :index_name + """ + ), + {"index_name": _INDEX_NAME}, + ).first() + if row is None: + return None + return str(row[0]), bool(row[1]) + + +def _index_is_intended() -> bool: + definition = _read_index_definition() + if definition is None or not definition[1]: + return False + normalized_definition = " ".join(definition[0].lower().split()) + return ( + normalized_definition.startswith("create index ") + and _INDEX_METHOD_AND_COLUMNS in normalized_definition + and " where " not in normalized_definition + ) + + +def _drop_index(*, concurrently: bool) -> None: + concurrent_clause = "CONCURRENTLY " if concurrently else "" + op.execute(f"DROP INDEX {concurrent_clause}IF EXISTS {_INDEX_NAME}") + + +def _create_index(*, concurrently: bool) -> None: + concurrent_clause = "CONCURRENTLY " if concurrently else "" + op.execute( + f""" + CREATE INDEX {concurrent_clause}IF NOT EXISTS {_INDEX_NAME} + ON document_chunks {_INDEX_COLUMNS} + """ + ) + + +def upgrade() -> None: + external_transaction = bool( + op.get_context().opts.get("knowhere_external_transaction", False) + ) + if external_transaction: + if _read_index_definition() is not None and not _index_is_intended(): + _drop_index(concurrently=False) + if _read_index_definition() is None: + _create_index(concurrently=False) + return + + # Index creation must not hold a write lock on document_chunks while the + # production corpus is being indexed. CONCURRENTLY cannot run inside the + # transaction Alembic normally opens, so switch to an autocommit block. + with op.get_context().autocommit_block(): + if _read_index_definition() is not None and not _index_is_intended(): + _drop_index(concurrently=True) + if _read_index_definition() is None: + _create_index(concurrently=True) + + +def downgrade() -> None: + external_transaction = bool( + op.get_context().opts.get("knowhere_external_transaction", False) + ) + if external_transaction: + _drop_index(concurrently=False) + else: + with op.get_context().autocommit_block(): + _drop_index(concurrently=True) diff --git a/apps/api/app/core/exception_handlers.py b/apps/api/app/core/exception_handlers.py index 1df2f085b..7d5a72680 100644 --- a/apps/api/app/core/exception_handlers.py +++ b/apps/api/app/core/exception_handlers.py @@ -48,7 +48,7 @@ """ import uuid -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Mapping from typing import List, cast from fastapi import FastAPI, HTTPException, Request @@ -88,7 +88,10 @@ def _get_request_id(request: Request) -> str: async def knowhere_exception_handler( - request: Request, exc: KnowhereException + request: Request, + exc: KnowhereException, + *, + response_headers: Mapping[str, str] | None = None, ) -> JSONResponse: """ This handler enforces the separation between: @@ -119,7 +122,10 @@ async def knowhere_exception_handler( exc.logging(request_id=request_id) # Always include request ID header for client-side correlation - headers = {"X-Request-ID": request_id} + # Preserve framework-provided headers such as ``Allow: POST`` for 405 + # responses and ``WWW-Authenticate`` for authentication challenges. + headers = dict(response_headers or {}) + headers["X-Request-ID"] = request_id retry_after = exc.details.get("retry_after") if retry_after: headers["Retry-After"] = str(retry_after) @@ -208,7 +214,11 @@ async def http_exception_handler(request: Request, exc: HTTPException) -> JSONRe ) # Delegate to central handler - return await knowhere_exception_handler(request, knowhere_exc) + return await knowhere_exception_handler( + request, + knowhere_exc, + response_headers=exc.headers, + ) async def validation_exception_handler( diff --git a/apps/api/tests/contract/test_exception_handlers_contract.py b/apps/api/tests/contract/test_exception_handlers_contract.py new file mode 100644 index 000000000..40eff1064 --- /dev/null +++ b/apps/api/tests/contract/test_exception_handlers_contract.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import sys +from pathlib import Path +from typing import cast + +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient + +from tests.support.import_environment import ( + configure_import_environment, + ensure_import_paths, +) + +configure_import_environment() +ensure_import_paths() + + +def _prepare_api_app_imports() -> None: + api_root = str(Path(__file__).resolve().parents[2]) + if api_root in sys.path: + sys.path.remove(api_root) + sys.path.insert(0, api_root) + + +def _create_post_only_app() -> FastAPI: + _prepare_api_app_imports() + + from app.core.exception_handlers import setup_exception_handlers + + app = FastAPI() + + @app.post("/v2/retrieval/query") + async def query_retrieval() -> dict[str, bool]: + return {"ok": True} + + setup_exception_handlers(app) + return app + + +async def test_get_to_post_only_route_returns_method_not_allowed() -> None: + app = _create_post_only_app() + transport = ASGITransport(app=app) + + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.get("/v2/retrieval/query") + + assert response.status_code == 405 + assert response.headers["allow"] == "POST" + + response_json = cast(dict[str, object], response.json()) + error = cast(dict[str, object], response_json["error"]) + assert response_json["success"] is False + assert error["code"] == "METHOD_NOT_ALLOWED" + assert error["message"] == "Method not allowed" diff --git a/apps/api/tests/contract/test_retrieval_snapshot_batching_contract.py b/apps/api/tests/contract/test_retrieval_snapshot_batching_contract.py new file mode 100644 index 000000000..a7239100a --- /dev/null +++ b/apps/api/tests/contract/test_retrieval_snapshot_batching_contract.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager +from typing import cast +from uuid import uuid4 + +from httpx import AsyncClient +from sqlalchemy import Executable, Result +from sqlalchemy.sql.selectable import Select + +from shared.services.retrieval.nav_snapshot import SnapshotSession, load_nav_snapshot +from tests.support.retrieval_snapshot_support import contract_db_session +from tests.support.contract_database import ContractDatabase + + +_USER_ID = "local-dev-user" + + +class _CountingSession: + def __init__(self, session: SnapshotSession) -> None: + self._session = session + self.chunk_query_count = 0 + + async def execute(self, statement: Executable) -> Result[tuple[object, ...]]: + selected_tables = ( + { + getattr(table, "name", "") + for table in statement.get_final_froms() + } + if isinstance(statement, Select) + else set() + ) + if "document_chunks" in selected_tables: + self.chunk_query_count += 1 + result = await self._session.execute(statement) + return cast(Result[tuple[object, ...]], result) + + +async def _seed_many_small_documents(namespace: str, *, document_count: int) -> None: + identifier = uuid4().hex[:8] + await ContractDatabase.execute( + """ + INSERT INTO jobs ( + job_id, user_id, job_type, status, source_type, version, + webhook_enabled, created_at, updated_at, credits_charged, billing_status + ) + SELECT + 'job_batch_' || :identifier || '_' || i, + :user_id, + 'document_ingestion', + 'done', + 'file', + 0, + false, + NOW(), + NOW(), + 0, + 'skipped' + FROM generate_series(1, :document_count) AS values(i) + """, + { + "identifier": identifier, + "user_id": _USER_ID, + "document_count": document_count, + }, + ) + await ContractDatabase.execute( + """ + INSERT INTO documents ( + document_id, user_id, namespace, status, current_job_result_id, + source_file_name, parse_track, created_at, updated_at + ) + SELECT + 'doc_batch_' || :identifier || '_' || i, + :user_id, + :namespace, + 'active', + NULL, + 'batch-' || i || '.pdf', + 'chunk', + NOW(), + NOW() + FROM generate_series(1, :document_count) AS values(i) + """, + { + "identifier": identifier, + "user_id": _USER_ID, + "namespace": namespace, + "document_count": document_count, + }, + ) + await ContractDatabase.execute( + """ + INSERT INTO job_results ( + id, job_id, document_id, delivery_mode, created_at, updated_at + ) + SELECT + 'result_batch_' || :identifier || '_' || i, + 'job_batch_' || :identifier || '_' || i, + 'doc_batch_' || :identifier || '_' || i, + 'inline', + NOW(), + NOW() + FROM generate_series(1, :document_count) AS values(i) + """, + {"identifier": identifier, "document_count": document_count}, + ) + await ContractDatabase.execute( + """ + UPDATE documents + SET current_job_result_id = 'result_batch_' || :identifier || '_' || i + FROM generate_series(1, :document_count) AS values(i) + WHERE document_id = 'doc_batch_' || :identifier || '_' || i + """, + {"identifier": identifier, "document_count": document_count}, + ) + await ContractDatabase.execute( + """ + INSERT INTO document_sections ( + section_id, user_id, namespace, document_id, job_result_id, + section_path, section_title, section_level, sort_order, created_at + ) + SELECT + 'section_batch_' || :identifier || '_' || i, + :user_id, + :namespace, + 'doc_batch_' || :identifier || '_' || i, + 'result_batch_' || :identifier || '_' || i, + 'batch-' || i || '.pdf/section', + 'section', + 1, + 1, + NOW() + FROM generate_series(1, :document_count) AS values(i) + """, + { + "identifier": identifier, + "user_id": _USER_ID, + "namespace": namespace, + "document_count": document_count, + }, + ) + await ContractDatabase.execute( + """ + INSERT INTO document_chunks ( + id, chunk_id, user_id, namespace, document_id, job_result_id, + section_id, chunk_type, content, source_chunk_path, + chunk_metadata, sort_order, created_at + ) + SELECT + 'dchunk_batch_' || :identifier || '_' || i, + 'chunk_batch_' || :identifier || '_' || i, + :user_id, + :namespace, + 'doc_batch_' || :identifier || '_' || i, + 'result_batch_' || :identifier || '_' || i, + 'section_batch_' || :identifier || '_' || i, + 'text', + 'content-' || i, + 'batch-' || i || '.pdf/section/chunk', + '{}'::json, + 1, + NOW() + FROM generate_series(1, :document_count) AS values(i) + """, + { + "identifier": identifier, + "user_id": _USER_ID, + "namespace": namespace, + "document_count": document_count, + }, + ) + + +async def test_snapshot_batches_chunks_across_many_small_documents( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + document_count = 25 + namespace = f"batch-documents-{uuid4().hex[:8]}" + async with developer_api_client_factory(): + await _seed_many_small_documents(namespace, document_count=document_count) + async with contract_db_session() as db: + counting_db = _CountingSession(db) + snapshot = await load_nav_snapshot( + counting_db, + user_id=_USER_ID, + namespace=namespace, + ) + + assert len(snapshot.document_ids) == document_count + assert counting_db.chunk_query_count == 1 diff --git a/apps/api/tests/contract/test_retrieval_snapshot_consistency_contract.py b/apps/api/tests/contract/test_retrieval_snapshot_consistency_contract.py new file mode 100644 index 000000000..64aaf4c0b --- /dev/null +++ b/apps/api/tests/contract/test_retrieval_snapshot_consistency_contract.py @@ -0,0 +1,182 @@ +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from contextlib import AbstractAsyncContextManager +from typing import cast +from uuid import uuid4 + +from httpx import AsyncClient +from sqlalchemy import Executable, Result + +from shared.services.retrieval.nav_snapshot import SnapshotSession, load_nav_snapshot +from tests.support.retrieval_snapshot_support import contract_db_session +from tests.support.contract_database import ContractDatabase + + +_USER_ID = "local-dev-user" + + +class _PublishingSession: + def __init__( + self, + session: SnapshotSession, + publish_revision: Callable[[], Awaitable[None]], + ) -> None: + self._session = session + self._publish_revision = publish_revision + self._has_published = False + + async def execute(self, statement: Executable) -> Result[tuple[object, ...]]: + result = await self._session.execute(statement) + if not self._has_published: + self._has_published = True + await self._publish_revision() + return cast(Result[tuple[object, ...]], result) + + +async def _seed_republished_document(namespace: str) -> tuple[str, str]: + identifier = uuid4().hex[:8] + document_id = f"doc_race_{identifier}" + old_result_id = f"result_race_old_{identifier}" + new_result_id = f"result_race_new_{identifier}" + await ContractDatabase.execute( + """ + INSERT INTO jobs ( + job_id, user_id, job_type, status, source_type, version, + webhook_enabled, created_at, updated_at, credits_charged, billing_status + ) VALUES + (:old_job_id, :user_id, 'document_ingestion', 'done', 'file', 0, + false, NOW(), NOW(), 0, 'skipped'), + (:new_job_id, :user_id, 'document_ingestion', 'done', 'file', 0, + false, NOW(), NOW(), 0, 'skipped') + """, + { + "old_job_id": f"job_race_old_{identifier}", + "new_job_id": f"job_race_new_{identifier}", + "user_id": _USER_ID, + }, + ) + await ContractDatabase.execute( + """ + INSERT INTO documents ( + document_id, user_id, namespace, status, current_job_result_id, + source_file_name, parse_track, created_at, updated_at + ) VALUES ( + :document_id, :user_id, :namespace, 'active', NULL, + 'republished.pdf', 'chunk', NOW(), NOW() + ) + """, + {"document_id": document_id, "user_id": _USER_ID, "namespace": namespace}, + ) + await ContractDatabase.execute( + """ + INSERT INTO job_results ( + id, job_id, document_id, delivery_mode, created_at, updated_at + ) VALUES + (:old_result_id, :old_job_id, :document_id, 'inline', NOW(), NOW()), + (:new_result_id, :new_job_id, :document_id, 'inline', NOW(), NOW()) + """, + { + "old_result_id": old_result_id, + "new_result_id": new_result_id, + "old_job_id": f"job_race_old_{identifier}", + "new_job_id": f"job_race_new_{identifier}", + "document_id": document_id, + }, + ) + await ContractDatabase.execute( + """ + INSERT INTO document_sections ( + section_id, user_id, namespace, document_id, job_result_id, + section_path, section_title, section_level, sort_order, created_at + ) VALUES + (:old_section_id, :user_id, :namespace, :document_id, :old_result_id, + 'republished.pdf/old', 'old', 1, 1, NOW()), + (:new_section_id, :user_id, :namespace, :document_id, :new_result_id, + 'republished.pdf/new', 'new', 1, 1, NOW()) + """, + { + "old_section_id": f"section_race_old_{identifier}", + "new_section_id": f"section_race_new_{identifier}", + "user_id": _USER_ID, + "namespace": namespace, + "document_id": document_id, + "old_result_id": old_result_id, + "new_result_id": new_result_id, + }, + ) + await ContractDatabase.execute( + """ + INSERT INTO document_chunks ( + id, chunk_id, user_id, namespace, document_id, job_result_id, + section_id, chunk_type, content, source_chunk_path, + chunk_metadata, sort_order, created_at + ) VALUES + (:old_id, :old_chunk_id, :user_id, :namespace, :document_id, + :old_result_id, :old_section_id, 'text', 'old content', + 'republished.pdf/old/chunk', '{}'::json, 1, NOW()), + (:new_id, :new_chunk_id, :user_id, :namespace, :document_id, + :new_result_id, :new_section_id, 'text', 'new content', + 'republished.pdf/new/chunk', '{}'::json, 1, NOW()) + """, + { + "old_id": f"dchunk_race_old_{identifier}", + "new_id": f"dchunk_race_new_{identifier}", + "old_chunk_id": f"chunk_race_old_{identifier}", + "new_chunk_id": f"chunk_race_new_{identifier}", + "user_id": _USER_ID, + "namespace": namespace, + "document_id": document_id, + "old_result_id": old_result_id, + "new_result_id": new_result_id, + "old_section_id": f"section_race_old_{identifier}", + "new_section_id": f"section_race_new_{identifier}", + }, + ) + await ContractDatabase.execute( + """ + UPDATE documents + SET current_job_result_id = :old_result_id + WHERE document_id = :document_id + """, + {"old_result_id": old_result_id, "document_id": document_id}, + ) + return document_id, new_result_id + + +async def test_snapshot_keeps_sections_and_chunks_on_the_captured_revision( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + namespace = f"revision-race-{uuid4().hex[:8]}" + async with developer_api_client_factory(): + document_id, new_result_id = await _seed_republished_document(namespace) + + async def publish_new_revision() -> None: + await ContractDatabase.execute( + """ + UPDATE documents + SET current_job_result_id = :new_result_id + WHERE document_id = :document_id + """, + {"new_result_id": new_result_id, "document_id": document_id}, + ) + + async with contract_db_session() as db: + snapshot = await load_nav_snapshot( + _PublishingSession(db, publish_new_revision), + user_id=_USER_ID, + namespace=namespace, + ) + + section_ids = list(snapshot.provider.children(document_id)) + chunks = [ + chunk + for section_id in section_ids + for chunk in snapshot.provider.self_units(section_id) + ] + assert [chunk.content for chunk in chunks] == ["old content"] + assert snapshot.chunk_ref_index[chunks[0].chunk_id]["section_path"] == ( + "republished.pdf/old" + ) diff --git a/apps/api/tests/contract/test_retrieval_snapshot_large_corpus_contract.py b/apps/api/tests/contract/test_retrieval_snapshot_large_corpus_contract.py new file mode 100644 index 000000000..a414e44b3 --- /dev/null +++ b/apps/api/tests/contract/test_retrieval_snapshot_large_corpus_contract.py @@ -0,0 +1,344 @@ +from __future__ import annotations + +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager +from typing import cast +from uuid import uuid4 + +from httpx import AsyncClient + +from shared.models.database.document import Document, DocumentChunk, DocumentSection +from shared.models.database.job_result import JobResult +from shared.services.retrieval.nav_snapshot import ( + SnapshotSession, + _CHUNK_BATCH_SIZE, + _REVISION_GROUP_SIZE, + load_nav_snapshot, +) +from sqlalchemy import Executable, Result, select +from sqlalchemy.engine import Row +from sqlalchemy.sql.selectable import Select +from tests.support.retrieval_snapshot_support import contract_db_session +from tests.support.contract_database import ContractDatabase + + +_USER_ID = "local-dev-user" +_DOCUMENT_COUNT = 100 +_CHUNKS_PER_DOCUMENT = 600 +_SECTIONS_PER_DOCUMENT = 8 +_TOTAL_CHUNKS = _DOCUMENT_COUNT * _CHUNKS_PER_DOCUMENT +LegacySnapshotRow = Row[ + tuple[ + str, + str, + str | None, + str, + str, + int, + str, + str | None, + dict[str, object], + str | None, + str | None, + ] +] + + +class _CountingSession: + def __init__(self, session: SnapshotSession) -> None: + self._session = session + self.chunk_query_count = 0 + + async def execute(self, statement: Executable) -> Result[tuple[object, ...]]: + if isinstance(statement, Select): + selected_tables = { + getattr(table, "name", "") + for table in statement.get_final_froms() + } + else: + selected_tables = set() + if "document_chunks" in selected_tables: + self.chunk_query_count += 1 + result = await self._session.execute(statement) + return cast(Result[tuple[object, ...]], result) + + +async def _seed_large_retrieval_corpus(namespace: str) -> None: + await ContractDatabase.execute( + """ + INSERT INTO jobs ( + job_id, user_id, job_type, status, source_type, version, + webhook_enabled, created_at, updated_at, credits_charged, billing_status + ) + SELECT + 'job_lg_' || i, + :user_id, + 'document_ingestion', + 'done', + 'file', + 0, + false, + NOW(), + NOW(), + 0, + 'skipped' + FROM generate_series(1, :document_count) AS values(i) + """, + {"user_id": _USER_ID, "document_count": _DOCUMENT_COUNT}, + ) + await ContractDatabase.execute( + """ + INSERT INTO documents ( + document_id, user_id, namespace, status, current_job_result_id, + source_file_name, parse_track, created_at, updated_at + ) + SELECT + 'doc_lg_' || i, + :user_id, + :namespace, + 'active', + NULL, + 'large-corpus-' || i || '.pdf', + 'chunk', + NOW(), + NOW() + FROM generate_series(1, :document_count) AS values(i) + """, + { + "user_id": _USER_ID, + "namespace": namespace, + "document_count": _DOCUMENT_COUNT, + }, + ) + await ContractDatabase.execute( + """ + INSERT INTO job_results ( + id, job_id, document_id, delivery_mode, created_at, updated_at + ) + SELECT + 'result_lg_' || i, + 'job_lg_' || i, + 'doc_lg_' || i, + 'inline', + NOW(), + NOW() + FROM generate_series(1, :document_count) AS values(i) + """, + {"document_count": _DOCUMENT_COUNT}, + ) + await ContractDatabase.execute( + """ + UPDATE documents + SET current_job_result_id = 'result_lg_' || i + FROM generate_series(1, :document_count) AS values(i) + WHERE documents.document_id = 'doc_lg_' || i + """, + {"document_count": _DOCUMENT_COUNT}, + ) + await ContractDatabase.execute( + """ + INSERT INTO document_sections ( + section_id, user_id, namespace, document_id, job_result_id, + section_path, section_title, section_level, sort_order, created_at + ) + SELECT + 'section_lg_' || i || '_' || section_number, + :user_id, + :namespace, + 'doc_lg_' || i, + 'result_lg_' || i, + 'large-corpus-' || i || '/section/' || section_number, + 'section-' || section_number, + 1, + section_number, + NOW() + FROM generate_series(1, :document_count) AS values(i) + CROSS JOIN generate_series(1, :sections_per_document) AS sections(section_number) + """, + { + "user_id": _USER_ID, + "namespace": namespace, + "document_count": _DOCUMENT_COUNT, + "sections_per_document": _SECTIONS_PER_DOCUMENT, + }, + ) + await ContractDatabase.execute( + """ + INSERT INTO document_chunks ( + id, chunk_id, user_id, namespace, document_id, job_result_id, + section_id, chunk_type, content, content_lexical_text, + path_lexical_text, content_search_text, path_search_text, + term_search_text, source_chunk_path, chunk_metadata, sort_order, + created_at + ) + SELECT + 'dchunk_lg_' || document_number || '_' || chunk_number, + 'chunk_lg_' || document_number || '_' || chunk_number, + :user_id, + :namespace, + 'doc_lg_' || document_number, + 'result_lg_' || document_number, + 'section_lg_' || document_number || '_' || section_number, + 'text', + repeat(md5(document_number::text || ':' || chunk_number::text), 64), + repeat(md5(document_number::text || ':' || chunk_number::text), 64), + 'large-corpus-' || document_number || '/section/' || section_number || '/' || chunk_number, + repeat(md5(document_number::text || ':' || chunk_number::text), 64), + 'large-corpus-' || document_number || '/section/' || section_number || '/' || chunk_number, + repeat(md5(document_number::text || ':' || chunk_number::text), 64), + 'large-corpus-' || document_number || '/section/' || section_number || '/' || chunk_number, + json_build_object( + 'tokens', ARRAY['benchmark', 'retrieval', 'document', document_number::text], + 'keywords', ARRAY['benchmark', 'production-shaped'], + 'summary', repeat('payload ', 32) + )::json, + chunk_number, + NOW() + FROM generate_series(1, :document_count) AS documents(document_number) + CROSS JOIN generate_series(1, :chunks_per_document) AS chunks(chunk_number) + CROSS JOIN LATERAL ( + SELECT ((chunk_number - 1) % :sections_per_document) + 1 AS section_number + ) AS section_values + """, + { + "user_id": _USER_ID, + "namespace": namespace, + "document_count": _DOCUMENT_COUNT, + "chunks_per_document": _CHUNKS_PER_DOCUMENT, + "sections_per_document": _SECTIONS_PER_DOCUMENT, + }, + ) + + +async def _load_legacy_rows( + namespace: str, +) -> list[LegacySnapshotRow]: + stmt = ( + select( + Document.document_id, + DocumentChunk.chunk_id, + DocumentChunk.section_id, + DocumentChunk.chunk_type, + DocumentChunk.content, + DocumentChunk.sort_order, + DocumentChunk.source_chunk_path, + DocumentChunk.file_path, + DocumentChunk.chunk_metadata, + DocumentSection.section_path, + JobResult.job_id, + ) + .join( + DocumentChunk, + (DocumentChunk.document_id == Document.document_id) + & (DocumentChunk.job_result_id == Document.current_job_result_id), + ) + .outerjoin( + DocumentSection, DocumentSection.section_id == DocumentChunk.section_id + ) + .outerjoin(JobResult, JobResult.id == DocumentChunk.job_result_id) + .where(Document.user_id == _USER_ID) + .where(Document.namespace == namespace) + .where(Document.status == "active") + .order_by( + Document.document_id, DocumentChunk.sort_order, DocumentChunk.chunk_id + ) + ) + async with contract_db_session() as db: + rows: list[LegacySnapshotRow] = list((await db.execute(stmt)).all()) + return rows + + +async def test_large_snapshot_keeps_all_retrieval_inputs_after_bounded_sql_load( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + namespace = f"large-corpus-{uuid4().hex[:8]}" + async with developer_api_client_factory(): + await _seed_large_retrieval_corpus(namespace) + legacy_rows = await _load_legacy_rows(namespace) + async with contract_db_session() as db: + counting_db = _CountingSession(db) + snapshot = await load_nav_snapshot( + counting_db, + user_id=_USER_ID, + namespace=namespace, + ) + expected_chunk_query_count = sum( + ( + min(_REVISION_GROUP_SIZE, _DOCUMENT_COUNT - group_start) + * _CHUNKS_PER_DOCUMENT + + _CHUNK_BATCH_SIZE + - 1 + ) + // _CHUNK_BATCH_SIZE + for group_start in range(0, _DOCUMENT_COUNT, _REVISION_GROUP_SIZE) + ) + assert counting_db.chunk_query_count == expected_chunk_query_count + + async with contract_db_session() as db: + bounded_snapshot = await load_nav_snapshot( + db, + user_id=_USER_ID, + namespace=namespace, + ) + + assert len(legacy_rows) == _TOTAL_CHUNKS + assert len(snapshot.document_ids) == _DOCUMENT_COUNT + assert len(bounded_snapshot.document_ids) == _DOCUMENT_COUNT + + optimized_rows = [ + ( + document_id, + chunk.chunk_id, + chunk.section_id, + chunk.chunk_type, + chunk.content, + chunk.sort_order, + chunk.source_chunk_path, + chunk.file_path, + chunk.metadata, + snapshot.chunk_ref_index[f"{document_id}:{chunk.chunk_id}"][ + "section_path" + ], + snapshot.chunk_ref_index[f"{document_id}:{chunk.chunk_id}"]["job_id"], + ) + for document_id in snapshot.document_ids + for section_id in snapshot.provider.children(document_id) + for chunk in snapshot.provider.self_units(section_id) + ] + optimized_rows.sort(key=lambda row: (row[0], row[5], row[1], row[2] or "")) + legacy_rows_projected = [ + ( + str(row[0]), + str(row[1]), + str(row[2]) if row[2] else None, + str(row[3]), + str(row[4]), + int(row[5]), + str(row[6]), + str(row[7] or ""), + row[8] if isinstance(row[8], dict) else {}, + str(row[9] or ""), + str(row[10]) if row[10] else None, + ) + for row in legacy_rows + ] + def row_order_key(row: tuple[object, ...]) -> tuple[str, ...]: + return ( + str(row[0]), + str(row[5]), + str(row[1]), + str(row[2] or ""), + str(row[3]), + str(row[4]), + str(row[6]), + str(row[7] or ""), + str(row[8]), + str(row[9] or ""), + str(row[10] or ""), + ) + legacy_rows_projected.sort(key=row_order_key) + optimized_rows.sort(key=row_order_key) + assert len(optimized_rows) == _TOTAL_CHUNKS + assert optimized_rows == legacy_rows_projected diff --git a/apps/api/tests/migrations/test_schema_contract.py b/apps/api/tests/migrations/test_schema_contract.py index 830a2cbd1..24047dccd 100644 --- a/apps/api/tests/migrations/test_schema_contract.py +++ b/apps/api/tests/migrations/test_schema_contract.py @@ -31,11 +31,24 @@ def _build_alembic_command_config(*, engine: Engine) -> Config: def _upgrade_to_heads(*, engine: Engine) -> None: config = _build_alembic_command_config(engine=engine) + # Let Alembic create its own connection. This is required for migrations + # that use PostgreSQL autocommit (for example CREATE INDEX CONCURRENTLY). + command.upgrade(config, "heads") + + +def _upgrade_to_heads_with_external_connection(*, engine: Engine) -> None: + config = _build_alembic_command_config(engine=engine) with engine.begin() as connection: config.attributes["connection"] = connection command.upgrade(config, "heads") +def _upgrade_to_snapshot_parents(*, engine: Engine) -> None: + config = _build_alembic_command_config(engine=engine) + command.upgrade(config, "f0d85d209e68") + command.upgrade(config, "fbe1c2d3e4f5") + + def _insert_job( connection: Connection, *, @@ -199,6 +212,73 @@ def test_should_seed_v2_job_polling_system_limit( assert result["description"] == "Job queries - prevent polling" +def test_should_index_document_chunks_in_snapshot_pagination_order( + migrated_head_engine: Engine, +) -> None: + with migrated_head_engine.begin() as connection: + index_definition = connection.execute( + text( + """ + SELECT indexdef + FROM pg_indexes + WHERE schemaname = current_schema() + AND tablename = 'document_chunks' + AND indexname = 'idx_document_chunks_revision_snapshot_order' + """ + ) + ).scalar_one() + + assert "(document_id, job_result_id, sort_order, chunk_id, id)" in str( + index_definition + ) + + +def test_should_upgrade_with_a_caller_owned_connection( + alembic_engine: Engine, +) -> None: + _upgrade_to_heads_with_external_connection(engine=alembic_engine) + + +def test_standalone_upgrade_should_preserve_a_caller_owned_connection( + standalone_alembic_engine: Engine, +) -> None: + _upgrade_to_heads_with_external_connection(engine=standalone_alembic_engine) + + +def test_should_replace_a_same_named_index_with_the_wrong_definition( + alembic_engine: Engine, +) -> None: + _upgrade_to_snapshot_parents(engine=alembic_engine) + with alembic_engine.begin() as connection: + connection.execute( + text( + """ + CREATE INDEX idx_document_chunks_revision_snapshot_order + ON document_chunks (document_id) + """ + ) + ) + + _upgrade_to_heads(engine=alembic_engine) + + with alembic_engine.begin() as connection: + index_definition = connection.execute( + text( + """ + SELECT indexdef + FROM pg_indexes + WHERE schemaname = current_schema() + AND tablename = 'document_chunks' + AND indexname = 'idx_document_chunks_revision_snapshot_order' + """ + ) + ).scalar_one() + + assert "(document_id, job_result_id, sort_order, chunk_id, id)" in str( + index_definition + ) + + def test_api_standalone_mode_should_create_auth_user_table_before_migrations( standalone_alembic_engine: Engine, ) -> None: diff --git a/apps/api/tests/support/retrieval_snapshot_support.py b/apps/api/tests/support/retrieval_snapshot_support.py new file mode 100644 index 000000000..b06b55f15 --- /dev/null +++ b/apps/api/tests/support/retrieval_snapshot_support.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager + +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from shared.testing.contract_runtime import get_contract_database_url + + +@asynccontextmanager +async def contract_db_session() -> AsyncGenerator[AsyncSession, None]: + engine = create_async_engine(get_contract_database_url(), future=True) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with session_factory() as session: + yield session + finally: + await engine.dispose() diff --git a/packages/shared-python/shared/core/response/ErrorCode.py b/packages/shared-python/shared/core/response/ErrorCode.py index 4c9820936..cf5dd2f8f 100644 --- a/packages/shared-python/shared/core/response/ErrorCode.py +++ b/packages/shared-python/shared/core/response/ErrorCode.py @@ -55,6 +55,7 @@ class ErrorCode(str, Enum): ) PERMISSION_DENIED = "PERMISSION_DENIED" # 403 - Caller lacks permission NOT_FOUND = "NOT_FOUND" # 404 - Resource does not exist + METHOD_NOT_ALLOWED = "METHOD_NOT_ALLOWED" # 405 - HTTP method is not supported ABORTED = "ABORTED" # 409 - Concurrency conflict ALREADY_EXISTS = "ALREADY_EXISTS" # 409 - Resource already exists RESOURCE_EXHAUSTED = ( @@ -89,6 +90,7 @@ class ErrorCodeMapper: ErrorCode.PAYMENT_REQUIRED: 402, ErrorCode.PERMISSION_DENIED: 403, ErrorCode.NOT_FOUND: 404, + ErrorCode.METHOD_NOT_ALLOWED: 405, ErrorCode.ABORTED: 409, ErrorCode.ALREADY_EXISTS: 409, ErrorCode.RESOURCE_EXHAUSTED: 429, @@ -110,6 +112,7 @@ class ErrorCodeMapper: 402: ErrorCode.PAYMENT_REQUIRED, 403: ErrorCode.PERMISSION_DENIED, 404: ErrorCode.NOT_FOUND, + 405: ErrorCode.METHOD_NOT_ALLOWED, 409: ErrorCode.ALREADY_EXISTS, 422: ErrorCode.INVALID_ARGUMENT, # Pydantic validation 429: ErrorCode.RESOURCE_EXHAUSTED, diff --git a/packages/shared-python/shared/models/database/document.py b/packages/shared-python/shared/models/database/document.py index c9af71407..f9681e178 100644 --- a/packages/shared-python/shared/models/database/document.py +++ b/packages/shared-python/shared/models/database/document.py @@ -202,6 +202,14 @@ class DocumentChunk(Base): Index("idx_document_chunks_scope", "user_id", "namespace"), Index("idx_document_chunks_chunk_id", "chunk_id"), Index("idx_document_chunks_doc_revision", "document_id", "job_result_id"), + Index( + "idx_document_chunks_revision_snapshot_order", + "document_id", + "job_result_id", + "sort_order", + "chunk_id", + "id", + ), Index("idx_document_chunks_section", "section_id"), Index( "idx_chunk_content_search_tsv", diff --git a/packages/shared-python/shared/services/retrieval/nav_snapshot.py b/packages/shared-python/shared/services/retrieval/nav_snapshot.py index 964d698b7..7dc60c6f3 100644 --- a/packages/shared-python/shared/services/retrieval/nav_snapshot.py +++ b/packages/shared-python/shared/services/retrieval/nav_snapshot.py @@ -8,10 +8,10 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Optional +from typing import Any, Protocol -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import Executable, literal, select, tuple_ +from sqlalchemy.engine import Result from shared.models.database.document import Document, DocumentChunk, DocumentSection from shared.models.database.job_result import JobResult @@ -24,6 +24,21 @@ from shared.services.retrieval.search.section_filters import is_excluded_section +# Keep each payload query bounded under the API's 30-second statement timeout. +# Ten-thousand-row keyset pages avoid OFFSET scans while keeping each payload +# statement bounded. The contract benchmark verifies this page size against +# the full 2 KiB content and metadata payload. +_CHUNK_BATCH_SIZE = 10_000 +_REVISION_GROUP_SIZE = 32 + + +class SnapshotSession(Protocol): + """Minimal database interface required by the snapshot loader.""" + + async def execute(self, statement: Executable) -> Result[tuple[object, ...]]: + raise NotImplementedError + + @dataclass(frozen=True) class NavSnapshot: """In-memory corpus for one map-nav episode.""" @@ -65,7 +80,7 @@ def build_nav_snapshot( async def load_nav_snapshot( - db: AsyncSession, + db: SnapshotSession, *, user_id: str, namespace: str, @@ -94,25 +109,39 @@ async def load_nav_snapshot( ) document_titles: dict[str, str] = {} - for document_id, source_file_name, _job in doc_rows: + current_job_result_ids: set[str] = set() + document_revisions: list[tuple[str, str]] = [] + for document_id, source_file_name, current_job_result_id in doc_rows: did = str(document_id) title = str(source_file_name or "").strip() or did document_titles[did] = title + if current_job_result_id: + job_result_id = str(current_job_result_id) + current_job_result_ids.add(job_result_id) + document_revisions.append((did, job_result_id)) + + job_result_rows = await db.execute( + select(JobResult.id, JobResult.job_id).where( + JobResult.id.in_(list(current_job_result_ids)) + ) + ) + job_id_by_result_id = { + str(job_result_id): str(job_id) + for job_result_id, job_id in job_result_rows.all() + if job_result_id and job_id + } sections_by_doc, section_path_by_id = await _load_sections( db, - user_id=user_id, - namespace=namespace, - exclude_document_ids=excluded_docs, + document_revisions=document_revisions, exclude_sections=excluded_secs, ) units_by_doc, chunk_ref_index = await _load_chunks( db, - user_id=user_id, - namespace=namespace, - exclude_document_ids=excluded_docs, + document_revisions=document_revisions, exclude_sections=excluded_secs, section_path_by_id=section_path_by_id, + job_id_by_result_id=job_id_by_result_id, ) # Keep only documents that still have sections after exclude filters. @@ -136,16 +165,15 @@ async def load_nav_snapshot( async def _load_sections( - db: AsyncSession, + db: SnapshotSession, *, - user_id: str, - namespace: str, - exclude_document_ids: list[str], + document_revisions: list[tuple[str, str]], exclude_sections: list[dict[str, str]], ) -> tuple[dict[str, list[SectionRow]], dict[str, str]]: + # Captured pairs replace DocumentSection.job_result_id == Document.current_job_result_id. stmt = ( select( - Document.document_id, + DocumentSection.document_id, DocumentSection.section_id, DocumentSection.parent_section_id, DocumentSection.section_path, @@ -154,18 +182,18 @@ async def _load_sections( DocumentSection.summary, DocumentSection.sort_order, ) - .join( - DocumentSection, - (DocumentSection.document_id == Document.document_id) - & (DocumentSection.job_result_id == Document.current_job_result_id), + .where( + tuple_( + DocumentSection.document_id, + DocumentSection.job_result_id, + ).in_(document_revisions) + ) + .order_by( + DocumentSection.document_id, + DocumentSection.sort_order, + DocumentSection.section_id, ) - .where(Document.user_id == user_id) - .where(Document.namespace == namespace) - .where(Document.status == "active") - .order_by(Document.document_id, DocumentSection.sort_order, DocumentSection.section_id) ) - if exclude_document_ids: - stmt = stmt.where(Document.document_id.notin_(list(exclude_document_ids))) by_doc: dict[str, list[SectionRow]] = {} path_by_id: dict[str, str] = {} @@ -194,89 +222,123 @@ async def _load_sections( async def _load_chunks( - db: AsyncSession, + db: SnapshotSession, *, - user_id: str, - namespace: str, - exclude_document_ids: list[str], + document_revisions: list[tuple[str, str]], exclude_sections: list[dict[str, str]], section_path_by_id: dict[str, str], + job_id_by_result_id: dict[str, str], ) -> tuple[dict[str, list[UnitRow]], dict[str, dict[str, Any]]]: - stmt = ( - select( - Document.document_id, - DocumentChunk.chunk_id, - DocumentChunk.section_id, - DocumentChunk.chunk_type, - DocumentChunk.content, - DocumentChunk.sort_order, - DocumentChunk.source_chunk_path, - DocumentChunk.file_path, - DocumentChunk.chunk_metadata, - DocumentSection.section_path, - JobResult.job_id, - ) - .join( - DocumentChunk, - (DocumentChunk.document_id == Document.document_id) - & (DocumentChunk.job_result_id == Document.current_job_result_id), - ) - .outerjoin( - DocumentSection, - DocumentSection.section_id == DocumentChunk.section_id, - ) - .outerjoin(JobResult, JobResult.id == DocumentChunk.job_result_id) - .where(Document.user_id == user_id) - .where(Document.namespace == namespace) - .where(Document.status == "active") - .order_by(Document.document_id, DocumentChunk.sort_order, DocumentChunk.chunk_id) - ) - if exclude_document_ids: - stmt = stmt.where(Document.document_id.notin_(list(exclude_document_ids))) - + # Captured pairs replace DocumentChunk.document_id == document_id and + # DocumentChunk.job_result_id == job_result_id. by_doc: dict[str, list[UnitRow]] = {} ref_index: dict[str, dict[str, Any]] = {} - for row in (await db.execute(stmt)).all(): - document_id = str(row[0]) - chunk_id = str(row[1] or "").strip() - section_id = str(row[2]) if row[2] else None - # Prefer joined section_path; fall back to kept section map. - section_path: Optional[str] = str(row[9]) if row[9] is not None else None - if section_path is None and section_id: - section_path = section_path_by_id.get(section_id) - if is_excluded_section( - document_id=document_id, - section_path=section_path, - exclude_sections=exclude_sections, - ): - continue - # Drop units whose section was filtered out of the tree. - if section_id and section_id not in section_path_by_id: - continue + for group_start in range(0, len(document_revisions), _REVISION_GROUP_SIZE): + revision_group = document_revisions[ + group_start : group_start + _REVISION_GROUP_SIZE + ] + last_key: tuple[str, str, int, str, str] | None = None + while True: + stmt = ( + select( + DocumentChunk.document_id, + DocumentChunk.job_result_id, + DocumentChunk.chunk_id, + DocumentChunk.section_id, + DocumentChunk.chunk_type, + DocumentChunk.content, + DocumentChunk.sort_order, + DocumentChunk.source_chunk_path, + DocumentChunk.file_path, + DocumentChunk.chunk_metadata, + DocumentChunk.id, + ) + .where( + tuple_( + DocumentChunk.document_id, + DocumentChunk.job_result_id, + ).in_(revision_group) + ) + .order_by( + DocumentChunk.document_id, + DocumentChunk.job_result_id, + DocumentChunk.sort_order, + DocumentChunk.chunk_id, + DocumentChunk.id, + ) + .limit(_CHUNK_BATCH_SIZE) + ) + if last_key is not None: + stmt = stmt.where( + tuple_( + DocumentChunk.document_id, + DocumentChunk.job_result_id, + DocumentChunk.sort_order, + DocumentChunk.chunk_id, + DocumentChunk.id, + ) + > tuple_( + literal(last_key[0]), + literal(last_key[1]), + literal(last_key[2]), + literal(last_key[3]), + literal(last_key[4]), + ) + ) - unit = UnitRow( - chunk_id=chunk_id, - section_id=section_id, - chunk_type=str(row[3] or "text"), - content=str(row[4] or ""), - sort_order=int(row[5] or 0), - source_chunk_path=str(row[6] or ""), - file_path=str(row[7] or ""), - metadata=_as_meta(row[8]), - ) - by_doc.setdefault(document_id, []).append(unit) - if chunk_id: - meta = { - "document_id": document_id, - "section_path": section_path, - "chunk_type": unit.chunk_type, - "file_path": unit.file_path or None, - "job_id": str(row[10]) if row[10] else None, - } - # Bare chunk_id (last-wins) plus doc-scoped key so the same - # chunk_id can appear under multiple documents. - ref_index[chunk_id] = meta - ref_index[f"{document_id}:{chunk_id}"] = meta + rows = (await db.execute(stmt)).all() + if not rows: + break + for row in rows: + document_id = str(row[0]) + job_result_id = str(row[1]) + chunk_id = str(row[2] or "").strip() + section_id = str(row[3]) if row[3] else None + section_path: str | None = ( + section_path_by_id.get(section_id) if section_id else None + ) + if is_excluded_section( + document_id=document_id, + section_path=section_path, + exclude_sections=exclude_sections, + ): + continue + if section_id and section_id not in section_path_by_id: + continue + + unit = UnitRow( + chunk_id=chunk_id, + section_id=section_id, + chunk_type=str(row[4] or "text"), + content=str(row[5] or ""), + sort_order=int(row[6] or 0), + source_chunk_path=str(row[7] or ""), + file_path=str(row[8] or ""), + metadata=_as_meta(row[9]), + ) + by_doc.setdefault(document_id, []).append(unit) + if chunk_id: + meta = { + "document_id": document_id, + "section_path": section_path, + "chunk_type": unit.chunk_type, + "file_path": unit.file_path or None, + "job_id": job_id_by_result_id.get(job_result_id), + } + # Bare chunk_id (last-wins) plus doc-scoped key so the same + # chunk_id can appear under multiple documents. + ref_index[chunk_id] = meta + ref_index[f"{document_id}:{chunk_id}"] = meta + last_row = rows[-1] + last_key = ( + str(last_row[0]), + str(last_row[1]), + int(last_row[6] or 0), + str(last_row[2] or ""), + str(last_row[10]), + ) + if len(rows) < _CHUNK_BATCH_SIZE: + break return by_doc, ref_index diff --git a/packages/shared-python/shared/tests/test_nav_snapshot.py b/packages/shared-python/shared/tests/test_nav_snapshot.py index 340c70687..f8bf76f61 100644 --- a/packages/shared-python/shared/tests/test_nav_snapshot.py +++ b/packages/shared-python/shared/tests/test_nav_snapshot.py @@ -93,15 +93,3 @@ def test_build_nav_snapshot_rejects_empty_corpus() -> None: units_by_doc={}, chunk_ref_index={}, ) - - -def test_load_nav_snapshot_joins_current_revision_only() -> None: - """Section/chunk loaders must bind rows to Document.current_job_result_id.""" - import inspect - - from shared.services.retrieval import nav_snapshot as nav_snapshot_mod - - sections_src = "".join(inspect.getsource(nav_snapshot_mod._load_sections).split()) - chunks_src = "".join(inspect.getsource(nav_snapshot_mod._load_chunks).split()) - assert "DocumentSection.job_result_id==Document.current_job_result_id" in sections_src - assert "DocumentChunk.job_result_id==Document.current_job_result_id" in chunks_src