Revert "feat: Stream mode task spliting (#767)"

This reverts commit 16ac0d8, reversing
changes made to 4ce7926.

Author: Wendong-Fan

backend/app/service/chat_service.py

@@ -20,8 +20,6 @@ class Action(str, Enum):
     update_task = "update_task"  # user -> backend
     task_state = "task_state"  # backend -> user
     new_task_state = "new_task_state"  # backend -> user
-    decompose_progress = "decompose_progress"  # backend -> user (streaming decomposition)
-    decompose_text = "decompose_text"  # backend -> user (raw streaming text)
     start = "start"  # user -> backend
     create_agent = "create_agent"  # backend -> user
     activate_agent = "activate_agent"  # backend -> user
@@ -66,17 +64,6 @@ class ActionTaskStateData(BaseModel):
     action: Literal[Action.task_state] = Action.task_state
     data: dict[Literal["task_id", "content", "state", "result", "failure_count"], str | int]
 
-
-class ActionDecomposeProgressData(BaseModel):
-    action: Literal[Action.decompose_progress] = Action.decompose_progress
-    data: dict
-
-
-class ActionDecomposeTextData(BaseModel):
-    action: Literal[Action.decompose_text] = Action.decompose_text
-    data: dict
-
-
 class ActionNewTaskStateData(BaseModel):
     action: Literal[Action.new_task_state] = Action.new_task_state
     data: dict[Literal["task_id", "content", "state", "result", "failure_count"], str | int]
@@ -240,8 +227,6 @@ class ActionSkipTaskData(BaseModel):
     | ActionAddTaskData
     | ActionRemoveTaskData
     | ActionSkipTaskData
-    | ActionDecomposeTextData
-    | ActionDecomposeProgressData
 )
 
 

backend/app/service/task.py

@@ -108,7 +108,6 @@ def __init__(
         prune_tool_calls_from_memory: bool = False,
         enable_snapshot_clean: bool = False,
         step_timeout: float | None = 900,
-        **kwargs: Any,
     ) -> None:
         super().__init__(
             system_message=system_message,
@@ -131,7 +130,6 @@ def __init__(
             prune_tool_calls_from_memory=prune_tool_calls_from_memory,
             enable_snapshot_clean=enable_snapshot_clean,
             step_timeout=step_timeout,
-            **kwargs,
         )
         self.api_task_id = api_task_id
         self.agent_name = agent_name
@@ -535,21 +533,6 @@ def agent_model(
         )
     )
 
-    # Build model config, defaulting to streaming for planner
-    extra_params = options.extra_params or {}
-    model_config: dict[str, Any] = {}
-    if options.is_cloud():
-        model_config["user"] = str(options.project_id)
-    model_config.update(
-        {
-            k: v
-            for k, v in extra_params.items()
-            if k not in ["model_platform", "model_type", "api_key", "url"]
-        }
-    )
-    if agent_name == Agents.task_agent:
-        model_config["stream"] = True
-
     return ListenChatAgent(
         options.project_id,
         agent_name,
@@ -559,15 +542,23 @@ def agent_model(
             model_type=options.model_type,
             api_key=options.api_key,
             url=options.api_url,
-            model_config_dict=model_config or None,
+            model_config_dict={
+                "user": str(options.project_id),
+            }
+            if options.is_cloud()
+            else None,
+            **{
+                k: v
+                for k, v in (options.extra_params or {}).items()
+                if k not in ["model_platform", "model_type", "api_key", "url"]
+            },
         ),
         # output_language=options.language,
         tools=tools,
         agent_id=agent_id,
         prune_tool_calls_from_memory=prune_tool_calls_from_memory,
         toolkits_to_register_agent=toolkits_to_register_agent,
         enable_snapshot_clean=enable_snapshot_clean,
-        stream_accumulate=False,
     )
 
 

backend/app/utils/agent.py

@@ -12,7 +12,6 @@
 from camel.societies.workforce.utils import TaskAssignResult
 from camel.societies.workforce.workforce_metrics import WorkforceMetrics
 from camel.societies.workforce.events import WorkerCreatedEvent
-from camel.societies.workforce.prompts import TASK_DECOMPOSE_PROMPT
 from camel.tasks.task import Task, TaskState, validate_task_content
 from app.component import code
 from app.exception.exception import UserException
@@ -66,22 +65,14 @@ def __init__(
         )
         logger.info(f"[WF-LIFECYCLE] ✅ Workforce.__init__ COMPLETED, id={id(self)}")
 
-    def eigent_make_sub_tasks(
-        self,
-        task: Task,
-        coordinator_context: str = "",
-        on_stream_batch=None,
-        on_stream_text=None,
-    ):
+    def eigent_make_sub_tasks(self, task: Task, coordinator_context: str = ""):
         """
         Split process_task method to eigent_make_sub_tasks and eigent_start method.
 
         Args:
             task: The main task to decompose
             coordinator_context: Optional context ONLY for coordinator agent during decomposition.
                                 This context will NOT be passed to subtasks or worker agents.
-            on_stream_batch: Optional callback for streaming batches signature (List[Task], bool)
-            on_stream_text: Optional callback for raw streaming text chunks
         """
         logger.info("=" * 80)
         logger.info("🧩 [DECOMPOSE] eigent_make_sub_tasks CALLED", extra={
@@ -112,15 +103,7 @@ def eigent_make_sub_tasks(
         logger.info(f"[DECOMPOSE] Workforce reset complete, state: {self._state.name}")
 
         logger.info(f"[DECOMPOSE] Calling handle_decompose_append_task")
-        subtasks = asyncio.run(
-            self.handle_decompose_append_task(
-                task, 
-                reset=False, 
-                coordinator_context=coordinator_context,
-                on_stream_batch=on_stream_batch, 
-                on_stream_text=on_stream_text
-            )
-        )
+        subtasks = asyncio.run(self.handle_decompose_append_task(task, reset=False, coordinator_context=coordinator_context))
         logger.info("=" * 80)
         logger.info(f"✅ [DECOMPOSE] Task decomposition COMPLETED", extra={
             "api_task_id": self.api_task_id,
@@ -159,45 +142,8 @@ async def eigent_start(self, subtasks: list[Task]):
                 self._state = WorkforceState.IDLE
                 logger.info(f"[WF-LIFECYCLE] Workforce state set to IDLE")
 
-    def _decompose_task(self, task: Task, stream_callback=None):
-        """Decompose task with optional streaming text callback."""
-
-        decompose_prompt = str(
-            TASK_DECOMPOSE_PROMPT.format(
-                content=task.content,
-                child_nodes_info=self._get_child_nodes_info(),
-                additional_info=task.additional_info,
-            )
-        )
-        self.task_agent.reset()
-        result = task.decompose(
-            self.task_agent, decompose_prompt, stream_callback=stream_callback
-        )
-
-        if isinstance(result, Generator):
-            def streaming_with_dependencies():
-                all_subtasks = []
-                for new_tasks in result:
-                    all_subtasks.extend(new_tasks)
-                    if new_tasks:
-                        self._update_dependencies_for_decomposition(
-                            task, all_subtasks
-                        )
-                    yield new_tasks
-            return streaming_with_dependencies()
-        else:
-            subtasks = result
-            if subtasks:
-                self._update_dependencies_for_decomposition(task, subtasks)
-            return subtasks
-
     async def handle_decompose_append_task(
-        self,
-        task: Task,
-        reset: bool = True,
-        coordinator_context: str = "",
-        on_stream_batch=None,
-        on_stream_text=None,
+        self, task: Task, reset: bool = True, coordinator_context: str = ""
     ) -> List[Task]:
         """
         Override to support coordinator_context parameter.
@@ -207,8 +153,6 @@ async def handle_decompose_append_task(
             task: The task to be processed
             reset: Should trigger workforce reset (Workforce must not be running)
             coordinator_context: Optional context ONLY for coordinator during decomposition
-            on_stream_batch: Optional callback for streaming batches signature (List[Task], bool)
-            on_stream_text: Optional callback for raw streaming text chunks
 
         Returns:
             List[Task]: The decomposed subtasks or the original task
@@ -242,23 +186,18 @@ async def handle_decompose_append_task(
             task.content = task_with_context
 
             logger.info(f"[DECOMPOSE] Calling _decompose_task with context")
-            subtasks_result = self._decompose_task(task, stream_callback=on_stream_text)
+            subtasks_result = self._decompose_task(task)
 
             task.content = original_content
         else:
             logger.info(f"[DECOMPOSE] Calling _decompose_task without context")
-            subtasks_result = self._decompose_task(task, stream_callback=on_stream_text)
+            subtasks_result = self._decompose_task(task)
 
         logger.info(f"[DECOMPOSE] _decompose_task returned, processing results")
         if isinstance(subtasks_result, Generator):
             subtasks = []
             for new_tasks in subtasks_result:
                 subtasks.extend(new_tasks)
-                if on_stream_batch:
-                    try:
-                        on_stream_batch(new_tasks, False)
-                    except Exception as e:
-                        logger.warning(f"Streaming callback failed: {e}")
             logger.info(f"[DECOMPOSE] Collected {len(subtasks)} subtasks from generator")
         else:
             subtasks = subtasks_result
@@ -279,12 +218,6 @@ async def handle_decompose_append_task(
             subtasks = [fallback_task]
             logger.info(f"[DECOMPOSE] Created fallback task: {fallback_task.id}")
 
-        if on_stream_batch:
-            try:
-                on_stream_batch(subtasks, True)
-            except Exception as e:
-                logger.warning(f"Final streaming callback failed: {e}")
-
         return subtasks
 
     async def _find_assignee(self, tasks: List[Task]) -> TaskAssignResult:

backend/app/utils/workforce.py

@@ -23,39 +23,7 @@ export const ProjectSection = React.forwardRef<HTMLDivElement, ProjectSectionPro
   onSkip,
   isPauseResumeLoading
 }, ref) => {
-  // Subscribe to store changes with throttling to prevent excessive re-renders
-  const [chatState, setChatState] = React.useState(() => chatStore.getState());
-
-  React.useEffect(() => {
-    let timeoutId: NodeJS.Timeout | null = null;
-    let latestState: any = null;
-
-    const unsubscribe = chatStore.subscribe((state) => {
-      latestState = state;
-
-      // Throttle updates to max once per 100ms
-      if (!timeoutId) {
-        timeoutId = setTimeout(() => {
-          if (latestState) {
-            setChatState(latestState);
-          }
-          timeoutId = null;
-        }, 100);
-      }
-    });
-
-    return () => {
-      unsubscribe();
-      if (timeoutId) {
-        clearTimeout(timeoutId);
-        // Apply final state on cleanup
-        if (latestState) {
-          setChatState(latestState);
-        }
-      }
-    };
-  }, [chatStore]);
-
+  const chatState = chatStore.getState();
   const activeTaskId = chatState.activeTaskId;
 
   if (!activeTaskId || !chatState.tasks[activeTaskId]) {
@@ -65,17 +33,8 @@ export const ProjectSection = React.forwardRef<HTMLDivElement, ProjectSectionPro
   const task = chatState.tasks[activeTaskId];
   const messages = task.messages || [];
 
-  // Create a stable key based on messages content to prevent excessive re-renders
-  const lastMessage = messages[messages.length - 1];
-  const messagesKey = React.useMemo(() => {
-    // Only re-compute when message count or last message changes
-    return `${messages.length}-${lastMessage?.id || ''}-${lastMessage?.content?.length || 0}`;
-  }, [messages.length, lastMessage?.id, lastMessage?.content?.length]);
-
-  // Memoize grouping to prevent re-creating objects on every render
-  const queryGroups = React.useMemo(() => {
-    return groupMessagesByQuery(messages);
-  }, [messagesKey]);
+  // Group messages by query cycles and show in chronological order (oldest first)
+  const queryGroups = groupMessagesByQuery(messages);
 
   return (
     <motion.div
@@ -193,7 +152,7 @@ function groupMessagesByQuery(messages: any[]) {
           otherMessages: []
         };
       }
-    } else {
+  } else {
       // Other messages (assistant responses, errors, etc.)
       if (currentGroup) {
         currentGroup.otherMessages.push(message);