diff --git a/README.md b/README.md
index c29572ba..bf8f98fd 100644
--- a/README.md
+++ b/README.md
@@ -155,9 +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 | 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 11130a27..6919a3cd 100644
--- a/analysis-plugins/README.md
+++ b/analysis-plugins/README.md
@@ -146,3 +146,13 @@ 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, 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 3f4db32c..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,56 @@ 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():
+ 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():
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/__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..e6cd10c8 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,8 +55,11 @@ 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 +6919,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}"
@@ -6960,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
}
@@ -6996,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/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" <
@{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/.gitignore b/python-ecosystem/rag-pipeline/.gitignore
index 292e697b..13eeb6e8 100644
--- a/python-ecosystem/rag-pipeline/.gitignore
+++ b/python-ecosystem/rag-pipeline/.gitignore
@@ -20,4 +20,4 @@ build/
logs/**
**/__pycache__/
-
+newrelic.ini
diff --git a/python-ecosystem/rag-pipeline/Dockerfile.observable b/python-ecosystem/rag-pipeline/Dockerfile.observable
new file mode 100644
index 00000000..908674aa
--- /dev/null
+++ b/python-ecosystem/rag-pipeline/Dockerfile.observable
@@ -0,0 +1,63 @@
+FROM python:3.11-slim AS builder
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y \
+ build-essential \
+ && rm -rf /var/lib/apt/lists/*
+
+COPY python-ecosystem/rag-pipeline/requirements.txt .
+RUN pip install --upgrade pip && \
+ pip install --no-cache-dir -r requirements.txt
+
+# --- Production stage (no build-essential, saves ~340MB) ---
+FROM python:3.11-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y \
+ curl \
+ && rm -rf /var/lib/apt/lists/*
+
+# Copy installed Python packages and executables, including newrelic-admin.
+COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
+COPY --from=builder /usr/local/bin /usr/local/bin
+
+COPY python-ecosystem/rag-pipeline/src/ ./src/
+COPY python-ecosystem/rag-pipeline/main.py .
+COPY analysis-plugins/ ./plugins/
+
+RUN groupadd -r appgroup && useradd -r -g appgroup appuser
+
+# Create cache directories with proper permissions
+RUN mkdir -p /tmp/.tiktoken_cache /tmp/.transformers_cache /tmp/.huggingface /tmp/.llama_index && \
+ chmod -R 777 /tmp && \
+ chown -R appuser:appgroup /app
+
+USER appuser
+
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONPATH=/app/src:/app/plugins/contracts/python
+ENV CODECROW_PLUGINS_ROOT=/app/plugins
+ENV HOME=/tmp
+ENV TIKTOKEN_CACHE_DIR=/tmp/.tiktoken_cache
+ENV TRANSFORMERS_CACHE=/tmp/.transformers_cache
+ENV HF_HOME=/tmp/.huggingface
+ENV LLAMA_INDEX_CACHE_DIR=/tmp/.llama_index
+# Allow concurrent indexing by running multiple Uvicorn workers
+ENV UVICORN_WORKERS=4
+
+# CPU Threading Optimization (can be overridden via .env)
+# These control parallelism for numerical operations
+ENV OMP_NUM_THREADS=4
+ENV MKL_NUM_THREADS=4
+ENV OPENBLAS_NUM_THREADS=4
+ENV NUMEXPR_NUM_THREADS=4
+
+# New Relic observability. The config file is mounted by production Compose.
+ENV NEW_RELIC_CONFIG_FILE=/app/newrelic.ini
+ENV NEW_RELIC_LOG_FILE_NAME=STDOUT
+
+EXPOSE 8001
+
+CMD ["newrelic-admin", "run-program", "python", "main.py"]
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/main.py b/python-ecosystem/rag-pipeline/main.py
index 310f9fc4..066b2dd0 100644
--- a/python-ecosystem/rag-pipeline/main.py
+++ b/python-ecosystem/rag-pipeline/main.py
@@ -1,10 +1,43 @@
"""
Main entry point for the RAG Pipeline API server
"""
+
+import os as _os
+
+# Load deployment configuration before importing modules that read env vars at
+# import time. Keep this before New Relic and application imports.
+try:
+ from dotenv import load_dotenv as _load_dotenv
+ _load_dotenv(interpolate=False)
+except Exception as _dotenv_err:
+ print(f"[ENV-BOOT] ERROR loading .env: {_dotenv_err}", flush=True)
+
+# New Relic APM must be initialized before any application imports.
+_nr_config = _os.environ.get("NEW_RELIC_CONFIG_FILE")
+print(f"[NR-BOOT] NEW_RELIC_CONFIG_FILE = {_nr_config!r}", flush=True)
+if _nr_config:
+ _nr_exists = _os.path.exists(_nr_config)
+ print(f"[NR-BOOT] Config file exists: {_nr_exists}", flush=True)
+ if _nr_exists:
+ try:
+ import newrelic.agent
+ print(
+ f"[NR-BOOT] newrelic.agent imported, version={newrelic.version}",
+ flush=True,
+ )
+ newrelic.agent.initialize(_nr_config)
+ print(
+ "[NR-BOOT] newrelic.agent.initialize() completed successfully",
+ flush=True,
+ )
+ except Exception as _nr_err:
+ print(f"[NR-BOOT] ERROR during initialization: {_nr_err}", flush=True)
+else:
+ print("[NR-BOOT] Skipping — no config file env var set", flush=True)
+
import os
import sys
import logging
-from dotenv import load_dotenv
# Configure logging early
logging.basicConfig(
@@ -12,7 +45,6 @@
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
-load_dotenv(interpolate=False)
# In community/self-hosted mode, fetch embedding config from the Java web-server
# before reading env vars. This is a no-op if CODECROW_WEB_SERVER_URL is not set.
diff --git a/python-ecosystem/rag-pipeline/requirements.local.txt b/python-ecosystem/rag-pipeline/requirements.local.txt
index e9596746..b111c990 100644
--- a/python-ecosystem/rag-pipeline/requirements.local.txt
+++ b/python-ecosystem/rag-pipeline/requirements.local.txt
@@ -60,4 +60,7 @@ httpx>=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
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 5b545449..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
@@ -5,7 +5,9 @@
"""
import logging
+import time
import uuid
+from dataclasses import dataclass
from datetime import datetime, timezone
from typing import List, Dict, Tuple
@@ -16,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."""
@@ -25,11 +41,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(
@@ -156,22 +182,211 @@ 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_result = self._upsert_resilient(
+ collection_name,
+ batch,
+ batch_offset=i,
+ )
+ successful += batch_result.successful
+ skipped_points.extend(batch_result.skipped_points)
+
+ return PointWriteResult(successful, tuple(skipped_points))
+
+ @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
+ 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
+ return any(
+ marker in message
+ for marker in (
+ "bad request",
+ "payload too large",
+ "request entity too large",
+ "request too large",
+ "validation error",
+ )
+ )
+
+ @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,
+ ) -> PointWriteResult:
+ """Write a batch with bounded retry and rejected-request subdivision."""
+ if not points:
+ return PointWriteResult()
+
+ 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 PointWriteResult(successful=len(points))
+ 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,
+ )
+ 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 "
+ "(%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(
+ "Skipping one exact index point rejected by Qdrant "
+ "(%s): %s",
+ self._point_label(points[0]),
+ error,
+ )
+ return PointWriteResult(skipped_points=(points[0],))
+
+ 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 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,
@@ -199,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 bbc84eb2..d06e81b4 100644
--- a/python-ecosystem/rag-pipeline/tests/test_point_operations.py
+++ b/python-ecosystem/rag-pipeline/tests/test_point_operations.py
@@ -1,8 +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():
@@ -59,6 +65,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"})
@@ -85,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()
@@ -98,12 +106,182 @@ 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"})
for index in range(4)
]
+ with pytest.raises(PointWriteInfrastructureError):
+ operations.process_and_upsert_chunks(
+ chunks,
+ "pending",
+ "workspace",
+ "project",
+ "main",
+ )
+
+ 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,
+ )
+
+ 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 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",
@@ -112,5 +290,10 @@ def test_process_stops_embedding_after_pending_write_failure():
"main",
)
- assert (successful, failed) == (0, 2)
- embed_model.get_text_embedding_batch.assert_called_once()
+ 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 ──