fix (tasks): removed context verification step before planning

Kabeer2004 · Kabeer2004 · commit 67914ad632e4 · 2025-08-17T16:18:05.000+05:30
diff --git a/src/client/components/tasks/TaskDetailsContent.js b/src/client/components/tasks/TaskDetailsContent.js
@@ -394,11 +394,6 @@ const TaskDetailsContent = ({
 		<div className="space-y-6">
 			{/* --- SWARM DETAILS (if applicable) --- */}
 			{displayTask.task_type === "swarm" && (
-				<SwarmDetailsSection swarmDetails={displayTask.swarm_details} />
-			)}
-
-			{/* --- Researched Context (if planning) --- */}
-			{displayTask.status === "planning" && displayTask.found_context && (
 				<div>
 					<label className="text-sm font-medium text-neutral-400 block mb-2">
 						Researched Context
@@ -465,20 +460,6 @@ const TaskDetailsContent = ({
 				</div>
 			</div>
 
-			{/* --- Researched Context (if planning) --- */}
-			{displayTask.status === "planning" && displayTask.found_context && (
-				<div>
-					<label className="text-sm font-medium text-neutral-400 block mb-2">
-						Researched Context
-					</label>
-					<div className="bg-neutral-800/50 p-3 rounded-lg text-sm text-neutral-300 whitespace-pre-wrap border border-neutral-700/50">
-						<ReactMarkdown className="prose prose-sm prose-invert">
-							{displayTask.found_context}
-						</ReactMarkdown>
-					</div>
-				</div>
-			)}
-
 			{/* --- DESCRIPTION --- */}
 			<div>
 				<label className="text-sm font-medium text-neutral-400 block mb-2">
diff --git a/src/server/main/db.py b/src/server/main/db.py
@@ -386,8 +386,7 @@ async def add_task(self, user_id: str, task_data: dict) -> str:
             "task_type": task_data.get("task_type", "single"),
             "swarm_details": task_data.get("swarm_details") # Will be None for single tasks
         }
-
-        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "found_context", "clarifying_questions", "result", "swarm_details"]
+        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "clarifying_questions", "result", "swarm_details"]
         _encrypt_doc(task_doc, SENSITIVE_TASK_FIELDS)
 
         await self.task_collection.insert_one(task_doc)
@@ -397,22 +396,22 @@ async def add_task(self, user_id: str, task_data: dict) -> str:
     async def get_task(self, task_id: str, user_id: str) -> Optional[Dict]:
         """Fetches a single task by its ID, ensuring it belongs to the user."""
         doc = await self.task_collection.find_one({"task_id": task_id, "user_id": user_id})
-        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "found_context", "clarifying_questions", "result", "swarm_details"]
+        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "clarifying_questions", "result", "swarm_details"]
         _decrypt_doc(doc, SENSITIVE_TASK_FIELDS)
         return doc
 
     async def get_all_tasks_for_user(self, user_id: str) -> List[Dict]:
         """Fetches all tasks for a given user."""
         cursor = self.task_collection.find({"user_id": user_id}).sort("created_at", -1)
         docs = await cursor.to_list(length=None)
-        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "found_context", "clarifying_questions", "result", "swarm_details"]
+        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "clarifying_questions", "result", "swarm_details"]
         _decrypt_docs(docs, SENSITIVE_TASK_FIELDS)
         return docs
 
     async def update_task(self, task_id: str, updates: Dict) -> bool:
         """Updates an existing task document."""
         updates["updated_at"] = datetime.datetime.now(datetime.timezone.utc)
-        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "found_context", "clarifying_questions", "result", "swarm_details"]
+        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "clarifying_questions", "result", "swarm_details"]
         _encrypt_doc(updates, SENSITIVE_TASK_FIELDS)
         result = await self.task_collection.update_one(
             {"task_id": task_id},
diff --git a/src/server/workers/executor/tasks.py b/src/server/workers/executor/tasks.py
@@ -266,14 +266,12 @@ async def async_execute_task_plan(task_id: str, user_id: str, run_id: str):
 
     plan_description = task.get("name", "Unnamed plan")
     original_context_str = json.dumps(original_context_data, indent=2, default=str) if original_context_data else "No original context provided."
-    found_context_str = task.get("found_context", "No additional context was found by the research agent.")
     block_id_prompt = f"The block_id for this task is '{block_id}'. You MUST pass this ID to the 'update_progress' tool in the 'block_id' parameter." if block_id else "This task did not originate from a tasks block."
 
     full_plan_prompt = (
         f"You are Sentient, a resourceful and autonomous executor agent. Your goal is to complete the user's request by intelligently following the provided plan.\n\n" # noqa
         f"**User Context:**\n- **User's Name:** {user_name}\n- **User's Location:** {user_location}\n- **Current Date & Time:** {current_user_time}\n\n"
         f"{trigger_event_prompt_section}"
-        f"**Retrieved Context (from research agent):**\n{found_context_str}\n\n"
         f"Your task ID is '{task_id}' and the current run ID is '{run_id}'.\n\n"
         f"The original context that triggered this plan is:\n---BEGIN CONTEXT---\n{original_context_str}\n---END CONTEXT---\n\n"
         f"**Primary Objective:** '{plan_description}'\n\n"
@@ -292,41 +290,91 @@ async def async_execute_task_plan(task_id: str, user_id: str, run_id: str):
     )
     
     try:
-        initial_messages = [{'role': 'user', 'content': "Begin executing the plan. Follow your instructions meticulously."}]
-        
         logger.info(f"Task {task_id}: Starting agent run.")
+        initial_messages = [{'role': 'user', 'content': "Begin executing the plan. Follow your instructions meticulously."}]
 
-        final_assistant_content = ""
-        final_history = []
-        for current_history in run_main_agent(
+        # Step 1: Fully exhaust the generator to let the agent run to completion.
+        # This prevents async UI updates from interfering with the agent's internal loop.
+        agent_generator = run_main_agent(
             system_message=full_plan_prompt,
             function_list=tools_config,
             messages=initial_messages
-        ):
-            final_history = current_history
-
-        # After the loop, process the complete response
-        if final_history and final_history[-1].get("role") == "assistant":
-            final_assistant_content = final_history[-1].get("content", "")
-
-        final_answer_content = ""
-        if final_assistant_content:
-            parsed_updates = parse_agent_string_to_updates(final_assistant_content)
-            for update in parsed_updates:
-                # Per user request, do not push thoughts to the log, but parse everything else.
-                if update.get("type") == "thought":
-                    continue
-                await add_progress_update(db, task_id, run_id, user_id, update, block_id)
-                if update.get("type") == "final_answer":
-                    final_answer_content = update.get("content")
+        )
+        all_history_steps = list(agent_generator)
+
+        if not all_history_steps:
+            raise Exception("Agent run produced no history, indicating an immediate failure.")
+
+        # Step 2: After completion, process all history steps to send UI updates.
+        last_history_len = len(initial_messages)
+        for current_history in all_history_steps:
+            new_messages = current_history[last_history_len:]
+            for msg in new_messages:
+                updates_to_push = []
+                if msg.get("role") == "assistant":
+                    if msg.get("content"):
+                        updates_to_push.extend(parse_agent_string_to_updates(msg["content"]))
+                    if msg.get("function_call"):
+                        fc = msg["function_call"]
+                        params_str = fc.get("arguments", "{}")
+                        params = JsonExtractor.extract_valid_json(params_str) or {"raw_parameters": params_str}
+                        updates_to_push.append({
+                            "type": "tool_call",
+                            "tool_name": fc.get("name"),
+                            "parameters": params
+                        })
+                elif msg.get("role") == "function":
+                    result_str = msg.get("content", "{}")
+                    result_content = JsonExtractor.extract_valid_json(result_str) or {"raw_result": result_str}
+                    is_error = isinstance(result_content, dict) and result_content.get("status") == "failure"
+                    updates_to_push.append({
+                        "type": "tool_result",
+                        "tool_name": msg.get("name"),
+                        "result": result_content.get("result", result_content.get("error", result_content)),
+                        "is_error": is_error
+                    })
+                
+                for update in updates_to_push:
+                    if update.get("type") != "thought":
+                        asyncio.create_task(add_progress_update(db, task_id, run_id, user_id, update, block_id))
+            last_history_len = len(current_history)
+
+        # Step 3: Check the final state for a valid answer.
+        final_history = all_history_steps[-1]
+        final_assistant_message = next((msg for msg in reversed(final_history) if msg.get("role") == "assistant"), None)
+        
+        has_final_answer = False
+        if final_assistant_message and final_assistant_message.get("content"):
+            parsed_updates = parse_agent_string_to_updates(final_assistant_message["content"])
+            if any(upd.get("type") == "final_answer" for upd in parsed_updates):
+                has_final_answer = True
+
+        if not has_final_answer:
+            error_message = "Agent finished execution without providing a final answer as required by its instructions. The task may be incomplete."
+            logger.error(f"Task {task_id}: {error_message}. Final history: {final_history}")
+            await add_progress_update(db, task_id, run_id, user_id, {"type": "error", "content": error_message}, block_id=block_id)
+            await update_task_run_status(db, task_id, run_id, "error", user_id, details={"error": error_message}, block_id=block_id)
+            
+            from workers.tasks import calculate_next_run
+            schedule_type = task.get('schedule', {}).get('type')
+            if schedule_type == 'recurring':
+                next_run_time, _ = calculate_next_run(task['schedule'], last_run=datetime.datetime.now(datetime.timezone.utc))
+                await db.tasks.update_one({"_id": task["_id"]}, {"$set": {"status": "active", "next_execution_at": next_run_time}})
+            elif schedule_type == 'triggered':
+                await db.tasks.update_one({"_id": task["_id"]}, {"$set": {"status": "active", "next_execution_at": None}})
+            else:
+                await db.tasks.update_one({"_id": task["_id"]}, {"$set": {"status": "error", "next_execution_at": None}})
+            return {"status": "error", "message": error_message}
 
+        # If we have a final answer, the execution was successful.
         logger.info(f"Task {task_id} execution phase completed. Dispatching to result generator.")
         await add_progress_update(db, task_id, run_id, user_id, {"type": "info", "content": "Execution finished. Generating final report..."}, block_id=block_id)
         await update_task_run_status(db, task_id, run_id, "completed", user_id, block_id=block_id)
         capture_event(user_id, "task_execution_succeeded", {"task_id": task_id, "run_id": run_id})
 
         # Call the new result generator task
         generate_task_result.delay(task_id, run_id, user_id)
+
         from workers.tasks import calculate_next_run
         schedule_type = task.get('schedule', {}).get('type')
         if schedule_type == 'recurring':
diff --git a/src/server/workers/planner/db.py b/src/server/workers/planner/db.py
@@ -110,7 +110,7 @@ async def create_initial_task(self, user_id: str, name: str, description: str, a
             "chat_history": [],
         }
 
-        SENSITIVE_TASK_FIELDS = ["name", "description", "original_context"]
+        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "clarifying_questions", "result", "swarm_details"]
         _encrypt_doc(task_doc, SENSITIVE_TASK_FIELDS)
 
         await self.tasks_collection.insert_one(task_doc)
@@ -120,7 +120,7 @@ async def create_initial_task(self, user_id: str, name: str, description: str, a
     async def update_task_field(self, task_id: str, fields: dict):
         """Updates specific fields of a task document."""
         if DB_ENCRYPTION_ENABLED:
-            SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "found_context", "clarifying_questions", "result", "swarm_details"]
+            SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "clarifying_questions", "result", "swarm_details"]
             encrypted_fields = {}
             for field, value in fields.items():
                 if field in SENSITIVE_TASK_FIELDS:
@@ -166,7 +166,7 @@ async def update_task_with_plan(self, task_id: str, plan_data: dict, is_change_r
     async def get_task(self, task_id: str) -> Optional[Dict]:
         """Fetches a single task by its ID."""
         doc = await self.tasks_collection.find_one({"task_id": task_id})
-        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "found_context", "clarifying_questions", "result", "swarm_details"]
+        SENSITIVE_TASK_FIELDS = ["name", "description", "plan", "runs", "original_context", "chat_history", "error", "clarifying_questions", "result", "swarm_details"]
         _decrypt_doc(doc, SENSITIVE_TASK_FIELDS)
         return doc
 
diff --git a/src/server/workers/planner/llm.py b/src/server/workers/planner/llm.py
@@ -9,23 +9,19 @@
 
 logger = logging.getLogger(__name__)
 
-def get_planner_agent(available_tools: dict, current_time_str: str, user_name: str, user_location: str, retrieved_context: dict = None):
+def get_planner_agent(available_tools: dict, current_time_str: str, user_name: str, user_location: str):
     """Initializes and returns a Qwen Assistant agent for planning."""
     
     # Format the MCP descriptions for the prompt
     # The keys are now the simple names (e.g., 'gmail')
     tools_list_str = "\n".join([f"- {name}: {desc}" for name, desc in available_tools.items()])
     
-    # Format the retrieved context for the prompt
-    context_str = json.dumps(retrieved_context, indent=2) if retrieved_context else "No additional context was found in memory."
-
     # Add current time to the prompt for better contextual planning
     system_prompt = prompts.SYSTEM_PROMPT.format(
         user_name=user_name,
         user_location=user_location,
         current_time=current_time_str,
-        available_tools=tools_list_str, # This is the new field
-        retrieved_context=context_str # This is the new field
+        available_tools=tools_list_str
     )
 
     # This function now returns the necessary components to run the agent with fallback
diff --git a/src/server/workers/planner/prompts.py b/src/server/workers/planner/prompts.py
diff --git a/src/server/workers/tasks.py b/src/server/workers/tasks.py
