feat(files): persist + expose indexation timestamp (files.created_at)

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

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,19 +317,34 @@ 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 {},
                     indexation_config,
                     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)
@@ -550,12 +578,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 {}
+        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,
+            # 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

openrag/services/persistence/migrations/alembic/versions/b7c1d2e3f4a5_add_files_indexed_at.py

@@ -0,0 +1,49 @@
+"""add files.indexed_at
+
+Revision ID: b7c1d2e3f4a5
+Revises: b7c8d9e0f1a2
+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 = "b7c8d9e0f1a2"
+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.
+
+    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", "indexed_at"):
+        op.add_column(
+            "files",
+            sa.Column(
+                "indexed_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.text("now()"),
+            ),
+        )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    if column_exists("files", "indexed_at"):
+        op.drop_column("files", "indexed_at")

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(
+        "indexed_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"),

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)

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,
     )
 

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"

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

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

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)