From e0d5aab2549a7e88e028f5688aac39fda403fd96 Mon Sep 17 00:00:00 2001 From: andyne13 Date: Fri, 19 Jun 2026 17:36:58 +0200 Subject: [PATCH 1/4] feat(files): persist + expose indexation timestamp (files.created_at) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The documents list reads from the Postgres files catalog, which had no timestamp column, so the admin UI "Indexed" column was always empty — the index time existed only in Milvus chunk metadata (shown in the file detail). Add a files.created_at column (timezone-aware, server_default now()), surface it in the file-listing dict (_row_to_dict, placed after the metadata spread so the column wins), and add an idempotent migration. Existing rows backfill to the migration run time via the server default; newly indexed files get their true insert time. The admin UI already renders this column. --- openrag/services/persistence/document_repo.py | 4 ++ .../b7c1d2e3f4a5_add_files_created_at.py | 46 +++++++++++++++++++ openrag/services/persistence/schema.py | 6 +++ 3 files changed, 56 insertions(+) create mode 100644 openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_created_at.py diff --git a/openrag/services/persistence/document_repo.py b/openrag/services/persistence/document_repo.py index 85e8d23f3..5abb4dcf0 100644 --- a/openrag/services/persistence/document_repo.py +++ b/openrag/services/persistence/document_repo.py @@ -550,12 +550,16 @@ def _row_to_dict(row: asyncpg.Record) -> dict: key flattened in). Used by the shim's pass-through calls. """ metadata = row["file_metadata"] or {} + created_at = row["created_at"] return { "partition": row["partition_name"], "file_id": row["file_id"], "relationship_id": row["relationship_id"], "parent_id": row["parent_id"], **metadata, + # Indexation timestamp lives on the row, not in file_metadata; + # placed after the spread so the column value always wins. + "created_at": created_at.isoformat() if created_at else None, } @staticmethod diff --git a/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_created_at.py b/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_created_at.py new file mode 100644 index 000000000..1105ef1ab --- /dev/null +++ b/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_created_at.py @@ -0,0 +1,46 @@ +"""add files.created_at + +Revision ID: b7c1d2e3f4a5 +Revises: 06dd2101ea3a +Create Date: 2026-06-19 00:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from schema_helpers import column_exists + +# revision identifiers, used by Alembic. +revision: str = "b7c1d2e3f4a5" +down_revision: str | Sequence[str] | None = "06dd2101ea3a" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Add the indexation timestamp to ``files``. + + Idempotent: ``Base.metadata.create_all()`` at app startup may have already + added the column from the SQLAlchemy model on existing deployments. + + Existing rows have no recorded index time, so the ``server_default`` backfills + them to the migration run time; newly indexed files get their true insert time. + """ + if not column_exists("files", "created_at"): + op.add_column( + "files", + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.text("now()"), + ), + ) + + +def downgrade() -> None: + """Downgrade schema.""" + if column_exists("files", "created_at"): + op.drop_column("files", "created_at") diff --git a/openrag/services/persistence/schema.py b/openrag/services/persistence/schema.py index a5d1467e5..948d215d4 100644 --- a/openrag/services/persistence/schema.py +++ b/openrag/services/persistence/schema.py @@ -139,6 +139,12 @@ ), Column("relationship_id", String, nullable=True, index=True), Column("parent_id", String, nullable=True, index=True), + Column( + "created_at", + DateTime(timezone=True), + server_default=text("now()"), + nullable=False, + ), UniqueConstraint("file_id", "partition_name", name="uix_file_id_partition"), Index("ix_partition_file", "partition_name", "file_id"), Index("ix_relationship_partition", "relationship_id", "partition_name"), From 28e53cbe0cdfe52162e6f5911963133497dce921 Mon Sep 17 00:00:00 2001 From: Ahmath-Gadji Date: Mon, 22 Jun 2026 12:03:38 +0000 Subject: [PATCH 2/4] fix(files): use indexed_at for the indexation timestamp column created_at is a reserved, client-supplied temporal field (provided in file_metadata at upload time for time-based filtering), so the indexation timestamp must not reuse that key. Rename the new files column, its migration, and the _row_to_dict surface from created_at to indexed_at. This matches the existing chunk-level indexed_at field and the admin UI, which reads `indexed_at ?? created_at`, so the document list's Indexed column populates while the client created_at stays available for filtering. --- openrag/services/persistence/document_repo.py | 11 +++++++---- ...d_at.py => b7c1d2e3f4a5_add_files_indexed_at.py} | 13 ++++++++----- openrag/services/persistence/schema.py | 2 +- 3 files changed, 16 insertions(+), 10 deletions(-) rename openrag/services/persistence/migrations/alembic/versions/{b7c1d2e3f4a5_add_files_created_at.py => b7c1d2e3f4a5_add_files_indexed_at.py} (76%) diff --git a/openrag/services/persistence/document_repo.py b/openrag/services/persistence/document_repo.py index 5abb4dcf0..90f1f1926 100644 --- a/openrag/services/persistence/document_repo.py +++ b/openrag/services/persistence/document_repo.py @@ -550,16 +550,19 @@ def _row_to_dict(row: asyncpg.Record) -> dict: key flattened in). Used by the shim's pass-through calls. """ metadata = row["file_metadata"] or {} - created_at = row["created_at"] + indexed_at = row["indexed_at"] return { "partition": row["partition_name"], "file_id": row["file_id"], "relationship_id": row["relationship_id"], "parent_id": row["parent_id"], **metadata, - # Indexation timestamp lives on the row, not in file_metadata; - # placed after the spread so the column value always wins. - "created_at": created_at.isoformat() if created_at else None, + # Authoritative system insert time, materialized on the row. Placed + # after the spread so the column wins over any ``indexed_at`` the + # copy/restore path copies into file_metadata from chunk metadata + # (``_file_metadata_from_chunk``). Distinct from the client-supplied + # ``created_at`` temporal field, which stays in file_metadata. + "indexed_at": indexed_at.isoformat() if indexed_at else None, } @staticmethod diff --git a/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_created_at.py b/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py similarity index 76% rename from openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_created_at.py rename to openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py index 1105ef1ab..8a2eeeec9 100644 --- a/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_created_at.py +++ b/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py @@ -1,4 +1,4 @@ -"""add files.created_at +"""add files.indexed_at Revision ID: b7c1d2e3f4a5 Revises: 06dd2101ea3a @@ -27,12 +27,15 @@ def upgrade() -> None: Existing rows have no recorded index time, so the ``server_default`` backfills them to the migration run time; newly indexed files get their true insert time. + + Named ``indexed_at`` (not ``created_at``) because ``created_at`` is the + reserved client-supplied temporal field stored in ``file_metadata``. """ - if not column_exists("files", "created_at"): + if not column_exists("files", "indexed_at"): op.add_column( "files", sa.Column( - "created_at", + "indexed_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()"), @@ -42,5 +45,5 @@ def upgrade() -> None: def downgrade() -> None: """Downgrade schema.""" - if column_exists("files", "created_at"): - op.drop_column("files", "created_at") + if column_exists("files", "indexed_at"): + op.drop_column("files", "indexed_at") diff --git a/openrag/services/persistence/schema.py b/openrag/services/persistence/schema.py index 948d215d4..cb64daae0 100644 --- a/openrag/services/persistence/schema.py +++ b/openrag/services/persistence/schema.py @@ -140,7 +140,7 @@ Column("relationship_id", String, nullable=True, index=True), Column("parent_id", String, nullable=True, index=True), Column( - "created_at", + "indexed_at", DateTime(timezone=True), server_default=text("now()"), nullable=False, From bb59b74749e9d9242d8a3cc5077527201fccc3f1 Mon Sep 17 00:00:00 2001 From: Ahmath-Gadji Date: Mon, 22 Jun 2026 12:04:10 +0000 Subject: [PATCH 3/4] feat(indexer): share one indexed_at across Milvus chunks and the files row The Milvus upsert and the Postgres catalog write each generated their own now(), so a file's chunk indexed_at and its files.indexed_at could drift. Mint a single timestamp in process_file and thread it to both sinks: the store stage forwards it to VectorStore.upsert (stamped on every chunk) and _write_catalog_record passes it to the files INSERT/UPDATE. Both arguments default to None, falling back to now()/the server default, so existing callers (direct upsert, copy/restore, create_document) are unaffected. Re-index (replace) refreshes indexed_at to match the re-upserted chunks. --- openrag/core/vector_stores/vector_store.py | 19 ++++- openrag/services/persistence/document_repo.py | 40 +++++++++-- openrag/services/storage/milvus_store.py | 14 +++- openrag/services/workers/indexer_actor.py | 11 ++- openrag/services/workers/stages/store.py | 10 ++- tests/unit/conftest.py | 2 +- .../workers/stages/test_pipeline_stages.py | 2 +- .../services/workers/test_batch_ingest.py | 2 +- .../services/workers/test_indexer_worker.py | 70 +++++++++++++------ .../services/workers/test_pipeline_builder.py | 2 +- 10 files changed, 134 insertions(+), 38 deletions(-) diff --git a/openrag/core/vector_stores/vector_store.py b/openrag/core/vector_stores/vector_store.py index 78761790c..08ab34d68 100644 --- a/openrag/core/vector_stores/vector_store.py +++ b/openrag/core/vector_stores/vector_store.py @@ -3,17 +3,30 @@ from __future__ import annotations from abc import ABC, abstractmethod -from typing import Any +from typing import TYPE_CHECKING, Any from openrag.core.models.chunk import Chunk +if TYPE_CHECKING: + from datetime import datetime + class VectorStore(ABC): """Base class for vector database backends.""" @abstractmethod - async def upsert(self, chunks: list[Chunk], collection: str = "default") -> int: - """Insert or update chunks. Returns count of upserted items.""" + async def upsert( + self, + chunks: list[Chunk], + collection: str = "default", + *, + indexed_at: datetime | None = None, + ) -> int: + """Insert or update chunks. Returns count of upserted items. + + ``indexed_at`` optionally pins the indexation timestamp stamped on the + chunks so it can match the catalog row; ``None`` means "use now". + """ ... @abstractmethod diff --git a/openrag/services/persistence/document_repo.py b/openrag/services/persistence/document_repo.py index 90f1f1926..b939e39fb 100644 --- a/openrag/services/persistence/document_repo.py +++ b/openrag/services/persistence/document_repo.py @@ -29,6 +29,8 @@ class (not on the ABC) so the Phase 7C shim can delegate to them unchanged. from core.ports.document_repo import DocumentRepository if TYPE_CHECKING: + from datetime import datetime + import asyncpg # Note on JSON: ``ConnectionManager.initialize`` registers a json/jsonb codec @@ -271,11 +273,15 @@ async def add_file_to_partition( # noqa: PLR0913 — legacy signature pinned relationship_id: str | None = None, parent_id: str | None = None, indexation_config: dict | None = None, + indexed_at: datetime | None = None, ) -> bool: """TODO(phase-9): remove. Mirror of legacy ``add_file_to_partition``. Creates the partition row on first use (legacy behaviour). Returns ``False`` if a row with the same (file_id, partition) already exists. + + ``indexed_at`` pins the indexation timestamp so it matches the Milvus + chunks; when ``None`` the ``files.indexed_at`` server default applies. """ async with self.pool.acquire() as conn: async with conn.transaction(): @@ -311,12 +317,16 @@ async def add_file_to_partition( # noqa: PLR0913 — legacy signature pinned user_id, ) - await conn.execute( - """ - INSERT INTO files (file_id, partition_name, file_metadata, - indexation_config, created_by, relationship_id, parent_id) - VALUES ($1, $2, $3::json, $4::jsonb, $5, $6, $7) - """, + columns = [ + "file_id", + "partition_name", + "file_metadata", + "indexation_config", + "created_by", + "relationship_id", + "parent_id", + ] + values: list[Any] = [ file_id, partition, file_metadata or {}, @@ -324,6 +334,17 @@ async def add_file_to_partition( # noqa: PLR0913 — legacy signature pinned user_id, relationship_id, parent_id, + ] + # Omit indexed_at to let the server default fire (legacy path). + if indexed_at is not None: + columns.append("indexed_at") + values.append(indexed_at) + # file_metadata is JSON, indexation_config is JSONB; the rest bind directly. + casts = {"file_metadata": "::json", "indexation_config": "::jsonb"} + placeholders = ", ".join(f"${i}{casts.get(col, '')}" for i, col in enumerate(columns, start=1)) + await conn.execute( + f"INSERT INTO files ({', '.join(columns)}) VALUES ({placeholders})", + *values, ) if user_id is not None: await conn.execute( @@ -393,12 +414,16 @@ async def update_file_in_partition( relationship_id: object = _UNSET, parent_id: object = _UNSET, indexation_config: object = _UNSET, + indexed_at: datetime | None = None, ) -> bool: """TODO(phase-9): remove. PUT-style in-place update. Preserves the underlying ``files.id`` so workspace FK rows stay valid. Pass ``relationship_id=None`` / ``parent_id=None`` explicitly to clear; omit the kwarg to leave the column alone. + + ``indexed_at`` refreshes the indexation timestamp on re-index so it + matches the freshly re-upserted Milvus chunks; ``None`` leaves it. """ sets: list[str] = [] params: list[Any] = [] @@ -414,6 +439,9 @@ async def update_file_in_partition( if indexation_config is not self._UNSET: params.append(indexation_config) sets.append(f"indexation_config = ${len(params)}::jsonb") + if indexed_at is not None: + params.append(indexed_at) + sets.append(f"indexed_at = ${len(params)}") if not sets: # Match legacy: report whether the row exists at all. return await self.file_exists_in_partition(file_id, partition) diff --git a/openrag/services/storage/milvus_store.py b/openrag/services/storage/milvus_store.py index bd4dca4c0..2ff4c0cf4 100644 --- a/openrag/services/storage/milvus_store.py +++ b/openrag/services/storage/milvus_store.py @@ -624,13 +624,23 @@ def _chunk_to_entity( # VectorStore ABC — writes # ------------------------------------------------------------------ - async def upsert(self, chunks: list[Chunk], collection: str = "default") -> int: + async def upsert( + self, + chunks: list[Chunk], + collection: str = "default", + *, + indexed_at: datetime | None = None, + ) -> int: """Insert pre-embedded chunks into the backing Milvus collection. ``chunk.partition`` is authoritative — the ``collection`` argument is accepted for ABC compatibility but does not override per-chunk partition values. Every chunk MUST carry a populated ``embedding``; embedding is an upstream pipeline concern, not a store concern. + + ``indexed_at`` lets the caller pin a single indexation timestamp so the + Milvus chunks and the Postgres ``files`` row agree; when omitted it + defaults to the current time (legacy behaviour). """ self._resolve_collection(collection) if not chunks: @@ -643,7 +653,7 @@ async def upsert(self, chunks: list[Chunk], collection: str = "default") -> int: collection_name=self._collection_name, ) - indexed_at = datetime.now(UTC).isoformat() + indexed_at = (indexed_at or datetime.now(UTC)).isoformat() order_metadata = self._gen_chunk_order_metadata(len(chunks)) entities = [ self._chunk_to_entity(c, indexed_at=indexed_at, order=o) diff --git a/openrag/services/workers/indexer_actor.py b/openrag/services/workers/indexer_actor.py index d582a788a..736ce9669 100644 --- a/openrag/services/workers/indexer_actor.py +++ b/openrag/services/workers/indexer_actor.py @@ -1,6 +1,7 @@ from __future__ import annotations import traceback +from datetime import datetime from pathlib import Path from typing import Any @@ -61,6 +62,8 @@ async def process_file( await self._tsm.set_state.remote(task_id, "SERIALIZING") try: document = _load_document(path, metadata, partition, indexation_config=indexation_config) + # One indexation timestamp for this file, shared by the Milvus chunks + # (via the store stage) and the Postgres catalog row, so they agree. row: dict[str, Any] = { "document": document, "partition": partition, @@ -72,7 +75,9 @@ async def process_file( "indexation_config": indexation_config, "embedder_name": embedder_name, } - await self._pipeline.run(row) + row = await self._pipeline.run(row) + indexed_at = row.get("indexed_at") + if self._document_repo is not None: await _write_catalog_record( doc_repo=self._document_repo, @@ -81,6 +86,7 @@ async def process_file( user=user, replace=replace, indexation_config=indexation_config, + indexed_at=indexed_at, ) if self._topic_tag_repo is not None: await _replace_topic_tags_if_needed( @@ -106,6 +112,7 @@ async def _write_catalog_record( user: dict[str, Any] | None, replace: bool, indexation_config: dict[str, Any] | None, + indexed_at: datetime | None = None, ) -> None: file_id = metadata.get("file_id", "") file_metadata = {key: value for key, value in metadata.items() if key != "page"} @@ -117,6 +124,7 @@ async def _write_catalog_record( file_metadata=file_metadata, relationship_id=metadata.get("relationship_id"), parent_id=metadata.get("parent_id"), + indexed_at=indexed_at, **config_kwargs, ) return @@ -128,6 +136,7 @@ async def _write_catalog_record( user_id=user.get("id") if user else None, relationship_id=metadata.get("relationship_id"), parent_id=metadata.get("parent_id"), + indexed_at=indexed_at, **config_kwargs, ) diff --git a/openrag/services/workers/stages/store.py b/openrag/services/workers/stages/store.py index 3f6bd50fc..4e81ca7f2 100644 --- a/openrag/services/workers/stages/store.py +++ b/openrag/services/workers/stages/store.py @@ -1,6 +1,7 @@ from __future__ import annotations from collections.abc import MutableMapping +from datetime import UTC, datetime from typing import Any from core.models.chunk import Chunk @@ -32,8 +33,15 @@ async def store_stage( await vector_store.ensure_collection("default", len(embedding)) effective_timeout = stage_timeout(timeout, len(chunks), per_item_timeout=per_chunk_timeout) + # One indexation timestamp shared by the Milvus chunks (via the upsert + # arg below) and the Postgres catalog row (read back from the row in the + # orchestrator). Keep it a ``datetime``: the catalog write binds it to a + # ``timestamptz`` column and asyncpg rejects a pre-stringified value. + indexed_at = datetime.now(UTC) + row["indexed_at"] = indexed_at + row["stored_count"] = await run_with_optional_timeout( - lambda: vector_store.upsert(chunks), + lambda: vector_store.upsert(chunks, indexed_at=indexed_at), effective_timeout, ) row["stage"] = "stored" diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index cc27f097c..5d8b26714 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -51,7 +51,7 @@ def __init__(self) -> None: self.collections: dict[str, dict[str, Any]] = {} self.search_results: list[dict[str, Any]] = [] - async def upsert(self, chunks: list[Any], collection: str = "default") -> int: + async def upsert(self, chunks: list[Any], collection: str = "default", *, indexed_at=None) -> int: store = self.collections.setdefault(collection, {}) for chunk in chunks: store[getattr(chunk, "id", id(chunk))] = chunk diff --git a/tests/unit/services/workers/stages/test_pipeline_stages.py b/tests/unit/services/workers/stages/test_pipeline_stages.py index c73f6f7ec..b809280dc 100644 --- a/tests/unit/services/workers/stages/test_pipeline_stages.py +++ b/tests/unit/services/workers/stages/test_pipeline_stages.py @@ -99,7 +99,7 @@ def __init__(self, count: int, error: Exception | None = None) -> None: self.calls: list[tuple[list[Chunk], str]] = [] self.ensure_calls: list[tuple[str, int]] = [] - async def upsert(self, chunks: list[Chunk], collection: str = "default") -> int: + async def upsert(self, chunks: list[Chunk], collection: str = "default", *, indexed_at=None) -> int: self.calls.append((chunks, collection)) if self.error is not None: raise self.error diff --git a/tests/unit/services/workers/test_batch_ingest.py b/tests/unit/services/workers/test_batch_ingest.py index bcf23cc7d..572dbf062 100644 --- a/tests/unit/services/workers/test_batch_ingest.py +++ b/tests/unit/services/workers/test_batch_ingest.py @@ -57,7 +57,7 @@ def __init__(self) -> None: self.calls: list[tuple] = [] self.ensure_calls: list[tuple[str, int]] = [] - async def upsert(self, chunks: list[Chunk], collection: str = "default") -> int: + async def upsert(self, chunks: list[Chunk], collection: str = "default", *, indexed_at=None) -> int: self.calls.append((chunks, collection)) return len(chunks) diff --git a/tests/unit/services/workers/test_indexer_worker.py b/tests/unit/services/workers/test_indexer_worker.py index 8a871be6d..5811d7a98 100644 --- a/tests/unit/services/workers/test_indexer_worker.py +++ b/tests/unit/services/workers/test_indexer_worker.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime from pathlib import Path from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -51,8 +52,8 @@ def __init__(self) -> None: self.calls: list[tuple] = [] self.ensure_calls: list[tuple[str, int]] = [] - async def upsert(self, chunks: list[Chunk], collection: str = "default") -> int: - self.calls.append((chunks, collection)) + async def upsert(self, chunks: list[Chunk], collection: str = "default", *, indexed_at=None) -> int: + self.calls.append((chunks, collection, indexed_at)) return len(chunks) async def ensure_collection(self, name: str, dimension: int, **kwargs: Any) -> None: @@ -275,19 +276,45 @@ async def test_process_file_creates_catalog_record_after_successful_pipeline(tmp user={"id": 42}, ) - assert repo.add_calls == [ - { - "file_id": "f1", - "partition": "p", - "file_metadata": {"file_id": "f1", "relationship_id": "rel", "parent_id": "parent"}, - "user_id": 42, - "relationship_id": "rel", - "parent_id": "parent", - } - ] + assert len(repo.add_calls) == 1 + add_call = repo.add_calls[0] + assert isinstance(add_call.pop("indexed_at"), datetime) + assert add_call == { + "file_id": "f1", + "partition": "p", + "file_metadata": {"file_id": "f1", "relationship_id": "rel", "parent_id": "parent"}, + "user_id": 42, + "relationship_id": "rel", + "parent_id": "parent", + } assert repo.update_calls == [] +@pytest.mark.asyncio +async def test_process_file_shares_one_indexed_at_between_store_and_catalog(tmp_path: Path) -> None: + path = tmp_path / "doc.txt" + path.write_bytes(b"content") + processed = ProcessedDocument(document_id="d1", text_blocks=[TextBlock(text="content")]) + chunks = [Chunk(id="c1", text="content", partition="p")] + store = FakeVectorStore() + repo = FakeDocumentRepo() + pipeline = build_indexing_pipeline( + parser=FakeParser(processed), + chunker=FakeChunker(chunks), + embedder=FakeEmbedder(), + vector_store=store, + ) + worker = IndexerWorker(pipeline=pipeline, task_state_manager=_fake_tsm(), document_repo=repo) + + await worker.process_file(task_id="t1", path=str(path), metadata={"file_id": "f1"}, partition="p", user={"id": 1}) + + store_indexed_at = store.calls[0][2] + catalog_indexed_at = repo.add_calls[0]["indexed_at"] + assert isinstance(store_indexed_at, datetime) + # The store and the catalog must receive the very same timestamp object/value. + assert store_indexed_at == catalog_indexed_at + + @pytest.mark.asyncio async def test_process_file_stores_indexation_config_snapshot_on_new_file(tmp_path: Path) -> None: path = tmp_path / "doc.txt" @@ -335,15 +362,16 @@ async def test_process_file_updates_catalog_record_on_replace(tmp_path: Path) -> replace=True, ) - assert repo.update_calls == [ - { - "file_id": "f1", - "partition": "p", - "file_metadata": {"file_id": "f1"}, - "relationship_id": None, - "parent_id": None, - } - ] + assert len(repo.update_calls) == 1 + update_call = repo.update_calls[0] + assert isinstance(update_call.pop("indexed_at"), datetime) + assert update_call == { + "file_id": "f1", + "partition": "p", + "file_metadata": {"file_id": "f1"}, + "relationship_id": None, + "parent_id": None, + } assert repo.add_calls == [] diff --git a/tests/unit/services/workers/test_pipeline_builder.py b/tests/unit/services/workers/test_pipeline_builder.py index 3cedd161b..40a75f676 100644 --- a/tests/unit/services/workers/test_pipeline_builder.py +++ b/tests/unit/services/workers/test_pipeline_builder.py @@ -46,7 +46,7 @@ def __init__(self) -> None: self.calls: list[tuple[list[Chunk], str]] = [] self.ensure_calls: list[tuple[str, int]] = [] - async def upsert(self, chunks: list[Chunk], collection: str = "default") -> int: + async def upsert(self, chunks: list[Chunk], collection: str = "default", *, indexed_at=None) -> int: self.calls.append((chunks, collection)) return len(chunks) From 8a3477278709b4110dc09c0bb80ce607f4644798 Mon Sep 17 00:00:00 2001 From: Ahmath-Gadji Date: Mon, 22 Jun 2026 12:47:40 +0000 Subject: [PATCH 4/4] fix(migrations): chain indexed_at after topic_tags to avoid multiple heads Rebasing onto refactor/hexagonal brought in the topic_tags migration (b7c8d9e0f1a2), which also descends from 06dd2101ea3a. Two siblings off the same parent give Alembic multiple heads, so `alembic upgrade head` fails at startup and the ServiceContainer never initializes (every request 500s). Re-parent indexed_at (b7c1d2e3f4a5) onto b7c8d9e0f1a2 for a single linear head. --- .../alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py b/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py index 8a2eeeec9..5b34c2588 100644 --- a/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py +++ b/openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py @@ -1,7 +1,7 @@ """add files.indexed_at Revision ID: b7c1d2e3f4a5 -Revises: 06dd2101ea3a +Revises: b7c8d9e0f1a2 Create Date: 2026-06-19 00:00:00.000000 """ @@ -14,7 +14,7 @@ # revision identifiers, used by Alembic. revision: str = "b7c1d2e3f4a5" -down_revision: str | Sequence[str] | None = "06dd2101ea3a" +down_revision: str | Sequence[str] | None = "b7c8d9e0f1a2" branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None