feat: API response corrections - add message_id, component tracking, and error handling

- Added ID generation utilities (message_id, component_id, output_id)
- Enhanced ComponentMetadata with component_output, hasError, and structured errors
- Renamed result_preview to result in pipeline responses
- Added hasError flag to ChatResponse model
- Removed internal fields (current_file, pipeline_s3_key) from API responses
- Updated session_manager to use generate_message_id()
- Enhanced pipeline_manager with component tracking and error handling
- Updated API routes to expose new response structures

All changes are backward compatible. No database migration required.

api_routes_v2.py

@@ -40,6 +40,7 @@ s3 = boto3.client("s3", region_name=AWS_REGION)
 # ========================
 
 class Message(BaseModel):
+    message_id: Optional[str] = None  # Unique message identifier
     role: str
     content: str
     timestamp: Optional[str] = None
@@ -48,9 +49,11 @@ class Message(BaseModel):
     fileUrl: Optional[str] = None 
 
 class ChatResponse(BaseModel):
+    message_id: Optional[str] = None  # Message ID for assistant response
     assistant_response: str
     output: Dict[str, Any] = Field(default_factory=dict)
     final_output: Optional[Dict[str, Any]] = None
+    hasError: bool = False  # Error flag
     exception: Optional[str] = None
     api_response: Dict[str, Any]
     intent: Dict[str, Any]
@@ -594,6 +597,10 @@ def _assistant_response_payload(
     """
     Create ChatResponse payload with all required fields.
     """
+    # Generate message_id for assistant response
+    from services.schemas import generate_message_id
+    message_id = generate_message_id()
+    
     # Persist assistant message to S3
     _add_and_mirror_message(chat_id, "assistant", friendly_response)
     
@@ -602,9 +609,11 @@ def _assistant_response_payload(
     file_metadata = session.get("file_metadata", {})
     
     return ChatResponse(
+        message_id=message_id,
         assistant_response=friendly_response,
         output=output or {},
         final_output=final_output,
+        hasError=bool(exception),
         exception=exception,
         api_response=api_data,
         intent=intent,
@@ -804,7 +813,7 @@ async def get_all_sessions(
                     "created_at": created_at,
                     "last_activity": last_activity,
                     "state": session.get("state", "unknown"),
-                    "current_file": session.get("current_file"),
+                    # REMOVED: current_file (not needed for session list)
                     "chat_name": session.get("chat_name"),  # CHANGE: added field
                     "stats": session.get("stats", {}),
                     "total_messages": len(session.get("conversation_history", [])),
@@ -866,6 +875,14 @@ async def get_session_history(
         enhanced_pipelines = []
         for pipeline_meta in pipelines_hist:
             enhanced_pipe = pipeline_meta.copy()
+            
+            # CHANGE: Rename result_preview to result
+            if "result_preview" in enhanced_pipe:
+                enhanced_pipe["result"] = enhanced_pipe.pop("result_preview")
+            
+            # CHANGE: Remove internal S3 keys from response
+            enhanced_pipe.pop("pipeline_s3_key", None)
+            
             # Load pipeline definition from S3 to get components/tools
             pipeline_s3_key = pipeline_meta.get("pipeline_s3_key")
             if pipeline_s3_key:
@@ -892,6 +909,11 @@ async def get_session_history(
                 enhanced_pipe["tools"] = []
                 enhanced_pipe["component_count"] = 0
                 enhanced_pipe["components"] = []
+            
+            # CHANGE: Add hasError field if not present
+            if "hasError" not in enhanced_pipe:
+                enhanced_pipe["hasError"] = enhanced_pipe.get("status") == "failed"
+            
             enhanced_pipelines.append(enhanced_pipe)
         
         # Ensure sorting by most recent first (updated_at or created_at)

services/pipeline_manager.py

@@ -17,7 +17,8 @@ from pymongo import MongoClient
 from services.s3_manager import get_s3_manager
 from services.schemas import (
     PipelineSchema, ComponentMetadata, PipelineStatus, ComponentStatus,
-    FinalOutputS3, ComponentOutputS3, schema_to_dict
+    FinalOutputS3, ComponentOutputS3, schema_to_dict,
+    generate_component_id, generate_output_id
 )
 import os
 
@@ -145,6 +146,7 @@ class PipelineManager:
         
         for i, step in enumerate(pipeline_steps):
             component = ComponentMetadata(
+                component_id=generate_component_id(),
                 component_name=step.get("tool_name", "unknown"),
                 order=i + 1,
                 status=ComponentStatus.PENDING
@@ -156,6 +158,7 @@ class PipelineManager:
         self.s3.upload_json(s3_key, pipeline_definition, add_prefix=False)
         
         # Create pipeline record
+        output_id = generate_output_id()
         pipeline_record = PipelineSchema(
             execution_id=execution_id,
             session_id=session_id,
@@ -165,7 +168,8 @@ class PipelineManager:
             created_by_message=created_by_message,
             status=PipelineStatus.PROPOSED,
             pipeline_definition_s3_key=s3_key,
-            components=components
+            components=components,
+            output_id=output_id
         )
         
         # Insert into MongoDB
@@ -220,7 +224,14 @@ class PipelineManager:
                     comp["success_message"] = success_message
                 
                 if error:
-                    comp["error"] = error
+                    comp["error"] = {"message": error} if isinstance(error, str) else error
+                    comp["hasError"] = True
+                else:
+                    comp["hasError"] = False
+                
+                # Store component output
+                if output:
+                    comp["component_output"] = output
                 
                 component_found = True
                 break
@@ -296,11 +307,15 @@ class PipelineManager:
         component_outputs = []
         for i, result in enumerate(components_results):
             comp_output = ComponentOutputS3(
+                component_id=result.get("component_id", generate_component_id()),
                 name=result.get("component_name", f"component_{i+1}"),
                 order=i + 1,
                 status=result.get("status", "unknown"),
+                component_output=result.get("result"),
+                hasError=result.get("status") == "failed",
+                error={"message": result.get("error")} if result.get("error") else None,
                 success_message=result.get("success_message"),
-                error=result.get("error"),
+                metadata=result.get("metadata"),
                 result=result.get("result")
             )
             component_outputs.append(comp_output.model_dump())
@@ -335,7 +350,9 @@ class PipelineManager:
                     "final_output_presigned_expires_at": presigned["presigned_expires_at"],
                     "completed_at": now.isoformat() + "Z",
                     "executor": executor,
-                    "result_preview": result_preview
+                    "result": result_preview,
+                    "hasError": workflow_status == "failed",
+                    "error": {"message": "Pipeline execution failed"} if workflow_status == "failed" else None
                 }
             }
         )
@@ -370,7 +387,8 @@ class PipelineManager:
             {
                 "$set": {
                     "status": PipelineStatus.FAILED,
-                    "error": error,
+                    "error": {"message": error} if isinstance(error, str) else error,
+                    "hasError": True,
                     "completed_at": now.isoformat() + "Z"
                 }
             }

services/schemas.py

@@ -13,6 +13,7 @@ from typing import Dict, Any, List, Optional
 from datetime import datetime
 from pydantic import BaseModel, Field
 from enum import Enum
+import uuid
 
 
 # ========================
@@ -24,6 +25,9 @@ class SessionState(str, Enum):
     INITIAL = "initial"
     PIPELINE_PROPOSED = "pipeline_proposed"
     EXECUTING = "executing"
+    IN_PROGRESS = "in_progress"  # Pipeline is running
+    COMPLETED = "completed"  # Pipeline completed successfully
+    FAILED = "failed"  # Pipeline failed
 
 
 class PipelineStatus(str, Enum):
@@ -38,7 +42,9 @@ class ComponentStatus(str, Enum):
     """Component execution states"""
     PENDING = "pending"
     EXECUTING = "executing"
+    RUNNING = "running"  # Alias for executing
     COMPLETED = "completed"
+    SUCCESS = "success"  # Alias for completed
     FAILED = "failed"
 
 
@@ -55,13 +61,17 @@ class MessageRole(str, Enum):
 
 class ComponentMetadata(BaseModel):
     """Component execution metadata within a pipeline"""
+    component_id: str  # Unique identifier for component
     component_name: str
     order: int
     status: ComponentStatus = ComponentStatus.PENDING
     started_at: Optional[datetime] = None
     completed_at: Optional[datetime] = None
+    component_output: Optional[Any] = None  # Output from component execution
+    hasError: bool = False  # Error flag
+    error: Optional[Dict[str, Any]] = None  # Error details {error_code, message, details}
     success_message: Optional[str] = None  # Component-wise success message
-    error: Optional[str] = None
+    metadata: Optional[Dict[str, Any]] = None  # Duration, tokens, etc.
 
 
 class SessionStats(BaseModel):
@@ -127,13 +137,15 @@ class PipelineSchema(BaseModel):
     pipeline_definition_s3_key: str  # Full pipeline definition in S3
     components: List[ComponentMetadata]  # Component metadata only
     final_output_s3_key: Optional[str] = None  # Contains ALL component outputs
+    output_id: Optional[str] = None  # ID for downloading output
     final_output_presigned_url: Optional[str] = None
     final_output_presigned_expires_at: Optional[datetime] = None
     executed_at: Optional[datetime] = None
     completed_at: Optional[datetime] = None
     executor: Optional[str] = None  # "bedrock", "crewai", "langchain"
-    result_preview: Optional[str] = None  # First 500 chars of result
-    error: Optional[str] = None
+    result: Optional[str] = None  # Renamed from result_preview - output if completed
+    hasError: bool = False  # Error flag
+    error: Optional[Dict[str, Any]] = None  # Error details
     metadata: Dict[str, Any] = Field(default_factory=dict)
     
     class Config:
@@ -198,11 +210,15 @@ class MessageS3Content(BaseModel):
 
 class ComponentOutputS3(BaseModel):
     """Schema for individual component output in final_output.json"""
+    component_id: str
     name: str
     order: int
     status: str
+    component_output: Optional[Any] = None
+    hasError: bool = False
+    error: Optional[Dict[str, Any]] = None
     success_message: Optional[str] = None
-    error: Optional[str] = None
+    metadata: Optional[Dict[str, Any]] = None
     result: Optional[Dict[str, Any]] = None
 
 
@@ -232,6 +248,20 @@ class PipelineDefinitionS3(BaseModel):
 # UTILITY FUNCTIONS
 # ========================
 
+def generate_message_id() -> str:
+    """Generate unique message ID"""
+    return f"msg_{uuid.uuid4().hex[:12]}"
+
+
+def generate_component_id() -> str:
+    """Generate unique component ID"""
+    return f"comp_{uuid.uuid4().hex[:12]}"
+
+
+def generate_output_id() -> str:
+    """Generate unique output ID for downloads"""
+    return f"output_{uuid.uuid4().hex[:12]}"
+
 def datetime_to_iso(dt: Optional[datetime]) -> Optional[str]:
     """Convert datetime to ISO string"""
     return dt.isoformat() + "Z" if dt else None

services/session_manager.py

@@ -20,7 +20,8 @@ from pymongo.errors import DuplicateKeyError, ConnectionFailure
 from services.s3_manager import get_s3_manager
 from services.schemas import (
     SessionSchema, MessageSchema, SessionState, MessageRole,
-    SessionStats, MessageS3Content, schema_to_dict, datetime_to_iso
+    SessionStats, MessageS3Content, schema_to_dict, datetime_to_iso,
+    generate_message_id
 )
 
 
@@ -110,7 +111,7 @@ class SessionManager:
         Returns:
             Message metadata dict
         """
-        message_id = str(uuid.uuid4())
+        message_id = generate_message_id()
         now = datetime.utcnow()
         
         # Store full message content in S3