sdlc 80 percent

src/langgraphagenticai/graph/graph_builder.py

@@ -2,7 +2,11 @@ from langgraph.graph import StateGraph, END, MessagesState
 from langgraph.prebuilt import tools_condition,ToolNode
 from langchain_core.prompts import ChatPromptTemplate
 import datetime
+from langgraph.types import interrupt, Command
+from langgraph.checkpoint.memory import MemorySaver
+import streamlit as st
 #module import
+
 from src.langgraphagenticai.node.sdlc_node import SDLCNode
 from src.langgraphagenticai.node.ai_news_node import AINewsNode
 from src.langgraphagenticai.node import travel_planner_node
@@ -150,21 +154,147 @@ class GraphBuilder:
         self.graph_builder.add_edge("fetch_news", "summarize_news")
         self.graph_builder.add_edge("summarize_news", "save_result")
         self.graph_builder.add_edge("save_result", END)
-        
-        
+    
+    
     def sdlc_workflow_build_graph(self):
-        
         sdlc_wf_node = SDLCNode(self.llm)
-        
         self.graph_builder = self.sdlc_graph_builder
+        try:
 
-        self.graph_builder.add_node("generate_user_stories", sdlc_wf_node.generate_user_stories)
-        self.graph_builder.add_node("generate_code", sdlc_wf_node.generate_code)
+            # Add all primary workflow nodes.
+            nodes = [
+                ("generate_user_stories", sdlc_wf_node.generate_user_stories),
+                ("product_owner_review", sdlc_wf_node.product_owner_review),
+                ("create_design_docs", sdlc_wf_node.create_design_docs),
+                ("revise_user_stories", sdlc_wf_node.revise_user_stories),
+                ("design_review", sdlc_wf_node.design_review),
+                ("generate_code", sdlc_wf_node.generate_code),
+                ("code_review", sdlc_wf_node.code_review),
+                ("security_review", sdlc_wf_node.security_review),
+                ("fix_code_after_code_review", sdlc_wf_node.fix_code_after_code_review),
+                ("fix_code_after_security", sdlc_wf_node.fix_code_after_security),
+                ("write_test_cases", sdlc_wf_node.write_test_cases),
+                ("test_cases_review", sdlc_wf_node.test_cases_review),
+                ("fix_test_cases", sdlc_wf_node.fix_test_cases),
+            ]
 
-        self.graph_builder.set_entry_point("generate_user_stories")
-        self.graph_builder.add_edge("generate_user_stories", "generate_code")
-        self.graph_builder.add_edge("generate_code", END)
-        
+            # Helper functions to wrap review nodes.
+            def human_loop_node(review_field):
+                def node(state):
+                    # Trigger an interrupt to surface the LLM-generated review.
+                    if st.session_state.user_decision is not  "approve":
+                        value = interrupt({
+                            "__interrupt__": True,
+                            "review": state.get(review_field, ""),
+                            "instruction": f"Please review the '{review_field}'. Approve or provide feedback to reject."
+                        })
+                    else :
+                        value = st.session_state.user_decision
+                        st.session_state.user_decision = ''
+                    return {"human_decision": value}
+                return node
+
+            def decision_node(previous_node):
+                def node(state):
+                    if state.get("human_decision") == "approve":
+                        state["decision"] = "approve"
+                    else:
+                        state["decision"] = "reject"
+                        state["feedback"] = state.get("human_decision")
+                    return state
+                return node
+
+            review_nodes = ["product_owner_review", "design_review", "code_review", "test_cases_review"]
+            additional_nodes = []
+            for review in review_nodes:
+                additional_nodes.append((f"human_loop_{review}", human_loop_node(review)))
+                # Set the previous node for rejection (adjust as needed):
+                if review == "product_owner_review":
+                    prev = "generate_user_stories"
+                elif review == "design_review":
+                    prev = "revise_user_stories"
+                elif review == "code_review":
+                    prev = "generate_code"
+                elif review == "test_cases_review":
+                    prev = "write_test_cases"
+                additional_nodes.append((f"decision_{review}", decision_node(prev)))
+
+            # Add all nodes to the graph.
+            for node_name, node_func in nodes + additional_nodes:
+                self.graph_builder.add_node(node_name, node_func)
+
+            # Set entry point.
+            if st.session_state.graph_stage == 'resumed':
+                self.graph_builder.set_entry_point(st.session_state.state['current_step'])
+            else:   
+                self.graph_builder.set_entry_point("generate_user_stories")
+
+            # ---- Build Flow Edges ----
+
+            # Wrap product_owner_review:
+            self.graph_builder.add_edge("generate_user_stories", "product_owner_review")
+            self.graph_builder.add_edge("product_owner_review", "human_loop_product_owner_review")
+            self.graph_builder.add_edge("human_loop_product_owner_review", "decision_product_owner_review")
+            self.graph_builder.add_conditional_edges(
+                "decision_product_owner_review",
+                lambda state: "approve" if state.get("decision") == "approve" else "reject",
+                {
+                    "approve": "create_design_docs",
+                    "reject": "generate_user_stories"
+                }
+            )
+
+            # Wrap design_review:
+            self.graph_builder.add_edge("revise_user_stories", "design_review")
+            self.graph_builder.add_edge("design_review", "human_loop_design_review")
+            self.graph_builder.add_edge("human_loop_design_review", "decision_design_review")
+            self.graph_builder.add_conditional_edges(
+                "decision_design_review",
+                lambda state: "approve" if state.get("decision") == "approve" else "reject",
+                {
+                    "approve": "generate_code",
+                    "reject": "revise_user_stories"
+                }
+            )
+
+            # Wrap code_review:
+            self.graph_builder.add_edge("generate_code", "code_review")
+            self.graph_builder.add_edge("code_review", "human_loop_code_review")
+            self.graph_builder.add_edge("human_loop_code_review", "decision_code_review")
+            self.graph_builder.add_conditional_edges(
+                "decision_code_review",
+                lambda state: "approve" if state.get("decision") == "approve" else "reject",
+                {
+                    "approve": "security_review",
+                    "reject": "generate_code"
+                }
+            )
+
+            # Wrap test_cases_review:
+            self.graph_builder.add_edge("write_test_cases", "test_cases_review")
+            self.graph_builder.add_edge("test_cases_review", "human_loop_test_cases_review")
+            self.graph_builder.add_edge("human_loop_test_cases_review", "decision_test_cases_review")
+            self.graph_builder.add_conditional_edges(
+                "decision_test_cases_review",
+                lambda state: "approve" if state.get("decision") == "approve" else "reject",
+                {
+                    "approve": "fix_test_cases",
+                    "reject": "write_test_cases"
+                }
+            )
+
+            # Other sequential edges.
+            self.graph_builder.add_edge("create_design_docs", "revise_user_stories")
+            self.graph_builder.add_edge("security_review", "fix_code_after_code_review")
+            self.graph_builder.add_edge("fix_code_after_code_review", "fix_code_after_security")
+            self.graph_builder.add_edge("fix_code_after_security", "write_test_cases")
+
+            # Set finish point at the end of the workflow.
+            self.graph_builder.set_finish_point("fix_test_cases")
+        except Exception as e:
+            print(e)
+
+        self.graph_builder
 
 
     def setup_graph(self, usecase: str):
@@ -184,7 +314,9 @@ class GraphBuilder:
         elif usecase =="AI News":
             self.ai_news_build_graph()
         elif usecase =="SDLC Workflow":
+            checkpointer = MemorySaver()
             self.sdlc_workflow_build_graph()
+            return self.graph_builder.compile(checkpointer=checkpointer)
         else:
             raise ValueError("Invalid use case selected.")
         return self.graph_builder.compile()

src/langgraphagenticai/main.py

@@ -18,6 +18,7 @@ def load_langgraph_agenticai_app():
     implementing exception handling for robustness.
     """
     try:
+        
         # Load UI
         ui = LoadStreamlitUI()
         user_input = ui.load_streamlit_ui()
@@ -25,11 +26,13 @@ def load_langgraph_agenticai_app():
         if not user_input:
             st.error("Error: Failed to load user input from the UI.")
             return
+        
+        user_message = ''
 
         # Text input for user message
         if st.session_state.IsFetchButtonClicked:
             user_message = st.session_state.timeframe 
-        elif st.session_state.IsSDLC :
+        elif st.session_state.IsSDLC or 'current_step' in st.session_state.state:
             user_message = st.session_state.state
         else :
             user_message = st.chat_input("Enter your message:")

src/langgraphagenticai/node/sdlc_node.py

@@ -1,38 +1,410 @@
+# sdlc_node.py
 import streamlit as st
+import logging
+from typing import Dict, Any
+
+from src.langgraphagenticai.ui.streamlitui.sdlcfeedback import SDLCUI
+
+# Configure logging for production (adjust level and handlers as necessary)
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
 
 class SDLCNode:
     def __init__(self, llm):
         self.llm = llm
+        self.logger = logging.getLogger(self.__class__.__name__)
 
-    def generate_user_stories(self, state: dict) -> dict:
-        prompt = f"""
-        Generate comprehensive user stories based on the following requirements:
-        {state['requirements']}
-        
-        {f"PO Feedback to incorporate: {state['po_feedback']}" if state.get('po_feedback') else ""}
-        
-        Format the user stories in Markdown with clear acceptance criteria.
+    def _update_state(self, updates: Dict[str, Any]) -> None:
         """
-        
-        response = self.llm.invoke(prompt)
-        
-        st.session_state.state["user_stories"] = response.content
-        st.session_state.state["current_step"] = "po_approval"
-        return {"user_stories": response.content, "current_step": "po_approval"}
+        Safely update st.session_state['state'] with provided updates.
+        """
+        try:
+            if 'state' not in st.session_state:
+                st.session_state['state'] = {}
+            st.session_state['state'].update(updates)
+        except Exception as e:
+            self.logger.exception("Failed to update session state.")
+            st.error(f"Error updating session state: {e}")
 
-    def po_approval(self, state: dict) -> dict:
-        return {"current_step": "po_approval"}
+    def refresh_ui(self):
+        try:
+            ui = SDLCUI()
+            ui.render()
+        except Exception as e:
+            self.logger.exception("Error refreshing UI.")
+            st.error(f"UI Refresh failed: {e}")
 
-    def generate_code(self, state: dict) -> dict:
-        prompt = f"""
-        Generate production-quality code based on these user stories:
-        {state['user_stories']}
-        
-        {f"Review Feedback to incorporate: {state['review_feedback']}" if state.get('review_feedback') else ""}
+    def generate_user_stories(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        if st.session_state.state['user_stories'] is not '':
+            return {"user_stories": st.session_state.state['user_stories']}
+        try:
+            prompt = f"""
+                Generate comprehensive user stories based on these requirements:
+                {state.get('requirements', 'No requirements provided')}
+
+                {f"PO Feedback to incorporate: {state.get('po_feedback', '')}" if state.get('po_feedback') else ""}
+                
+                Format as Markdown with clear acceptance criteria.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "user_stories": content,
+                "current_result": content,
+                "current_step": "product_owner_review"
+            })
+            return {"user_stories": content}
+        except Exception as e:
+            self.logger.exception("Error in generate_user_stories.")
+            st.error(f"Failed to generate user stories: {e}")
+            return {"error": str(e)}
+
+    def product_owner_review(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        if st.session_state.state['po_feedback'] is not '':
+            return {"po_feedback": st.session_state.state['po_feedback']}
         
-        Include proper error handling, comments, and follow best practices.
-        """
+        try:
+            prompt = f"""
+                Please review the following user stories and provide your feedback.
+                
+                User Stories:
+                {state.get('user_stories', 'No user stories available')}
+                
+                If you approve, simply type "approve". Otherwise, provide detailed feedback.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "po_feedback": content,
+                "current_result": content,
+                "current_step": "create_design_docs"  # Updated to next node
+            })
+            st.session_state.user_decision = None
+            return {"po_feedback": content}
+        except Exception as e:
+            self.logger.exception("Error in product_owner_review.")
+            st.error(f"Product owner review failed: {e}")
+            return {"error": str(e)}
+
+    def create_design_docs(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        if st.session_state.state['design_docs'] is not '':
+            return {"design_docs": st.session_state.state['design_docs']}
+        try:
+            prompt = f"""
+                Create comprehensive design documents for the following user stories:
+                {state.get('user_stories', 'No user stories available')}
+                
+                Include both functional and technical specifications.
+                Provide architecture diagrams in Mermaid format.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "design_docs": content,
+                "current_result": content,
+                "current_step": "revise_user_stories"  # Updated to next node
+            })
+            return {"design_docs": content}
+        except Exception as e:
+            self.logger.exception("Error in create_design_docs.")
+            st.error(f"Failed to create design documents: {e}")
+            return {"error": str(e)}
+
+    def revise_user_stories(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        if st.session_state.state['user_stories'] is not '':
+            return {"user_stories": st.session_state.state['user_stories']}
+        try:
+            prompt = f"""
+                Revise the user stories based on the following Product Owner feedback:
+                
+                Original Stories: {state.get('user_stories', 'No stories generated')}
+                Feedback: {state.get('po_feedback', 'No feedback provided')}
+                
+                Please maintain Markdown format and include clear acceptance criteria.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "user_stories": content,
+                "current_result": content,
+                "current_step": "design_review"  # Updated to next node
+            })
+            return {"user_stories": content}
+        except Exception as e:
+            self.logger.exception("Error in revise_user_stories.")
+            st.error(f"Failed to revise user stories: {e}")
+            return {"error": str(e)}
+
+    def design_review(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        if st.session_state.state['design_feedback'] is not '':
+            return {"design_feedback": st.session_state.state['design_feedback']}
+        try:
+            prompt = f"""
+                Please review the following design documents and provide your feedback.
+                
+                Design Documents:
+                {state.get('design_docs', 'No design documents available')}
+                
+                If you approve, type "approve". Otherwise, provide detailed design feedback.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "design_feedback": content,
+                "current_result": content,
+                "current_step": "generate_code"  # Updated to next node
+            })
+            st.session_state.user_decision = None
+            return {"design_feedback": content}
+        except Exception as e:
+            self.logger.exception("Error in design_review.")
+            st.error(f"Design review failed: {e}")
+            return {"error": str(e)}
+
+    def generate_code(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        if st.session_state.state['generate_code'] is not '':
+            return {"generate_code": st.session_state.state['generate_code']}
+            
+        try:
+            prompt = f"""
+                Generate production-quality code for the following design:
+                {state.get('design_docs', 'No design documents available')}
+                
+                {f"Code Review Feedback: {state.get('review_feedback', '')}" if state.get('review_feedback') else ""}
+                
+                Include error handling, clear comments, and follow best coding practices.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "generate_code": content,
+                "current_result": content,
+                "current_step": "code_review"  # Updated to next node
+            })
+            return {"generate_code": content}
+        except Exception as e:
+            self.logger.exception("Error in generate_code.")
+            st.error(f"Failed to generate code: {e}")
+            return {"error": str(e)}
+
+    def code_review(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        if st.session_state.state['review_feedback'] is not '':
+            return {"review_feedback": st.session_state.state['review_feedback']}
+        try:
+            prompt = f"""
+                Please review the following generated code and provide your feedback.
+                
+                Code:
+                {state.get('generated_code', 'No code generated')}
+                
+                If the code is acceptable, type "approve". Otherwise, provide specific code review feedback.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "code_review": content,
+                "current_result": content,
+                "current_step": "security_review"  # Updated to next node
+            })
+            st.session_state.user_decision = None
+            return {"review_feedback": content}
+        except Exception as e:
+            self.logger.exception("Error in code_review.")
+            st.error(f"Code review failed: {e}")
+            return {"error": str(e)}
+
+    def security_review(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        if st.session_state.state['security_feedback'] is not '':
+            return {"security_feedback": st.session_state.state['security_feedback']}
+        try:
+            prompt = f"""
+                Please review the following code for potential security vulnerabilities:
+                
+                Code:
+                {state.get('generated_code', 'No code available')}
+                
+                Provide any security-related feedback.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "security_feedback": content,
+                "current_result": content,
+                "current_step": "fix_code_after_code_review"  # Updated to next node
+            })
+            st.session_state.user_decision = None
+            return {"security_feedback": content, "current_step": "fix_code_after_code_review"}
+        except Exception as e:
+            self.logger.exception("Error in security_review.")
+            st.error(f"Security review failed: {e}")
+            return {"error": str(e)}
+
+    def fix_code_after_code_review(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            prompt = f"""
+                Fix the following code based on the code review feedback provided.
+                
+                Original Code: {state.get('generated_code', 'No code available')}
+                Feedback: {state.get('review_feedback', 'No feedback provided')}
+                
+                Ensure that functionality is preserved while addressing all feedback.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "generated_code": content,
+                "current_result": content,
+                "current_step": "fix_code_after_security"  # Updated to next node
+            })
+            st.session_state.user_decision = None
+            return {"generated_code": content}
+        except Exception as e:
+            self.logger.exception("Error in fix_code_after_code_review.")
+            st.error(f"Failed to fix code after review: {e}")
+            return {"error": str(e)}
+
+    def fix_code_after_security(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            prompt = f"""
+                Improve the security of the following code based on the security review feedback:
+                
+                Code: {state.get('generated_code', 'No code available')}
+                Security Feedback: {state.get('security_feedback', 'No security feedback provided')}
+                
+                Apply best practices for security and fix any vulnerabilities.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "generated_code": content,
+                "current_result": content,
+                "current_step": "write_test_cases"  # Updated to next node
+            })
+            return {"generated_code": content}
+        except Exception as e:
+            self.logger.exception("Error in fix_code_after_security.")
+            st.error(f"Failed to fix code for security: {e}")
+            return {"error": str(e)}
+
+    def write_test_cases(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            prompt = f"""
+                Create  test cases for the following code in brief :
+                {state.get('generated_code', 'No code available')}
+                
+                Include positive, negative, edge, and security test cases.
+                Format the tests as a Markdown table with test steps and expected results.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "test_cases": content,
+                "current_result": content,
+                "current_step": "test_cases_review"  # Updated to next node
+            })
+            return {"test_cases": content}
+        except Exception as e:
+            self.logger.exception("Error in write_test_cases.")
+            st.error(f"Failed to write test cases: {e}")
+            return {"error": str(e)}
+
+    def test_cases_review(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            prompt = f"""
+                Please review the following test cases and provide your feedback.
+                
+                Test Cases:
+                {state.get('test_cases', 'No test cases generated')}
+                
+                If the test cases are acceptable, type "approve". Otherwise, provide detailed feedback.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "test_feedback": content,
+                "current_step": "fix_test_cases"  # Updated to next node
+            })
+            # st.session_state.user_decision = None
+            return {"test_feedback": content}
+        except Exception as e:
+            self.logger.exception("Error in test_cases_review.")
+            # st.error(f"Test cases review failed: {e}")
+            return {"error": str(e)}
         
-        response = self.llm.invoke(prompt)
-        st.session_state.state["generated_code"] = response.content 
-        return {"generated_code": response.content, "current_step": "code_review"}
+    def decision_test_cases_review(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            prompt = f"""
+                Based on the following test cases and the review feedback, please decide whether the test cases are acceptable or if further modifications are required.
+                
+                Test Cases:
+                {state.get('test_cases', 'No test cases available')}
+                
+                Review Feedback:
+                {state.get('test_feedback', 'No feedback provided')}
+                
+                If the test cases are acceptable, output "approved". Otherwise, provide the modifications required.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "decision_test_cases": content,
+                "current_result": content,
+                "current_step": "fix_test_cases"  # or update to the next step in your flow if needed
+            })
+            return {"decision_test_cases": content}
+        except Exception as e:
+            self.logger.exception("Error in decision_test_cases_review.")
+            st.error(f"Decision on test cases review failed: {e}")
+            return {"error": str(e)}
+
+
+    def fix_test_cases(self, state: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            prompt = f"""
+                Improve the test cases based on the following feedback:
+                
+                Original Test Cases: {state.get('test_cases', 'No test cases available')}
+                Feedback: {state.get('test_feedback', 'No feedback provided')}
+                
+                Ensure that all feedback points are addressed and that the Markdown table format is maintained.
+            """
+            response = self.llm.invoke(prompt)
+            content = getattr(response, 'content', None)
+            if content is None:
+                raise ValueError("LLM response does not contain content.")
+            self._update_state({
+                "test_cases": content,
+                "current_step": "completed"  # Final step
+            })
+            return {"test_cases": content}
+        except Exception as e:
+            self.logger.exception("Error in fix_test_cases.")
+            # st.error(f"Failed to fix test cases: {e}")
+            return {"error": str(e)}

src/langgraphagenticai/state/state.py

@@ -19,18 +19,53 @@ class PlannerState(TypedDict):
     start_date: str 
     end_date: str    
     
+# class SDLCState(TypedDict):
+#     current_step: Literal[
+#         "requirements",
+#         "generate_user_stories",
+#         "po_approval",
+#         "generate_code",
+#         "code_review",
+#         "completed"
+#     ]
+#     requirements: Optional[str]
+#     user_stories: Optional[str]
+#     po_feedback: Optional[str]
+#     generated_code: Optional[str]
+#     review_feedback: Optional[str]
+#     decision: Optional[Literal["approved", "feedback"]]
+    
+# from typing import TypedDict, Optional, Literal
+
 class SDLCState(TypedDict):
     current_step: Literal[
         "requirements",
         "generate_user_stories",
-        "po_approval",
+        "product_owner_review",
+        "create_design_docs",
+        "design_review",
         "generate_code",
         "code_review",
-        "completed"
+        "fix_code_after_code_review",
+        "security_review",
+        "fix_code_after_security",
+        "write_test_cases",
+        "test_cases_review",
+        "fix_test_cases",
+        "end"
     ]
+    decision: Optional[Literal["approved", "feedback"]]
+    # Content fields
     requirements: Optional[str]
     user_stories: Optional[str]
-    po_feedback: Optional[str]
+    design_docs: Optional[str]
     generated_code: Optional[str]
-    review_feedback: Optional[str]
-    decision: Optional[Literal["approved", "feedback"]]
+    test_cases: Optional[str]
+    # Feedback fields
+    po_feedback: Optional[str]          # Product Owner feedback
+    design_feedback: Optional[str]      # Design Review feedback
+    code_feedback: Optional[str]        # Code Review feedback
+    security_feedback: Optional[str]    # Security Review feedback
+    test_case_feedback: Optional[str]   # Test Case Review feedback
+    # For human input during interrupts.
+    human_decision: Optional[str]

src/langgraphagenticai/ui/streamlitui/display_result.py

@@ -1,7 +1,11 @@
+import uuid
 import streamlit as st
 from langchain_core.messages import HumanMessage,AIMessage,ToolMessage
 import json
 
+from langgraph.types import interrupt, Command
+
+from src.langgraphagenticai.ui.streamlitui.sdlcfeedback import SDLCUI
 from src.langgraphagenticai.tools.customtool import APPOINTMENTS
 from src.langgraphagenticai.tools.customer_support_tools import customers_database, data_protection_checks
 
@@ -11,7 +15,6 @@ class DisplayResultStreamlit:
         self.graph = graph
         self.user_message = user_message
     
-    
     def display_result_on_ui(self):
         usecase= self.usecase
         graph = self.graph
@@ -117,18 +120,105 @@ class DisplayResultStreamlit:
         elif usecase == "SDLC Workflow":
             initial_state = self.user_message
             # Invoke the workflow node for generating user stories
-            result = graph.invoke(initial_state)
-            if 'user_stories' in result:
-                with st.expander('generated userstories'):
-                    st.markdown(result["user_stories"])
-            if 'generated_code' in result:
-                with st.expander('generated code'):
-                    st.markdown(result['generated_code'])
-                   
-        # display graph
-        if graph:
-            st.write('state graph - workflow')
-            st.image(graph.get_graph(xray=True).draw_mermaid_png())
+            
+            if 'graph_config' not in st.session_state:
+                st.session_state.graph_config = {"configurable": {"thread_id": uuid.uuid4()}}
+           
+            # Create a placeholder to show output or interrupts.
+            output_placeholder = st.empty()
+                
+            ui = SDLCUI()
+            col1, col2 = st.columns(2)  # Create two columns
+            st.session_state.state['Final_Result'] = []
+            if 'Final_Result' not in st.session_state:
+                st.session_state['Final_Result'] = []
+            with col1:
+                st.subheader('Final Result')
+                
+            with col2:
+                st.subheader('human in loop : __interrupt__')
+            if  st.session_state.graph_stage =='initial' and 'current_step' in st.session_state.state and st.session_state.state['current_step']!='' :
+                graph_stream = graph.stream(st.session_state["state"], config=st.session_state.graph_config )
+                if graph_stream:
+                    
+                    for event in graph_stream:
+                            with col1:
+                                if "__interrupt__" not in event:
+                                    for d in event.values():
+                                        st.session_state.state["Final_Result"].append(d)
+                                        if d:
+                                            for key, value in d.items():
+                                                with st.expander(label=key):
+                                                    st.markdown(value)
+                            with col2:
+                                if "__interrupt__" in event:
+                                    st.session_state.graph_stage = "waiting"
+                                    st.rerun()
+                                    break
+                else:
+                    st.session_state.graph_stage = "finished"
+            # --- Stage 2: Display Human Input UI ---
+            if st.session_state.graph_stage == "waiting":
+                with col2:
+                    col2_1, col2_2 = st.columns(2)
+                    st.info(f"Current Steps : {st.session_state.state['current_step']}")
+
+                    feedback = st.text_area("Feedback (enter text to reject)", key="feedback_input")
+                    with col2_1:
+                        if st.button("✅ Approve"):
+                            st.session_state.user_decision = "approve"
+                            st.session_state.graph_stage = "resumed"
+                            st.rerun()
+                    with col2_2:
+                        if st.button("📝 Request Change"):
+                            st.session_state.user_decision = feedback if feedback else "reject"
+                            st.session_state.graph_stage = "resumed"
+                            st.rerun()
+                            
+                    if st.session_state.graph_stage == "waiting":
+                        st.stop()
+
+            # --- Stage 3: Resume Graph Execution ---
+            # When resuming after interrupt
+            if st.session_state.graph_stage == "resumed":
+                resume_state = {
+                    "human_decision": st.session_state.user_decision,
+                }
+                for event in graph.stream(
+                    st.session_state["state"],
+                    config=st.session_state.thread_config
+                ):
+                    with col1:
+                        if "__interrupt__" not in event:
+                            for d in event.values():
+                                st.session_state.state["Final_Result"].append(d)
+                                if d:
+                                    for key, value in d.items():
+                                        with st.expander(label=key):
+                                            st.markdown(value)
+                    with col2:
+                        if "__interrupt__" in event:
+                            st.session_state.graph_stage = "waiting"
+                            st.rerun()
+                            break
+                                
+                        
+                # Determine if the workflow has reached the finish point (fix_test_cases node reached).
+                if st.session_state.state['current_step']=='completed' and any(isinstance(event, dict) and event.get("fix_test_cases") ):
+                    st.session_state.graph_stage = "finished"
+                    ui.render_end(state=st.session_state.state)
+                else:
+                    st.session_state.graph_stage = "initial"
+                st.rerun()
+
+            # --- Stage 4: Workflow Finished ---
+            if st.session_state.graph_stage == "finished":
+                st.write("### Workflow Complete")
+                ui.render_end(state=st.session_state.state)
+                if graph:
+                    st.write('state graph - workflow')
+                    st.image(graph.get_graph(xray=True).draw_mermaid_png())
+            
             
     def _display_travel_planner_results(self):
         # Extract travel parameters from message
@@ -174,7 +264,6 @@ class DisplayResultStreamlit:
             # Display destination and dates
             st.markdown(f"{sections['Destination']}")
            
-
     def _display_tool_calls(self, message):
         """
         Displays details of tool calls made during the itinerary generation.

src/langgraphagenticai/ui/streamlitui/loadui.py

@@ -1,7 +1,9 @@
+import uuid
 import streamlit as st
 import os
 from datetime import date
 
+from src.langgraphagenticai.ui.streamlitui.sdlcfeedback import SDLCUI
 from src.langgraphagenticai.ui.uiconfigfile import Config
 from langchain_core.messages import AIMessage, HumanMessage
 
@@ -14,14 +16,20 @@ class LoadStreamlitUI:
     def initialize_session(self):
         return {
             "current_step": "requirements",
+            "decision": None,
             "requirements": "",
             "user_stories": "",
+            "design_docs": "",
+            "code": "",
+            "test_cases": "",
             "po_feedback": "",
-            "generated_code": "",
+            "design_feedback":"",
+            'generate_code':"",
             "review_feedback": "",
-            "decision": None
+            "security_feedback": "",
+            "test_feedback": "",
+            "human_decision": ""
         }
-
     def render_requirements(self):
         st.markdown("## 📝 Requirements Submission")
         st.session_state.state["requirements"] = st.text_area(
@@ -40,6 +48,8 @@ class LoadStreamlitUI:
         st.session_state.timeframe = ''
         st.session_state.IsFetchButtonClicked = False
         st.session_state.IsSDLC = False
+        ui = SDLCUI()
+        
         
         
 
@@ -73,6 +83,9 @@ class LoadStreamlitUI:
                 # Validate API key
                 if not self.user_controls["TAVILY_API_KEY"]:
                     st.warning("⚠️ Please enter your TAVILY_API_KEY key to proceed. Don't have? refer : https://app.tavily.com/home")
+        
+        if self.user_controls['selected_usecase']!="SDLC Workflow":
+            st.session_state['state'] = ''
         if self.user_controls['selected_usecase'] == "Appointment Receptionist":
             col1, col2 = st.columns(2)
             with col1:
@@ -144,11 +157,37 @@ class LoadStreamlitUI:
         # Added for SDLC Workflow
         elif self.user_controls['selected_usecase']=="SDLC Workflow":
             st.subheader(" SDLC Workflow ")
+            # if 'status' in st.session_state.state and st.session_state.state['status']== "__interrupt__" and 'decision' in st.session_state.state and st.session_state.state['decision']==None :
+            #     ui.render()
+            if "requirements" in st.session_state.state and  st.session_state.state['current_step']=="requirements":
+                ui.render_requirements(st.session_state.state)
+            # if st.session_state.state !='':
+                
+            #     ui.render()
+                
+                
             
-            
-            if "state" not in st.session_state:
+            if st.button("start workflow"):
+                st.session_state.state = ''
                 st.session_state.state = self.initialize_session()
-            self.render_requirements()
+                # Initialize session state variables.
+                if "thread_config" not in st.session_state:
+                    st.session_state.thread_config = {"configurable": {"thread_id": uuid.uuid4()}}
+                if "graph_stage" not in st.session_state:
+                    st.session_state.graph_stage = "initial"  # Stages: initial, waiting, resumed, finished.
+                if "output_chunks" not in st.session_state:
+                    st.session_state.output_chunks = []
+                if "user_decision" not in st.session_state:
+                    st.session_state.user_decision = None
+                st.session_state.IsSDLC = True
+                ui.render_requirements(st.session_state.state)
+                
+                # if st.session_state.state['current_step']=="requirements":
+                    # self.render_requirements()
+            if 'requirements' in st.session_state.state and  st.session_state.state['requirements']=='':
+                    st.stop()   
+            
+                
             
         
         return self.user_controls

src/langgraphagenticai/ui/streamlitui/sdlcfeedback.py

@@ -0,0 +1,234 @@
+# sdlc_feedback.py
+import streamlit as st
+
+class SDLCUI:
+    def render(self):
+        if "state" not in st.session_state:
+            st.session_state.state = {"current_step": "user_input"}
+        state = st.session_state.state
+
+        st.markdown(f"**Current Step:** {state['current_step'].replace('_', ' ').title()}")
+
+        step_method = getattr(self, f"render_{state['current_step']}", None)
+        if step_method:
+            step_method(state)
+        else:
+            st.error(f"Unknown step: {state['current_step']}")
+
+    def render_requirements(self, state):
+        st.markdown("### Requirements Input")
+        requirements = st.text_area("Enter your requirements:", height=200)
+        if st.button("Submit Requirements"):
+            state["requirements"] = requirements
+            state["current_step"] = "generate_user_stories"
+            st.session_state.state["current_step"] = "generate_user_stories"
+       
+            
+            
+
+    def render_generate_user_stories(self, state):
+        st.markdown("### Generated User Stories")
+        if st.session_state.state['user_stories']!='':
+            with st.expander("View User Stories"):
+                st.markdown(st.session_state.state['user_stories'])
+            if st.button("Continue to Product Owner Review"):
+                state["current_step"] = "product_owner_review"
+                st.session_state.state["current_step"] = "product_owner_review"
+            # st.rerun()
+
+    def render_product_owner_review(self, state):
+        st.markdown("### Product Owner Review")
+        with st.expander("View Current User Stories"):
+            st.markdown(state.get("user_stories", ""))
+        st.markdown("### Review Actions")
+        feedback = st.text_area("Feedback:", height=150, key='Feedback')
+        
+        col1, col2 = st.columns(2)
+        with col1:
+            if st.button("🚀 Approve Design Phase", key="approve_design"):
+                state["decision"] = "approved"
+                state["current_step"] = "create_design_docs"
+                st.session_state.user_decision = "approve"
+                st.session_state.graph_stage = "resumed"
+                st.rerun()
+        with col2:
+            if st.button("🔧 Request Revisions", key="request_revisions"):
+                state["decision"] = "feedback"
+                state["po_feedback"] = feedback
+                state["current_step"] = "revise_user_stories"
+                st.rerun()
+
+    def render_create_design_docs(self, state):
+        st.markdown("### Design Documents")
+        with st.expander("View Design Documents"):
+            st.markdown(state.get("design_docs", ""))
+        if st.button("Proceed to Design Review"):
+            state["current_step"] = "design_review"
+            st.rerun()
+
+    def render_design_review(self, state):
+        st.markdown("### Design Review")
+        with st.expander("View Design Documents"):
+            st.markdown(state.get("design_docs", ""))
+        st.markdown("### Review Actions")
+        feedback = st.text_area("Design Feedback:", height=150, key="design_feedback")
+        
+        col1, col2 = st.columns(2)
+        with col1:
+            if st.button("🚀 Approve Implementation", key="approve_implementation"):
+                state["decision"] = "approved"
+                state["current_step"] = "generate_code"
+                st.session_state.user_decision = "approve"
+                st.session_state.graph_stage = "resumed"
+                st.rerun()
+        with col2:
+            if st.button("🔧 Request Design Changes", key="request_design_changes"):
+                state["decision"] = "feedback"
+                state["design_feedback"] = feedback
+                state["current_step"] = "create_design_docs"
+                st.rerun()
+
+    def render_generate_code(self, state):
+        st.markdown("### Generated Code")
+        with st.expander("View Code Implementation"):
+            st.code(state.get("code", ""), language='python')
+        if st.button("Proceed to Code Review"):
+            state["current_step"] = "code_review"
+            st.rerun()
+
+    def render_code_review(self, state):
+        st.markdown("### Code Review")
+        with st.expander("View Current Code"):
+            st.code(state.get("code", ""), language='python')
+        st.markdown("### Review Actions")
+        feedback = st.text_area("Code Feedback:", height=150, key="code_feedback")
+        
+        col1, col2 = st.columns(2)
+        with col1:
+            if st.button("🚀 Approve for Security Review", key="approve_security"):
+                state["decision"] = "approved"
+                state["current_step"] = "security_review"
+                st.session_state.user_decision = "approve"
+                st.session_state.graph_stage = "resumed"
+                st.rerun()
+        with col2:
+            if st.button("🔧 Request Code Fixes", key="request_code_fixes"):
+                state["decision"] = "feedback"
+                state["review_feedback"] = feedback
+                state["current_step"] = "fix_code_after_code_review"
+                st.rerun()
+
+    def render_security_review(self, state):
+        st.markdown("### Security Review")
+        with st.expander("View Code for Security Audit"):
+            st.code(state.get("code", ""), language='python')
+        st.markdown("### Security Findings")
+        feedback = st.text_area("Security Feedback:", height=150, key="security_feedback")
+        
+        col1, col2 = st.columns(2)
+        with col1:
+            if st.button("🚀 Approve for Testing", key="approve_testing"):
+                state["decision"] = "approved"
+                state["current_step"] = "write_test_cases"
+                st.session_state.user_decision = "approve"
+                st.session_state.graph_stage = "resumed"
+                st.rerun()
+        with col2:
+            if st.button("🔧 Request Security Fixes", key="request_security_fixes"):
+                state["decision"] = "feedback"
+                state["security_feedback"] = feedback
+                state["current_step"] = "fix_code_after_security"
+                st.rerun()
+
+    def render_write_test_cases(self, state):
+        st.markdown("### Test Cases")
+        with st.expander("View Test Cases"):
+            st.markdown(state.get("test_cases", ""))
+        if st.button("Proceed to Test Review"):
+            state["current_step"] = "test_cases_review"
+            st.rerun()
+
+    def render_test_cases_review(self, state):
+        st.markdown("### Test Cases Review")
+        with st.expander("View Current Test Cases"):
+            st.markdown(state.get("test_cases", ""))
+        st.markdown("### Review Actions")
+        feedback = st.text_area("Test Feedback:", height=150, key="test_feedback")
+        
+        col1, col2 = st.columns(2)
+        with col1:
+            if st.button("🚀 Final Approval", key="final_approve"):
+                state["decision"] = "approved"
+                state["current_step"] = "end"
+                st.rerun()
+        with col2:
+            if st.button("🔧 Improve Test Cases", key="request_test_fixes"):
+                state["decision"] = "feedback"
+                state["test_feedback"] = feedback
+                state["current_step"] = "fix_test_cases"
+                st.rerun()
+
+    def render_end(self, state):
+        st.success("✅ SDLC Process Completed Successfully!")
+        with st.expander("Requirements"):
+            st.write(state.get("requirements", ""))
+        with st.expander("Decision"):
+            st.write(state.get("decision", ""))
+        with st.expander("User Stories"):
+            st.write(state.get("user_stories", ""))
+        with st.expander("Design Documents"):
+            st.write(state.get("design_docs", ""))
+        with st.expander("Design Feedback"):
+            st.write(state.get("design_feedback", ""))
+        with st.expander("Generated Code"):
+            st.write(state.get("generate_code", ""))
+        with st.expander("Test Cases"):
+            st.write(state.get("test_cases", ""))
+        with st.expander("PO Feedback"):
+            st.write(state.get("po_feedback", ""))
+        with st.expander("Review Feedback"):
+            st.write(state.get("review_feedback", ""))
+        with st.expander("Security Feedback"):
+            st.write(state.get("security_feedback", ""))
+        with st.expander("Test Feedback"):
+            st.write(state.get("test_feedback", ""))
+                
+            
+            
+        with st.expander("Final Artifacts", expanded=True):
+            st.markdown("### Download Assets")
+            st.download_button("📥 Requirements", 
+                            data=state.get("requirements", ""), 
+                            file_name="requirements.md")
+            st.download_button("📥 User Stories", 
+                            data=state.get("user_stories", ""), 
+                            file_name="user_stories.md")
+            st.download_button("📥 Design Documents", 
+                            data=state.get("design_docs", ""), 
+                            file_name="design_docs.md")
+            st.download_button("📥 Generated Code", 
+                            data=state.get("code", ""), 
+                            file_name="generated_code.md")
+            st.download_button("📥 Test Cases", 
+                            data=state.get("test_cases", ""), 
+                            file_name="test_cases.md")
+            st.download_button("📥 Product Owner Feedback", 
+                            data=state.get("po_feedback", ""), 
+                            file_name="po_feedback.md")
+            st.download_button("📥 Design Feedback", 
+                            data=state.get("design_feedback", ""), 
+                            file_name="design_feedback.md")
+            st.download_button("📥 Code Review Feedback", 
+                            data=state.get("review_feedback", ""), 
+                            file_name="code_review_feedback.md")
+            st.download_button("📥 Security Feedback", 
+                            data=state.get("security_feedback", ""), 
+                            file_name="security_feedback.md")
+            st.download_button("📥 Test Cases Feedback", 
+                            data=state.get("test_feedback", ""), 
+                            file_name="test_feedback.md")
+
+                
+        if st.button("🔄 Restart Process"):
+            st.session_state.clear()
+            st.rerun()