From 47dc4601759a51780c24b26ecedc1fc6c952c8e0 Mon Sep 17 00:00:00 2001 From: rostislav Date: Wed, 29 Jul 2026 10:42:25 +0300 Subject: [PATCH 1/3] fix(rag): tolerate custom Magento XML and recover Qdrant writes - exclude nested custom etc XML from Magento architecture parsing - keep custom XML available for generic semantic indexing - retry transient Qdrant upsert failures with bounded backoff - split rejected batches to isolate oversized or invalid points - preserve atomic index publication after exhausted recovery - add regression coverage and update RAG/plugin documentation --- README.md | 1 + analysis-plugins/README.md | 8 + .../python/tests/test_builtin_plugins.py | 42 +++++ .../codecrow_plugin_magento/__init__.py | 3 +- .../codecrow_plugin_magento/architecture.py | 26 +++- .../codecrow_plugin_magento/repository.py | 5 +- .../core/index_manager/point_operations.py | 144 +++++++++++++++++- .../tests/test_point_operations.py | 114 ++++++++++++++ 8 files changed, 332 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index c29572ba..6d204991 100644 --- a/README.md +++ b/README.md @@ -155,6 +155,7 @@ normal review pipeline continues. | Retrieval | Qdrant semantic search combined with deterministic metadata and exact-path lookup | | Stored Context | Semantic source chunks plus exact-source, architecture, graph, plugin snapshot, and repository-detection points | | Full Reindex | Builds a pending collection generation and atomically swaps the project alias only after successful completion | +| Resilient Writes | Retries transient Qdrant writes and subdivides rejected batches; an unrecoverable point keeps the pending generation inactive | | Incremental Reindex | Applies one pinned commit change set and replaces changed semantic chunks together with affected graph and state groups | | PR Context | Uses an immutable, commit-pinned PR overlay so changed files do not retrieve stale base-branch copies | | Compatibility Guard | Returns HTTP 409 for stale or incompatible representation/plugin state before a review-model call; recovery is a full reindex with matching service images | diff --git a/analysis-plugins/README.md b/analysis-plugins/README.md index 11130a27..f79f845a 100644 --- a/analysis-plugins/README.md +++ b/analysis-plugins/README.md @@ -146,3 +146,11 @@ applies it before full/incremental file loading and PR overlay construction; and inference maps generated/excluded files to explicit non-reviewable hunk dispositions before Stage 0. Empty registries keep the generic fallback, while a failed policy contribution aborts instead of silently changing the disposition. + +Magento classifies XML as module configuration only when it is directly below an +`etc` directory or directly below a known Magento area such as `etc/frontend`. +Custom descendants such as `etc/samples/*.xml` stay `full` project content and +are handled by the generic indexing path. DTDs and entities in those ordinary +documents are not resolved by the Magento repository analyzer. Actual Magento +configuration XML remains architecture input and fails closed when it contains a +DTD or entity declaration. diff --git a/analysis-plugins/contracts/python/tests/test_builtin_plugins.py b/analysis-plugins/contracts/python/tests/test_builtin_plugins.py index 3f4db32c..174db091 100644 --- a/analysis-plugins/contracts/python/tests/test_builtin_plugins.py +++ b/analysis-plugins/contracts/python/tests/test_builtin_plugins.py @@ -689,6 +689,48 @@ def test_magento_rejects_xml_entities_without_resolution(): assert [diagnostic.code for diagnostic in diagnostics] == ["magento-unsafe-xml"] +def test_magento_keeps_custom_entity_bearing_xml_outside_architecture_analysis(): + catalog = PluginCatalog.discover(PLUGINS_ROOT) + runtime = PluginRuntime(catalog) + capabilities = ProjectSelector(catalog.registry).select(_facts()) + path = "app/code/Punchout/Gateway/etc/samples/cxml_inv_po.xml" + artifact = FileArtifact( + path, + """ + + +""", + ) + + assert runtime.file_disposition( + path, + capabilities, + ) is FileDisposition.FULL + + handle = runtime.start_repository_analysis( + capabilities, + "0123456789abcdef", + ) + handle.ingest(tuple(sorted(( + artifact, + FileArtifact( + "app/code/Punchout/Gateway/etc/module.xml", + '', + ), + FileArtifact( + "app/etc/config.php", + " ['Punchout_Gateway' => 1]];", + ), + ), key=lambda value: value.path))) + analysis, diagnostics = handle.finish() + + assert diagnostics == () + assert all( + path not in packet.paths + for packet in analysis.packets + ) + + def test_magento_file_policy_keeps_architecture_without_vendor_test_vectors(): catalog = PluginCatalog.discover(PLUGINS_ROOT) runtime = PluginRuntime(catalog) diff --git a/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/__init__.py b/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/__init__.py index e9d8bff4..3c141d1f 100644 --- a/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/__init__.py +++ b/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/__init__.py @@ -11,6 +11,7 @@ ValidationResult, ) +from .architecture import is_magento_config_xml from .repository import MagentoRepositorySession @@ -101,7 +102,7 @@ def file_disposition(self, path: str): ): return PluginOutcome.handled(FileDisposition.ARCHITECTURE_ONLY) if normalized.endswith(".xml") and ( - "/etc/" in normalized + is_magento_config_xml(path) or "/layout/" in normalized or "/ui_component/" in normalized ): diff --git a/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/architecture.py b/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/architecture.py index 7de2f55b..dd72f78e 100644 --- a/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/architecture.py +++ b/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/architecture.py @@ -18,6 +18,29 @@ ) +def is_magento_config_xml(path: str) -> bool: + """Return whether an XML path participates in Magento config merging. + + A module configuration file is either directly below ``etc`` or below one + known Magento area. Deeper/custom directories such as ``etc/samples`` are + ordinary project content, even though their path contains an ``etc`` + segment. Treating every descendant as Magento configuration makes valid + payload formats such as cXML part of repository architecture analysis. + """ + normalized = path.replace("\\", "/").strip("/").casefold() + if not normalized.endswith(".xml"): + return False + marker = "/etc/" + candidate = f"/{normalized}" + if marker not in candidate: + return False + tail = candidate.split(marker, 1)[1].split("/") + return ( + len(tail) == 1 + or (len(tail) == 2 and tail[0] in MAGENTO_AREAS) + ) + + def tag(element: ET.Element) -> str: return element.tag.rsplit("}", 1)[-1] @@ -32,7 +55,8 @@ def safe_xml(plugin_id: str, path: str, content: str): if " None: for fact in packet.facts } for path in sorted(self.artifacts): - if not path.endswith(".xml") or "/etc/" not in f"/{path}": + if not is_magento_config_xml(path): continue module = self._module_for_path(path, modules) is_application_config = path.startswith("app/etc/") @@ -6916,7 +6917,7 @@ def ingest(self, artifacts: tuple[FileArtifact, ...]) -> None: self.artifacts.pop(path, None) continue filename = PurePosixPath(path).name - is_config = path.endswith(".xml") and "/etc/" in f"/{path}" + is_config = is_magento_config_xml(path) is_schema_whitelist = ( filename == "db_schema_whitelist.json" and "/etc/" in f"/{path}" diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/point_operations.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/point_operations.py index 5b545449..727d3f0c 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/point_operations.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/point_operations.py @@ -5,6 +5,7 @@ """ import logging +import time import uuid from datetime import datetime, timezone from typing import List, Dict, Tuple @@ -25,11 +26,21 @@ def __init__( embed_model, batch_size: int = 50, embedding_dim: int | None = None, + upsert_max_attempts: int = 3, + upsert_retry_base_seconds: float = 0.25, ): + if batch_size <= 0: + raise ValueError("batch_size must be positive") + if upsert_max_attempts <= 0: + raise ValueError("upsert_max_attempts must be positive") + if upsert_retry_base_seconds < 0: + raise ValueError("upsert_retry_base_seconds cannot be negative") self.client = client self.embed_model = embed_model self.batch_size = batch_size self.embedding_dim = embedding_dim + self.upsert_max_attempts = upsert_max_attempts + self.upsert_retry_base_seconds = upsert_retry_base_seconds @staticmethod def generate_point_id( @@ -161,17 +172,136 @@ def upsert_points( for i in range(0, len(points), self.batch_size): batch = points[i:i + self.batch_size] + batch_successful, batch_failed = self._upsert_resilient( + collection_name, + batch, + batch_offset=i, + ) + successful += batch_successful + failed += batch_failed + + return successful, failed + + @staticmethod + def _status_code(exception: Exception) -> int | None: + status_code = getattr(exception, "status_code", None) + if isinstance(status_code, int): + return status_code + response = getattr(exception, "response", None) + status_code = getattr(response, "status_code", None) + return status_code if isinstance(status_code, int) else None + + @classmethod + def _is_batch_shape_failure(cls, exception: Exception | None) -> bool: + """Return whether a smaller request can isolate a rejected point.""" + if exception is None: + return False + status_code = cls._status_code(exception) + if status_code in {400, 413, 422}: + return True + message = str(exception).casefold() + return any( + marker in message + for marker in ( + "bad request", + "payload too large", + "request entity too large", + "request too large", + "validation error", + "vector dimension", + "wrong input", + ) + ) + + @staticmethod + def _point_label(point: PointStruct) -> str: + payload = point.payload or {} + path = payload.get("path", "") + return f"id={point.id} path={path}" + + def _upsert_resilient( + self, + collection_name: str, + points: List[PointStruct], + *, + batch_offset: int, + ) -> tuple[int, int]: + """Write a batch with bounded retry and rejected-request subdivision.""" + if not points: + return 0, 0 + + error = None + for attempt in range(1, self.upsert_max_attempts + 1): try: self.client.upsert( collection_name=collection_name, - points=batch + points=points, ) - successful += len(batch) - except Exception as e: - logger.error(f"Failed to upsert batch starting at {i}: {e}") - failed += len(batch) - - return successful, failed + return len(points), 0 + except Exception as exception: + error = exception + if self._is_batch_shape_failure(exception): + break + if attempt >= self.upsert_max_attempts: + logger.error( + "Qdrant upsert failed after %s attempts at point " + "offset %s (%s points): %s", + self.upsert_max_attempts, + batch_offset, + len(points), + exception, + ) + return 0, len(points) + delay = self.upsert_retry_base_seconds * (2 ** (attempt - 1)) + logger.warning( + "Qdrant upsert failed at point offset %s " + "(%s points, attempt %s/%s); retrying in %.2fs: %s", + batch_offset, + len(points), + attempt, + self.upsert_max_attempts, + delay, + exception, + ) + if delay: + time.sleep(delay) + + if len(points) == 1: + logger.error( + "Qdrant rejected one exact index point after bounded recovery " + "(%s): %s", + self._point_label(points[0]), + error, + ) + return 0, 1 + + midpoint = len(points) // 2 + left = points[:midpoint] + right = points[midpoint:] + + logger.warning( + "Qdrant rejected a %s-point request at offset %s; " + "isolating it into %s and %s points: %s", + len(points), + batch_offset, + len(left), + len(right), + error, + ) + left_result = self._upsert_resilient( + collection_name, + left, + batch_offset=batch_offset, + ) + right_result = self._upsert_resilient( + collection_name, + right, + batch_offset=batch_offset + midpoint, + ) + return ( + left_result[0] + right_result[0], + left_result[1] + right_result[1], + ) def process_and_upsert_chunks( self, diff --git a/python-ecosystem/rag-pipeline/tests/test_point_operations.py b/python-ecosystem/rag-pipeline/tests/test_point_operations.py index bbc84eb2..03b300af 100644 --- a/python-ecosystem/rag-pipeline/tests/test_point_operations.py +++ b/python-ecosystem/rag-pipeline/tests/test_point_operations.py @@ -1,6 +1,8 @@ +import uuid from unittest.mock import MagicMock from llama_index.core.schema import TextNode +from qdrant_client.models import PointStruct from rag_pipeline.core.index_manager.point_operations import PointOperations @@ -59,6 +61,8 @@ def test_process_streams_embedding_and_pending_writes_in_point_batches(): embed_model, embedding_dim=3, batch_size=2, + upsert_max_attempts=1, + upsert_retry_base_seconds=0, ) chunks = [ TextNode(text=f"chunk-{index}", metadata={"path": "Service.php"}) @@ -98,6 +102,8 @@ def test_process_stops_embedding_after_pending_write_failure(): embed_model, embedding_dim=3, batch_size=2, + upsert_max_attempts=1, + upsert_retry_base_seconds=0, ) chunks = [ TextNode(text=f"chunk-{index}", metadata={"path": "Service.php"}) @@ -114,3 +120,111 @@ def test_process_stops_embedding_after_pending_write_failure(): assert (successful, failed) == (0, 2) embed_model.get_text_embedding_batch.assert_called_once() + + +def _points(count: int): + return [ + PointStruct( + id=str(uuid.uuid4()), + vector=[0.1, 0.2, 0.3], + payload={"path": f"src/chunk-{index}.php"}, + ) + for index in range(count) + ] + + +def test_upsert_retries_transient_qdrant_failure_without_losing_points(): + client = MagicMock() + client.upsert.side_effect = [ + RuntimeError("temporary transport failure"), + None, + ] + operations = PointOperations( + client, + MagicMock(), + batch_size=4, + upsert_max_attempts=3, + upsert_retry_base_seconds=0, + ) + + successful, failed = operations.upsert_points("pending", _points(4)) + + assert (successful, failed) == (4, 0) + assert client.upsert.call_count == 2 + + +def test_upsert_splits_rejected_request_until_smaller_slices_succeed(): + class RequestTooLarge(RuntimeError): + status_code = 413 + + client = MagicMock() + + def reject_multi_point_request(**kwargs): + if len(kwargs["points"]) > 1: + raise RequestTooLarge("request entity too large") + + client.upsert.side_effect = reject_multi_point_request + operations = PointOperations( + client, + MagicMock(), + batch_size=4, + upsert_max_attempts=3, + upsert_retry_base_seconds=0, + ) + + successful, failed = operations.upsert_points("pending", _points(4)) + + assert (successful, failed) == (4, 0) + assert client.upsert.call_count == 7 + accepted = [ + call.kwargs["points"] + for call in client.upsert.call_args_list + if len(call.kwargs["points"]) == 1 + ] + assert len(accepted) == 4 + + +def test_upsert_isolates_one_rejected_point_without_losing_valid_siblings(): + class InvalidPoint(RuntimeError): + status_code = 400 + + points = _points(4) + rejected_id = points[2].id + client = MagicMock() + + def reject_one_point(**kwargs): + if any(point.id == rejected_id for point in kwargs["points"]): + raise InvalidPoint("bad request") + + client.upsert.side_effect = reject_one_point + operations = PointOperations( + client, + MagicMock(), + batch_size=4, + upsert_retry_base_seconds=0, + ) + + successful, failed = operations.upsert_points("pending", points) + + assert (successful, failed) == (3, 1) + assert any( + call.kwargs["points"] == [points[2]] + for call in client.upsert.call_args_list + ) + + +def test_upsert_stops_subdivision_when_qdrant_is_globally_unavailable(): + client = MagicMock() + client.upsert.side_effect = RuntimeError("qdrant unavailable") + operations = PointOperations( + client, + MagicMock(), + batch_size=4, + upsert_max_attempts=2, + upsert_retry_base_seconds=0, + ) + + successful, failed = operations.upsert_points("pending", _points(4)) + + assert (successful, failed) == (0, 4) + assert client.upsert.call_count == 2 From 24d7b259f9c918b8a8ef5f3f862b8cdc156317ef Mon Sep 17 00:00:00 2001 From: rostislav Date: Wed, 29 Jul 2026 11:51:31 +0300 Subject: [PATCH 2/3] feat: newrelic for RAG Pipeline service --- deployment/ci/ci-build.sh | 2 +- deployment/ci/server-deploy.sh | 5 +- deployment/ci/server-init.sh | 36 +++++++---- deployment/config/.gitignore | 3 +- deployment/config/rag-pipeline/.env.sample | 5 ++ deployment/docker-compose.prod.yml | 3 + python-ecosystem/rag-pipeline/.gitignore | 2 +- .../rag-pipeline/Dockerfile.observable | 63 +++++++++++++++++++ python-ecosystem/rag-pipeline/main.py | 36 ++++++++++- .../rag-pipeline/requirements.local.txt | 3 + .../rag-pipeline/requirements.txt | 3 + .../rag-pipeline/src/rag_pipeline/api/api.py | 11 ++++ 12 files changed, 154 insertions(+), 18 deletions(-) create mode 100644 python-ecosystem/rag-pipeline/Dockerfile.observable diff --git a/deployment/ci/ci-build.sh b/deployment/ci/ci-build.sh index 68cbe1ac..0c9f1edf 100755 --- a/deployment/ci/ci-build.sh +++ b/deployment/ci/ci-build.sh @@ -169,7 +169,7 @@ set_image_definition() { rag-pipeline) IMAGE_NAME="codecrow/rag-pipeline" CONTEXT="." - DOCKERFILE="python-ecosystem/rag-pipeline/Dockerfile" + DOCKERFILE="python-ecosystem/rag-pipeline/Dockerfile.observable" ;; web-frontend) IMAGE_NAME="codecrow/web-frontend" diff --git a/deployment/ci/server-deploy.sh b/deployment/ci/server-deploy.sh index 5a3dcb54..f0a59312 100755 --- a/deployment/ci/server-deploy.sh +++ b/deployment/ci/server-deploy.sh @@ -71,7 +71,10 @@ if codecrow_includes_service "inference-orchestrator" "${SELECTED_SERVICES[@]}"; fi if codecrow_includes_service "rag-pipeline" "${SELECTED_SERVICES[@]}"; then - REQUIRED_CONFIGS+=("$CONFIG_DIR/rag-pipeline/.env") + REQUIRED_CONFIGS+=( + "$CONFIG_DIR/rag-pipeline/.env" + "$CONFIG_DIR/rag-pipeline/newrelic.ini" + ) fi for cfg in "${REQUIRED_CONFIGS[@]}"; do diff --git a/deployment/ci/server-init.sh b/deployment/ci/server-init.sh index 1e452c32..038b492a 100755 --- a/deployment/ci/server-init.sh +++ b/deployment/ci/server-init.sh @@ -83,26 +83,37 @@ SAMPLE fi done -# New Relic config for inference-orchestrator (Python agent) -NR_INI="$DEPLOY_DIR/config/inference-orchestrator/newrelic.ini" -if [ ! -f "$NR_INI" ]; then - cat > "$NR_INI" <<'SAMPLE' +# New Relic configs for Python services (different app_name per service) +for nr_svc in inference-orchestrator rag-pipeline; do + case "$nr_svc" in + inference-orchestrator) + NR_APP_NAME="CodeCrow Inference Orchestrator" + ;; + rag-pipeline) + NR_APP_NAME="CodeCrow RAG Pipeline" + ;; + esac + + NR_INI="$DEPLOY_DIR/config/$nr_svc/newrelic.ini" + if [ ! -f "$NR_INI" ]; then + cat > "$NR_INI" <=0.27.0,<1.0.0 # System monitoring psutil>=5.9.0,<6.0.0 +# Application performance monitoring +newrelic==11.5.0 + debugpy==1.8.1 diff --git a/python-ecosystem/rag-pipeline/requirements.txt b/python-ecosystem/rag-pipeline/requirements.txt index 817a8d5b..cecbc8f0 100644 --- a/python-ecosystem/rag-pipeline/requirements.txt +++ b/python-ecosystem/rag-pipeline/requirements.txt @@ -59,3 +59,6 @@ httpx>=0.27.0,<1.0.0 # System monitoring psutil>=5.9.0,<6.0.0 + +# Application performance monitoring +newrelic==11.5.0 diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/api/api.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/api/api.py index 4d4eb1b3..d3d3251a 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/api/api.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/api/api.py @@ -5,6 +5,7 @@ and includes all routers. This is the thin orchestration layer. """ import logging +import os from contextlib import asynccontextmanager from typing import Optional from fastapi import FastAPI @@ -81,6 +82,16 @@ async def lifespan(app: FastAPI): app.include_router(pr_router) app.include_router(inspect_router) +# Uvicorn loads this module by import string in every worker. Wrap the exported +# application here so each worker receives top-level ASGI instrumentation. +if os.environ.get("NEW_RELIC_CONFIG_FILE"): + try: + import newrelic.agent + app = newrelic.agent.ASGIApplicationWrapper(app) + logger.info("New Relic ASGI wrapper applied") + except Exception as exc: + logger.warning("New Relic ASGI wrapper failed: %s", exc) + if __name__ == "__main__": import uvicorn From 5088a7440c98411060bdd1f2e88795e75458bdeb Mon Sep 17 00:00:00 2001 From: rostislav Date: Wed, 29 Jul 2026 13:27:07 +0300 Subject: [PATCH 3/3] fix(rag): quarantine invalid files instead of failing repository indexing - isolate loader, parser, plugin, embedding, and point-write failures - continue indexing valid files and deterministic architecture context - treat recoverable plugin diagnostics as file-level skips - remove stale chunks when changed files become invalid - preserve atomic rollback for infrastructure and integrity failures - report skipped file and chunk counts - document indexing quarantine and failure semantics - add full, incremental, PR-overlay, plugin, and Qdrant coverage Deployment note: fully reindex existing branches because the neutral and plugin implementation fingerprints changed. --- README.md | 4 +- analysis-plugins/README.md | 6 +- .../contracts/python/codecrow_plugins/api.py | 14 ++ .../python/codecrow_plugins/runtime.py | 22 +- .../python/tests/test_builtin_plugins.py | 10 +- .../tests/test_magento_repository_analysis.py | 9 +- .../test_repository_runtime_resilience.py | 59 ++++++ .../codecrow_plugin_magento/architecture.py | 4 + .../codecrow_plugin_magento/repository.py | 23 +- deployment/.env.sample | 2 + .../config/inference-orchestrator/.env.sample | 1 + deployment/docker-compose.prod.yml | 19 +- deployment/docker-compose.yml | 19 +- java-ecosystem/libs/analysis-engine/pom.xml | 1 + .../aiclient/AiAnalysisClient.java | 103 +++++++-- .../aiclient/AiAnalysisClientTest.java | 66 +++++- .../codecrow/queue/RedisQueueService.java | 8 + .../codecrow/queue/RedisQueueServiceTest.java | 14 ++ .../IsolatedReviewProducerReplayTest.java | 10 + .../src/api/routers/health.py | 10 +- .../src/server/queue_consumer.py | 59 ++++++ .../tests/test_queue_consumer.py | 35 +++ .../rag-pipeline/integration/conftest.py | 9 + .../src/rag_pipeline/api/routers/pr.py | 45 ++-- .../core/index_manager/indexer.py | 199 ++++++++++++------ .../core/index_manager/point_operations.py | 128 +++++++++-- .../rag_pipeline/core/index_representation.py | 20 +- .../rag_pipeline/core/splitter/splitter.py | 25 +++ .../src/rag_pipeline/models/config.py | 2 + .../src/rag_pipeline/services/base.py | 30 +-- .../rag_pipeline/services/semantic_search.py | 2 +- .../tests/test_deterministic_context.py | 4 +- .../test_incremental_repository_overlay.py | 32 ++- .../tests/test_index_representation.py | 30 ++- .../rag-pipeline/tests/test_indexer.py | 98 ++++++--- .../tests/test_point_operations.py | 93 ++++++-- .../rag-pipeline/tests/test_router_pr.py | 26 ++- .../rag-pipeline/tests/test_services.py | 11 +- .../rag-pipeline/tests/test_splitter.py | 22 ++ 39 files changed, 1033 insertions(+), 241 deletions(-) create mode 100644 analysis-plugins/contracts/python/tests/test_repository_runtime_resilience.py diff --git a/README.md b/README.md index 6d204991..bf8f98fd 100644 --- a/README.md +++ b/README.md @@ -155,10 +155,10 @@ normal review pipeline continues. | Retrieval | Qdrant semantic search combined with deterministic metadata and exact-path lookup | | Stored Context | Semantic source chunks plus exact-source, architecture, graph, plugin snapshot, and repository-detection points | | Full Reindex | Builds a pending collection generation and atomically swaps the project alias only after successful completion | -| Resilient Writes | Retries transient Qdrant writes and subdivides rejected batches; an unrecoverable point keeps the pending generation inactive | +| Resilient Writes | Quarantines malformed files and exact rejected points while retaining valid content; systemic Qdrant failures still prevent activation | | Incremental Reindex | Applies one pinned commit change set and replaces changed semantic chunks together with affected graph and state groups | | PR Context | Uses an immutable, commit-pinned PR overlay so changed files do not retrieve stale base-branch copies | -| Compatibility Guard | Returns HTTP 409 for stale or incompatible representation/plugin state before a review-model call; recovery is a full reindex with matching service images | +| Compatibility Guard | Source-build fingerprints are provenance only and never force a reindex; durable plugin descriptor, snapshot-integrity, and Qdrant vector-shape checks remain enforced | | Prompt Budget | Plugin and RAG evidence share bounded context budgets; plugins cannot create an additional model stage | The Vector Storage Explorer exposes the different point types and their diff --git a/analysis-plugins/README.md b/analysis-plugins/README.md index f79f845a..6919a3cd 100644 --- a/analysis-plugins/README.md +++ b/analysis-plugins/README.md @@ -152,5 +152,7 @@ Magento classifies XML as module configuration only when it is directly below an Custom descendants such as `etc/samples/*.xml` stay `full` project content and are handled by the generic indexing path. DTDs and entities in those ordinary documents are not resolved by the Magento repository analyzer. Actual Magento -configuration XML remains architecture input and fails closed when it contains a -DTD or entity declaration. +configuration XML remains architecture input, but a malformed file or one with a +DTD/entity declaration is quarantined from the architecture snapshot instead of +failing the repository index. Other valid plugin inputs continue to contribute +deterministic context. Runtime/plugin contract failures remain fatal. diff --git a/analysis-plugins/contracts/python/codecrow_plugins/api.py b/analysis-plugins/contracts/python/codecrow_plugins/api.py index df30a0e6..44a8064d 100644 --- a/analysis-plugins/contracts/python/codecrow_plugins/api.py +++ b/analysis-plugins/contracts/python/codecrow_plugins/api.py @@ -225,12 +225,18 @@ class PluginDiagnostic: code: str message: str plugin_id: str | None = None + path: str | None = None + recoverable: bool = False def __post_init__(self) -> None: _non_blank(self.code, "diagnostic code") _non_blank(self.message, "diagnostic message") if self.plugin_id is not None: _plugin_id(self.plugin_id) + if self.path is not None: + normalized = normalize_path(self.path) + if normalized != self.path: + raise ValueError("diagnostic path must already be normalized") T = TypeVar("T") @@ -537,12 +543,20 @@ class RepositoryAnalysis: packets: tuple[ArchitecturePacket, ...] = () snapshots: tuple[RepositorySnapshot, ...] = () contexts: tuple[RepositoryContext, ...] = () + diagnostics: tuple[PluginDiagnostic, ...] = () def __post_init__(self) -> None: _sorted_unique(self.symbols, "repository symbols") _sorted_unique(self.packets, "architecture packets") _sorted_unique(self.snapshots, "repository snapshots") _sorted_unique(self.contexts, "repository contexts") + if any( + not isinstance(diagnostic, PluginDiagnostic) + for diagnostic in self.diagnostics + ): + raise ValueError( + "repository diagnostics must contain PluginDiagnostic values" + ) @dataclass(frozen=True, order=True) diff --git a/analysis-plugins/contracts/python/codecrow_plugins/runtime.py b/analysis-plugins/contracts/python/codecrow_plugins/runtime.py index b63b3edc..885102ce 100644 --- a/analysis-plugins/contracts/python/codecrow_plugins/runtime.py +++ b/analysis-plugins/contracts/python/codecrow_plugins/runtime.py @@ -401,15 +401,18 @@ def ingest(self, artifacts: tuple[FileArtifact, ...]) -> None: raise ValueError("repository artifacts must be path-sorted") retained: list[tuple[str, object]] = [] for plugin_id, session in self._sessions: - try: - session.ingest(artifacts) - retained.append((plugin_id, session)) - except Exception as exception: - self._diagnostics.append(PluginDiagnostic( - code="plugin-repository-ingest-exception", - message=f"{type(exception).__name__}: {exception}", - plugin_id=plugin_id, - )) + for artifact in artifacts: + try: + session.ingest((artifact,)) + except Exception as exception: + self._diagnostics.append(PluginDiagnostic( + code="plugin-repository-file-skipped", + message=f"{type(exception).__name__}: {exception}", + plugin_id=plugin_id, + path=artifact.path, + recoverable=True, + )) + retained.append((plugin_id, session)) self._sessions = retained def finish(self) -> tuple[RepositoryAnalysis, tuple[PluginDiagnostic, ...]]: @@ -444,6 +447,7 @@ def finish(self) -> tuple[RepositoryAnalysis, tuple[PluginDiagnostic, ...]]: plugin_id=plugin_id, )) continue + self._diagnostics.extend(contribution.diagnostics) symbols.update(contribution.symbols) if len(symbols) > self._runtime.MAX_REPOSITORY_SYMBOLS: self._diagnostics.append(PluginDiagnostic( diff --git a/analysis-plugins/contracts/python/tests/test_builtin_plugins.py b/analysis-plugins/contracts/python/tests/test_builtin_plugins.py index 174db091..acda5000 100644 --- a/analysis-plugins/contracts/python/tests/test_builtin_plugins.py +++ b/analysis-plugins/contracts/python/tests/test_builtin_plugins.py @@ -662,7 +662,7 @@ def test_magento_emits_effective_di_from_repository_state_only(): ) -def test_magento_rejects_xml_entities_without_resolution(): +def test_magento_quarantines_unsafe_config_without_failing_repository_analysis(): catalog = PluginCatalog.discover(PLUGINS_ROOT) runtime = PluginRuntime(catalog) capabilities = ProjectSelector(catalog.registry).select(_facts()) @@ -685,8 +685,14 @@ def test_magento_rejects_xml_entities_without_resolution(): ), key=lambda value: value.path))) analysis, diagnostics = handle.finish() - assert analysis.packets == () + assert analysis.snapshots + assert all( + artifact.path not in packet.paths + for packet in analysis.packets + ) assert [diagnostic.code for diagnostic in diagnostics] == ["magento-unsafe-xml"] + assert diagnostics[0].recoverable is True + assert diagnostics[0].path == artifact.path def test_magento_keeps_custom_entity_bearing_xml_outside_architecture_analysis(): diff --git a/analysis-plugins/contracts/python/tests/test_magento_repository_analysis.py b/analysis-plugins/contracts/python/tests/test_magento_repository_analysis.py index da47d357..989831f0 100644 --- a/analysis-plugins/contracts/python/tests/test_magento_repository_analysis.py +++ b/analysis-plugins/contracts/python/tests/test_magento_repository_analysis.py @@ -2046,7 +2046,7 @@ def test_magento_di_emits_effective_virtual_and_php_inherited_object_arguments() } <= child_packet_paths -def test_magento_di_virtual_type_argument_cycle_fails_closed(): +def test_magento_di_virtual_type_argument_cycle_is_quarantined(): catalog = PluginCatalog.discover(PLUGINS_ROOT) plugin = catalog.implementation("magento") started = plugin.start_repository_analysis("virtual-cycle") @@ -2071,10 +2071,11 @@ def test_magento_di_virtual_type_argument_cycle_fails_closed(): outcome = started.value.finish(RepositoryAnalysis()) - assert outcome.status is OutcomeStatus.FAILED - assert outcome.diagnostic.code == ( + assert outcome.status is OutcomeStatus.HANDLED + assert [diagnostic.code for diagnostic in outcome.value.diagnostics] == [ "magento-di-argument-inheritance-cycle" - ) + ] + assert outcome.value.diagnostics[0].recoverable is True def test_magento_base_view_configuration_is_related_to_area_variant(): diff --git a/analysis-plugins/contracts/python/tests/test_repository_runtime_resilience.py b/analysis-plugins/contracts/python/tests/test_repository_runtime_resilience.py new file mode 100644 index 00000000..78b4f7ef --- /dev/null +++ b/analysis-plugins/contracts/python/tests/test_repository_runtime_resilience.py @@ -0,0 +1,59 @@ +from types import SimpleNamespace + +from codecrow_plugins import ( + FileArtifact, + PluginDiagnostic, + PluginOutcome, + RepositoryAnalysis, +) +from codecrow_plugins.runtime import RepositoryAnalysisHandle + + +class _FileIsolatingSession: + def __init__(self): + self.ingested = [] + + def ingest(self, artifacts): + artifact = artifacts[0] + if artifact.path == "bad.xml": + raise ValueError("invalid project file") + self.ingested.append(artifact.path) + + def finish(self, _dependencies): + return PluginOutcome.handled(RepositoryAnalysis( + diagnostics=(PluginDiagnostic( + code="project-warning", + message="recoverable repository diagnostic", + plugin_id="test-plugin", + path="warning.xml", + recoverable=True, + ),), + )) + + +def test_repository_runtime_quarantines_ingest_failure_and_keeps_session(): + session = _FileIsolatingSession() + runtime = SimpleNamespace( + MAX_REPOSITORY_SYMBOLS=10, + MAX_ARCHITECTURE_PACKETS=10, + ) + handle = RepositoryAnalysisHandle( + runtime, + [("test-plugin", session)], + [], + ) + + handle.ingest(( + FileArtifact("bad.xml", ""), + FileArtifact("good.xml", ""), + )) + _analysis, diagnostics = handle.finish() + + assert session.ingested == ["good.xml"] + assert [ + (diagnostic.code, diagnostic.path, diagnostic.recoverable) + for diagnostic in diagnostics + ] == [ + ("plugin-repository-file-skipped", "bad.xml", True), + ("project-warning", "warning.xml", True), + ] diff --git a/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/architecture.py b/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/architecture.py index dd72f78e..e6cd10c8 100644 --- a/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/architecture.py +++ b/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/architecture.py @@ -58,6 +58,8 @@ def safe_xml(plugin_id: str, path: str, content: str): "DTD/entity declarations are forbidden in Magento " f"configuration {path}", plugin_id, + path=path, + recoverable=True, ) try: return ET.fromstring(content), None @@ -66,6 +68,8 @@ def safe_xml(plugin_id: str, path: str, content: str): "magento-invalid-xml", f"Cannot parse {path}: {exception}", plugin_id, + path=path, + recoverable=True, ) diff --git a/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/repository.py b/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/repository.py index edc7d244..ec6c694a 100644 --- a/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/repository.py +++ b/analysis-plugins/frameworks/magento/python/codecrow_plugin_magento/repository.py @@ -210,6 +210,7 @@ def __init__( self.graph = PacketGraph(plugin_id) self._roots: dict[str, object] = {} self._diagnostics: list[PluginDiagnostic] = [] + self.invalid_paths: set[str] = set() self._effective_di_arguments: dict[str, dict[str, dict]] = {} self._template_layout_sources: dict[ str, @@ -302,6 +303,7 @@ def _xml(self, path: str): root, diagnostic = safe_xml(self.plugin_id, path, self.artifacts[path]) if diagnostic: self._diagnostics.append(diagnostic) + self.invalid_paths.add(path) self._roots[path] = root return root @@ -6961,7 +6963,16 @@ def finish(self, dependencies: RepositoryAnalysis): resolver = MagentoRepositoryResolver(self.plugin_id, self.artifacts, dependencies.symbols) analysis, diagnostics = resolver.resolve() if diagnostics: - return PluginOutcome.failed(diagnostics[0]) + for diagnostic in diagnostics: + logger.warning( + "Skipping invalid Magento repository input " + "(code=%s path=%s): %s", + diagnostic.code, + diagnostic.path or "", + diagnostic.message, + ) + for path in resolver.invalid_paths: + self.artifacts.pop(path, None) related_paths = { path for packet in analysis.packets for path in packet.paths } @@ -6997,4 +7008,14 @@ def finish(self, dependencies: RepositoryAnalysis): packets=analysis.packets, snapshots=(snapshot,), contexts=contexts, + diagnostics=tuple( + PluginDiagnostic( + code=diagnostic.code, + message=diagnostic.message, + plugin_id=diagnostic.plugin_id, + path=diagnostic.path, + recoverable=True, + ) + for diagnostic in diagnostics + ), )) diff --git a/deployment/.env.sample b/deployment/.env.sample index c761514a..890cd26a 100644 --- a/deployment/.env.sample +++ b/deployment/.env.sample @@ -35,7 +35,9 @@ PR_ENRICHMENT_MAX_TOTAL_SIZE_BYTES=20971520 # Review workers emit liveness while a normal Stage 0-3 job is active. The Java # producer fails only after this much silence; total healthy runtime is unbounded. # ANALYSIS_QUEUE_HEARTBEAT_SECONDS=30 +# ANALYSIS_QUEUE_ADMISSION_TIMEOUT_MINUTES=5 # ANALYSIS_QUEUE_INACTIVITY_TIMEOUT_MINUTES=15 +# ANALYSIS_CONSUMER_HEARTBEAT_SECONDS=5 # Opt-in evidence capture for real BYOK review calls. Unlike prompt dry-run, # this invokes the configured provider and stores proprietary source/prompts and diff --git a/deployment/config/inference-orchestrator/.env.sample b/deployment/config/inference-orchestrator/.env.sample index 1df8e46d..6f4524f5 100644 --- a/deployment/config/inference-orchestrator/.env.sample +++ b/deployment/config/inference-orchestrator/.env.sample @@ -13,6 +13,7 @@ SERVICE_SECRET=change-me-to-a-random-secret # REDIS_URL=redis://localhost:6379/1 # MAX_CONCURRENT_REVIEWS=4 # ANALYSIS_QUEUE_HEARTBEAT_SECONDS=30 +# ANALYSIS_CONSUMER_HEARTBEAT_SECONDS=5 # MAX_CONCURRENT_COMMANDS=10 # COMMAND_TIMEOUT_SECONDS=600 # REVIEW_TIMEOUT_SECONDS=1500 diff --git a/deployment/docker-compose.prod.yml b/deployment/docker-compose.prod.yml index 566c9c8b..88d8f9d7 100644 --- a/deployment/docker-compose.prod.yml +++ b/deployment/docker-compose.prod.yml @@ -167,6 +167,7 @@ services: PR_ENRICHMENT_MAX_FILE_SIZE_BYTES: ${PR_ENRICHMENT_MAX_FILE_SIZE_BYTES:-5242880} PR_ENRICHMENT_MAX_TOTAL_SIZE_BYTES: ${PR_ENRICHMENT_MAX_TOTAL_SIZE_BYTES:-20971520} ANALYSIS_QUEUE_INACTIVITY_TIMEOUT_MINUTES: ${ANALYSIS_QUEUE_INACTIVITY_TIMEOUT_MINUTES:-15} + ANALYSIS_QUEUE_ADMISSION_TIMEOUT_MINUTES: ${ANALYSIS_QUEUE_ADMISSION_TIMEOUT_MINUTES:-5} ANALYSIS_PROMPT_DRY_RUN_ENABLED: ${ANALYSIS_PROMPT_DRY_RUN_ENABLED:-false} ANALYSIS_PROMPT_DRY_RUN_PROJECT_IDS: ${ANALYSIS_PROMPT_DRY_RUN_PROJECT_IDS:-} LOGGING_FILE_NAME: /app/logs/codecrow-pipeline-agent.log @@ -177,6 +178,8 @@ services: condition: service_healthy redis: condition: service_healthy + inference-orchestrator: + condition: service_healthy fix-permissions: condition: service_completed_successfully networks: @@ -203,9 +206,12 @@ services: ports: - "127.0.0.1:${INFERENCE_ORCHESTRATOR_HOST_PORT:-8000}:8000" depends_on: - - rag-pipeline - - web-server - - redis + rag-pipeline: + condition: service_healthy + web-server: + condition: service_healthy + redis: + condition: service_healthy environment: # API access for Platform MCP (internal network only) CODECROW_API_URL: http://codecrow-web-application:8081 @@ -214,6 +220,7 @@ services: INTERNAL_API_SECRET: ${INTERNAL_API_SECRET:?INTERNAL_API_SECRET must be set in .env} SERVICE_SECRET: ${INTERNAL_API_SECRET:?INTERNAL_API_SECRET must be set in .env} ANALYSIS_QUEUE_HEARTBEAT_SECONDS: ${ANALYSIS_QUEUE_HEARTBEAT_SECONDS:-30} + ANALYSIS_CONSUMER_HEARTBEAT_SECONDS: ${ANALYSIS_CONSUMER_HEARTBEAT_SECONDS:-5} ANALYSIS_PROMPT_DRY_RUN_ENABLED: ${ANALYSIS_PROMPT_DRY_RUN_ENABLED:-false} ANALYSIS_PROMPT_DRY_RUN_SYNTHETIC_FINDINGS_PER_FILE: ${ANALYSIS_PROMPT_DRY_RUN_SYNTHETIC_FINDINGS_PER_FILE:-6} ANALYSIS_PROMPT_DRY_RUN_SYNTHETIC_FINDINGS_MAX_TOTAL: ${ANALYSIS_PROMPT_DRY_RUN_SYNTHETIC_FINDINGS_MAX_TOTAL:-24} @@ -232,6 +239,12 @@ services: - ./config/inference-orchestrator/.env:/app/.env - ./config/inference-orchestrator/newrelic.ini:/app/newrelic.ini:ro - inference_orchestrator_logs:/app/logs + healthcheck: + test: ["CMD", "curl", "-f", "http://127.0.0.1:8000/health"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 60s restart: unless-stopped extra_hosts: - "host.docker.internal:host-gateway" diff --git a/deployment/docker-compose.yml b/deployment/docker-compose.yml index f9b9ecaa..1ada0e0b 100644 --- a/deployment/docker-compose.yml +++ b/deployment/docker-compose.yml @@ -173,6 +173,7 @@ services: PR_ENRICHMENT_MAX_FILE_SIZE_BYTES: ${PR_ENRICHMENT_MAX_FILE_SIZE_BYTES:-5242880} PR_ENRICHMENT_MAX_TOTAL_SIZE_BYTES: ${PR_ENRICHMENT_MAX_TOTAL_SIZE_BYTES:-20971520} ANALYSIS_QUEUE_INACTIVITY_TIMEOUT_MINUTES: ${ANALYSIS_QUEUE_INACTIVITY_TIMEOUT_MINUTES:-15} + ANALYSIS_QUEUE_ADMISSION_TIMEOUT_MINUTES: ${ANALYSIS_QUEUE_ADMISSION_TIMEOUT_MINUTES:-5} ANALYSIS_PROMPT_DRY_RUN_ENABLED: ${ANALYSIS_PROMPT_DRY_RUN_ENABLED:-false} ANALYSIS_PROMPT_DRY_RUN_PROJECT_IDS: ${ANALYSIS_PROMPT_DRY_RUN_PROJECT_IDS:-} #JAVA_OPTS: "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5006" @@ -185,6 +186,8 @@ services: condition: service_healthy redis: condition: service_healthy + inference-orchestrator: + condition: service_healthy fix-permissions: condition: service_completed_successfully networks: @@ -212,15 +215,19 @@ services: ports: - "127.0.0.1:${INFERENCE_ORCHESTRATOR_HOST_PORT:-8015}:8000" depends_on: - - rag-pipeline - - web-server - - redis + rag-pipeline: + condition: service_healthy + web-server: + condition: service_healthy + redis: + condition: service_healthy environment: CODECROW_API_URL: http://codecrow-web-application:8081 PLATFORM_MCP_JAR: /app/codecrow-platform-mcp-1.0.jar INTERNAL_API_SECRET: ${INTERNAL_API_SECRET:?INTERNAL_API_SECRET must be set in .env} SERVICE_SECRET: ${INTERNAL_API_SECRET:?INTERNAL_API_SECRET must be set in .env} ANALYSIS_QUEUE_HEARTBEAT_SECONDS: ${ANALYSIS_QUEUE_HEARTBEAT_SECONDS:-30} + ANALYSIS_CONSUMER_HEARTBEAT_SECONDS: ${ANALYSIS_CONSUMER_HEARTBEAT_SECONDS:-5} ANALYSIS_PROMPT_DRY_RUN_ENABLED: ${ANALYSIS_PROMPT_DRY_RUN_ENABLED:-false} ANALYSIS_PROMPT_DRY_RUN_SYNTHETIC_FINDINGS_PER_FILE: ${ANALYSIS_PROMPT_DRY_RUN_SYNTHETIC_FINDINGS_PER_FILE:-6} ANALYSIS_PROMPT_DRY_RUN_SYNTHETIC_FINDINGS_MAX_TOTAL: ${ANALYSIS_PROMPT_DRY_RUN_SYNTHETIC_FINDINGS_MAX_TOTAL:-24} @@ -236,6 +243,12 @@ services: volumes: - ./config/inference-orchestrator/.env:/app/.env - inference_orchestrator_logs:/app/logs + healthcheck: + test: ["CMD", "curl", "-f", "http://127.0.0.1:8000/health"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 60s restart: unless-stopped extra_hosts: - "host.docker.internal:host-gateway" diff --git a/java-ecosystem/libs/analysis-engine/pom.xml b/java-ecosystem/libs/analysis-engine/pom.xml index fb7b342b..ab125b6d 100644 --- a/java-ecosystem/libs/analysis-engine/pom.xml +++ b/java-ecosystem/libs/analysis-engine/pom.xml @@ -169,6 +169,7 @@ @{argLine} + -javaagent:${settings.localRepository}/net/bytebuddy/byte-buddy-agent/${byte-buddy.version}/byte-buddy-agent-${byte-buddy.version}.jar --add-opens org.rostilos.codecrow.analysisengine/org.rostilos.codecrow.analysisengine.service=com.fasterxml.jackson.databind --add-opens org.rostilos.codecrow.core/org.rostilos.codecrow.core.model.branch=org.rostilos.codecrow.analysisengine --add-opens org.rostilos.codecrow.core/org.rostilos.codecrow.core.model.project=org.rostilos.codecrow.analysisengine diff --git a/java-ecosystem/libs/analysis-engine/src/main/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClient.java b/java-ecosystem/libs/analysis-engine/src/main/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClient.java index 30789ca3..c7a10c7d 100644 --- a/java-ecosystem/libs/analysis-engine/src/main/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClient.java +++ b/java-ecosystem/libs/analysis-engine/src/main/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClient.java @@ -37,16 +37,27 @@ public class AiAnalysisClient { static final String INACTIVITY_TIMEOUT_MINUTES_KEY = "ANALYSIS_QUEUE_INACTIVITY_TIMEOUT_MINUTES"; + static final String ADMISSION_TIMEOUT_MINUTES_KEY = + "ANALYSIS_QUEUE_ADMISSION_TIMEOUT_MINUTES"; + static final String CONSUMER_HEARTBEAT_KEY = + "codecrow:analysis:consumer:heartbeat"; private static final long DEFAULT_INACTIVITY_TIMEOUT_MINUTES = 15L; + private static final long DEFAULT_ADMISSION_TIMEOUT_MINUTES = 5L; private final long inactivityTimeoutMillis; + private final long admissionTimeoutMillis; @Autowired public AiAnalysisClient( @Qualifier("aiRestTemplate") RestTemplate restTemplate, RedisQueueService queueService, ObjectMapper objectMapper) { - this(restTemplate, queueService, objectMapper, resolveInactivityTimeoutMillis()); + this( + restTemplate, + queueService, + objectMapper, + resolveInactivityTimeoutMillis(), + resolveAdmissionTimeoutMillis()); } AiAnalysisClient( @@ -54,6 +65,20 @@ public AiAnalysisClient( RedisQueueService queueService, ObjectMapper objectMapper, long inactivityTimeoutMillis) { + this( + restTemplate, + queueService, + objectMapper, + inactivityTimeoutMillis, + resolveAdmissionTimeoutMillis()); + } + + AiAnalysisClient( + RestTemplate restTemplate, + RedisQueueService queueService, + ObjectMapper objectMapper, + long inactivityTimeoutMillis, + long admissionTimeoutMillis) { // restTemplate kept in constructor for backward compatibility but no longer // used this.queueService = queueService; @@ -61,7 +86,11 @@ public AiAnalysisClient( if (inactivityTimeoutMillis <= 0) { throw new IllegalArgumentException("Review queue inactivity timeout must be positive"); } + if (admissionTimeoutMillis <= 0) { + throw new IllegalArgumentException("Review queue admission timeout must be positive"); + } this.inactivityTimeoutMillis = inactivityTimeoutMillis; + this.admissionTimeoutMillis = admissionTimeoutMillis; } public Map performAnalysis(AiAnalysisRequest request) @@ -76,10 +105,16 @@ public Map performAnalysis(AiAnalysisRequest request, String jobId = UUID.randomUUID().toString(); String eventQueueKey = "codecrow:analysis:events:" + jobId; String jobsQueueKey = "codecrow:analysis:jobs"; + String jsonPayload = null; try { log.info("Sending async analysis request to Redis queue (Job ID: {})", jobId); + if (!queueService.hasKey(CONSUMER_HEARTBEAT_KEY)) { + throw new IOException( + "Inference Orchestrator is unavailable: no live review queue consumer"); + } + // Wrap the request with the jobId boolean promptDryRun = PromptDryRunMode.isEnabledForProject(request.getProjectId()); Map requestPayload = buildSerializableRequestPayload(request); @@ -100,7 +135,7 @@ public Map performAnalysis(AiAnalysisRequest request, "job_id", jobId, "request", requestPayload); - String jsonPayload = objectMapper.writeValueAsString(jobPayload); + jsonPayload = objectMapper.writeValueAsString(jobPayload); // Push the job to the Redis queue queueService.leftPush(jobsQueueKey, jsonPayload); @@ -114,14 +149,30 @@ public Map performAnalysis(AiAnalysisRequest request, queueService.setExpiry(eventQueueKey, eventTtlMinutes); long lastActivityTime = System.currentTimeMillis(); + long queuedAt = lastActivityTime; + boolean workerAcknowledged = false; // Poll the event queue for progress or final result while (true) { - if (System.currentTimeMillis() - lastActivityTime > inactivityTimeoutMillis) { + long now = System.currentTimeMillis(); + if (!workerAcknowledged) { + if (!queueService.hasKey(CONSUMER_HEARTBEAT_KEY)) { + throw new IOException( + "Inference Orchestrator became unavailable before " + + "the review job was admitted"); + } + if (now - queuedAt > admissionTimeoutMillis) { + throw new IOException( + "Review was not admitted by Inference Orchestrator within " + + TimeUnit.MILLISECONDS.toSeconds(admissionTimeoutMillis) + + " seconds"); + } + } + if (now - lastActivityTime > inactivityTimeoutMillis) { if (queueService.listContains(jobsQueueKey, jsonPayload)) { - // Capacity-bound work is still durably owned by Redis. - // Treat that exact queue membership as liveness and - // forward it so the caller can renew its analysis lock. + // Queue membership is durable ownership, but only a fresh + // consumer heartbeat proves that capacity can eventually + // admit it. Admission itself remains time-bounded. lastActivityTime = System.currentTimeMillis(); forwardEvent(eventHandler, Map.of( "type", "status", @@ -140,6 +191,7 @@ public Map performAnalysis(AiAnalysisRequest request, continue; // Timeout on BRPOP, continue to check inactivity } lastActivityTime = System.currentTimeMillis(); + workerAcknowledged = true; try { Map event = objectMapper.readValue(eventJson, Map.class); @@ -184,10 +236,24 @@ public Map performAnalysis(AiAnalysisRequest request, log.error("Failed to communicate with AI async queue", e); throw new IOException("AI queue communication failed: " + e.getMessage(), e); } finally { + if (jsonPayload != null) { + try { + queueService.removeFromList(jobsQueueKey, jsonPayload); + } catch (Exception cleanupError) { + log.warn( + "Failed to remove analysis job {} from the pending queue: {}", + jobId, + cleanupError.getMessage()); + } + } try { // Clean up the event queue if we exit early or successfully queueService.deleteKey(eventQueueKey); - } catch (Exception ignored) { + } catch (Exception cleanupError) { + log.warn( + "Failed to delete analysis event queue {}: {}", + eventQueueKey, + cleanupError.getMessage()); } } } @@ -206,23 +272,34 @@ private void forwardEvent( } private static long resolveInactivityTimeoutMillis() { - String configured = System.getProperty(INACTIVITY_TIMEOUT_MINUTES_KEY); + return resolvePositiveMinutes( + INACTIVITY_TIMEOUT_MINUTES_KEY, + DEFAULT_INACTIVITY_TIMEOUT_MINUTES); + } + + private static long resolveAdmissionTimeoutMillis() { + return resolvePositiveMinutes( + ADMISSION_TIMEOUT_MINUTES_KEY, + DEFAULT_ADMISSION_TIMEOUT_MINUTES); + } + + private static long resolvePositiveMinutes(String key, long defaultMinutes) { + String configured = System.getProperty(key); if (configured == null || configured.isBlank()) { - configured = System.getenv(INACTIVITY_TIMEOUT_MINUTES_KEY); + configured = System.getenv(key); } if (configured == null || configured.isBlank()) { - return TimeUnit.MINUTES.toMillis(DEFAULT_INACTIVITY_TIMEOUT_MINUTES); + return TimeUnit.MINUTES.toMillis(defaultMinutes); } try { long minutes = Long.parseLong(configured.trim()); if (minutes <= 0) { - throw new IllegalArgumentException( - INACTIVITY_TIMEOUT_MINUTES_KEY + " must be a positive integer"); + throw new IllegalArgumentException(key + " must be a positive integer"); } return TimeUnit.MINUTES.toMillis(minutes); } catch (NumberFormatException error) { throw new IllegalArgumentException( - INACTIVITY_TIMEOUT_MINUTES_KEY + " must be a positive integer", + key + " must be a positive integer", error); } } diff --git a/java-ecosystem/libs/analysis-engine/src/test/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClientTest.java b/java-ecosystem/libs/analysis-engine/src/test/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClientTest.java index 822e2276..209f5a46 100644 --- a/java-ecosystem/libs/analysis-engine/src/test/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClientTest.java +++ b/java-ecosystem/libs/analysis-engine/src/test/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClientTest.java @@ -207,6 +207,8 @@ public Map getReconciliationFileContents() { void setUp() throws Exception { objectMapper = new ObjectMapper(); client = new AiAnalysisClient(restTemplate, queueService, objectMapper); + lenient().when(queueService.hasKey(AiAnalysisClient.CONSUMER_HEARTBEAT_KEY)) + .thenReturn(true); System.setProperty(PromptDryRunMode.ENABLED_KEY, "false"); System.clearProperty(PromptDryRunMode.PROJECT_IDS_KEY); } @@ -555,8 +557,8 @@ void shouldAllowLongReviewWhileWorkerRemainsActive() throws Exception { } @Test - @DisplayName("should keep waiting while the exact job remains durably queued") - void shouldTreatExactQueueMembershipAsAdmissionLiveness() throws Exception { + @DisplayName("should report durable queue wait while a live consumer can still admit it") + void shouldReportDurableQueueWaitWithLiveConsumer() throws Exception { AiAnalysisClient shortInactivityClient = new AiAnalysisClient( restTemplate, queueService, objectMapper, 20L); String terminal = objectMapper.writeValueAsString(Map.of( @@ -596,6 +598,66 @@ void shouldTreatExactQueueMembershipAsAdmissionLiveness() throws Exception { @DisplayName("performAnalysis() error paths") class PerformAnalysisErrorTests { + @Test + @DisplayName("should fail before enqueue when no review consumer is alive") + void shouldFailWhenInferenceOrchestratorIsUnavailable() { + when(queueService.hasKey(AiAnalysisClient.CONSUMER_HEARTBEAT_KEY)) + .thenReturn(false); + + assertThatThrownBy(() -> client.performAnalysis(mockRequest)) + .isInstanceOf(IOException.class) + .hasMessageContaining("no live review queue consumer"); + + verify(queueService, never()).leftPush(anyString(), anyString()); + } + + @Test + @DisplayName("should fail queued work when the review consumer heartbeat expires") + void shouldFailWhenConsumerDisappearsBeforeAdmission() { + AiAnalysisClient shortAdmissionClient = new AiAnalysisClient( + restTemplate, + queueService, + objectMapper, + 1_000L, + 1_000L); + when(queueService.hasKey(AiAnalysisClient.CONSUMER_HEARTBEAT_KEY)) + .thenReturn(true, true, false); + when(queueService.rightPop(anyString(), anyLong())) + .thenReturn(null); + + assertThatThrownBy(() -> + shortAdmissionClient.performAnalysis(mockRequest)) + .isInstanceOf(IOException.class) + .hasMessageContaining( + "became unavailable before the review job was admitted"); + + verify(queueService).removeFromList( + eq("codecrow:analysis:jobs"), + anyString()); + } + + @Test + @DisplayName("should fail work that exceeds the bounded admission wait") + void shouldFailWhenWorkerCapacityDoesNotAdmitJobInTime() { + AiAnalysisClient shortAdmissionClient = new AiAnalysisClient( + restTemplate, + queueService, + objectMapper, + 1_000L, + 10L); + when(queueService.rightPop(anyString(), anyLong())) + .thenAnswer(invocation -> { + Thread.sleep(12L); + return null; + }); + + assertThatThrownBy(() -> + shortAdmissionClient.performAnalysis(mockRequest)) + .isInstanceOf(IOException.class) + .hasMessageContaining( + "was not admitted by Inference Orchestrator"); + } + @Test @DisplayName("should throw IOException when AI service returns error event") void shouldThrowWhenErrorEventReceived() throws Exception { diff --git a/java-ecosystem/libs/queue/src/main/java/org/rostilos/codecrow/queue/RedisQueueService.java b/java-ecosystem/libs/queue/src/main/java/org/rostilos/codecrow/queue/RedisQueueService.java index 3e1075f0..4019fea6 100644 --- a/java-ecosystem/libs/queue/src/main/java/org/rostilos/codecrow/queue/RedisQueueService.java +++ b/java-ecosystem/libs/queue/src/main/java/org/rostilos/codecrow/queue/RedisQueueService.java @@ -27,6 +27,14 @@ public boolean listContains(String queueKey, String payload) { return redisTemplate.opsForList().indexOf(queueKey, payload) != null; } + public void removeFromList(String queueKey, String payload) { + redisTemplate.opsForList().remove(queueKey, 1, payload); + } + + public boolean hasKey(String key) { + return Boolean.TRUE.equals(redisTemplate.hasKey(key)); + } + public void setExpiry(String key, long timeoutMinutes) { redisTemplate.expire(key, timeoutMinutes, TimeUnit.MINUTES); } diff --git a/java-ecosystem/libs/queue/src/test/java/org/rostilos/codecrow/queue/RedisQueueServiceTest.java b/java-ecosystem/libs/queue/src/test/java/org/rostilos/codecrow/queue/RedisQueueServiceTest.java index f571ff03..be2603c3 100644 --- a/java-ecosystem/libs/queue/src/test/java/org/rostilos/codecrow/queue/RedisQueueServiceTest.java +++ b/java-ecosystem/libs/queue/src/test/java/org/rostilos/codecrow/queue/RedisQueueServiceTest.java @@ -102,6 +102,20 @@ void listContains_shouldReturnFalseWhenPayloadIsNotQueued() { assertThat(service.listContains("q", "payload")).isFalse(); } + @Test + void removeFromList_shouldDeleteOnlyTheExactQueuedPayload() { + service.removeFromList("q", "payload"); + + verify(listOperations).remove("q", 1, "payload"); + } + + @Test + void hasKey_shouldRequireAnExistingHeartbeatKey() { + when(redisTemplate.hasKey("heartbeat")).thenReturn(true); + + assertThat(service.hasKey("heartbeat")).isTrue(); + } + // ── setExpiry ──────────────────────────────────────────────────────── @Test diff --git a/java-ecosystem/services/pipeline-agent/src/test/java/org/rostilos/codecrow/pipelineagent/generic/service/IsolatedReviewProducerReplayTest.java b/java-ecosystem/services/pipeline-agent/src/test/java/org/rostilos/codecrow/pipelineagent/generic/service/IsolatedReviewProducerReplayTest.java index 6a6aa89f..260796d0 100644 --- a/java-ecosystem/services/pipeline-agent/src/test/java/org/rostilos/codecrow/pipelineagent/generic/service/IsolatedReviewProducerReplayTest.java +++ b/java-ecosystem/services/pipeline-agent/src/test/java/org/rostilos/codecrow/pipelineagent/generic/service/IsolatedReviewProducerReplayTest.java @@ -405,6 +405,16 @@ public void setExpiry(String key, long timeoutMinutes) { // The disconnected producer has no Redis state to expire. } + @Override + public boolean hasKey(String key) { + return true; + } + + @Override + public void removeFromList(String queueKey, String payload) { + // The synthetic response consumes the envelope immediately. + } + @Override public void deleteKey(String key) { // The disconnected producer has no Redis state to delete. diff --git a/python-ecosystem/inference-orchestrator/src/api/routers/health.py b/python-ecosystem/inference-orchestrator/src/api/routers/health.py index 40d0da39..ce2435e4 100644 --- a/python-ecosystem/inference-orchestrator/src/api/routers/health.py +++ b/python-ecosystem/inference-orchestrator/src/api/routers/health.py @@ -1,12 +1,18 @@ """ Health check endpoints. """ -from fastapi import APIRouter +from fastapi import APIRouter, HTTPException, Request router = APIRouter(tags=["health"]) @router.get("/health") -async def health(): +async def health(request: Request): """Health check endpoint.""" + queue_consumer = getattr(request.app.state, "queue_consumer", None) + if queue_consumer is None or not await queue_consumer.is_healthy(): + raise HTTPException( + status_code=503, + detail="review queue consumer is unavailable", + ) return {"status": "ok"} diff --git a/python-ecosystem/inference-orchestrator/src/server/queue_consumer.py b/python-ecosystem/inference-orchestrator/src/server/queue_consumer.py index 2b986e16..92cd73f8 100644 --- a/python-ecosystem/inference-orchestrator/src/server/queue_consumer.py +++ b/python-ecosystem/inference-orchestrator/src/server/queue_consumer.py @@ -30,6 +30,19 @@ def __init__(self, review_service: ReviewService): self.is_running = False self._redis: Optional[redis.Redis] = None self._task: Optional[asyncio.Task] = None + self._consumer_heartbeat_task: Optional[asyncio.Task] = None + self.consumer_heartbeat_key = "codecrow:analysis:consumer:heartbeat" + self.consumer_heartbeat_seconds = max( + 1.0, + float(os.environ.get( + "ANALYSIS_CONSUMER_HEARTBEAT_SECONDS", + "5", + )), + ) + self.consumer_heartbeat_ttl_seconds = max( + 15, + int(self.consumer_heartbeat_seconds * 3), + ) # Bound concurrent job processing to prevent memory pressure max_concurrent = int(os.environ.get("MAX_CONCURRENT_REVIEWS", "4")) self._job_semaphore = asyncio.Semaphore(max_concurrent) @@ -52,6 +65,10 @@ async def start(self): health_check_interval=30, ) self.is_running = True + await self._publish_consumer_heartbeat() + self._consumer_heartbeat_task = asyncio.create_task( + self._consumer_heartbeat_loop() + ) self._task = asyncio.create_task(self._consume_loop()) async def stop(self): @@ -63,11 +80,53 @@ async def stop(self): await self._task except asyncio.CancelledError: pass + if self._consumer_heartbeat_task: + self._consumer_heartbeat_task.cancel() + try: + await self._consumer_heartbeat_task + except asyncio.CancelledError: + pass if self._redis: await self._redis.aclose() logger.info("Redis Queue Consumer stopped") + async def _publish_consumer_heartbeat(self): + if self._redis: + await self._redis.set( + self.consumer_heartbeat_key, + "alive", + ex=self.consumer_heartbeat_ttl_seconds, + ) + + async def _consumer_heartbeat_loop(self): + while self.is_running: + try: + await self._publish_consumer_heartbeat() + except asyncio.CancelledError: + raise + except Exception: + logger.exception("Failed to publish review consumer heartbeat") + await asyncio.sleep(self.consumer_heartbeat_seconds) + + async def is_healthy(self) -> bool: + if ( + not self.is_running + or self._redis is None + or self._task is None + or self._task.done() + or self._consumer_heartbeat_task is None + or self._consumer_heartbeat_task.done() + ): + return False + try: + return bool(await asyncio.wait_for( + self._redis.exists(self.consumer_heartbeat_key), + timeout=2, + )) + except Exception: + return False + async def _consume_loop(self): """Infinite loop blocking on the Redis queue for new jobs.""" logger.info(f"Listening for jobs on '{self.job_queue_key}'...") diff --git a/python-ecosystem/inference-orchestrator/tests/test_queue_consumer.py b/python-ecosystem/inference-orchestrator/tests/test_queue_consumer.py index 970d0a54..da73898a 100644 --- a/python-ecosystem/inference-orchestrator/tests/test_queue_consumer.py +++ b/python-ecosystem/inference-orchestrator/tests/test_queue_consumer.py @@ -186,6 +186,7 @@ async def test_start_uses_blocking_read_safe_redis_timeouts(): review_service = MagicMock() redis_client = MagicMock() redis_client.aclose = AsyncMock() + redis_client.set = AsyncMock() consumer = RedisQueueConsumer(review_service) consumer._consume_loop = AsyncMock() @@ -203,6 +204,40 @@ async def test_start_uses_blocking_read_safe_redis_timeouts(): "socket_timeout": 30, "health_check_interval": 30, } + redis_client.set.assert_awaited() + + +@pytest.mark.asyncio(loop_scope="function") +async def test_review_consumer_heartbeat_has_a_short_expiry(): + consumer = RedisQueueConsumer(MagicMock()) + consumer._redis = MagicMock() + consumer._redis.set = AsyncMock() + consumer.consumer_heartbeat_ttl_seconds = 15 + + await consumer._publish_consumer_heartbeat() + + consumer._redis.set.assert_awaited_once_with( + "codecrow:analysis:consumer:heartbeat", + "alive", + ex=15, + ) + + +@pytest.mark.asyncio(loop_scope="function") +async def test_review_consumer_health_requires_live_tasks_and_heartbeat(): + consumer = RedisQueueConsumer(MagicMock()) + consumer.is_running = True + consumer._redis = MagicMock() + consumer._redis.exists = AsyncMock(return_value=1) + consumer._task = MagicMock() + consumer._task.done.return_value = False + consumer._consumer_heartbeat_task = MagicMock() + consumer._consumer_heartbeat_task.done.return_value = False + + assert await consumer.is_healthy() is True + + consumer._task.done.return_value = True + assert await consumer.is_healthy() is False @pytest.mark.asyncio(loop_scope="function") diff --git a/python-ecosystem/rag-pipeline/integration/conftest.py b/python-ecosystem/rag-pipeline/integration/conftest.py index ced4b60a..d980637e 100644 --- a/python-ecosystem/rag-pipeline/integration/conftest.py +++ b/python-ecosystem/rag-pipeline/integration/conftest.py @@ -78,6 +78,15 @@ def rag_app(_mock_qdrant, _mock_embedding): mock_im = MagicMock() mock_im.embed_model = _mock_embedding mock_im.qdrant_client = _mock_qdrant + mock_im.splitter.split_documents_resilient.side_effect = ( + lambda documents, capabilities=None: ( + mock_im.splitter.split_documents( + documents, + capabilities=capabilities, + ), + (), + ) + ) MockIM.return_value = mock_im mock_qs = MagicMock() diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/api/routers/pr.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/api/routers/pr.py index f97f5a27..f618c1de 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/api/routers/pr.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/api/routers/pr.py @@ -155,7 +155,7 @@ def index_pr_files(request: PRIndexRequest): stored_plugin_ids, stored_fingerprint, stored_descriptor_fingerprint, - stored_implementation_fingerprint, + _stored_implementation_fingerprint, ) = load_repository_snapshots( index_manager.qdrant_client, collection_name, @@ -195,22 +195,15 @@ def index_pr_files(request: PRIndexRequest): stored_plugin_ids ) ) - current_stored_implementation_fingerprint = ( - index_manager.plugin_catalog.implementation_fingerprint( - stored_plugin_ids - ) - ) if ( stored_descriptor_fingerprint != current_stored_descriptor_fingerprint - or stored_implementation_fingerprint - != current_stored_implementation_fingerprint ): raise HTTPException( status_code=409, detail=( f"target branch '{target_branch}' was indexed with " - "different or unknown plugin implementation content; " + "different or unknown plugin descriptors; " f"reindex target branch '{target_branch}' before review" ), ) @@ -484,10 +477,16 @@ def index_pr_files(request: PRIndexRequest): ) documents.append(doc) - chunks = index_manager.splitter.split_documents( - documents, - capabilities=capabilities, - ) if documents else [] + if documents: + chunks, split_skipped_paths = ( + index_manager.splitter.split_documents_resilient( + documents, + capabilities=capabilities, + ) + ) + else: + chunks = [] + split_skipped_paths = () # Add PR metadata to all chunks for chunk in chunks: @@ -553,12 +552,17 @@ def index_pr_files(request: PRIndexRequest): )) handle.ingest(artifacts) analysis, diagnostics = handle.finish() - if diagnostics: - summary = "; ".join( - f"{item.plugin_id or 'plugin'}:{item.code}: {item.message}" - for item in diagnostics[:10] + repository_skipped_paths = ( + index_manager._indexer + .accept_recoverable_repository_diagnostics( + diagnostics, + "PR repository overlay", ) - raise RuntimeError(f"PR repository overlay is incomplete: {summary}") + ) + split_skipped_paths = tuple(sorted({ + *split_skipped_paths, + *repository_skipped_paths, + })) changed_paths = { file_info.path for file_info in active_overlay_files @@ -621,6 +625,9 @@ def index_pr_files(request: PRIndexRequest): request.project, point_id_branch, ) + skipped_points = ( + len(chunks) + len(architecture_nodes) - successful + ) logger.info( "Indexed PR #%s: %s semantic/architecture points from %s changed files (%s architecture packets)", @@ -636,6 +643,8 @@ def index_pr_files(request: PRIndexRequest): "files_processed": len(request.files), "chunks_indexed": successful, "chunks_failed": 0, + "chunks_skipped": skipped_points, + "skipped_files": list(split_skipped_paths), "architecture_packets_indexed": len(architecture_nodes), "generation_fingerprint": generation_fingerprint, "overlay_representation_fingerprint": ( diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/indexer.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/indexer.py index 15f08ce7..38a29578 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/indexer.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/indexer.py @@ -94,6 +94,43 @@ def __init__( config ) + @staticmethod + def accept_recoverable_repository_diagnostics( + diagnostics, + phase: str, + ) -> set[str]: + """Log quarantined project inputs and reject only runtime-level faults.""" + recoverable = [ + diagnostic for diagnostic in diagnostics + if diagnostic.recoverable + ] + fatal = [ + diagnostic for diagnostic in diagnostics + if not diagnostic.recoverable + ] + for diagnostic in recoverable: + logger.warning( + "Skipping invalid repository input during %s " + "(plugin=%s code=%s path=%s): %s", + phase, + diagnostic.plugin_id or "plugin", + diagnostic.code, + diagnostic.path or "", + diagnostic.message, + ) + if fatal: + summary = "; ".join( + f"{diagnostic.plugin_id or 'plugin'}:{diagnostic.code}: " + f"{diagnostic.message}" + for diagnostic in fatal[:10] + ) + raise RuntimeError(f"{phase} failed: {summary}") + return { + diagnostic.path + for diagnostic in recoverable + if diagnostic.path is not None + } + @staticmethod def _architecture_nodes( analysis, @@ -576,7 +613,8 @@ def index_repository( document_count = 0 chunk_count = 0 successful_chunks = 0 - failed_chunks = 0 + skipped_chunk_count = 0 + skipped_file_paths: set[str] = set() preserved_point_count = 0 try: @@ -609,8 +647,21 @@ def index_repository( documents = self.loader.load_file_batch( file_batch, repo_path_obj, workspace, project, branch, commit, - strict=True, + strict=False, ) + loaded_paths = { + document.metadata["path"] for document in documents + } + missing_paths = { + clean_archive_path(Path(path).as_posix()) + for path in file_batch + } - loaded_paths + for path in sorted(missing_paths): + logger.warning( + "Skipping repository file that could not be loaded: %s", + path, + ) + skipped_file_paths.update(missing_paths) if not documents: continue @@ -634,14 +685,19 @@ def index_repository( for document in documents if document.metadata["path"] in semantic_paths ] - document_count += len(semantic_documents) if not semantic_documents: del documents continue - chunks = self.splitter.split_documents( + chunks, split_skipped_paths = ( + self.splitter.split_documents_resilient( semantic_documents, capabilities=capabilities, + ) + ) + skipped_file_paths.update(split_skipped_paths) + document_count += ( + len(semantic_documents) - len(split_skipped_paths) ) identity_metadata = _plugin_identity_metadata( capabilities, @@ -663,12 +719,13 @@ def index_repository( chunks, pending_collection_name, workspace, project, branch ) successful_chunks += success - failed_chunks += failed - + skipped_chunk_count += failed if failed: - raise RuntimeError( - f"Index write was partial in batch {batch_num}: " - f"{failed} of {batch_chunk_count} chunks failed" + logger.warning( + "Skipped %s rejected chunks in batch %s; " + "continuing repository indexing", + failed, + batch_num, ) logger.info( @@ -688,12 +745,12 @@ def index_repository( snapshot_nodes = [] if analysis_handle is not None: repository_analysis, diagnostics = analysis_handle.finish() - if diagnostics: - summary = "; ".join( - f"{diagnostic.plugin_id or 'plugin'}:{diagnostic.code}: {diagnostic.message}" - for diagnostic in diagnostics[:10] + skipped_file_paths.update( + self.accept_recoverable_repository_diagnostics( + diagnostics, + "repository architecture analysis", ) - raise RuntimeError(f"Repository architecture analysis is incomplete: {summary}") + ) architecture_nodes = self._architecture_nodes( repository_analysis, @@ -761,11 +818,12 @@ def index_repository( branch, ) successful_chunks += success - failed_chunks += failed + skipped_chunk_count += failed if failed: - raise RuntimeError( - "Architecture context index write was partial: " - f"{failed} of {architecture_count} chunks failed" + logger.warning( + "Skipped %s rejected deterministic context points; " + "continuing repository indexing", + failed, ) logger.info( "Indexed %s deterministic architecture packets, %s exact source parts, " @@ -778,21 +836,15 @@ def index_repository( logger.info( f"Streaming indexing complete: {document_count} files, " - f"{successful_chunks}/{chunk_count} chunks indexed ({failed_chunks} failed)" + f"{successful_chunks}/{chunk_count} chunks indexed " + f"({skipped_chunk_count} skipped across " + f"{len(skipped_file_paths)} files)" ) - if failed_chunks or successful_chunks != chunk_count: - raise RuntimeError( - "Index generation is incomplete: " - f"expected={chunk_count}, successful={successful_chunks}, failed={failed_chunks}" - ) - # Verify and perform atomic swap pending_info = self.point_ops.client.get_collection(pending_collection_name) actual_point_count = int(pending_info.points_count or 0) expected_point_count = preserved_point_count + successful_chunks - if actual_point_count == 0: - raise RuntimeError("Pending collection is empty after indexing") if actual_point_count != expected_point_count: raise RuntimeError( "Pending collection point count is incomplete: " @@ -818,7 +870,12 @@ def index_repository( try: self.stats_manager.store_metadata( - workspace, project, branch, commit, document_count, chunk_count + workspace, + project, + branch, + commit, + document_count, + successful_chunks, ) except Exception: self._rollback_atomic_swap(alias_name, old_target) @@ -839,6 +896,8 @@ def index_repository( namespace=namespace, document_count=document_count, chunk_count=successful_chunks, + skipped_file_count=len(skipped_file_paths), + skipped_chunk_count=skipped_chunk_count, last_updated=datetime.now(timezone.utc).isoformat(), workspace=workspace, project=project, @@ -873,6 +932,10 @@ def _rollback_atomic_swap(self, alias_name: str, old_target: Optional[str]) -> N class FileOperations: """Apply rollback-safe file and neutral repository-graph updates.""" + + accept_recoverable_repository_diagnostics = staticmethod( + RepositoryIndexer.accept_recoverable_repository_diagnostics + ) def __init__( self, @@ -1009,24 +1072,32 @@ def _replace_points( point.id for point in new_points if str(point.id) not in old_ids ] - successful, failed = self.point_ops.upsert_points( - collection_name, - new_points, - ) - if failed or successful != len(new_points): + try: + write_result = self.point_ops.upsert_points_detailed( + collection_name, + new_points, + ) + except Exception: self._restore_old_points( collection_name, old_points, new_only_ids, ) - raise RuntimeError( - "incremental index write was partial: " - f"expected={len(new_points)}, successful={successful}, failed={failed}" + raise + + skipped_ids = { + str(point.id) for point in write_result.skipped_points + } + if skipped_ids: + logger.warning( + "Quarantining %s rejected points during incremental replacement", + len(skipped_ids), ) + accepted_new_ids = new_ids - skipped_ids stale_ids = [ record.id for point_id, record in old_points.items() - if point_id not in new_ids + if point_id not in accepted_new_ids ] try: self._delete_point_ids(collection_name, stale_ids) @@ -1037,7 +1108,7 @@ def _replace_points( new_only_ids, ) raise - return successful + return write_result.successful def _apply_change_set( self, @@ -1123,7 +1194,7 @@ def _apply_change_set( plugin_ids, _fingerprint, stored_descriptor_fingerprint, - stored_implementation_fingerprint, + _stored_implementation_fingerprint, ) = load_repository_snapshots( self.client, collection_name, @@ -1184,11 +1255,9 @@ def _apply_change_set( ) if ( stored_descriptor_fingerprint != current_descriptor_fingerprint - or stored_implementation_fingerprint - != implementation_fingerprint ): raise IncrementalIndexPreconditionError( - "indexed repository plugin implementation is incompatible " + "indexed repository plugin descriptors are incompatible " f"with branch '{branch}'; reindex the branch before applying " "an incremental update" ) @@ -1242,14 +1311,16 @@ def _apply_change_set( documents_by_path = { document.metadata["path"]: document for document in documents } - missing = [ + missing = sorted( path for path in active_updated_paths if path not in documents_by_path - ] + ) if missing: - raise RuntimeError( - "cannot apply exact repository overlay; changed files were not loaded: " - + ", ".join(missing[:10]) + logger.warning( + "Quarantining %s changed repository files that could not " + "be loaded during incremental indexing: %s", + len(missing), + ", ".join(missing[:10]), ) artifacts = tuple(sorted(( *( @@ -1261,7 +1332,7 @@ def _apply_change_set( ), *( FileArtifact(path=path, content="", deleted=True) - for path in active_deleted_paths + for path in (*active_deleted_paths, *missing) ), ), key=lambda artifact: artifact.path)) handle = self.plugin_runtime.start_repository_analysis( @@ -1272,24 +1343,32 @@ def _apply_change_set( ) handle.ingest(artifacts) analysis, diagnostics = handle.finish() - if diagnostics: - summary = "; ".join( - f"{item.plugin_id or 'plugin'}:{item.code}: {item.message}" - for item in diagnostics[:10] - ) - raise RuntimeError( - f"incremental repository overlay is incomplete: {summary}" - ) + self.accept_recoverable_repository_diagnostics( + diagnostics, + "incremental repository overlay", + ) semantic_documents = [ document for document in documents if dispositions.get(document.metadata["path"], FileDisposition.FULL) is FileDisposition.FULL ] - semantic_nodes = self.splitter.split_documents( - semantic_documents, - capabilities=capabilities, - ) if semantic_documents else [] + if semantic_documents: + semantic_nodes, split_skipped_paths = ( + self.splitter.split_documents_resilient( + semantic_documents, + capabilities=capabilities, + ) + ) + if split_skipped_paths: + logger.warning( + "Quarantining %s changed files that failed semantic " + "parsing/enrichment: %s", + len(split_skipped_paths), + ", ".join(split_skipped_paths[:10]), + ) + else: + semantic_nodes = [] identity_metadata = _plugin_identity_metadata( capabilities, implementation_fingerprint, diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/point_operations.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/point_operations.py index 727d3f0c..e8d37baa 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/point_operations.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_manager/point_operations.py @@ -7,6 +7,7 @@ import logging import time import uuid +from dataclasses import dataclass from datetime import datetime, timezone from typing import List, Dict, Tuple @@ -17,6 +18,20 @@ logger = logging.getLogger(__name__) +class PointWriteInfrastructureError(RuntimeError): + """Raised when Qdrant is unavailable or rejects the index representation.""" + + +@dataclass(frozen=True) +class PointWriteResult: + successful: int = 0 + skipped_points: tuple[PointStruct, ...] = () + + @property + def failed(self) -> int: + return len(self.skipped_points) + + class PointOperations: """Handles point embedding and upsert operations.""" @@ -167,20 +182,29 @@ def upsert_points( Returns: Tuple of (successful_count, failed_count) """ + result = self.upsert_points_detailed(collection_name, points) + return result.successful, result.failed + + def upsert_points_detailed( + self, + collection_name: str, + points: List[PointStruct], + ) -> PointWriteResult: + """Upsert points and retain the identities of quarantined points.""" successful = 0 - failed = 0 - + skipped_points: list[PointStruct] = [] + for i in range(0, len(points), self.batch_size): batch = points[i:i + self.batch_size] - batch_successful, batch_failed = self._upsert_resilient( + batch_result = self._upsert_resilient( collection_name, batch, batch_offset=i, ) - successful += batch_successful - failed += batch_failed + successful += batch_result.successful + skipped_points.extend(batch_result.skipped_points) - return successful, failed + return PointWriteResult(successful, tuple(skipped_points)) @staticmethod def _status_code(exception: Exception) -> int | None: @@ -196,10 +220,23 @@ def _is_batch_shape_failure(cls, exception: Exception | None) -> bool: """Return whether a smaller request can isolate a rejected point.""" if exception is None: return False + message = str(exception).casefold() + if any( + marker in message + for marker in ( + "vector dimension", + "dimension mismatch", + "expected dimension", + "vector name", + "collection not found", + "doesn't exist", + "does not exist", + ) + ): + return False status_code = cls._status_code(exception) if status_code in {400, 413, 422}: return True - message = str(exception).casefold() return any( marker in message for marker in ( @@ -208,8 +245,6 @@ def _is_batch_shape_failure(cls, exception: Exception | None) -> bool: "request entity too large", "request too large", "validation error", - "vector dimension", - "wrong input", ) ) @@ -225,10 +260,10 @@ def _upsert_resilient( points: List[PointStruct], *, batch_offset: int, - ) -> tuple[int, int]: + ) -> PointWriteResult: """Write a batch with bounded retry and rejected-request subdivision.""" if not points: - return 0, 0 + return PointWriteResult() error = None for attempt in range(1, self.upsert_max_attempts + 1): @@ -237,7 +272,7 @@ def _upsert_resilient( collection_name=collection_name, points=points, ) - return len(points), 0 + return PointWriteResult(successful=len(points)) except Exception as exception: error = exception if self._is_batch_shape_failure(exception): @@ -251,7 +286,10 @@ def _upsert_resilient( len(points), exception, ) - return 0, len(points) + raise PointWriteInfrastructureError( + "Qdrant index storage is unavailable or incompatible " + f"after {self.upsert_max_attempts} attempts" + ) from exception delay = self.upsert_retry_base_seconds * (2 ** (attempt - 1)) logger.warning( "Qdrant upsert failed at point offset %s " @@ -268,12 +306,12 @@ def _upsert_resilient( if len(points) == 1: logger.error( - "Qdrant rejected one exact index point after bounded recovery " + "Skipping one exact index point rejected by Qdrant " "(%s): %s", self._point_label(points[0]), error, ) - return 0, 1 + return PointWriteResult(skipped_points=(points[0],)) midpoint = len(points) // 2 left = points[:midpoint] @@ -298,10 +336,57 @@ def _upsert_resilient( right, batch_offset=batch_offset + midpoint, ) - return ( - left_result[0] + right_result[0], - left_result[1] + right_result[1], + return PointWriteResult( + successful=left_result.successful + right_result.successful, + skipped_points=( + *left_result.skipped_points, + *right_result.skipped_points, + ), ) + + def _embed_resilient( + self, + chunk_data: List[Tuple[str, TextNode]], + ) -> tuple[List[PointStruct], int]: + """Embed a slice, isolating only provider-rejected input chunks.""" + if not chunk_data: + return [], 0 + try: + return self.embed_and_create_points(chunk_data), 0 + except MemoryError: + raise + except Exception as exception: + if not self._is_batch_shape_failure(exception): + raise + if len(chunk_data) == 1: + chunk = chunk_data[0][1] + logger.error( + "Skipping one embedding input rejected by the provider " + "(path=%s): %s", + chunk.metadata.get("path", ""), + exception, + ) + return [], 1 + + midpoint = len(chunk_data) // 2 + logger.warning( + "Embedding provider rejected a %s-chunk request; " + "isolating it into %s and %s chunks: %s", + len(chunk_data), + midpoint, + len(chunk_data) - midpoint, + exception, + ) + left_points, left_skipped = self._embed_resilient( + chunk_data[:midpoint] + ) + right_points, right_skipped = self._embed_resilient( + chunk_data[midpoint:] + ) + return ( + [*left_points, *right_points], + left_skipped + right_skipped, + ) def process_and_upsert_chunks( self, @@ -329,16 +414,17 @@ def process_and_upsert_chunks( # work count, but operators can now observe steady progress instead of # waiting for every chunk from a 50-file document batch to finish. for i in range(0, len(chunk_data), self.batch_size): - point_batch = self.embed_and_create_points( + point_batch, embedding_skipped = self._embed_resilient( chunk_data[i:i + self.batch_size] ) + failed += embedding_skipped + if not point_batch: + continue batch_successful, batch_failed = self.upsert_points( collection_name, point_batch, ) successful += batch_successful failed += batch_failed - if batch_failed: - break return successful, failed diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_representation.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_representation.py index 5ac74923..69850427 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_representation.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/index_representation.py @@ -4,6 +4,7 @@ import hashlib import json +import logging from importlib import metadata as importlib_metadata from pathlib import Path from typing import Mapping, Optional @@ -11,6 +12,8 @@ from qdrant_client.models import FieldCondition, Filter, MatchValue +logger = logging.getLogger(__name__) + INDEX_REPRESENTATION_PAYLOAD_KEY = "index_representation_fingerprint" # These inputs can change persistent target-branch point text, metadata, or @@ -211,7 +214,14 @@ def require_compatible_branch_representation( *, expected_fingerprint: Optional[str] = None, ) -> bool: - """Fail closed for a legacy or differently produced indexed branch.""" + """Return whether a branch exists without gating it on source-code hashes. + + The fingerprint remains stored as build provenance for diagnostics. It is + deliberately not a compatibility boundary: operational changes to the RAG + host must not force customers to rebuild otherwise usable embeddings. + Structural incompatibilities are enforced by Qdrant and by the persisted + plugin descriptor/snapshot contracts at the points where they are used. + """ exists, stored = read_branch_index_representation( client, collection_name, @@ -221,9 +231,9 @@ def require_compatible_branch_representation( return False expected = expected_fingerprint or index_representation_fingerprint() if stored != expected: - raise IndexCompatibilityError( - f"branch '{branch}' was indexed with different or unknown neutral " - "RAG representation content; fully reindex the branch before " - "retrieval or incremental updates" + logger.info( + "Branch '%s' has a different or legacy neutral RAG build " + "fingerprint; accepting the existing index without reindexing", + branch, ) return True diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/splitter/splitter.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/splitter/splitter.py index 316d7c26..e69495f9 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/core/splitter/splitter.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/core/splitter/splitter.py @@ -254,6 +254,31 @@ def split_documents( return all_nodes + def split_documents_resilient( + self, + documents: List[LlamaDocument], + capabilities: Any = None, + ) -> tuple[List[TextNode], tuple[str, ...]]: + """Split documents independently and quarantine file-local failures.""" + nodes: List[TextNode] = [] + skipped_paths: list[str] = [] + for document in documents: + path = document.metadata.get("path", "unknown") + try: + nodes.extend(self.split_documents( + [document], + capabilities=capabilities, + )) + except MemoryError: + raise + except Exception: + skipped_paths.append(path) + logger.exception( + "Skipping repository file that failed parsing/enrichment: %s", + path, + ) + return nodes, tuple(skipped_paths) + @staticmethod def _attach_syntax_metadata( nodes: List[TextNode], diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/models/config.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/models/config.py index cad2f995..e5b26dc3 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/models/config.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/models/config.py @@ -210,6 +210,8 @@ class IndexStats(BaseModel): namespace: str document_count: int chunk_count: int + skipped_file_count: int = 0 + skipped_chunk_count: int = 0 last_updated: str workspace: str project: str diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/services/base.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/services/base.py index 7f650e67..bd93a607 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/services/base.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/services/base.py @@ -16,7 +16,6 @@ from ..utils.utils import make_project_namespace from ..core.embedding_factory import create_embedding_model, get_embedding_model_info from ..core.index_representation import ( - INDEX_REPRESENTATION_PAYLOAD_KEY, index_representation_fingerprint, require_compatible_branch_representation, ) @@ -42,9 +41,7 @@ def __init__(self, config: RAGConfig, plugin_catalog=None): index_representation_fingerprint(config) ) self._compatible_branch_cache: set[tuple[str, str]] = set() - self._plugin_identity_cache: Dict[ - tuple[str, ...], tuple[str, str] - ] = {} + self._plugin_identity_cache: Dict[tuple[str, ...], str] = {} self.qdrant_client = QdrantClient( url=config.qdrant_url, api_key=config.qdrant_api_key or None, @@ -61,18 +58,7 @@ def __init__(self, config: RAGConfig, plugin_catalog=None): self._index_cache_lock = threading.Lock() def _plugin_identity_compatible(self, metadata: Dict[str, Any]) -> bool: - """Return whether indexed plugin code matches the running RAG build. - - Branches share one collection and a full reindex intentionally preserves - points from other branches. Build-content identity is therefore checked - per result rather than assumed from the active collection alias. - """ - if ( - self.plugin_catalog is not None - and metadata.get(INDEX_REPRESENTATION_PAYLOAD_KEY) - != self.index_representation_fingerprint - ): - return False + """Validate durable plugin descriptors, not deployment source hashes.""" if self.plugin_catalog is None: return True @@ -86,17 +72,15 @@ def _plugin_identity_compatible(self, metadata: Dict[str, Any]) -> bool: expected = self._plugin_identity_cache.get(normalized_ids) if expected is None: try: - expected = ( - self.plugin_catalog.registry.fingerprint_for(normalized_ids), - self.plugin_catalog.implementation_fingerprint(normalized_ids), + expected = self.plugin_catalog.registry.fingerprint_for( + normalized_ids ) except (KeyError, TypeError, ValueError): return False self._plugin_identity_cache[normalized_ids] = expected return ( - metadata.get("plugin_descriptor_fingerprint") == expected[0] - and metadata.get("plugin_implementation_fingerprint") == expected[1] + metadata.get("plugin_descriptor_fingerprint") == expected ) def _require_compatible_branches( @@ -104,7 +88,7 @@ def _require_compatible_branches( collection_name: str, branches: List[str], ) -> None: - """Preflight branch representation once so stale indexes never look empty.""" + """Confirm branch presence; build fingerprints are provenance only.""" for branch in dict.fromkeys(branches): cache_key = (collection_name, branch) if cache_key in self._compatible_branch_cache: @@ -128,7 +112,7 @@ def _filter_plugin_compatible_points(self, points: List[Any]) -> List[Any]: if omitted: logger.warning( "Discarded %d indexed point(s) with stale or unknown plugin " - "descriptor/build-content identity", + "descriptor identity", omitted, ) return compatible diff --git a/python-ecosystem/rag-pipeline/src/rag_pipeline/services/semantic_search.py b/python-ecosystem/rag-pipeline/src/rag_pipeline/services/semantic_search.py index 058dc2ce..0dd12f34 100644 --- a/python-ecosystem/rag-pipeline/src/rag_pipeline/services/semantic_search.py +++ b/python-ecosystem/rag-pipeline/src/rag_pipeline/services/semantic_search.py @@ -147,7 +147,7 @@ def semantic_search_multi_branch( if incompatible_count: logger.warning( "Semantic search discarded %d result(s) with stale or " - "unknown plugin descriptor/build-content identity", + "unknown plugin descriptor identity", incompatible_count, ) diff --git a/python-ecosystem/rag-pipeline/tests/test_deterministic_context.py b/python-ecosystem/rag-pipeline/tests/test_deterministic_context.py index dc0d6731..b98f4d4a 100644 --- a/python-ecosystem/rag-pipeline/tests/test_deterministic_context.py +++ b/python-ecosystem/rag-pipeline/tests/test_deterministic_context.py @@ -126,7 +126,7 @@ def test_single_branch_returns_all_non_pr(self): result = svc._apply_branch_priority([pt1, pt2], "main", ["main"], set()) assert len(result) == 2 - def test_stale_target_plugin_build_cannot_hide_fresh_base_context(self): + def test_older_target_plugin_build_remains_eligible(self): catalog = MagicMock() catalog.registry.fingerprint_for.return_value = "sha256:descriptor" catalog.implementation_fingerprint.return_value = "sha256:implementation" @@ -159,7 +159,7 @@ def test_stale_target_plugin_build_cannot_hide_fresh_base_context(self): set(), ) - assert [point.id for point in result] == ["base"] + assert [point.id for point in result] == ["target"] # ───────────────────────────────────────────────────────────── diff --git a/python-ecosystem/rag-pipeline/tests/test_incremental_repository_overlay.py b/python-ecosystem/rag-pipeline/tests/test_incremental_repository_overlay.py index bebdd471..8303dd66 100644 --- a/python-ecosystem/rag-pipeline/tests/test_incremental_repository_overlay.py +++ b/python-ecosystem/rag-pipeline/tests/test_incremental_repository_overlay.py @@ -31,6 +31,20 @@ def get_text_embedding_batch(self, texts): return [[1.0, 0.0, 0.0, 0.0] for _ in texts] +def _splitter_mock(): + splitter = MagicMock() + splitter.split_documents_resilient.side_effect = ( + lambda documents, capabilities=None: ( + splitter.split_documents( + documents, + capabilities=capabilities, + ), + (), + ) + ) + return splitter + + def _write_repository( root: Path, implementation: str, @@ -202,7 +216,7 @@ def test_incremental_di_change_restores_snapshot_and_replaces_effective_graph(tm excluded_patterns=(), max_file_size_bytes=1_000_000, )) - splitter = MagicMock() + splitter = _splitter_mock() stats = MagicMock() operations = FileOperations( client, @@ -353,7 +367,7 @@ def factory_facts(): excluded_patterns=(), max_file_size_bytes=1_000_000, )) - splitter = MagicMock() + splitter = _splitter_mock() splitter.split_documents.return_value = [] operations = FileOperations( client, @@ -497,7 +511,7 @@ def proxy_facts(): excluded_patterns=(), max_file_size_bytes=1_000_000, )) - splitter = MagicMock() + splitter = _splitter_mock() operations = FileOperations( client, point_ops, @@ -660,7 +674,9 @@ def test_incremental_update_rejects_missing_repository_analysis_snapshots( point_ops.prepare_chunks_for_embedding.assert_not_called() -def test_generic_change_set_updates_and_deletes_through_one_generation(tmp_path): +def test_generic_change_set_accepts_older_build_and_updates_one_generation( + tmp_path, +): catalog = discover_builtin_plugins() runtime = PluginRuntime(catalog) selector = ProjectSelector(catalog.registry) @@ -677,9 +693,7 @@ def test_generic_change_set_updates_and_deletes_through_one_generation(tmp_path) catalog.registry, ) capabilities = selector.select(facts) - implementation_fingerprint = catalog.implementation_fingerprint( - capabilities.repository_plugins - ) + implementation_fingerprint = "sha256:older-rag-build" client = QdrantClient(":memory:") collection = "repository" @@ -742,7 +756,7 @@ def test_generic_change_set_updates_and_deletes_through_one_generation(tmp_path) _write("docs/a.md", "new a") (tmp_path / "docs/b.md").unlink() - splitter = MagicMock() + splitter = _splitter_mock() splitter.split_documents.side_effect = lambda documents, capabilities: [ TextNode( text=document.text, @@ -1167,7 +1181,7 @@ def test_framework_change_set_updates_relation_and_deletes_related_source_togeth _write_repository(tmp_path, replacement) (tmp_path / deleted_path).unlink() - splitter = MagicMock() + splitter = _splitter_mock() operations = FileOperations( client, point_ops, diff --git a/python-ecosystem/rag-pipeline/tests/test_index_representation.py b/python-ecosystem/rag-pipeline/tests/test_index_representation.py index 93b2574d..3e5cc6f5 100644 --- a/python-ecosystem/rag-pipeline/tests/test_index_representation.py +++ b/python-ecosystem/rag-pipeline/tests/test_index_representation.py @@ -196,26 +196,38 @@ def test_branch_identity_distinguishes_absent_legacy_and_current_points(): "collection", "main", ) == (True, None) - with pytest.raises(IndexCompatibilityError, match="fully reindex"): - require_compatible_branch_representation( - client, - "collection", - "main", - expected_fingerprint="sha256:current", - ) + assert require_compatible_branch_representation( + client, + "collection", + "main", + expected_fingerprint="sha256:current", + ) is True client.scroll = lambda **_kwargs: ( [SimpleNamespace(payload={ - INDEX_REPRESENTATION_PAYLOAD_KEY: "sha256:current", + INDEX_REPRESENTATION_PAYLOAD_KEY: "sha256:older-build", })], None, ) - require_compatible_branch_representation( + assert require_compatible_branch_representation( client, "collection", "main", expected_fingerprint="sha256:current", + ) is True + + client.scroll = lambda **_kwargs: ( + [SimpleNamespace(payload={ + INDEX_REPRESENTATION_PAYLOAD_KEY: "sha256:current", + })], + None, ) + assert require_compatible_branch_representation( + client, + "collection", + "main", + expected_fingerprint="sha256:current", + ) is True def test_branch_identity_pages_past_pr_points_and_accepts_missing_pr_as_legacy(): diff --git a/python-ecosystem/rag-pipeline/tests/test_indexer.py b/python-ecosystem/rag-pipeline/tests/test_indexer.py index 6ab01c60..896973fa 100644 --- a/python-ecosystem/rag-pipeline/tests/test_indexer.py +++ b/python-ecosystem/rag-pipeline/tests/test_indexer.py @@ -20,6 +20,7 @@ RepositoryIndexer, FileOperations, DOCUMENT_BATCH_SIZE, INSERT_BATCH_SIZE, ) +from rag_pipeline.core.index_manager.point_operations import PointWriteResult from rag_pipeline.models.config import IndexStats @@ -46,6 +47,15 @@ def _mock_components(): point_ops.client = MagicMock() branch_mgr.get_branch_point_count.return_value = 1 branch_mgr.stream_copy_points_to_collection.return_value = 0 + splitter.split_documents_resilient.side_effect = ( + lambda documents, capabilities=None: ( + splitter.split_documents( + documents, + capabilities=capabilities, + ), + (), + ) + ) return coll_mgr, branch_mgr, point_ops, stats_mgr, splitter, loader @@ -406,7 +416,7 @@ def test_indexing_failure_cleans_up(self): coll_mgr.delete_collection.assert_called_with("pending") - def test_partial_vector_write_never_swaps_alias(self): + def test_rejected_vector_point_is_skipped_and_valid_index_is_published(self): config = _mock_config() coll_mgr, branch_mgr, point_ops, stats_mgr, splitter, loader = _mock_components() loader.iter_repository_files.return_value = iter(["a.php"]) @@ -415,6 +425,10 @@ def test_partial_vector_write_never_swaps_alias(self): ] splitter.split_documents.return_value = [MagicMock(), MagicMock()] point_ops.process_and_upsert_chunks.return_value = (1, 1) + point_ops.client.get_collection.return_value = SimpleNamespace( + points_count=1 + ) + branch_mgr.get_branch_point_count.return_value = 1 coll_mgr.create_pending_collection.return_value = "pending" coll_mgr.alias_exists.return_value = True coll_mgr.collection_exists.return_value = True @@ -422,17 +436,21 @@ def test_partial_vector_write_never_swaps_alias(self): indexer = RepositoryIndexer(config, coll_mgr, branch_mgr, point_ops, stats_mgr, splitter, loader) - with pytest.raises(RuntimeError, match="partial"): - indexer.index_repository("/repo", "ws", "proj", "main", "abc123", "alias1") + result = indexer.index_repository( + "/repo", "ws", "proj", "main", "abc123", "alias1" + ) - coll_mgr.atomic_alias_swap.assert_not_called() - stats_mgr.store_metadata.assert_not_called() + assert result.chunk_count == 1 + assert result.skipped_chunk_count == 1 + coll_mgr.atomic_alias_swap.assert_called_once() + stats_mgr.store_metadata.assert_called_once() def test_repository_architecture_is_streamed_and_indexed_as_context(self, tmp_path): from codecrow_plugins import ( ArchitecturePacket, FileDisposition, GraphFact, + PluginDiagnostic, RepositoryAnalysis, ) @@ -480,7 +498,13 @@ def test_repository_architecture_is_streamed_and_indexed_as_context(self, tmp_pa related_paths=("app/code/Acme/Module.php",), ),), ),)), - (), + (PluginDiagnostic( + code="magento-invalid-xml", + message="Cannot parse etc/invalid.xml", + plugin_id="magento", + path="etc/invalid.xml", + recoverable=True, + ),), ) runtime = MagicMock() runtime.file_disposition.return_value = FileDisposition.FULL @@ -513,6 +537,7 @@ def test_repository_architecture_is_streamed_and_indexed_as_context(self, tmp_pa "app/code/Acme/etc/di.xml", ] assert result.chunk_count == 3 + assert result.skipped_file_count == 1 # ───────────────────────────────────────────────────────────── @@ -722,6 +747,20 @@ def _make_file_ops(self): point_ops.upsert_points.side_effect = ( lambda _collection, points: (len(points), 0) ) + point_ops.upsert_points_detailed.side_effect = ( + lambda _collection, points: PointWriteResult( + successful=len(points) + ) + ) + splitter.split_documents_resilient.side_effect = ( + lambda documents, capabilities=None: ( + splitter.split_documents( + documents, + capabilities=capabilities, + ), + (), + ) + ) stats_mgr.get_project_stats.return_value = MagicMock(spec=IndexStats) return FileOperations(client, point_ops, coll_mgr, stats_mgr, splitter, loader) @@ -745,8 +784,8 @@ def test_update_files_success(self, tmp_path): collection_name="coll", ) - ops.splitter.split_documents.assert_called_once() - ops.point_ops.upsert_points.assert_called_once() + ops.splitter.split_documents_resilient.assert_called_once() + ops.point_ops.upsert_points_detailed.assert_called_once() ops.client.delete.assert_not_called() def test_update_files_no_documents(self, tmp_path): @@ -766,7 +805,7 @@ def test_update_files_no_documents(self, tmp_path): collection_name="coll", ) ops.splitter.split_documents.assert_not_called() - ops.point_ops.upsert_points.assert_called_once_with("coll", []) + ops.point_ops.upsert_points_detailed.assert_called_once_with("coll", []) def test_delete_files(self): ops = self._make_file_ops() @@ -778,10 +817,10 @@ def test_delete_files(self): branch="main", collection_name="coll", ) - ops.point_ops.upsert_points.assert_called_once_with("coll", []) + ops.point_ops.upsert_points_detailed.assert_called_once_with("coll", []) ops.client.delete.assert_not_called() - def test_partial_replacement_restores_previous_points(self): + def test_rejected_replacement_point_is_quarantined(self): from qdrant_client.models import PointStruct ops = self._make_file_ops() @@ -801,20 +840,27 @@ def test_partial_replacement_restores_previous_points(self): vector=[0.0, 1.0], payload={"path": "new.php", "branch": "main"}, )] - ops.point_ops.upsert_points.side_effect = None - ops.point_ops.upsert_points.return_value = (0, 1) - - with pytest.raises(RuntimeError, match="write was partial"): - ops._replace_points( - [node], - [old_record], - "coll", - "ws", - "project", - "main", - ) + ops.point_ops.upsert_points_detailed.side_effect = None + ops.point_ops.upsert_points_detailed.return_value = PointWriteResult( + skipped_points=( + PointStruct( + id=new_id, + vector=[0.0, 1.0], + payload={"path": "new.php", "branch": "main"}, + ), + ), + ) + + successful = ops._replace_points( + [node], + [old_record], + "coll", + "ws", + "project", + "main", + ) - restored = ops.client.upsert.call_args.kwargs["points"] - assert [str(point.id) for point in restored] == [old_id] + assert successful == 0 + ops.client.upsert.assert_not_called() deleted = ops.client.delete.call_args.kwargs["points_selector"] - assert [str(point_id) for point_id in deleted.points] == [new_id] + assert [str(point_id) for point_id in deleted.points] == [old_id] diff --git a/python-ecosystem/rag-pipeline/tests/test_point_operations.py b/python-ecosystem/rag-pipeline/tests/test_point_operations.py index 03b300af..d06e81b4 100644 --- a/python-ecosystem/rag-pipeline/tests/test_point_operations.py +++ b/python-ecosystem/rag-pipeline/tests/test_point_operations.py @@ -1,10 +1,14 @@ import uuid from unittest.mock import MagicMock +import pytest from llama_index.core.schema import TextNode from qdrant_client.models import PointStruct -from rag_pipeline.core.index_manager.point_operations import PointOperations +from rag_pipeline.core.index_manager.point_operations import ( + PointOperations, + PointWriteInfrastructureError, +) def test_architecture_context_uses_zero_vector_without_embedding_request(): @@ -89,7 +93,7 @@ def test_process_streams_embedding_and_pending_writes_in_point_batches(): ] -def test_process_stops_embedding_after_pending_write_failure(): +def test_process_fails_on_qdrant_wide_unavailability(): client = MagicMock() client.upsert.side_effect = RuntimeError("qdrant unavailable") embed_model = MagicMock() @@ -110,15 +114,15 @@ def test_process_stops_embedding_after_pending_write_failure(): for index in range(4) ] - successful, failed = operations.process_and_upsert_chunks( - chunks, - "pending", - "workspace", - "project", - "main", - ) + with pytest.raises(PointWriteInfrastructureError): + operations.process_and_upsert_chunks( + chunks, + "pending", + "workspace", + "project", + "main", + ) - assert (successful, failed) == (0, 2) embed_model.get_text_embedding_batch.assert_called_once() @@ -224,7 +228,72 @@ def test_upsert_stops_subdivision_when_qdrant_is_globally_unavailable(): upsert_retry_base_seconds=0, ) - successful, failed = operations.upsert_points("pending", _points(4)) + with pytest.raises(PointWriteInfrastructureError): + operations.upsert_points("pending", _points(4)) + + assert client.upsert.call_count == 2 + + +def test_vector_dimension_mismatch_is_systemic_not_a_skippable_point(): + class DimensionMismatch(RuntimeError): + status_code = 400 + + client = MagicMock() + client.upsert.side_effect = DimensionMismatch( + "vector dimension mismatch: expected dimension 4" + ) + operations = PointOperations( + client, + MagicMock(), + batch_size=4, + upsert_max_attempts=2, + upsert_retry_base_seconds=0, + ) + + with pytest.raises(PointWriteInfrastructureError): + operations.upsert_points("pending", _points(4)) - assert (successful, failed) == (0, 4) assert client.upsert.call_count == 2 + + +def test_process_isolates_provider_rejected_embedding_input(): + class InvalidEmbeddingInput(RuntimeError): + status_code = 400 + + client = MagicMock() + embed_model = MagicMock() + + def embed(texts): + if "invalid" in texts: + raise InvalidEmbeddingInput("bad request") + return [[0.1, 0.2, 0.3] for _ in texts] + + embed_model.get_text_embedding_batch.side_effect = embed + operations = PointOperations( + client, + embed_model, + embedding_dim=3, + batch_size=3, + upsert_retry_base_seconds=0, + ) + chunks = [ + TextNode(text="valid-one", metadata={"path": "one.php"}), + TextNode(text="invalid", metadata={"path": "bad.php"}), + TextNode(text="valid-two", metadata={"path": "two.php"}), + ] + + successful, failed = operations.process_and_upsert_chunks( + chunks, + "pending", + "workspace", + "project", + "main", + ) + + assert (successful, failed) == (2, 1) + written_paths = { + point.payload["path"] + for call in client.upsert.call_args_list + for point in call.kwargs["points"] + } + assert written_paths == {"one.php", "two.php"} diff --git a/python-ecosystem/rag-pipeline/tests/test_router_pr.py b/python-ecosystem/rag-pipeline/tests/test_router_pr.py index 9eedde48..729e254d 100644 --- a/python-ecosystem/rag-pipeline/tests/test_router_pr.py +++ b/python-ecosystem/rag-pipeline/tests/test_router_pr.py @@ -34,6 +34,15 @@ def _make_index_manager(): im._get_project_collection_name.return_value = "rag_ws__proj" im._collection_manager.collection_exists.return_value = True im.splitter.split_documents.return_value = [] + im.splitter.split_documents_resilient.side_effect = ( + lambda documents, capabilities=None: ( + im.splitter.split_documents( + documents, + capabilities=capabilities, + ), + (), + ) + ) im._point_ops.embed_and_create_points.return_value = [] im._point_ops.upsert_points.return_value = (0, 0) im._point_ops.process_and_upsert_chunks.return_value = (0, 0) @@ -723,12 +732,14 @@ def test_request_descriptor_identity_mismatch_is_rejected_before_mutation( assert "plugin descriptors do not match" in exc_info.value.detail im._file_ops._replace_points.assert_not_called() + @patch("rag_pipeline.api.routers.pr.build_overlay_capabilities") @patch("rag_pipeline.api.routers.pr.load_repository_snapshots") @patch("rag_pipeline.api.routers.pr._get_index_manager") - def test_stored_plugin_implementation_mismatch_requires_reindex( + def test_stored_plugin_implementation_mismatch_does_not_require_reindex( self, mock_get, mock_load_snapshots, + mock_build_capabilities, ): im = _make_index_manager() mock_get.return_value = im @@ -741,16 +752,15 @@ def test_stored_plugin_implementation_mismatch_requires_reindex( ) req = _request([]) req.repository_plugins = ["java"] + im.plugin_runtime.repository_analysis_plugins.return_value = () + mock_build_capabilities.return_value = _capabilities("java") from rag_pipeline.api.routers.pr import index_pr_files - with pytest.raises(HTTPException) as exc_info: - index_pr_files(req) + result = index_pr_files(req) - assert exc_info.value.status_code == 409 - assert "different or unknown plugin implementation" in exc_info.value.detail - assert "reindex target branch 'main'" in exc_info.value.detail - im._file_ops._replace_points.assert_not_called() + assert result["status"] == "indexed" + im._file_ops._replace_points.assert_called_once() @patch("rag_pipeline.api.routers.pr.load_repository_snapshots") @patch("rag_pipeline.api.routers.pr._get_index_manager") @@ -777,7 +787,7 @@ def test_legacy_index_without_plugin_content_identity_requires_reindex( index_pr_files(req) assert exc_info.value.status_code == 409 - assert "different or unknown plugin implementation" in exc_info.value.detail + assert "different or unknown plugin descriptors" in exc_info.value.detail im._file_ops._replace_points.assert_not_called() diff --git a/python-ecosystem/rag-pipeline/tests/test_services.py b/python-ecosystem/rag-pipeline/tests/test_services.py index ae9bf762..f4768369 100644 --- a/python-ecosystem/rag-pipeline/tests/test_services.py +++ b/python-ecosystem/rag-pipeline/tests/test_services.py @@ -100,7 +100,7 @@ def test_get_project_collection_name(self, MockQdrant, mock_info, mock_create): @patch("rag_pipeline.services.base.create_embedding_model") @patch("rag_pipeline.services.base.get_embedding_model_info") @patch("rag_pipeline.services.base.QdrantClient") - def test_plugin_identity_requires_current_descriptor_and_build_content( + def test_plugin_identity_requires_descriptor_but_accepts_older_build_content( self, MockQdrant, mock_info, mock_create ): from rag_pipeline.services.base import RAGQueryBase @@ -124,13 +124,16 @@ def test_plugin_identity_requires_current_descriptor_and_build_content( assert base._plugin_identity_compatible({ **current, "plugin_implementation_fingerprint": "sha256:old", - }) is False + "index_representation_fingerprint": "sha256:older-host", + }) is True assert base._plugin_identity_compatible({ - "plugin_ids": ["python", "fastapi"], + **current, + "plugin_descriptor_fingerprint": "sha256:other-descriptor", }) is False catalog.registry.fingerprint_for.assert_called_once_with( ("python", "fastapi") ) + catalog.implementation_fingerprint.assert_not_called() class TestRAGQueryService: @@ -223,7 +226,7 @@ def test_search_post_filters_boolean_plugin_points(self): filters = index.as_retriever.call_args.kwargs["filters"].filters assert [metadata_filter.key for metadata_filter in filters] == ["branch"] - def test_search_discards_stale_plugin_build_before_result_limit(self): + def test_search_discards_incompatible_plugin_descriptor_before_result_limit(self): from rag_pipeline.services.semantic_search import SemanticSearchMixin stale_node = SimpleNamespace( diff --git a/python-ecosystem/rag-pipeline/tests/test_splitter.py b/python-ecosystem/rag-pipeline/tests/test_splitter.py index 06a3c3e3..169c9412 100644 --- a/python-ecosystem/rag-pipeline/tests/test_splitter.py +++ b/python-ecosystem/rag-pipeline/tests/test_splitter.py @@ -6,6 +6,8 @@ from unittest.mock import MagicMock, patch, PropertyMock from dataclasses import dataclass +from llama_index.core.schema import Document + from rag_pipeline.core.splitter.splitter import ( ASTCodeSplitter, ASTChunk, @@ -775,6 +777,26 @@ def test_is_ast_supported_unknown(self): assert ASTCodeSplitter.is_ast_supported("test.xyz") is False +def test_resilient_splitter_quarantines_only_the_failing_file(): + splitter = ASTCodeSplitter() + good_node = MagicMock() + + def split(documents, capabilities=None): + path = documents[0].metadata["path"] + if path == "src/bad.custom": + raise ValueError("invalid project syntax") + return [good_node] + + splitter.split_documents = MagicMock(side_effect=split) + nodes, skipped_paths = splitter.split_documents_resilient([ + Document(text="valid", metadata={"path": "src/good.custom"}), + Document(text="invalid", metadata={"path": "src/bad.custom"}), + ]) + + assert nodes == [good_node] + assert skipped_paths == ("src/bad.custom",) + + # ── _get_text_splitter ──