fix(reliability): stabilize analysis queues and RAG lifecycle

deployment/config/inference-orchestrator/.env.sample

@@ -16,6 +16,7 @@ SERVICE_SECRET=change-me-to-a-random-secret
 # ANALYSIS_CONSUMER_HEARTBEAT_SECONDS=5
 # MAX_CONCURRENT_COMMANDS=10
 # COMMAND_TIMEOUT_SECONDS=600
+# COMMAND_EVENT_TTL_SECONDS=3600
 # REVIEW_TIMEOUT_SECONDS=1500
 # REVIEW_GLOBAL_RAG_QUERY_TIMEOUT_SECONDS=5
 

deployment/config/java-shared/application.properties.sample

@@ -192,6 +192,19 @@ logging.level.org.hibernate.orm.jdbc.bind=OFF
 # Repair interval for readable Qdrant current-branch aliases of active generations.
 #codecrow.rag.operator-alias.reconcile-interval-ms=300000
 #codecrow.rag.operator-alias.reconcile-initial-delay-ms=15000
+# Exact-generation operation recovery. A live build heartbeats its persisted
+# operation; a published operation also repairs an interrupted job/status handoff.
+#codecrow.rag.generation.stale-after-minutes=30
+#codecrow.rag.generation.recovery-interval-ms=300000
+#codecrow.rag.generation.recovery-initial-delay-ms=60000
+# Legacy unsealed incremental updates use a separate database job lease because
+# they do not own an immutable generation operation.
+#codecrow.rag.legacy-job.lease-seconds=120
+#codecrow.rag.legacy-job.heartbeat-interval-seconds=15
+#codecrow.rag.legacy-job.heartbeat-threads=4
+#codecrow.rag.legacy-job.recovery-interval-ms=30000
+#codecrow.rag.legacy-job.recovery-initial-delay-ms=30000
+#codecrow.rag.legacy-job.recovery-batch-size=100
 # Shared VCS acquisition threshold for incremental RAG and reconciliation
 # fallback. The legacy codecrow.rag.incremental.archive-file-threshold property
 # remains a fallback when this property is not set.
@@ -200,6 +213,10 @@ logging.level.org.hibernate.orm.jdbc.bind=OFF
 # Analysis Lock Configuration
 # Lock timeout - maximum time for a single analysis to hold a lock (in minutes)
 #analysis.lock.timeout.minutes=30
+# Independent lease-renewal scheduler capacity (default 4, minimum 2 threads)
+#analysis.lock.heartbeat.threads=4
+# Independent lease-renewal interval (seconds); bounded to one-third of the lease
+#analysis.lock.heartbeat.interval.seconds=60
 # RAG indexing lock timeout - longer timeout for RAG indexing operations (in minutes, default 6 hours)
 #analysis.lock.rag.timeout.minutes=360
 # Lock wait timeout - maximum time to wait for a lock to be released (in minutes)

deployment/config/rag-pipeline/.env.sample

@@ -49,6 +49,7 @@ SERVICE_SECRET=change-me-to-a-random-secret
 # QDRANT_URL=http://qdrant:6333
 # QDRANT_API_KEY=
 # QDRANT_COLLECTION_PREFIX=codecrow
+# QDRANT_TIMEOUT_SECONDS=30
 # QDRANT_VECTORS_ON_DISK=true
 # QDRANT_UPSERT_BATCH_SIZE=128
 

java-ecosystem/libs/analysis-engine/src/main/java/org/rostilos/codecrow/analysisengine/aiclient/AiAnalysisClient.java

@@ -12,11 +12,13 @@
 import org.rostilos.codecrow.queue.RedisQueueService;
 import org.rostilos.codecrow.core.model.rag.RagBranchIndexGenerationStatus;
 import org.rostilos.codecrow.core.persistence.repository.rag.RagBranchIndexGenerationRepository;
+import org.rostilos.codecrow.core.persistence.repository.rag.RagBranchIndexRepository;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestTemplate;
 
 import java.io.IOException;
 import java.security.GeneralSecurityException;
+import java.time.OffsetDateTime;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -40,6 +42,9 @@ public class AiAnalysisClient {
     @Autowired(required = false)
     private RagBranchIndexGenerationRepository branchGenerationRepository;
 
+    @Autowired(required = false)
+    private RagBranchIndexRepository branchIndexRepository;
+
     static final String INACTIVITY_TIMEOUT_MINUTES_KEY =
             "ANALYSIS_QUEUE_INACTIVITY_TIMEOUT_MINUTES";
     static final String ADMISSION_TIMEOUT_MINUTES_KEY =
@@ -353,11 +358,16 @@ private Map<String, Object> buildSerializableRequestPayload(AiAnalysisRequest re
         payload.put("previousCommitHash", request.getPreviousCommitHash());
         payload.put("currentCommitHash", request.getCurrentCommitHash());
         payload.put("baseCommitHash", request.getBaseCommitHash());
-        if (branchGenerationRepository != null
+        if (request.getRagEnabled()
+                && branchGenerationRepository != null
+                && branchIndexRepository != null
                 && request.getProjectId() != null
                 && request.getTargetBranchName() != null
                 && request.getBaseCommitHash() != null) {
-            branchGenerationRepository.findAvailableExactGeneration(
+            int accessed = branchIndexRepository.markAccessedIfUnclaimed(
+                    request.getProjectId(), request.getTargetBranchName(), OffsetDateTime.now());
+            if (accessed > 0) {
+                branchGenerationRepository.findAvailableExactGeneration(
                             request.getProjectId(),
                             request.getTargetBranchName(),
                             request.getBaseCommitHash(),
@@ -371,6 +381,7 @@ private Map<String, Object> buildSerializableRequestPayload(AiAnalysisRequest re
                         payload.put("ragBaseGenerationManifestSha256",
                                 generation.getManifestDigest());
                     });
+            }
         }
         payload.put("previousCodeAnalysisIssues", request.getPreviousCodeAnalysisIssues());
         payload.put("reconciliationFileContents", request.getReconciliationFileContents());

java-ecosystem/libs/analysis-engine/src/main/java/org/rostilos/codecrow/analysisengine/aiclient/AiCommandClient.java

@@ -4,6 +4,7 @@
 import org.rostilos.codecrow.queue.RedisQueueService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.io.IOException;
@@ -24,10 +25,35 @@ public class AiCommandClient {
     private final RedisQueueService queueService;
     private final ObjectMapper objectMapper;
     private static final int COMMAND_TIMEOUT_MINUTES = 30;
-
+    static final String CONSUMER_HEARTBEAT_KEY =
+            "codecrow:commands:consumer:heartbeat";
+    private static final long DEFAULT_ADMISSION_TIMEOUT_MINUTES = 5L;
+    private static final String ADMISSION_TIMEOUT_MINUTES_KEY =
+            "COMMAND_QUEUE_ADMISSION_TIMEOUT_MINUTES";
+    private final long commandTimeoutMillis;
+    private final long admissionTimeoutMillis;
+
+    @Autowired
     public AiCommandClient(RedisQueueService queueService, ObjectMapper objectMapper) {
+        this(
+                queueService,
+                objectMapper,
+                TimeUnit.MINUTES.toMillis(COMMAND_TIMEOUT_MINUTES),
+                resolveAdmissionTimeoutMillis());
+    }
+
+    AiCommandClient(
+            RedisQueueService queueService,
+            ObjectMapper objectMapper,
+            long commandTimeoutMillis,
+            long admissionTimeoutMillis) {
         this.queueService = queueService;
         this.objectMapper = objectMapper;
+        if (commandTimeoutMillis <= 0 || admissionTimeoutMillis <= 0) {
+            throw new IllegalArgumentException("Command queue timeouts must be positive");
+        }
+        this.commandTimeoutMillis = commandTimeoutMillis;
+        this.admissionTimeoutMillis = admissionTimeoutMillis;
     }
 
     /**
@@ -85,31 +111,52 @@ private Map<String, Object> executeAsyncJob(
             Consumer<Map<String, Object>> eventHandler) throws IOException {
         String eventQueueKey = "codecrow:analysis:events:" + jobId;
         String jobsQueueKey = "codecrow:queue:commands";
+        String jsonPayload = null;
 
         try {
+            if (!queueService.hasKey(CONSUMER_HEARTBEAT_KEY)) {
+                throw new IOException(
+                        "Inference Orchestrator is unavailable: no live command queue consumer");
+            }
             Map<String, Object> jobPayload = Map.of(
                     "job_id", jobId,
                     "command_type", commandType,
                     "request", request);
 
-            String jsonPayload = objectMapper.writeValueAsString(jobPayload);
+            jsonPayload = objectMapper.writeValueAsString(jobPayload);
             queueService.leftPush(jobsQueueKey, jsonPayload);
             queueService.setExpiry(eventQueueKey, COMMAND_TIMEOUT_MINUTES + 1);
 
             long startTime = System.currentTimeMillis();
-            long timeoutMillis = TimeUnit.MINUTES.toMillis(COMMAND_TIMEOUT_MINUTES);
+            boolean workerAcknowledged = false;
 
             while (true) {
-                if (System.currentTimeMillis() - startTime > timeoutMillis) {
+                long now = System.currentTimeMillis();
+                if (!workerAcknowledged) {
+                    if (!queueService.hasKey(CONSUMER_HEARTBEAT_KEY)) {
+                        throw new IOException(
+                                "Inference Orchestrator became unavailable before "
+                                        + "the command job was admitted");
+                    }
+                    if (now - startTime > admissionTimeoutMillis) {
+                        throw new IOException(
+                                "AI command was not admitted by Inference Orchestrator within "
+                                        + TimeUnit.MILLISECONDS.toSeconds(admissionTimeoutMillis)
+                                        + " seconds");
+                    }
+                }
+                if (now - startTime > commandTimeoutMillis) {
                     throw new IOException(
-                            "AI command timed out after " + COMMAND_TIMEOUT_MINUTES + " minutes for Job: " + jobId);
+                            "AI command timed out after " + commandTimeoutMillis
+                                    + "ms for Job: " + jobId);
                 }
 
                 String eventJson = queueService.rightPop(eventQueueKey, 5);
 
                 if (eventJson == null) {
                     continue; // Timeout on rightPop, continue to check overall timeout
                 }
+                workerAcknowledged = true;
 
                 try {
                     Map<String, Object> event = objectMapper.readValue(eventJson, Map.class);
@@ -150,13 +197,42 @@ private Map<String, Object> executeAsyncJob(
             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 command job {} from pending queue: {}",
+                            jobId, cleanupError.getMessage());
+                }
+            }
             try {
                 queueService.deleteKey(eventQueueKey);
             } catch (Exception ignored) {
             }
         }
     }
 
+    private static long resolveAdmissionTimeoutMillis() {
+        String configured = System.getProperty(ADMISSION_TIMEOUT_MINUTES_KEY);
+        if (configured == null || configured.isBlank()) {
+            configured = System.getenv(ADMISSION_TIMEOUT_MINUTES_KEY);
+        }
+        if (configured == null || configured.isBlank()) {
+            return TimeUnit.MINUTES.toMillis(DEFAULT_ADMISSION_TIMEOUT_MINUTES);
+        }
+        try {
+            long minutes = Long.parseLong(configured.trim());
+            if (minutes <= 0) {
+                throw new IllegalArgumentException(
+                        ADMISSION_TIMEOUT_MINUTES_KEY + " must be a positive integer");
+            }
+            return TimeUnit.MINUTES.toMillis(minutes);
+        } catch (NumberFormatException invalid) {
+            throw new IllegalArgumentException(
+                    ADMISSION_TIMEOUT_MINUTES_KEY + " must be a positive integer", invalid);
+        }
+    }
+
     /**
      * Request object for summarize endpoint.
      */