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

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

@@ -617,6 +617,11 @@ async def resume_session_stream(
     active_session, last_message_id = await stream_registry.get_active_session(
         session_id, user_id
     )
+    logger.info(
+        f"[STREAM_DIAG] resume_session_stream: session={session_id}, "
+        f"has_active={active_session is not None}, "
+        f"status={active_session.status if active_session else 'N/A'}"
+    )
 
     if not active_session:
         return Response(status_code=204)

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/processor.py

@@ -264,6 +264,12 @@ async def _execute_async(
                         exc_info=True,
                     )
 
+            logger.info(
+                f"[STREAM_DIAG] stream_fn loop ended, calling mark_session_completed, "
+                f"session_id={entry.session_id}, cancel={cancel.is_set()}, "
+                f"turn_id={entry.turn_id}"
+            )
+
             error_message = "Operation cancelled" if cancel.is_set() else None
             await stream_registry.mark_session_completed(
                 entry.session_id, error_message=error_message

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(

autogpt_platform/backend/backend/copilot/sdk/security_hooks_test.py

@@ -208,19 +208,22 @@ def test_bash_builtin_blocked_message_clarity():
 
 @pytest.fixture()
 def _hooks():
-    """Create security hooks and return the PreToolUse handler."""
+    """Create security hooks and return (pre, post, post_failure) handlers."""
     from .security_hooks import create_security_hooks
 
     hooks = create_security_hooks(user_id="u1", sdk_cwd=SDK_CWD, max_subtasks=2)
     pre = hooks["PreToolUse"][0].hooks[0]
-    return pre
+    post = hooks["PostToolUse"][0].hooks[0]
+    post_failure = hooks["PostToolUseFailure"][0].hooks[0]
+    return pre, post, post_failure
 
 
 @pytest.mark.skipif(not _sdk_available(), reason="claude_agent_sdk not installed")
 @pytest.mark.asyncio
 async def test_task_background_blocked(_hooks):
     """Task with run_in_background=true must be denied."""
-    result = await _hooks(
+    pre, _, _ = _hooks
+    result = await pre(
         {"tool_name": "Task", "tool_input": {"run_in_background": True, "prompt": "x"}},
         tool_use_id=None,
         context={},
@@ -233,9 +236,10 @@ async def test_task_background_blocked(_hooks):
 @pytest.mark.asyncio
 async def test_task_foreground_allowed(_hooks):
     """Task without run_in_background should be allowed."""
-    result = await _hooks(
+    pre, _, _ = _hooks
+    result = await pre(
         {"tool_name": "Task", "tool_input": {"prompt": "do stuff"}},
-        tool_use_id=None,
+        tool_use_id="tu-1",
         context={},
     )
     assert not _is_denied(result)
@@ -245,25 +249,102 @@ async def test_task_foreground_allowed(_hooks):
 @pytest.mark.asyncio
 async def test_task_limit_enforced(_hooks):
     """Task spawns beyond max_subtasks should be denied."""
+    pre, _, _ = _hooks
     # First two should pass
-    for _ in range(2):
-        result = await _hooks(
+    for i in range(2):
+        result = await pre(
             {"tool_name": "Task", "tool_input": {"prompt": "ok"}},
-            tool_use_id=None,
+            tool_use_id=f"tu-limit-{i}",
             context={},
         )
         assert not _is_denied(result)
 
     # Third should be denied (limit=2)
-    result = await _hooks(
+    result = await pre(
         {"tool_name": "Task", "tool_input": {"prompt": "over limit"}},
-        tool_use_id=None,
+        tool_use_id="tu-limit-2",
         context={},
     )
     assert _is_denied(result)
     assert "Maximum" in _reason(result)
 
 
+@pytest.mark.skipif(not _sdk_available(), reason="claude_agent_sdk not installed")
+@pytest.mark.asyncio
+async def test_task_slot_released_on_completion(_hooks):
+    """Completing a Task should free a slot so new Tasks can be spawned."""
+    pre, post, _ = _hooks
+    # Fill both slots
+    for i in range(2):
+        result = await pre(
+            {"tool_name": "Task", "tool_input": {"prompt": "ok"}},
+            tool_use_id=f"tu-comp-{i}",
+            context={},
+        )
+        assert not _is_denied(result)
+
+    # Third should be denied — at capacity
+    result = await pre(
+        {"tool_name": "Task", "tool_input": {"prompt": "over"}},
+        tool_use_id="tu-comp-2",
+        context={},
+    )
+    assert _is_denied(result)
+
+    # Complete first task — frees a slot
+    await post(
+        {"tool_name": "Task", "tool_input": {}},
+        tool_use_id="tu-comp-0",
+        context={},
+    )
+
+    # Now a new Task should be allowed
+    result = await pre(
+        {"tool_name": "Task", "tool_input": {"prompt": "after release"}},
+        tool_use_id="tu-comp-3",
+        context={},
+    )
+    assert not _is_denied(result)
+
+
+@pytest.mark.skipif(not _sdk_available(), reason="claude_agent_sdk not installed")
+@pytest.mark.asyncio
+async def test_task_slot_released_on_failure(_hooks):
+    """A failed Task should also free its concurrency slot."""
+    pre, _, post_failure = _hooks
+    # Fill both slots
+    for i in range(2):
+        result = await pre(
+            {"tool_name": "Task", "tool_input": {"prompt": "ok"}},
+            tool_use_id=f"tu-fail-{i}",
+            context={},
+        )
+        assert not _is_denied(result)
+
+    # At capacity
+    result = await pre(
+        {"tool_name": "Task", "tool_input": {"prompt": "over"}},
+        tool_use_id="tu-fail-2",
+        context={},
+    )
+    assert _is_denied(result)
+
+    # Fail first task — should free a slot
+    await post_failure(
+        {"tool_name": "Task", "tool_input": {}, "error": "something broke"},
+        tool_use_id="tu-fail-0",
+        context={},
+    )
+
+    # New Task should be allowed
+    result = await pre(
+        {"tool_name": "Task", "tool_input": {"prompt": "after failure"}},
+        tool_use_id="tu-fail-3",
+        context={},
+    )
+    assert not _is_denied(result)
+
+
 # -- _is_tool_error_or_denial ------------------------------------------------
 
 
@@ -298,7 +379,9 @@ def test_background_task_denial(self):
     def test_subtask_limit_denial(self):
         assert (
             _is_tool_error_or_denial(
-                "Maximum 2 sub-tasks per session. Please continue in the main conversation."
+                "Maximum 2 concurrent sub-tasks. "
+                "Wait for running sub-tasks to finish, "
+                "or continue in the main conversation."
             )
             is True
         )

autogpt_platform/backend/backend/copilot/sdk/service.py

@@ -725,6 +725,25 @@ async def _next_msg() -> Any:
                                     - len(adapter.resolved_tool_calls),
                                 )
 
+                        # Log ResultMessage details for debugging
+                        if isinstance(sdk_msg, ResultMessage):
+                            logger.info(
+                                "[SDK] [%s] Received: ResultMessage %s "
+                                "(unresolved=%d, current=%d, resolved=%d)",
+                                session_id[:12],
+                                sdk_msg.subtype,
+                                len(adapter.current_tool_calls)
+                                - len(adapter.resolved_tool_calls),
+                                len(adapter.current_tool_calls),
+                                len(adapter.resolved_tool_calls),
+                            )
+                            if sdk_msg.subtype in ("error", "error_during_execution"):
+                                logger.error(
+                                    "[SDK] [%s] SDK execution failed with error: %s",
+                                    session_id[:12],
+                                    sdk_msg.result or "(no error message provided)",
+                                )
+
                         for response in adapter.convert_message(sdk_msg):
                             if isinstance(response, StreamStart):
                                 continue
@@ -749,6 +768,15 @@ async def _next_msg() -> Any:
                                     extra,
                                 )
 
+                            # Log errors being sent to frontend
+                            if isinstance(response, StreamError):
+                                logger.error(
+                                    "[SDK] [%s] Sending error to frontend: %s (code=%s)",
+                                    session_id[:12],
+                                    response.errorText,
+                                    response.code,
+                                )
+
                             yield response
 
                             if isinstance(response, StreamTextDelta):

autogpt_platform/backend/backend/copilot/stream_registry.py

@@ -707,6 +707,10 @@ async def mark_session_completed(
         True if session was newly marked completed, False if already completed/failed
     """
     status: Literal["completed", "failed"] = "failed" if error_message else "completed"
+    logger.info(
+        f"[STREAM_DIAG] mark_session_completed called, session={session_id}, "
+        f"status={status}, error={error_message!r}"
+    )
 
     redis = await get_redis_async()
     meta_key = _get_session_meta_key(session_id)

autogpt_platform/frontend/src/app/(platform)/copilot/useChatSession.ts

@@ -46,8 +46,15 @@ export function useChatSession() {
   // resume which fires before hydration).
   const hasActiveStream = useMemo(() => {
     if (sessionQuery.data?.status !== 200) return false;
-    return !!sessionQuery.data.data.active_stream;
-  }, [sessionQuery.data]);
+    const active = !!sessionQuery.data.data.active_stream;
+    console.info("[STREAM_DIAG] session data", {
+      sessionId,
+      hasActiveStream: active,
+      messageCount: sessionQuery.data.data.messages?.length ?? 0,
+      ts: Date.now(),
+    });
+    return active;
+  }, [sessionQuery.data, sessionId]);
 
   // Memoize so the effect in useCopilotPage doesn't infinite-loop on a new
   // array reference every render. Re-derives only when query data changes.
@@ -119,5 +126,6 @@ export function useChatSession() {
     isSessionError: sessionQuery.isError,
     createSession,
     isCreatingSession,
+    refetchSession: sessionQuery.refetch,
   };
 }