feat(rag): add active generation schema and store (#438 PR1)

src/xagent/core/tools/core/RAG_tools/LanceDB/schema_manager.py

@@ -16,6 +16,7 @@
     "ensure_parses_table",
     "ensure_chunks_table",
     "ensure_embeddings_table",
+    "ensure_active_generations_table",
     "ensure_main_pointers_table",
     "ensure_prompt_templates_table",
     "ensure_ingestion_runs_table",
@@ -34,6 +35,27 @@ def _safe_close_table(table: Any) -> None:
             pass
 
 
+def _add_nullable_string_columns(
+    conn: DBConnection, table_name: str, columns: dict[str, str]
+) -> None:
+    """Add missing nullable string columns to an existing LanceDB table."""
+    if not _table_exists(conn, table_name):
+        return
+    table = conn.open_table(table_name)
+    try:
+        existing = set(table.schema.names)
+        to_add = {name: expr for name, expr in columns.items() if name not in existing}
+        if to_add:
+            logger.info(
+                "Migrating '%s' table: adding columns %s",
+                table_name,
+                list(to_add.keys()),
+            )
+            table.add_columns(to_add)
+    finally:
+        _safe_close_table(table)
+
+
 def _table_exists(conn: DBConnection, name: str) -> bool:
     try:
         return name in list_table_names(conn)
@@ -352,6 +374,9 @@ def ensure_chunks_table(conn: DBConnection) -> None:
     _validate_schema_fields(conn, "chunks", required_fields)
     if _table_exists(conn, "chunks"):
         _migrate_table_user_id_to_int64(conn, "chunks")
+        _add_nullable_string_columns(
+            conn, "chunks", {"generation_id": "cast(null as string)"}
+        )
 
         val_table = conn.open_table("chunks")
         try:
@@ -373,6 +398,7 @@ def ensure_chunks_table(conn: DBConnection) -> None:
             pa.field("json_path", pa.string()),
             pa.field("chunk_hash", pa.string()),
             pa.field("config_hash", pa.string()),
+            pa.field("generation_id", pa.string()),
             pa.field("created_at", pa.timestamp("us")),
             pa.field("metadata", pa.string()),
             pa.field("user_id", pa.int64()),
@@ -417,6 +443,14 @@ def ensure_embeddings_table(
     _validate_schema_fields(conn, table_name, required_fields)
     if _table_exists(conn, table_name):
         _migrate_table_user_id_to_int64(conn, table_name)
+        _add_nullable_string_columns(
+            conn,
+            table_name,
+            {
+                "generation_id": "cast(null as string)",
+                "config_hash": "cast(null as string)",
+            },
+        )
 
         val_table = conn.open_table(table_name)
         try:
@@ -441,6 +475,8 @@ def ensure_embeddings_table(
             pa.field("vector_dimension", pa.int32()),
             pa.field("text", pa.large_string()),
             pa.field("chunk_hash", pa.string()),
+            pa.field("config_hash", pa.string()),
+            pa.field("generation_id", pa.string()),
             pa.field("created_at", pa.timestamp("us")),
             pa.field("metadata", pa.string()),
             pa.field("user_id", pa.int64()),
@@ -453,6 +489,30 @@ def ensure_embeddings_table(
     )
 
 
+def ensure_active_generations_table(conn: DBConnection) -> None:
+    """Ensure the active_generations table exists with proper schema.
+
+    Tracks the published searchable generation per document scope:
+    (collection, doc_id, parse_hash, user_id, model_tag).
+    """
+    schema = pa.schema(
+        [
+            pa.field("collection", pa.string()),
+            pa.field("doc_id", pa.string()),
+            pa.field("parse_hash", pa.string()),
+            pa.field("user_id", pa.int64()),
+            pa.field("model_tag", pa.string()),
+            pa.field("generation_id", pa.string()),
+            pa.field("config_hash", pa.string()),
+            pa.field("created_at", pa.timestamp("us")),
+            pa.field("updated_at", pa.timestamp("us")),
+            pa.field("published_at", pa.timestamp("us")),
+            pa.field("operator", pa.string()),
+        ]
+    )
+    _create_table(conn, "active_generations", schema=schema)
+
+
 def ensure_main_pointers_table(conn: DBConnection) -> None:
     """Ensure the main_pointers table exists with proper schema.
 

src/xagent/core/tools/core/RAG_tools/kb/storage_shim.py

@@ -6,6 +6,7 @@
 
 if TYPE_CHECKING:
     from ..storage.contracts import (
+        ActiveGenerationStore,
         IngestionStatusStore,
         KBWriteCoordinator,
         MainPointerStore,
@@ -59,6 +60,10 @@ def get_main_pointer_store(self) -> MainPointerStore:
         """Return the legacy main pointer store singleton."""
         return self._storage_factory.get_main_pointer_store()
 
+    def get_active_generation_store(self) -> ActiveGenerationStore:
+        """Return the legacy active generation store singleton."""
+        return self._storage_factory.get_active_generation_store()
+
     def reset_kb_write_coordinator(self) -> None:
         """Clear legacy storage singletons and coordinator-backed shim state."""
         self._storage_factory.reset_all()

src/xagent/core/tools/core/RAG_tools/storage/__init__.py

@@ -4,6 +4,7 @@
 """
 
 from .contracts import (
+    ActiveGenerationStore,
     IngestionStatusStore,
     KBWriteCoordinator,
     MainPointerStore,
@@ -13,6 +14,7 @@
 )
 from .factory import (
     StorageFactory,
+    get_active_generation_store,
     get_ingestion_status_store,
     get_kb_write_coordinator,
     get_main_pointer_store,
@@ -38,6 +40,7 @@
     "IngestionStatusStore",
     "PromptTemplateStore",
     "MainPointerStore",
+    "ActiveGenerationStore",
     # Factory
     "StorageFactory",
     "get_kb_write_coordinator",
@@ -51,6 +54,7 @@
     "get_ingestion_status_store",
     "get_prompt_template_store",
     "get_main_pointer_store",
+    "get_active_generation_store",
     "reset_kb_write_coordinator",
     "reset_rag_storage_for_tests",
 ]

src/xagent/core/tools/core/RAG_tools/storage/contracts.py

@@ -56,6 +56,7 @@
         "json_path",
         "chunk_hash",
         "config_hash",
+        "generation_id",
         "metadata",
         # embeddings table
         "model",
@@ -71,6 +72,8 @@
         "semantic_id",
         "technical_id",
         "operator",
+        # active_generations table
+        "published_at",
         # prompt_templates table
         "id",
         "name",
@@ -1794,3 +1797,93 @@ async def delete_main_pointer_async(
         user_id: Optional[int] = None,
     ) -> bool:
         """Async version of delete_main_pointer."""
+
+
+class _UnsetUserIdFilter:
+    """Sentinel type for omitted active-generation user_id filters."""
+
+
+USER_ID_FILTER_UNSET = _UnsetUserIdFilter()
+UserIdFilter = Union[Optional[int], _UnsetUserIdFilter]
+
+
+class ActiveGenerationStore(ABC):
+    """Published active-generation pointer contract for ingestion and retrieval.
+
+    Each pointer identifies the single searchable generation for a document scope:
+    (collection, doc_id, parse_hash, user_id, model_tag).
+    """
+
+    @abstractmethod
+    def publish_active_generation(
+        self,
+        collection: str,
+        doc_id: str,
+        parse_hash: str,
+        user_id: Optional[int],
+        model_tag: Optional[str],
+        generation_id: str,
+        config_hash: str,
+        operator: Optional[str] = None,
+    ) -> None:
+        """Publish or replace the active generation pointer for a scope."""
+
+    @abstractmethod
+    def get_active_generation(
+        self,
+        collection: str,
+        doc_id: str,
+        parse_hash: str,
+        user_id: Optional[int],
+        model_tag: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Return the active generation pointer for a scope, if any."""
+
+    @abstractmethod
+    def list_active_generations(
+        self,
+        collection: str,
+        *,
+        doc_id: Optional[str] = None,
+        user_id: UserIdFilter = USER_ID_FILTER_UNSET,
+        model_tag: Optional[str] = None,
+        limit: int = 1000,
+    ) -> List[Dict[str, Any]]:
+        """List active generation pointers for a collection."""
+
+    @abstractmethod
+    async def publish_active_generation_async(
+        self,
+        collection: str,
+        doc_id: str,
+        parse_hash: str,
+        user_id: Optional[int],
+        model_tag: Optional[str],
+        generation_id: str,
+        config_hash: str,
+        operator: Optional[str] = None,
+    ) -> None:
+        """Async version of publish_active_generation."""
+
+    @abstractmethod
+    async def get_active_generation_async(
+        self,
+        collection: str,
+        doc_id: str,
+        parse_hash: str,
+        user_id: Optional[int],
+        model_tag: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Async version of get_active_generation."""
+
+    @abstractmethod
+    async def list_active_generations_async(
+        self,
+        collection: str,
+        *,
+        doc_id: Optional[str] = None,
+        user_id: UserIdFilter = USER_ID_FILTER_UNSET,
+        model_tag: Optional[str] = None,
+        limit: int = 1000,
+    ) -> List[Dict[str, Any]]:
+        """Async version of list_active_generations."""

src/xagent/core/tools/core/RAG_tools/storage/factory.py

@@ -17,6 +17,7 @@
 from typing import TYPE_CHECKING, Any, Optional, cast
 
 from .contracts import (
+    ActiveGenerationStore,
     IngestionStatusStore,
     KBWriteCoordinator,
     MainPointerStore,
@@ -25,6 +26,7 @@
     VectorIndexStore,
 )
 from .lancedb_stores import (
+    LanceDBActiveGenerationStore,
     LanceDBIngestionStatusStore,
     LanceDBMainPointerStore,
     LanceDBMetadataStore,
@@ -96,6 +98,7 @@ def __init__(self) -> None:
         self._ingestion_status_store: Optional[IngestionStatusStore] = None
         self._prompt_template_store: Optional[PromptTemplateStore] = None
         self._main_pointer_store: Optional[MainPointerStore] = None
+        self._active_generation_store: Optional[ActiveGenerationStore] = None
         self._coordinator: Optional[KBWriteCoordinator] = None
 
     @classmethod
@@ -126,6 +129,7 @@ def reset_all(self) -> None:
             self._ingestion_status_store = None
             self._prompt_template_store = None
             self._main_pointer_store = None
+            self._active_generation_store = None
             self._coordinator = None
 
         # Keep semantic KB coordinator state aligned with factory-backed stores.
@@ -217,6 +221,16 @@ def get_prompt_template_store(self) -> PromptTemplateStore:
                     self._prompt_template_store = LanceDBPromptTemplateStore()
         return self._prompt_template_store
 
+    # --- ActiveGenerationStore ---
+
+    def get_active_generation_store(self) -> ActiveGenerationStore:
+        """Get or create active generation pointer store."""
+        if self._active_generation_store is None:
+            with self._lock:
+                if self._active_generation_store is None:
+                    self._active_generation_store = LanceDBActiveGenerationStore()
+        return self._active_generation_store
+
     # --- MainPointerStore ---
 
     def get_main_pointer_store(self) -> MainPointerStore:
@@ -359,6 +373,15 @@ def get_main_pointer_store() -> MainPointerStore:
     return _get_storage_shim().get_main_pointer_store()
 
 
+def get_active_generation_store() -> ActiveGenerationStore:
+    """Get active generation pointer store.
+
+    Returns:
+        LanceDBActiveGenerationStore instance.
+    """
+    return _get_storage_shim().get_active_generation_store()
+
+
 # ============================================================================
 # Default Coordinator Implementation
 # ============================================================================