From 60ac85777e8aba3cf6a36df93aabc517ae713c44 Mon Sep 17 00:00:00 2001 From: Deep Kumar Singh Kushwah Date: Fri, 19 Jun 2026 02:31:54 +0530 Subject: [PATCH 1/6] feat(gateway,backends): config-driven OpenAI LLM + content hydration Wire real backends from rag.yaml so the gateway runs end-to-end RAG instead of noop-only: - rag_backends.llm.OpenAILLM (chat completions) behind a new [openai] extra - _build_{embedder,vector_store,llm}_from_config in wiring; vector-store lifespan init - IngestPipeline gains an optional content_store; PgVectorStore.put_content / hydrate_content persist chunk text so retrieved refs can be hydrated for rerank / pack / generate (the vector store holds vectors only) - query path now hydrates whenever pack or generate need text, not just rerank - gateway [backends] extra pulls rag-backends[openai] + rag-embedders[openai] Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/gateway/pyproject.toml | 2 +- apps/gateway/src/rag_gateway/app.py | 52 +++++-- apps/gateway/src/rag_gateway/query.py | 6 +- apps/gateway/src/rag_gateway/wiring.py | 135 ++++++++++++++++- packages/backends/pyproject.toml | 3 + .../backends/src/rag_backends/llm/__init__.py | 12 ++ .../backends/src/rag_backends/llm/openai.py | 136 ++++++++++++++++++ .../src/rag_backends/vector/pgvector.py | 104 ++++++++++++++ packages/ingest/src/rag_ingest/pipeline.py | 7 + uv.lock | 12 +- 10 files changed, 450 insertions(+), 19 deletions(-) create mode 100644 packages/backends/src/rag_backends/llm/__init__.py create mode 100644 packages/backends/src/rag_backends/llm/openai.py diff --git a/apps/gateway/pyproject.toml b/apps/gateway/pyproject.toml index cc98735..a17a457 100644 --- a/apps/gateway/pyproject.toml +++ b/apps/gateway/pyproject.toml @@ -53,7 +53,7 @@ dependencies = [ # default gateway install stays light; required only when rag.yaml selects # ``corpus_store.provider: postgres``. ``build_corpus_store_from_config`` # imports it lazily and raises a clear error if it is missing. -backends = ["rag-backends"] +backends = ["rag-backends[openai]", "rag-embedders[openai]"] [tool.uv.sources] rag-core = { workspace = true } diff --git a/apps/gateway/src/rag_gateway/app.py b/apps/gateway/src/rag_gateway/app.py index 6d7a18f..8f7858b 100644 --- a/apps/gateway/src/rag_gateway/app.py +++ b/apps/gateway/src/rag_gateway/app.py @@ -181,15 +181,28 @@ def _enforce_residency(state: Any, tenant_id: str) -> None: # --------------------------------------------------------------------------- # Default wiring — dev / smoke; production replaces this from rag.yaml in 3.5 # --------------------------------------------------------------------------- -def build_default_ingest_pipeline() -> IngestPipeline: - """Construct a fully-wired ``IngestPipeline`` for dev / demo use.""" +def build_default_ingest_pipeline( + *, + embedder: Any | None = None, + index_backend: Any | None = None, + content_store: Any | None = None, +) -> IngestPipeline: + """Construct a fully-wired ``IngestPipeline`` for dev / demo use. + + ``embedder`` / ``index_backend`` override the noop defaults so the + config-driven wiring can supply a real embedder + vector store; ingest + then embeds chunks and writes their vectors to that store, making them + retrievable by the query path (which must share the same store). + ``content_store`` (when set) persists chunk text for query-time hydration. + """ return IngestPipeline( chunker=HeadingAwareChunker(), enricher=DefaultEnricher(), - embedder=NoopEmbedder(dimension=8), - index_backend=NoopVectorStore(), + embedder=embedder or NoopEmbedder(dimension=8), + index_backend=index_backend or NoopVectorStore(), parser_registry=default_registry().select, pii_processor=PiiProcessor(detector=RegexPIIDetector()), + content_store=content_store, ) @@ -213,6 +226,8 @@ def build_default_retrieval_router( *, breaker_registry: BreakerRegistry | None = None, base_weights: HybridWeights | None = None, + vector_backend: Any | None = None, + query_embedder: Any | None = None, ) -> RetrievalRouter: """Noop-backed HybridRetriever + RetrievalRouter. @@ -231,7 +246,7 @@ def build_default_retrieval_router( — they never fail, so the breakers stay closed — but exercises the full machinery + status surface; real backends inherit the protection unchanged. """ - vector: VectorRetrievalBackend = NoopVectorStore() + vector: VectorRetrievalBackend = vector_backend or NoopVectorStore() keyword: KeywordRetrievalBackend = NoopKeywordStore() graph: GraphRetrievalBackend = NoopGraphStore() if breaker_registry is not None: @@ -246,7 +261,7 @@ def build_default_retrieval_router( config = RouterConfig(base_weights=base_weights) if base_weights is not None else None return RetrievalRouter( hybrid=hybrid, - embedder=NoopEmbedder(dimension=8), + embedder=query_embedder or NoopEmbedder(dimension=8), config=config, ) @@ -295,13 +310,14 @@ def build_default_fallback_chain( return FallbackChain(router=retrieval_router, config=config) -def build_default_reranker() -> RerankPipeline: +def build_default_reranker(*, hydrator: Any | None = None) -> RerankPipeline: """Two-stage cascade backed by ``NoopReranker``. - ``hydrator`` is a stub that returns empty chunks — production - callers wire their backend's ``hydrate`` method. For the demo, - the rerank stage runs but produces an empty hydrated list, - which is fine: ``/v1/query`` falls back to the chunk-ref view. + ``hydrator`` (a ``*RetrievalBackend.hydrate``-shaped callable turning + ChunkRefs into Chunks-with-content) is wired by config-driven serving to + the vector store's content hydrator, so rerank / generation see real text. + When omitted, a stub returns empty chunks (retrieval-only) — ``/v1/query`` + falls back to the chunk-ref view. """ async def _noop_hydrator(ctx: RequestContext, refs: list[Any]) -> list[Any]: @@ -309,7 +325,7 @@ async def _noop_hydrator(ctx: RequestContext, refs: list[Any]) -> list[Any]: return RerankPipeline( reranker=NoopReranker(), - hydrator=_noop_hydrator, + hydrator=hydrator or _noop_hydrator, ) @@ -956,6 +972,18 @@ async def ingest_document( # subscriber never blocks the upload response). Only on a real ingest. if result.status is IngestStatus.ingested: request.app.state.webhook_dispatcher.schedule(ctx, ingest_completed_event(ctx, result)) + # Bump the corpus's managed document_count. Best-effort: the document + # is already ingested, so a counter hiccup must never fail the upload + # (degrade-open) — the count is a display aid, not a correctness gate. + try: + await request.app.state.corpus_store.increment_document_count( + ctx, CorpusId(corpus_id) + ) + except Exception: # noqa: BLE001 — cosmetic count; never fail the ingest + _log.warning( + "ingest.corpus_count_bump_failed", + extra={"tenant_id": str(tenant), "corpus_id": corpus_id}, + ) # A new document changes the corpus, so drop its cached retrievals. # Best-effort: the in-memory tier returns 0 (relies on the corpus # version key), the Redis L1 tier drops tagged entries precisely. diff --git a/apps/gateway/src/rag_gateway/query.py b/apps/gateway/src/rag_gateway/query.py index e75e70b..3c9b2bb 100644 --- a/apps/gateway/src/rag_gateway/query.py +++ b/apps/gateway/src/rag_gateway/query.py @@ -716,7 +716,11 @@ async def _handle_query( # chunk-ref view (no hydrated chunks / citations this request) rather than # 5xx — Step 7.2's chaos gate kills this backend and asserts the survival. chunks: list[Chunk] = [] - if body.rerank and chunk_refs: + # Hydrate (via the rerank pipeline) whenever a downstream stage needs + # chunk text — rerank, pack, or generate. With the noop reranker this is + # pure hydration; a real reranker also reorders. Ref-only retrieval skips + # it so id-only callers pay nothing. + if (body.rerank or body.pack or body.generate) and chunk_refs: with timings.stage("rerank"): try: chunks = await deps.reranker.rerank( diff --git a/apps/gateway/src/rag_gateway/wiring.py b/apps/gateway/src/rag_gateway/wiring.py index 955dd22..b87a62d 100644 --- a/apps/gateway/src/rag_gateway/wiring.py +++ b/apps/gateway/src/rag_gateway/wiring.py @@ -29,6 +29,8 @@ from __future__ import annotations +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from typing import TYPE_CHECKING, Any from rag_breaker import BreakerRegistry, CircuitBreakerConfig @@ -732,6 +734,93 @@ def build_ab_router_from_config( ) +def _build_embedder_from_config(cfg: RagConfig) -> Any | None: + """Construct a real embedder from ``cfg.backends.embedder``. + + Returns ``None`` for ``provider: none`` (caller keeps build_app's noop + default). Only providers wired here are supported; anything else fails + fast at boot so a misconfigured ``rag.yaml`` never silently degrades. + Imported lazily so the provider SDK stays an optional dependency. + """ + from rag_config import EmbedderProvider + + emb = cfg.backends.embedder + if emb.provider is EmbedderProvider.NONE: + return None + conf = dict(emb.config) + dim = conf.get("dimension") + if emb.provider is EmbedderProvider.OPENAI: + from rag_embedders.openai import OpenAIEmbedder + + return OpenAIEmbedder( + model=emb.model or "text-embedding-3-small", + api_key=conf.get("api_key"), + base_url=conf.get("base_url"), + target_dimension=int(dim) if dim is not None else None, + ) + raise ValueError(f"embedder provider {emb.provider.value!r} is not wired in this build") + + +def _build_vector_store_from_config(cfg: RagConfig) -> Any | None: + """Construct a real vector store from ``cfg.backends.vector_store`` (pgvector).""" + from rag_config import VectorStoreProvider + + vec = cfg.backends.vector_store + if vec.provider is VectorStoreProvider.NONE: + return None + conf = dict(vec.config) + if vec.provider is VectorStoreProvider.PGVECTOR: + dsn = conf.get("dsn") + if not dsn: + raise ValueError("vector_store.provider=pgvector requires config.dsn") + from rag_backends.vector.pgvector import PgVectorStore + + return PgVectorStore( + dsn=str(dsn), + table=str(conf.get("table", "rag_vector_store")), + dimension=int(conf.get("dimension", 1536)), + ) + raise ValueError(f"vector_store provider {vec.provider.value!r} is not wired in this build") + + +def _build_llm_from_config(cfg: RagConfig) -> Any | None: + """Construct a real LLM from ``cfg.backends.llm`` (OpenAI chat completions).""" + from rag_config import LLMProvider + + llm_cfg = cfg.backends.llm + if llm_cfg.provider is LLMProvider.NONE: + return None + conf = dict(llm_cfg.config) + if llm_cfg.provider is LLMProvider.OPENAI: + from rag_backends.llm.openai import OpenAILLM + + return OpenAILLM( + model=llm_cfg.model or "gpt-4o-mini", + api_key=conf.get("api_key"), + base_url=conf.get("base_url"), + ) + raise ValueError(f"llm provider {llm_cfg.provider.value!r} is not wired in this build") + + +def _register_vector_store_init(app: FastAPI, vector_store: Any) -> None: + """Run the vector store's async, idempotent ``initialize()`` before serving. + + ``PgVectorStore`` lazily creates its connection pool but NOT its schema, so + ``initialize()`` (CREATE EXTENSION / TABLE / ANN index) must run once. + Starlette 1.x is lifespan-only (no ``on_event``), so we wrap the app's + existing lifespan to run the idempotent init first, then yield to it. + """ + previous = app.router.lifespan_context + + @asynccontextmanager + async def _lifespan(app_: FastAPI) -> AsyncIterator[None]: + await vector_store.initialize() + async with previous(app_): + yield + + app.router.lifespan_context = _lifespan + + def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: """Build the gateway app with a config-driven corpus store + router. @@ -748,6 +837,8 @@ def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: from rag_gateway.app import ( build_app, build_default_fallback_chain, + build_default_ingest_pipeline, + build_default_reranker, build_default_retrieval_router, ) @@ -763,9 +854,44 @@ def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: breaker_registry: BreakerRegistry | None = overrides.pop("breaker_registry", None) if breaker_registry is None and cfg.breakers.enabled: breaker_registry = BreakerRegistry(config=_breaker_runtime_config(cfg)) + + # Real retrieval / generation backends from ``cfg.backends`` (e.g. OpenAI + + # pgvector). ONE embedder + vector store is shared by ingest (embed → + # bulk_index) and the query path (embed → retrieve_ids) so ingested vectors + # are retrievable; each is ``None`` for ``provider: none``, leaving + # build_app's noop defaults untouched. Real retrieval needs BOTH an embedder + # and a vector store (one without the other can't embed-then-search). + embedder = _build_embedder_from_config(cfg) + vector_store = _build_vector_store_from_config(cfg) + llm = _build_llm_from_config(cfg) + real_retrieval = embedder is not None and vector_store is not None + base_router = overrides.pop("retrieval_router", None) or build_default_retrieval_router( - breaker_registry=breaker_registry + breaker_registry=breaker_registry, + vector_backend=vector_store if real_retrieval else None, + query_embedder=embedder if real_retrieval else None, ) + if real_retrieval and "ingest_pipeline" not in overrides: + overrides["ingest_pipeline"] = build_default_ingest_pipeline( + embedder=embedder, index_backend=vector_store, content_store=vector_store + ) + if real_retrieval and vector_store is not None and "reranker" not in overrides: + # Hydrate retrieved refs into Chunks-with-text via the vector store's + # content sidecar so reranking + generation have real evidence to use. + overrides["reranker"] = build_default_reranker(hydrator=vector_store.hydrate_content) + if llm is not None and "llm" not in overrides: + overrides["llm"] = llm + if embedder is not None and "embedder" not in overrides: + overrides["embedder"] = embedder + if real_retrieval and "understanding" not in overrides: + # Embed the RAW query with the real embedder. The default understanding + # wires HyDE with a dim-8 NoopEmbedder whose query vector would mismatch + # the real store and get the vector source silently dropped; a bare + # pipeline leaves hyde_vector=None so the router embeds the query text + # directly via its embedder (router._resolve_vector_input). + from rag_query.pipeline import QueryUnderstandingPipeline + + overrides["understanding"] = QueryUnderstandingPipeline() retrieval_router = ( build_default_fallback_chain(base_router, config=_retrieval_fallback_config(cfg)) @@ -906,7 +1032,7 @@ def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: audit_days=_r.audit_days, ) - return build_app( + app = build_app( auth=auth_backend, sso_enabled=cfg.sso.enabled, scim_enabled=scim_enabled, @@ -932,6 +1058,11 @@ def build_app_from_config(cfg: RagConfig, **overrides: Any) -> FastAPI: key_manager=key_manager, **overrides, ) + # pgvector needs its extension/table/index created before first use; do it on + # startup (idempotent) now that the app — and its event loop — exist. + if real_retrieval: + _register_vector_store_init(app, vector_store) + return app __all__ = [ diff --git a/packages/backends/pyproject.toml b/packages/backends/pyproject.toml index 971b3ca..f20bd7a 100644 --- a/packages/backends/pyproject.toml +++ b/packages/backends/pyproject.toml @@ -45,6 +45,9 @@ elasticsearch = ["elasticsearch[async]>=8.15"] # bindings). Ships pre-built wheels for ubuntu / macos / windows so no Rust # toolchain is required at install time. tantivy = ["tantivy>=0.22"] +# OpenAI chat-completions client (rag_backends.llm.OpenAILLM) — opt-in so the +# default install stays light; the gateway pulls it via its ``backends`` extra. +openai = ["openai>=1.30"] # Step 2.4 — knowledge-graph backends. Neo4j and Memgraph share the # official ``neo4j`` async driver (Memgraph speaks the same Bolt + Cypher # protocol); NetworkX is the pure-Python in-process option. diff --git a/packages/backends/src/rag_backends/llm/__init__.py b/packages/backends/src/rag_backends/llm/__init__.py new file mode 100644 index 0000000..4baad03 --- /dev/null +++ b/packages/backends/src/rag_backends/llm/__init__.py @@ -0,0 +1,12 @@ +"""Real LLM provider clients (the ``llm`` SPI's production implementations). + +Mirrors the ``vector/`` sub-package: concrete, dependency-gated clients behind +the dependency-free ``rag_core.spi.llm.LLM`` ABC. Imported lazily by the +gateway's config-driven wiring so the SDK stays optional. +""" + +from __future__ import annotations + +from rag_backends.llm.openai import OpenAILLM + +__all__ = ["OpenAILLM"] diff --git a/packages/backends/src/rag_backends/llm/openai.py b/packages/backends/src/rag_backends/llm/openai.py new file mode 100644 index 0000000..c015fda --- /dev/null +++ b/packages/backends/src/rag_backends/llm/openai.py @@ -0,0 +1,136 @@ +"""OpenAILLM — async OpenAI Chat Completions client behind the LLM SPI. + +Requires the ``[openai]`` extra (``pip install "rag-backends[openai]"``). Reads +the API key from the standard ``OPENAI_API_KEY`` env var by default; pass +``api_key=`` / ``base_url=`` to override (e.g. Azure / a compatible gateway). + +Mirrors :class:`rag_embedders.openai.OpenAIEmbedder`: the ``openai`` SDK is +imported lazily so constructing the object never requires the dependency, and a +missing package raises a clear :class:`~rag_core.errors.BackendError`. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any + +from rag_core.errors import BackendError +from rag_core.spi.llm import LLM, LLMMessage, LLMResponse +from rag_core.types import RequestContext + + +class OpenAILLM(LLM): + """Async OpenAI chat-completions client. + + ``model`` selects the chat model (e.g. ``gpt-4o-mini``). Token counts are + taken from the provider ``usage`` block when present (0 on the streaming + path, where usage is not emitted per-chunk). + """ + + def __init__( + self, + *, + model: str = "gpt-4o-mini", + api_key: str | None = None, + base_url: str | None = None, + organization: str | None = None, + ) -> None: + self._model = model + self._api_key = api_key + self._base_url = base_url + self._organization = organization + self._client: Any | None = None + + @property + def model(self) -> str: + return self._model + + def _get_client(self) -> Any: + if self._client is not None: + return self._client + try: + from openai import AsyncOpenAI + except ImportError as exc: # pragma: no cover - exercised via env without extra + raise BackendError( + "OpenAILLM requires the [openai] extra: pip install 'rag-backends[openai]'" + ) from exc + kwargs: dict[str, Any] = {} + if self._api_key is not None: + kwargs["api_key"] = self._api_key + if self._base_url is not None: + kwargs["base_url"] = self._base_url + if self._organization is not None: + kwargs["organization"] = self._organization + self._client = AsyncOpenAI(**kwargs) + return self._client + + @staticmethod + def _to_openai(messages: list[LLMMessage]) -> list[dict[str, str]]: + return [{"role": m.role, "content": m.content} for m in messages] + + async def complete( + self, + ctx: RequestContext, + messages: list[LLMMessage], + max_tokens: int = 1024, + temperature: float = 0.0, + stop: list[str] | None = None, + ) -> LLMResponse: + client = self._get_client() + try: + response = await client.chat.completions.create( + model=self._model, + messages=self._to_openai(messages), + max_tokens=max_tokens, + temperature=temperature, + stop=stop, + ) + except Exception as exc: # provider / network error → typed domain error + raise BackendError(f"OpenAI completion failed: {exc}") from exc + choice = response.choices[0] + usage = response.usage + return LLMResponse( + content=choice.message.content or "", + model=response.model, + input_tokens=int(getattr(usage, "prompt_tokens", 0) or 0), + output_tokens=int(getattr(usage, "completion_tokens", 0) or 0), + metadata={"finish_reason": choice.finish_reason}, + ) + + async def stream( + self, + ctx: RequestContext, + messages: list[LLMMessage], + max_tokens: int = 1024, + temperature: float = 0.0, + stop: list[str] | None = None, + ) -> AsyncIterator[str]: + client = self._get_client() + try: + stream = await client.chat.completions.create( + model=self._model, + messages=self._to_openai(messages), + max_tokens=max_tokens, + temperature=temperature, + stop=stop, + stream=True, + ) + except Exception as exc: # pragma: no cover - network path + raise BackendError(f"OpenAI stream failed: {exc}") from exc + + async def _gen() -> AsyncIterator[str]: + async for chunk in stream: + if not chunk.choices: + continue + delta = chunk.choices[0].delta.content + if delta: + yield delta + + return _gen() + + async def health(self) -> bool: + try: + self._get_client() + except Exception: + return False + return True diff --git a/packages/backends/src/rag_backends/vector/pgvector.py b/packages/backends/src/rag_backends/vector/pgvector.py index 8f8fd1e..b0123fd 100644 --- a/packages/backends/src/rag_backends/vector/pgvector.py +++ b/packages/backends/src/rag_backends/vector/pgvector.py @@ -24,9 +24,11 @@ from rag_core.filter import FilterExpr from rag_core.spi.vector_store import VectorStore from rag_core.types import ( + Chunk, ChunkId, ChunkRef, CorpusId, + DocumentId, Embedding, IndexHint, RequestContext, @@ -114,6 +116,40 @@ AND chunk_id = ANY($2) """ +# Sidecar table holding chunk *text* (+ the fields needed to rebuild a Chunk) +# so retrieved chunk_refs can be hydrated for reranking / generation. The +# vector table stays vectors-only per the VectorStore SPI; text lives here. +_DDL_CONTENT_TABLE = """ +CREATE TABLE IF NOT EXISTS {table} ( + tenant_id TEXT NOT NULL, + chunk_id TEXT NOT NULL, + document_id TEXT NOT NULL DEFAULT '', + corpus_id TEXT NOT NULL DEFAULT '', + position INT NOT NULL DEFAULT 0, + content TEXT NOT NULL DEFAULT '', + acl_labels TEXT[] NOT NULL DEFAULT '{{}}', + PRIMARY KEY (tenant_id, chunk_id) +) +""" + +_SQL_CONTENT_UPSERT = """ +INSERT INTO {table} + (tenant_id, chunk_id, document_id, corpus_id, position, content, acl_labels) +VALUES ($1, $2, $3, $4, $5, $6, $7) +ON CONFLICT (tenant_id, chunk_id) DO UPDATE SET + document_id = EXCLUDED.document_id, + corpus_id = EXCLUDED.corpus_id, + position = EXCLUDED.position, + content = EXCLUDED.content, + acl_labels = EXCLUDED.acl_labels +""" + +_SQL_CONTENT_SELECT = """ +SELECT chunk_id, document_id, corpus_id, position, content, acl_labels +FROM {table} +WHERE tenant_id = $1 AND chunk_id = ANY($2) +""" + def _to_np(v: list[float]) -> np.ndarray[Any, np.dtype[np.float32]]: return np.array(v, dtype=np.float32) @@ -164,6 +200,7 @@ def __init__( self._dimension = dimension self._min_conn = min_connections self._max_conn = max_connections + self._content_table = f"{table}_content" self._pool: asyncpg.Pool | None = None # ------------------------------------------------------------------ @@ -206,6 +243,7 @@ async def initialize(self, *, hint: IndexHint | None = None) -> None: if index_ddl is not None: await conn.execute(index_ddl) await conn.execute(_DDL_ACL_INDEX.format(table=self._table)) + await conn.execute(_DDL_CONTENT_TABLE.format(table=self._content_table)) _log.info( "pgvector schema ready", extra={"table": self._table, "dim": self._dimension, "variant": variant}, @@ -306,6 +344,72 @@ async def bulk_delete(self, ctx: RequestContext, chunk_ids: list[ChunkId]) -> No [str(c) for c in chunk_ids], ) + async def put_content(self, ctx: RequestContext, chunks: list[Chunk]) -> None: + """Persist chunk text (+ rebuild fields) so refs can be hydrated later. + + Text lives in a sidecar table keyed by ``(tenant_id, chunk_id)`` — the + vector table stays vectors-only per the SPI. Chunks without inline + ``content`` (content_ref-only) are skipped. + """ + rows = [ + ( + str(ctx.tenant_id), + str(c.id), + str(c.document_id), + str(c.corpus_id) if c.corpus_id else "", + c.position, + c.content, + sorted(c.acl_labels), + ) + for c in chunks + if c.content is not None + ] + if not rows: + return + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.executemany(_SQL_CONTENT_UPSERT.format(table=self._content_table), rows) + + async def hydrate_content( + self, ctx: RequestContext, chunk_refs: list[ChunkRef] + ) -> list[Chunk]: + """Hydrate ``chunk_refs`` into full :class:`Chunk`\\s with text. + + Satisfies the reranker ``Hydrator`` protocol + (``(ctx, refs) -> list[Chunk]``). Preserves input order + per-ref + score; refs with no stored content are dropped. + """ + if not chunk_refs: + return [] + ids = [str(r.chunk_id) for r in chunk_refs] + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + _SQL_CONTENT_SELECT.format(table=self._content_table), + str(ctx.tenant_id), + ids, + ) + by_id = {row["chunk_id"]: row for row in rows} + tenant = TenantId(ctx.tenant_id) + out: list[Chunk] = [] + for ref in chunk_refs: + row = by_id.get(str(ref.chunk_id)) + if row is None: + continue + out.append( + Chunk( + id=ChunkId(row["chunk_id"]), + document_id=DocumentId(row["document_id"]), + tenant_id=tenant, + corpus_id=CorpusId(row["corpus_id"]), + content=row["content"], + position=int(row["position"]), + acl_labels=frozenset(row["acl_labels"]), + score=ref.score, + ) + ) + return out + async def health(self) -> bool: try: pool = await self._get_pool() diff --git a/packages/ingest/src/rag_ingest/pipeline.py b/packages/ingest/src/rag_ingest/pipeline.py index 7b6b6d1..29b3bdd 100644 --- a/packages/ingest/src/rag_ingest/pipeline.py +++ b/packages/ingest/src/rag_ingest/pipeline.py @@ -79,6 +79,7 @@ class _Backends: select_parser: ParserSelector pii_processor: Any | None # rag_pii.PiiProcessor; Any to avoid the dep policy_engine: Any | None # rag_policy.PolicyEngine; Any to avoid the dep + content_store: Any | None = None # optional chunk-text store (hydration) class IngestPipeline: @@ -94,6 +95,7 @@ def __init__( parser_registry: ParserSelector, pii_processor: Any | None = None, policy_engine: Any | None = None, + content_store: Any | None = None, ) -> None: self._b = _Backends( chunker=chunker, @@ -103,6 +105,7 @@ def __init__( select_parser=parser_registry, pii_processor=pii_processor, policy_engine=policy_engine, + content_store=content_store, ) async def ingest_document( @@ -141,6 +144,10 @@ async def ingest_document( embeddings = await _embed(self._b, ctx, filtered) await self._b.index_backend.bulk_index(ctx, embeddings) + if self._b.content_store is not None: + # Persist chunk text so retrieved refs can be hydrated for + # reranking / generation (the vector store holds vectors only). + await self._b.content_store.put_content(ctx, filtered) metadata: dict[str, Any] = { "parser": type(self._b.select_parser(effective_mime)).__name__, diff --git a/uv.lock b/uv.lock index f00d48e..964eca5 100644 --- a/uv.lock +++ b/uv.lock @@ -7196,6 +7196,9 @@ neo4j = [ networkx = [ { name = "networkx" }, ] +openai = [ + { name = "openai" }, +] pinecone = [ { name = "pinecone" }, ] @@ -7228,6 +7231,7 @@ requires-dist = [ { name = "neo4j", marker = "extra == 'neo4j'", specifier = ">=5.20" }, { name = "networkx", marker = "extra == 'networkx'", specifier = ">=3.2" }, { name = "numpy", specifier = ">=1.26" }, + { name = "openai", marker = "extra == 'openai'", specifier = ">=1.30" }, { name = "pgvector", specifier = ">=0.3" }, { name = "pinecone", marker = "extra == 'pinecone'", specifier = ">=5.3" }, { name = "psycopg", extras = ["binary"], marker = "extra == 'postgres-cdc'", specifier = ">=3.2" }, @@ -7241,7 +7245,7 @@ requires-dist = [ { name = "types-aiobotocore", extras = ["s3"], specifier = ">=2.13" }, { name = "weaviate-client", marker = "extra == 'weaviate'", specifier = ">=4.9" }, ] -provides-extras = ["gcs", "azure", "postgres-cdc", "weaviate", "pinecone", "elasticsearch", "tantivy", "neo4j", "memgraph", "networkx", "kms-gcp", "kms-azure", "kms-vault", "dev"] +provides-extras = ["gcs", "azure", "postgres-cdc", "weaviate", "pinecone", "elasticsearch", "tantivy", "openai", "neo4j", "memgraph", "networkx", "kms-gcp", "kms-azure", "kms-vault", "dev"] [[package]] name = "rag-breaker" @@ -7562,7 +7566,8 @@ dependencies = [ [package.optional-dependencies] backends = [ - { name = "rag-backends" }, + { name = "rag-backends", extra = ["openai"] }, + { name = "rag-embedders", extra = ["openai"] }, ] [package.metadata] @@ -7576,7 +7581,7 @@ requires-dist = [ { name = "protobuf", specifier = ">=5.27" }, { name = "python-multipart", specifier = ">=0.0.9" }, { name = "rag-agent", editable = "packages/agent" }, - { name = "rag-backends", marker = "extra == 'backends'", editable = "packages/backends" }, + { name = "rag-backends", extras = ["openai"], marker = "extra == 'backends'", editable = "packages/backends" }, { name = "rag-breaker", editable = "packages/breaker" }, { name = "rag-cache", editable = "packages/cache" }, { name = "rag-chunker", editable = "packages/chunker" }, @@ -7585,6 +7590,7 @@ requires-dist = [ { name = "rag-core", editable = "packages/core" }, { name = "rag-drift", editable = "packages/drift" }, { name = "rag-embedders", editable = "packages/embedders" }, + { name = "rag-embedders", extras = ["openai"], marker = "extra == 'backends'", editable = "packages/embedders" }, { name = "rag-enricher", editable = "packages/enricher" }, { name = "rag-feedback", editable = "packages/feedback" }, { name = "rag-guard", editable = "packages/guard" }, From 628f4084da4fd4ce3f992977811199d676c8f3c5 Mon Sep 17 00:00:00 2001 From: Deep Kumar Singh Kushwah Date: Fri, 19 Jun 2026 02:32:11 +0530 Subject: [PATCH 2/6] =?UTF-8?q?feat(gateway):=20corpus=20&=20webhook=20upd?= =?UTF-8?q?ate=20(PATCH)=20=E2=80=94=20completes=20Step=203.10=20CRUD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Edit action on the Corpora and Webhooks pages was a dead placeholder (BUG-002 / BUG-003): the update (U) of CRUD was unimplemented across UI, gateway, and store. This adds the backend half: - CorpusStore.update + SubscriptionStore.update (SPI + noop + pg, tenant-scoped; embedding model/dim immutable on corpora, signing secret stays masked on subs) - PATCH /v1/corpora/{id} and PATCH /v1/webhooks/subscriptions/{id} on the gateway - CreateCorpusRequest/update wire types in gateway_types + webhook_types - contract + gateway-route tests; OpenAPI re-exported (drift gate) Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/gateway/src/rag_gateway/corpora.py | 49 +++- apps/gateway/src/rag_gateway/webhooks.py | 34 +++ apps/gateway/tests/test_corpora.py | 71 +++++ apps/gateway/tests/test_webhooks.py | 57 ++++ dist/openapi.json | 245 ++++++++++++++++++ dist/openapi.yaml | 174 +++++++++++++ .../rag_backends/corpus/pg_corpus_store.py | 29 ++- packages/core/src/rag_core/gateway_types.py | 27 ++ .../core/src/rag_core/spi/corpus_store.py | 55 +++- .../src/rag_core/spi/noop/corpus_store.py | 29 +++ .../rag_core/spi/noop/subscription_store.py | 29 +++ .../src/rag_core/spi/subscription_store.py | 19 ++ packages/core/src/rag_core/webhook_types.py | 19 ++ tests/contract/test_corpus_store.py | 49 ++++ 14 files changed, 883 insertions(+), 3 deletions(-) diff --git a/apps/gateway/src/rag_gateway/corpora.py b/apps/gateway/src/rag_gateway/corpora.py index 3942bce..5251e6b 100644 --- a/apps/gateway/src/rag_gateway/corpora.py +++ b/apps/gateway/src/rag_gateway/corpora.py @@ -26,7 +26,13 @@ from fastapi import APIRouter, Request, Response from fastapi.responses import JSONResponse from rag_core import get_logger -from rag_core.gateway_types import Corpus, CorpusList, CreateCorpusRequest, GatewayError +from rag_core.gateway_types import ( + Corpus, + CorpusList, + CreateCorpusRequest, + GatewayError, + UpdateCorpusRequest, +) from rag_core.types import CorpusId, RequestContext from rag_gateway.query import _require_ctx_for_handler @@ -165,6 +171,47 @@ async def delete_corpus(request: Request, corpus_id: str) -> Response: ) return Response(status_code=204) + @router.patch( + "/v1/corpora/{corpus_id}", + response_model=Corpus, + responses={ + 401: {"model": GatewayError, "description": "Missing or invalid auth"}, + 404: {"model": GatewayError, "description": "Corpus not found / not visible"}, + }, + summary="Update a corpus", + ) + async def update_corpus( + request: Request, + corpus_id: str, + body: UpdateCorpusRequest, + ) -> JSONResponse | Corpus: + """Patch a corpus's mutable fields (display name / description / metadata). + + PATCH semantics: only the fields present in the body change. Embedding + model + dimension are immutable (they define the vector space). + Cross-tenant / unknown ids return 404 — the same invisibility rule as + ``GET`` — so a caller can't probe for or mutate another tenant's corpora. + """ + ctx = _require_ctx_for_handler(request) + store = request.app.state.corpus_store + updated: Corpus | None = await store.update( + ctx, + CorpusId(corpus_id), + display_name=body.display_name, + description=body.description, + metadata=body.metadata, + ) + if updated is None: + return JSONResponse(status_code=404, content=_not_found_body(corpus_id, ctx)) + _log.info( + "corpus.updated", + extra={ + "rag_tenant_id": str(ctx.tenant_id), + "rag_corpus_id": corpus_id, + }, + ) + return updated + return router diff --git a/apps/gateway/src/rag_gateway/webhooks.py b/apps/gateway/src/rag_gateway/webhooks.py index 19ec4e7..7ca6ad3 100644 --- a/apps/gateway/src/rag_gateway/webhooks.py +++ b/apps/gateway/src/rag_gateway/webhooks.py @@ -24,6 +24,7 @@ from rag_core.webhook_types import ( CreateSubscriptionRequest, SubscriptionList, + UpdateSubscriptionRequest, WebhookEventType, WebhookSubscription, WebhookTestResult, @@ -120,6 +121,39 @@ async def delete_subscription(request: Request, subscription_id: str) -> Respons return _not_found(subscription_id, ctx) return Response(status_code=204) + @router.patch( + "/v1/webhooks/subscriptions/{subscription_id}", + response_model=WebhookSubscription, + responses={ + 401: {"model": GatewayError, "description": "Missing or invalid auth"}, + 404: {"model": GatewayError, "description": "Subscription not found"}, + }, + summary="Update a subscription (secret masked)", + ) + async def update_subscription( + request: Request, subscription_id: str, body: UpdateSubscriptionRequest + ) -> JSONResponse | WebhookSubscription: + ctx = _require_ctx_for_handler(request) + store = request.app.state.subscription_store + updated: WebhookSubscription | None = await store.update( + ctx, + subscription_id, + url=body.url, + event_types=body.event_types, + description=body.description, + active=body.active, + ) + if updated is None: + return _not_found(subscription_id, ctx) + _log.info( + "webhook.subscription_updated", + extra={ + "rag_tenant_id": str(ctx.tenant_id), + "rag_webhook_subscription_id": updated.id, + }, + ) + return updated.redacted() + @router.post( "/v1/webhooks/subscriptions/{subscription_id}/test", response_model=WebhookTestResult, diff --git a/apps/gateway/tests/test_corpora.py b/apps/gateway/tests/test_corpora.py index 6a20924..e4710ef 100644 --- a/apps/gateway/tests/test_corpora.py +++ b/apps/gateway/tests/test_corpora.py @@ -97,3 +97,74 @@ def test_tenant_isolation_on_get_and_delete() -> None: assert client.delete(f"/v1/corpora/{created['id']}", headers=OTHER).status_code == 404 # ...and acme still has it. assert client.get(f"/v1/corpora/{created['id']}", headers=HEADERS).status_code == 200 + + +# ---- update (PATCH /v1/corpora/{id}, Step 3.10 / BUG-002) ----------------- +def test_update_patches_fields_and_persists() -> None: + client = _client() + created = _create( + client, embedding_model="text-embedding-3-small", embedding_dimension=1536 + ) + resp = client.patch( + f"/v1/corpora/{created['id']}", + json={"display_name": "Renamed KB", "description": "updated"}, + headers=HEADERS, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["display_name"] == "Renamed KB" + assert body["description"] == "updated" + # embedding model / dimension are immutable — unchanged by the patch. + assert body["embedding_dimension"] == 1536 + # persisted: a subsequent GET reflects the change. + got = client.get(f"/v1/corpora/{created['id']}", headers=HEADERS).json() + assert got["display_name"] == "Renamed KB" + + +def test_update_only_patches_provided_fields() -> None: + client = _client() + created = _create(client, description="keep me") + resp = client.patch( + f"/v1/corpora/{created['id']}", json={"display_name": "New"}, headers=HEADERS + ) + assert resp.status_code == 200 + body = resp.json() + assert body["display_name"] == "New" + assert body["description"] == "keep me" # omitted → untouched + + +def test_update_unknown_returns_404() -> None: + client = _client() + resp = client.patch( + "/v1/corpora/corpus_nope", json={"display_name": "x"}, headers=HEADERS + ) + assert resp.status_code == 404 + assert resp.json()["error"]["code"] == "corpus_not_found" + + +def test_update_rejects_blank_display_name() -> None: + client = _client() + created = _create(client) + resp = client.patch( + f"/v1/corpora/{created['id']}", json={"display_name": ""}, headers=HEADERS + ) + assert resp.status_code == 422 # pydantic min_length + + +def test_update_requires_auth() -> None: + client = _client() + resp = client.patch("/v1/corpora/corpus_x", json={"display_name": "x"}) + assert resp.status_code == 401 + + +def test_update_tenant_isolation() -> None: + client = _client() + created = _create(client) # owned by acme + # globex can't mutate it... + resp = client.patch( + f"/v1/corpora/{created['id']}", json={"display_name": "Hacked"}, headers=OTHER + ) + assert resp.status_code == 404 + # ...and acme's copy is unchanged. + got = client.get(f"/v1/corpora/{created['id']}", headers=HEADERS).json() + assert got["display_name"] == "Support KB" diff --git a/apps/gateway/tests/test_webhooks.py b/apps/gateway/tests/test_webhooks.py index d3df65c..4a20429 100644 --- a/apps/gateway/tests/test_webhooks.py +++ b/apps/gateway/tests/test_webhooks.py @@ -117,3 +117,60 @@ def test_ingest_emits_ingest_completed() -> None: assert event.type.value == "ingest.completed" assert str(event.tenant_id) == "acme" assert event.data["status"] == resp.json()["status"] + + +# ---- update (PATCH, Step 3.10 / BUG-003) ---------------------------------- +def test_update_patches_fields_and_masks_secret() -> None: + client = _client() + created = _create(client, event_types=["ingest.completed"]) + resp = client.patch( + f"/v1/webhooks/subscriptions/{created['id']}", + json={ + "url": "https://example.test/updated", + "description": "new", + "event_types": ["drift.detected"], + }, + headers=HEADERS, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["url"] == "https://example.test/updated" + assert body["description"] == "new" + assert body["event_types"] == ["drift.detected"] + # the signing secret is masked on the update response (like reads) + assert body["secret"] == "whsec_••••" + # persisted + got = client.get(f"/v1/webhooks/subscriptions/{created['id']}", headers=HEADERS).json() + assert got["url"] == "https://example.test/updated" + + +def test_update_only_patches_provided_fields() -> None: + client = _client() + created = _create(client, description="keep", event_types=["ingest.completed"]) + resp = client.patch( + f"/v1/webhooks/subscriptions/{created['id']}", json={"active": False}, headers=HEADERS + ) + assert resp.status_code == 200 + body = resp.json() + assert body["active"] is False + assert body["description"] == "keep" # omitted → untouched + assert body["event_types"] == ["ingest.completed"] # omitted → untouched + + +def test_update_unknown_returns_404() -> None: + client = _client() + resp = client.patch( + "/v1/webhooks/subscriptions/whsub_nope", json={"active": False}, headers=HEADERS + ) + assert resp.status_code == 404 + assert resp.json()["error"]["code"] == "webhook_subscription_not_found" + + +def test_update_tenant_isolation() -> None: + client = _client() + created = _create(client) # tenant acme + other = {"x-tenant-id": "globex", "x-principal-id": "svc"} + resp = client.patch( + f"/v1/webhooks/subscriptions/{created['id']}", json={"active": False}, headers=other + ) + assert resp.status_code == 404 diff --git a/dist/openapi.json b/dist/openapi.json index c32db2b..2795f99 100644 --- a/dist/openapi.json +++ b/dist/openapi.json @@ -4626,6 +4626,110 @@ "title": "TrustLevel", "type": "string" }, + "UpdateCorpusRequest": { + "description": "``PATCH /v1/corpora/{id}`` body — partial update of mutable fields.\n\nPATCH semantics: only the fields present in the request change; omitted\nfields are left untouched. ``embedding_model`` / ``embedding_dimension``\nare intentionally **not** updatable — they define the corpus's vector\nspace, and changing them would invalidate every embedding already indexed\nunder it. ``id`` / ``tenant_id`` / ``document_count`` are likewise\nimmutable from this surface.", + "properties": { + "description": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "New description (omit to leave unchanged).", + "title": "Description" + }, + "display_name": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "description": "New display name (omit to leave unchanged).", + "title": "Display Name" + }, + "metadata": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "description": "Replacement metadata blob (omit to leave unchanged).", + "title": "Metadata" + } + }, + "title": "UpdateCorpusRequest", + "type": "object" + }, + "UpdateSubscriptionRequest": { + "description": "``PATCH /v1/webhooks/subscriptions/{id}`` body — partial update.\n\nPATCH semantics: only the fields present in the request change. The signing\n``secret`` and ``id`` are immutable from this surface (rotate the secret by\nrecreating the subscription); ``active`` toggles delivery on/off.", + "properties": { + "active": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "description": "Enable/disable delivery (omit to keep).", + "title": "Active" + }, + "description": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "New description (omit to keep).", + "title": "Description" + }, + "event_types": { + "anyOf": [ + { + "items": { + "$ref": "#/components/schemas/WebhookEventType" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "description": "Replacement event allow-list (omit to keep).", + "title": "Event Types" + }, + "url": { + "anyOf": [ + { + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ], + "description": "New endpoint (omit to keep).", + "title": "Url" + } + }, + "title": "UpdateSubscriptionRequest", + "type": "object" + }, "ValidationError": { "properties": { "ctx": { @@ -6169,6 +6273,77 @@ "tags": [ "corpora" ] + }, + "patch": { + "description": "Patch a corpus's mutable fields (display name / description / metadata).\n\nPATCH semantics: only the fields present in the body change. Embedding\nmodel + dimension are immutable (they define the vector space).\nCross-tenant / unknown ids return 404 — the same invisibility rule as\n``GET`` — so a caller can't probe for or mutate another tenant's corpora.", + "operationId": "update_corpus_v1_corpora__corpus_id__patch", + "parameters": [ + { + "in": "path", + "name": "corpus_id", + "required": true, + "schema": { + "title": "Corpus Id", + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UpdateCorpusRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Corpus" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/GatewayError" + } + } + }, + "description": "Missing or invalid auth" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/GatewayError" + } + } + }, + "description": "Corpus not found / not visible" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Update a corpus", + "tags": [ + "corpora" + ] } }, "/v1/embeddings": { @@ -7517,6 +7692,76 @@ "tags": [ "webhooks" ] + }, + "patch": { + "operationId": "update_subscription_v1_webhooks_subscriptions__subscription_id__patch", + "parameters": [ + { + "in": "path", + "name": "subscription_id", + "required": true, + "schema": { + "title": "Subscription Id", + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UpdateSubscriptionRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/WebhookSubscription" + } + } + }, + "description": "Successful Response" + }, + "401": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/GatewayError" + } + } + }, + "description": "Missing or invalid auth" + }, + "404": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/GatewayError" + } + } + }, + "description": "Subscription not found" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Update a subscription (secret masked)", + "tags": [ + "webhooks" + ] } }, "/v1/webhooks/subscriptions/{subscription_id}/test": { diff --git a/dist/openapi.yaml b/dist/openapi.yaml index 9ade1b5..e1d9a75 100644 --- a/dist/openapi.yaml +++ b/dist/openapi.yaml @@ -4065,6 +4065,83 @@ components: - user_supplied title: TrustLevel type: string + UpdateCorpusRequest: + description: '``PATCH /v1/corpora/{id}`` body — partial update of mutable fields. + + + PATCH semantics: only the fields present in the request change; omitted + + fields are left untouched. ``embedding_model`` / ``embedding_dimension`` + + are intentionally **not** updatable — they define the corpus''s vector + + space, and changing them would invalidate every embedding already indexed + + under it. ``id`` / ``tenant_id`` / ``document_count`` are likewise + + immutable from this surface.' + properties: + description: + anyOf: + - type: string + - type: 'null' + description: New description (omit to leave unchanged). + title: Description + display_name: + anyOf: + - minLength: 1 + type: string + - type: 'null' + description: New display name (omit to leave unchanged). + title: Display Name + metadata: + anyOf: + - additionalProperties: true + type: object + - type: 'null' + description: Replacement metadata blob (omit to leave unchanged). + title: Metadata + title: UpdateCorpusRequest + type: object + UpdateSubscriptionRequest: + description: '``PATCH /v1/webhooks/subscriptions/{id}`` body — partial update. + + + PATCH semantics: only the fields present in the request change. The signing + + ``secret`` and ``id`` are immutable from this surface (rotate the secret by + + recreating the subscription); ``active`` toggles delivery on/off.' + properties: + active: + anyOf: + - type: boolean + - type: 'null' + description: Enable/disable delivery (omit to keep). + title: Active + description: + anyOf: + - type: string + - type: 'null' + description: New description (omit to keep). + title: Description + event_types: + anyOf: + - items: + $ref: '#/components/schemas/WebhookEventType' + type: array + - type: 'null' + description: Replacement event allow-list (omit to keep). + title: Event Types + url: + anyOf: + - minLength: 1 + type: string + - type: 'null' + description: New endpoint (omit to keep). + title: Url + title: UpdateSubscriptionRequest + type: object ValidationError: properties: ctx: @@ -5139,6 +5216,60 @@ paths: summary: Describe a single corpus tags: - corpora + patch: + description: 'Patch a corpus''s mutable fields (display name / description / + metadata). + + + PATCH semantics: only the fields present in the body change. Embedding + + model + dimension are immutable (they define the vector space). + + Cross-tenant / unknown ids return 404 — the same invisibility rule as + + ``GET`` — so a caller can''t probe for or mutate another tenant''s corpora.' + operationId: update_corpus_v1_corpora__corpus_id__patch + parameters: + - in: path + name: corpus_id + required: true + schema: + title: Corpus Id + type: string + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/UpdateCorpusRequest' + required: true + responses: + '200': + content: + application/json: + schema: + $ref: '#/components/schemas/Corpus' + description: Successful Response + '401': + content: + application/json: + schema: + $ref: '#/components/schemas/GatewayError' + description: Missing or invalid auth + '404': + content: + application/json: + schema: + $ref: '#/components/schemas/GatewayError' + description: Corpus not found / not visible + '422': + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + description: Validation Error + summary: Update a corpus + tags: + - corpora /v1/embeddings: post: description: 'Embed text via the wired ``Embedder`` SPI. @@ -6083,6 +6214,49 @@ paths: summary: Describe a subscription (secret masked) tags: - webhooks + patch: + operationId: update_subscription_v1_webhooks_subscriptions__subscription_id__patch + parameters: + - in: path + name: subscription_id + required: true + schema: + title: Subscription Id + type: string + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/UpdateSubscriptionRequest' + required: true + responses: + '200': + content: + application/json: + schema: + $ref: '#/components/schemas/WebhookSubscription' + description: Successful Response + '401': + content: + application/json: + schema: + $ref: '#/components/schemas/GatewayError' + description: Missing or invalid auth + '404': + content: + application/json: + schema: + $ref: '#/components/schemas/GatewayError' + description: Subscription not found + '422': + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + description: Validation Error + summary: Update a subscription (secret masked) + tags: + - webhooks /v1/webhooks/subscriptions/{subscription_id}/test: post: operationId: test_subscription_v1_webhooks_subscriptions__subscription_id__test_post diff --git a/packages/backends/src/rag_backends/corpus/pg_corpus_store.py b/packages/backends/src/rag_backends/corpus/pg_corpus_store.py index 523af0f..92526b0 100644 --- a/packages/backends/src/rag_backends/corpus/pg_corpus_store.py +++ b/packages/backends/src/rag_backends/corpus/pg_corpus_store.py @@ -34,7 +34,7 @@ from __future__ import annotations import json -from datetime import datetime +from datetime import datetime, timezone from typing import Any import asyncpg @@ -222,6 +222,33 @@ async def delete(self, ctx: RequestContext, corpus_id: CorpusId) -> bool: ) return row is not None + async def update( + self, + ctx: RequestContext, + corpus_id: CorpusId, + *, + display_name: str | None = None, + description: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> Corpus | None: + # Tenant-scoped read-modify-write: ``get`` enforces the cross-tenant + # invisibility rule, so a miss (or another tenant's id) returns None + # without mutating anything. Reuses ``upsert`` for the write (same + # path as ``create``), which also splits out the routing profile. + current = await self.get(ctx, corpus_id) + if current is None: + return None + patch: dict[str, Any] = {"updated_at": datetime.now(timezone.utc)} + if display_name is not None: + patch["display_name"] = display_name + if description is not None: + patch["description"] = description + if metadata is not None: + patch["metadata"] = metadata + updated = current.model_copy(update=patch) + await self.upsert(updated) + return updated + async def health(self) -> bool: try: pool = await self._get_pool() diff --git a/packages/core/src/rag_core/gateway_types.py b/packages/core/src/rag_core/gateway_types.py index 06e245c..8250a16 100644 --- a/packages/core/src/rag_core/gateway_types.py +++ b/packages/core/src/rag_core/gateway_types.py @@ -124,6 +124,32 @@ class CreateCorpusRequest(BaseModel): metadata: dict[str, Any] = Field(default_factory=dict) +class UpdateCorpusRequest(BaseModel): + """``PATCH /v1/corpora/{id}`` body — partial update of mutable fields. + + PATCH semantics: only the fields present in the request change; omitted + fields are left untouched. ``embedding_model`` / ``embedding_dimension`` + are intentionally **not** updatable — they define the corpus's vector + space, and changing them would invalidate every embedding already indexed + under it. ``id`` / ``tenant_id`` / ``document_count`` are likewise + immutable from this surface. + """ + + model_config = {"frozen": True} + + display_name: str | None = Field( + default=None, + min_length=1, + description="New display name (omit to leave unchanged).", + ) + description: str | None = Field( + default=None, description="New description (omit to leave unchanged)." + ) + metadata: dict[str, Any] | None = Field( + default=None, description="Replacement metadata blob (omit to leave unchanged)." + ) + + # --------------------------------------------------------------------------- # Request shapes # --------------------------------------------------------------------------- @@ -623,6 +649,7 @@ class GatewayError(BaseModel): "Corpus", "CorpusList", "CreateCorpusRequest", + "UpdateCorpusRequest", "ExperimentAssignment", "FeedbackAck", "FeedbackRequest", diff --git a/packages/core/src/rag_core/spi/corpus_store.py b/packages/core/src/rag_core/spi/corpus_store.py index 22595cc..7849e40 100644 --- a/packages/core/src/rag_core/spi/corpus_store.py +++ b/packages/core/src/rag_core/spi/corpus_store.py @@ -23,7 +23,8 @@ from __future__ import annotations import abc -from typing import TYPE_CHECKING +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any from rag_core.spi._base import HealthCheckMixin from rag_core.types import CorpusId, RequestContext @@ -95,5 +96,57 @@ async def delete(self, ctx: RequestContext, corpus_id: CorpusId) -> bool: corpora). Implementations MUST NOT raise on a missing id. """ + @abc.abstractmethod + async def update( + self, + ctx: RequestContext, + corpus_id: CorpusId, + *, + display_name: str | None = None, + description: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> Corpus | None: + """Patch mutable fields of a corpus in ``ctx.tenant_id``; return it. + + Only non-``None`` arguments are applied (PATCH semantics); omitted + fields are left unchanged. ``embedding_model`` / ``embedding_dimension`` + are immutable — they define the corpus's vector space — so are not + updatable here. Implementations MUST bump ``updated_at``. + + Returns the updated corpus, or ``None`` when no corpus with that id is + visible to ``ctx.tenant_id`` (missing, or owned by a different tenant — + the same cross-tenant invisibility rule as :meth:`get` / :meth:`delete`, + so a caller can't probe for or mutate another tenant's corpora). + Implementations MUST NOT raise on a missing id. + """ + + async def increment_document_count( + self, + ctx: RequestContext, + corpus_id: CorpusId, + delta: int = 1, + ) -> Corpus | None: + """Adjust the *managed* ``document_count`` by ``delta`` (default +1). + + Called by the ingest path when a document is added to (or removed from) + a corpus. ``document_count`` is managed, not user-editable — hence it + is absent from :meth:`update`. Concrete default: read-modify-write via + :meth:`get` + :meth:`create` (every impl's ``create`` upserts by id), + bumping ``updated_at``; backends MAY override with an atomic increment. + Tenant-scoped and clamped at zero; returns the updated corpus, or + ``None`` when not visible to ``ctx.tenant_id`` (same invisibility rule + as :meth:`get`, so a caller can't probe another tenant's corpora). + """ + current = await self.get(ctx, corpus_id) + if current is None: + return None + bumped = current.model_copy( + update={ + "document_count": max(0, current.document_count + delta), + "updated_at": datetime.now(UTC), + } + ) + return await self.create(ctx, bumped) + __all__ = ["CorpusStore"] diff --git a/packages/core/src/rag_core/spi/noop/corpus_store.py b/packages/core/src/rag_core/spi/noop/corpus_store.py index 81d4bfa..fd61cc1 100644 --- a/packages/core/src/rag_core/spi/noop/corpus_store.py +++ b/packages/core/src/rag_core/spi/noop/corpus_store.py @@ -7,6 +7,9 @@ from __future__ import annotations +from datetime import datetime, timezone +from typing import Any + from rag_core.gateway_types import Corpus from rag_core.spi.corpus_store import CorpusStore from rag_core.types import CorpusId, RequestContext @@ -52,6 +55,32 @@ async def delete(self, ctx: RequestContext, corpus_id: CorpusId) -> bool: # (or probe for) tenant B's corpus by id. return self._by_key.pop((str(ctx.tenant_id), str(corpus_id)), None) is not None + async def update( + self, + ctx: RequestContext, + corpus_id: CorpusId, + *, + display_name: str | None = None, + description: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> Corpus | None: + # Tenant-scoped read-modify-write: a miss (or another tenant's id) + # returns None without mutating anything — same invisibility rule as get. + key = (str(ctx.tenant_id), str(corpus_id)) + current = self._by_key.get(key) + if current is None: + return None + patch: dict[str, Any] = {"updated_at": datetime.now(timezone.utc)} + if display_name is not None: + patch["display_name"] = display_name + if description is not None: + patch["description"] = description + if metadata is not None: + patch["metadata"] = metadata + updated = current.model_copy(update=patch) + self._by_key[key] = updated + return updated + async def health(self) -> bool: return True diff --git a/packages/core/src/rag_core/spi/noop/subscription_store.py b/packages/core/src/rag_core/spi/noop/subscription_store.py index 9a19071..e1675e6 100644 --- a/packages/core/src/rag_core/spi/noop/subscription_store.py +++ b/packages/core/src/rag_core/spi/noop/subscription_store.py @@ -2,6 +2,8 @@ from __future__ import annotations +from datetime import datetime, timezone + from rag_core.spi.subscription_store import SubscriptionStore from rag_core.types import RequestContext from rag_core.webhook_types import WebhookEventType, WebhookSubscription @@ -40,5 +42,32 @@ async def list_for_event( async def delete(self, ctx: RequestContext, subscription_id: str) -> bool: return self._bucket(ctx).pop(subscription_id, None) is not None + async def update( + self, + ctx: RequestContext, + subscription_id: str, + *, + url: str | None = None, + event_types: list[WebhookEventType] | None = None, + description: str | None = None, + active: bool | None = None, + ) -> WebhookSubscription | None: + bucket = self._bucket(ctx) + current = bucket.get(subscription_id) + if current is None: + return None + patch: dict[str, object] = {"updated_at": datetime.now(timezone.utc)} + if url is not None: + patch["url"] = url + if event_types is not None: + patch["event_types"] = event_types + if description is not None: + patch["description"] = description + if active is not None: + patch["active"] = active + updated = current.model_copy(update=patch) + bucket[subscription_id] = updated + return updated + async def health(self) -> bool: return True diff --git a/packages/core/src/rag_core/spi/subscription_store.py b/packages/core/src/rag_core/spi/subscription_store.py index 0a590e3..c6d47a5 100644 --- a/packages/core/src/rag_core/spi/subscription_store.py +++ b/packages/core/src/rag_core/spi/subscription_store.py @@ -53,3 +53,22 @@ async def list_for_event( @abc.abstractmethod async def delete(self, ctx: RequestContext, subscription_id: str) -> bool: """Delete the subscription; return True if it existed, False otherwise.""" + + @abc.abstractmethod + async def update( + self, + ctx: RequestContext, + subscription_id: str, + *, + url: str | None = None, + event_types: list[WebhookEventType] | None = None, + description: str | None = None, + active: bool | None = None, + ) -> WebhookSubscription | None: + """Patch mutable fields of a subscription in ``ctx.tenant_id``; return it. + + Only non-``None`` args are applied (PATCH semantics). ``secret`` / ``id`` + are immutable here. Returns ``None`` if no subscription with that id is + owned by ``ctx.tenant_id`` (the cross-tenant invisibility rule); MUST NOT + raise on a missing id. + """ diff --git a/packages/core/src/rag_core/webhook_types.py b/packages/core/src/rag_core/webhook_types.py index b0a7dc4..9b5b839 100644 --- a/packages/core/src/rag_core/webhook_types.py +++ b/packages/core/src/rag_core/webhook_types.py @@ -200,6 +200,24 @@ class CreateSubscriptionRequest(BaseModel): description: str = "" +class UpdateSubscriptionRequest(BaseModel): + """``PATCH /v1/webhooks/subscriptions/{id}`` body — partial update. + + PATCH semantics: only the fields present in the request change. The signing + ``secret`` and ``id`` are immutable from this surface (rotate the secret by + recreating the subscription); ``active`` toggles delivery on/off. + """ + + model_config = {"frozen": True} + + url: str | None = Field(default=None, min_length=1, description="New endpoint (omit to keep).") + event_types: list[WebhookEventType] | None = Field( + default=None, description="Replacement event allow-list (omit to keep)." + ) + description: str | None = Field(default=None, description="New description (omit to keep).") + active: bool | None = Field(default=None, description="Enable/disable delivery (omit to keep).") + + class SubscriptionList(BaseModel): """``GET /v1/webhooks/subscriptions`` response.""" @@ -237,6 +255,7 @@ async def publish(self, ctx: RequestContext, event: WebhookEvent) -> list[Webhoo __all__ = [ "CreateSubscriptionRequest", + "UpdateSubscriptionRequest", "DeliveryStatus", "SubscriptionList", "WebhookDelivery", diff --git a/tests/contract/test_corpus_store.py b/tests/contract/test_corpus_store.py index a20bfd5..09824a5 100644 --- a/tests/contract/test_corpus_store.py +++ b/tests/contract/test_corpus_store.py @@ -120,3 +120,52 @@ async def test_delete_cross_tenant_is_a_noop() -> None: assert await store.delete(_ctx("tenant-a"), CorpusId("wiki")) is False # still present for its real owner. assert await store.get(_ctx("tenant-b"), CorpusId("wiki")) is not None + + +# ---- update (admin write surface, Step 3.10 / BUG-002) -------------------- +async def test_update_patches_display_name_and_description() -> None: + store = NoopCorpusStore( + seed=[_corpus("docs", description="old", embedding_model="m", embedding_dimension=8)] + ) + updated = await store.update( + _ctx(), CorpusId("docs"), display_name="New Name", description="new" + ) + assert updated is not None + assert updated.display_name == "New Name" + assert updated.description == "new" + # embedding model / dimension are immutable — never touched by update. + assert updated.embedding_model == "m" + assert updated.embedding_dimension == 8 + # and the change is persisted, not just returned. + again = await store.get(_ctx(), CorpusId("docs")) + assert again is not None and again.display_name == "New Name" + + +async def test_update_only_changes_provided_fields() -> None: + store = NoopCorpusStore(seed=[_corpus("docs", display_name="Keep", description="old")]) + updated = await store.update(_ctx(), CorpusId("docs"), description="new") + assert updated is not None + assert updated.display_name == "Keep" # omitted → untouched + assert updated.description == "new" + + +async def test_update_bumps_updated_at() -> None: + seed = _corpus("docs") + store = NoopCorpusStore(seed=[seed]) + updated = await store.update(_ctx(), CorpusId("docs"), display_name="X") + assert updated is not None + assert updated.updated_at >= seed.updated_at + + +async def test_update_missing_returns_none() -> None: + store = NoopCorpusStore() + assert await store.update(_ctx(), CorpusId("nope"), display_name="X") is None + + +async def test_update_cross_tenant_returns_none() -> None: + """Tenant A must NOT be able to mutate tenant B's corpus by id.""" + store = NoopCorpusStore(seed=[_corpus("wiki", tenant="tenant-b", display_name="Orig")]) + assert await store.update(_ctx("tenant-a"), CorpusId("wiki"), display_name="Hacked") is None + # unchanged for its real owner. + out = await store.get(_ctx("tenant-b"), CorpusId("wiki")) + assert out is not None and out.display_name == "Orig" From be7a42c259a5e7a3c5d3eabab2879dbec27e23bd Mon Sep 17 00:00:00 2001 From: Deep Kumar Singh Kushwah Date: Fri, 19 Jun 2026 02:32:27 +0530 Subject: [PATCH 3/6] fix(admin-ui): Edit dialogs, demo-source badges, hydration & stale copy Resolves the admin-console bugs found in this session's testing pass: - BUG-002/003: wire the corpus & webhook Edit dialogs to the new PATCH routes (updateCorpus / updateWebhook in api.ts), with demo-mode local simulation - BUG-004: each seed-fallback Live-Status card (breakers/quotas/drift/feedback/ cost/experiments) renders a per-source SourceBadge so seed data isn't shown as Live - BUG-007: suppressHydrationWarning on the Metrics 'Captured' locale timestamp - BUG-008: drop stale 'lands with Phase 6' copy on API Keys + Tenants (project is GA) - BUG-009: Playground 'Demo data' badge only shows after a mock run, not on load - chat: default RAG scope to the tenant's live corpora once loaded Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/admin-ui/src/app/api-keys/page.tsx | 2 +- apps/admin-ui/src/app/chat/page.tsx | 10 ++- apps/admin-ui/src/app/corpora/page.tsx | 95 ++++++++++++++++++++++- apps/admin-ui/src/app/metrics/page.tsx | 2 +- apps/admin-ui/src/app/playground/page.tsx | 14 +++- apps/admin-ui/src/app/status/page.tsx | 20 ++--- apps/admin-ui/src/app/tenants/page.tsx | 2 +- apps/admin-ui/src/app/webhooks/page.tsx | 94 +++++++++++++++++++++- apps/admin-ui/src/lib/api.ts | 22 ++++++ apps/admin-ui/test/status.test.tsx | 5 +- 10 files changed, 245 insertions(+), 21 deletions(-) diff --git a/apps/admin-ui/src/app/api-keys/page.tsx b/apps/admin-ui/src/app/api-keys/page.tsx index 4a16f7b..587ac1f 100644 --- a/apps/admin-ui/src/app/api-keys/page.tsx +++ b/apps/admin-ui/src/app/api-keys/page.tsx @@ -52,7 +52,7 @@ export default function ApiKeysPage() { />
- API-key management is a preview — the backend lands with Phase 6 governance. Data shown here is local. + API-key management is a preview — the data shown here is local and not yet backed by the gateway.
diff --git a/apps/admin-ui/src/app/chat/page.tsx b/apps/admin-ui/src/app/chat/page.tsx index 7c067a9..0a34954 100644 --- a/apps/admin-ui/src/app/chat/page.tsx +++ b/apps/admin-ui/src/app/chat/page.tsx @@ -130,7 +130,7 @@ function RagSummary({ msg }: { msg: ThreadMessage }) { function ChatTab({ models }: { models: ModelInfo[] }) { const { tenant } = useApp(); - const { rows: corpora } = useLive( + const { rows: corpora, source: corporaSource } = useLive( (tid) => CORPORA.filter((c) => c.tenant_id === tid), fetchCorpora, ); @@ -157,6 +157,14 @@ function ChatTab({ models }: { models: ModelInfo[] }) { setMessages([]); setRagCorpusIds([]); }, [tenant.id]); + // Default RAG scope to all of the tenant's live corpora once loaded, so chat + // grounds on your data with an explicit corpus_ids (empty would search all + // anyway, but the explicit payload is clearer). Live data only. + useEffect(() => { + if (corporaSource === "live" && corpora.length > 0) { + setRagCorpusIds(corpora.map((c) => c.id)); + } + }, [corporaSource, corpora]); useEffect( () => () => { if (streamTimer.current) clearInterval(streamTimer.current); diff --git a/apps/admin-ui/src/app/corpora/page.tsx b/apps/admin-ui/src/app/corpora/page.tsx index 9616187..c35c5ec 100644 --- a/apps/admin-ui/src/app/corpora/page.tsx +++ b/apps/admin-ui/src/app/corpora/page.tsx @@ -26,7 +26,7 @@ import { TableSkeleton, Textarea, } from "@/components/ui/primitives"; -import { createCorpus, deleteCorpus, fetchCorpora } from "@/lib/api"; +import { createCorpus, deleteCorpus, fetchCorpora, updateCorpus } from "@/lib/api"; import { GATEWAY_URL } from "@/lib/config"; import { CORPORA } from "@/lib/mock"; import { absTime, relTime } from "@/lib/time"; @@ -50,6 +50,7 @@ export default function CorporaPage() { const [selected, setSelected] = useState(null); const [creating, setCreating] = useState(false); + const [editing, setEditing] = useState(null); const [deleting, setDeleting] = useState(null); return ( @@ -127,7 +128,9 @@ export default function CorporaPage() { } onClick={() => setSelected(c)}> View details - }>Edit + } onClick={() => setEditing(c)}> + Edit + } onClick={() => setDeleting(c)}> Delete @@ -225,6 +228,34 @@ export default function CorporaPage() { /> )} + {editing && ( + setEditing(null)} + onSave={async ({ display_name, description }) => { + let updated: Corpus; + if (GATEWAY_URL && source === "live") { + try { + updated = await updateCorpus(tenant.id, editing.id, { display_name, description }); + } catch { + toast({ title: "Failed to update corpus", variant: "error" }); + return; + } + } else { + updated = { + ...editing, + display_name, + description, + updated_at: new Date().toISOString(), + }; + } + setRows((prev) => prev.map((c) => (c.id === updated.id ? updated : c))); + setEditing(null); + toast({ title: "Corpus updated", description: updated.display_name, variant: "success" }); + }} + /> + )} + setDeleting(null)} @@ -326,3 +357,63 @@ function NewCorpusDialog({ ); } + +function EditCorpusDialog({ + corpus, + onClose, + onSave, +}: { + corpus: Corpus; + onClose: () => void; + onSave: (v: { display_name: string; description: string }) => void; +}) { + const [name, setName] = useState(corpus.display_name); + const [description, setDescription] = useState(corpus.description ?? ""); + const dirty = + name.trim() !== corpus.display_name || description.trim() !== (corpus.description ?? ""); + + return ( + +
+ + setName(e.target.value)} + placeholder="Support Knowledge Base" + /> + + +