Merge pull request #437 from microsoft/macae-v3-dev-marktayl

feat: Macae v3 dev marktayl

Author: blessing-sanusi

src/backend/v3/api/router.py

@@ -306,26 +306,48 @@ async def plan_approval(human_feedback: messages.PlanApprovalResponse, request:
             status_code=401, detail="Missing or invalid user information"
         )
     # Set the approval in the orchestration config
-    if user_id and human_feedback.plan_id:
-        if orchestration_config and human_feedback.plan_id in orchestration_config.approvals:
-            orchestration_config.approvals[human_feedback.plan_id] = human_feedback.approved
+    if user_id and human_feedback.plan_dot_id:
+        if orchestration_config and human_feedback.plan_dot_id in orchestration_config.approvals:
+            orchestration_config.approvals[human_feedback.plan_dot_id] = human_feedback.approved
             track_event_if_configured(
                 "PlanApprovalReceived",
                 {
-                    "plan_id": human_feedback.plan_id,
+                    "plan_id": human_feedback.plan_dot_id,
                     "approved": human_feedback.approved,
                     "user_id": user_id,
                     "feedback": human_feedback.feedback
                 },
             )
             return {"status": "approval recorded"}
         else:
-            logging.warning(f"No orchestration or plan found for plan_id: {human_feedback.plan_id}")
+            logging.warning(f"No orchestration or plan found for plan_id: {human_feedback.plan_dot_id}")
             raise HTTPException(status_code=404, detail="No active plan found for approval")
 
 @app_v3.post("/user_clarification")
 async def user_clarification(human_feedback: messages.UserClarificationResponse, request: Request):
-    pass
+    """ Endpoint to receive plan approval or rejection from the user. """
+    authenticated_user = get_authenticated_user_details(request_headers=request.headers)
+    user_id = authenticated_user["user_principal_id"]
+    if not user_id:
+        raise HTTPException(
+            status_code=401, detail="Missing or invalid user information"
+        )
+    # Set the approval in the orchestration config
+    if user_id and human_feedback.request_id:
+        if orchestration_config and human_feedback.request_id in orchestration_config.clarifications:
+            orchestration_config.clarifications[human_feedback.request_id] = human_feedback.answer
+            track_event_if_configured(
+                "PlanApprovalReceived",
+                {
+                    "request_id": human_feedback.request_id,
+                    "answer": human_feedback.answer,
+                    "user_id": user_id,
+                },
+            )
+            return {"status": "clarification recorded"}
+        else:
+            logging.warning(f"No orchestration or plan found for request_id: {human_feedback.request_id}")
+            raise HTTPException(status_code=404, detail="No active plan found for clarification")
 
 
 @app_v3.post("/upload_team_config")

src/backend/v3/config/settings.py

@@ -7,7 +7,6 @@
 import contextvars
 import json
 import logging
-import uuid
 from typing import Dict, Optional
 
 from common.config.app_config import config
@@ -76,10 +75,11 @@ class OrchestrationConfig:
     """Configuration for orchestration settings."""
 
     def __init__(self):
-        self.orchestrations = {}
-        self.plans = {}       # job_id -> current plan
-        self.approvals = {}   # job_id -> True/False/None
-        self.sockets = {}     # job_id -> WebSocket
+        self.orchestrations: Dict[str, MagenticOrchestration] = {} # user_id -> orchestration instance
+        self.plans: Dict[str, any] = {} # plan_id -> plan details
+        self.approvals: Dict[str, bool] = {} # plan_dot_id -> approval status
+        self.sockets: Dict[str, WebSocket] = {} # user_id -> WebSocket
+        self.clarifications: Dict[str, str] = {} # plan_dot_id -> clarification response
 
     def get_current_orchestration(self, user_id: str) -> MagenticOrchestration:
         """get existing orchestration instance."""

src/backend/v3/magentic_agents/proxy_agent.py

@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft. All rights reserved.
 """ Proxy agent that prompts for human clarification."""
 
+import asyncio
 import logging
 import uuid
 from collections.abc import AsyncIterable
@@ -20,7 +21,8 @@
 from typing_extensions import override
 from v3.callbacks.response_handlers import (agent_response_callback,
                                             streaming_agent_response_callback)
-from v3.config.settings import current_user_id
+from v3.config.settings import (connection_config, current_user_id,
+                                orchestration_config)
 from v3.models.messages import (UserClarificationRequest,
                                 UserClarificationResponse)
 
@@ -144,24 +146,30 @@ async def invoke(self, message: str,*, thread: AgentThread | None = None,**kwarg
             construct_thread=lambda: DummyAgentThread(),
             expected_type=DummyAgentThread,
         )
-        # Send clarification request via response handlers
+
+        # Send clarification request via streaming callbacks
         clarification_request = f"I need clarification about: {message}"
+
+        clarification_message = UserClarificationRequest(
+            question=clarification_request,
+            request_id=str(uuid.uuid4())  # Unique ID for the request
+        )
+
+        # Send the approval request to the user's WebSocket
+        await connection_config.send_status_update_async({
+            "type": "user_clarification_request", 
+            "data": clarification_message
+        })
 
-        clarification_message = self._create_message_content(clarification_request, thread.id)
-        await self._trigger_response_callbacks(clarification_message)
-        
-        # Get human input - replace this with awaiting a websocket call or API handler when available
-        human_response = input("Please provide clarification: ").strip()
+        # Get human input
+        human_response = await self._wait_for_user_clarification(clarification_message.request_id)
 
         if not human_response:
             human_response = "No additional clarification provided."
 
         response = f"Human clarification: {human_response}"
 
-        # Send response via response handlers
-        response_message = self._create_message_content(response, thread.id)
-
-        chat_message = response_message
+        chat_message = self._create_message_content(response, thread.id)
 
         yield AgentResponseItem(
             message=chat_message,
@@ -188,11 +196,23 @@ async def invoke_stream(self, messages, thread=None, **kwargs) -> AsyncIterator[
 
         # Send clarification request via streaming callbacks
         clarification_request = f"I need clarification about: {message}"
-        self._create_message_content(clarification_request, thread.id)
-        await self._trigger_streaming_callbacks(clarification_request)
+        #self._create_message_content(clarification_request, thread.id)
+        # await self._trigger_streaming_callbacks(clarification_request)
+
+        clarification_message = UserClarificationRequest(
+            question=clarification_request,
+            request_id=str(uuid.uuid4())  # Unique ID for the request
+        )
+
+        # Send the approval request to the user's WebSocket
+        # The user_id will be automatically retrieved from context
+        await connection_config.send_status_update_async({
+            "type": "user_clarification_request", 
+            "data": clarification_message
+        })
 
         # Get human input - replace with websocket call when available
-        human_response = input("Please provide clarification: ").strip()
+        human_response = await self._wait_for_user_clarification(clarification_message.request_id)
 
         if not human_response:
             human_response = "No additional clarification provided."
@@ -206,6 +226,15 @@ async def invoke_stream(self, messages, thread=None, **kwargs) -> AsyncIterator[
             thread=thread
         )
 
+    async def _wait_for_user_clarification(self, request_id:str) -> Optional[UserClarificationResponse]: 
+        """Wait for user clarification response."""
+        # To do: implement timeout and error handling
+        if request_id not in orchestration_config.clarifications:
+            orchestration_config.clarifications[request_id] = None
+        while orchestration_config.clarifications[request_id] is None:
+            await asyncio.sleep(0.2)
+        return UserClarificationResponse(request_id=request_id,answer=orchestration_config.clarifications[request_id])
+    
     async def get_response(self, chat_history, **kwargs):
         """Get response from the agent - required by Agent base class."""
         # Extract the latest user message
@@ -220,6 +249,7 @@ async def get_response(self, chat_history, **kwargs):
             role=AuthorRole.ASSISTANT,
             content="No clarification provided."
         )
+    
 
 async def create_proxy_agent(user_id: str = None):
     """Factory function for human proxy agent."""

src/backend/v3/models/messages.py

@@ -1,5 +1,6 @@
 """Messages from the backend to the frontend via WebSocket."""
 
+import uuid
 from dataclasses import dataclass
 from typing import Any, Dict, List, Literal, Optional
 
@@ -48,7 +49,7 @@ class PlanApprovalRequest:
 @dataclass(slots=True)
 class PlanApprovalResponse:
     """Response for plan approval from the frontend."""
-    plan_id: str
+    plan_dot_id: str
     approved: bool
     feedback: str | None = None
 
@@ -70,20 +71,19 @@ class ReplanApprovalResponse:
 class UserClarificationRequest:
     """Request for user clarification from the frontend."""
     question: str
-    context: dict | None = None
+    request_id: str
 
 @dataclass(slots=True)
 class UserClarificationResponse:
     """Response for user clarification from the frontend."""
-    plan_id: str | None
+    request_id: str
     answer: str = ""
 
 @dataclass(slots=True)
 class FinalResultMessage:
     """Final result message from the backend to the frontend."""
     result: str
     summary: str | None = None
-    context: dict | None = None
 
 @dataclass(slots=True)
 class HumanFeedback(KernelBaseModel):

src/backend/v3/orchestration/human_approval_manager.py

@@ -67,6 +67,8 @@ async def plan(self, magentic_context: MagenticContext) -> Any:
         plan = await super().plan(magentic_context)
         self.magentic_plan = self.plan_to_obj( magentic_context, self.task_ledger)
 
+        self.magentic_plan.user_id = current_user_id.get()
+
         # Request approval from the user before executing the plan
         approval_message = messages.PlanApprovalRequest(
             plan=self.magentic_plan,
@@ -76,9 +78,6 @@ async def plan(self, magentic_context: MagenticContext) -> Any:
                 "participant_descriptions": magentic_context.participant_descriptions
             } if hasattr(magentic_context, 'participant_descriptions') else {}
         )
-
-        # Send the current plan to the frontend via WebSocket
-        #await connection_config.send_status_update_async(approval_message,)
 
         # Send the approval request to the user's WebSocket
         # The user_id will be automatically retrieved from context
@@ -87,8 +86,8 @@ async def plan(self, magentic_context: MagenticContext) -> Any:
             "data": approval_message
         })
 
-        # Wait for user approval (you'll need to implement this)
-        approval_response = await self._wait_for_user_approval()
+        # Wait for user approval
+        approval_response = await self._wait_for_user_approval(approval_message.plan.id)
 
         if approval_response and approval_response.approved:
             print("Plan approved - proceeding with execution...")
@@ -106,23 +105,15 @@ async def plan(self, magentic_context: MagenticContext) -> Any:
             # )
 
 
-    async def _wait_for_user_approval(self, plan_id: Optional[str] = None) -> Optional[messages.PlanApprovalResponse]: # plan_id will not be optional in future
+    async def _wait_for_user_approval(self, plan_dot_id: Optional[str] = None) -> Optional[messages.PlanApprovalResponse]: # plan_id will not be optional in future
         """Wait for user approval response."""
-        # Temporarily use console input for approval - will switch to WebSocket or API in future
-        # response = input("\nApprove this execution plan? [y/n]: ").strip().lower()
-        # if response in ['y', 'yes']:
-        #     return messages.PlanApprovalResponse(approved=True, plan_id=plan_id if plan_id else "input")
-        # elif response in ['n', 'no']:
-        #     return messages.PlanApprovalResponse(approved=False, plan_id=plan_id if plan_id else "input")
-        # else:
-        #     print("Invalid input. Please enter 'y' for yes or 'n' for no.")
-        #     return await self._wait_for_user_approval()
-        # In future, implement actual waiting for WebSocket or API response here
-        if plan_id not in orchestration_config.approvals:
-            orchestration_config.approvals[plan_id] = None
-        while orchestration_config.approvals[plan_id] is None:
+        
+        # To do: implement timeout and error handling
+        if plan_dot_id not in orchestration_config.approvals:
+            orchestration_config.approvals[plan_dot_id] = None
+        while orchestration_config.approvals[plan_dot_id] is None:
             await asyncio.sleep(0.2)
-        return messages.PlanApprovalResponse(approved=orchestration_config.approvals[plan_id], plan_id=plan_id)
+        return messages.PlanApprovalResponse(approved=orchestration_config.approvals[plan_dot_id], plan_dot_id=plan_dot_id)
 
 
     async def prepare_final_answer(self, magentic_context: MagenticContext) -> ChatMessageContent: