feat(copilot): wire up stop button to cancel executor tasks (#12171)

## Summary

- The stop button was completely disconnected — clicking it only aborted
the client-side SSE fetch while the executor kept running indefinitely
- The executor already had full cancel infrastructure (RabbitMQ FANOUT
consumer, `CancelCoPilotEvent`, `threading.Event`, periodic cancel
checks) but nobody ever published a cancel message
- This PR wires up the missing pieces: a cancel REST endpoint, a publish
function, and frontend integration

## Changes

- **`executor/utils.py`**: Add `enqueue_cancel_task()` to publish
`CancelCoPilotEvent` to the existing FANOUT exchange
- **`routes.py`**: Add `POST /sessions/{session_id}/cancel` that finds
the active task, publishes cancel, and polls Redis until the task
confirms stopped (up to 10s timeout)
- **`cancel/route.ts`**: Next.js API proxy route for the cancel endpoint
- **`useCopilotPage.ts`**: Wrap AI SDK's `stop()` to also call the
backend cancel API — `sdkStop()` fires first for instant UI feedback,
then the cancel API waits for executor confirmation

## Test plan

- [ ] Start a copilot chat session and send a message
- [ ] Click "Stop generating" while streaming
- [ ] Verify executor logs show `Received cancel for {task_id}` and
`Cancelled during streaming`
- [ ] Verify the cancel endpoint returns `{"cancelled": true}` (not
timeout)
- [ ] Verify frontend transitions to idle state
- [ ] Verify clicking stop when no task is running returns
`{"cancelled": false, "reason": "no_active_task"}`

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

@@ -18,7 +18,7 @@ from backend.copilot.completion_handler import (
     process_operation_success,
 )
 from backend.copilot.config import ChatConfig
-from backend.copilot.executor.utils import enqueue_copilot_task
+from backend.copilot.executor.utils import enqueue_cancel_task, enqueue_copilot_task
 from backend.copilot.model import (
     ChatMessage,
     ChatSession,
@@ -132,6 +132,14 @@ class ListSessionsResponse(BaseModel):
     total: int
 
 
+class CancelTaskResponse(BaseModel):
+    """Response model for the cancel task endpoint."""
+
+    cancelled: bool
+    task_id: str | None = None
+    reason: str | None = None
+
+
 class OperationCompleteRequest(BaseModel):
     """Request model for external completion webhook."""
 
@@ -314,6 +322,57 @@ async def get_session(
     )
 
 
+@router.post(
+    "/sessions/{session_id}/cancel",
+    status_code=200,
+)
+async def cancel_session_task(
+    session_id: str,
+    user_id: Annotated[str | None, Depends(auth.get_user_id)],
+) -> CancelTaskResponse:
+    """Cancel the active streaming task for a session.
+
+    Publishes a cancel event to the executor via RabbitMQ FANOUT, then
+    polls Redis until the task status flips from ``running`` or a timeout
+    (5 s) is reached.  Returns only after the cancellation is confirmed.
+    """
+    await _validate_and_get_session(session_id, user_id)
+
+    active_task, _ = await stream_registry.get_active_task_for_session(
+        session_id, user_id
+    )
+    if not active_task:
+        return CancelTaskResponse(cancelled=False, reason="no_active_task")
+
+    task_id = active_task.task_id
+    await enqueue_cancel_task(task_id)
+    logger.info(
+        f"[CANCEL] Published cancel for task ...{task_id[-8:]} "
+        f"session ...{session_id[-8:]}"
+    )
+
+    # Poll until the executor confirms the task is no longer running.
+    # Keep max_wait below typical reverse-proxy read timeouts.
+    poll_interval = 0.5
+    max_wait = 5.0
+    waited = 0.0
+    while waited < max_wait:
+        await asyncio.sleep(poll_interval)
+        waited += poll_interval
+        task = await stream_registry.get_task(task_id)
+        if task is None or task.status != "running":
+            logger.info(
+                f"[CANCEL] Task ...{task_id[-8:]} confirmed stopped "
+                f"(status={task.status if task else 'gone'}) after {waited:.1f}s"
+            )
+            return CancelTaskResponse(cancelled=True, task_id=task_id)
+
+    logger.warning(f"[CANCEL] Task ...{task_id[-8:]} not confirmed after {max_wait}s")
+    return CancelTaskResponse(
+        cancelled=True, task_id=task_id, reason="cancel_published_not_confirmed"
+    )
+
+
 @router.post(
     "/sessions/{session_id}/stream",
 )

autogpt_platform/backend/backend/copilot/executor/utils.py

@@ -205,3 +205,20 @@ async def enqueue_copilot_task(
         message=entry.model_dump_json(),
         exchange=COPILOT_EXECUTION_EXCHANGE,
     )
+
+
+async def enqueue_cancel_task(task_id: str) -> None:
+    """Publish a cancel request for a running CoPilot task.
+
+    Sends a ``CancelCoPilotEvent`` to the FANOUT exchange so all executor
+    pods receive the cancellation signal.
+    """
+    from backend.util.clients import get_async_copilot_queue
+
+    event = CancelCoPilotEvent(task_id=task_id)
+    queue_client = await get_async_copilot_queue()
+    await queue_client.publish_message(
+        routing_key="",  # FANOUT ignores routing key
+        message=event.model_dump_json(),
+        exchange=COPILOT_CANCEL_EXCHANGE,
+    )

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

@@ -1,5 +1,6 @@
 import {
   getGetV2ListSessionsQueryKey,
+  postV2CancelSessionTask,
   useDeleteV2DeleteSession,
   useGetV2ListSessions,
 } from "@/app/api/__generated__/endpoints/chat/chat";
@@ -95,7 +96,7 @@ export function useCopilotPage() {
   const {
     messages,
     sendMessage,
-    stop,
+    stop: sdkStop,
     status,
     error,
     setMessages,
@@ -108,6 +109,52 @@ export function useCopilotPage() {
     // call resumeStream() manually after hydration + active_stream detection.
   });
 
+  // Wrap AI SDK's stop() to also cancel the backend executor task.
+  // sdkStop() aborts the SSE fetch instantly (UI feedback), then we fire
+  // the cancel API to actually stop the executor and wait for confirmation.
+  async function stop() {
+    sdkStop();
+
+    // Mark any in-progress tool parts as errored so spinners stop.
+    setMessages((prev) =>
+      prev.map((msg) => ({
+        ...msg,
+        parts: msg.parts.map((part) =>
+          "state" in part &&
+          (part.state === "input-streaming" || part.state === "input-available")
+            ? {
+                ...part,
+                state: "output-error" as const,
+                errorText: "Cancelled",
+              }
+            : part,
+        ),
+      })),
+    );
+
+    if (!sessionId) return;
+    try {
+      const res = await postV2CancelSessionTask(sessionId);
+      if (
+        res.status === 200 &&
+        "reason" in res.data &&
+        res.data.reason === "cancel_published_not_confirmed"
+      ) {
+        toast({
+          title: "Stop may take a moment",
+          description:
+            "The cancel was sent but not yet confirmed. The task should stop shortly.",
+        });
+      }
+    } catch {
+      toast({
+        title: "Could not stop the task",
+        description: "The task may still be running in the background.",
+        variant: "destructive",
+      });
+    }
+  }
+
   // Abort the stream if the backend doesn't start sending data within 12s.
   const stopRef = useRef(stop);
   stopRef.current = stop;

autogpt_platform/frontend/src/app/api/openapi.json

@@ -1263,6 +1263,44 @@
         }
       }
     },
+    "/api/chat/sessions/{session_id}/cancel": {
+      "post": {
+        "tags": ["v2", "chat", "chat"],
+        "summary": "Cancel Session Task",
+        "description": "Cancel the active streaming task for a session.\n\nPublishes a cancel event to the executor via RabbitMQ FANOUT, then\npolls Redis until the task status flips from ``running`` or a timeout\n(5 s) is reached.  Returns only after the cancellation is confirmed.",
+        "operationId": "postV2CancelSessionTask",
+        "security": [{ "HTTPBearerJWT": [] }],
+        "parameters": [
+          {
+            "name": "session_id",
+            "in": "path",
+            "required": true,
+            "schema": { "type": "string", "title": "Session Id" }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": { "$ref": "#/components/schemas/CancelTaskResponse" }
+              }
+            }
+          },
+          "401": {
+            "$ref": "#/components/responses/HTTP401NotAuthenticatedError"
+          },
+          "422": {
+            "description": "Validation Error",
+            "content": {
+              "application/json": {
+                "schema": { "$ref": "#/components/schemas/HTTPValidationError" }
+              }
+            }
+          }
+        }
+      }
+    },
     "/api/chat/sessions/{session_id}/stream": {
       "get": {
         "tags": ["v2", "chat", "chat"],
@@ -7537,6 +7575,23 @@
         "required": ["file"],
         "title": "Body_postV2Upload submission media"
       },
+      "CancelTaskResponse": {
+        "properties": {
+          "cancelled": { "type": "boolean", "title": "Cancelled" },
+          "task_id": {
+            "anyOf": [{ "type": "string" }, { "type": "null" }],
+            "title": "Task Id"
+          },
+          "reason": {
+            "anyOf": [{ "type": "string" }, { "type": "null" }],
+            "title": "Reason"
+          }
+        },
+        "type": "object",
+        "required": ["cancelled"],
+        "title": "CancelTaskResponse",
+        "description": "Response model for the cancel task endpoint."
+      },
       "ChangelogEntry": {
         "properties": {
           "version": { "type": "string", "title": "Version" },