fix(backend/frontend): error handling, stream reconnection, and chat switching

autogpt_platform/backend/backend/api/features/chat/routes.py

@@ -621,11 +621,10 @@ async def resume_session_stream(
     if not active_session:
         return Response(status_code=204)
 
-    # Subscribe from the beginning ("0-0") to replay all chunks for this turn.
-    # This is necessary because hydrated messages filter out incomplete tool calls
-    # to avoid "No tool invocation found" errors. The resume stream delivers
-    # those tool calls fresh with proper SDK state.
-    # The AI SDK's deduplication will handle any duplicate chunks.
+    # Always replay from the beginning ("0-0") on resume.
+    # We can't use last_message_id because it's the latest ID in the backend
+    # stream, not the latest the frontend received — the gap causes lost
+    # messages. The frontend deduplicates replayed content.
     subscriber_queue = await stream_registry.subscribe_to_session(
         session_id=session_id,
         user_id=user_id,

autogpt_platform/backend/backend/copilot/config.py

@@ -85,7 +85,7 @@ class ChatConfig(BaseSettings):
     )
     claude_agent_max_subtasks: int = Field(
         default=10,
-        description="Max number of sub-agent Tasks the SDK can spawn per session.",
+        description="Max number of concurrent sub-agent Tasks the SDK can run per session.",
     )
     claude_agent_use_resume: bool = Field(
         default=True,

autogpt_platform/backend/backend/copilot/executor/manager.py

@@ -4,6 +4,7 @@
 RabbitMQ and processes them using a thread pool, following the graph executor pattern.
 """
 
+import asyncio
 import logging
 import os
 import threading
@@ -409,14 +410,19 @@ def ack_message(reject: bool, requeue: bool):
 
         def on_run_done(f: Future):
             logger.info(f"Run completed for {session_id}")
+            error_msg = None
             try:
                 if exec_error := f.exception():
-                    logger.error(f"Execution for {session_id} failed: {exec_error}")
+                    error_msg = str(exec_error) or type(exec_error).__name__
+                    logger.error(f"Execution for {session_id} failed: {error_msg}")
                     ack_message(reject=True, requeue=False)
                 else:
                     ack_message(reject=False, requeue=False)
+            except asyncio.CancelledError:
+                logger.info(f"Run completion callback cancelled for {session_id}")
             except BaseException as e:
-                logger.exception(f"Error in run completion callback: {e}")
+                error_msg = str(e) or type(e).__name__
+                logger.exception(f"Error in run completion callback: {error_msg}")
             finally:
                 # Release the cluster lock
                 if session_id in self._task_locks:

autogpt_platform/backend/backend/copilot/executor/processor.py

@@ -125,7 +125,10 @@ def cleanup(self):
             )
             future.result(timeout=5)
         except Exception as e:
-            logger.warning(f"[CoPilotExecutor] Worker {self.tid} cleanup error: {e}")
+            error_msg = str(e) or type(e).__name__
+            logger.warning(
+                f"[CoPilotExecutor] Worker {self.tid} cleanup error: {error_msg}"
+            )
 
         # Stop the event loop
         self.execution_loop.call_soon_threadsafe(self.execution_loop.stop)
@@ -157,47 +160,30 @@ def execute(
 
         start_time = time.monotonic()
 
-        try:
-            # Run the async execution in our event loop
-            future = asyncio.run_coroutine_threadsafe(
-                self._execute_async(entry, cancel, cluster_lock, log),
-                self.execution_loop,
-            )
-
-            # Wait for completion, checking cancel periodically
-            while not future.done():
-                try:
-                    future.result(timeout=1.0)
-                except asyncio.TimeoutError:
-                    if cancel.is_set():
-                        log.info("Cancellation requested")
-                        future.cancel()
-                        break
-                    # Refresh cluster lock to maintain ownership
-                    cluster_lock.refresh()
+        # Run the async execution in our event loop
+        future = asyncio.run_coroutine_threadsafe(
+            self._execute_async(entry, cancel, cluster_lock, log),
+            self.execution_loop,
+        )
 
-            if not future.cancelled():
-                # Get result to propagate any exceptions
-                future.result()
+        # Wait for completion, checking cancel periodically
+        while not future.done():
+            try:
+                future.result(timeout=1.0)
+            except asyncio.TimeoutError:
+                if cancel.is_set():
+                    log.info("Cancellation requested")
+                    future.cancel()
+                    break
+                # Refresh cluster lock to maintain ownership
+                cluster_lock.refresh()
 
-            elapsed = time.monotonic() - start_time
-            log.info(f"Execution completed in {elapsed:.2f}s")
+        if not future.cancelled():
+            # Get result to propagate any exceptions
+            future.result()
 
-        except BaseException as e:
-            elapsed = time.monotonic() - start_time
-            log.error(f"Execution failed after {elapsed:.2f}s: {e}")
-            # Safety net: if _execute_async's error handler failed to mark
-            # the session (e.g. RuntimeError from SDK cleanup), do it here.
-            try:
-                asyncio.run_coroutine_threadsafe(
-                    stream_registry.mark_session_completed(
-                        entry.session_id, error_message=str(e) or "Unknown error"
-                    ),
-                    self.execution_loop,
-                ).result(timeout=5.0)
-            except Exception as cleanup_err:
-                log.error(f"Safety net mark_session_completed failed: {cleanup_err}")
-            raise
+        elapsed = time.monotonic() - start_time
+        log.info(f"Execution completed in {elapsed:.2f}s")
 
     async def _execute_async(
         self,
@@ -219,6 +205,7 @@ async def _execute_async(
         """
         last_refresh = time.monotonic()
         refresh_interval = 30.0  # Refresh lock every 30 seconds
+        error_msg = None
 
         try:
             # Choose service based on LaunchDarkly flag
@@ -264,17 +251,26 @@ async def _execute_async(
                         exc_info=True,
                     )
 
-            error_message = "Operation cancelled" if cancel.is_set() else None
-            await stream_registry.mark_session_completed(
-                entry.session_id, error_message=error_message
-            )
+            # Stream loop completed
+            if cancel.is_set():
+                log.info("Stream cancelled by user")
 
         except BaseException as e:
-            log.error(f"Turn failed: {e}")
+            # Handle all exceptions (including CancelledError) with appropriate logging
+            if isinstance(e, asyncio.CancelledError):
+                log.info("Turn cancelled")
+                error_msg = "Operation cancelled"
+            else:
+                error_msg = str(e) or type(e).__name__
+                log.error(f"Turn failed: {error_msg}")
+            raise
+        finally:
+            # If no exception but user cancelled, still mark as cancelled
+            if not error_msg and cancel.is_set():
+                error_msg = "Operation cancelled"
             try:
                 await stream_registry.mark_session_completed(
-                    entry.session_id, error_message=str(e) or "Unknown error"
+                    entry.session_id, error_message=error_msg
                 )
             except Exception as mark_err:
-                log.error(f"mark_session_completed also failed: {mark_err}")
-            raise
+                log.error(f"Failed to mark session completed: {mark_err}")

autogpt_platform/backend/backend/copilot/sdk/security_hooks.py

@@ -160,7 +160,7 @@ def create_security_hooks(
     Args:
         user_id: Current user ID for isolation validation
         sdk_cwd: SDK working directory for workspace-scoped tool validation
-        max_subtasks: Maximum Task (sub-agent) spawns allowed per session
+        max_subtasks: Maximum concurrent Task (sub-agent) spawns allowed per session
         on_stop: Callback ``(transcript_path, sdk_session_id)`` invoked when
             the SDK finishes processing — used to read the JSONL transcript
             before the CLI process exits.
@@ -172,16 +172,16 @@ def create_security_hooks(
         from claude_agent_sdk import HookMatcher
         from claude_agent_sdk.types import HookContext, HookInput, SyncHookJSONOutput
 
-        # Per-session counter for Task sub-agent spawns
-        task_spawn_count = 0
+        # Per-session tracking for Task sub-agent concurrency.
+        # Set of tool_use_ids that consumed a slot — len() is the active count.
+        task_tool_use_ids: set[str] = set()
 
         async def pre_tool_use_hook(
             input_data: HookInput,
             tool_use_id: str | None,
             context: HookContext,
         ) -> SyncHookJSONOutput:
             """Combined pre-tool-use validation hook."""
-            nonlocal task_spawn_count
             _ = context  # unused but required by signature
             tool_name = cast(str, input_data.get("tool_name", ""))
             tool_input = cast(dict[str, Any], input_data.get("tool_input", {}))
@@ -200,18 +200,18 @@ async def pre_tool_use_hook(
                             "(remove the run_in_background parameter)."
                         ),
                     )
-                if task_spawn_count >= max_subtasks:
+                if len(task_tool_use_ids) >= max_subtasks:
                     logger.warning(
                         f"[SDK] Task limit reached ({max_subtasks}), user={user_id}"
                     )
                     return cast(
                         SyncHookJSONOutput,
                         _deny(
-                            f"Maximum {max_subtasks} sub-tasks per session. "
-                            "Please continue in the main conversation."
+                            f"Maximum {max_subtasks} concurrent sub-tasks. "
+                            "Wait for running sub-tasks to finish, "
+                            "or continue in the main conversation."
                         ),
                     )
-                task_spawn_count += 1
 
             # Strip MCP prefix for consistent validation
             is_copilot_tool = tool_name.startswith(MCP_TOOL_PREFIX)
@@ -229,9 +229,24 @@ async def pre_tool_use_hook(
             if result:
                 return cast(SyncHookJSONOutput, result)
 
+            # Reserve the Task slot only after all validations pass
+            if tool_name == "Task" and tool_use_id is not None:
+                task_tool_use_ids.add(tool_use_id)
+
             logger.debug(f"[SDK] Tool start: {tool_name}, user={user_id}")
             return cast(SyncHookJSONOutput, {})
 
+        def _release_task_slot(tool_name: str, tool_use_id: str | None) -> None:
+            """Release a Task concurrency slot if one was reserved."""
+            if tool_name == "Task" and tool_use_id in task_tool_use_ids:
+                task_tool_use_ids.discard(tool_use_id)
+                logger.info(
+                    "[SDK] Task slot released, active=%d/%d, user=%s",
+                    len(task_tool_use_ids),
+                    max_subtasks,
+                    user_id,
+                )
+
         async def post_tool_use_hook(
             input_data: HookInput,
             tool_use_id: str | None,
@@ -246,6 +261,8 @@ async def post_tool_use_hook(
             """
             _ = context
             tool_name = cast(str, input_data.get("tool_name", ""))
+
+            _release_task_slot(tool_name, tool_use_id)
             is_builtin = not tool_name.startswith(MCP_TOOL_PREFIX)
             logger.info(
                 "[SDK] PostToolUse: %s (builtin=%s, tool_use_id=%s)",
@@ -289,6 +306,9 @@ async def post_tool_failure_hook(
                 f"[SDK] Tool failed: {tool_name}, error={error}, "
                 f"user={user_id}, tool_use_id={tool_use_id}"
             )
+
+            _release_task_slot(tool_name, tool_use_id)
+
             return cast(SyncHookJSONOutput, {})
 
         async def pre_compact_hook(