feat(chat): improve agent generator error propagation (#11884)

## Summary
- Add helper functions in `service.py` to create standardized error
responses with `error_type` classification
- Update service functions to return error dicts instead of `None`,
preserving error details from the Agent Generator microservice
- Update `core.py` to pass through error responses properly
- Update `create_agent.py` to handle error responses with user-friendly
messages based on error type

## Error Types Now Propagated
| Error Type | Description | User Message |
|------------|-------------|--------------|
| `llm_parse_error` | LLM returned unparseable response | "The AI had
trouble understanding this request" |
| `llm_timeout` / `timeout` | Request timed out | "The request took too
long" |
| `llm_rate_limit` / `rate_limit` | Rate limited | "The service is
currently busy" |
| `validation_error` | Agent validation failed | "The generated agent
failed validation" |
| `connection_error` | Could not connect to Agent Generator | Generic
error message |
| `http_error` | HTTP error from Agent Generator | Generic error message
|
| `unknown` | Unclassified error | Generic error message |

## Motivation
This enables better debugging for issues like SECRT-1817 where
decomposition failed due to transient LLM errors but the root cause was
unclear in the logs. Now:
1. Error details from the Agent Generator microservice are preserved
2. Users get more helpful error messages based on error type
3. Debugging is easier with `error_type` in response details

## Related PR
- Agent Generator side:
https://github.com/Significant-Gravitas/AutoGPT-Agent-Generator/pull/102

## Test Plan
- [ ] Test decomposition with various error scenarios (timeout, parse
error)
- [ ] Verify user-friendly messages are shown based on error type
- [ ] Check that error details are logged properly

.github/copilot-instructions.md

@@ -160,7 +160,7 @@ pnpm storybook                      # Start component development server
 
 **Backend Entry Points:**
 
-- `backend/backend/server/server.py` - FastAPI application setup
+- `backend/backend/api/rest_api.py` - FastAPI application setup
 - `backend/backend/data/` - Database models and user management
 - `backend/blocks/` - Agent execution blocks and logic
 
@@ -219,7 +219,7 @@ Agents are built using a visual block-based system where each block performs a s
 
 ### API Development
 
-1. Update routes in `/backend/backend/server/routers/`
+1. Update routes in `/backend/backend/api/features/`
 2. Add/update Pydantic models in same directory
 3. Write tests alongside route files
 4. For `data/*.py` changes, validate user ID checks
@@ -285,7 +285,7 @@ Agents are built using a visual block-based system where each block performs a s
 
 ### Security Guidelines
 
-**Cache Protection Middleware** (`/backend/backend/server/middleware/security.py`):
+**Cache Protection Middleware** (`/backend/backend/api/middleware/security.py`):
 
 - Default: Disables caching for ALL endpoints with `Cache-Control: no-store, no-cache, must-revalidate, private`
 - Uses allow list approach for cacheable paths (static assets, health checks, public pages)

.gitignore

@@ -178,4 +178,5 @@ autogpt_platform/backend/settings.py
 *.ign.*
 .test-contents
 .claude/settings.local.json
+CLAUDE.local.md
 /autogpt_platform/backend/logs

autogpt_platform/CLAUDE.md

@@ -6,141 +6,30 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 AutoGPT Platform is a monorepo containing:
 
-- **Backend** (`/backend`): Python FastAPI server with async support
-- **Frontend** (`/frontend`): Next.js React application
-- **Shared Libraries** (`/autogpt_libs`): Common Python utilities
+- **Backend** (`backend`): Python FastAPI server with async support
+- **Frontend** (`frontend`): Next.js React application
+- **Shared Libraries** (`autogpt_libs`): Common Python utilities
 
-## Essential Commands
+## Component Documentation
 
-### Backend Development
+- **Backend**: See @backend/CLAUDE.md for backend-specific commands, architecture, and development tasks
+- **Frontend**: See @frontend/CLAUDE.md for frontend-specific commands, architecture, and development patterns
 
-```bash
-# Install dependencies
-cd backend && poetry install
-
-# Run database migrations
-poetry run prisma migrate dev
-
-# Start all services (database, redis, rabbitmq, clamav)
-docker compose up -d
-
-# Run the backend server
-poetry run serve
-
-# Run tests
-poetry run test
-
-# Run specific test
-poetry run pytest path/to/test_file.py::test_function_name
-
-# Run block tests (tests that validate all blocks work correctly)
-poetry run pytest backend/blocks/test/test_block.py -xvs
-
-# Run tests for a specific block (e.g., GetCurrentTimeBlock)
-poetry run pytest 'backend/blocks/test/test_block.py::test_available_blocks[GetCurrentTimeBlock]' -xvs
-
-# Lint and format
-# prefer format if you want to just "fix" it and only get the errors that can't be autofixed
-poetry run format  # Black + isort
-poetry run lint    # ruff
-```
-
-More details can be found in TESTING.md
-
-#### Creating/Updating Snapshots
-
-When you first write a test or when the expected output changes:
-
-```bash
-poetry run pytest path/to/test.py --snapshot-update
-```
-
-⚠️ **Important**: Always review snapshot changes before committing! Use `git diff` to verify the changes are expected.
-
-### Frontend Development
-
-```bash
-# Install dependencies
-cd frontend && pnpm i
-
-# Generate API client from OpenAPI spec
-pnpm generate:api
-
-# Start development server
-pnpm dev
-
-# Run E2E tests
-pnpm test
-
-# Run Storybook for component development
-pnpm storybook
-
-# Build production
-pnpm build
-
-# Format and lint
-pnpm format
-
-# Type checking
-pnpm types
-```
-
-## Architecture Overview
-
-### Backend Architecture
-
-- **API Layer**: FastAPI with REST and WebSocket endpoints
-- **Database**: PostgreSQL with Prisma ORM, includes pgvector for embeddings
-- **Queue System**: RabbitMQ for async task processing
-- **Execution Engine**: Separate executor service processes agent workflows
-- **Authentication**: JWT-based with Supabase integration
-- **Security**: Cache protection middleware prevents sensitive data caching in browsers/proxies
-
-### Frontend Architecture
-
-- **Framework**: Next.js 15 App Router (client-first approach)
-- **Data Fetching**: Type-safe generated API hooks via Orval + React Query
-- **State Management**: React Query for server state, co-located UI state in components/hooks
-- **Component Structure**: Separate render logic (`.tsx`) from business logic (`use*.ts` hooks)
-- **Workflow Builder**: Visual graph editor using @xyflow/react
-- **UI Components**: shadcn/ui (Radix UI primitives) with Tailwind CSS styling
-- **Icons**: Phosphor Icons only
-- **Feature Flags**: LaunchDarkly integration
-- **Error Handling**: ErrorCard for render errors, toast for mutations, Sentry for exceptions
-- **Testing**: Playwright for E2E, Storybook for component development
-
-### Key Concepts
+## Key Concepts
 
 1. **Agent Graphs**: Workflow definitions stored as JSON, executed by the backend
-2. **Blocks**: Reusable components in `/backend/blocks/` that perform specific tasks
+2. **Blocks**: Reusable components in `backend/backend/blocks/` that perform specific tasks
 3. **Integrations**: OAuth and API connections stored per user
 4. **Store**: Marketplace for sharing agent templates
 5. **Virus Scanning**: ClamAV integration for file upload security
 
-### Testing Approach
-
-- Backend uses pytest with snapshot testing for API responses
-- Test files are colocated with source files (`*_test.py`)
-- Frontend uses Playwright for E2E tests
-- Component testing via Storybook
-
-### Database Schema
-
-Key models (defined in `/backend/schema.prisma`):
-
-- `User`: Authentication and profile data
-- `AgentGraph`: Workflow definitions with version control
-- `AgentGraphExecution`: Execution history and results
-- `AgentNode`: Individual nodes in a workflow
-- `StoreListing`: Marketplace listings for sharing agents
-
 ### Environment Configuration
 
 #### Configuration Files
 
-- **Backend**: `/backend/.env.default` (defaults) → `/backend/.env` (user overrides)
-- **Frontend**: `/frontend/.env.default` (defaults) → `/frontend/.env` (user overrides)
-- **Platform**: `/.env.default` (Supabase/shared defaults) → `/.env` (user overrides)
+- **Backend**: `backend/.env.default` (defaults) → `backend/.env` (user overrides)
+- **Frontend**: `frontend/.env.default` (defaults) → `frontend/.env` (user overrides)
+- **Platform**: `.env.default` (Supabase/shared defaults) → `.env` (user overrides)
 
 #### Docker Environment Loading Order
 
@@ -156,130 +45,12 @@ Key models (defined in `/backend/schema.prisma`):
 - Backend/Frontend services use YAML anchors for consistent configuration
 - Supabase services (`db/docker/docker-compose.yml`) follow the same pattern
 
-### Common Development Tasks
-
-**Adding a new block:**
-
-Follow the comprehensive [Block SDK Guide](../../../docs/content/platform/block-sdk-guide.md) which covers:
-
-- Provider configuration with `ProviderBuilder`
-- Block schema definition
-- Authentication (API keys, OAuth, webhooks)
-- Testing and validation
-- File organization
-
-Quick steps:
-
-1. Create new file in `/backend/backend/blocks/`
-2. Configure provider using `ProviderBuilder` in `_config.py`
-3. Inherit from `Block` base class
-4. Define input/output schemas using `BlockSchema`
-5. Implement async `run` method
-6. Generate unique block ID using `uuid.uuid4()`
-7. Test with `poetry run pytest backend/blocks/test/test_block.py`
-
-Note: when making many new blocks analyze the interfaces for each of these blocks and picture if they would go well together in a graph based editor or would they struggle to connect productively?
-ex: do the inputs and outputs tie well together?
-
-If you get any pushback or hit complex block conditions check the new_blocks guide in the docs.
-
-**Handling files in blocks with `store_media_file()`:**
-
-When blocks need to work with files (images, videos, documents), use `store_media_file()` from `backend.util.file`. The `return_format` parameter determines what you get back:
-
-| Format | Use When | Returns |
-|--------|----------|---------|
-| `"for_local_processing"` | Processing with local tools (ffmpeg, MoviePy, PIL) | Local file path (e.g., `"image.png"`) |
-| `"for_external_api"` | Sending content to external APIs (Replicate, OpenAI) | Data URI (e.g., `"data:image/png;base64,..."`) |
-| `"for_block_output"` | Returning output from your block | Smart: `workspace://` in CoPilot, data URI in graphs |
-
-**Examples:**
-```python
-# INPUT: Need to process file locally with ffmpeg
-local_path = await store_media_file(
-    file=input_data.video,
-    execution_context=execution_context,
-    return_format="for_local_processing",
-)
-# local_path = "video.mp4" - use with Path/ffmpeg/etc
-
-# INPUT: Need to send to external API like Replicate
-image_b64 = await store_media_file(
-    file=input_data.image,
-    execution_context=execution_context,
-    return_format="for_external_api",
-)
-# image_b64 = "data:image/png;base64,iVBORw0..." - send to API
-
-# OUTPUT: Returning result from block
-result_url = await store_media_file(
-    file=generated_image_url,
-    execution_context=execution_context,
-    return_format="for_block_output",
-)
-yield "image_url", result_url
-# In CoPilot: result_url = "workspace://abc123"
-# In graphs:  result_url = "data:image/png;base64,..."
-```
-
-**Key points:**
-- `for_block_output` is the ONLY format that auto-adapts to execution context
-- Always use `for_block_output` for block outputs unless you have a specific reason not to
-- Never hardcode workspace checks - let `for_block_output` handle it
-
-**Modifying the API:**
-
-1. Update route in `/backend/backend/server/routers/`
-2. Add/update Pydantic models in same directory
-3. Write tests alongside the route file
-4. Run `poetry run test` to verify
-
-### Frontend guidelines:
-
-See `/frontend/CONTRIBUTING.md` for complete patterns. Quick reference:
-
-1. **Pages**: Create in `src/app/(platform)/feature-name/page.tsx`
-   - Add `usePageName.ts` hook for logic
-   - Put sub-components in local `components/` folder
-2. **Components**: Structure as `ComponentName/ComponentName.tsx` + `useComponentName.ts` + `helpers.ts`
-   - Use design system components from `src/components/` (atoms, molecules, organisms)
-   - Never use `src/components/__legacy__/*`
-3. **Data fetching**: Use generated API hooks from `@/app/api/__generated__/endpoints/`
-   - Regenerate with `pnpm generate:api`
-   - Pattern: `use{Method}{Version}{OperationName}`
-4. **Styling**: Tailwind CSS only, use design tokens, Phosphor Icons only
-5. **Testing**: Add Storybook stories for new components, Playwright for E2E
-6. **Code conventions**: Function declarations (not arrow functions) for components/handlers
-
-- Component props should be `interface Props { ... }` (not exported) unless the interface needs to be used outside the component
-- Separate render logic from business logic (component.tsx + useComponent.ts + helpers.ts)
-- Colocate state when possible and avoid creating large components, use sub-components ( local `/components` folder next to the parent component ) when sensible
-- Avoid large hooks, abstract logic into `helpers.ts` files when sensible
-- Use function declarations for components, arrow functions only for callbacks
-- No barrel files or `index.ts` re-exports
-- Do not use `useCallback` or `useMemo` unless asked to optimise a given function
-- Avoid comments at all times unless the code is very complex
-- Do not type hook returns, let Typescript infer as much as possible
-- Never type with `any`, if not types available use `unknown`
-
-### Security Implementation
-
-**Cache Protection Middleware:**
-
-- Located in `/backend/backend/server/middleware/security.py`
-- Default behavior: Disables caching for ALL endpoints with `Cache-Control: no-store, no-cache, must-revalidate, private`
-- Uses an allow list approach - only explicitly permitted paths can be cached
-- Cacheable paths include: static assets (`/static/*`, `/_next/static/*`), health checks, public store pages, documentation
-- Prevents sensitive data (auth tokens, API keys, user data) from being cached by browsers/proxies
-- To allow caching for a new endpoint, add it to `CACHEABLE_PATHS` in the middleware
-- Applied to both main API server and external API applications
-
 ### Creating Pull Requests
 
-- Create the PR aginst the `dev` branch of the repository.
-- Ensure the branch name is descriptive (e.g., `feature/add-new-block`)/
-- Use conventional commit messages (see below)/
-- Fill out the .github/PULL_REQUEST_TEMPLATE.md template as the PR description/
+- Create the PR against the `dev` branch of the repository.
+- Ensure the branch name is descriptive (e.g., `feature/add-new-block`)
+- Use conventional commit messages (see below)
+- Fill out the .github/PULL_REQUEST_TEMPLATE.md template as the PR description
 - Run the github pre-commit hooks to ensure code quality.
 
 ### Reviewing/Revising Pull Requests

autogpt_platform/backend/CLAUDE.md

@@ -0,0 +1,170 @@
+# CLAUDE.md - Backend
+
+This file provides guidance to Claude Code when working with the backend.
+
+## Essential Commands
+
+To run something with Python package dependencies you MUST use `poetry run ...`.
+
+```bash
+# Install dependencies
+poetry install
+
+# Run database migrations
+poetry run prisma migrate dev
+
+# Start all services (database, redis, rabbitmq, clamav)
+docker compose up -d
+
+# Run the backend as a whole
+poetry run app
+
+# Run tests
+poetry run test
+
+# Run specific test
+poetry run pytest path/to/test_file.py::test_function_name
+
+# Run block tests (tests that validate all blocks work correctly)
+poetry run pytest backend/blocks/test/test_block.py -xvs
+
+# Run tests for a specific block (e.g., GetCurrentTimeBlock)
+poetry run pytest 'backend/blocks/test/test_block.py::test_available_blocks[GetCurrentTimeBlock]' -xvs
+
+# Lint and format
+# prefer format if you want to just "fix" it and only get the errors that can't be autofixed
+poetry run format  # Black + isort
+poetry run lint    # ruff
+```
+
+More details can be found in @TESTING.md
+
+### Creating/Updating Snapshots
+
+When you first write a test or when the expected output changes:
+
+```bash
+poetry run pytest path/to/test.py --snapshot-update
+```
+
+⚠️ **Important**: Always review snapshot changes before committing! Use `git diff` to verify the changes are expected.
+
+## Architecture
+
+- **API Layer**: FastAPI with REST and WebSocket endpoints
+- **Database**: PostgreSQL with Prisma ORM, includes pgvector for embeddings
+- **Queue System**: RabbitMQ for async task processing
+- **Execution Engine**: Separate executor service processes agent workflows
+- **Authentication**: JWT-based with Supabase integration
+- **Security**: Cache protection middleware prevents sensitive data caching in browsers/proxies
+
+## Testing Approach
+
+- Uses pytest with snapshot testing for API responses
+- Test files are colocated with source files (`*_test.py`)
+
+## Database Schema
+
+Key models (defined in `schema.prisma`):
+
+- `User`: Authentication and profile data
+- `AgentGraph`: Workflow definitions with version control
+- `AgentGraphExecution`: Execution history and results
+- `AgentNode`: Individual nodes in a workflow
+- `StoreListing`: Marketplace listings for sharing agents
+
+## Environment Configuration
+
+- **Backend**: `.env.default` (defaults) → `.env` (user overrides)
+
+## Common Development Tasks
+
+### Adding a new block
+
+Follow the comprehensive [Block SDK Guide](@../../docs/content/platform/block-sdk-guide.md) which covers:
+
+- Provider configuration with `ProviderBuilder`
+- Block schema definition
+- Authentication (API keys, OAuth, webhooks)
+- Testing and validation
+- File organization
+
+Quick steps:
+
+1. Create new file in `backend/blocks/`
+2. Configure provider using `ProviderBuilder` in `_config.py`
+3. Inherit from `Block` base class
+4. Define input/output schemas using `BlockSchema`
+5. Implement async `run` method
+6. Generate unique block ID using `uuid.uuid4()`
+7. Test with `poetry run pytest backend/blocks/test/test_block.py`
+
+Note: when making many new blocks analyze the interfaces for each of these blocks and picture if they would go well together in a graph-based editor or would they struggle to connect productively?
+ex: do the inputs and outputs tie well together?
+
+If you get any pushback or hit complex block conditions check the new_blocks guide in the docs.
+
+#### Handling files in blocks with `store_media_file()`
+
+When blocks need to work with files (images, videos, documents), use `store_media_file()` from `backend.util.file`. The `return_format` parameter determines what you get back:
+
+| Format | Use When | Returns |
+|--------|----------|---------|
+| `"for_local_processing"` | Processing with local tools (ffmpeg, MoviePy, PIL) | Local file path (e.g., `"image.png"`) |
+| `"for_external_api"` | Sending content to external APIs (Replicate, OpenAI) | Data URI (e.g., `"data:image/png;base64,..."`) |
+| `"for_block_output"` | Returning output from your block | Smart: `workspace://` in CoPilot, data URI in graphs |
+
+**Examples:**
+
+```python
+# INPUT: Need to process file locally with ffmpeg
+local_path = await store_media_file(
+    file=input_data.video,
+    execution_context=execution_context,
+    return_format="for_local_processing",
+)
+# local_path = "video.mp4" - use with Path/ffmpeg/etc
+
+# INPUT: Need to send to external API like Replicate
+image_b64 = await store_media_file(
+    file=input_data.image,
+    execution_context=execution_context,
+    return_format="for_external_api",
+)
+# image_b64 = "data:image/png;base64,iVBORw0..." - send to API
+
+# OUTPUT: Returning result from block
+result_url = await store_media_file(
+    file=generated_image_url,
+    execution_context=execution_context,
+    return_format="for_block_output",
+)
+yield "image_url", result_url
+# In CoPilot: result_url = "workspace://abc123"
+# In graphs:  result_url = "data:image/png;base64,..."
+```
+
+**Key points:**
+
+- `for_block_output` is the ONLY format that auto-adapts to execution context
+- Always use `for_block_output` for block outputs unless you have a specific reason not to
+- Never hardcode workspace checks - let `for_block_output` handle it
+
+### Modifying the API
+
+1. Update route in `backend/api/features/`
+2. Add/update Pydantic models in same directory
+3. Write tests alongside the route file
+4. Run `poetry run test` to verify
+
+## Security Implementation
+
+### Cache Protection Middleware
+
+- Located in `backend/api/middleware/security.py`
+- Default behavior: Disables caching for ALL endpoints with `Cache-Control: no-store, no-cache, must-revalidate, private`
+- Uses an allow list approach - only explicitly permitted paths can be cached
+- Cacheable paths include: static assets (`static/*`, `_next/static/*`), health checks, public store pages, documentation
+- Prevents sensitive data (auth tokens, API keys, user data) from being cached by browsers/proxies
+- To allow caching for a new endpoint, add it to `CACHEABLE_PATHS` in the middleware
+- Applied to both main API server and external API applications

autogpt_platform/backend/TESTING.md

@@ -138,7 +138,7 @@ If the test doesn't need the `user_id` specifically, mocking is not necessary as
 
 #### Using Global Auth Fixtures
 
-Two global auth fixtures are provided by `backend/server/conftest.py`:
+Two global auth fixtures are provided by `backend/api/conftest.py`:
 
 - `mock_jwt_user` - Regular user with `test_user_id` ("test-user-id")
 - `mock_jwt_admin` - Admin user with `admin_user_id` ("admin-user-id")

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

@@ -17,7 +17,7 @@ router = fastapi.APIRouter(
 )
 
 
-# Taken from backend/server/v2/store/db.py
+# Taken from backend/api/features/store/db.py
 def sanitize_query(query: str | None) -> str | None:
     if query is None:
         return query

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

@@ -1,325 +0,0 @@
-"""RabbitMQ consumer for operation completion messages.
-
-This module provides a consumer that listens for completion notifications
-from external services (like Agent Generator) and triggers the appropriate
-stream registry and chat service updates.
-"""
-
-import asyncio
-import logging
-
-import orjson
-from pydantic import BaseModel
-
-from backend.data.rabbitmq import (
-    AsyncRabbitMQ,
-    Exchange,
-    ExchangeType,
-    Queue,
-    RabbitMQConfig,
-)
-
-from . import service as chat_service
-from . import stream_registry
-from .response_model import StreamError, StreamFinish, StreamToolOutputAvailable
-from .tools.models import ErrorResponse
-
-logger = logging.getLogger(__name__)
-
-# Queue and exchange configuration
-OPERATION_COMPLETE_EXCHANGE = Exchange(
-    name="chat_operations",
-    type=ExchangeType.DIRECT,
-    durable=True,
-)
-
-OPERATION_COMPLETE_QUEUE = Queue(
-    name="chat_operation_complete",
-    durable=True,
-    exchange=OPERATION_COMPLETE_EXCHANGE,
-    routing_key="operation.complete",
-)
-
-RABBITMQ_CONFIG = RabbitMQConfig(
-    exchanges=[OPERATION_COMPLETE_EXCHANGE],
-    queues=[OPERATION_COMPLETE_QUEUE],
-)
-
-
-class OperationCompleteMessage(BaseModel):
-    """Message format for operation completion notifications."""
-
-    operation_id: str
-    task_id: str
-    success: bool
-    result: dict | str | None = None
-    error: str | None = None
-
-
-class ChatCompletionConsumer:
-    """Consumer for chat operation completion messages from RabbitMQ."""
-
-    def __init__(self):
-        self._rabbitmq: AsyncRabbitMQ | None = None
-        self._consumer_task: asyncio.Task | None = None
-        self._running = False
-
-    async def start(self) -> None:
-        """Start the completion consumer."""
-        if self._running:
-            logger.warning("Completion consumer already running")
-            return
-
-        self._rabbitmq = AsyncRabbitMQ(RABBITMQ_CONFIG)
-        await self._rabbitmq.connect()
-
-        self._running = True
-        self._consumer_task = asyncio.create_task(self._consume_messages())
-        logger.info("Chat completion consumer started")
-
-    async def stop(self) -> None:
-        """Stop the completion consumer."""
-        self._running = False
-
-        if self._consumer_task:
-            self._consumer_task.cancel()
-            try:
-                await self._consumer_task
-            except asyncio.CancelledError:
-                pass
-            self._consumer_task = None
-
-        if self._rabbitmq:
-            await self._rabbitmq.disconnect()
-            self._rabbitmq = None
-
-        logger.info("Chat completion consumer stopped")
-
-    async def _consume_messages(self) -> None:
-        """Main message consumption loop with retry logic."""
-        max_retries = 10
-        retry_delay = 5  # seconds
-        retry_count = 0
-
-        while self._running and retry_count < max_retries:
-            if not self._rabbitmq:
-                logger.error("RabbitMQ not initialized")
-                return
-
-            try:
-                channel = await self._rabbitmq.get_channel()
-                queue = await channel.get_queue(OPERATION_COMPLETE_QUEUE.name)
-
-                # Reset retry count on successful connection
-                retry_count = 0
-
-                async with queue.iterator() as queue_iter:
-                    async for message in queue_iter:
-                        if not self._running:
-                            return
-
-                        try:
-                            async with message.process():
-                                await self._handle_message(message.body)
-                        except Exception as e:
-                            logger.error(
-                                f"Error processing completion message: {e}",
-                                exc_info=True,
-                            )
-                            # Message will be requeued due to exception
-
-            except asyncio.CancelledError:
-                logger.info("Consumer cancelled")
-                return
-            except Exception as e:
-                retry_count += 1
-                logger.error(
-                    f"Consumer error (retry {retry_count}/{max_retries}): {e}",
-                    exc_info=True,
-                )
-                if self._running and retry_count < max_retries:
-                    await asyncio.sleep(retry_delay)
-                else:
-                    logger.error("Max retries reached, stopping consumer")
-                    return
-
-    async def _handle_message(self, body: bytes) -> None:
-        """Handle a single completion message."""
-        try:
-            data = orjson.loads(body)
-            message = OperationCompleteMessage(**data)
-        except Exception as e:
-            logger.error(f"Failed to parse completion message: {e}")
-            return
-
-        logger.info(
-            f"Received completion for operation {message.operation_id} "
-            f"(task_id={message.task_id}, success={message.success})"
-        )
-
-        # Find task in registry
-        task = await stream_registry.find_task_by_operation_id(message.operation_id)
-        if task is None:
-            # Try to look up by task_id directly
-            task = await stream_registry.get_task(message.task_id)
-
-        if task is None:
-            logger.warning(
-                f"Task not found for operation {message.operation_id} "
-                f"(task_id={message.task_id})"
-            )
-            return
-
-        if message.success:
-            await self._handle_success(task, message)
-        else:
-            await self._handle_failure(task, message)
-
-    async def _handle_success(
-        self,
-        task: stream_registry.ActiveTask,
-        message: OperationCompleteMessage,
-    ) -> None:
-        """Handle successful operation completion."""
-        # Publish result to stream registry
-        result_output = message.result if message.result else {"status": "completed"}
-        await stream_registry.publish_chunk(
-            task.task_id,
-            StreamToolOutputAvailable(
-                toolCallId=task.tool_call_id,
-                toolName=task.tool_name,
-                output=(
-                    result_output
-                    if isinstance(result_output, str)
-                    else orjson.dumps(result_output).decode("utf-8")
-                ),
-                success=True,
-            ),
-        )
-
-        # Update pending operation in database
-        result_str = (
-            message.result
-            if isinstance(message.result, str)
-            else (
-                orjson.dumps(message.result).decode("utf-8")
-                if message.result
-                else '{"status": "completed"}'
-            )
-        )
-        await chat_service._update_pending_operation(
-            session_id=task.session_id,
-            tool_call_id=task.tool_call_id,
-            result=result_str,
-        )
-
-        # Generate LLM continuation with streaming
-        await chat_service._generate_llm_continuation_with_streaming(
-            session_id=task.session_id,
-            user_id=task.user_id,
-            task_id=task.task_id,
-        )
-
-        # Mark task as completed and release Redis lock
-        await stream_registry.mark_task_completed(task.task_id, status="completed")
-        await chat_service._mark_operation_completed(task.tool_call_id)
-
-        logger.info(
-            f"Successfully processed completion for task {task.task_id} "
-            f"(operation {message.operation_id})"
-        )
-
-    async def _handle_failure(
-        self,
-        task: stream_registry.ActiveTask,
-        message: OperationCompleteMessage,
-    ) -> None:
-        """Handle failed operation completion."""
-        error_msg = message.error or "Operation failed"
-
-        # Publish error to stream registry followed by finish event
-        await stream_registry.publish_chunk(
-            task.task_id,
-            StreamError(errorText=error_msg),
-        )
-        await stream_registry.publish_chunk(task.task_id, StreamFinish())
-
-        # Update pending operation with error
-        error_response = ErrorResponse(
-            message=error_msg,
-            error=message.error,
-        )
-        await chat_service._update_pending_operation(
-            session_id=task.session_id,
-            tool_call_id=task.tool_call_id,
-            result=error_response.model_dump_json(),
-        )
-
-        # Mark task as failed and release Redis lock
-        await stream_registry.mark_task_completed(task.task_id, status="failed")
-        await chat_service._mark_operation_completed(task.tool_call_id)
-
-        logger.info(
-            f"Processed failure for task {task.task_id} "
-            f"(operation {message.operation_id}): {error_msg}"
-        )
-
-
-# Module-level consumer instance
-_consumer: ChatCompletionConsumer | None = None
-
-
-async def start_completion_consumer() -> None:
-    """Start the global completion consumer."""
-    global _consumer
-    if _consumer is None:
-        _consumer = ChatCompletionConsumer()
-    await _consumer.start()
-
-
-async def stop_completion_consumer() -> None:
-    """Stop the global completion consumer."""
-    global _consumer
-    if _consumer:
-        await _consumer.stop()
-        _consumer = None
-
-
-async def publish_operation_complete(
-    operation_id: str,
-    task_id: str,
-    success: bool,
-    result: dict | str | None = None,
-    error: str | None = None,
-) -> None:
-    """Publish an operation completion message.
-
-    This is a helper function for testing or for services that want to
-    publish completion messages directly.
-
-    Args:
-        operation_id: The operation ID that completed.
-        task_id: The task ID associated with the operation.
-        success: Whether the operation succeeded.
-        result: The result data (for success).
-        error: The error message (for failure).
-    """
-    message = OperationCompleteMessage(
-        operation_id=operation_id,
-        task_id=task_id,
-        success=success,
-        result=result,
-        error=error,
-    )
-
-    rabbitmq = AsyncRabbitMQ(RABBITMQ_CONFIG)
-    try:
-        await rabbitmq.connect()
-        await rabbitmq.publish_message(
-            routing_key="operation.complete",
-            message=message.model_dump_json(),
-            exchange=OPERATION_COMPLETE_EXCHANGE,
-        )
-        logger.info(f"Published completion for operation {operation_id}")
-    finally:
-        await rabbitmq.disconnect()

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

@@ -44,20 +44,6 @@ class ChatConfig(BaseSettings):
         description="TTL in seconds for long-running operation tracking in Redis (safety net if pod dies)",
     )
 
-    # Stream registry configuration for SSE reconnection
-    stream_ttl: int = Field(
-        default=3600,
-        description="TTL in seconds for stream data in Redis (1 hour)",
-    )
-    stream_max_length: int = Field(
-        default=1000,
-        description="Maximum number of messages to store per stream",
-    )
-    internal_api_key: str | None = Field(
-        default=None,
-        description="API key for internal webhook callbacks (env: CHAT_INTERNAL_API_KEY)",
-    )
-
     # Langfuse Prompt Management Configuration
     # Note: Langfuse credentials are in Settings().secrets (settings.py)
     langfuse_prompt_name: str = Field(
@@ -96,14 +82,6 @@ class ChatConfig(BaseSettings):
                 v = "https://openrouter.ai/api/v1"
         return v
 
-    @field_validator("internal_api_key", mode="before")
-    @classmethod
-    def get_internal_api_key(cls, v):
-        """Get internal API key from environment if not provided."""
-        if v is None:
-            v = os.getenv("CHAT_INTERNAL_API_KEY")
-        return v
-
     # Prompt paths for different contexts
     PROMPT_PATHS: dict[str, str] = {
         "default": "prompts/chat_system.md",

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

@@ -4,19 +4,16 @@ import logging
 from collections.abc import AsyncGenerator
 from typing import Annotated
 
-import orjson
 from autogpt_libs import auth
-from fastapi import APIRouter, Depends, Header, HTTPException, Query, Security
+from fastapi import APIRouter, Depends, Query, Security
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
 from backend.util.exceptions import NotFoundError
 
 from . import service as chat_service
-from . import stream_registry
 from .config import ChatConfig
 from .model import ChatSession, create_chat_session, get_chat_session, get_user_sessions
-from .response_model import StreamFinish, StreamHeartbeat
 
 config = ChatConfig()
 
@@ -84,14 +81,6 @@ class ListSessionsResponse(BaseModel):
     total: int
 
 
-class OperationCompleteRequest(BaseModel):
-    """Request model for external completion webhook."""
-
-    success: bool
-    result: dict | str | None = None
-    error: str | None = None
-
-
 # ========== Routes ==========
 
 
@@ -377,267 +366,6 @@ async def session_assign_user(
     return {"status": "ok"}
 
 
-# ========== Task Streaming (SSE Reconnection) ==========
-
-
-@router.get(
-    "/tasks/{task_id}/stream",
-)
-async def stream_task(
-    task_id: str,
-    user_id: str | None = Depends(auth.get_user_id),
-    last_message_id: str = Query(
-        default="0-0",
-        description="Last Redis Stream message ID received (e.g., '1706540123456-0'). Use '0-0' for full replay.",
-    ),
-):
-    """
-    Reconnect to a long-running task's SSE stream.
-
-    When a long-running operation (like agent generation) starts, the client
-    receives a task_id. If the connection drops, the client can reconnect
-    using this endpoint to resume receiving updates.
-
-    Args:
-        task_id: The task ID from the operation_started response.
-        user_id: Authenticated user ID for ownership validation.
-        last_message_id: Last Redis Stream message ID received ("0-0" for full replay).
-
-    Returns:
-        StreamingResponse: SSE-formatted response chunks starting after last_message_id.
-
-    Raises:
-        NotFoundError: If task_id is not found or user doesn't have access.
-    """
-    # Get subscriber queue from stream registry
-    subscriber_queue = await stream_registry.subscribe_to_task(
-        task_id=task_id,
-        user_id=user_id,
-        last_message_id=last_message_id,
-    )
-
-    if subscriber_queue is None:
-        raise NotFoundError(f"Task {task_id} not found or access denied.")
-
-    async def event_generator() -> AsyncGenerator[str, None]:
-        import asyncio
-
-        chunk_count = 0
-        heartbeat_interval = 15.0  # Send heartbeat every 15 seconds
-        try:
-            while True:
-                try:
-                    # Wait for next chunk with timeout for heartbeats
-                    chunk = await asyncio.wait_for(
-                        subscriber_queue.get(), timeout=heartbeat_interval
-                    )
-                    chunk_count += 1
-                    yield chunk.to_sse()
-
-                    # Check for finish signal
-                    if isinstance(chunk, StreamFinish):
-                        logger.info(
-                            f"Task stream completed for task {task_id}, "
-                            f"chunk_count={chunk_count}"
-                        )
-                        break
-                except asyncio.TimeoutError:
-                    # Send heartbeat to keep connection alive
-                    yield StreamHeartbeat().to_sse()
-        except Exception as e:
-            logger.error(f"Error in task stream {task_id}: {e}", exc_info=True)
-        finally:
-            # Unsubscribe when client disconnects or stream ends
-            await stream_registry.unsubscribe_from_task(task_id, subscriber_queue)
-
-        # AI SDK protocol termination
-        yield "data: [DONE]\n\n"
-
-    return StreamingResponse(
-        event_generator(),
-        media_type="text/event-stream",
-        headers={
-            "Cache-Control": "no-cache",
-            "Connection": "keep-alive",
-            "X-Accel-Buffering": "no",
-            "x-vercel-ai-ui-message-stream": "v1",
-        },
-    )
-
-
-@router.get(
-    "/tasks/{task_id}",
-)
-async def get_task_status(
-    task_id: str,
-    user_id: str | None = Depends(auth.get_user_id),
-) -> dict:
-    """
-    Get the status of a long-running task.
-
-    Args:
-        task_id: The task ID to check.
-        user_id: Authenticated user ID for ownership validation.
-
-    Returns:
-        dict: Task status including task_id, status, tool_name, and operation_id.
-
-    Raises:
-        NotFoundError: If task_id is not found or user doesn't have access.
-    """
-    task = await stream_registry.get_task(task_id)
-
-    if task is None:
-        raise NotFoundError(f"Task {task_id} not found.")
-
-    # Validate ownership
-    if user_id and task.user_id and task.user_id != user_id:
-        raise NotFoundError(f"Task {task_id} not found.")
-
-    return {
-        "task_id": task.task_id,
-        "session_id": task.session_id,
-        "status": task.status,
-        "tool_name": task.tool_name,
-        "operation_id": task.operation_id,
-        "created_at": task.created_at.isoformat(),
-    }
-
-
-# ========== External Completion Webhook ==========
-
-
-@router.post(
-    "/operations/{operation_id}/complete",
-    status_code=200,
-)
-async def complete_operation(
-    operation_id: str,
-    request: OperationCompleteRequest,
-    x_api_key: str | None = Header(default=None),
-) -> dict:
-    """
-    External completion webhook for long-running operations.
-
-    Called by Agent Generator (or other services) when an operation completes.
-    This triggers the stream registry to publish completion and continue LLM generation.
-
-    Args:
-        operation_id: The operation ID to complete.
-        request: Completion payload with success status and result/error.
-        x_api_key: Internal API key for authentication.
-
-    Returns:
-        dict: Status of the completion.
-
-    Raises:
-        HTTPException: If API key is invalid or operation not found.
-    """
-    # Validate internal API key - reject if not configured or invalid
-    if not config.internal_api_key:
-        logger.error(
-            "Operation complete webhook rejected: CHAT_INTERNAL_API_KEY not configured"
-        )
-        raise HTTPException(
-            status_code=503,
-            detail="Webhook not available: internal API key not configured",
-        )
-    if x_api_key != config.internal_api_key:
-        raise HTTPException(status_code=401, detail="Invalid API key")
-
-    # Find task by operation_id
-    task = await stream_registry.find_task_by_operation_id(operation_id)
-    if task is None:
-        raise HTTPException(
-            status_code=404,
-            detail=f"Operation {operation_id} not found",
-        )
-
-    logger.info(
-        f"Received completion webhook for operation {operation_id} "
-        f"(task_id={task.task_id}, success={request.success})"
-    )
-
-    if request.success:
-        # Publish result to stream registry
-        from .response_model import StreamToolOutputAvailable
-
-        result_output = request.result if request.result else {"status": "completed"}
-        await stream_registry.publish_chunk(
-            task.task_id,
-            StreamToolOutputAvailable(
-                toolCallId=task.tool_call_id,
-                toolName=task.tool_name,
-                output=(
-                    result_output
-                    if isinstance(result_output, str)
-                    else orjson.dumps(result_output).decode("utf-8")
-                ),
-                success=True,
-            ),
-        )
-
-        # Update pending operation in database
-        from . import service as svc
-
-        result_str = (
-            request.result
-            if isinstance(request.result, str)
-            else (
-                orjson.dumps(request.result).decode("utf-8")
-                if request.result
-                else '{"status": "completed"}'
-            )
-        )
-        await svc._update_pending_operation(
-            session_id=task.session_id,
-            tool_call_id=task.tool_call_id,
-            result=result_str,
-        )
-
-        # Generate LLM continuation with streaming
-        await svc._generate_llm_continuation_with_streaming(
-            session_id=task.session_id,
-            user_id=task.user_id,
-            task_id=task.task_id,
-        )
-
-        # Mark task as completed and release Redis lock
-        await stream_registry.mark_task_completed(task.task_id, status="completed")
-        await svc._mark_operation_completed(task.tool_call_id)
-    else:
-        # Publish error to stream registry
-        from .response_model import StreamError
-
-        error_msg = request.error or "Operation failed"
-        await stream_registry.publish_chunk(
-            task.task_id,
-            StreamError(errorText=error_msg),
-        )
-        # Send finish event to end the stream
-        await stream_registry.publish_chunk(task.task_id, StreamFinish())
-
-        # Update pending operation with error
-        from . import service as svc
-        from .tools.models import ErrorResponse
-
-        error_response = ErrorResponse(
-            message=error_msg,
-            error=request.error,
-        )
-        await svc._update_pending_operation(
-            session_id=task.session_id,
-            tool_call_id=task.tool_call_id,
-            result=error_response.model_dump_json(),
-        )
-
-        # Mark task as failed and release Redis lock
-        await stream_registry.mark_task_completed(task.task_id, status="failed")
-        await svc._mark_operation_completed(task.tool_call_id)
-
-    return {"status": "ok", "task_id": task.task_id}
-
-
 # ========== Health Check ==========
 
 

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

@@ -26,7 +26,6 @@ from backend.util.exceptions import NotFoundError
 from backend.util.settings import Settings
 
 from . import db as chat_db
-from . import stream_registry
 from .config import ChatConfig
 from .model import (
     ChatMessage,
@@ -1611,9 +1610,8 @@ async def _yield_tool_call(
             )
             return
 
-        # Generate operation ID and task ID
+        # Generate operation ID
         operation_id = str(uuid_module.uuid4())
-        task_id = str(uuid_module.uuid4())
 
         # Build a user-friendly message based on tool and arguments
         if tool_name == "create_agent":
@@ -1656,16 +1654,6 @@ async def _yield_tool_call(
 
         # Wrap session save and task creation in try-except to release lock on failure
         try:
-            # Create task in stream registry for SSE reconnection support
-            await stream_registry.create_task(
-                task_id=task_id,
-                session_id=session.session_id,
-                user_id=session.user_id,
-                tool_call_id=tool_call_id,
-                tool_name=tool_name,
-                operation_id=operation_id,
-            )
-
             # Save assistant message with tool_call FIRST (required by LLM)
             assistant_message = ChatMessage(
                 role="assistant",
@@ -1687,27 +1675,23 @@ async def _yield_tool_call(
             session.messages.append(pending_message)
             await upsert_chat_session(session)
             logger.info(
-                f"Saved pending operation {operation_id} (task_id={task_id}) "
-                f"for tool {tool_name} in session {session.session_id}"
+                f"Saved pending operation {operation_id} for tool {tool_name} "
+                f"in session {session.session_id}"
             )
 
             # Store task reference in module-level set to prevent GC before completion
-            bg_task = asyncio.create_task(
-                _execute_long_running_tool_with_streaming(
+            task = asyncio.create_task(
+                _execute_long_running_tool(
                     tool_name=tool_name,
                     parameters=arguments,
                     tool_call_id=tool_call_id,
                     operation_id=operation_id,
-                    task_id=task_id,
                     session_id=session.session_id,
                     user_id=session.user_id,
                 )
             )
-            _background_tasks.add(bg_task)
-            bg_task.add_done_callback(_background_tasks.discard)
-
-            # Associate the asyncio task with the stream registry task
-            await stream_registry.set_task_asyncio_task(task_id, bg_task)
+            _background_tasks.add(task)
+            task.add_done_callback(_background_tasks.discard)
         except Exception as e:
             # Roll back appended messages to prevent data corruption on subsequent saves
             if (
@@ -1725,11 +1709,6 @@ async def _yield_tool_call(
 
             # Release the Redis lock since the background task won't be spawned
             await _mark_operation_completed(tool_call_id)
-            # Mark stream registry task as failed if it was created
-            try:
-                await stream_registry.mark_task_completed(task_id, status="failed")
-            except Exception:
-                pass
             logger.error(
                 f"Failed to setup long-running tool {tool_name}: {e}", exc_info=True
             )
@@ -1743,7 +1722,6 @@ async def _yield_tool_call(
                 message=started_msg,
                 operation_id=operation_id,
                 tool_name=tool_name,
-                task_id=task_id,  # Include task_id for SSE reconnection
             ).model_dump_json(),
             success=True,
         )
@@ -1813,9 +1791,6 @@ async def _execute_long_running_tool(
 
     This function runs independently of the SSE connection, so the operation
     survives if the user closes their browser tab.
-
-    NOTE: This is the legacy function without stream registry support.
-    Use _execute_long_running_tool_with_streaming for new implementations.
     """
     try:
         # Load fresh session (not stale reference)
@@ -1863,128 +1838,6 @@ async def _execute_long_running_tool(
         await _mark_operation_completed(tool_call_id)
 
 
-async def _execute_long_running_tool_with_streaming(
-    tool_name: str,
-    parameters: dict[str, Any],
-    tool_call_id: str,
-    operation_id: str,
-    task_id: str,
-    session_id: str,
-    user_id: str | None,
-) -> None:
-    """Execute a long-running tool with stream registry support for SSE reconnection.
-
-    This function runs independently of the SSE connection, publishes progress
-    to the stream registry, and survives if the user closes their browser tab.
-    Clients can reconnect via GET /chat/tasks/{task_id}/stream to resume streaming.
-
-    If the external service returns a 202 Accepted (async), this function exits
-    early and lets the RabbitMQ completion consumer handle the rest.
-    """
-    # Track whether we delegated to async processing - if so, the RabbitMQ
-    # completion consumer will handle cleanup, not us
-    delegated_to_async = False
-
-    try:
-        # Load fresh session (not stale reference)
-        session = await get_chat_session(session_id, user_id)
-        if not session:
-            logger.error(f"Session {session_id} not found for background tool")
-            await stream_registry.mark_task_completed(task_id, status="failed")
-            return
-
-        # Pass operation_id and task_id to the tool for async processing
-        enriched_parameters = {
-            **parameters,
-            "_operation_id": operation_id,
-            "_task_id": task_id,
-        }
-
-        # Execute the actual tool
-        result = await execute_tool(
-            tool_name=tool_name,
-            parameters=enriched_parameters,
-            tool_call_id=tool_call_id,
-            user_id=user_id,
-            session=session,
-        )
-
-        # Check if the tool result indicates async processing
-        # (e.g., Agent Generator returned 202 Accepted)
-        try:
-            result_data = orjson.loads(result.output) if result.output else {}
-            if result_data.get("status") == "accepted":
-                logger.info(
-                    f"Tool {tool_name} delegated to async processing "
-                    f"(operation_id={operation_id}, task_id={task_id}). "
-                    f"RabbitMQ completion consumer will handle the rest."
-                )
-                # Don't publish result, don't continue with LLM, and don't cleanup
-                # The RabbitMQ consumer will handle everything when the external
-                # service completes and publishes to the queue
-                delegated_to_async = True
-                return
-        except (orjson.JSONDecodeError, TypeError):
-            pass  # Not JSON or not async - continue normally
-
-        # Publish tool result to stream registry
-        await stream_registry.publish_chunk(task_id, result)
-
-        # Update the pending message with result
-        result_str = (
-            result.output
-            if isinstance(result.output, str)
-            else orjson.dumps(result.output).decode("utf-8")
-        )
-        await _update_pending_operation(
-            session_id=session_id,
-            tool_call_id=tool_call_id,
-            result=result_str,
-        )
-
-        logger.info(
-            f"Background tool {tool_name} completed for session {session_id} "
-            f"(task_id={task_id})"
-        )
-
-        # Generate LLM continuation and stream chunks to registry
-        await _generate_llm_continuation_with_streaming(
-            session_id=session_id,
-            user_id=user_id,
-            task_id=task_id,
-        )
-
-        # Mark task as completed in stream registry
-        await stream_registry.mark_task_completed(task_id, status="completed")
-
-    except Exception as e:
-        logger.error(f"Background tool {tool_name} failed: {e}", exc_info=True)
-        error_response = ErrorResponse(
-            message=f"Tool {tool_name} failed: {str(e)}",
-        )
-
-        # Publish error to stream registry followed by finish event
-        await stream_registry.publish_chunk(
-            task_id,
-            StreamError(errorText=str(e)),
-        )
-        await stream_registry.publish_chunk(task_id, StreamFinish())
-
-        await _update_pending_operation(
-            session_id=session_id,
-            tool_call_id=tool_call_id,
-            result=error_response.model_dump_json(),
-        )
-
-        # Mark task as failed in stream registry
-        await stream_registry.mark_task_completed(task_id, status="failed")
-    finally:
-        # Only cleanup if we didn't delegate to async processing
-        # For async path, the RabbitMQ completion consumer handles cleanup
-        if not delegated_to_async:
-            await _mark_operation_completed(tool_call_id)
-
-
 async def _update_pending_operation(
     session_id: str,
     tool_call_id: str,
@@ -2111,128 +1964,3 @@ async def _generate_llm_continuation(
 
     except Exception as e:
         logger.error(f"Failed to generate LLM continuation: {e}", exc_info=True)
-
-
-async def _generate_llm_continuation_with_streaming(
-    session_id: str,
-    user_id: str | None,
-    task_id: str,
-) -> None:
-    """Generate an LLM response with streaming to the stream registry.
-
-    This is called by background tasks to continue the conversation
-    after a tool result is saved. Chunks are published to the stream registry
-    so reconnecting clients can receive them.
-    """
-    import uuid as uuid_module
-
-    try:
-        # Load fresh session from DB (bypass cache to get the updated tool result)
-        await invalidate_session_cache(session_id)
-        session = await get_chat_session(session_id, user_id)
-        if not session:
-            logger.error(f"Session {session_id} not found for LLM continuation")
-            return
-
-        # Build system prompt
-        system_prompt, _ = await _build_system_prompt(user_id)
-
-        # Build messages in OpenAI format
-        messages = session.to_openai_messages()
-        if system_prompt:
-            from openai.types.chat import ChatCompletionSystemMessageParam
-
-            system_message = ChatCompletionSystemMessageParam(
-                role="system",
-                content=system_prompt,
-            )
-            messages = [system_message] + messages
-
-        # Build extra_body for tracing
-        extra_body: dict[str, Any] = {
-            "posthogProperties": {
-                "environment": settings.config.app_env.value,
-            },
-        }
-        if user_id:
-            extra_body["user"] = user_id[:128]
-            extra_body["posthogDistinctId"] = user_id
-        if session_id:
-            extra_body["session_id"] = session_id[:128]
-
-        # Make streaming LLM call (no tools - just text response)
-        from typing import cast
-
-        from openai.types.chat import ChatCompletionMessageParam
-
-        # Generate unique IDs for AI SDK protocol
-        message_id = str(uuid_module.uuid4())
-        text_block_id = str(uuid_module.uuid4())
-
-        # Publish start event
-        await stream_registry.publish_chunk(task_id, StreamStart(messageId=message_id))
-        await stream_registry.publish_chunk(task_id, StreamTextStart(id=text_block_id))
-
-        # Stream the response
-        stream = await client.chat.completions.create(
-            model=config.model,
-            messages=cast(list[ChatCompletionMessageParam], messages),
-            extra_body=extra_body,
-            stream=True,
-        )
-
-        assistant_content = ""
-        async for chunk in stream:
-            if chunk.choices and chunk.choices[0].delta.content:
-                delta = chunk.choices[0].delta.content
-                assistant_content += delta
-                # Publish delta to stream registry
-                await stream_registry.publish_chunk(
-                    task_id,
-                    StreamTextDelta(id=text_block_id, delta=delta),
-                )
-
-        # Publish end events
-        await stream_registry.publish_chunk(task_id, StreamTextEnd(id=text_block_id))
-
-        if assistant_content:
-            # Reload session from DB to avoid race condition with user messages
-            fresh_session = await get_chat_session(session_id, user_id)
-            if not fresh_session:
-                logger.error(
-                    f"Session {session_id} disappeared during LLM continuation"
-                )
-                return
-
-            # Save assistant message to database
-            assistant_message = ChatMessage(
-                role="assistant",
-                content=assistant_content,
-            )
-            fresh_session.messages.append(assistant_message)
-
-            # Save to database (not cache) to persist the response
-            await upsert_chat_session(fresh_session)
-
-            # Invalidate cache so next poll/refresh gets fresh data
-            await invalidate_session_cache(session_id)
-
-            logger.info(
-                f"Generated streaming LLM continuation for session {session_id} "
-                f"(task_id={task_id}), response length: {len(assistant_content)}"
-            )
-        else:
-            logger.warning(
-                f"Streaming LLM continuation returned empty response for {session_id}"
-            )
-
-    except Exception as e:
-        logger.error(
-            f"Failed to generate streaming LLM continuation: {e}", exc_info=True
-        )
-        # Publish error to stream registry followed by finish event
-        await stream_registry.publish_chunk(
-            task_id,
-            StreamError(errorText=f"Failed to generate response: {e}"),
-        )
-        await stream_registry.publish_chunk(task_id, StreamFinish())

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

@@ -1,648 +0,0 @@
-"""Stream registry for managing reconnectable SSE streams.
-
-This module provides a registry for tracking active streaming tasks and their
-messages. It supports:
-- Creating tasks with unique IDs for long-running operations
-- Publishing stream messages to both Redis Streams and in-memory queues
-- Subscribing to tasks with replay of missed messages
-- Looking up tasks by operation_id for webhook callbacks
-- Cross-pod real-time delivery via Redis pub/sub
-"""
-
-import asyncio
-import logging
-from dataclasses import dataclass, field
-from datetime import datetime, timezone
-from typing import Any, Literal
-
-import orjson
-
-from backend.data.redis_client import get_redis_async
-
-from .config import ChatConfig
-from .response_model import StreamBaseResponse, StreamFinish
-
-logger = logging.getLogger(__name__)
-config = ChatConfig()
-
-# Track active pub/sub listeners for cross-pod delivery
-_pubsub_listeners: dict[str, asyncio.Task] = {}
-
-
-@dataclass
-class ActiveTask:
-    """Represents an active streaming task."""
-
-    task_id: str
-    session_id: str
-    user_id: str | None
-    tool_call_id: str
-    tool_name: str
-    operation_id: str
-    status: Literal["running", "completed", "failed"] = "running"
-    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
-    queue: asyncio.Queue[StreamBaseResponse] = field(default_factory=asyncio.Queue)
-    asyncio_task: asyncio.Task | None = None
-    # Lock for atomic status checks and subscriber management
-    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
-    # Set of subscriber queues for fan-out
-    subscribers: set[asyncio.Queue[StreamBaseResponse]] = field(default_factory=set)
-
-
-# Module-level registry for active tasks
-_active_tasks: dict[str, ActiveTask] = {}
-
-# Redis key patterns
-TASK_META_PREFIX = "chat:task:meta:"  # Hash for task metadata
-TASK_STREAM_PREFIX = "chat:stream:"  # Redis Stream for messages
-TASK_OP_PREFIX = "chat:task:op:"  # Operation ID -> task_id mapping
-TASK_PUBSUB_PREFIX = "chat:task:pubsub:"  # Pub/sub channel for cross-pod delivery
-
-
-def _get_task_meta_key(task_id: str) -> str:
-    """Get Redis key for task metadata."""
-    return f"{TASK_META_PREFIX}{task_id}"
-
-
-def _get_task_stream_key(task_id: str) -> str:
-    """Get Redis key for task message stream."""
-    return f"{TASK_STREAM_PREFIX}{task_id}"
-
-
-def _get_operation_mapping_key(operation_id: str) -> str:
-    """Get Redis key for operation_id to task_id mapping."""
-    return f"{TASK_OP_PREFIX}{operation_id}"
-
-
-def _get_task_pubsub_channel(task_id: str) -> str:
-    """Get Redis pub/sub channel for task cross-pod delivery."""
-    return f"{TASK_PUBSUB_PREFIX}{task_id}"
-
-
-async def create_task(
-    task_id: str,
-    session_id: str,
-    user_id: str | None,
-    tool_call_id: str,
-    tool_name: str,
-    operation_id: str,
-) -> ActiveTask:
-    """Create a new streaming task in memory and Redis.
-
-    Args:
-        task_id: Unique identifier for the task
-        session_id: Chat session ID
-        user_id: User ID (may be None for anonymous)
-        tool_call_id: Tool call ID from the LLM
-        tool_name: Name of the tool being executed
-        operation_id: Operation ID for webhook callbacks
-
-    Returns:
-        The created ActiveTask instance
-    """
-    task = ActiveTask(
-        task_id=task_id,
-        session_id=session_id,
-        user_id=user_id,
-        tool_call_id=tool_call_id,
-        tool_name=tool_name,
-        operation_id=operation_id,
-    )
-
-    # Store in memory registry
-    _active_tasks[task_id] = task
-
-    # Store metadata in Redis for durability
-    redis = await get_redis_async()
-    meta_key = _get_task_meta_key(task_id)
-    op_key = _get_operation_mapping_key(operation_id)
-
-    await redis.hset(  # type: ignore[misc]
-        meta_key,
-        mapping={
-            "task_id": task_id,
-            "session_id": session_id,
-            "user_id": user_id or "",
-            "tool_call_id": tool_call_id,
-            "tool_name": tool_name,
-            "operation_id": operation_id,
-            "status": task.status,
-            "created_at": task.created_at.isoformat(),
-        },
-    )
-    await redis.expire(meta_key, config.stream_ttl)
-
-    # Create operation_id -> task_id mapping for webhook lookups
-    await redis.set(op_key, task_id, ex=config.stream_ttl)
-
-    logger.info(
-        f"Created streaming task {task_id} for operation {operation_id} "
-        f"in session {session_id}"
-    )
-
-    return task
-
-
-async def publish_chunk(
-    task_id: str,
-    chunk: StreamBaseResponse,
-) -> str:
-    """Publish a chunk to the task's stream.
-
-    Delivers to in-memory subscribers first (for real-time), then persists to
-    Redis Stream (for replay). This order ensures live subscribers get messages
-    even if Redis temporarily fails.
-
-    Args:
-        task_id: Task ID to publish to
-        chunk: The stream response chunk to publish
-
-    Returns:
-        The Redis Stream message ID (format: "timestamp-sequence"), or "0-0" if
-        Redis persistence failed
-    """
-    # Deliver to in-memory subscribers FIRST for real-time updates
-    task = _active_tasks.get(task_id)
-    if task:
-        async with task.lock:
-            for subscriber_queue in task.subscribers:
-                try:
-                    subscriber_queue.put_nowait(chunk)
-                except asyncio.QueueFull:
-                    logger.warning(
-                        f"Subscriber queue full for task {task_id}, dropping chunk"
-                    )
-
-    # Then persist to Redis Stream for replay (with error handling)
-    message_id = "0-0"
-    chunk_json = chunk.model_dump_json()
-    try:
-        redis = await get_redis_async()
-        stream_key = _get_task_stream_key(task_id)
-
-        # Add to Redis Stream with auto-generated ID
-        # The ID format is "timestamp-sequence" which gives us ordering
-        raw_id = await redis.xadd(
-            stream_key,
-            {"data": chunk_json},
-            maxlen=config.stream_max_length,
-        )
-        message_id = raw_id if isinstance(raw_id, str) else raw_id.decode()
-
-        # Publish to pub/sub for cross-pod real-time delivery
-        pubsub_channel = _get_task_pubsub_channel(task_id)
-        await redis.publish(pubsub_channel, chunk_json)
-
-        logger.debug(f"Published chunk to task {task_id}, message_id={message_id}")
-    except Exception as e:
-        logger.error(
-            f"Failed to persist chunk to Redis for task {task_id}: {e}",
-            exc_info=True,
-        )
-
-    return message_id
-
-
-async def subscribe_to_task(
-    task_id: str,
-    user_id: str | None,
-    last_message_id: str = "0-0",
-) -> asyncio.Queue[StreamBaseResponse] | None:
-    """Subscribe to a task's stream with replay of missed messages.
-
-    Args:
-        task_id: Task ID to subscribe to
-        user_id: User ID for ownership validation
-        last_message_id: Last Redis Stream message ID received ("0-0" for full replay)
-
-    Returns:
-        An asyncio Queue that will receive stream chunks, or None if task not found
-        or user doesn't have access
-    """
-    # Check in-memory first
-    task = _active_tasks.get(task_id)
-
-    if task:
-        # Validate ownership
-        if user_id and task.user_id and task.user_id != user_id:
-            logger.warning(
-                f"User {user_id} attempted to subscribe to task {task_id} "
-                f"owned by {task.user_id}"
-            )
-            return None
-
-        # Create a new queue for this subscriber
-        subscriber_queue: asyncio.Queue[StreamBaseResponse] = asyncio.Queue()
-
-        # Replay from Redis Stream
-        redis = await get_redis_async()
-        stream_key = _get_task_stream_key(task_id)
-
-        # Track the last message ID we've seen for gap detection
-        replay_last_id = last_message_id
-
-        # Read all messages from stream starting after last_message_id
-        # xread returns messages with ID > last_message_id
-        messages = await redis.xread({stream_key: last_message_id}, block=0, count=1000)
-
-        if messages:
-            # messages format: [[stream_name, [(id, {data: json}), ...]]]
-            for _stream_name, stream_messages in messages:
-                for msg_id, msg_data in stream_messages:
-                    # Track the last message ID we've processed
-                    replay_last_id = (
-                        msg_id if isinstance(msg_id, str) else msg_id.decode()
-                    )
-                    if b"data" in msg_data:
-                        try:
-                            chunk_data = orjson.loads(msg_data[b"data"])
-                            # Reconstruct the appropriate response type
-                            chunk = _reconstruct_chunk(chunk_data)
-                            if chunk:
-                                await subscriber_queue.put(chunk)
-                        except Exception as e:
-                            logger.warning(f"Failed to replay message: {e}")
-
-        # Atomically check status and register subscriber under lock
-        # This prevents race condition where task completes between check and subscribe
-        should_start_pubsub = False
-        async with task.lock:
-            if task.status == "running":
-                # Register this subscriber for live updates
-                task.subscribers.add(subscriber_queue)
-                # Start pub/sub listener if this is the first subscriber
-                should_start_pubsub = len(task.subscribers) == 1
-                logger.debug(
-                    f"Registered subscriber for task {task_id}, "
-                    f"total subscribers: {len(task.subscribers)}"
-                )
-            else:
-                # Task is done, add finish marker
-                await subscriber_queue.put(StreamFinish())
-
-        # After registering, do a second read to catch any messages published
-        # between the first read and registration (closes the race window)
-        if task.status == "running":
-            gap_messages = await redis.xread(
-                {stream_key: replay_last_id}, block=0, count=1000
-            )
-            if gap_messages:
-                for _stream_name, stream_messages in gap_messages:
-                    for _msg_id, msg_data in stream_messages:
-                        if b"data" in msg_data:
-                            try:
-                                chunk_data = orjson.loads(msg_data[b"data"])
-                                chunk = _reconstruct_chunk(chunk_data)
-                                if chunk:
-                                    await subscriber_queue.put(chunk)
-                            except Exception as e:
-                                logger.warning(f"Failed to replay gap message: {e}")
-
-        # Start pub/sub listener outside the lock to avoid deadlocks
-        if should_start_pubsub:
-            await start_pubsub_listener(task_id)
-
-        return subscriber_queue
-
-    # Try to load from Redis if not in memory
-    redis = await get_redis_async()
-    meta_key = _get_task_meta_key(task_id)
-    meta: dict[Any, Any] = await redis.hgetall(meta_key)  # type: ignore[misc]
-
-    if not meta:
-        logger.warning(f"Task {task_id} not found in memory or Redis")
-        return None
-
-    # Validate ownership
-    task_user_id = meta.get(b"user_id", b"").decode() or None
-    if user_id and task_user_id and task_user_id != user_id:
-        logger.warning(
-            f"User {user_id} attempted to subscribe to task {task_id} "
-            f"owned by {task_user_id}"
-        )
-        return None
-
-    # Replay from Redis Stream only (task is not in memory, so it's completed/crashed)
-    subscriber_queue = asyncio.Queue()
-    stream_key = _get_task_stream_key(task_id)
-
-    # Read all messages starting after last_message_id
-    messages = await redis.xread({stream_key: last_message_id}, block=0, count=1000)
-
-    if messages:
-        for _stream_name, stream_messages in messages:
-            for _msg_id, msg_data in stream_messages:
-                if b"data" in msg_data:
-                    try:
-                        chunk_data = orjson.loads(msg_data[b"data"])
-                        chunk = _reconstruct_chunk(chunk_data)
-                        if chunk:
-                            await subscriber_queue.put(chunk)
-                    except Exception as e:
-                        logger.warning(f"Failed to replay message: {e}")
-
-    # Add finish marker since task is not active
-    await subscriber_queue.put(StreamFinish())
-
-    return subscriber_queue
-
-
-async def mark_task_completed(
-    task_id: str,
-    status: Literal["completed", "failed"] = "completed",
-) -> None:
-    """Mark a task as completed and publish final event.
-
-    Args:
-        task_id: Task ID to mark as completed
-        status: Final status ("completed" or "failed")
-    """
-    task = _active_tasks.get(task_id)
-
-    if task:
-        # Acquire lock to prevent new subscribers during completion
-        async with task.lock:
-            task.status = status
-            # Send finish event directly to all current subscribers
-            finish_event = StreamFinish()
-            for subscriber_queue in task.subscribers:
-                try:
-                    subscriber_queue.put_nowait(finish_event)
-                except asyncio.QueueFull:
-                    logger.warning(
-                        f"Subscriber queue full for task {task_id} during completion"
-                    )
-            # Clear subscribers since task is done
-            task.subscribers.clear()
-
-        # Stop pub/sub listener since task is done
-        await stop_pubsub_listener(task_id)
-
-        # Also publish to Redis Stream for replay (and pub/sub for cross-pod)
-        await publish_chunk(task_id, StreamFinish())
-
-        # Remove from active tasks after a short delay to allow subscribers to finish
-        async def _cleanup():
-            await asyncio.sleep(5)
-            _active_tasks.pop(task_id, None)
-            logger.info(f"Cleaned up task {task_id} from memory")
-
-        asyncio.create_task(_cleanup())
-
-    # Update Redis metadata
-    redis = await get_redis_async()
-    meta_key = _get_task_meta_key(task_id)
-    await redis.hset(meta_key, "status", status)  # type: ignore[misc]
-
-    logger.info(f"Marked task {task_id} as {status}")
-
-
-async def find_task_by_operation_id(operation_id: str) -> ActiveTask | None:
-    """Find a task by its operation ID.
-
-    Used by webhook callbacks to locate the task to update.
-
-    Args:
-        operation_id: Operation ID to search for
-
-    Returns:
-        ActiveTask if found, None otherwise
-    """
-    # Check in-memory first
-    for task in _active_tasks.values():
-        if task.operation_id == operation_id:
-            return task
-
-    # Try Redis lookup
-    redis = await get_redis_async()
-    op_key = _get_operation_mapping_key(operation_id)
-    task_id = await redis.get(op_key)
-
-    if task_id:
-        task_id_str = task_id.decode() if isinstance(task_id, bytes) else task_id
-        # Check if task is in memory
-        if task_id_str in _active_tasks:
-            return _active_tasks[task_id_str]
-
-        # Load metadata from Redis
-        meta_key = _get_task_meta_key(task_id_str)
-        meta: dict[Any, Any] = await redis.hgetall(meta_key)  # type: ignore[misc]
-
-        if meta:
-            # Reconstruct task object (not fully active, but has metadata)
-            return ActiveTask(
-                task_id=meta.get(b"task_id", b"").decode(),
-                session_id=meta.get(b"session_id", b"").decode(),
-                user_id=meta.get(b"user_id", b"").decode() or None,
-                tool_call_id=meta.get(b"tool_call_id", b"").decode(),
-                tool_name=meta.get(b"tool_name", b"").decode(),
-                operation_id=operation_id,
-                status=meta.get(b"status", b"running").decode(),  # type: ignore
-            )
-
-    return None
-
-
-async def get_task(task_id: str) -> ActiveTask | None:
-    """Get a task by its ID.
-
-    Args:
-        task_id: Task ID to look up
-
-    Returns:
-        ActiveTask if found, None otherwise
-    """
-    # Check in-memory first
-    if task_id in _active_tasks:
-        return _active_tasks[task_id]
-
-    # Try Redis lookup
-    redis = await get_redis_async()
-    meta_key = _get_task_meta_key(task_id)
-    meta: dict[Any, Any] = await redis.hgetall(meta_key)  # type: ignore[misc]
-
-    if meta:
-        return ActiveTask(
-            task_id=meta.get(b"task_id", b"").decode(),
-            session_id=meta.get(b"session_id", b"").decode(),
-            user_id=meta.get(b"user_id", b"").decode() or None,
-            tool_call_id=meta.get(b"tool_call_id", b"").decode(),
-            tool_name=meta.get(b"tool_name", b"").decode(),
-            operation_id=meta.get(b"operation_id", b"").decode(),
-            status=meta.get(b"status", b"running").decode(),  # type: ignore[arg-type]
-        )
-
-    return None
-
-
-def _reconstruct_chunk(chunk_data: dict) -> StreamBaseResponse | None:
-    """Reconstruct a StreamBaseResponse from JSON data.
-
-    Args:
-        chunk_data: Parsed JSON data from Redis
-
-    Returns:
-        Reconstructed response object, or None if unknown type
-    """
-    from .response_model import (
-        ResponseType,
-        StreamError,
-        StreamFinish,
-        StreamHeartbeat,
-        StreamStart,
-        StreamTextDelta,
-        StreamTextEnd,
-        StreamTextStart,
-        StreamToolInputAvailable,
-        StreamToolInputStart,
-        StreamToolOutputAvailable,
-        StreamUsage,
-    )
-
-    chunk_type = chunk_data.get("type")
-
-    try:
-        if chunk_type == ResponseType.START.value:
-            return StreamStart(**chunk_data)
-        elif chunk_type == ResponseType.FINISH.value:
-            return StreamFinish(**chunk_data)
-        elif chunk_type == ResponseType.TEXT_START.value:
-            return StreamTextStart(**chunk_data)
-        elif chunk_type == ResponseType.TEXT_DELTA.value:
-            return StreamTextDelta(**chunk_data)
-        elif chunk_type == ResponseType.TEXT_END.value:
-            return StreamTextEnd(**chunk_data)
-        elif chunk_type == ResponseType.TOOL_INPUT_START.value:
-            return StreamToolInputStart(**chunk_data)
-        elif chunk_type == ResponseType.TOOL_INPUT_AVAILABLE.value:
-            return StreamToolInputAvailable(**chunk_data)
-        elif chunk_type == ResponseType.TOOL_OUTPUT_AVAILABLE.value:
-            return StreamToolOutputAvailable(**chunk_data)
-        elif chunk_type == ResponseType.ERROR.value:
-            return StreamError(**chunk_data)
-        elif chunk_type == ResponseType.USAGE.value:
-            return StreamUsage(**chunk_data)
-        elif chunk_type == ResponseType.HEARTBEAT.value:
-            return StreamHeartbeat(**chunk_data)
-        else:
-            logger.warning(f"Unknown chunk type: {chunk_type}")
-            return None
-    except Exception as e:
-        logger.warning(f"Failed to reconstruct chunk of type {chunk_type}: {e}")
-        return None
-
-
-async def set_task_asyncio_task(task_id: str, asyncio_task: asyncio.Task) -> None:
-    """Associate an asyncio.Task with an ActiveTask.
-
-    Args:
-        task_id: Task ID
-        asyncio_task: The asyncio Task to associate
-    """
-    task = _active_tasks.get(task_id)
-    if task:
-        task.asyncio_task = asyncio_task
-
-
-async def unsubscribe_from_task(
-    task_id: str,
-    subscriber_queue: asyncio.Queue[StreamBaseResponse],
-) -> None:
-    """Unsubscribe a queue from a task's stream.
-
-    Should be called when a client disconnects to clean up resources.
-    Also stops the pub/sub listener if there are no more local subscribers.
-
-    Args:
-        task_id: Task ID to unsubscribe from
-        subscriber_queue: The queue to remove from subscribers
-    """
-    task = _active_tasks.get(task_id)
-    if task:
-        async with task.lock:
-            task.subscribers.discard(subscriber_queue)
-            remaining = len(task.subscribers)
-            logger.debug(
-                f"Unsubscribed from task {task_id}, "
-                f"remaining subscribers: {remaining}"
-            )
-            # Stop pub/sub listener if no more local subscribers
-            if remaining == 0:
-                await stop_pubsub_listener(task_id)
-
-
-async def start_pubsub_listener(task_id: str) -> None:
-    """Start listening to Redis pub/sub for cross-pod delivery.
-
-    This enables real-time updates when another pod publishes chunks for a task
-    that has local subscribers on this pod.
-
-    Args:
-        task_id: Task ID to listen for
-    """
-    if task_id in _pubsub_listeners:
-        return  # Already listening
-
-    task = _active_tasks.get(task_id)
-    if not task:
-        return
-
-    async def _listener():
-        try:
-            redis = await get_redis_async()
-            pubsub = redis.pubsub()
-            channel = _get_task_pubsub_channel(task_id)
-            await pubsub.subscribe(channel)
-            logger.debug(f"Started pub/sub listener for task {task_id}")
-
-            async for message in pubsub.listen():
-                if message["type"] != "message":
-                    continue
-
-                try:
-                    chunk_data = orjson.loads(message["data"])
-                    chunk = _reconstruct_chunk(chunk_data)
-                    if chunk:
-                        # Deliver to local subscribers
-                        local_task = _active_tasks.get(task_id)
-                        if local_task:
-                            async with local_task.lock:
-                                for queue in local_task.subscribers:
-                                    try:
-                                        queue.put_nowait(chunk)
-                                    except asyncio.QueueFull:
-                                        pass
-                        # Stop listening if this was a finish event
-                        if isinstance(chunk, StreamFinish):
-                            break
-                except Exception as e:
-                    logger.warning(f"Error processing pub/sub message: {e}")
-
-            await pubsub.unsubscribe(channel)
-            await pubsub.close()
-        except asyncio.CancelledError:
-            pass
-        except Exception as e:
-            logger.error(f"Pub/sub listener error for task {task_id}: {e}")
-        finally:
-            _pubsub_listeners.pop(task_id, None)
-            logger.debug(f"Stopped pub/sub listener for task {task_id}")
-
-    listener_task = asyncio.create_task(_listener())
-    _pubsub_listeners[task_id] = listener_task
-
-
-async def stop_pubsub_listener(task_id: str) -> None:
-    """Stop the pub/sub listener for a task.
-
-    Args:
-        task_id: Task ID to stop listening for
-    """
-    listener = _pubsub_listeners.pop(task_id, None)
-    if listener and not listener.done():
-        listener.cancel()
-        try:
-            await listener
-        except asyncio.CancelledError:
-            pass
-        logger.debug(f"Cancelled pub/sub listener for task {task_id}")

autogpt_platform/backend/backend/api/features/chat/tools/agent_generator/__init__.py

@@ -9,6 +9,7 @@ from .core import (
     json_to_graph,
     save_agent_to_library,
 )
+from .errors import get_user_message_for_error
 from .service import health_check as check_external_service_health
 from .service import is_external_service_configured
 
@@ -25,4 +26,6 @@ __all__ = [
     # Service
     "is_external_service_configured",
     "check_external_service_health",
+    # Error handling
+    "get_user_message_for_error",
 ]

autogpt_platform/backend/backend/api/features/chat/tools/agent_generator/core.py

@@ -57,34 +57,26 @@ async def decompose_goal(description: str, context: str = "") -> dict[str, Any]
     return await decompose_goal_external(description, context)
 
 
-async def generate_agent(
-    instructions: dict[str, Any],
-    operation_id: str | None = None,
-    task_id: str | None = None,
-) -> dict[str, Any] | None:
+async def generate_agent(instructions: dict[str, Any]) -> dict[str, Any] | None:
     """Generate agent JSON from instructions.
 
     Args:
         instructions: Structured instructions from decompose_goal
-        operation_id: Operation ID for async processing (enables RabbitMQ callback)
-        task_id: Task ID for async processing (enables RabbitMQ callback)
 
     Returns:
-        Agent JSON dict, {"status": "accepted"} for async, or None on error
+        Agent JSON dict, error dict {"type": "error", ...}, or None on error
 
     Raises:
         AgentGeneratorNotConfiguredError: If the external service is not configured.
     """
     _check_service_configured()
     logger.info("Calling external Agent Generator service for generate_agent")
-    result = await generate_agent_external(instructions, operation_id, task_id)
-
-    # Don't modify async response
-    if result and result.get("status") == "accepted":
-        return result
-
+    result = await generate_agent_external(instructions)
     if result:
-        # Ensure required fields
+        # Check if it's an error response - pass through as-is
+        if isinstance(result, dict) and result.get("type") == "error":
+            return result
+        # Ensure required fields for successful agent generation
         if "id" not in result:
             result["id"] = str(uuid.uuid4())
         if "version" not in result:
@@ -264,10 +256,7 @@ async def get_agent_as_json(
 
 
 async def generate_agent_patch(
-    update_request: str,
-    current_agent: dict[str, Any],
-    operation_id: str | None = None,
-    task_id: str | None = None,
+    update_request: str, current_agent: dict[str, Any]
 ) -> dict[str, Any] | None:
     """Update an existing agent using natural language.
 
@@ -279,17 +268,14 @@ async def generate_agent_patch(
     Args:
         update_request: Natural language description of changes
         current_agent: Current agent JSON
-        operation_id: Operation ID for async processing (enables RabbitMQ callback)
-        task_id: Task ID for async processing (enables RabbitMQ callback)
 
     Returns:
-        Updated agent JSON, clarifying questions dict, {"status": "accepted"} for async, or None on error
+        Updated agent JSON, clarifying questions dict {"type": "clarifying_questions", ...},
+        error dict {"type": "error", ...}, or None on unexpected error
 
     Raises:
         AgentGeneratorNotConfiguredError: If the external service is not configured.
     """
     _check_service_configured()
     logger.info("Calling external Agent Generator service for generate_agent_patch")
-    return await generate_agent_patch_external(
-        update_request, current_agent, operation_id, task_id
-    )
+    return await generate_agent_patch_external(update_request, current_agent)

autogpt_platform/backend/backend/api/features/chat/tools/agent_generator/errors.py

@@ -0,0 +1,43 @@
+"""Error handling utilities for agent generator."""
+
+
+def get_user_message_for_error(
+    error_type: str,
+    operation: str = "process the request",
+    llm_parse_message: str | None = None,
+    validation_message: str | None = None,
+) -> str:
+    """Get a user-friendly error message based on error type.
+
+    This function maps internal error types to user-friendly messages,
+    providing a consistent experience across different agent operations.
+
+    Args:
+        error_type: The error type from the external service
+            (e.g., "llm_parse_error", "timeout", "rate_limit")
+        operation: Description of what operation failed, used in the default
+            message (e.g., "analyze the goal", "generate the agent")
+        llm_parse_message: Custom message for llm_parse_error type
+        validation_message: Custom message for validation_error type
+
+    Returns:
+        User-friendly error message suitable for display to the user
+    """
+    if error_type == "llm_parse_error":
+        return (
+            llm_parse_message
+            or "The AI had trouble processing this request. Please try again."
+        )
+    elif error_type == "validation_error":
+        return (
+            validation_message
+            or "The request failed validation. Please try rephrasing."
+        )
+    elif error_type == "patch_error":
+        return "Failed to apply the changes. Please try a different approach."
+    elif error_type in ("timeout", "llm_timeout"):
+        return "The request took too long. Please try again."
+    elif error_type in ("rate_limit", "llm_rate_limit"):
+        return "The service is currently busy. Please try again in a moment."
+    else:
+        return f"Failed to {operation}. Please try again."

autogpt_platform/backend/backend/api/features/chat/tools/agent_generator/service.py

@@ -14,6 +14,70 @@ from backend.util.settings import Settings
 
 logger = logging.getLogger(__name__)
 
+
+def _create_error_response(
+    error_message: str,
+    error_type: str = "unknown",
+    details: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Create a standardized error response dict.
+
+    Args:
+        error_message: Human-readable error message
+        error_type: Machine-readable error type
+        details: Optional additional error details
+
+    Returns:
+        Error dict with type="error" and error details
+    """
+    response: dict[str, Any] = {
+        "type": "error",
+        "error": error_message,
+        "error_type": error_type,
+    }
+    if details:
+        response["details"] = details
+    return response
+
+
+def _classify_http_error(e: httpx.HTTPStatusError) -> tuple[str, str]:
+    """Classify an HTTP error into error_type and message.
+
+    Args:
+        e: The HTTP status error
+
+    Returns:
+        Tuple of (error_type, error_message)
+    """
+    status = e.response.status_code
+    if status == 429:
+        return "rate_limit", f"Agent Generator rate limited: {e}"
+    elif status == 503:
+        return "service_unavailable", f"Agent Generator unavailable: {e}"
+    elif status == 504 or status == 408:
+        return "timeout", f"Agent Generator timed out: {e}"
+    else:
+        return "http_error", f"HTTP error calling Agent Generator: {e}"
+
+
+def _classify_request_error(e: httpx.RequestError) -> tuple[str, str]:
+    """Classify a request error into error_type and message.
+
+    Args:
+        e: The request error
+
+    Returns:
+        Tuple of (error_type, error_message)
+    """
+    error_str = str(e).lower()
+    if "timeout" in error_str or "timed out" in error_str:
+        return "timeout", f"Agent Generator request timed out: {e}"
+    elif "connect" in error_str:
+        return "connection_error", f"Could not connect to Agent Generator: {e}"
+    else:
+        return "request_error", f"Request error calling Agent Generator: {e}"
+
+
 _client: httpx.AsyncClient | None = None
 _settings: Settings | None = None
 
@@ -67,7 +131,8 @@ async def decompose_goal_external(
         - {"type": "instructions", "steps": [...]}
         - {"type": "unachievable_goal", ...}
         - {"type": "vague_goal", ...}
-        Or None on error
+        - {"type": "error", "error": "...", "error_type": "..."} on error
+        Or None on unexpected error
     """
     client = _get_client()
 
@@ -83,8 +148,13 @@ async def decompose_goal_external(
         data = response.json()
 
         if not data.get("success"):
-            logger.error(f"External service returned error: {data.get('error')}")
-            return None
+            error_msg = data.get("error", "Unknown error from Agent Generator")
+            error_type = data.get("error_type", "unknown")
+            logger.error(
+                f"Agent Generator decomposition failed: {error_msg} "
+                f"(type: {error_type})"
+            )
+            return _create_error_response(error_msg, error_type)
 
         # Map the response to the expected format
         response_type = data.get("type")
@@ -106,122 +176,113 @@ async def decompose_goal_external(
                 "type": "vague_goal",
                 "suggested_goal": data.get("suggested_goal"),
             }
+        elif response_type == "error":
+            # Pass through error from the service
+            return _create_error_response(
+                data.get("error", "Unknown error"),
+                data.get("error_type", "unknown"),
+            )
         else:
             logger.error(
                 f"Unknown response type from external service: {response_type}"
             )
-            return None
+            return _create_error_response(
+                f"Unknown response type from Agent Generator: {response_type}",
+                "invalid_response",
+            )
 
     except httpx.HTTPStatusError as e:
-        logger.error(f"HTTP error calling external agent generator: {e}")
-        return None
+        error_type, error_msg = _classify_http_error(e)
+        logger.error(error_msg)
+        return _create_error_response(error_msg, error_type)
     except httpx.RequestError as e:
-        logger.error(f"Request error calling external agent generator: {e}")
-        return None
+        error_type, error_msg = _classify_request_error(e)
+        logger.error(error_msg)
+        return _create_error_response(error_msg, error_type)
     except Exception as e:
-        logger.error(f"Unexpected error calling external agent generator: {e}")
-        return None
+        error_msg = f"Unexpected error calling Agent Generator: {e}"
+        logger.error(error_msg)
+        return _create_error_response(error_msg, "unexpected_error")
 
 
 async def generate_agent_external(
     instructions: dict[str, Any],
-    operation_id: str | None = None,
-    task_id: str | None = None,
 ) -> dict[str, Any] | None:
     """Call the external service to generate an agent from instructions.
 
     Args:
         instructions: Structured instructions from decompose_goal
-        operation_id: Operation ID for async processing (enables RabbitMQ callback)
-        task_id: Task ID for async processing (enables RabbitMQ callback)
 
     Returns:
-        Agent JSON dict, or {"status": "accepted"} for async, or None on error
+        Agent JSON dict on success, or error dict {"type": "error", ...} on error
     """
     client = _get_client()
 
-    # Build request payload
-    payload: dict[str, Any] = {"instructions": instructions}
-    if operation_id and task_id:
-        payload["operation_id"] = operation_id
-        payload["task_id"] = task_id
-
     try:
-        response = await client.post("/api/generate-agent", json=payload)
-
-        # Handle 202 Accepted for async processing
-        if response.status_code == 202:
-            logger.info(
-                f"Agent Generator accepted async request "
-                f"(operation_id={operation_id}, task_id={task_id})"
-            )
-            return {"status": "accepted", "operation_id": operation_id, "task_id": task_id}
-
+        response = await client.post(
+            "/api/generate-agent", json={"instructions": instructions}
+        )
         response.raise_for_status()
         data = response.json()
 
         if not data.get("success"):
-            logger.error(f"External service returned error: {data.get('error')}")
-            return None
+            error_msg = data.get("error", "Unknown error from Agent Generator")
+            error_type = data.get("error_type", "unknown")
+            logger.error(
+                f"Agent Generator generation failed: {error_msg} "
+                f"(type: {error_type})"
+            )
+            return _create_error_response(error_msg, error_type)
 
         return data.get("agent_json")
 
     except httpx.HTTPStatusError as e:
-        logger.error(f"HTTP error calling external agent generator: {e}")
-        return None
+        error_type, error_msg = _classify_http_error(e)
+        logger.error(error_msg)
+        return _create_error_response(error_msg, error_type)
     except httpx.RequestError as e:
-        logger.error(f"Request error calling external agent generator: {e}")
-        return None
+        error_type, error_msg = _classify_request_error(e)
+        logger.error(error_msg)
+        return _create_error_response(error_msg, error_type)
     except Exception as e:
-        logger.error(f"Unexpected error calling external agent generator: {e}")
-        return None
+        error_msg = f"Unexpected error calling Agent Generator: {e}"
+        logger.error(error_msg)
+        return _create_error_response(error_msg, "unexpected_error")
 
 
 async def generate_agent_patch_external(
-    update_request: str,
-    current_agent: dict[str, Any],
-    operation_id: str | None = None,
-    task_id: str | None = None,
+    update_request: str, current_agent: dict[str, Any]
 ) -> dict[str, Any] | None:
     """Call the external service to generate a patch for an existing agent.
 
     Args:
         update_request: Natural language description of changes
         current_agent: Current agent JSON
-        operation_id: Operation ID for async processing (enables RabbitMQ callback)
-        task_id: Task ID for async processing (enables RabbitMQ callback)
 
     Returns:
-        Updated agent JSON, clarifying questions dict, {"status": "accepted"} for async, or None on error
+        Updated agent JSON, clarifying questions dict, or error dict on error
     """
     client = _get_client()
 
-    # Build request payload
-    payload: dict[str, Any] = {
-        "update_request": update_request,
-        "current_agent_json": current_agent,
-    }
-    if operation_id and task_id:
-        payload["operation_id"] = operation_id
-        payload["task_id"] = task_id
-
     try:
-        response = await client.post("/api/update-agent", json=payload)
-
-        # Handle 202 Accepted for async processing
-        if response.status_code == 202:
-            logger.info(
-                f"Agent Generator accepted async update request "
-                f"(operation_id={operation_id}, task_id={task_id})"
-            )
-            return {"status": "accepted", "operation_id": operation_id, "task_id": task_id}
-
+        response = await client.post(
+            "/api/update-agent",
+            json={
+                "update_request": update_request,
+                "current_agent_json": current_agent,
+            },
+        )
         response.raise_for_status()
         data = response.json()
 
         if not data.get("success"):
-            logger.error(f"External service returned error: {data.get('error')}")
-            return None
+            error_msg = data.get("error", "Unknown error from Agent Generator")
+            error_type = data.get("error_type", "unknown")
+            logger.error(
+                f"Agent Generator patch generation failed: {error_msg} "
+                f"(type: {error_type})"
+            )
+            return _create_error_response(error_msg, error_type)
 
         # Check if it's clarifying questions
         if data.get("type") == "clarifying_questions":
@@ -230,18 +291,28 @@ async def generate_agent_patch_external(
                 "questions": data.get("questions", []),
             }
 
+        # Check if it's an error passed through
+        if data.get("type") == "error":
+            return _create_error_response(
+                data.get("error", "Unknown error"),
+                data.get("error_type", "unknown"),
+            )
+
         # Otherwise return the updated agent JSON
         return data.get("agent_json")
 
     except httpx.HTTPStatusError as e:
-        logger.error(f"HTTP error calling external agent generator: {e}")
-        return None
+        error_type, error_msg = _classify_http_error(e)
+        logger.error(error_msg)
+        return _create_error_response(error_msg, error_type)
     except httpx.RequestError as e:
-        logger.error(f"Request error calling external agent generator: {e}")
-        return None
+        error_type, error_msg = _classify_request_error(e)
+        logger.error(error_msg)
+        return _create_error_response(error_msg, error_type)
     except Exception as e:
-        logger.error(f"Unexpected error calling external agent generator: {e}")
-        return None
+        error_msg = f"Unexpected error calling Agent Generator: {e}"
+        logger.error(error_msg)
+        return _create_error_response(error_msg, "unexpected_error")
 
 
 async def get_blocks_external() -> list[dict[str, Any]] | None:

autogpt_platform/backend/backend/api/features/chat/tools/create_agent.py

@@ -9,13 +9,13 @@ from .agent_generator import (
     AgentGeneratorNotConfiguredError,
     decompose_goal,
     generate_agent,
+    get_user_message_for_error,
     save_agent_to_library,
 )
 from .base import BaseTool
 from .models import (
     AgentPreviewResponse,
     AgentSavedResponse,
-    AsyncProcessingResponse,
     ClarificationNeededResponse,
     ClarifyingQuestion,
     ErrorResponse,
@@ -96,10 +96,6 @@ class CreateAgentTool(BaseTool):
         save = kwargs.get("save", True)
         session_id = session.session_id if session else None
 
-        # Extract async processing params (passed by long-running tool handler)
-        operation_id = kwargs.get("_operation_id")
-        task_id = kwargs.get("_task_id")
-
         if not description:
             return ErrorResponse(
                 message="Please provide a description of what the agent should do.",
@@ -122,11 +118,29 @@ class CreateAgentTool(BaseTool):
 
         if decomposition_result is None:
             return ErrorResponse(
-                message="Failed to analyze the goal. The agent generation service may be unavailable or timed out. Please try again.",
+                message="Failed to analyze the goal. The agent generation service may be unavailable. Please try again.",
                 error="decomposition_failed",
+                details={"description": description[:100]},
+                session_id=session_id,
+            )
+
+        # Check if the result is an error from the external service
+        if decomposition_result.get("type") == "error":
+            error_msg = decomposition_result.get("error", "Unknown error")
+            error_type = decomposition_result.get("error_type", "unknown")
+            user_message = get_user_message_for_error(
+                error_type,
+                operation="analyze the goal",
+                llm_parse_message="The AI had trouble understanding this request. Please try rephrasing your goal.",
+            )
+            return ErrorResponse(
+                message=user_message,
+                error=f"decomposition_failed:{error_type}",
                 details={
-                    "description": description[:100]
-                },  # Include context for debugging
+                    "description": description[:100],
+                    "service_error": error_msg,
+                    "error_type": error_type,
+                },
                 session_id=session_id,
             )
 
@@ -178,11 +192,7 @@ class CreateAgentTool(BaseTool):
 
         # Step 2: Generate agent JSON (external service handles fixing and validation)
         try:
-            agent_json = await generate_agent(
-                decomposition_result,
-                operation_id=operation_id,
-                task_id=task_id,
-            )
+            agent_json = await generate_agent(decomposition_result)
         except AgentGeneratorNotConfiguredError:
             return ErrorResponse(
                 message=(
@@ -195,24 +205,30 @@ class CreateAgentTool(BaseTool):
 
         if agent_json is None:
             return ErrorResponse(
-                message="Failed to generate the agent. The agent generation service may be unavailable or timed out. Please try again.",
+                message="Failed to generate the agent. The agent generation service may be unavailable. Please try again.",
                 error="generation_failed",
-                details={
-                    "description": description[:100]
-                },  # Include context for debugging
+                details={"description": description[:100]},
                 session_id=session_id,
             )
 
-        # Check if Agent Generator accepted for async processing
-        if agent_json.get("status") == "accepted":
-            logger.info(
-                f"Agent generation delegated to async processing "
-                f"(operation_id={operation_id}, task_id={task_id})"
+        # Check if the result is an error from the external service
+        if isinstance(agent_json, dict) and agent_json.get("type") == "error":
+            error_msg = agent_json.get("error", "Unknown error")
+            error_type = agent_json.get("error_type", "unknown")
+            user_message = get_user_message_for_error(
+                error_type,
+                operation="generate the agent",
+                llm_parse_message="The AI had trouble generating the agent. Please try again or simplify your goal.",
+                validation_message="The generated agent failed validation. Please try rephrasing your goal.",
             )
-            return AsyncProcessingResponse(
-                message="Agent generation started. You'll be notified when it's complete.",
-                operation_id=operation_id,
-                task_id=task_id,
+            return ErrorResponse(
+                message=user_message,
+                error=f"generation_failed:{error_type}",
+                details={
+                    "description": description[:100],
+                    "service_error": error_msg,
+                    "error_type": error_type,
+                },
                 session_id=session_id,
             )
 

autogpt_platform/backend/backend/api/features/chat/tools/edit_agent.py

@@ -9,13 +9,13 @@ from .agent_generator import (
     AgentGeneratorNotConfiguredError,
     generate_agent_patch,
     get_agent_as_json,
+    get_user_message_for_error,
     save_agent_to_library,
 )
 from .base import BaseTool
 from .models import (
     AgentPreviewResponse,
     AgentSavedResponse,
-    AsyncProcessingResponse,
     ClarificationNeededResponse,
     ClarifyingQuestion,
     ErrorResponse,
@@ -103,10 +103,6 @@ class EditAgentTool(BaseTool):
         save = kwargs.get("save", True)
         session_id = session.session_id if session else None
 
-        # Extract async processing params (passed by long-running tool handler)
-        operation_id = kwargs.get("_operation_id")
-        task_id = kwargs.get("_task_id")
-
         if not agent_id:
             return ErrorResponse(
                 message="Please provide the agent ID to edit.",
@@ -138,12 +134,7 @@ class EditAgentTool(BaseTool):
 
         # Step 2: Generate updated agent (external service handles fixing and validation)
         try:
-            result = await generate_agent_patch(
-                update_request,
-                current_agent,
-                operation_id=operation_id,
-                task_id=task_id,
-            )
+            result = await generate_agent_patch(update_request, current_agent)
         except AgentGeneratorNotConfiguredError:
             return ErrorResponse(
                 message=(
@@ -162,16 +153,25 @@ class EditAgentTool(BaseTool):
                 session_id=session_id,
             )
 
-        # Check if Agent Generator accepted for async processing
-        if result.get("status") == "accepted":
-            logger.info(
-                f"Agent edit delegated to async processing "
-                f"(operation_id={operation_id}, task_id={task_id})"
+        # Check if the result is an error from the external service
+        if isinstance(result, dict) and result.get("type") == "error":
+            error_msg = result.get("error", "Unknown error")
+            error_type = result.get("error_type", "unknown")
+            user_message = get_user_message_for_error(
+                error_type,
+                operation="generate the changes",
+                llm_parse_message="The AI had trouble generating the changes. Please try again or simplify your request.",
+                validation_message="The generated changes failed validation. Please try rephrasing your request.",
             )
-            return AsyncProcessingResponse(
-                message="Agent edit started. You'll be notified when it's complete.",
-                operation_id=operation_id,
-                task_id=task_id,
+            return ErrorResponse(
+                message=user_message,
+                error=f"update_generation_failed:{error_type}",
+                details={
+                    "agent_id": agent_id,
+                    "changes": changes[:100],
+                    "service_error": error_msg,
+                    "error_type": error_type,
+                },
                 session_id=session_id,
             )
 

autogpt_platform/backend/backend/api/features/chat/tools/models.py

@@ -352,15 +352,11 @@ class OperationStartedResponse(ToolResponseBase):
 
     This is returned immediately to the client while the operation continues
     to execute. The user can close the tab and check back later.
-
-    The task_id can be used to reconnect to the SSE stream via
-    GET /chat/tasks/{task_id}/stream?last_idx=0
     """
 
     type: ResponseType = ResponseType.OPERATION_STARTED
     operation_id: str
     tool_name: str
-    task_id: str | None = None  # For SSE reconnection
 
 
 class OperationPendingResponse(ToolResponseBase):
@@ -384,20 +380,3 @@ class OperationInProgressResponse(ToolResponseBase):
 
     type: ResponseType = ResponseType.OPERATION_IN_PROGRESS
     tool_call_id: str
-
-
-class AsyncProcessingResponse(ToolResponseBase):
-    """Response when an operation has been delegated to async processing.
-
-    This is returned by tools when the external service accepts the request
-    for async processing (HTTP 202 Accepted). The RabbitMQ completion consumer
-    will handle the result when the external service completes.
-
-    The status field is specifically "accepted" to allow the long-running tool
-    handler to detect this response and skip LLM continuation.
-    """
-
-    type: ResponseType = ResponseType.OPERATION_STARTED
-    status: str = "accepted"  # Must be "accepted" for detection
-    operation_id: str | None = None
-    task_id: str | None = None

autogpt_platform/backend/backend/api/rest_api.py

@@ -22,10 +22,6 @@ import backend.api.features.admin.store_admin_routes
 import backend.api.features.builder
 import backend.api.features.builder.routes
 import backend.api.features.chat.routes as chat_routes
-from backend.api.features.chat.completion_consumer import (
-    start_completion_consumer,
-    stop_completion_consumer,
-)
 import backend.api.features.executions.review.routes
 import backend.api.features.library.db
 import backend.api.features.library.model
@@ -122,21 +118,9 @@ async def lifespan_context(app: fastapi.FastAPI):
     await backend.data.graph.migrate_llm_models(DEFAULT_LLM_MODEL)
     await backend.integrations.webhooks.utils.migrate_legacy_triggered_graphs()
 
-    # Start chat completion consumer for RabbitMQ notifications
-    try:
-        await start_completion_consumer()
-    except Exception as e:
-        logger.warning(f"Could not start chat completion consumer: {e}")
-
     with launch_darkly_context():
         yield
 
-    # Stop chat completion consumer
-    try:
-        await stop_completion_consumer()
-    except Exception as e:
-        logger.warning(f"Error stopping chat completion consumer: {e}")
-
     try:
         await shutdown_cloud_storage_handler()
     except Exception as e:

autogpt_platform/backend/test/agent_generator/test_service.py

@@ -151,15 +151,20 @@ class TestDecomposeGoalExternal:
     @pytest.mark.asyncio
     async def test_decompose_goal_handles_http_error(self):
         """Test decomposition handles HTTP errors gracefully."""
+        mock_response = MagicMock()
+        mock_response.status_code = 500
         mock_client = AsyncMock()
         mock_client.post.side_effect = httpx.HTTPStatusError(
-            "Server error", request=MagicMock(), response=MagicMock()
+            "Server error", request=MagicMock(), response=mock_response
         )
 
         with patch.object(service, "_get_client", return_value=mock_client):
             result = await service.decompose_goal_external("Build a chatbot")
 
-        assert result is None
+        assert result is not None
+        assert result.get("type") == "error"
+        assert result.get("error_type") == "http_error"
+        assert "Server error" in result.get("error", "")
 
     @pytest.mark.asyncio
     async def test_decompose_goal_handles_request_error(self):
@@ -170,7 +175,10 @@ class TestDecomposeGoalExternal:
         with patch.object(service, "_get_client", return_value=mock_client):
             result = await service.decompose_goal_external("Build a chatbot")
 
-        assert result is None
+        assert result is not None
+        assert result.get("type") == "error"
+        assert result.get("error_type") == "connection_error"
+        assert "Connection failed" in result.get("error", "")
 
     @pytest.mark.asyncio
     async def test_decompose_goal_handles_service_error(self):
@@ -179,6 +187,7 @@ class TestDecomposeGoalExternal:
         mock_response.json.return_value = {
             "success": False,
             "error": "Internal error",
+            "error_type": "internal_error",
         }
         mock_response.raise_for_status = MagicMock()
 
@@ -188,7 +197,10 @@ class TestDecomposeGoalExternal:
         with patch.object(service, "_get_client", return_value=mock_client):
             result = await service.decompose_goal_external("Build a chatbot")
 
-        assert result is None
+        assert result is not None
+        assert result.get("type") == "error"
+        assert result.get("error") == "Internal error"
+        assert result.get("error_type") == "internal_error"
 
 
 class TestGenerateAgentExternal:
@@ -236,7 +248,10 @@ class TestGenerateAgentExternal:
         with patch.object(service, "_get_client", return_value=mock_client):
             result = await service.generate_agent_external({"steps": []})
 
-        assert result is None
+        assert result is not None
+        assert result.get("type") == "error"
+        assert result.get("error_type") == "connection_error"
+        assert "Connection failed" in result.get("error", "")
 
 
 class TestGenerateAgentPatchExternal:

autogpt_platform/frontend/CLAUDE.md

@@ -0,0 +1,76 @@
+# CLAUDE.md - Frontend
+
+This file provides guidance to Claude Code when working with the frontend.
+
+## Essential Commands
+
+```bash
+# Install dependencies
+pnpm i
+
+# Generate API client from OpenAPI spec
+pnpm generate:api
+
+# Start development server
+pnpm dev
+
+# Run E2E tests
+pnpm test
+
+# Run Storybook for component development
+pnpm storybook
+
+# Build production
+pnpm build
+
+# Format and lint
+pnpm format
+
+# Type checking
+pnpm types
+```
+
+### Code Style
+
+- Fully capitalize acronyms in symbols, e.g. `graphID`, `useBackendAPI`
+- Use function declarations (not arrow functions) for components/handlers
+
+## Architecture
+
+- **Framework**: Next.js 15 App Router (client-first approach)
+- **Data Fetching**: Type-safe generated API hooks via Orval + React Query
+- **State Management**: React Query for server state, co-located UI state in components/hooks
+- **Component Structure**: Separate render logic (`.tsx`) from business logic (`use*.ts` hooks)
+- **Workflow Builder**: Visual graph editor using @xyflow/react
+- **UI Components**: shadcn/ui (Radix UI primitives) with Tailwind CSS styling
+- **Icons**: Phosphor Icons only
+- **Feature Flags**: LaunchDarkly integration
+- **Error Handling**: ErrorCard for render errors, toast for mutations, Sentry for exceptions
+- **Testing**: Playwright for E2E, Storybook for component development
+
+## Environment Configuration
+
+`.env.default` (defaults) → `.env` (user overrides)
+
+## Feature Development
+
+See @CONTRIBUTING.md for complete patterns. Quick reference:
+
+1. **Pages**: Create in `src/app/(platform)/feature-name/page.tsx`
+   - Extract component logic into custom hooks grouped by concern, not by component. Each hook should represent a cohesive domain of functionality (e.g., useSearch, useFilters, usePagination) rather than bundling all state into one useComponentState hook.
+     - Put each hook in its own `.ts` file
+   - Put sub-components in local `components/` folder
+   - Component props should be `type Props = { ... }` (not exported) unless it needs to be used outside the component
+2. **Components**: Structure as `ComponentName/ComponentName.tsx` + `useComponentName.ts` + `helpers.ts`
+   - Use design system components from `src/components/` (atoms, molecules, organisms)
+   - Never use `src/components/__legacy__/*`
+3. **Data fetching**: Use generated API hooks from `@/app/api/__generated__/endpoints/`
+   - Regenerate with `pnpm generate:api`
+   - Pattern: `use{Method}{Version}{OperationName}`
+4. **Styling**: Tailwind CSS only, use design tokens, Phosphor Icons only
+5. **Testing**: Add Storybook stories for new components, Playwright for E2E
+6. **Code conventions**:
+   - Use function declarations (not arrow functions) for components/handlers
+   - Do not use `useCallback` or `useMemo` unless asked to optimise a given function
+   - Do not type hook returns, let Typescript infer as much as possible
+   - Never type with `any` unless a variable/attribute can ACTUALLY be of any type

autogpt_platform/frontend/src/lib/autogpt-server-api/types.ts

@@ -516,7 +516,7 @@ export type GraphValidationErrorResponse = {
 
 /* *** LIBRARY *** */
 
-/* Mirror of backend/server/v2/library/model.py:LibraryAgent */
+/* Mirror of backend/api/features/library/model.py:LibraryAgent */
 export type LibraryAgent = {
   id: LibraryAgentID;
   graph_id: GraphID;
@@ -616,7 +616,7 @@ export enum LibraryAgentSortEnum {
 
 /* *** CREDENTIALS *** */
 
-/* Mirror of backend/server/integrations/router.py:CredentialsMetaResponse */
+/* Mirror of backend/api/features/integrations/router.py:CredentialsMetaResponse */
 export type CredentialsMetaResponse = {
   id: string;
   provider: CredentialsProviderName;
@@ -628,13 +628,13 @@ export type CredentialsMetaResponse = {
   is_system?: boolean;
 };
 
-/* Mirror of backend/server/integrations/router.py:CredentialsDeletionResponse */
+/* Mirror of backend/api/features/integrations/router.py:CredentialsDeletionResponse */
 export type CredentialsDeleteResponse = {
   deleted: true;
   revoked: boolean | null;
 };
 
-/* Mirror of backend/server/integrations/router.py:CredentialsDeletionNeedsConfirmationResponse */
+/* Mirror of backend/api/features/integrations/router.py:CredentialsDeletionNeedsConfirmationResponse */
 export type CredentialsDeleteNeedConfirmationResponse = {
   deleted: false;
   need_confirmation: true;
@@ -888,7 +888,7 @@ export type Schedule = {
 
 export type ScheduleID = Brand<string, "ScheduleID">;
 
-/* Mirror of backend/server/routers/v1.py:ScheduleCreationRequest */
+/* Mirror of backend/api/features/v1.py:ScheduleCreationRequest */
 export type ScheduleCreatable = {
   graph_id: GraphID;
   graph_version: number;

docs/platform/contributing/oauth-integration-flow.md

@@ -25,7 +25,7 @@ This document focuses on the **API Integration OAuth flow** used for connecting
 ### 2. Backend API Trust Boundary
 - **Location**: Server-side FastAPI application
 - **Components**:
-  - Integration router (`/backend/backend/server/integrations/router.py`)
+  - Integration router (`/backend/backend/api/features/integrations/router.py`)
   - OAuth handlers (`/backend/backend/integrations/oauth/`)
   - Credentials store (`/backend/backend/integrations/credentials_store.py`)
 - **Trust Level**: Trusted - server-controlled environment

docs/platform/ollama.md

@@ -246,7 +246,7 @@ If you encounter any issues, verify that:
   ```bash
   ollama pull llama3.2
   ```
-- If using a custom model, ensure it's added to the model list in `backend/server/model.py`
+- If using a custom model, ensure it's added to the model list in `backend/api/model.py`
 
 #### Docker Issues
 - Ensure Docker daemon is running: