fix-bodystreambuffer-hf

.streamlit/config.toml

@@ -3,13 +3,20 @@ headless = true
 port = 8501
 enableCORS = false
 enableXsrfProtection = false
+enableWebsocketCompression = false
+maxMessageSize = 700
+maxUploadSize = 200
 
 [browser]
 gatherUsageStats = false
 
+[runner]
+fastReruns = false
+enforceSerializableSessionState = false
+
 [theme]
 base = "light"
 primaryColor = "#4facfe"
 backgroundColor = "#ffffff"
 secondaryBackgroundColor = "#f0f2f6"
-textColor = "#262730"
+textColor = "#262730"

config.toml

@@ -3,10 +3,17 @@ headless = true
 port = 8501
 enableCORS = false
 enableXsrfProtection = false
+enableWebsocketCompression = false
+maxMessageSize = 700
+maxUploadSize = 200
 
 [browser]
 gatherUsageStats = false
 
+[runner]
+fastReruns = false
+enforceSerializableSessionState = false
+
 [theme]
 base = "light"
 primaryColor = "#4facfe"

modules/results.py

@@ -24,6 +24,11 @@ from modules.admin import get_active_prompt
 from utils.feedback_system import (
     display_feedback_section as display_feedback_section_util)
 from utils.translations import t, get_language
+from utils.streamlit_runtime import (
+    make_progress_callback,
+    is_connection_abort_error,
+    is_huggingface_space,
+)
 
 # Import CropDrive integration for user ID
 try:
@@ -665,17 +670,27 @@ def show_results_page():
                 status_text = st.empty()
                 time_estimate = st.empty()
                 step_indicator = st.empty()
+                stream_indicator = st.empty()
                 
                 # Add a "system is working" indicator
                 working_indicator = st.empty()
             
             # Process the new analysis with enhanced progress tracking
+            results_data = None
             try:
                 # Add a note about keeping the page open
                 st.info(f"⏳ **{t('results_important', 'Important')}:** {t('results_keep_page_open', 'Please keep this page open during analysis. The process may take 5-8 minutes.')}")
                 
                 logger.info("🔍 DEBUG - Calling process_new_analysis...")
-                results_data = process_new_analysis(st.session_state.analysis_data, progress_bar, status_text, time_estimate, step_indicator, working_indicator)
+                results_data = process_new_analysis(
+                    st.session_state.analysis_data,
+                    progress_bar,
+                    status_text,
+                    time_estimate,
+                    step_indicator,
+                    working_indicator,
+                    stream_indicator,
+                )
                 logger.info(f"🔍 DEBUG - process_new_analysis returned: {type(results_data)}, success={results_data.get('success') if results_data else 'None'}")
             except KeyboardInterrupt:
                 logger.warning("Analysis interrupted by user")
@@ -687,17 +702,31 @@ def show_results_page():
                 logger.error(f"Traceback: {traceback.format_exc()}")
                 
                 # Handle specific error types
-                if "bodystreambuffer" in error_msg or "aborted" in error_msg:
-                    results_data = {'success': False, 'message': 'Connection was interrupted during analysis. Please refresh the page and try again.'}
+                if is_connection_abort_error(e):
+                    results_data = {
+                        'success': False,
+                        'message': 'Connection was interrupted during analysis. Please refresh the page and try again.',
+                        'connection_aborted': True,
+                    }
                 elif "timeout" in error_msg:
                     results_data = {'success': False, 'message': 'Analysis timed out. Please try again with smaller files or check your connection.'}
                 else:
                     results_data = {'success': False, 'message': f'Processing error: {str(e)}'}
             finally:
-                # Clear the analysis_data from session state after processing (success or failure)
-                if 'analysis_data' in st.session_state:
+                # Keep analysis_data on connection abort so the user can retry without re-uploading
+                should_clear_input = (
+                    results_data
+                    and results_data.get('success')
+                    and not results_data.get('connection_aborted')
+                )
+                if should_clear_input and 'analysis_data' in st.session_state:
                     logger.info("🔍 DEBUG - Clearing analysis_data from session state")
                     del st.session_state.analysis_data
+                elif results_data and results_data.get('connection_aborted'):
+                    st.warning(
+                        "Connection interrupted. Your uploaded files are still saved — "
+                        "click Refresh, then start analysis again."
+                    )
             
             if results_data and results_data.get('success', False):
                 # Clear progress container
@@ -1379,7 +1408,15 @@ def reconstruct_firestore_data(data):
         logger.error(f"❌ Error reconstructing Firestore data: {e}")
         return data
 
-def process_new_analysis(analysis_data, progress_bar, status_text, time_estimate=None, step_indicator=None, working_indicator=None):
+def process_new_analysis(
+    analysis_data,
+    progress_bar,
+    status_text,
+    time_estimate=None,
+    step_indicator=None,
+    working_indicator=None,
+    stream_indicator=None,
+):
     """Process new analysis data from uploaded files"""
     try:
         import time
@@ -1657,15 +1694,11 @@ def process_new_analysis(analysis_data, progress_bar, status_text, time_estimate
         analysis_engine = AnalysisEngine()
         
         # Analysis processing (optimized - no delays)
-        status_text.text("🔬 **Step 4/5:** Running comprehensive agricultural analysis... 🔄")
-        time_estimate.text("⏱️ Processing data patterns and generating insights...")
-        step_indicator.text(f"📋 Step {current_step} of {total_steps} - AI Analysis")
-        
-        status_text.text("🔬 **Step 4/5:** Running comprehensive agricultural analysis... ✅")
+        status_text.markdown("🔬 **Step 4/5:** Running comprehensive agricultural analysis... 🔄")
         if time_estimate:
-            time_estimate.text("⏱️ Estimated time remaining: ~15 seconds")
+            time_estimate.markdown("⏱️ Processing data patterns and generating insights...")
         if step_indicator:
-            step_indicator.text(f"📋 Step {current_step} of {total_steps}")
+            step_indicator.markdown(f"📋 Step {current_step} of {total_steps} - AI Analysis")
         
         # Transform data structure to match analysis engine expectations
         transformed_soil_samples = []
@@ -1792,13 +1825,26 @@ def process_new_analysis(analysis_data, progress_bar, status_text, time_estimate
             status_text.text("🔬 **Step 4/5:** Starting comprehensive AI analysis... 🔄")
             progress_bar.progress(70)
             if time_estimate:
-                time_estimate.text("⏱️ This may take 5-8 minutes. Please keep this page open...")
-            
+                remaining = "3-6 minutes" if is_huggingface_space() else "5-8 minutes"
+                time_estimate.markdown(
+                    f"⏱️ This may take {remaining}. Please keep this page open..."
+                )
+
+            progress_callback = make_progress_callback(
+                progress_bar=progress_bar,
+                status_text=status_text,
+                time_estimate=time_estimate,
+                step_indicator=step_indicator,
+                working_indicator=working_indicator,
+                stream_indicator=stream_indicator,
+            )
+
             analysis_results = analysis_engine.generate_comprehensive_analysis(
                 soil_data=transformed_soil_data,
                 leaf_data=transformed_leaf_data,
                 land_yield_data=land_yield_data,
-                prompt_text=active_prompt.get('prompt_text', '')
+                prompt_text=active_prompt.get('prompt_text', ''),
+                progress_callback=progress_callback,
             )
             logger.info(f"✅ Analysis completed successfully")
             logger.info(f"🔍 Analysis results keys: {list(analysis_results.keys()) if isinstance(analysis_results, dict) else 'None'}")
@@ -1814,11 +1860,15 @@ def process_new_analysis(analysis_data, progress_bar, status_text, time_estimate
             
             # Check if it's a timeout/connection error
             error_msg = str(e).lower()
-            if "timeout" in error_msg or "aborted" in error_msg or "bodystreambuffer" in error_msg or "connection" in error_msg:
-                st.error("❌ **Analysis Timeout**: The analysis took too long or the connection was interrupted.")
-                st.info("💡 **Tip:** Large files may take longer to process. Please ensure you have a stable internet connection and try again.")
-                st.info("💡 **Alternative:** Try refreshing the page and starting the analysis again.")
-                return {'success': False, 'message': 'Analysis timeout or connection interrupted'}
+            if is_connection_abort_error(e) or "timeout" in error_msg:
+                st.error("❌ **Connection Interrupted**: The analysis stream was aborted before completion.")
+                st.info("💡 **Tip:** Keep this browser tab open and retry. On Hugging Face, long AI runs now stream progress to prevent disconnects.")
+                st.info("💡 **Alternative:** Refresh the page and run the analysis again — your uploaded files are still saved.")
+                return {
+                    'success': False,
+                    'message': 'Analysis timeout or connection interrupted',
+                    'connection_aborted': True,
+                }
             elif "quota" in error_msg or "429" in error_msg:
                 st.error("❌ **API Quota Exceeded**: The analysis service is temporarily unavailable due to high demand.")
                 st.info("💡 **Tip:** Please try again in a few minutes.")

utils/analysis_engine.py

@@ -2592,8 +2592,11 @@ class PromptAnalyzer:
             # Use optimal settings from memory: temperature=0.0 for maximum accuracy [[memory:7795938]]
             temperature = 0.0  
             
-            # Use maximum available tokens for Gemini 2.5 Pro [[memory:7795941]]
-            max_tokens = 65536  # Gemini 2.5 Pro maximum output tokens
+            # HF Spaces: smaller outputs finish faster and reduce websocket timeouts
+            if os.getenv("SPACE_ID") or os.getenv("HF_HOME"):
+                max_tokens = 16384
+            else:
+                max_tokens = 65536
             
             # Ensure the API key is available to all client layers
             try:
@@ -2710,7 +2713,8 @@ class PromptAnalyzer:
     def generate_step_analysis(self, step: Dict[str, str], soil_params: Dict[str, Any], 
                              leaf_params: Dict[str, Any], land_yield_data: Dict[str, Any],
                              previous_results: List[Dict[str, Any]] = None, total_steps: int = None, 
-                             runtime_ctx: Dict[str, Any] = None) -> Dict[str, Any]:
+                             runtime_ctx: Dict[str, Any] = None,
+                             progress_callback: Any = None) -> Dict[str, Any]:
         """Generate analysis for a specific step using LLM"""
         try:
             # Ensure LLM is available before proceeding
@@ -3119,45 +3123,78 @@ class PromptAnalyzer:
             
             Please provide your analysis in the requested JSON format. Be specific and detailed in your findings and recommendations. Use the research references to support your analysis where relevant."""
             
+            if progress_callback and hasattr(progress_callback, 'on_step_start'):
+                try:
+                    progress_callback.on_step_start(
+                        int(step['number']),
+                        step.get('title', f"Step {step['number']}"),
+                        total_steps or 1,
+                    )
+                except Exception:
+                    pass
+
             # Generate response using Google Gemini with retries
             self.logger.info(f"Generating LLM response for Step {step['number']}")
             last_err = None
             for attempt in range(1, (getattr(self.ai_config, 'retry_attempts', 3) or 3) + 1):
                 try:
                     if hasattr(self, '_use_direct_gemini') and self._use_direct_gemini:
-                        # Use direct Gemini API
+                        # Use direct Gemini API with streaming to keep HF websocket alive
                         import google.generativeai as genai
                         combined_prompt = f"{system_prompt}\n\n{human_prompt}"
                         generation_config = genai.types.GenerationConfig(
                             temperature=self._temperature,
                             max_output_tokens=self._max_tokens,
                         )
-                        resp_obj = self.llm.generate_content(
+                        response_text = ""
+                        stream = self.llm.generate_content(
                             combined_prompt,
                             generation_config=generation_config,
-                            safety_settings=getattr(self, '_safety_settings', None)
+                            safety_settings=getattr(self, '_safety_settings', None),
+                            stream=True,
                         )
+                        for chunk in stream:
+                            chunk_text = getattr(chunk, "text", None) or ""
+                            if chunk_text:
+                                response_text += chunk_text
+                                if progress_callback and hasattr(progress_callback, 'on_stream_chunk'):
+                                    try:
+                                        progress_callback.on_stream_chunk(
+                                            int(step['number']),
+                                            len(response_text),
+                                        )
+                                    except Exception:
+                                        pass
+
                         class GeminiResponse:
                             def __init__(self, content):
                                 self.content = content
-                        
-                        # Check if response is valid
-                        if not resp_obj.candidates or len(resp_obj.candidates) == 0:
-                            raise Exception(f"No response candidates generated. Safety filters may have blocked content.")
-                        
-                        candidate = resp_obj.candidates[0]
-                        if hasattr(candidate, 'finish_reason') and candidate.finish_reason != 1:  # 1 = STOP (successful completion)
-                            finish_reason_names = {0: "UNSPECIFIED", 1: "STOP", 2: "MAX_TOKENS", 3: "SAFETY", 4: "RECITATION", 5: "OTHER"}
-                            reason_name = finish_reason_names.get(candidate.finish_reason, f"UNKNOWN_{candidate.finish_reason}")
-                            raise Exception(f"Response generation failed with finish_reason: {reason_name} ({candidate.finish_reason}). This may be due to safety filters or content policy violations.")
-                        
-                        if not hasattr(resp_obj, 'text') or not resp_obj.text:
+
+                        if not response_text.strip():
                             raise Exception("Empty response from Gemini API. This may be due to safety filters.")
-                        
-                        response = GeminiResponse(resp_obj.text)
+
+                        response = GeminiResponse(response_text)
                     else:
-                        # Use LangChain client
-                        response = self.llm.invoke(system_prompt + "\n\n" + human_prompt)
+                        # Use LangChain client with streaming when available
+                        response_text = ""
+                        if hasattr(self.llm, "stream"):
+                            for chunk in self.llm.stream(system_prompt + "\n\n" + human_prompt):
+                                chunk_text = getattr(chunk, "content", None) or str(chunk)
+                                response_text += chunk_text
+                                if progress_callback and hasattr(progress_callback, 'on_stream_chunk'):
+                                    try:
+                                        progress_callback.on_stream_chunk(
+                                            int(step['number']),
+                                            len(response_text),
+                                        )
+                                    except Exception:
+                                        pass
+                            class LangChainResponse:
+                                def __init__(self, content):
+                                    self.content = content
+                            response = LangChainResponse(response_text)
+                        else:
+                            response = self.llm.invoke(system_prompt + "\n\n" + human_prompt)
                     last_err = None
                     break
                 except Exception as e:
@@ -7955,12 +7992,20 @@ class AnalysisEngine:
             return {}
 
     def generate_comprehensive_analysis(self, soil_data: Dict[str, Any], leaf_data: Dict[str, Any],
-                                      land_yield_data: Dict[str, Any], prompt_text: str) -> Dict[str, Any]:
+                                      land_yield_data: Dict[str, Any], prompt_text: str,
+                                      progress_callback: Any = None) -> Dict[str, Any]:
         """Generate comprehensive analysis with all components (enhanced)"""
         try:
             self.logger.info("Starting enhanced comprehensive analysis")
             start_time = datetime.now()
 
+            def _report_phase(message: str, fraction: float) -> None:
+                if progress_callback and hasattr(progress_callback, 'on_phase'):
+                    try:
+                        progress_callback.on_phase(message, fraction)
+                    except Exception:
+                        pass
+
             # Initialize previous_results for comprehensive analysis (no prior steps)
             previous_results = []
 
@@ -7987,6 +8032,8 @@ class AnalysisEngine:
             else:
                 self.logger.info("No structured leaf data found in session state")
 
+            _report_phase("🧮 Preprocessing soil and leaf data...", 0.05)
+
             # Step 1: Preprocess raw data (skip if already processed)
             self.logger.info("Preprocessing data...")
             if not (soil_data and 'parameter_statistics' in soil_data):
@@ -8075,6 +8122,8 @@ class AnalysisEngine:
             self.logger.info("Generating economic forecast...")
             economic_forecast = self.results_generator.generate_economic_forecast(land_yield_data, recommendations, previous_results)
 
+            _report_phase("📊 Comparing results against MPOB standards...", 0.2)
+
             # Step 6: Process prompt steps with LLM (enhanced)
             self.logger.info("Processing analysis steps...")
             steps = self.prompt_analyzer.extract_steps_from_prompt(prompt_text)
@@ -8091,11 +8140,23 @@ class AnalysisEngine:
                     # Inject runtime context for real-time, seasonal adjustments
                     runtime_ctx = self._get_runtime_context()
                     step_result = self.prompt_analyzer.generate_step_analysis(
-                        step, soil_params, leaf_params, land_yield_data, step_results, len(steps), runtime_ctx
+                        step,
+                        soil_params,
+                        leaf_params,
+                        land_yield_data,
+                        step_results,
+                        len(steps),
+                        runtime_ctx,
+                        progress_callback=progress_callback,
                     )
                     # Normalize structure (remove item_0 keys, parse inner JSON, drop raw dumps)
                     step_result = self._normalize_step_result(step_result)
                     step_results.append(step_result)
+                    if progress_callback and hasattr(progress_callback, 'on_step_complete'):
+                        try:
+                            progress_callback.on_step_complete(int(step.get('number', len(step_results))))
+                        except Exception:
+                            pass
                 except Exception as step_error:
                     self.logger.error(f"Error processing step {step.get('number', 'unknown')}: {str(step_error)}")
                     # Add fallback step result

utils/streamlit_runtime.py

@@ -0,0 +1,137 @@
+"""Streamlit runtime helpers for long-running jobs on Hugging Face Spaces."""
+
+from __future__ import annotations
+
+import os
+import time
+from typing import Any, Callable, Optional
+
+
+def is_huggingface_space() -> bool:
+    """Detect Hugging Face Spaces runtime."""
+    return bool(
+        os.getenv("SPACE_ID")
+        or os.getenv("HF_HOME")
+        or os.getenv("SYSTEM")
+    )
+
+
+class AnalysisProgressReporter:
+    """Keeps the Streamlit websocket alive during long Gemini calls."""
+
+    def __init__(
+        self,
+        progress_bar: Any = None,
+        status_text: Any = None,
+        time_estimate: Any = None,
+        step_indicator: Any = None,
+        working_indicator: Any = None,
+        stream_indicator: Any = None,
+        base_progress: float = 70.0,
+        progress_span: float = 25.0,
+    ):
+        self.progress_bar = progress_bar
+        self.status_text = status_text
+        self.time_estimate = time_estimate
+        self.step_indicator = step_indicator
+        self.working_indicator = working_indicator
+        self.stream_indicator = stream_indicator
+        self.base_progress = base_progress
+        self.progress_span = progress_span
+        self._last_flush = 0.0
+        self._total_steps = 1
+
+    def on_phase(self, message: str, fraction: float = 0.0) -> None:
+        self._update(
+            progress=self.base_progress + (self.progress_span * fraction),
+            status=message,
+            working=message,
+        )
+
+    def on_step_start(self, step_number: int, step_title: str, total_steps: int) -> None:
+        self._total_steps = max(total_steps, 1)
+        fraction = max(0.0, (step_number - 1) / self._total_steps)
+        self.on_phase(
+            f"🤖 **AI Step {step_number}/{total_steps}:** {step_title}...",
+            fraction=fraction,
+        )
+
+    def on_stream_chunk(self, step_number: int, chars_received: int) -> None:
+        now = time.time()
+        if now - self._last_flush < 0.75:
+            return
+        self._last_flush = now
+        fraction = min(0.99, ((step_number - 1) + min(chars_received / 4000.0, 0.95)) / self._total_steps)
+        progress = int(self.base_progress + (self.progress_span * fraction))
+        stream_msg = f"Receiving AI output... {chars_received:,} characters"
+        self._update(
+            progress=progress,
+            status=f"🤖 **AI Step {step_number}/{self._total_steps}:** {stream_msg}",
+            working=stream_msg,
+            stream=stream_msg,
+        )
+
+    def on_step_complete(self, step_number: int) -> None:
+        fraction = min(1.0, step_number / self._total_steps)
+        self.on_phase(
+            f"✅ **AI Step {step_number}/{self._total_steps}** complete",
+            fraction=fraction,
+        )
+
+    def _update(
+        self,
+        progress: float,
+        status: str,
+        working: Optional[str] = None,
+        stream: Optional[str] = None,
+    ) -> None:
+        try:
+            if self.progress_bar is not None:
+                self.progress_bar.progress(min(max(int(progress), 0), 100))
+            if self.status_text is not None:
+                self.status_text.markdown(status)
+            if self.working_indicator is not None and working:
+                self.working_indicator.markdown(f"🔄 **Processing:** {working}")
+            if self.stream_indicator is not None and stream:
+                self.stream_indicator.caption(stream)
+            if self.time_estimate is not None:
+                self.time_estimate.markdown(
+                    "⏱️ Analysis is still running — keep this tab open."
+                )
+            if self.step_indicator is not None:
+                self.step_indicator.markdown("📋 AI analysis in progress")
+        except Exception:
+            pass
+
+
+def make_progress_callback(
+    progress_bar: Any = None,
+    status_text: Any = None,
+    time_estimate: Any = None,
+    step_indicator: Any = None,
+    working_indicator: Any = None,
+    stream_indicator: Any = None,
+) -> AnalysisProgressReporter:
+    return AnalysisProgressReporter(
+        progress_bar=progress_bar,
+        status_text=status_text,
+        time_estimate=time_estimate,
+        step_indicator=step_indicator,
+        working_indicator=working_indicator,
+        stream_indicator=stream_indicator,
+    )
+
+
+def is_connection_abort_error(error: Exception | str) -> bool:
+    message = str(error).lower()
+    return any(
+        token in message
+        for token in (
+            "bodystreambuffer",
+            "aborted",
+            "connection reset",
+            "connection interrupted",
+            "client disconnected",
+            "stream closed",
+        )
+    )