fix(backend/chat): Use distributed locks for chat session mutations

.branchlet.json

@@ -29,7 +29,8 @@
   "postCreateCmd": [
     "cd autogpt_platform/autogpt_libs && poetry install",
     "cd autogpt_platform/backend && poetry install && poetry run prisma generate",
-    "cd autogpt_platform/frontend && pnpm install"
+    "cd autogpt_platform/frontend && pnpm install",
+    "cd docs && pip install -r requirements.txt"
   ],
   "terminalCommand": "code .",
   "deleteBranchWithWorktree": false

.github/copilot-instructions.md

@@ -160,7 +160,7 @@ pnpm storybook                      # Start component development server
 
 **Backend Entry Points:**
 
-- `backend/backend/api/rest_api.py` - FastAPI application setup
+- `backend/backend/server/server.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/api/features/`
+1. Update routes in `/backend/backend/server/routers/`
 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/api/middleware/security.py`):
+**Cache Protection Middleware** (`/backend/backend/server/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,5 +178,4 @@ autogpt_platform/backend/settings.py
 *.ign.*
 .test-contents
 .claude/settings.local.json
-CLAUDE.local.md
 /autogpt_platform/backend/logs

AGENTS.md

@@ -16,6 +16,7 @@ See `docs/content/platform/getting-started.md` for setup instructions.
 - Format Python code with `poetry run format`.
 - Format frontend code using `pnpm format`.
 
+
 ## Frontend guidelines:
 
 See `/frontend/CONTRIBUTING.md` for complete patterns. Quick reference:
@@ -32,17 +33,14 @@ See `/frontend/CONTRIBUTING.md` for complete patterns. Quick reference:
 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 strictly needed
 - Avoid comments at all times unless the code is very complex
-- 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`, if not types available use `unknown`
 
 ## Testing
 
@@ -51,8 +49,22 @@ See `/frontend/CONTRIBUTING.md` for complete patterns. Quick reference:
 
 Always run the relevant linters and tests before committing.
 Use conventional commit messages for all commits (e.g. `feat(backend): add API`).
-Types: - feat - fix - refactor - ci - dx (developer experience)
-Scopes: - platform - platform/library - platform/marketplace - backend - backend/executor - frontend - frontend/library - frontend/marketplace - blocks
+  Types:
+    - feat
+    - fix
+    - refactor
+    - ci
+    - dx (developer experience)
+  Scopes:
+    - platform
+      - platform/library
+      - platform/marketplace
+      - backend
+        - backend/executor
+      - frontend
+        - frontend/library
+        - frontend/marketplace
+      - blocks
 
 ## Pull requests
 

README.md

@@ -54,7 +54,7 @@ Before proceeding with the installation, ensure your system meets the following
 ### Updated Setup Instructions:
 We've moved to a fully maintained and regularly updated documentation site.
 
-👉 [Follow the official self-hosting guide here](https://agpt.co/docs/platform/getting-started/getting-started)
+👉 [Follow the official self-hosting guide here](https://docs.agpt.co/platform/getting-started/)
 
 
 This tutorial assumes you have Docker, VSCode, git and npm installed.

autogpt_platform/CLAUDE.md

@@ -6,30 +6,152 @@ 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
 
-## Component Documentation
+## Essential Commands
 
-- **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
+### Backend Development
 
-## Key Concepts
+```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
+```
+
+**📖 Complete Guide**: See `/frontend/CONTRIBUTING.md` and `/frontend/.cursorrules` for comprehensive frontend patterns.
+
+**Key Frontend Conventions:**
+
+- Separate render logic from data/behavior in components
+- Use generated API hooks from `@/app/api/__generated__/endpoints/`
+- Use function declarations (not arrow functions) for components/handlers
+- Use design system components from `src/components/` (atoms, molecules, organisms)
+- Only use Phosphor Icons
+- Never use `src/components/__legacy__/*` or deprecated `BackendAPI`
+
+## 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
 
 1. **Agent Graphs**: Workflow definitions stored as JSON, executed by the backend
-2. **Blocks**: Reusable components in `backend/backend/blocks/` that perform specific tasks
+2. **Blocks**: Reusable components in `/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
 
@@ -45,12 +167,83 @@ AutoGPT Platform is a monorepo containing:
 - 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.
+
+**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 strictly needed
+- Avoid comments at all times unless the code is very complex
+
+### 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 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
+- 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/
 - Run the github pre-commit hooks to ensure code quality.
 
 ### Reviewing/Revising Pull Requests

autogpt_platform/backend/.env.default

@@ -178,10 +178,5 @@ AYRSHARE_JWT_KEY=
 SMARTLEAD_API_KEY=
 ZEROBOUNCE_API_KEY=
 
-# PostHog Analytics
-# Get API key from https://posthog.com - Project Settings > Project API Key
-POSTHOG_API_KEY=
-POSTHOG_HOST=https://eu.i.posthog.com
-
 # Other Services
 AUTOMOD_API_KEY=

autogpt_platform/backend/CLAUDE.md

@@ -1,170 +0,0 @@
-# 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/api/conftest.py`:
+Two global auth fixtures are provided by `backend/server/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/external/v1/routes.py

@@ -86,8 +86,6 @@ async def execute_graph_block(
     obj = backend.data.block.get_block(block_id)
     if not obj:
         raise HTTPException(status_code=404, detail=f"Block #{block_id} not found.")
-    if obj.disabled:
-        raise HTTPException(status_code=403, detail=f"Block #{block_id} is disabled.")
 
     output = defaultdict(list)
     async for name, data in obj.execute(data):

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

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

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

@@ -33,15 +33,9 @@ class ChatConfig(BaseSettings):
 
     stream_timeout: int = Field(default=300, description="Stream timeout in seconds")
     max_retries: int = Field(default=3, description="Maximum number of retries")
-    max_agent_runs: int = Field(default=30, description="Maximum number of agent runs")
+    max_agent_runs: int = Field(default=3, description="Maximum number of agent runs")
     max_agent_schedules: int = Field(
-        default=30, description="Maximum number of agent schedules"
-    )
-
-    # Long-running operation configuration
-    long_running_operation_ttl: int = Field(
-        default=600,
-        description="TTL in seconds for long-running operation tracking in Redis (safety net if pod dies)",
+        default=3, description="Maximum number of agent schedules"
     )
 
     # Langfuse Prompt Management Configuration

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

@@ -247,45 +247,3 @@ async def get_chat_session_message_count(session_id: str) -> int:
     """Get the number of messages in a chat session."""
     count = await PrismaChatMessage.prisma().count(where={"sessionId": session_id})
     return count
-
-
-async def update_tool_message_content(
-    session_id: str,
-    tool_call_id: str,
-    new_content: str,
-) -> bool:
-    """Update the content of a tool message in chat history.
-
-    Used by background tasks to update pending operation messages with final results.
-
-    Args:
-        session_id: The chat session ID.
-        tool_call_id: The tool call ID to find the message.
-        new_content: The new content to set.
-
-    Returns:
-        True if a message was updated, False otherwise.
-    """
-    try:
-        result = await PrismaChatMessage.prisma().update_many(
-            where={
-                "sessionId": session_id,
-                "toolCallId": tool_call_id,
-            },
-            data={
-                "content": new_content,
-            },
-        )
-        if result == 0:
-            logger.warning(
-                f"No message found to update for session {session_id}, "
-                f"tool_call_id {tool_call_id}"
-            )
-            return False
-        return True
-    except Exception as e:
-        logger.error(
-            f"Failed to update tool message for session {session_id}, "
-            f"tool_call_id {tool_call_id}: {e}"
-        )
-        return False

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

@@ -1,9 +1,8 @@
-import asyncio
 import logging
 import uuid
+from contextlib import asynccontextmanager
 from datetime import UTC, datetime
 from typing import Any
-from weakref import WeakValueDictionary
 
 from openai.types.chat import (
     ChatCompletionAssistantMessageParam,
@@ -52,28 +51,36 @@ def _get_session_cache_key(session_id: str) -> str:
     return f"{CHAT_SESSION_CACHE_PREFIX}{session_id}"
 
 
-# Session-level locks to prevent race conditions during concurrent upserts.
-# Uses WeakValueDictionary to automatically garbage collect locks when no longer referenced,
-# preventing unbounded memory growth while maintaining lock semantics for active sessions.
-# Invalidation: Locks are auto-removed by GC when no coroutine holds a reference (after
-# async with lock: completes). Explicit cleanup also occurs in delete_chat_session().
-_session_locks: WeakValueDictionary[str, asyncio.Lock] = WeakValueDictionary()
-_session_locks_mutex = asyncio.Lock()
+CHAT_SESSION_LOCK_PREFIX = "chat:session_lock:"
+CHAT_SESSION_LOCK_TIMEOUT = 60  # seconds
 
 
-async def _get_session_lock(session_id: str) -> asyncio.Lock:
-    """Get or create a lock for a specific session to prevent concurrent upserts.
+@asynccontextmanager
+async def _session_lock(session_id: str):
+    """Distributed lock for a chat session using Redis.
 
-    Uses WeakValueDictionary for automatic cleanup: locks are garbage collected
-    when no coroutine holds a reference to them, preventing memory leaks from
-    unbounded growth of session locks.
+    Provides system-wide locking across horizontally scaled backend instances
+    to prevent race conditions during concurrent session mutations.
     """
-    async with _session_locks_mutex:
-        lock = _session_locks.get(session_id)
-        if lock is None:
-            lock = asyncio.Lock()
-            _session_locks[session_id] = lock
-        return lock
+    async_redis = await get_redis_async()
+    lock_key = _get_session_lock_key(session_id)
+    lock = async_redis.lock(lock_key, timeout=CHAT_SESSION_LOCK_TIMEOUT)
+    try:
+        await lock.acquire()
+        yield
+    finally:
+        if await lock.locked() and await lock.owned():
+            try:
+                await lock.release()
+            except Exception as e:
+                logger.warning(
+                    f"Failed to release lock for chat session #{session_id}: {e}"
+                )
+
+
+def _get_session_lock_key(session_id: str) -> str:
+    """Get the Redis lock key for a chat session."""
+    return f"{CHAT_SESSION_LOCK_PREFIX}{session_id}"
 
 
 class ChatMessage(BaseModel):
@@ -295,21 +302,6 @@ async def cache_chat_session(session: ChatSession) -> None:
     await _cache_session(session)
 
 
-async def invalidate_session_cache(session_id: str) -> None:
-    """Invalidate a chat session from Redis cache.
-
-    Used by background tasks to ensure fresh data is loaded on next access.
-    This is best-effort - Redis failures are logged but don't fail the operation.
-    """
-    try:
-        redis_key = _get_session_cache_key(session_id)
-        async_redis = await get_redis_async()
-        await async_redis.delete(redis_key)
-    except Exception as e:
-        # Best-effort: log but don't fail - cache will expire naturally
-        logger.warning(f"Failed to invalidate session cache for {session_id}: {e}")
-
-
 async def _get_session_from_db(session_id: str) -> ChatSession | None:
     """Get a chat session from the database."""
     prisma_session = await chat_db.get_chat_session(session_id)
@@ -454,10 +446,8 @@ async def upsert_chat_session(
             callers are aware of the persistence failure.
         RedisError: If the cache write fails (after successful DB write).
     """
-    # Acquire session-specific lock to prevent concurrent upserts
-    lock = await _get_session_lock(session.session_id)
-
-    async with lock:
+    # Acquire distributed session-specific lock to prevent concurrent upserts
+    async with _session_lock(session.session_id):
         # Get existing message count from DB for incremental saves
         existing_message_count = await chat_db.get_chat_session_message_count(
             session.session_id
@@ -568,7 +558,7 @@ async def delete_chat_session(session_id: str, user_id: str | None = None) -> bo
     if not deleted:
         return False
 
-    # Only invalidate cache and clean up lock after DB confirms deletion
+    # Invalidate cache after DB confirms deletion
     try:
         redis_key = _get_session_cache_key(session_id)
         async_redis = await get_redis_async()
@@ -576,10 +566,6 @@ async def delete_chat_session(session_id: str, user_id: str | None = None) -> bo
     except Exception as e:
         logger.warning(f"Failed to delete session {session_id} from cache: {e}")
 
-    # Clean up session lock (belt-and-suspenders with WeakValueDictionary)
-    async with _session_locks_mutex:
-        _session_locks.pop(session_id, None)
-
     return True
 
 

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

@@ -31,7 +31,6 @@ class ResponseType(str, Enum):
     # Other
     ERROR = "error"
     USAGE = "usage"
-    HEARTBEAT = "heartbeat"
 
 
 class StreamBaseResponse(BaseModel):
@@ -143,20 +142,3 @@ class StreamError(StreamBaseResponse):
     details: dict[str, Any] | None = Field(
         default=None, description="Additional error details"
     )
-
-
-class StreamHeartbeat(StreamBaseResponse):
-    """Heartbeat to keep SSE connection alive during long-running operations.
-
-    Uses SSE comment format (: comment) which is ignored by clients but keeps
-    the connection alive through proxies and load balancers.
-    """
-
-    type: ResponseType = ResponseType.HEARTBEAT
-    toolCallId: str | None = Field(
-        default=None, description="Tool call ID if heartbeat is for a specific tool"
-    )
-
-    def to_sse(self) -> str:
-        """Convert to SSE comment format to keep connection alive."""
-        return ": heartbeat\n\n"

autogpt_platform/backend/backend/api/features/chat/tools/IDEAS.md

@@ -1,79 +0,0 @@
-# CoPilot Tools - Future Ideas
-
-## Multimodal Image Support for CoPilot
-
-**Problem:** CoPilot uses a vision-capable model but can't "see" workspace images. When a block generates an image and returns `workspace://abc123`, CoPilot can't evaluate it (e.g., checking blog thumbnail quality).
-
-**Backend Solution:**
-When preparing messages for the LLM, detect `workspace://` image references and convert them to proper image content blocks:
-
-```python
-# Before sending to LLM, scan for workspace image references
-# and inject them as image content parts
-
-# Example message transformation:
-# FROM: {"role": "assistant", "content": "Generated image: workspace://abc123"}
-# TO:   {"role": "assistant", "content": [
-#         {"type": "text", "text": "Generated image: workspace://abc123"},
-#         {"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}}
-#       ]}
-```
-
-**Where to implement:**
-- In the chat stream handler before calling the LLM
-- Or in a message preprocessing step
-- Need to fetch image from workspace, convert to base64, add as image content
-
-**Considerations:**
-- Only do this for image MIME types (image/png, image/jpeg, etc.)
-- May want a size limit (don't pass 10MB images)
-- Track which images were "shown" to the AI for frontend indicator
-- Cost implications - vision API calls are more expensive
-
-**Frontend Solution:**
-Show visual indicator on workspace files in chat:
-- If AI saw the image: normal display
-- If AI didn't see it: overlay icon saying "AI can't see this image"
-
-Requires response metadata indicating which `workspace://` refs were passed to the model.
-
----
-
-## Output Post-Processing Layer for run_block
-
-**Problem:** Many blocks produce large outputs that:
-- Consume massive context (100KB base64 image = ~133KB tokens)
-- Can't fit in conversation
-- Break things and cause high LLM costs
-
-**Proposed Solution:** Instead of modifying individual blocks or `store_media_file()`, implement a centralized output processor in `run_block.py` that handles outputs before they're returned to CoPilot.
-
-**Benefits:**
-1. **Centralized** - one place to handle all output processing
-2. **Future-proof** - new blocks automatically get output processing
-3. **Keeps blocks pure** - they don't need to know about context constraints
-4. **Handles all large outputs** - not just images
-
-**Processing Rules:**
-- Detect base64 data URIs → save to workspace, return `workspace://` reference
-- Truncate very long strings (>N chars) with truncation note
-- Summarize large arrays/lists (e.g., "Array with 1000 items, first 5: [...]")
-- Handle nested large outputs in dicts recursively
-- Cap total output size
-
-**Implementation Location:** `run_block.py` after block execution, before returning `BlockOutputResponse`
-
-**Example:**
-```python
-def _process_outputs_for_context(
-    outputs: dict[str, list[Any]],
-    workspace_manager: WorkspaceManager,
-    max_string_length: int = 10000,
-    max_array_preview: int = 5,
-) -> dict[str, list[Any]]:
-    """Process block outputs to prevent context bloat."""
-    processed = {}
-    for name, values in outputs.items():
-        processed[name] = [_process_value(v, workspace_manager) for v in values]
-    return processed
-```

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

@@ -1,10 +1,8 @@
-import logging
 from typing import TYPE_CHECKING, Any
 
 from openai.types.chat import ChatCompletionToolParam
 
 from backend.api.features.chat.model import ChatSession
-from backend.api.features.chat.tracking import track_tool_called
 
 from .add_understanding import AddUnderstandingTool
 from .agent_output import AgentOutputTool
@@ -18,18 +16,10 @@ from .get_doc_page import GetDocPageTool
 from .run_agent import RunAgentTool
 from .run_block import RunBlockTool
 from .search_docs import SearchDocsTool
-from .workspace_files import (
-    DeleteWorkspaceFileTool,
-    ListWorkspaceFilesTool,
-    ReadWorkspaceFileTool,
-    WriteWorkspaceFileTool,
-)
 
 if TYPE_CHECKING:
     from backend.api.features.chat.response_model import StreamToolOutputAvailable
 
-logger = logging.getLogger(__name__)
-
 # Single source of truth for all tools
 TOOL_REGISTRY: dict[str, BaseTool] = {
     "add_understanding": AddUnderstandingTool(),
@@ -43,11 +33,6 @@ TOOL_REGISTRY: dict[str, BaseTool] = {
     "view_agent_output": AgentOutputTool(),
     "search_docs": SearchDocsTool(),
     "get_doc_page": GetDocPageTool(),
-    # Workspace tools for CoPilot file operations
-    "list_workspace_files": ListWorkspaceFilesTool(),
-    "read_workspace_file": ReadWorkspaceFileTool(),
-    "write_workspace_file": WriteWorkspaceFileTool(),
-    "delete_workspace_file": DeleteWorkspaceFileTool(),
 }
 
 # Export individual tool instances for backwards compatibility
@@ -60,11 +45,6 @@ tools: list[ChatCompletionToolParam] = [
 ]
 
 
-def get_tool(tool_name: str) -> BaseTool | None:
-    """Get a tool instance by name."""
-    return TOOL_REGISTRY.get(tool_name)
-
-
 async def execute_tool(
     tool_name: str,
     parameters: dict[str, Any],
@@ -73,20 +53,7 @@ async def execute_tool(
     tool_call_id: str,
 ) -> "StreamToolOutputAvailable":
     """Execute a tool by name."""
-    tool = get_tool(tool_name)
+    tool = TOOL_REGISTRY.get(tool_name)
     if not tool:
         raise ValueError(f"Tool {tool_name} not found")
-
-    # Track tool call in PostHog
-    logger.info(
-        f"Tracking tool call: tool={tool_name}, user={user_id}, "
-        f"session={session.session_id}, call_id={tool_call_id}"
-    )
-    track_tool_called(
-        user_id=user_id,
-        session_id=session.session_id,
-        tool_name=tool_name,
-        tool_call_id=tool_call_id,
-    )
-
     return await tool.execute(user_id, session, tool_call_id, **parameters)

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

@@ -3,6 +3,8 @@
 import logging
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 from backend.data.understanding import (
     BusinessUnderstandingInput,
@@ -59,6 +61,7 @@ and automations for the user's specific needs."""
         """Requires authentication to store user-specific data."""
         return True
 
+    @observe(as_type="tool", name="add_understanding")
     async def _execute(
         self,
         user_id: str | None,

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

@@ -2,54 +2,27 @@
 
 from .core import (
     AgentGeneratorNotConfiguredError,
-    AgentJsonValidationError,
-    AgentSummary,
-    DecompositionResult,
-    DecompositionStep,
-    LibraryAgentSummary,
-    MarketplaceAgentSummary,
     decompose_goal,
-    enrich_library_agents_from_steps,
-    extract_search_terms_from_steps,
-    extract_uuids_from_text,
     generate_agent,
     generate_agent_patch,
     get_agent_as_json,
-    get_all_relevant_agents_for_generation,
-    get_library_agent_by_graph_id,
-    get_library_agent_by_id,
-    get_library_agents_for_generation,
     json_to_graph,
     save_agent_to_library,
-    search_marketplace_agents_for_generation,
 )
-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
 
 __all__ = [
-    "AgentGeneratorNotConfiguredError",
-    "AgentJsonValidationError",
-    "AgentSummary",
-    "DecompositionResult",
-    "DecompositionStep",
-    "LibraryAgentSummary",
-    "MarketplaceAgentSummary",
-    "check_external_service_health",
+    # Core functions
     "decompose_goal",
-    "enrich_library_agents_from_steps",
-    "extract_search_terms_from_steps",
-    "extract_uuids_from_text",
     "generate_agent",
     "generate_agent_patch",
-    "get_agent_as_json",
-    "get_all_relevant_agents_for_generation",
-    "get_library_agent_by_graph_id",
-    "get_library_agent_by_id",
-    "get_library_agents_for_generation",
-    "get_user_message_for_error",
-    "is_external_service_configured",
-    "json_to_graph",
     "save_agent_to_library",
-    "search_marketplace_agents_for_generation",
+    "get_agent_as_json",
+    "json_to_graph",
+    # Exceptions
+    "AgentGeneratorNotConfiguredError",
+    # Service
+    "is_external_service_configured",
+    "check_external_service_health",
 ]

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

@@ -1,22 +1,11 @@
 """Core agent generation functions."""
 
 import logging
-import re
 import uuid
-from typing import Any, NotRequired, TypedDict
+from typing import Any
 
 from backend.api.features.library import db as library_db
-from backend.api.features.store import db as store_db
-from backend.data.graph import (
-    Graph,
-    Link,
-    Node,
-    create_graph,
-    get_graph,
-    get_graph_all_versions,
-    get_store_listed_graphs,
-)
-from backend.util.exceptions import DatabaseError, NotFoundError
+from backend.data.graph import Graph, Link, Node, create_graph
 
 from .service import (
     decompose_goal_external,
@@ -27,74 +16,6 @@ from .service import (
 
 logger = logging.getLogger(__name__)
 
-AGENT_EXECUTOR_BLOCK_ID = "e189baac-8c20-45a1-94a7-55177ea42565"
-
-
-class ExecutionSummary(TypedDict):
-    """Summary of a single execution for quality assessment."""
-
-    status: str
-    correctness_score: NotRequired[float]
-    activity_summary: NotRequired[str]
-
-
-class LibraryAgentSummary(TypedDict):
-    """Summary of a library agent for sub-agent composition.
-
-    Includes recent executions to help the LLM decide whether to use this agent.
-    Each execution shows status, correctness_score (0-1), and activity_summary.
-    """
-
-    graph_id: str
-    graph_version: int
-    name: str
-    description: str
-    input_schema: dict[str, Any]
-    output_schema: dict[str, Any]
-    recent_executions: NotRequired[list[ExecutionSummary]]
-
-
-class MarketplaceAgentSummary(TypedDict):
-    """Summary of a marketplace agent for sub-agent composition."""
-
-    name: str
-    description: str
-    sub_heading: str
-    creator: str
-    is_marketplace_agent: bool
-
-
-class DecompositionStep(TypedDict, total=False):
-    """A single step in decomposed instructions."""
-
-    description: str
-    action: str
-    block_name: str
-    tool: str
-    name: str
-
-
-class DecompositionResult(TypedDict, total=False):
-    """Result from decompose_goal - can be instructions, questions, or error."""
-
-    type: str
-    steps: list[DecompositionStep]
-    questions: list[dict[str, Any]]
-    error: str
-    error_type: str
-
-
-AgentSummary = LibraryAgentSummary | MarketplaceAgentSummary | dict[str, Any]
-
-
-def _to_dict_list(
-    agents: list[AgentSummary] | list[dict[str, Any]] | None,
-) -> list[dict[str, Any]] | None:
-    """Convert typed agent summaries to plain dicts for external service calls."""
-    if agents is None:
-        return None
-    return [dict(a) for a in agents]
-
 
 class AgentGeneratorNotConfiguredError(Exception):
     """Raised when the external Agent Generator service is not configured."""
@@ -115,422 +36,15 @@ def _check_service_configured() -> None:
         )
 
 
-_UUID_PATTERN = re.compile(
-    r"[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}",
-    re.IGNORECASE,
-)
-
-
-def extract_uuids_from_text(text: str) -> list[str]:
-    """Extract all UUID v4 strings from text.
-
-    Args:
-        text: Text that may contain UUIDs (e.g., user's goal description)
-
-    Returns:
-        List of unique UUIDs found in the text (lowercase)
-    """
-    matches = _UUID_PATTERN.findall(text)
-    return list({m.lower() for m in matches})
-
-
-async def get_library_agent_by_id(
-    user_id: str, agent_id: str
-) -> LibraryAgentSummary | None:
-    """Fetch a specific library agent by its ID (library agent ID or graph_id).
-
-    This function tries multiple lookup strategies:
-    1. First tries to find by graph_id (AgentGraph primary key)
-    2. If not found, tries to find by library agent ID (LibraryAgent primary key)
-
-    This handles both cases:
-    - User provides graph_id (e.g., from AgentExecutorBlock)
-    - User provides library agent ID (e.g., from library URL)
-
-    Args:
-        user_id: The user ID
-        agent_id: The ID to look up (can be graph_id or library agent ID)
-
-    Returns:
-        LibraryAgentSummary if found, None otherwise
-    """
-    try:
-        agent = await library_db.get_library_agent_by_graph_id(user_id, agent_id)
-        if agent:
-            logger.debug(f"Found library agent by graph_id: {agent.name}")
-            return LibraryAgentSummary(
-                graph_id=agent.graph_id,
-                graph_version=agent.graph_version,
-                name=agent.name,
-                description=agent.description,
-                input_schema=agent.input_schema,
-                output_schema=agent.output_schema,
-            )
-    except DatabaseError:
-        raise
-    except Exception as e:
-        logger.debug(f"Could not fetch library agent by graph_id {agent_id}: {e}")
-
-    try:
-        agent = await library_db.get_library_agent(agent_id, user_id)
-        if agent:
-            logger.debug(f"Found library agent by library_id: {agent.name}")
-            return LibraryAgentSummary(
-                graph_id=agent.graph_id,
-                graph_version=agent.graph_version,
-                name=agent.name,
-                description=agent.description,
-                input_schema=agent.input_schema,
-                output_schema=agent.output_schema,
-            )
-    except NotFoundError:
-        logger.debug(f"Library agent not found by library_id: {agent_id}")
-    except DatabaseError:
-        raise
-    except Exception as e:
-        logger.warning(
-            f"Could not fetch library agent by library_id {agent_id}: {e}",
-            exc_info=True,
-        )
-
-    return None
-
-
-get_library_agent_by_graph_id = get_library_agent_by_id
-
-
-async def get_library_agents_for_generation(
-    user_id: str,
-    search_query: str | None = None,
-    exclude_graph_id: str | None = None,
-    max_results: int = 15,
-) -> list[LibraryAgentSummary]:
-    """Fetch user's library agents formatted for Agent Generator.
-
-    Uses search-based fetching to return relevant agents instead of all agents.
-    This is more scalable for users with large libraries.
-
-    Includes recent_executions list to help the LLM assess agent quality:
-    - Each execution has status, correctness_score (0-1), and activity_summary
-    - This gives the LLM concrete examples of recent performance
-
-    Args:
-        user_id: The user ID
-        search_query: Optional search term to find relevant agents (user's goal/description)
-        exclude_graph_id: Optional graph ID to exclude (prevents circular references)
-        max_results: Maximum number of agents to return (default 15)
-
-    Returns:
-        List of LibraryAgentSummary with schemas and recent executions for sub-agent composition
-    """
-    try:
-        response = await library_db.list_library_agents(
-            user_id=user_id,
-            search_term=search_query,
-            page=1,
-            page_size=max_results,
-            include_executions=True,
-        )
-
-        results: list[LibraryAgentSummary] = []
-        for agent in response.agents:
-            if exclude_graph_id is not None and agent.graph_id == exclude_graph_id:
-                continue
-
-            summary = LibraryAgentSummary(
-                graph_id=agent.graph_id,
-                graph_version=agent.graph_version,
-                name=agent.name,
-                description=agent.description,
-                input_schema=agent.input_schema,
-                output_schema=agent.output_schema,
-            )
-            if agent.recent_executions:
-                exec_summaries: list[ExecutionSummary] = []
-                for ex in agent.recent_executions:
-                    exec_sum = ExecutionSummary(status=ex.status)
-                    if ex.correctness_score is not None:
-                        exec_sum["correctness_score"] = ex.correctness_score
-                    if ex.activity_summary:
-                        exec_sum["activity_summary"] = ex.activity_summary
-                    exec_summaries.append(exec_sum)
-                summary["recent_executions"] = exec_summaries
-            results.append(summary)
-        return results
-    except DatabaseError:
-        raise
-    except Exception as e:
-        logger.warning(f"Failed to fetch library agents: {e}")
-        return []
-
-
-async def search_marketplace_agents_for_generation(
-    search_query: str,
-    max_results: int = 10,
-) -> list[LibraryAgentSummary]:
-    """Search marketplace agents formatted for Agent Generator.
-
-    Fetches marketplace agents and their full schemas so they can be used
-    as sub-agents in generated workflows.
-
-    Args:
-        search_query: Search term to find relevant public agents
-        max_results: Maximum number of agents to return (default 10)
-
-    Returns:
-        List of LibraryAgentSummary with full input/output schemas
-    """
-    try:
-        response = await store_db.get_store_agents(
-            search_query=search_query,
-            page=1,
-            page_size=max_results,
-        )
-
-        agents_with_graphs = [
-            agent for agent in response.agents if agent.agent_graph_id
-        ]
-
-        if not agents_with_graphs:
-            return []
-
-        graph_ids = [agent.agent_graph_id for agent in agents_with_graphs]
-        graphs = await get_store_listed_graphs(*graph_ids)
-
-        results: list[LibraryAgentSummary] = []
-        for agent in agents_with_graphs:
-            graph_id = agent.agent_graph_id
-            if graph_id and graph_id in graphs:
-                graph = graphs[graph_id]
-                results.append(
-                    LibraryAgentSummary(
-                        graph_id=graph.id,
-                        graph_version=graph.version,
-                        name=agent.agent_name,
-                        description=agent.description,
-                        input_schema=graph.input_schema,
-                        output_schema=graph.output_schema,
-                    )
-                )
-        return results
-    except Exception as e:
-        logger.warning(f"Failed to search marketplace agents: {e}")
-        return []
-
-
-async def get_all_relevant_agents_for_generation(
-    user_id: str,
-    search_query: str | None = None,
-    exclude_graph_id: str | None = None,
-    include_library: bool = True,
-    include_marketplace: bool = True,
-    max_library_results: int = 15,
-    max_marketplace_results: int = 10,
-) -> list[AgentSummary]:
-    """Fetch relevant agents from library and/or marketplace.
-
-    Searches both user's library and marketplace by default.
-    Explicitly mentioned UUIDs in the search query are always looked up.
-
-    Args:
-        user_id: The user ID
-        search_query: Search term to find relevant agents (user's goal/description)
-        exclude_graph_id: Optional graph ID to exclude (prevents circular references)
-        include_library: Whether to search user's library (default True)
-        include_marketplace: Whether to also search marketplace (default True)
-        max_library_results: Max library agents to return (default 15)
-        max_marketplace_results: Max marketplace agents to return (default 10)
-
-    Returns:
-        List of AgentSummary with full schemas (both library and marketplace agents)
-    """
-    agents: list[AgentSummary] = []
-    seen_graph_ids: set[str] = set()
-
-    if search_query:
-        mentioned_uuids = extract_uuids_from_text(search_query)
-        for graph_id in mentioned_uuids:
-            if graph_id == exclude_graph_id:
-                continue
-            agent = await get_library_agent_by_graph_id(user_id, graph_id)
-            agent_graph_id = agent.get("graph_id") if agent else None
-            if agent and agent_graph_id and agent_graph_id not in seen_graph_ids:
-                agents.append(agent)
-                seen_graph_ids.add(agent_graph_id)
-                logger.debug(
-                    f"Found explicitly mentioned agent: {agent.get('name') or 'Unknown'}"
-                )
-
-    if include_library:
-        library_agents = await get_library_agents_for_generation(
-            user_id=user_id,
-            search_query=search_query,
-            exclude_graph_id=exclude_graph_id,
-            max_results=max_library_results,
-        )
-        for agent in library_agents:
-            graph_id = agent.get("graph_id")
-            if graph_id and graph_id not in seen_graph_ids:
-                agents.append(agent)
-                seen_graph_ids.add(graph_id)
-
-    if include_marketplace and search_query:
-        marketplace_agents = await search_marketplace_agents_for_generation(
-            search_query=search_query,
-            max_results=max_marketplace_results,
-        )
-        for agent in marketplace_agents:
-            graph_id = agent.get("graph_id")
-            if graph_id and graph_id not in seen_graph_ids:
-                agents.append(agent)
-                seen_graph_ids.add(graph_id)
-
-    return agents
-
-
-def extract_search_terms_from_steps(
-    decomposition_result: DecompositionResult | dict[str, Any],
-) -> list[str]:
-    """Extract search terms from decomposed instruction steps.
-
-    Analyzes the decomposition result to extract relevant keywords
-    for additional library agent searches.
-
-    Args:
-        decomposition_result: Result from decompose_goal containing steps
-
-    Returns:
-        List of unique search terms extracted from steps
-    """
-    search_terms: list[str] = []
-
-    if decomposition_result.get("type") != "instructions":
-        return search_terms
-
-    steps = decomposition_result.get("steps", [])
-    if not steps:
-        return search_terms
-
-    step_keys: list[str] = ["description", "action", "block_name", "tool", "name"]
-
-    for step in steps:
-        for key in step_keys:
-            value = step.get(key)  # type: ignore[union-attr]
-            if isinstance(value, str) and len(value) > 3:
-                search_terms.append(value)
-
-    seen: set[str] = set()
-    unique_terms: list[str] = []
-    for term in search_terms:
-        term_lower = term.lower()
-        if term_lower not in seen:
-            seen.add(term_lower)
-            unique_terms.append(term)
-
-    return unique_terms
-
-
-async def enrich_library_agents_from_steps(
-    user_id: str,
-    decomposition_result: DecompositionResult | dict[str, Any],
-    existing_agents: list[AgentSummary] | list[dict[str, Any]],
-    exclude_graph_id: str | None = None,
-    include_marketplace: bool = True,
-    max_additional_results: int = 10,
-) -> list[AgentSummary] | list[dict[str, Any]]:
-    """Enrich library agents list with additional searches based on decomposed steps.
-
-    This implements two-phase search: after decomposition, we search for additional
-    relevant agents based on the specific steps identified.
-
-    Args:
-        user_id: The user ID
-        decomposition_result: Result from decompose_goal containing steps
-        existing_agents: Already fetched library agents from initial search
-        exclude_graph_id: Optional graph ID to exclude
-        include_marketplace: Whether to also search marketplace
-        max_additional_results: Max additional agents per search term (default 10)
-
-    Returns:
-        Combined list of library agents (existing + newly discovered)
-    """
-    search_terms = extract_search_terms_from_steps(decomposition_result)
-
-    if not search_terms:
-        return existing_agents
-
-    existing_ids: set[str] = set()
-    existing_names: set[str] = set()
-
-    for agent in existing_agents:
-        agent_name = agent.get("name")
-        if agent_name and isinstance(agent_name, str):
-            existing_names.add(agent_name.lower())
-        graph_id = agent.get("graph_id")  # type: ignore[call-overload]
-        if graph_id and isinstance(graph_id, str):
-            existing_ids.add(graph_id)
-
-    all_agents: list[AgentSummary] | list[dict[str, Any]] = list(existing_agents)
-
-    for term in search_terms[:3]:
-        try:
-            additional_agents = await get_all_relevant_agents_for_generation(
-                user_id=user_id,
-                search_query=term,
-                exclude_graph_id=exclude_graph_id,
-                include_marketplace=include_marketplace,
-                max_library_results=max_additional_results,
-                max_marketplace_results=5,
-            )
-
-            for agent in additional_agents:
-                agent_name = agent.get("name")
-                if not agent_name or not isinstance(agent_name, str):
-                    continue
-                agent_name_lower = agent_name.lower()
-
-                if agent_name_lower in existing_names:
-                    continue
-
-                graph_id = agent.get("graph_id")  # type: ignore[call-overload]
-                if graph_id and graph_id in existing_ids:
-                    continue
-
-                all_agents.append(agent)
-                existing_names.add(agent_name_lower)
-                if graph_id and isinstance(graph_id, str):
-                    existing_ids.add(graph_id)
-
-        except DatabaseError:
-            logger.error(f"Database error searching for agents with term '{term}'")
-            raise
-        except Exception as e:
-            logger.warning(
-                f"Failed to search for additional agents with term '{term}': {e}"
-            )
-
-    logger.debug(
-        f"Enriched library agents: {len(existing_agents)} initial + "
-        f"{len(all_agents) - len(existing_agents)} additional = {len(all_agents)} total"
-    )
-
-    return all_agents
-
-
-async def decompose_goal(
-    description: str,
-    context: str = "",
-    library_agents: list[AgentSummary] | None = None,
-) -> DecompositionResult | None:
+async def decompose_goal(description: str, context: str = "") -> dict[str, Any] | None:
     """Break down a goal into steps or return clarifying questions.
 
     Args:
         description: Natural language goal description
         context: Additional context (e.g., answers to previous questions)
-        library_agents: User's library agents available for sub-agent composition
 
     Returns:
-        DecompositionResult with either:
+        Dict with either:
         - {"type": "clarifying_questions", "questions": [...]}
         - {"type": "instructions", "steps": [...]}
         Or None on error
@@ -540,36 +54,26 @@ async def decompose_goal(
     """
     _check_service_configured()
     logger.info("Calling external Agent Generator service for decompose_goal")
-    result = await decompose_goal_external(
-        description, context, _to_dict_list(library_agents)
-    )
-    return result  # type: ignore[return-value]
+    return await decompose_goal_external(description, context)
 
 
-async def generate_agent(
-    instructions: DecompositionResult | dict[str, Any],
-    library_agents: list[AgentSummary] | list[dict[str, Any]] | 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
-        library_agents: User's library agents available for sub-agent composition
 
     Returns:
-        Agent JSON dict, error dict {"type": "error", ...}, or None on error
+        Agent JSON dict 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(
-        dict(instructions), _to_dict_list(library_agents)
-    )
+    result = await generate_agent_external(instructions)
     if result:
-        if isinstance(result, dict) and result.get("type") == "error":
-            return result
+        # Ensure required fields
         if "id" not in result:
             result["id"] = str(uuid.uuid4())
         if "version" not in result:
@@ -579,12 +83,6 @@ async def generate_agent(
     return result
 
 
-class AgentJsonValidationError(Exception):
-    """Raised when agent JSON is invalid or missing required fields."""
-
-    pass
-
-
 def json_to_graph(agent_json: dict[str, Any]) -> Graph:
     """Convert agent JSON dict to Graph model.
 
@@ -593,55 +91,25 @@ def json_to_graph(agent_json: dict[str, Any]) -> Graph:
 
     Returns:
         Graph ready for saving
-
-    Raises:
-        AgentJsonValidationError: If required fields are missing from nodes or links
     """
     nodes = []
-    for idx, n in enumerate(agent_json.get("nodes", [])):
-        block_id = n.get("block_id")
-        if not block_id:
-            node_id = n.get("id", f"index_{idx}")
-            raise AgentJsonValidationError(
-                f"Node '{node_id}' is missing required field 'block_id'"
-            )
+    for n in agent_json.get("nodes", []):
         node = Node(
             id=n.get("id", str(uuid.uuid4())),
-            block_id=block_id,
+            block_id=n["block_id"],
             input_default=n.get("input_default", {}),
             metadata=n.get("metadata", {}),
         )
         nodes.append(node)
 
     links = []
-    for idx, link_data in enumerate(agent_json.get("links", [])):
-        source_id = link_data.get("source_id")
-        sink_id = link_data.get("sink_id")
-        source_name = link_data.get("source_name")
-        sink_name = link_data.get("sink_name")
-
-        missing_fields = []
-        if not source_id:
-            missing_fields.append("source_id")
-        if not sink_id:
-            missing_fields.append("sink_id")
-        if not source_name:
-            missing_fields.append("source_name")
-        if not sink_name:
-            missing_fields.append("sink_name")
-
-        if missing_fields:
-            link_id = link_data.get("id", f"index_{idx}")
-            raise AgentJsonValidationError(
-                f"Link '{link_id}' is missing required fields: {', '.join(missing_fields)}"
-            )
-
+    for link_data in agent_json.get("links", []):
         link = Link(
             id=link_data.get("id", str(uuid.uuid4())),
-            source_id=source_id,
-            sink_id=sink_id,
-            source_name=source_name,
-            sink_name=sink_name,
+            source_id=link_data["source_id"],
+            sink_id=link_data["sink_id"],
+            source_name=link_data["source_name"],
+            sink_name=link_data["sink_name"],
             is_static=link_data.get("is_static", False),
         )
         links.append(link)
@@ -662,40 +130,22 @@ def _reassign_node_ids(graph: Graph) -> None:
 
     This is needed when creating a new version to avoid unique constraint violations.
     """
+    # Create mapping from old node IDs to new UUIDs
     id_map = {node.id: str(uuid.uuid4()) for node in graph.nodes}
 
+    # Reassign node IDs
     for node in graph.nodes:
         node.id = id_map[node.id]
 
+    # Update link references to use new node IDs
     for link in graph.links:
-        link.id = str(uuid.uuid4())
+        link.id = str(uuid.uuid4())  # Also give links new IDs
         if link.source_id in id_map:
             link.source_id = id_map[link.source_id]
         if link.sink_id in id_map:
             link.sink_id = id_map[link.sink_id]
 
 
-def _populate_agent_executor_user_ids(agent_json: dict[str, Any], user_id: str) -> None:
-    """Populate user_id in AgentExecutorBlock nodes.
-
-    The external agent generator creates AgentExecutorBlock nodes with empty user_id.
-    This function fills in the actual user_id so sub-agents run with correct permissions.
-
-    Args:
-        agent_json: Agent JSON dict (modified in place)
-        user_id: User ID to set
-    """
-    for node in agent_json.get("nodes", []):
-        if node.get("block_id") == AGENT_EXECUTOR_BLOCK_ID:
-            input_default = node.get("input_default") or {}
-            if not input_default.get("user_id"):
-                input_default["user_id"] = user_id
-                node["input_default"] = input_default
-                logger.debug(
-                    f"Set user_id for AgentExecutorBlock node {node.get('id')}"
-                )
-
-
 async def save_agent_to_library(
     agent_json: dict[str, Any], user_id: str, is_update: bool = False
 ) -> tuple[Graph, Any]:
@@ -709,27 +159,33 @@ async def save_agent_to_library(
     Returns:
         Tuple of (created Graph, LibraryAgent)
     """
-    # Populate user_id in AgentExecutorBlock nodes before conversion
-    _populate_agent_executor_user_ids(agent_json, user_id)
+    from backend.data.graph import get_graph_all_versions
 
     graph = json_to_graph(agent_json)
 
     if is_update:
+        # For updates, keep the same graph ID but increment version
+        # and reassign node/link IDs to avoid conflicts
         if graph.id:
             existing_versions = await get_graph_all_versions(graph.id, user_id)
             if existing_versions:
                 latest_version = max(v.version for v in existing_versions)
                 graph.version = latest_version + 1
+                # Reassign node IDs (but keep graph ID the same)
                 _reassign_node_ids(graph)
                 logger.info(f"Updating agent {graph.id} to version {graph.version}")
     else:
+        # For new agents, always generate a fresh UUID to avoid collisions
         graph.id = str(uuid.uuid4())
         graph.version = 1
+        # Reassign all node IDs as well
         _reassign_node_ids(graph)
         logger.info(f"Creating new agent with ID {graph.id}")
 
+    # Save to database
     created_graph = await create_graph(graph, user_id)
 
+    # Add to user's library (or update existing library agent)
     library_agents = await library_db.create_library_agent(
         graph=created_graph,
         user_id=user_id,
@@ -741,31 +197,25 @@ async def save_agent_to_library(
 
 
 async def get_agent_as_json(
-    agent_id: str, user_id: str | None
+    graph_id: str, user_id: str | None
 ) -> dict[str, Any] | None:
     """Fetch an agent and convert to JSON format for editing.
 
     Args:
-        agent_id: Graph ID or library agent ID
+        graph_id: Graph ID or library agent ID
         user_id: User ID
 
     Returns:
         Agent as JSON dict or None if not found
     """
-    graph = await get_graph(agent_id, version=None, user_id=user_id)
-
-    if not graph and user_id:
-        try:
-            library_agent = await library_db.get_library_agent(agent_id, user_id)
-            graph = await get_graph(
-                library_agent.graph_id, version=None, user_id=user_id
-            )
-        except NotFoundError:
-            pass
+    from backend.data.graph import get_graph
 
+    # Try to get the graph (version=None gets the active version)
+    graph = await get_graph(graph_id, version=None, user_id=user_id)
     if not graph:
         return None
 
+    # Convert to JSON format
     nodes = []
     for node in graph.nodes:
         nodes.append(
@@ -803,9 +253,7 @@ async def get_agent_as_json(
 
 
 async def generate_agent_patch(
-    update_request: str,
-    current_agent: dict[str, Any],
-    library_agents: list[AgentSummary] | None = None,
+    update_request: str, current_agent: dict[str, Any]
 ) -> dict[str, Any] | None:
     """Update an existing agent using natural language.
 
@@ -817,17 +265,13 @@ async def generate_agent_patch(
     Args:
         update_request: Natural language description of changes
         current_agent: Current agent JSON
-        library_agents: User's library agents available for sub-agent composition
 
     Returns:
-        Updated agent JSON, clarifying questions dict {"type": "clarifying_questions", ...},
-        error dict {"type": "error", ...}, or None on unexpected error
+        Updated agent JSON, clarifying questions dict, 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_patch")
-    return await generate_agent_patch_external(
-        update_request, current_agent, _to_dict_list(library_agents)
-    )
+    return await generate_agent_patch_external(update_request, current_agent)

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

@@ -1,95 +0,0 @@
-"""Error handling utilities for agent generator."""
-
-import re
-
-
-def _sanitize_error_details(details: str) -> str:
-    """Sanitize error details to remove sensitive information.
-
-    Strips common patterns that could expose internal system info:
-    - File paths (Unix and Windows)
-    - Database connection strings
-    - URLs with credentials
-    - Stack trace internals
-
-    Args:
-        details: Raw error details string
-
-    Returns:
-        Sanitized error details safe for user display
-    """
-    sanitized = re.sub(
-        r"/[a-zA-Z0-9_./\-]+\.(py|js|ts|json|yaml|yml)", "[path]", details
-    )
-    sanitized = re.sub(r"[A-Z]:\\[a-zA-Z0-9_\\.\\-]+", "[path]", sanitized)
-    sanitized = re.sub(
-        r"(postgres|mysql|mongodb|redis)://[^\s]+", "[database_url]", sanitized
-    )
-    sanitized = re.sub(r"https?://[^:]+:[^@]+@[^\s]+", "[url]", sanitized)
-    sanitized = re.sub(r", line \d+", "", sanitized)
-    sanitized = re.sub(r'File "[^"]+",?', "", sanitized)
-
-    return sanitized.strip()
-
-
-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,
-    error_details: 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
-        error_details: Optional additional details about the error
-
-    Returns:
-        User-friendly error message suitable for display to the user
-    """
-    base_message = ""
-
-    if error_type == "llm_parse_error":
-        base_message = (
-            llm_parse_message
-            or "The AI had trouble processing this request. Please try again."
-        )
-    elif error_type == "validation_error":
-        base_message = (
-            validation_message
-            or "The generated agent failed validation. "
-            "This usually happens when the agent structure doesn't match "
-            "what the platform expects. Please try simplifying your goal "
-            "or breaking it into smaller parts."
-        )
-    elif error_type == "patch_error":
-        base_message = (
-            "Failed to apply the changes. The modification couldn't be "
-            "validated. Please try a different approach or simplify the change."
-        )
-    elif error_type in ("timeout", "llm_timeout"):
-        base_message = (
-            "The request took too long to process. This can happen with "
-            "complex agents. Please try again or simplify your goal."
-        )
-    elif error_type in ("rate_limit", "llm_rate_limit"):
-        base_message = "The service is currently busy. Please try again in a moment."
-    else:
-        base_message = f"Failed to {operation}. Please try again."
-
-    if error_details:
-        details = _sanitize_error_details(error_details)
-        if len(details) > 200:
-            details = details[:200] + "..."
-        base_message += f"\n\nTechnical details: {details}"
-
-    return base_message

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

@@ -14,70 +14,6 @@ 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
 
@@ -117,16 +53,13 @@ def _get_client() -> httpx.AsyncClient:
 
 
 async def decompose_goal_external(
-    description: str,
-    context: str = "",
-    library_agents: list[dict[str, Any]] | None = None,
+    description: str, context: str = ""
 ) -> dict[str, Any] | None:
     """Call the external service to decompose a goal.
 
     Args:
         description: Natural language goal description
         context: Additional context (e.g., answers to previous questions)
-        library_agents: User's library agents available for sub-agent composition
 
     Returns:
         Dict with either:
@@ -134,17 +67,15 @@ async def decompose_goal_external(
         - {"type": "instructions", "steps": [...]}
         - {"type": "unachievable_goal", ...}
         - {"type": "vague_goal", ...}
-        - {"type": "error", "error": "...", "error_type": "..."} on error
-        Or None on unexpected error
+        Or None on error
     """
     client = _get_client()
 
-    if context:
-        description = f"{description}\n\nAdditional context from user:\n{context}"
-
+    # Build the request payload
     payload: dict[str, Any] = {"description": description}
-    if library_agents:
-        payload["library_agents"] = library_agents
+    if context:
+        # The external service uses user_instruction for additional context
+        payload["user_instruction"] = context
 
     try:
         response = await client.post("/api/decompose-description", json=payload)
@@ -152,13 +83,8 @@ async def decompose_goal_external(
         data = response.json()
 
         if not data.get("success"):
-            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)
+            logger.error(f"External service returned error: {data.get('error')}")
+            return None
 
         # Map the response to the expected format
         response_type = data.get("type")
@@ -180,120 +106,88 @@ 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 _create_error_response(
-                f"Unknown response type from Agent Generator: {response_type}",
-                "invalid_response",
-            )
+            return None
 
     except httpx.HTTPStatusError as e:
-        error_type, error_msg = _classify_http_error(e)
-        logger.error(error_msg)
-        return _create_error_response(error_msg, error_type)
+        logger.error(f"HTTP error calling external agent generator: {e}")
+        return None
     except httpx.RequestError as e:
-        error_type, error_msg = _classify_request_error(e)
-        logger.error(error_msg)
-        return _create_error_response(error_msg, error_type)
+        logger.error(f"Request error calling external agent generator: {e}")
+        return None
     except Exception as e:
-        error_msg = f"Unexpected error calling Agent Generator: {e}"
-        logger.error(error_msg)
-        return _create_error_response(error_msg, "unexpected_error")
+        logger.error(f"Unexpected error calling external agent generator: {e}")
+        return None
 
 
 async def generate_agent_external(
-    instructions: dict[str, Any],
-    library_agents: list[dict[str, Any]] | None = None,
+    instructions: dict[str, Any]
 ) -> dict[str, Any] | None:
     """Call the external service to generate an agent from instructions.
 
     Args:
         instructions: Structured instructions from decompose_goal
-        library_agents: User's library agents available for sub-agent composition
 
     Returns:
-        Agent JSON dict on success, or error dict {"type": "error", ...} on error
+        Agent JSON dict or None on error
     """
     client = _get_client()
 
-    payload: dict[str, Any] = {"instructions": instructions}
-    if library_agents:
-        payload["library_agents"] = library_agents
-
     try:
-        response = await client.post("/api/generate-agent", json=payload)
+        response = await client.post(
+            "/api/generate-agent", json={"instructions": instructions}
+        )
         response.raise_for_status()
         data = response.json()
 
         if not data.get("success"):
-            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} (type: {error_type})"
-            )
-            return _create_error_response(error_msg, error_type)
+            logger.error(f"External service returned error: {data.get('error')}")
+            return None
 
         return data.get("agent_json")
 
     except httpx.HTTPStatusError as e:
-        error_type, error_msg = _classify_http_error(e)
-        logger.error(error_msg)
-        return _create_error_response(error_msg, error_type)
+        logger.error(f"HTTP error calling external agent generator: {e}")
+        return None
     except httpx.RequestError as e:
-        error_type, error_msg = _classify_request_error(e)
-        logger.error(error_msg)
-        return _create_error_response(error_msg, error_type)
+        logger.error(f"Request error calling external agent generator: {e}")
+        return None
     except Exception as e:
-        error_msg = f"Unexpected error calling Agent Generator: {e}"
-        logger.error(error_msg)
-        return _create_error_response(error_msg, "unexpected_error")
+        logger.error(f"Unexpected error calling external agent generator: {e}")
+        return None
 
 
 async def generate_agent_patch_external(
-    update_request: str,
-    current_agent: dict[str, Any],
-    library_agents: list[dict[str, Any]] | 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
-        library_agents: User's library agents available for sub-agent composition
 
     Returns:
-        Updated agent JSON, clarifying questions dict, or error dict on error
+        Updated agent JSON, clarifying questions dict, or None on error
     """
     client = _get_client()
 
-    payload: dict[str, Any] = {
-        "update_request": update_request,
-        "current_agent_json": current_agent,
-    }
-    if library_agents:
-        payload["library_agents"] = library_agents
-
     try:
-        response = await client.post("/api/update-agent", json=payload)
+        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"):
-            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)
+            logger.error(f"External service returned error: {data.get('error')}")
+            return None
 
         # Check if it's clarifying questions
         if data.get("type") == "clarifying_questions":
@@ -302,28 +196,18 @@ 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:
-        error_type, error_msg = _classify_http_error(e)
-        logger.error(error_msg)
-        return _create_error_response(error_msg, error_type)
+        logger.error(f"HTTP error calling external agent generator: {e}")
+        return None
     except httpx.RequestError as e:
-        error_type, error_msg = _classify_request_error(e)
-        logger.error(error_msg)
-        return _create_error_response(error_msg, error_type)
+        logger.error(f"Request error calling external agent generator: {e}")
+        return None
     except Exception as e:
-        error_msg = f"Unexpected error calling Agent Generator: {e}"
-        logger.error(error_msg)
-        return _create_error_response(error_msg, "unexpected_error")
+        logger.error(f"Unexpected error calling external agent generator: {e}")
+        return None
 
 
 async def get_blocks_external() -> list[dict[str, Any]] | None:

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

@@ -5,6 +5,7 @@ import re
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
+from langfuse import observe
 from pydantic import BaseModel, field_validator
 
 from backend.api.features.chat.model import ChatSession
@@ -328,6 +329,7 @@ class AgentOutputTool(BaseTool):
             total_executions=len(available_executions) if available_executions else 1,
         )
 
+    @observe(as_type="tool", name="view_agent_output")
     async def _execute(
         self,
         user_id: str | None,

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

@@ -1,7 +1,6 @@
 """Shared agent search functionality for find_agent and find_library_agent tools."""
 
 import logging
-import re
 from typing import Literal
 
 from backend.api.features.library import db as library_db
@@ -20,85 +19,6 @@ logger = logging.getLogger(__name__)
 
 SearchSource = Literal["marketplace", "library"]
 
-_UUID_PATTERN = re.compile(
-    r"^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$",
-    re.IGNORECASE,
-)
-
-
-def _is_uuid(text: str) -> bool:
-    """Check if text is a valid UUID v4."""
-    return bool(_UUID_PATTERN.match(text.strip()))
-
-
-async def _get_library_agent_by_id(user_id: str, agent_id: str) -> AgentInfo | None:
-    """Fetch a library agent by ID (library agent ID or graph_id).
-
-    Tries multiple lookup strategies:
-    1. First by graph_id (AgentGraph primary key)
-    2. Then by library agent ID (LibraryAgent primary key)
-
-    Args:
-        user_id: The user ID
-        agent_id: The ID to look up (can be graph_id or library agent ID)
-
-    Returns:
-        AgentInfo if found, None otherwise
-    """
-    try:
-        agent = await library_db.get_library_agent_by_graph_id(user_id, agent_id)
-        if agent:
-            logger.debug(f"Found library agent by graph_id: {agent.name}")
-            return AgentInfo(
-                id=agent.id,
-                name=agent.name,
-                description=agent.description or "",
-                source="library",
-                in_library=True,
-                creator=agent.creator_name,
-                status=agent.status.value,
-                can_access_graph=agent.can_access_graph,
-                has_external_trigger=agent.has_external_trigger,
-                new_output=agent.new_output,
-                graph_id=agent.graph_id,
-            )
-    except DatabaseError:
-        raise
-    except Exception as e:
-        logger.warning(
-            f"Could not fetch library agent by graph_id {agent_id}: {e}",
-            exc_info=True,
-        )
-
-    try:
-        agent = await library_db.get_library_agent(agent_id, user_id)
-        if agent:
-            logger.debug(f"Found library agent by library_id: {agent.name}")
-            return AgentInfo(
-                id=agent.id,
-                name=agent.name,
-                description=agent.description or "",
-                source="library",
-                in_library=True,
-                creator=agent.creator_name,
-                status=agent.status.value,
-                can_access_graph=agent.can_access_graph,
-                has_external_trigger=agent.has_external_trigger,
-                new_output=agent.new_output,
-                graph_id=agent.graph_id,
-            )
-    except NotFoundError:
-        logger.debug(f"Library agent not found by library_id: {agent_id}")
-    except DatabaseError:
-        raise
-    except Exception as e:
-        logger.warning(
-            f"Could not fetch library agent by library_id {agent_id}: {e}",
-            exc_info=True,
-        )
-
-    return None
-
 
 async def search_agents(
     query: str,
@@ -149,37 +69,29 @@ async def search_agents(
                         is_featured=False,
                     )
                 )
-        else:
-            if _is_uuid(query):
-                logger.info(f"Query looks like UUID, trying direct lookup: {query}")
-                agent = await _get_library_agent_by_id(user_id, query)  # type: ignore[arg-type]
-                if agent:
-                    agents.append(agent)
-                    logger.info(f"Found agent by direct ID lookup: {agent.name}")
-
-            if not agents:
-                logger.info(f"Searching user library for: {query}")
-                results = await library_db.list_library_agents(
-                    user_id=user_id,  # type: ignore[arg-type]
-                    search_term=query,
-                    page_size=10,
-                )
-                for agent in results.agents:
-                    agents.append(
-                        AgentInfo(
-                            id=agent.id,
-                            name=agent.name,
-                            description=agent.description or "",
-                            source="library",
-                            in_library=True,
-                            creator=agent.creator_name,
-                            status=agent.status.value,
-                            can_access_graph=agent.can_access_graph,
-                            has_external_trigger=agent.has_external_trigger,
-                            new_output=agent.new_output,
-                            graph_id=agent.graph_id,
-                        )
+        else:  # library
+            logger.info(f"Searching user library for: {query}")
+            results = await library_db.list_library_agents(
+                user_id=user_id,  # type: ignore[arg-type]
+                search_term=query,
+                page_size=10,
+            )
+            for agent in results.agents:
+                agents.append(
+                    AgentInfo(
+                        id=agent.id,
+                        name=agent.name,
+                        description=agent.description or "",
+                        source="library",
+                        in_library=True,
+                        creator=agent.creator_name,
+                        status=agent.status.value,
+                        can_access_graph=agent.can_access_graph,
+                        has_external_trigger=agent.has_external_trigger,
+                        new_output=agent.new_output,
+                        graph_id=agent.graph_id,
                     )
+                )
         logger.info(f"Found {len(agents)} agents in {source}")
     except NotFoundError:
         pass

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

@@ -36,16 +36,6 @@ class BaseTool:
         """Whether this tool requires authentication."""
         return False
 
-    @property
-    def is_long_running(self) -> bool:
-        """Whether this tool is long-running and should execute in background.
-
-        Long-running tools (like agent generation) are executed via background
-        tasks to survive SSE disconnections. The result is persisted to chat
-        history and visible when the user refreshes.
-        """
-        return False
-
     def as_openai_tool(self) -> ChatCompletionToolParam:
         """Convert to OpenAI tool format."""
         return ChatCompletionToolParam(

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

@@ -3,15 +3,14 @@
 import logging
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 
 from .agent_generator import (
     AgentGeneratorNotConfiguredError,
     decompose_goal,
-    enrich_library_agents_from_steps,
     generate_agent,
-    get_all_relevant_agents_for_generation,
-    get_user_message_for_error,
     save_agent_to_library,
 )
 from .base import BaseTool
@@ -45,10 +44,6 @@ class CreateAgentTool(BaseTool):
     def requires_auth(self) -> bool:
         return True
 
-    @property
-    def is_long_running(self) -> bool:
-        return True
-
     @property
     def parameters(self) -> dict[str, Any]:
         return {
@@ -80,6 +75,7 @@ class CreateAgentTool(BaseTool):
             "required": ["description"],
         }
 
+    @observe(as_type="tool", name="create_agent")
     async def _execute(
         self,
         user_id: str | None,
@@ -105,24 +101,9 @@ class CreateAgentTool(BaseTool):
                 session_id=session_id,
             )
 
-        library_agents = None
-        if user_id:
-            try:
-                library_agents = await get_all_relevant_agents_for_generation(
-                    user_id=user_id,
-                    search_query=description,
-                    include_marketplace=True,
-                )
-                logger.debug(
-                    f"Found {len(library_agents)} relevant agents for sub-agent composition"
-                )
-            except Exception as e:
-                logger.warning(f"Failed to fetch library agents: {e}")
-
+        # Step 1: Decompose goal into steps
         try:
-            decomposition_result = await decompose_goal(
-                description, context, library_agents
-            )
+            decomposition_result = await decompose_goal(description, context)
         except AgentGeneratorNotConfiguredError:
             return ErrorResponse(
                 message=(
@@ -135,31 +116,12 @@ class CreateAgentTool(BaseTool):
 
         if decomposition_result is None:
             return ErrorResponse(
-                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,
-            )
-
-        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],
-                    "service_error": error_msg,
-                    "error_type": error_type,
-                },
+                message="Failed to analyze the goal. Please try rephrasing.",
+                error="Decomposition failed",
                 session_id=session_id,
             )
 
+        # Check if LLM returned clarifying questions
         if decomposition_result.get("type") == "clarifying_questions":
             questions = decomposition_result.get("questions", [])
             return ClarificationNeededResponse(
@@ -178,6 +140,7 @@ class CreateAgentTool(BaseTool):
                 session_id=session_id,
             )
 
+        # Check for unachievable/vague goals
         if decomposition_result.get("type") == "unachievable_goal":
             suggested = decomposition_result.get("suggested_goal", "")
             reason = decomposition_result.get("reason", "")
@@ -204,22 +167,9 @@ class CreateAgentTool(BaseTool):
                 session_id=session_id,
             )
 
-        if user_id and library_agents is not None:
-            try:
-                library_agents = await enrich_library_agents_from_steps(
-                    user_id=user_id,
-                    decomposition_result=decomposition_result,
-                    existing_agents=library_agents,
-                    include_marketplace=True,
-                )
-                logger.debug(
-                    f"After enrichment: {len(library_agents)} total agents for sub-agent composition"
-                )
-            except Exception as e:
-                logger.warning(f"Failed to enrich library agents from steps: {e}")
-
+        # Step 2: Generate agent JSON (external service handles fixing and validation)
         try:
-            agent_json = await generate_agent(decomposition_result, library_agents)
+            agent_json = await generate_agent(decomposition_result)
         except AgentGeneratorNotConfiguredError:
             return ErrorResponse(
                 message=(
@@ -232,34 +182,8 @@ class CreateAgentTool(BaseTool):
 
         if agent_json is None:
             return ErrorResponse(
-                message="Failed to generate the agent. The agent generation service may be unavailable. Please try again.",
-                error="generation_failed",
-                details={"description": description[:100]},
-                session_id=session_id,
-            )
-
-        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=(
-                    "I wasn't able to create a valid agent for this request. "
-                    "The generated workflow had some structural issues. "
-                    "Please try simplifying your goal or breaking it into smaller steps."
-                ),
-                error_details=error_msg,
-            )
-            return ErrorResponse(
-                message=user_message,
-                error=f"generation_failed:{error_type}",
-                details={
-                    "description": description[:100],
-                    "service_error": error_msg,
-                    "error_type": error_type,
-                },
+                message="Failed to generate the agent. Please try again.",
+                error="Generation failed",
                 session_id=session_id,
             )
 
@@ -268,6 +192,7 @@ class CreateAgentTool(BaseTool):
         node_count = len(agent_json.get("nodes", []))
         link_count = len(agent_json.get("links", []))
 
+        # Step 3: Preview or save
         if not save:
             return AgentPreviewResponse(
                 message=(
@@ -282,6 +207,7 @@ class CreateAgentTool(BaseTool):
                 session_id=session_id,
             )
 
+        # Save to library
         if not user_id:
             return ErrorResponse(
                 message="You must be logged in to save agents.",
@@ -299,7 +225,7 @@ class CreateAgentTool(BaseTool):
                 agent_id=created_graph.id,
                 agent_name=created_graph.name,
                 library_agent_id=library_agent.id,
-                library_agent_link=f"/library/agents/{library_agent.id}",
+                library_agent_link=f"/library/{library_agent.id}",
                 agent_page_link=f"/build?flowID={created_graph.id}",
                 session_id=session_id,
             )

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

@@ -3,14 +3,14 @@
 import logging
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 
 from .agent_generator import (
     AgentGeneratorNotConfiguredError,
     generate_agent_patch,
     get_agent_as_json,
-    get_all_relevant_agents_for_generation,
-    get_user_message_for_error,
     save_agent_to_library,
 )
 from .base import BaseTool
@@ -44,10 +44,6 @@ class EditAgentTool(BaseTool):
     def requires_auth(self) -> bool:
         return True
 
-    @property
-    def is_long_running(self) -> bool:
-        return True
-
     @property
     def parameters(self) -> dict[str, Any]:
         return {
@@ -85,6 +81,7 @@ class EditAgentTool(BaseTool):
             "required": ["agent_id", "changes"],
         }
 
+    @observe(as_type="tool", name="edit_agent")
     async def _execute(
         self,
         user_id: str | None,
@@ -118,6 +115,7 @@ class EditAgentTool(BaseTool):
                 session_id=session_id,
             )
 
+        # Step 1: Fetch current agent
         current_agent = await get_agent_as_json(agent_id, user_id)
 
         if current_agent is None:
@@ -127,30 +125,14 @@ class EditAgentTool(BaseTool):
                 session_id=session_id,
             )
 
-        library_agents = None
-        if user_id:
-            try:
-                graph_id = current_agent.get("id")
-                library_agents = await get_all_relevant_agents_for_generation(
-                    user_id=user_id,
-                    search_query=changes,
-                    exclude_graph_id=graph_id,
-                    include_marketplace=True,
-                )
-                logger.debug(
-                    f"Found {len(library_agents)} relevant agents for sub-agent composition"
-                )
-            except Exception as e:
-                logger.warning(f"Failed to fetch library agents: {e}")
-
+        # Build the update request with context
         update_request = changes
         if context:
             update_request = f"{changes}\n\nAdditional context:\n{context}"
 
+        # Step 2: Generate updated agent (external service handles fixing and validation)
         try:
-            result = await generate_agent_patch(
-                update_request, current_agent, library_agents
-            )
+            result = await generate_agent_patch(update_request, current_agent)
         except AgentGeneratorNotConfiguredError:
             return ErrorResponse(
                 message=(
@@ -163,34 +145,12 @@ class EditAgentTool(BaseTool):
 
         if result is None:
             return ErrorResponse(
-                message="Failed to generate changes. The agent generation service may be unavailable or timed out. Please try again.",
-                error="update_generation_failed",
-                details={"agent_id": agent_id, "changes": changes[:100]},
-                session_id=session_id,
-            )
-
-        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.",
-                error_details=error_msg,
-            )
-            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,
-                },
+                message="Failed to generate changes. Please try rephrasing.",
+                error="Update generation failed",
                 session_id=session_id,
             )
 
+        # Check if LLM returned clarifying questions
         if result.get("type") == "clarifying_questions":
             questions = result.get("questions", [])
             return ClarificationNeededResponse(
@@ -209,6 +169,7 @@ class EditAgentTool(BaseTool):
                 session_id=session_id,
             )
 
+        # Result is the updated agent JSON
         updated_agent = result
 
         agent_name = updated_agent.get("name", "Updated Agent")
@@ -216,6 +177,7 @@ class EditAgentTool(BaseTool):
         node_count = len(updated_agent.get("nodes", []))
         link_count = len(updated_agent.get("links", []))
 
+        # Step 3: Preview or save
         if not save:
             return AgentPreviewResponse(
                 message=(
@@ -231,6 +193,7 @@ class EditAgentTool(BaseTool):
                 session_id=session_id,
             )
 
+        # Save to library (creates a new version)
         if not user_id:
             return ErrorResponse(
                 message="You must be logged in to save agents.",
@@ -248,7 +211,7 @@ class EditAgentTool(BaseTool):
                 agent_id=created_graph.id,
                 agent_name=created_graph.name,
                 library_agent_id=library_agent.id,
-                library_agent_link=f"/library/agents/{library_agent.id}",
+                library_agent_link=f"/library/{library_agent.id}",
                 agent_page_link=f"/build?flowID={created_graph.id}",
                 session_id=session_id,
             )

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

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 
 from .agent_search import search_agents
@@ -35,6 +37,7 @@ class FindAgentTool(BaseTool):
             "required": ["query"],
         }
 
+    @observe(as_type="tool", name="find_agent")
     async def _execute(
         self, user_id: str | None, session: ChatSession, **kwargs
     ) -> ToolResponseBase:

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

@@ -1,6 +1,7 @@
 import logging
 from typing import Any
 
+from langfuse import observe
 from prisma.enums import ContentType
 
 from backend.api.features.chat.model import ChatSession
@@ -55,6 +56,7 @@ class FindBlockTool(BaseTool):
     def requires_auth(self) -> bool:
         return True
 
+    @observe(as_type="tool", name="find_block")
     async def _execute(
         self,
         user_id: str | None,
@@ -107,8 +109,7 @@ class FindBlockTool(BaseTool):
                 block_id = result["content_id"]
                 block = get_block(block_id)
 
-                # Skip disabled blocks
-                if block and not block.disabled:
+                if block:
                     # Get input/output schemas
                     input_schema = {}
                     output_schema = {}

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

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 
 from .agent_search import search_agents
@@ -41,6 +43,7 @@ class FindLibraryAgentTool(BaseTool):
     def requires_auth(self) -> bool:
         return True
 
+    @observe(as_type="tool", name="find_library_agent")
     async def _execute(
         self, user_id: str | None, session: ChatSession, **kwargs
     ) -> ToolResponseBase:

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

@@ -4,6 +4,8 @@ import logging
 from pathlib import Path
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 from backend.api.features.chat.tools.base import BaseTool
 from backend.api.features.chat.tools.models import (
@@ -71,6 +73,7 @@ class GetDocPageTool(BaseTool):
         url_path = path.rsplit(".", 1)[0] if "." in path else path
         return f"{DOCS_BASE_URL}/{url_path}"
 
+    @observe(as_type="tool", name="get_doc_page")
     async def _execute(
         self,
         user_id: str | None,

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

@@ -28,16 +28,6 @@ class ResponseType(str, Enum):
     BLOCK_OUTPUT = "block_output"
     DOC_SEARCH_RESULTS = "doc_search_results"
     DOC_PAGE = "doc_page"
-    # Workspace response types
-    WORKSPACE_FILE_LIST = "workspace_file_list"
-    WORKSPACE_FILE_CONTENT = "workspace_file_content"
-    WORKSPACE_FILE_METADATA = "workspace_file_metadata"
-    WORKSPACE_FILE_WRITTEN = "workspace_file_written"
-    WORKSPACE_FILE_DELETED = "workspace_file_deleted"
-    # Long-running operation types
-    OPERATION_STARTED = "operation_started"
-    OPERATION_PENDING = "operation_pending"
-    OPERATION_IN_PROGRESS = "operation_in_progress"
 
 
 # Base response model
@@ -344,39 +334,3 @@ class BlockOutputResponse(ToolResponseBase):
     block_name: str
     outputs: dict[str, list[Any]]
     success: bool = True
-
-
-# Long-running operation models
-class OperationStartedResponse(ToolResponseBase):
-    """Response when a long-running operation has been started in the background.
-
-    This is returned immediately to the client while the operation continues
-    to execute. The user can close the tab and check back later.
-    """
-
-    type: ResponseType = ResponseType.OPERATION_STARTED
-    operation_id: str
-    tool_name: str
-
-
-class OperationPendingResponse(ToolResponseBase):
-    """Response stored in chat history while a long-running operation is executing.
-
-    This is persisted to the database so users see a pending state when they
-    refresh before the operation completes.
-    """
-
-    type: ResponseType = ResponseType.OPERATION_PENDING
-    operation_id: str
-    tool_name: str
-
-
-class OperationInProgressResponse(ToolResponseBase):
-    """Response when an operation is already in progress.
-
-    Returned for idempotency when the same tool_call_id is requested again
-    while the background task is still running.
-    """
-
-    type: ResponseType = ResponseType.OPERATION_IN_PROGRESS
-    tool_call_id: str

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

@@ -3,14 +3,11 @@
 import logging
 from typing import Any
 
+from langfuse import observe
 from pydantic import BaseModel, Field, field_validator
 
 from backend.api.features.chat.config import ChatConfig
 from backend.api.features.chat.model import ChatSession
-from backend.api.features.chat.tracking import (
-    track_agent_run_success,
-    track_agent_scheduled,
-)
 from backend.api.features.library import db as library_db
 from backend.data.graph import GraphModel
 from backend.data.model import CredentialsMetaInput
@@ -158,6 +155,7 @@ class RunAgentTool(BaseTool):
         """All operations require authentication."""
         return True
 
+    @observe(as_type="tool", name="run_agent")
     async def _execute(
         self,
         user_id: str | None,
@@ -455,16 +453,6 @@ class RunAgentTool(BaseTool):
             session.successful_agent_runs.get(library_agent.graph_id, 0) + 1
         )
 
-        # Track in PostHog
-        track_agent_run_success(
-            user_id=user_id,
-            session_id=session_id,
-            graph_id=library_agent.graph_id,
-            graph_name=library_agent.name,
-            execution_id=execution.id,
-            library_agent_id=library_agent.id,
-        )
-
         library_agent_link = f"/library/agents/{library_agent.id}"
         return ExecutionStartedResponse(
             message=(
@@ -546,18 +534,6 @@ class RunAgentTool(BaseTool):
             session.successful_agent_schedules.get(library_agent.graph_id, 0) + 1
         )
 
-        # Track in PostHog
-        track_agent_scheduled(
-            user_id=user_id,
-            session_id=session_id,
-            graph_id=library_agent.graph_id,
-            graph_name=library_agent.name,
-            schedule_id=result.id,
-            schedule_name=schedule_name,
-            cron=cron,
-            library_agent_id=library_agent.id,
-        )
-
         library_agent_link = f"/library/agents/{library_agent.id}"
         return ExecutionStartedResponse(
             message=(

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

@@ -1,15 +1,15 @@
 """Tool for executing blocks directly."""
 
 import logging
-import uuid
 from collections import defaultdict
 from typing import Any
 
+from langfuse import observe
+
 from backend.api.features.chat.model import ChatSession
 from backend.data.block import get_block
 from backend.data.execution import ExecutionContext
 from backend.data.model import CredentialsMetaInput
-from backend.data.workspace import get_or_create_workspace
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.util.exceptions import BlockError
 
@@ -130,6 +130,7 @@ class RunBlockTool(BaseTool):
 
         return matched_credentials, missing_credentials
 
+    @observe(as_type="tool", name="run_block")
     async def _execute(
         self,
         user_id: str | None,
@@ -178,11 +179,6 @@ class RunBlockTool(BaseTool):
                 message=f"Block '{block_id}' not found",
                 session_id=session_id,
             )
-        if block.disabled:
-            return ErrorResponse(
-                message=f"Block '{block_id}' is disabled",
-                session_id=session_id,
-            )
 
         logger.info(f"Executing block {block.name} ({block_id}) for user {user_id}")
 
@@ -225,48 +221,11 @@ class RunBlockTool(BaseTool):
             )
 
         try:
-            # Get or create user's workspace for CoPilot file operations
-            workspace = await get_or_create_workspace(user_id)
-
-            # Generate synthetic IDs for CoPilot context
-            # Each chat session is treated as its own agent with one continuous run
-            # This means:
-            # - graph_id (agent) = session (memories scoped to session when limit_to_agent=True)
-            # - graph_exec_id (run) = session (memories scoped to session when limit_to_run=True)
-            # - node_exec_id = unique per block execution
-            synthetic_graph_id = f"copilot-session-{session.session_id}"
-            synthetic_graph_exec_id = f"copilot-session-{session.session_id}"
-            synthetic_node_id = f"copilot-node-{block_id}"
-            synthetic_node_exec_id = (
-                f"copilot-{session.session_id}-{uuid.uuid4().hex[:8]}"
-            )
-
-            # Create unified execution context with all required fields
-            execution_context = ExecutionContext(
-                # Execution identity
-                user_id=user_id,
-                graph_id=synthetic_graph_id,
-                graph_exec_id=synthetic_graph_exec_id,
-                graph_version=1,  # Versions are 1-indexed
-                node_id=synthetic_node_id,
-                node_exec_id=synthetic_node_exec_id,
-                # Workspace with session scoping
-                workspace_id=workspace.id,
-                session_id=session.session_id,
-            )
-
-            # Prepare kwargs for block execution
-            # Keep individual kwargs for backwards compatibility with existing blocks
+            # Fetch actual credentials and prepare kwargs for block execution
+            # Create execution context with defaults (blocks may require it)
             exec_kwargs: dict[str, Any] = {
                 "user_id": user_id,
-                "execution_context": execution_context,
-                # Legacy: individual kwargs for blocks not yet using execution_context
-                "workspace_id": workspace.id,
-                "graph_exec_id": synthetic_graph_exec_id,
-                "node_exec_id": synthetic_node_exec_id,
-                "node_id": synthetic_node_id,
-                "graph_version": 1,  # Versions are 1-indexed
-                "graph_id": synthetic_graph_id,
+                "execution_context": ExecutionContext(),
             }
 
             for field_name, cred_meta in matched_credentials.items():

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

@@ -3,6 +3,7 @@
 import logging
 from typing import Any
 
+from langfuse import observe
 from prisma.enums import ContentType
 
 from backend.api.features.chat.model import ChatSession
@@ -87,6 +88,7 @@ class SearchDocsTool(BaseTool):
         url_path = path.rsplit(".", 1)[0] if "." in path else path
         return f"{DOCS_BASE_URL}/{url_path}"
 
+    @observe(as_type="tool", name="search_docs")
     async def _execute(
         self,
         user_id: str | None,

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

@@ -8,7 +8,7 @@ from backend.api.features.library import model as library_model
 from backend.api.features.store import db as store_db
 from backend.data import graph as graph_db
 from backend.data.graph import GraphModel
-from backend.data.model import Credentials, CredentialsFieldInfo, CredentialsMetaInput
+from backend.data.model import CredentialsFieldInfo, CredentialsMetaInput
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.util.exceptions import NotFoundError
 
@@ -266,14 +266,13 @@ async def match_user_credentials_to_graph(
         credential_requirements,
         _node_fields,
     ) in aggregated_creds.items():
-        # Find first matching credential by provider, type, and scopes
+        # Find first matching credential by provider and type
         matching_cred = next(
             (
                 cred
                 for cred in available_creds
                 if cred.provider in credential_requirements.provider
                 and cred.type in credential_requirements.supported_types
-                and _credential_has_required_scopes(cred, credential_requirements)
             ),
             None,
         )
@@ -297,17 +296,10 @@ async def match_user_credentials_to_graph(
                     f"{credential_field_name} (validation failed: {e})"
                 )
         else:
-            # Build a helpful error message including scope requirements
-            error_parts = [
-                f"provider in {list(credential_requirements.provider)}",
-                f"type in {list(credential_requirements.supported_types)}",
-            ]
-            if credential_requirements.required_scopes:
-                error_parts.append(
-                    f"scopes including {list(credential_requirements.required_scopes)}"
-                )
             missing_creds.append(
-                f"{credential_field_name} (requires {', '.join(error_parts)})"
+                f"{credential_field_name} "
+                f"(requires provider in {list(credential_requirements.provider)}, "
+                f"type in {list(credential_requirements.supported_types)})"
             )
 
     logger.info(
@@ -317,28 +309,6 @@ async def match_user_credentials_to_graph(
     return graph_credentials_inputs, missing_creds
 
 
-def _credential_has_required_scopes(
-    credential: Credentials,
-    requirements: CredentialsFieldInfo,
-) -> bool:
-    """
-    Check if a credential has all the scopes required by the block.
-
-    For OAuth2 credentials, verifies that the credential's scopes are a superset
-    of the required scopes. For other credential types, returns True (no scope check).
-    """
-    # Only OAuth2 credentials have scopes to check
-    if credential.type != "oauth2":
-        return True
-
-    # If no scopes are required, any credential matches
-    if not requirements.required_scopes:
-        return True
-
-    # Check that credential scopes are a superset of required scopes
-    return set(credential.scopes).issuperset(requirements.required_scopes)
-
-
 async def check_user_has_required_credentials(
     user_id: str,
     required_credentials: list[CredentialsMetaInput],

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

@@ -1,620 +0,0 @@
-"""CoPilot tools for workspace file operations."""
-
-import base64
-import logging
-from typing import Any, Optional
-
-from pydantic import BaseModel
-
-from backend.api.features.chat.model import ChatSession
-from backend.data.workspace import get_or_create_workspace
-from backend.util.settings import Config
-from backend.util.virus_scanner import scan_content_safe
-from backend.util.workspace import WorkspaceManager
-
-from .base import BaseTool
-from .models import ErrorResponse, ResponseType, ToolResponseBase
-
-logger = logging.getLogger(__name__)
-
-
-class WorkspaceFileInfoData(BaseModel):
-    """Data model for workspace file information (not a response itself)."""
-
-    file_id: str
-    name: str
-    path: str
-    mime_type: str
-    size_bytes: int
-
-
-class WorkspaceFileListResponse(ToolResponseBase):
-    """Response containing list of workspace files."""
-
-    type: ResponseType = ResponseType.WORKSPACE_FILE_LIST
-    files: list[WorkspaceFileInfoData]
-    total_count: int
-
-
-class WorkspaceFileContentResponse(ToolResponseBase):
-    """Response containing workspace file content (legacy, for small text files)."""
-
-    type: ResponseType = ResponseType.WORKSPACE_FILE_CONTENT
-    file_id: str
-    name: str
-    path: str
-    mime_type: str
-    content_base64: str
-
-
-class WorkspaceFileMetadataResponse(ToolResponseBase):
-    """Response containing workspace file metadata and download URL (prevents context bloat)."""
-
-    type: ResponseType = ResponseType.WORKSPACE_FILE_METADATA
-    file_id: str
-    name: str
-    path: str
-    mime_type: str
-    size_bytes: int
-    download_url: str
-    preview: str | None = None  # First 500 chars for text files
-
-
-class WorkspaceWriteResponse(ToolResponseBase):
-    """Response after writing a file to workspace."""
-
-    type: ResponseType = ResponseType.WORKSPACE_FILE_WRITTEN
-    file_id: str
-    name: str
-    path: str
-    size_bytes: int
-
-
-class WorkspaceDeleteResponse(ToolResponseBase):
-    """Response after deleting a file from workspace."""
-
-    type: ResponseType = ResponseType.WORKSPACE_FILE_DELETED
-    file_id: str
-    success: bool
-
-
-class ListWorkspaceFilesTool(BaseTool):
-    """Tool for listing files in user's workspace."""
-
-    @property
-    def name(self) -> str:
-        return "list_workspace_files"
-
-    @property
-    def description(self) -> str:
-        return (
-            "List files in the user's workspace. "
-            "Returns file names, paths, sizes, and metadata. "
-            "Optionally filter by path prefix."
-        )
-
-    @property
-    def parameters(self) -> dict[str, Any]:
-        return {
-            "type": "object",
-            "properties": {
-                "path_prefix": {
-                    "type": "string",
-                    "description": (
-                        "Optional path prefix to filter files "
-                        "(e.g., '/documents/' to list only files in documents folder). "
-                        "By default, only files from the current session are listed."
-                    ),
-                },
-                "limit": {
-                    "type": "integer",
-                    "description": "Maximum number of files to return (default 50, max 100)",
-                    "minimum": 1,
-                    "maximum": 100,
-                },
-                "include_all_sessions": {
-                    "type": "boolean",
-                    "description": (
-                        "If true, list files from all sessions. "
-                        "Default is false (only current session's files)."
-                    ),
-                },
-            },
-            "required": [],
-        }
-
-    @property
-    def requires_auth(self) -> bool:
-        return True
-
-    async def _execute(
-        self,
-        user_id: str | None,
-        session: ChatSession,
-        **kwargs,
-    ) -> ToolResponseBase:
-        session_id = session.session_id
-
-        if not user_id:
-            return ErrorResponse(
-                message="Authentication required",
-                session_id=session_id,
-            )
-
-        path_prefix: Optional[str] = kwargs.get("path_prefix")
-        limit = min(kwargs.get("limit", 50), 100)
-        include_all_sessions: bool = kwargs.get("include_all_sessions", False)
-
-        try:
-            workspace = await get_or_create_workspace(user_id)
-            # Pass session_id for session-scoped file access
-            manager = WorkspaceManager(user_id, workspace.id, session_id)
-
-            files = await manager.list_files(
-                path=path_prefix,
-                limit=limit,
-                include_all_sessions=include_all_sessions,
-            )
-            total = await manager.get_file_count(
-                path=path_prefix,
-                include_all_sessions=include_all_sessions,
-            )
-
-            file_infos = [
-                WorkspaceFileInfoData(
-                    file_id=f.id,
-                    name=f.name,
-                    path=f.path,
-                    mime_type=f.mimeType,
-                    size_bytes=f.sizeBytes,
-                )
-                for f in files
-            ]
-
-            scope_msg = "all sessions" if include_all_sessions else "current session"
-            return WorkspaceFileListResponse(
-                files=file_infos,
-                total_count=total,
-                message=f"Found {len(files)} files in workspace ({scope_msg})",
-                session_id=session_id,
-            )
-
-        except Exception as e:
-            logger.error(f"Error listing workspace files: {e}", exc_info=True)
-            return ErrorResponse(
-                message=f"Failed to list workspace files: {str(e)}",
-                error=str(e),
-                session_id=session_id,
-            )
-
-
-class ReadWorkspaceFileTool(BaseTool):
-    """Tool for reading file content from workspace."""
-
-    # Size threshold for returning full content vs metadata+URL
-    # Files larger than this return metadata with download URL to prevent context bloat
-    MAX_INLINE_SIZE_BYTES = 32 * 1024  # 32KB
-    # Preview size for text files
-    PREVIEW_SIZE = 500
-
-    @property
-    def name(self) -> str:
-        return "read_workspace_file"
-
-    @property
-    def description(self) -> str:
-        return (
-            "Read a file from the user's workspace. "
-            "Specify either file_id or path to identify the file. "
-            "For small text files, returns content directly. "
-            "For large or binary files, returns metadata and a download URL. "
-            "Paths are scoped to the current session by default. "
-            "Use /sessions/<session_id>/... for cross-session access."
-        )
-
-    @property
-    def parameters(self) -> dict[str, Any]:
-        return {
-            "type": "object",
-            "properties": {
-                "file_id": {
-                    "type": "string",
-                    "description": "The file's unique ID (from list_workspace_files)",
-                },
-                "path": {
-                    "type": "string",
-                    "description": (
-                        "The virtual file path (e.g., '/documents/report.pdf'). "
-                        "Scoped to current session by default."
-                    ),
-                },
-                "force_download_url": {
-                    "type": "boolean",
-                    "description": (
-                        "If true, always return metadata+URL instead of inline content. "
-                        "Default is false (auto-selects based on file size/type)."
-                    ),
-                },
-            },
-            "required": [],  # At least one must be provided
-        }
-
-    @property
-    def requires_auth(self) -> bool:
-        return True
-
-    def _is_text_mime_type(self, mime_type: str) -> bool:
-        """Check if the MIME type is a text-based type."""
-        text_types = [
-            "text/",
-            "application/json",
-            "application/xml",
-            "application/javascript",
-            "application/x-python",
-            "application/x-sh",
-        ]
-        return any(mime_type.startswith(t) for t in text_types)
-
-    async def _execute(
-        self,
-        user_id: str | None,
-        session: ChatSession,
-        **kwargs,
-    ) -> ToolResponseBase:
-        session_id = session.session_id
-
-        if not user_id:
-            return ErrorResponse(
-                message="Authentication required",
-                session_id=session_id,
-            )
-
-        file_id: Optional[str] = kwargs.get("file_id")
-        path: Optional[str] = kwargs.get("path")
-        force_download_url: bool = kwargs.get("force_download_url", False)
-
-        if not file_id and not path:
-            return ErrorResponse(
-                message="Please provide either file_id or path",
-                session_id=session_id,
-            )
-
-        try:
-            workspace = await get_or_create_workspace(user_id)
-            # Pass session_id for session-scoped file access
-            manager = WorkspaceManager(user_id, workspace.id, session_id)
-
-            # Get file info
-            if file_id:
-                file_info = await manager.get_file_info(file_id)
-                if file_info is None:
-                    return ErrorResponse(
-                        message=f"File not found: {file_id}",
-                        session_id=session_id,
-                    )
-                target_file_id = file_id
-            else:
-                # path is guaranteed to be non-None here due to the check above
-                assert path is not None
-                file_info = await manager.get_file_info_by_path(path)
-                if file_info is None:
-                    return ErrorResponse(
-                        message=f"File not found at path: {path}",
-                        session_id=session_id,
-                    )
-                target_file_id = file_info.id
-
-            # Decide whether to return inline content or metadata+URL
-            is_small_file = file_info.sizeBytes <= self.MAX_INLINE_SIZE_BYTES
-            is_text_file = self._is_text_mime_type(file_info.mimeType)
-
-            # Return inline content for small text files (unless force_download_url)
-            if is_small_file and is_text_file and not force_download_url:
-                content = await manager.read_file_by_id(target_file_id)
-                content_b64 = base64.b64encode(content).decode("utf-8")
-
-                return WorkspaceFileContentResponse(
-                    file_id=file_info.id,
-                    name=file_info.name,
-                    path=file_info.path,
-                    mime_type=file_info.mimeType,
-                    content_base64=content_b64,
-                    message=f"Successfully read file: {file_info.name}",
-                    session_id=session_id,
-                )
-
-            # Return metadata + workspace:// reference for large or binary files
-            # This prevents context bloat (100KB file = ~133KB as base64)
-            # Use workspace:// format so frontend urlTransform can add proxy prefix
-            download_url = f"workspace://{target_file_id}"
-
-            # Generate preview for text files
-            preview: str | None = None
-            if is_text_file:
-                try:
-                    content = await manager.read_file_by_id(target_file_id)
-                    preview_text = content[: self.PREVIEW_SIZE].decode(
-                        "utf-8", errors="replace"
-                    )
-                    if len(content) > self.PREVIEW_SIZE:
-                        preview_text += "..."
-                    preview = preview_text
-                except Exception:
-                    pass  # Preview is optional
-
-            return WorkspaceFileMetadataResponse(
-                file_id=file_info.id,
-                name=file_info.name,
-                path=file_info.path,
-                mime_type=file_info.mimeType,
-                size_bytes=file_info.sizeBytes,
-                download_url=download_url,
-                preview=preview,
-                message=f"File: {file_info.name} ({file_info.sizeBytes} bytes). Use download_url to retrieve content.",
-                session_id=session_id,
-            )
-
-        except FileNotFoundError as e:
-            return ErrorResponse(
-                message=str(e),
-                session_id=session_id,
-            )
-        except Exception as e:
-            logger.error(f"Error reading workspace file: {e}", exc_info=True)
-            return ErrorResponse(
-                message=f"Failed to read workspace file: {str(e)}",
-                error=str(e),
-                session_id=session_id,
-            )
-
-
-class WriteWorkspaceFileTool(BaseTool):
-    """Tool for writing files to workspace."""
-
-    @property
-    def name(self) -> str:
-        return "write_workspace_file"
-
-    @property
-    def description(self) -> str:
-        return (
-            "Write or create a file in the user's workspace. "
-            "Provide the content as a base64-encoded string. "
-            f"Maximum file size is {Config().max_file_size_mb}MB. "
-            "Files are saved to the current session's folder by default. "
-            "Use /sessions/<session_id>/... for cross-session access."
-        )
-
-    @property
-    def parameters(self) -> dict[str, Any]:
-        return {
-            "type": "object",
-            "properties": {
-                "filename": {
-                    "type": "string",
-                    "description": "Name for the file (e.g., 'report.pdf')",
-                },
-                "content_base64": {
-                    "type": "string",
-                    "description": "Base64-encoded file content",
-                },
-                "path": {
-                    "type": "string",
-                    "description": (
-                        "Optional virtual path where to save the file "
-                        "(e.g., '/documents/report.pdf'). "
-                        "Defaults to '/{filename}'. Scoped to current session."
-                    ),
-                },
-                "mime_type": {
-                    "type": "string",
-                    "description": (
-                        "Optional MIME type of the file. "
-                        "Auto-detected from filename if not provided."
-                    ),
-                },
-                "overwrite": {
-                    "type": "boolean",
-                    "description": "Whether to overwrite if file exists at path (default: false)",
-                },
-            },
-            "required": ["filename", "content_base64"],
-        }
-
-    @property
-    def requires_auth(self) -> bool:
-        return True
-
-    async def _execute(
-        self,
-        user_id: str | None,
-        session: ChatSession,
-        **kwargs,
-    ) -> ToolResponseBase:
-        session_id = session.session_id
-
-        if not user_id:
-            return ErrorResponse(
-                message="Authentication required",
-                session_id=session_id,
-            )
-
-        filename: str = kwargs.get("filename", "")
-        content_b64: str = kwargs.get("content_base64", "")
-        path: Optional[str] = kwargs.get("path")
-        mime_type: Optional[str] = kwargs.get("mime_type")
-        overwrite: bool = kwargs.get("overwrite", False)
-
-        if not filename:
-            return ErrorResponse(
-                message="Please provide a filename",
-                session_id=session_id,
-            )
-
-        if not content_b64:
-            return ErrorResponse(
-                message="Please provide content_base64",
-                session_id=session_id,
-            )
-
-        # Decode content
-        try:
-            content = base64.b64decode(content_b64)
-        except Exception:
-            return ErrorResponse(
-                message="Invalid base64-encoded content",
-                session_id=session_id,
-            )
-
-        # Check size
-        max_file_size = Config().max_file_size_mb * 1024 * 1024
-        if len(content) > max_file_size:
-            return ErrorResponse(
-                message=f"File too large. Maximum size is {Config().max_file_size_mb}MB",
-                session_id=session_id,
-            )
-
-        try:
-            # Virus scan
-            await scan_content_safe(content, filename=filename)
-
-            workspace = await get_or_create_workspace(user_id)
-            # Pass session_id for session-scoped file access
-            manager = WorkspaceManager(user_id, workspace.id, session_id)
-
-            file_record = await manager.write_file(
-                content=content,
-                filename=filename,
-                path=path,
-                mime_type=mime_type,
-                overwrite=overwrite,
-            )
-
-            return WorkspaceWriteResponse(
-                file_id=file_record.id,
-                name=file_record.name,
-                path=file_record.path,
-                size_bytes=file_record.sizeBytes,
-                message=f"Successfully wrote file: {file_record.name}",
-                session_id=session_id,
-            )
-
-        except ValueError as e:
-            return ErrorResponse(
-                message=str(e),
-                session_id=session_id,
-            )
-        except Exception as e:
-            logger.error(f"Error writing workspace file: {e}", exc_info=True)
-            return ErrorResponse(
-                message=f"Failed to write workspace file: {str(e)}",
-                error=str(e),
-                session_id=session_id,
-            )
-
-
-class DeleteWorkspaceFileTool(BaseTool):
-    """Tool for deleting files from workspace."""
-
-    @property
-    def name(self) -> str:
-        return "delete_workspace_file"
-
-    @property
-    def description(self) -> str:
-        return (
-            "Delete a file from the user's workspace. "
-            "Specify either file_id or path to identify the file. "
-            "Paths are scoped to the current session by default. "
-            "Use /sessions/<session_id>/... for cross-session access."
-        )
-
-    @property
-    def parameters(self) -> dict[str, Any]:
-        return {
-            "type": "object",
-            "properties": {
-                "file_id": {
-                    "type": "string",
-                    "description": "The file's unique ID (from list_workspace_files)",
-                },
-                "path": {
-                    "type": "string",
-                    "description": (
-                        "The virtual file path (e.g., '/documents/report.pdf'). "
-                        "Scoped to current session by default."
-                    ),
-                },
-            },
-            "required": [],  # At least one must be provided
-        }
-
-    @property
-    def requires_auth(self) -> bool:
-        return True
-
-    async def _execute(
-        self,
-        user_id: str | None,
-        session: ChatSession,
-        **kwargs,
-    ) -> ToolResponseBase:
-        session_id = session.session_id
-
-        if not user_id:
-            return ErrorResponse(
-                message="Authentication required",
-                session_id=session_id,
-            )
-
-        file_id: Optional[str] = kwargs.get("file_id")
-        path: Optional[str] = kwargs.get("path")
-
-        if not file_id and not path:
-            return ErrorResponse(
-                message="Please provide either file_id or path",
-                session_id=session_id,
-            )
-
-        try:
-            workspace = await get_or_create_workspace(user_id)
-            # Pass session_id for session-scoped file access
-            manager = WorkspaceManager(user_id, workspace.id, session_id)
-
-            # Determine the file_id to delete
-            target_file_id: str
-            if file_id:
-                target_file_id = file_id
-            else:
-                # path is guaranteed to be non-None here due to the check above
-                assert path is not None
-                file_info = await manager.get_file_info_by_path(path)
-                if file_info is None:
-                    return ErrorResponse(
-                        message=f"File not found at path: {path}",
-                        session_id=session_id,
-                    )
-                target_file_id = file_info.id
-
-            success = await manager.delete_file(target_file_id)
-
-            if not success:
-                return ErrorResponse(
-                    message=f"File not found: {target_file_id}",
-                    session_id=session_id,
-                )
-
-            return WorkspaceDeleteResponse(
-                file_id=target_file_id,
-                success=True,
-                message="File deleted successfully",
-                session_id=session_id,
-            )
-
-        except Exception as e:
-            logger.error(f"Error deleting workspace file: {e}", exc_info=True)
-            return ErrorResponse(
-                message=f"Failed to delete workspace file: {str(e)}",
-                error=str(e),
-                session_id=session_id,
-            )

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

@@ -1,250 +0,0 @@
-"""PostHog analytics tracking for the chat system."""
-
-import atexit
-import logging
-from typing import Any
-
-from posthog import Posthog
-
-from backend.util.settings import Settings
-
-logger = logging.getLogger(__name__)
-settings = Settings()
-
-# PostHog client instance (lazily initialized)
-_posthog_client: Posthog | None = None
-
-
-def _shutdown_posthog() -> None:
-    """Flush and shutdown PostHog client on process exit."""
-    if _posthog_client is not None:
-        _posthog_client.flush()
-        _posthog_client.shutdown()
-
-
-atexit.register(_shutdown_posthog)
-
-
-def _get_posthog_client() -> Posthog | None:
-    """Get or create the PostHog client instance."""
-    global _posthog_client
-    if _posthog_client is not None:
-        return _posthog_client
-
-    if not settings.secrets.posthog_api_key:
-        logger.debug("PostHog API key not configured, analytics disabled")
-        return None
-
-    _posthog_client = Posthog(
-        settings.secrets.posthog_api_key,
-        host=settings.secrets.posthog_host,
-    )
-    logger.info(
-        f"PostHog client initialized with host: {settings.secrets.posthog_host}"
-    )
-    return _posthog_client
-
-
-def _get_base_properties() -> dict[str, Any]:
-    """Get base properties included in all events."""
-    return {
-        "environment": settings.config.app_env.value,
-        "source": "chat_copilot",
-    }
-
-
-def track_user_message(
-    user_id: str | None,
-    session_id: str,
-    message_length: int,
-) -> None:
-    """Track when a user sends a message in chat.
-
-    Args:
-        user_id: The user's ID (or None for anonymous)
-        session_id: The chat session ID
-        message_length: Length of the user's message
-    """
-    client = _get_posthog_client()
-    if not client:
-        return
-
-    try:
-        properties = {
-            **_get_base_properties(),
-            "session_id": session_id,
-            "message_length": message_length,
-        }
-        client.capture(
-            distinct_id=user_id or f"anonymous_{session_id}",
-            event="copilot_message_sent",
-            properties=properties,
-        )
-    except Exception as e:
-        logger.warning(f"Failed to track user message: {e}")
-
-
-def track_tool_called(
-    user_id: str | None,
-    session_id: str,
-    tool_name: str,
-    tool_call_id: str,
-) -> None:
-    """Track when a tool is called in chat.
-
-    Args:
-        user_id: The user's ID (or None for anonymous)
-        session_id: The chat session ID
-        tool_name: Name of the tool being called
-        tool_call_id: Unique ID of the tool call
-    """
-    client = _get_posthog_client()
-    if not client:
-        logger.info("PostHog client not available for tool tracking")
-        return
-
-    try:
-        properties = {
-            **_get_base_properties(),
-            "session_id": session_id,
-            "tool_name": tool_name,
-            "tool_call_id": tool_call_id,
-        }
-        distinct_id = user_id or f"anonymous_{session_id}"
-        logger.info(
-            f"Sending copilot_tool_called event to PostHog: distinct_id={distinct_id}, "
-            f"tool_name={tool_name}"
-        )
-        client.capture(
-            distinct_id=distinct_id,
-            event="copilot_tool_called",
-            properties=properties,
-        )
-    except Exception as e:
-        logger.warning(f"Failed to track tool call: {e}")
-
-
-def track_agent_run_success(
-    user_id: str,
-    session_id: str,
-    graph_id: str,
-    graph_name: str,
-    execution_id: str,
-    library_agent_id: str,
-) -> None:
-    """Track when an agent is successfully run.
-
-    Args:
-        user_id: The user's ID
-        session_id: The chat session ID
-        graph_id: ID of the agent graph
-        graph_name: Name of the agent
-        execution_id: ID of the execution
-        library_agent_id: ID of the library agent
-    """
-    client = _get_posthog_client()
-    if not client:
-        return
-
-    try:
-        properties = {
-            **_get_base_properties(),
-            "session_id": session_id,
-            "graph_id": graph_id,
-            "graph_name": graph_name,
-            "execution_id": execution_id,
-            "library_agent_id": library_agent_id,
-        }
-        client.capture(
-            distinct_id=user_id,
-            event="copilot_agent_run_success",
-            properties=properties,
-        )
-    except Exception as e:
-        logger.warning(f"Failed to track agent run: {e}")
-
-
-def track_agent_scheduled(
-    user_id: str,
-    session_id: str,
-    graph_id: str,
-    graph_name: str,
-    schedule_id: str,
-    schedule_name: str,
-    cron: str,
-    library_agent_id: str,
-) -> None:
-    """Track when an agent is successfully scheduled.
-
-    Args:
-        user_id: The user's ID
-        session_id: The chat session ID
-        graph_id: ID of the agent graph
-        graph_name: Name of the agent
-        schedule_id: ID of the schedule
-        schedule_name: Name of the schedule
-        cron: Cron expression for the schedule
-        library_agent_id: ID of the library agent
-    """
-    client = _get_posthog_client()
-    if not client:
-        return
-
-    try:
-        properties = {
-            **_get_base_properties(),
-            "session_id": session_id,
-            "graph_id": graph_id,
-            "graph_name": graph_name,
-            "schedule_id": schedule_id,
-            "schedule_name": schedule_name,
-            "cron": cron,
-            "library_agent_id": library_agent_id,
-        }
-        client.capture(
-            distinct_id=user_id,
-            event="copilot_agent_scheduled",
-            properties=properties,
-        )
-    except Exception as e:
-        logger.warning(f"Failed to track agent schedule: {e}")
-
-
-def track_trigger_setup(
-    user_id: str,
-    session_id: str,
-    graph_id: str,
-    graph_name: str,
-    trigger_type: str,
-    library_agent_id: str,
-) -> None:
-    """Track when a trigger is set up for an agent.
-
-    Args:
-        user_id: The user's ID
-        session_id: The chat session ID
-        graph_id: ID of the agent graph
-        graph_name: Name of the agent
-        trigger_type: Type of trigger (e.g., 'webhook')
-        library_agent_id: ID of the library agent
-    """
-    client = _get_posthog_client()
-    if not client:
-        return
-
-    try:
-        properties = {
-            **_get_base_properties(),
-            "session_id": session_id,
-            "graph_id": graph_id,
-            "graph_name": graph_name,
-            "trigger_type": trigger_type,
-            "library_agent_id": library_agent_id,
-        }
-        client.capture(
-            distinct_id=user_id,
-            event="copilot_trigger_setup",
-            properties=properties,
-        )
-    except Exception as e:
-        logger.warning(f"Failed to track trigger setup: {e}")

autogpt_platform/backend/backend/api/features/executions/review/review_routes_test.py

@@ -164,9 +164,9 @@ async def test_process_review_action_approve_success(
     """Test successful review approval"""
     # Mock the route functions
 
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     mock_get_reviews_for_user.return_value = {"test_node_123": sample_pending_review}
 
@@ -244,9 +244,9 @@ async def test_process_review_action_reject_success(
     """Test successful review rejection"""
     # Mock the route functions
 
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     mock_get_reviews_for_user.return_value = {"test_node_123": sample_pending_review}
 
@@ -339,9 +339,9 @@ async def test_process_review_action_mixed_success(
 
     # Mock the route functions
 
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     mock_get_reviews_for_user.return_value = {
         "test_node_123": sample_pending_review,
@@ -463,9 +463,9 @@ async def test_process_review_action_review_not_found(
     test_user_id: str,
 ) -> None:
     """Test error when review is not found"""
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     # Return empty dict to simulate review not found
     mock_get_reviews_for_user.return_value = {}
@@ -506,7 +506,7 @@ async def test_process_review_action_review_not_found(
     response = await client.post("/api/review/action", json=request_data)
 
     assert response.status_code == 404
-    assert "Review(s) not found" in response.json()["detail"]
+    assert "No pending review found" in response.json()["detail"]
 
 
 @pytest.mark.asyncio(loop_scope="session")
@@ -517,9 +517,9 @@ async def test_process_review_action_partial_failure(
     test_user_id: str,
 ) -> None:
     """Test handling of partial failures in review processing"""
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     mock_get_reviews_for_user.return_value = {"test_node_123": sample_pending_review}
 
@@ -567,9 +567,9 @@ async def test_process_review_action_invalid_node_exec_id(
     test_user_id: str,
 ) -> None:
     """Test failure when trying to process review with invalid node execution ID"""
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     # Return empty dict to simulate review not found
     mock_get_reviews_for_user.return_value = {}
@@ -596,7 +596,7 @@ async def test_process_review_action_invalid_node_exec_id(
 
     # Returns 404 when review is not found
     assert response.status_code == 404
-    assert "Review(s) not found" in response.json()["detail"]
+    assert "No pending review found" in response.json()["detail"]
 
 
 @pytest.mark.asyncio(loop_scope="session")
@@ -607,9 +607,9 @@ async def test_process_review_action_auto_approve_creates_auto_approval_records(
     test_user_id: str,
 ) -> None:
     """Test that auto_approve_future_actions flag creates auto-approval records"""
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     mock_get_reviews_for_user.return_value = {"test_node_123": sample_pending_review}
 
@@ -737,9 +737,9 @@ async def test_process_review_action_without_auto_approve_still_loads_settings(
     test_user_id: str,
 ) -> None:
     """Test that execution context is created with settings even without auto-approve"""
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     mock_get_reviews_for_user.return_value = {"test_node_123": sample_pending_review}
 
@@ -885,9 +885,9 @@ async def test_process_review_action_auto_approve_only_applies_to_approved_revie
         reviewed_at=FIXED_NOW,
     )
 
-    # Mock get_reviews_by_node_exec_ids (called to find the graph_exec_id)
+    # Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
     # Need to return both reviews in WAITING state (before processing)
     approved_review_waiting = PendingHumanReviewModel(
@@ -1031,9 +1031,9 @@ async def test_process_review_action_per_review_auto_approve_granularity(
     test_user_id: str,
 ) -> None:
     """Test that auto-approval can be set per-review (granular control)"""
-    # Mock get_reviews_by_node_exec_ids - return different reviews based on node_exec_id
+    # Mock get_pending_reviews_by_node_exec_ids - return different reviews based on node_exec_id
     mock_get_reviews_for_user = mocker.patch(
-        "backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
+        "backend.api.features.executions.review.routes.get_pending_reviews_by_node_exec_ids"
     )
 
     # Create a mapping of node_exec_id to review

autogpt_platform/backend/backend/api/features/executions/review/routes.py

@@ -14,9 +14,9 @@ from backend.data.execution import (
 from backend.data.graph import get_graph_settings
 from backend.data.human_review import (
     create_auto_approval_record,
+    get_pending_reviews_by_node_exec_ids,
     get_pending_reviews_for_execution,
     get_pending_reviews_for_user,
-    get_reviews_by_node_exec_ids,
     has_pending_reviews_for_graph_exec,
     process_all_reviews_for_execution,
 )
@@ -137,17 +137,17 @@ async def process_review_action(
             detail="At least one review must be provided",
         )
 
-    # Batch fetch all requested reviews (regardless of status for idempotent handling)
-    reviews_map = await get_reviews_by_node_exec_ids(
+    # Batch fetch all requested reviews
+    reviews_map = await get_pending_reviews_by_node_exec_ids(
         list(all_request_node_ids), user_id
     )
 
-    # Validate all reviews were found (must exist, any status is OK for now)
+    # Validate all reviews were found
     missing_ids = all_request_node_ids - set(reviews_map.keys())
     if missing_ids:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"Review(s) not found: {', '.join(missing_ids)}",
+            detail=f"No pending review found for node execution(s): {', '.join(missing_ids)}",
         )
 
     # Validate all reviews belong to the same execution

autogpt_platform/backend/backend/api/features/library/db.py

@@ -21,7 +21,7 @@ from backend.data.model import CredentialsMetaInput
 from backend.integrations.creds_manager import IntegrationCredentialsManager
 from backend.integrations.webhooks.graph_lifecycle_hooks import on_graph_activate
 from backend.util.clients import get_scheduler_client
-from backend.util.exceptions import DatabaseError, InvalidInputError, NotFoundError
+from backend.util.exceptions import DatabaseError, NotFoundError
 from backend.util.json import SafeJson
 from backend.util.models import Pagination
 from backend.util.settings import Config
@@ -39,7 +39,6 @@ async def list_library_agents(
     sort_by: library_model.LibraryAgentSort = library_model.LibraryAgentSort.UPDATED_AT,
     page: int = 1,
     page_size: int = 50,
-    include_executions: bool = False,
 ) -> library_model.LibraryAgentResponse:
     """
     Retrieves a paginated list of LibraryAgent records for a given user.
@@ -50,9 +49,6 @@ async def list_library_agents(
         sort_by: Sorting field (createdAt, updatedAt, isFavorite, isCreatedByUser).
         page: Current page (1-indexed).
         page_size: Number of items per page.
-        include_executions: Whether to include execution data for status calculation.
-            Defaults to False for performance (UI fetches status separately).
-            Set to True when accurate status/metrics are needed (e.g., agent generator).
 
     Returns:
         A LibraryAgentResponse containing the list of agents and pagination details.
@@ -68,11 +64,11 @@ async def list_library_agents(
 
     if page < 1 or page_size < 1:
         logger.warning(f"Invalid pagination: page={page}, page_size={page_size}")
-        raise InvalidInputError("Invalid pagination input")
+        raise DatabaseError("Invalid pagination input")
 
     if search_term and len(search_term.strip()) > 100:
         logger.warning(f"Search term too long: {repr(search_term)}")
-        raise InvalidInputError("Search term is too long")
+        raise DatabaseError("Search term is too long")
 
     where_clause: prisma.types.LibraryAgentWhereInput = {
         "userId": user_id,
@@ -80,6 +76,7 @@ async def list_library_agents(
         "isArchived": False,
     }
 
+    # Build search filter if applicable
     if search_term:
         where_clause["OR"] = [
             {
@@ -96,6 +93,7 @@ async def list_library_agents(
             },
         ]
 
+    # Determine sorting
     order_by: prisma.types.LibraryAgentOrderByInput | None = None
 
     if sort_by == library_model.LibraryAgentSort.CREATED_AT:
@@ -107,7 +105,7 @@ async def list_library_agents(
         library_agents = await prisma.models.LibraryAgent.prisma().find_many(
             where=where_clause,
             include=library_agent_include(
-                user_id, include_nodes=False, include_executions=include_executions
+                user_id, include_nodes=False, include_executions=False
             ),
             order=order_by,
             skip=(page - 1) * page_size,
@@ -177,7 +175,7 @@ async def list_favorite_library_agents(
 
     if page < 1 or page_size < 1:
         logger.warning(f"Invalid pagination: page={page}, page_size={page_size}")
-        raise InvalidInputError("Invalid pagination input")
+        raise DatabaseError("Invalid pagination input")
 
     where_clause: prisma.types.LibraryAgentWhereInput = {
         "userId": user_id,

autogpt_platform/backend/backend/api/features/library/model.py

@@ -9,7 +9,6 @@ import pydantic
 from backend.data.block import BlockInput
 from backend.data.graph import GraphModel, GraphSettings, GraphTriggerInfo
 from backend.data.model import CredentialsMetaInput, is_credentials_field_name
-from backend.util.json import loads as json_loads
 from backend.util.models import Pagination
 
 if TYPE_CHECKING:
@@ -17,10 +16,10 @@ if TYPE_CHECKING:
 
 
 class LibraryAgentStatus(str, Enum):
-    COMPLETED = "COMPLETED"
-    HEALTHY = "HEALTHY"
-    WAITING = "WAITING"
-    ERROR = "ERROR"
+    COMPLETED = "COMPLETED"  # All runs completed
+    HEALTHY = "HEALTHY"  # Agent is running (not all runs have completed)
+    WAITING = "WAITING"  # Agent is queued or waiting to start
+    ERROR = "ERROR"  # Agent is in an error state
 
 
 class MarketplaceListingCreator(pydantic.BaseModel):
@@ -40,30 +39,6 @@ class MarketplaceListing(pydantic.BaseModel):
     creator: MarketplaceListingCreator
 
 
-class RecentExecution(pydantic.BaseModel):
-    """Summary of a recent execution for quality assessment.
-
-    Used by the LLM to understand the agent's recent performance with specific examples
-    rather than just aggregate statistics.
-    """
-
-    status: str
-    correctness_score: float | None = None
-    activity_summary: str | None = None
-
-
-def _parse_settings(settings: dict | str | None) -> GraphSettings:
-    """Parse settings from database, handling both dict and string formats."""
-    if settings is None:
-        return GraphSettings()
-    try:
-        if isinstance(settings, str):
-            settings = json_loads(settings)
-        return GraphSettings.model_validate(settings)
-    except Exception:
-        return GraphSettings()
-
-
 class LibraryAgent(pydantic.BaseModel):
     """
     Represents an agent in the library, including metadata for display and
@@ -73,7 +48,7 @@ class LibraryAgent(pydantic.BaseModel):
     id: str
     graph_id: str
     graph_version: int
-    owner_user_id: str
+    owner_user_id: str  # ID of user who owns/created this agent graph
 
     image_url: str | None
 
@@ -89,7 +64,7 @@ class LibraryAgent(pydantic.BaseModel):
     description: str
     instructions: str | None = None
 
-    input_schema: dict[str, Any]
+    input_schema: dict[str, Any]  # Should be BlockIOObjectSubSchema in frontend
     output_schema: dict[str, Any]
     credentials_input_schema: dict[str, Any] | None = pydantic.Field(
         description="Input schema for credentials required by the agent",
@@ -106,19 +81,25 @@ class LibraryAgent(pydantic.BaseModel):
     )
     trigger_setup_info: Optional[GraphTriggerInfo] = None
 
+    # Indicates whether there's a new output (based on recent runs)
     new_output: bool
-    execution_count: int = 0
-    success_rate: float | None = None
-    avg_correctness_score: float | None = None
-    recent_executions: list[RecentExecution] = pydantic.Field(
-        default_factory=list,
-        description="List of recent executions with status, score, and summary",
-    )
+
+    # Whether the user can access the underlying graph
     can_access_graph: bool
+
+    # Indicates if this agent is the latest version
     is_latest_version: bool
+
+    # Whether the agent is marked as favorite by the user
     is_favorite: bool
+
+    # Recommended schedule cron (from marketplace agents)
     recommended_schedule_cron: str | None = None
+
+    # User-specific settings for this library agent
     settings: GraphSettings = pydantic.Field(default_factory=GraphSettings)
+
+    # Marketplace listing information if the agent has been published
     marketplace_listing: Optional["MarketplaceListing"] = None
 
     @staticmethod
@@ -142,6 +123,7 @@ class LibraryAgent(pydantic.BaseModel):
         agent_updated_at = agent.AgentGraph.updatedAt
         lib_agent_updated_at = agent.updatedAt
 
+        # Compute updated_at as the latest between library agent and graph
         updated_at = (
             max(agent_updated_at, lib_agent_updated_at)
             if agent_updated_at
@@ -154,6 +136,7 @@ class LibraryAgent(pydantic.BaseModel):
             creator_name = agent.Creator.name or "Unknown"
             creator_image_url = agent.Creator.avatarUrl or ""
 
+        # Logic to calculate status and new_output
         week_ago = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
             days=7
         )
@@ -162,55 +145,13 @@ class LibraryAgent(pydantic.BaseModel):
         status = status_result.status
         new_output = status_result.new_output
 
-        execution_count = len(executions)
-        success_rate: float | None = None
-        avg_correctness_score: float | None = None
-        if execution_count > 0:
-            success_count = sum(
-                1
-                for e in executions
-                if e.executionStatus == prisma.enums.AgentExecutionStatus.COMPLETED
-            )
-            success_rate = (success_count / execution_count) * 100
-
-            correctness_scores = []
-            for e in executions:
-                if e.stats and isinstance(e.stats, dict):
-                    score = e.stats.get("correctness_score")
-                    if score is not None and isinstance(score, (int, float)):
-                        correctness_scores.append(float(score))
-            if correctness_scores:
-                avg_correctness_score = sum(correctness_scores) / len(
-                    correctness_scores
-                )
-
-        recent_executions: list[RecentExecution] = []
-        for e in executions:
-            exec_score: float | None = None
-            exec_summary: str | None = None
-            if e.stats and isinstance(e.stats, dict):
-                score = e.stats.get("correctness_score")
-                if score is not None and isinstance(score, (int, float)):
-                    exec_score = float(score)
-                summary = e.stats.get("activity_status")
-                if summary is not None and isinstance(summary, str):
-                    exec_summary = summary
-            exec_status = (
-                e.executionStatus.value
-                if hasattr(e.executionStatus, "value")
-                else str(e.executionStatus)
-            )
-            recent_executions.append(
-                RecentExecution(
-                    status=exec_status,
-                    correctness_score=exec_score,
-                    activity_summary=exec_summary,
-                )
-            )
-
+        # Check if user can access the graph
         can_access_graph = agent.AgentGraph.userId == agent.userId
+
+        # Hard-coded to True until a method to check is implemented
         is_latest_version = True
 
+        # Build marketplace_listing if available
         marketplace_listing_data = None
         if store_listing and store_listing.ActiveVersion and profile:
             creator_data = MarketplaceListingCreator(
@@ -249,15 +190,11 @@ class LibraryAgent(pydantic.BaseModel):
             has_sensitive_action=graph.has_sensitive_action,
             trigger_setup_info=graph.trigger_setup_info,
             new_output=new_output,
-            execution_count=execution_count,
-            success_rate=success_rate,
-            avg_correctness_score=avg_correctness_score,
-            recent_executions=recent_executions,
             can_access_graph=can_access_graph,
             is_latest_version=is_latest_version,
             is_favorite=agent.isFavorite,
             recommended_schedule_cron=agent.AgentGraph.recommendedScheduleCron,
-            settings=_parse_settings(agent.settings),
+            settings=GraphSettings.model_validate(agent.settings),
             marketplace_listing=marketplace_listing_data,
         )
 
@@ -283,15 +220,18 @@ def _calculate_agent_status(
     if not executions:
         return AgentStatusResult(status=LibraryAgentStatus.COMPLETED, new_output=False)
 
+    # Track how many times each execution status appears
     status_counts = {status: 0 for status in prisma.enums.AgentExecutionStatus}
     new_output = False
 
     for execution in executions:
+        # Check if there's a completed run more recent than `recent_threshold`
         if execution.createdAt >= recent_threshold:
             if execution.executionStatus == prisma.enums.AgentExecutionStatus.COMPLETED:
                 new_output = True
         status_counts[execution.executionStatus] += 1
 
+    # Determine the final status based on counts
     if status_counts[prisma.enums.AgentExecutionStatus.FAILED] > 0:
         return AgentStatusResult(status=LibraryAgentStatus.ERROR, new_output=new_output)
     elif status_counts[prisma.enums.AgentExecutionStatus.QUEUED] > 0:

autogpt_platform/backend/backend/api/features/library/routes/agents.py

@@ -1,3 +1,4 @@
+import logging
 from typing import Literal, Optional
 
 import autogpt_libs.auth as autogpt_auth_lib
@@ -5,11 +6,15 @@ from fastapi import APIRouter, Body, HTTPException, Query, Security, status
 from fastapi.responses import Response
 from prisma.enums import OnboardingStep
 
+import backend.api.features.store.exceptions as store_exceptions
 from backend.data.onboarding import complete_onboarding_step
+from backend.util.exceptions import DatabaseError, NotFoundError
 
 from .. import db as library_db
 from .. import model as library_model
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(
     prefix="/agents",
     tags=["library", "private"],
@@ -21,6 +26,10 @@ router = APIRouter(
     "",
     summary="List Library Agents",
     response_model=library_model.LibraryAgentResponse,
+    responses={
+        200: {"description": "List of library agents"},
+        500: {"description": "Server error", "content": {"application/json": {}}},
+    },
 )
 async def list_library_agents(
     user_id: str = Security(autogpt_auth_lib.get_user_id),
@@ -44,19 +53,43 @@ async def list_library_agents(
 ) -> library_model.LibraryAgentResponse:
     """
     Get all agents in the user's library (both created and saved).
+
+    Args:
+        user_id: ID of the authenticated user.
+        search_term: Optional search term to filter agents by name/description.
+        filter_by: List of filters to apply (favorites, created by user).
+        sort_by: List of sorting criteria (created date, updated date).
+        page: Page number to retrieve.
+        page_size: Number of agents per page.
+
+    Returns:
+        A LibraryAgentResponse containing agents and pagination metadata.
+
+    Raises:
+        HTTPException: If a server/database error occurs.
     """
-    return await library_db.list_library_agents(
-        user_id=user_id,
-        search_term=search_term,
-        sort_by=sort_by,
-        page=page,
-        page_size=page_size,
-    )
+    try:
+        return await library_db.list_library_agents(
+            user_id=user_id,
+            search_term=search_term,
+            sort_by=sort_by,
+            page=page,
+            page_size=page_size,
+        )
+    except Exception as e:
+        logger.error(f"Could not list library agents for user #{user_id}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=str(e),
+        ) from e
 
 
 @router.get(
     "/favorites",
     summary="List Favorite Library Agents",
+    responses={
+        500: {"description": "Server error", "content": {"application/json": {}}},
+    },
 )
 async def list_favorite_library_agents(
     user_id: str = Security(autogpt_auth_lib.get_user_id),
@@ -73,12 +106,30 @@ async def list_favorite_library_agents(
 ) -> library_model.LibraryAgentResponse:
     """
     Get all favorite agents in the user's library.
+
+    Args:
+        user_id: ID of the authenticated user.
+        page: Page number to retrieve.
+        page_size: Number of agents per page.
+
+    Returns:
+        A LibraryAgentResponse containing favorite agents and pagination metadata.
+
+    Raises:
+        HTTPException: If a server/database error occurs.
     """
-    return await library_db.list_favorite_library_agents(
-        user_id=user_id,
-        page=page,
-        page_size=page_size,
-    )
+    try:
+        return await library_db.list_favorite_library_agents(
+            user_id=user_id,
+            page=page,
+            page_size=page_size,
+        )
+    except Exception as e:
+        logger.error(f"Could not list favorite library agents for user #{user_id}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=str(e),
+        ) from e
 
 
 @router.get("/{library_agent_id}", summary="Get Library Agent")
@@ -111,6 +162,10 @@ async def get_library_agent_by_graph_id(
     summary="Get Agent By Store ID",
     tags=["store", "library"],
     response_model=library_model.LibraryAgent | None,
+    responses={
+        200: {"description": "Library agent found"},
+        404: {"description": "Agent not found"},
+    },
 )
 async def get_library_agent_by_store_listing_version_id(
     store_listing_version_id: str,
@@ -119,15 +174,32 @@ async def get_library_agent_by_store_listing_version_id(
     """
     Get Library Agent from Store Listing Version ID.
     """
-    return await library_db.get_library_agent_by_store_version_id(
-        store_listing_version_id, user_id
-    )
+    try:
+        return await library_db.get_library_agent_by_store_version_id(
+            store_listing_version_id, user_id
+        )
+    except NotFoundError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(e),
+        )
+    except Exception as e:
+        logger.error(f"Could not fetch library agent from store version ID: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=str(e),
+        ) from e
 
 
 @router.post(
     "",
     summary="Add Marketplace Agent",
     status_code=status.HTTP_201_CREATED,
+    responses={
+        201: {"description": "Agent added successfully"},
+        404: {"description": "Store listing version not found"},
+        500: {"description": "Server error"},
+    },
 )
 async def add_marketplace_agent_to_library(
     store_listing_version_id: str = Body(embed=True),
@@ -138,19 +210,59 @@ async def add_marketplace_agent_to_library(
 ) -> library_model.LibraryAgent:
     """
     Add an agent from the marketplace to the user's library.
+
+    Args:
+        store_listing_version_id: ID of the store listing version to add.
+        user_id: ID of the authenticated user.
+
+    Returns:
+        library_model.LibraryAgent: Agent added to the library
+
+    Raises:
+        HTTPException(404): If the listing version is not found.
+        HTTPException(500): If a server/database error occurs.
     """
-    agent = await library_db.add_store_agent_to_library(
-        store_listing_version_id=store_listing_version_id,
-        user_id=user_id,
-    )
-    if source != "onboarding":
-        await complete_onboarding_step(user_id, OnboardingStep.MARKETPLACE_ADD_AGENT)
-    return agent
+    try:
+        agent = await library_db.add_store_agent_to_library(
+            store_listing_version_id=store_listing_version_id,
+            user_id=user_id,
+        )
+        if source != "onboarding":
+            await complete_onboarding_step(
+                user_id, OnboardingStep.MARKETPLACE_ADD_AGENT
+            )
+        return agent
+
+    except store_exceptions.AgentNotFoundError as e:
+        logger.warning(
+            f"Could not find store listing version {store_listing_version_id} "
+            "to add to library"
+        )
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
+    except DatabaseError as e:
+        logger.error(f"Database error while adding agent to library: {e}", e)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={"message": str(e), "hint": "Inspect DB logs for details."},
+        ) from e
+    except Exception as e:
+        logger.error(f"Unexpected error while adding agent to library: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={
+                "message": str(e),
+                "hint": "Check server logs for more information.",
+            },
+        ) from e
 
 
 @router.patch(
     "/{library_agent_id}",
     summary="Update Library Agent",
+    responses={
+        200: {"description": "Agent updated successfully"},
+        500: {"description": "Server error"},
+    },
 )
 async def update_library_agent(
     library_agent_id: str,
@@ -159,21 +271,52 @@ async def update_library_agent(
 ) -> library_model.LibraryAgent:
     """
     Update the library agent with the given fields.
+
+    Args:
+        library_agent_id: ID of the library agent to update.
+        payload: Fields to update (auto_update_version, is_favorite, etc.).
+        user_id: ID of the authenticated user.
+
+    Raises:
+        HTTPException(500): If a server/database error occurs.
     """
-    return await library_db.update_library_agent(
-        library_agent_id=library_agent_id,
-        user_id=user_id,
-        auto_update_version=payload.auto_update_version,
-        graph_version=payload.graph_version,
-        is_favorite=payload.is_favorite,
-        is_archived=payload.is_archived,
-        settings=payload.settings,
-    )
+    try:
+        return await library_db.update_library_agent(
+            library_agent_id=library_agent_id,
+            user_id=user_id,
+            auto_update_version=payload.auto_update_version,
+            graph_version=payload.graph_version,
+            is_favorite=payload.is_favorite,
+            is_archived=payload.is_archived,
+            settings=payload.settings,
+        )
+    except NotFoundError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(e),
+        ) from e
+    except DatabaseError as e:
+        logger.error(f"Database error while updating library agent: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={"message": str(e), "hint": "Verify DB connection."},
+        ) from e
+    except Exception as e:
+        logger.error(f"Unexpected error while updating library agent: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={"message": str(e), "hint": "Check server logs."},
+        ) from e
 
 
 @router.delete(
     "/{library_agent_id}",
     summary="Delete Library Agent",
+    responses={
+        204: {"description": "Agent deleted successfully"},
+        404: {"description": "Agent not found"},
+        500: {"description": "Server error"},
+    },
 )
 async def delete_library_agent(
     library_agent_id: str,
@@ -181,11 +324,28 @@ async def delete_library_agent(
 ) -> Response:
     """
     Soft-delete the specified library agent.
+
+    Args:
+        library_agent_id: ID of the library agent to delete.
+        user_id: ID of the authenticated user.
+
+    Returns:
+        204 No Content if successful.
+
+    Raises:
+        HTTPException(404): If the agent does not exist.
+        HTTPException(500): If a server/database error occurs.
     """
-    await library_db.delete_library_agent(
-        library_agent_id=library_agent_id, user_id=user_id
-    )
-    return Response(status_code=status.HTTP_204_NO_CONTENT)
+    try:
+        await library_db.delete_library_agent(
+            library_agent_id=library_agent_id, user_id=user_id
+        )
+        return Response(status_code=status.HTTP_204_NO_CONTENT)
+    except NotFoundError as e:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(e),
+        ) from e
 
 
 @router.post("/{library_agent_id}/fork", summary="Fork Library Agent")

autogpt_platform/backend/backend/api/features/library/routes_test.py

@@ -118,6 +118,21 @@ async def test_get_library_agents_success(
     )
 
 
+def test_get_library_agents_error(mocker: pytest_mock.MockFixture, test_user_id: str):
+    mock_db_call = mocker.patch("backend.api.features.library.db.list_library_agents")
+    mock_db_call.side_effect = Exception("Test error")
+
+    response = client.get("/agents?search_term=test")
+    assert response.status_code == 500
+    mock_db_call.assert_called_once_with(
+        user_id=test_user_id,
+        search_term="test",
+        sort_by=library_model.LibraryAgentSort.UPDATED_AT,
+        page=1,
+        page_size=15,
+    )
+
+
 @pytest.mark.asyncio
 async def test_get_favorite_library_agents_success(
     mocker: pytest_mock.MockFixture,
@@ -175,6 +190,23 @@ async def test_get_favorite_library_agents_success(
     )
 
 
+def test_get_favorite_library_agents_error(
+    mocker: pytest_mock.MockFixture, test_user_id: str
+):
+    mock_db_call = mocker.patch(
+        "backend.api.features.library.db.list_favorite_library_agents"
+    )
+    mock_db_call.side_effect = Exception("Test error")
+
+    response = client.get("/agents/favorites")
+    assert response.status_code == 500
+    mock_db_call.assert_called_once_with(
+        user_id=test_user_id,
+        page=1,
+        page_size=15,
+    )
+
+
 def test_add_agent_to_library_success(
     mocker: pytest_mock.MockFixture, test_user_id: str
 ):
@@ -226,3 +258,19 @@ def test_add_agent_to_library_success(
         store_listing_version_id="test-version-id", user_id=test_user_id
     )
     mock_complete_onboarding.assert_awaited_once()
+
+
+def test_add_agent_to_library_error(mocker: pytest_mock.MockFixture, test_user_id: str):
+    mock_db_call = mocker.patch(
+        "backend.api.features.library.db.add_store_agent_to_library"
+    )
+    mock_db_call.side_effect = Exception("Test error")
+
+    response = client.post(
+        "/agents", json={"store_listing_version_id": "test-version-id"}
+    )
+    assert response.status_code == 500
+    assert "detail" in response.json()  # Verify error response structure
+    mock_db_call.assert_called_once_with(
+        store_listing_version_id="test-version-id", user_id=test_user_id
+    )

autogpt_platform/backend/backend/api/features/store/content_handlers.py

@@ -188,10 +188,6 @@ class BlockHandler(ContentHandler):
             try:
                 block_instance = block_cls()
 
-                # Skip disabled blocks - they shouldn't be indexed
-                if block_instance.disabled:
-                    continue
-
                 # Build searchable text from block metadata
                 parts = []
                 if hasattr(block_instance, "name") and block_instance.name:
@@ -252,19 +248,12 @@ class BlockHandler(ContentHandler):
         from backend.data.block import get_blocks
 
         all_blocks = get_blocks()
-
-        # Filter out disabled blocks - they're not indexed
-        enabled_block_ids = [
-            block_id
-            for block_id, block_cls in all_blocks.items()
-            if not block_cls().disabled
-        ]
-        total_blocks = len(enabled_block_ids)
+        total_blocks = len(all_blocks)
 
         if total_blocks == 0:
             return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}
 
-        block_ids = enabled_block_ids
+        block_ids = list(all_blocks.keys())
         placeholders = ",".join([f"${i+1}" for i in range(len(block_ids))])
 
         embedded_result = await query_raw_with_schema(

autogpt_platform/backend/backend/api/features/store/content_handlers_test.py

@@ -81,7 +81,6 @@ async def test_block_handler_get_missing_items(mocker):
     mock_block_instance.name = "Calculator Block"
     mock_block_instance.description = "Performs calculations"
     mock_block_instance.categories = [MagicMock(value="MATH")]
-    mock_block_instance.disabled = False
     mock_block_instance.input_schema.model_json_schema.return_value = {
         "properties": {"expression": {"description": "Math expression to evaluate"}}
     }
@@ -117,18 +116,11 @@ async def test_block_handler_get_stats(mocker):
     """Test BlockHandler returns correct stats."""
     handler = BlockHandler()
 
-    # Mock get_blocks - each block class returns an instance with disabled=False
-    def make_mock_block_class():
-        mock_class = MagicMock()
-        mock_instance = MagicMock()
-        mock_instance.disabled = False
-        mock_class.return_value = mock_instance
-        return mock_class
-
+    # Mock get_blocks
     mock_blocks = {
-        "block-1": make_mock_block_class(),
-        "block-2": make_mock_block_class(),
-        "block-3": make_mock_block_class(),
+        "block-1": MagicMock(),
+        "block-2": MagicMock(),
+        "block-3": MagicMock(),
     }
 
     # Mock embedded count query (2 blocks have embeddings)
@@ -317,7 +309,6 @@ async def test_block_handler_handles_missing_attributes():
     mock_block_class = MagicMock()
     mock_block_instance = MagicMock()
     mock_block_instance.name = "Minimal Block"
-    mock_block_instance.disabled = False
     # No description, categories, or schema
     del mock_block_instance.description
     del mock_block_instance.categories
@@ -351,7 +342,6 @@ async def test_block_handler_skips_failed_blocks():
     good_instance.name = "Good Block"
     good_instance.description = "Works fine"
     good_instance.categories = []
-    good_instance.disabled = False
     good_block.return_value = good_instance
 
     bad_block = MagicMock()

autogpt_platform/backend/backend/api/features/store/db.py

@@ -112,7 +112,6 @@ async def get_store_agents(
                             description=agent["description"],
                             runs=agent["runs"],
                             rating=agent["rating"],
-                            agent_graph_id=agent.get("agentGraphId", ""),
                         )
                         store_agents.append(store_agent)
                     except Exception as e:
@@ -171,7 +170,6 @@ async def get_store_agents(
                         description=agent.description,
                         runs=agent.runs,
                         rating=agent.rating,
-                        agent_graph_id=agent.agentGraphId,
                     )
                     # Add to the list only if creation was successful
                     store_agents.append(store_agent)

autogpt_platform/backend/backend/api/features/store/embeddings.py

@@ -454,7 +454,6 @@ async def backfill_all_content_types(batch_size: int = 10) -> dict[str, Any]:
     total_processed = 0
     total_success = 0
     total_failed = 0
-    all_errors: dict[str, int] = {}  # Aggregate errors across all content types
 
     # Process content types in explicit order
     processing_order = [
@@ -500,12 +499,23 @@ async def backfill_all_content_types(batch_size: int = 10) -> dict[str, Any]:
             success = sum(1 for result in results if result is True)
             failed = len(results) - success
 
-            # Aggregate errors across all content types
+            # Aggregate unique errors to avoid Sentry spam
             if failed > 0:
+                # Group errors by type and message
+                error_summary: dict[str, int] = {}
                 for result in results:
                     if isinstance(result, Exception):
                         error_key = f"{type(result).__name__}: {str(result)}"
-                        all_errors[error_key] = all_errors.get(error_key, 0) + 1
+                        error_summary[error_key] = error_summary.get(error_key, 0) + 1
+
+                # Log aggregated error summary
+                error_details = ", ".join(
+                    f"{error} ({count}x)" for error, count in error_summary.items()
+                )
+                logger.error(
+                    f"{content_type.value}: {failed}/{len(results)} embeddings failed. "
+                    f"Errors: {error_details}"
+                )
 
             results_by_type[content_type.value] = {
                 "processed": len(missing_items),
@@ -532,13 +542,6 @@ async def backfill_all_content_types(batch_size: int = 10) -> dict[str, Any]:
                 "error": str(e),
             }
 
-    # Log aggregated errors once at the end
-    if all_errors:
-        error_details = ", ".join(
-            f"{error} ({count}x)" for error, count in all_errors.items()
-        )
-        logger.error(f"Embedding backfill errors: {error_details}")
-
     return {
         "by_type": results_by_type,
         "totals": {

autogpt_platform/backend/backend/api/features/store/hybrid_search.py

@@ -600,7 +600,6 @@ async def hybrid_search(
                 sa.featured,
                 sa.is_available,
                 sa.updated_at,
-                sa."agentGraphId",
                 -- Searchable text for BM25 reranking
                 COALESCE(sa.agent_name, '') || ' ' || COALESCE(sa.sub_heading, '') || ' ' || COALESCE(sa.description, '') as searchable_text,
                 -- Semantic score
@@ -660,7 +659,6 @@ async def hybrid_search(
                 featured,
                 is_available,
                 updated_at,
-                "agentGraphId",
                 searchable_text,
                 semantic_score,
                 lexical_score,

autogpt_platform/backend/backend/api/features/store/model.py

@@ -38,7 +38,6 @@ class StoreAgent(pydantic.BaseModel):
     description: str
     runs: int
     rating: float
-    agent_graph_id: str
 
 
 class StoreAgentsResponse(pydantic.BaseModel):

autogpt_platform/backend/backend/api/features/store/model_test.py

@@ -26,13 +26,11 @@ def test_store_agent():
         description="Test description",
         runs=50,
         rating=4.5,
-        agent_graph_id="test-graph-id",
     )
     assert agent.slug == "test-agent"
     assert agent.agent_name == "Test Agent"
     assert agent.runs == 50
     assert agent.rating == 4.5
-    assert agent.agent_graph_id == "test-graph-id"
 
 
 def test_store_agents_response():
@@ -48,7 +46,6 @@ def test_store_agents_response():
                 description="Test description",
                 runs=50,
                 rating=4.5,
-                agent_graph_id="test-graph-id",
             )
         ],
         pagination=store_model.Pagination(

autogpt_platform/backend/backend/api/features/store/routes_test.py

@@ -82,7 +82,6 @@ def test_get_agents_featured(
                 description="Featured agent description",
                 runs=100,
                 rating=4.5,
-                agent_graph_id="test-graph-1",
             )
         ],
         pagination=store_model.Pagination(
@@ -128,7 +127,6 @@ def test_get_agents_by_creator(
                 description="Creator agent description",
                 runs=50,
                 rating=4.0,
-                agent_graph_id="test-graph-2",
             )
         ],
         pagination=store_model.Pagination(
@@ -174,7 +172,6 @@ def test_get_agents_sorted(
                 description="Top agent description",
                 runs=1000,
                 rating=5.0,
-                agent_graph_id="test-graph-3",
             )
         ],
         pagination=store_model.Pagination(
@@ -220,7 +217,6 @@ def test_get_agents_search(
                 description="Specific search term description",
                 runs=75,
                 rating=4.2,
-                agent_graph_id="test-graph-search",
             )
         ],
         pagination=store_model.Pagination(
@@ -266,7 +262,6 @@ def test_get_agents_category(
                 description="Category agent description",
                 runs=60,
                 rating=4.1,
-                agent_graph_id="test-graph-category",
             )
         ],
         pagination=store_model.Pagination(
@@ -311,7 +306,6 @@ def test_get_agents_pagination(
                 description=f"Agent {i} description",
                 runs=i * 10,
                 rating=4.0,
-                agent_graph_id="test-graph-2",
             )
             for i in range(5)
         ],

autogpt_platform/backend/backend/api/features/store/test_cache_delete.py

@@ -33,7 +33,6 @@ class TestCacheDeletion:
                     description="Test description",
                     runs=100,
                     rating=4.5,
-                    agent_graph_id="test-graph-id",
                 )
             ],
             pagination=Pagination(

autogpt_platform/backend/backend/api/features/v1.py

@@ -261,36 +261,14 @@ async def get_onboarding_agents(
     return await get_recommended_agents(user_id)
 
 
-class OnboardingStatusResponse(pydantic.BaseModel):
-    """Response for onboarding status check."""
-
-    is_onboarding_enabled: bool
-    is_chat_enabled: bool
-
-
 @v1_router.get(
     "/onboarding/enabled",
     summary="Is onboarding enabled",
     tags=["onboarding", "public"],
-    response_model=OnboardingStatusResponse,
+    dependencies=[Security(requires_user)],
 )
-async def is_onboarding_enabled(
-    user_id: Annotated[str, Security(get_user_id)],
-) -> OnboardingStatusResponse:
-    # Check if chat is enabled for user
-    is_chat_enabled = await is_feature_enabled(Flag.CHAT, user_id, False)
-
-    # If chat is enabled, skip legacy onboarding
-    if is_chat_enabled:
-        return OnboardingStatusResponse(
-            is_onboarding_enabled=False,
-            is_chat_enabled=True,
-        )
-
-    return OnboardingStatusResponse(
-        is_onboarding_enabled=await onboarding_enabled(),
-        is_chat_enabled=False,
-    )
+async def is_onboarding_enabled() -> bool:
+    return await onboarding_enabled()
 
 
 @v1_router.post(
@@ -386,8 +364,6 @@ async def execute_graph_block(
     obj = get_block(block_id)
     if not obj:
         raise HTTPException(status_code=404, detail=f"Block #{block_id} not found.")
-    if obj.disabled:
-        raise HTTPException(status_code=403, detail=f"Block #{block_id} is disabled.")
 
     user = await get_user_by_id(user_id)
     if not user:

autogpt_platform/backend/backend/api/features/v1_test.py

@@ -138,7 +138,6 @@ def test_execute_graph_block(
     """Test execute block endpoint"""
     # Mock block
     mock_block = Mock()
-    mock_block.disabled = False
 
     async def mock_execute(*args, **kwargs):
         yield "output1", {"data": "result1"}

autogpt_platform/backend/backend/api/features/workspace/__init__.py

@@ -1 +0,0 @@
-# Workspace API feature module

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

@@ -1,122 +0,0 @@
-"""
-Workspace API routes for managing user file storage.
-"""
-
-import logging
-import re
-from typing import Annotated
-from urllib.parse import quote
-
-import fastapi
-from autogpt_libs.auth.dependencies import get_user_id, requires_user
-from fastapi.responses import Response
-
-from backend.data.workspace import get_workspace, get_workspace_file
-from backend.util.workspace_storage import get_workspace_storage
-
-
-def _sanitize_filename_for_header(filename: str) -> str:
-    """
-    Sanitize filename for Content-Disposition header to prevent header injection.
-
-    Removes/replaces characters that could break the header or inject new headers.
-    Uses RFC5987 encoding for non-ASCII characters.
-    """
-    # Remove CR, LF, and null bytes (header injection prevention)
-    sanitized = re.sub(r"[\r\n\x00]", "", filename)
-    # Escape quotes
-    sanitized = sanitized.replace('"', '\\"')
-    # For non-ASCII, use RFC5987 filename* parameter
-    # Check if filename has non-ASCII characters
-    try:
-        sanitized.encode("ascii")
-        return f'attachment; filename="{sanitized}"'
-    except UnicodeEncodeError:
-        # Use RFC5987 encoding for UTF-8 filenames
-        encoded = quote(sanitized, safe="")
-        return f"attachment; filename*=UTF-8''{encoded}"
-
-
-logger = logging.getLogger(__name__)
-
-router = fastapi.APIRouter(
-    dependencies=[fastapi.Security(requires_user)],
-)
-
-
-def _create_streaming_response(content: bytes, file) -> Response:
-    """Create a streaming response for file content."""
-    return Response(
-        content=content,
-        media_type=file.mimeType,
-        headers={
-            "Content-Disposition": _sanitize_filename_for_header(file.name),
-            "Content-Length": str(len(content)),
-        },
-    )
-
-
-async def _create_file_download_response(file) -> Response:
-    """
-    Create a download response for a workspace file.
-
-    Handles both local storage (direct streaming) and GCS (signed URL redirect
-    with fallback to streaming).
-    """
-    storage = await get_workspace_storage()
-
-    # For local storage, stream the file directly
-    if file.storagePath.startswith("local://"):
-        content = await storage.retrieve(file.storagePath)
-        return _create_streaming_response(content, file)
-
-    # For GCS, try to redirect to signed URL, fall back to streaming
-    try:
-        url = await storage.get_download_url(file.storagePath, expires_in=300)
-        # If we got back an API path (fallback), stream directly instead
-        if url.startswith("/api/"):
-            content = await storage.retrieve(file.storagePath)
-            return _create_streaming_response(content, file)
-        return fastapi.responses.RedirectResponse(url=url, status_code=302)
-    except Exception as e:
-        # Log the signed URL failure with context
-        logger.error(
-            f"Failed to get signed URL for file {file.id} "
-            f"(storagePath={file.storagePath}): {e}",
-            exc_info=True,
-        )
-        # Fall back to streaming directly from GCS
-        try:
-            content = await storage.retrieve(file.storagePath)
-            return _create_streaming_response(content, file)
-        except Exception as fallback_error:
-            logger.error(
-                f"Fallback streaming also failed for file {file.id} "
-                f"(storagePath={file.storagePath}): {fallback_error}",
-                exc_info=True,
-            )
-            raise
-
-
-@router.get(
-    "/files/{file_id}/download",
-    summary="Download file by ID",
-)
-async def download_file(
-    user_id: Annotated[str, fastapi.Security(get_user_id)],
-    file_id: str,
-) -> Response:
-    """
-    Download a file by its ID.
-
-    Returns the file content directly or redirects to a signed URL for GCS.
-    """
-    workspace = await get_workspace(user_id)
-    if workspace is None:
-        raise fastapi.HTTPException(status_code=404, detail="Workspace not found")
-
-    file = await get_workspace_file(file_id, workspace.id)
-    if file is None:
-        raise fastapi.HTTPException(status_code=404, detail="File not found")
-
-    return await _create_file_download_response(file)

autogpt_platform/backend/backend/api/rest_api.py

@@ -32,7 +32,6 @@ import backend.api.features.postmark.postmark
 import backend.api.features.store.model
 import backend.api.features.store.routes
 import backend.api.features.v1
-import backend.api.features.workspace.routes as workspace_routes
 import backend.data.block
 import backend.data.db
 import backend.data.graph
@@ -53,7 +52,6 @@ from backend.util.exceptions import (
 )
 from backend.util.feature_flag import initialize_launchdarkly, shutdown_launchdarkly
 from backend.util.service import UnhealthyServiceError
-from backend.util.workspace_storage import shutdown_workspace_storage
 
 from .external.fastapi_app import external_api
 from .features.analytics import router as analytics_router
@@ -126,11 +124,6 @@ async def lifespan_context(app: fastapi.FastAPI):
     except Exception as e:
         logger.warning(f"Error shutting down cloud storage handler: {e}")
 
-    try:
-        await shutdown_workspace_storage()
-    except Exception as e:
-        logger.warning(f"Error shutting down workspace storage: {e}")
-
     await backend.data.db.disconnect()
 
 
@@ -322,11 +315,6 @@ app.include_router(
     tags=["v2", "chat"],
     prefix="/api/chat",
 )
-app.include_router(
-    workspace_routes.router,
-    tags=["workspace"],
-    prefix="/api/workspace",
-)
 app.include_router(
     backend.api.features.oauth.router,
     tags=["oauth"],

autogpt_platform/backend/backend/api/ws_api.py

@@ -66,24 +66,18 @@ async def event_broadcaster(manager: ConnectionManager):
     execution_bus = AsyncRedisExecutionEventBus()
     notification_bus = AsyncRedisNotificationEventBus()
 
-    try:
+    async def execution_worker():
+        async for event in execution_bus.listen("*"):
+            await manager.send_execution_update(event)
 
-        async def execution_worker():
-            async for event in execution_bus.listen("*"):
-                await manager.send_execution_update(event)
+    async def notification_worker():
+        async for notification in notification_bus.listen("*"):
+            await manager.send_notification(
+                user_id=notification.user_id,
+                payload=notification.payload,
+            )
 
-        async def notification_worker():
-            async for notification in notification_bus.listen("*"):
-                await manager.send_notification(
-                    user_id=notification.user_id,
-                    payload=notification.payload,
-                )
-
-        await asyncio.gather(execution_worker(), notification_worker())
-    finally:
-        # Ensure PubSub connections are closed on any exit to prevent leaks
-        await execution_bus.close()
-        await notification_bus.close()
+    await asyncio.gather(execution_worker(), notification_worker())
 
 
 async def authenticate_websocket(websocket: WebSocket) -> str:

autogpt_platform/backend/backend/blocks/ai_image_customizer.py

@@ -13,7 +13,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import (
     APIKeyCredentials,
     CredentialsField,
@@ -118,13 +117,11 @@ class AIImageCustomizerBlock(Block):
                 "credentials": TEST_CREDENTIALS_INPUT,
             },
             test_output=[
-                # Output will be a workspace ref or data URI depending on context
-                ("image_url", lambda x: x.startswith(("workspace://", "data:"))),
+                ("image_url", "https://replicate.delivery/generated-image.jpg"),
             ],
             test_mock={
-                # Use data URI to avoid HTTP requests during tests
                 "run_model": lambda *args, **kwargs: MediaFileType(
-                    "data:image/jpeg;base64,/9j/4AAQSkZJRgABAgAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/2wBDAQkJCQwLDBgNDRgyIRwhMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjL/wAARCAABAAEDASIAAhEBAxEB/8QAHwAAAQUBAQEBAQEAAAAAAAAAAAECAwQFBgcICQoL/8QAtRAAAgEDAwIEAwUFBAQAAAF9AQIDAAQRBRIhMUEGE1FhByJxFDKBkaEII0KxwRVS0fAkM2JyggkKFhcYGRolJicoKSo0NTY3ODk6Q0RFRkdISUpTVFVWV1hZWmNkZWZnaGlqc3R1dnd4eXqDhIWGh4iJipKTlJWWl5iZmqKjpKWmp6ipqrKztLW2t7i5usLDxMXGx8jJytLT1NXW19jZ2uHi4+Tl5ufo6erx8vP09fb3+Pn6/8QAHwEAAwEBAQEBAQEBAQAAAAAAAAECAwQFBgcICQoL/8QAtREAAgECBAQDBAcFBAQAAQJ3AAECAxEEBSExBhJBUQdhcRMiMoEIFEKRobHBCSMzUvAVYnLRChYkNOEl8RcYGRomJygpKjU2Nzg5OkNERUZHSElKU1RVVldYWVpjZGVmZ2hpanN0dXZ3eHl6goOEhYaHiImKkpOUlZaXmJmaoqOkpaanqKmqsrO0tba3uLm6wsPExcbHyMnK0tPU1dbX2Nna4uPk5ebn6Onq8vP09fb3+Pn6/9oADAMBAAIRAxEAPwD3+iiigD//2Q=="
+                    "https://replicate.delivery/generated-image.jpg"
                 ),
             },
             test_credentials=TEST_CREDENTIALS,
@@ -135,7 +132,8 @@ class AIImageCustomizerBlock(Block):
         input_data: Input,
         *,
         credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         try:
@@ -143,9 +141,10 @@ class AIImageCustomizerBlock(Block):
             processed_images = await asyncio.gather(
                 *(
                     store_media_file(
+                        graph_exec_id=graph_exec_id,
                         file=img,
-                        execution_context=execution_context,
-                        return_format="for_external_api",  # Get content for Replicate API
+                        user_id=user_id,
+                        return_content=True,
                     )
                     for img in input_data.images
                 )
@@ -159,14 +158,7 @@ class AIImageCustomizerBlock(Block):
                 aspect_ratio=input_data.aspect_ratio.value,
                 output_format=input_data.output_format.value,
             )
-
-            # Store the generated image to the user's workspace for persistence
-            stored_url = await store_media_file(
-                file=result,
-                execution_context=execution_context,
-                return_format="for_block_output",
-            )
-            yield "image_url", stored_url
+            yield "image_url", result
         except Exception as e:
             yield "error", str(e)
 

autogpt_platform/backend/backend/blocks/ai_image_generator_block.py

@@ -6,7 +6,6 @@ from replicate.client import Client as ReplicateClient
 from replicate.helpers import FileOutput
 
 from backend.data.block import Block, BlockCategory, BlockSchemaInput, BlockSchemaOutput
-from backend.data.execution import ExecutionContext
 from backend.data.model import (
     APIKeyCredentials,
     CredentialsField,
@@ -14,8 +13,6 @@ from backend.data.model import (
     SchemaField,
 )
 from backend.integrations.providers import ProviderName
-from backend.util.file import store_media_file
-from backend.util.type import MediaFileType
 
 
 class ImageSize(str, Enum):
@@ -168,13 +165,11 @@ class AIImageGeneratorBlock(Block):
             test_output=[
                 (
                     "image_url",
-                    # Test output is a data URI since we now store images
-                    lambda x: x.startswith("data:image/"),
+                    "https://replicate.delivery/generated-image.webp",
                 ),
             ],
             test_mock={
-                # Return a data URI directly so store_media_file doesn't need to download
-                "_run_client": lambda *args, **kwargs: "data:image/webp;base64,UklGRiQAAABXRUJQVlA4IBgAAAAwAQCdASoBAAEAAQAcJYgCdAEO"
+                "_run_client": lambda *args, **kwargs: "https://replicate.delivery/generated-image.webp"
             },
         )
 
@@ -323,24 +318,11 @@ class AIImageGeneratorBlock(Block):
         style_text = style_map.get(style, "")
         return f"{style_text} of" if style_text else ""
 
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
-    ):
+    async def run(self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs):
         try:
             url = await self.generate_image(input_data, credentials)
             if url:
-                # Store the generated image to the user's workspace/execution folder
-                stored_url = await store_media_file(
-                    file=MediaFileType(url),
-                    execution_context=execution_context,
-                    return_format="for_block_output",
-                )
-                yield "image_url", stored_url
+                yield "image_url", url
             else:
                 yield "error", "Image generation returned an empty result."
         except Exception as e:

autogpt_platform/backend/backend/blocks/ai_shortform_video_block.py

@@ -13,7 +13,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import (
     APIKeyCredentials,
     CredentialsField,
@@ -22,9 +21,7 @@ from backend.data.model import (
 )
 from backend.integrations.providers import ProviderName
 from backend.util.exceptions import BlockExecutionError
-from backend.util.file import store_media_file
 from backend.util.request import Requests
-from backend.util.type import MediaFileType
 
 TEST_CREDENTIALS = APIKeyCredentials(
     id="01234567-89ab-cdef-0123-456789abcdef",
@@ -274,10 +271,7 @@ class AIShortformVideoCreatorBlock(Block):
                 "voice": Voice.LILY,
                 "video_style": VisualMediaType.STOCK_VIDEOS,
             },
-            test_output=(
-                "video_url",
-                lambda x: x.startswith(("workspace://", "data:")),
-            ),
+            test_output=("video_url", "https://example.com/video.mp4"),
             test_mock={
                 "create_webhook": lambda *args, **kwargs: (
                     "test_uuid",
@@ -286,21 +280,15 @@ class AIShortformVideoCreatorBlock(Block):
                 "create_video": lambda *args, **kwargs: {"pid": "test_pid"},
                 "check_video_status": lambda *args, **kwargs: {
                     "status": "ready",
-                    "videoUrl": "data:video/mp4;base64,AAAA",
+                    "videoUrl": "https://example.com/video.mp4",
                 },
-                # Use data URI to avoid HTTP requests during tests
-                "wait_for_video": lambda *args, **kwargs: "data:video/mp4;base64,AAAA",
+                "wait_for_video": lambda *args, **kwargs: "https://example.com/video.mp4",
             },
             test_credentials=TEST_CREDENTIALS,
         )
 
     async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
+        self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs
     ) -> BlockOutput:
         # Create a new Webhook.site URL
         webhook_token, webhook_url = await self.create_webhook()
@@ -352,13 +340,7 @@ class AIShortformVideoCreatorBlock(Block):
             )
             video_url = await self.wait_for_video(credentials.api_key, pid)
             logger.debug(f"Video ready: {video_url}")
-            # Store the generated video to the user's workspace for persistence
-            stored_url = await store_media_file(
-                file=MediaFileType(video_url),
-                execution_context=execution_context,
-                return_format="for_block_output",
-            )
-            yield "video_url", stored_url
+            yield "video_url", video_url
 
 
 class AIAdMakerVideoCreatorBlock(Block):
@@ -465,10 +447,7 @@ class AIAdMakerVideoCreatorBlock(Block):
                     "https://cdn.revid.ai/uploads/1747076315114-image.png",
                 ],
             },
-            test_output=(
-                "video_url",
-                lambda x: x.startswith(("workspace://", "data:")),
-            ),
+            test_output=("video_url", "https://example.com/ad.mp4"),
             test_mock={
                 "create_webhook": lambda *args, **kwargs: (
                     "test_uuid",
@@ -477,21 +456,14 @@ class AIAdMakerVideoCreatorBlock(Block):
                 "create_video": lambda *args, **kwargs: {"pid": "test_pid"},
                 "check_video_status": lambda *args, **kwargs: {
                     "status": "ready",
-                    "videoUrl": "data:video/mp4;base64,AAAA",
+                    "videoUrl": "https://example.com/ad.mp4",
                 },
-                "wait_for_video": lambda *args, **kwargs: "data:video/mp4;base64,AAAA",
+                "wait_for_video": lambda *args, **kwargs: "https://example.com/ad.mp4",
             },
             test_credentials=TEST_CREDENTIALS,
         )
 
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
-    ):
+    async def run(self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs):
         webhook_token, webhook_url = await self.create_webhook()
 
         payload = {
@@ -559,13 +531,7 @@ class AIAdMakerVideoCreatorBlock(Block):
             raise RuntimeError("Failed to create video: No project ID returned")
 
         video_url = await self.wait_for_video(credentials.api_key, pid)
-        # Store the generated video to the user's workspace for persistence
-        stored_url = await store_media_file(
-            file=MediaFileType(video_url),
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-        yield "video_url", stored_url
+        yield "video_url", video_url
 
 
 class AIScreenshotToVideoAdBlock(Block):
@@ -660,10 +626,7 @@ class AIScreenshotToVideoAdBlock(Block):
                 "script": "Amazing numbers!",
                 "screenshot_url": "https://cdn.revid.ai/uploads/1747080376028-image.png",
             },
-            test_output=(
-                "video_url",
-                lambda x: x.startswith(("workspace://", "data:")),
-            ),
+            test_output=("video_url", "https://example.com/screenshot.mp4"),
             test_mock={
                 "create_webhook": lambda *args, **kwargs: (
                     "test_uuid",
@@ -672,21 +635,14 @@ class AIScreenshotToVideoAdBlock(Block):
                 "create_video": lambda *args, **kwargs: {"pid": "test_pid"},
                 "check_video_status": lambda *args, **kwargs: {
                     "status": "ready",
-                    "videoUrl": "data:video/mp4;base64,AAAA",
+                    "videoUrl": "https://example.com/screenshot.mp4",
                 },
-                "wait_for_video": lambda *args, **kwargs: "data:video/mp4;base64,AAAA",
+                "wait_for_video": lambda *args, **kwargs: "https://example.com/screenshot.mp4",
             },
             test_credentials=TEST_CREDENTIALS,
         )
 
-    async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
-    ):
+    async def run(self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs):
         webhook_token, webhook_url = await self.create_webhook()
 
         payload = {
@@ -754,10 +710,4 @@ class AIScreenshotToVideoAdBlock(Block):
             raise RuntimeError("Failed to create video: No project ID returned")
 
         video_url = await self.wait_for_video(credentials.api_key, pid)
-        # Store the generated video to the user's workspace for persistence
-        stored_url = await store_media_file(
-            file=MediaFileType(video_url),
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-        yield "video_url", stored_url
+        yield "video_url", video_url

autogpt_platform/backend/backend/blocks/bannerbear/text_overlay.py

@@ -6,7 +6,6 @@ if TYPE_CHECKING:
 
 from pydantic import SecretStr
 
-from backend.data.execution import ExecutionContext
 from backend.sdk import (
     APIKeyCredentials,
     Block,
@@ -18,8 +17,6 @@ from backend.sdk import (
     Requests,
     SchemaField,
 )
-from backend.util.file import store_media_file
-from backend.util.type import MediaFileType
 
 from ._config import bannerbear
 
@@ -138,17 +135,15 @@ class BannerbearTextOverlayBlock(Block):
             },
             test_output=[
                 ("success", True),
-                # Output will be a workspace ref or data URI depending on context
-                ("image_url", lambda x: x.startswith(("workspace://", "data:"))),
+                ("image_url", "https://cdn.bannerbear.com/test-image.jpg"),
                 ("uid", "test-uid-123"),
                 ("status", "completed"),
             ],
             test_mock={
-                # Use data URI to avoid HTTP requests during tests
                 "_make_api_request": lambda *args, **kwargs: {
                     "uid": "test-uid-123",
                     "status": "completed",
-                    "image_url": "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRofHh0aHBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/wAALCAABAAEBAREA/8QAHwAAAQUBAQEBAQEAAAAAAAAAAAECAwQFBgcICQoL/8QAtRAAAgEDAwIEAwUFBAQAAAF9AQIDAAQRBRIhMUEGE1FhByJxFDKBkaEII0KxwRVS0fAkM2JyggkKFhcYGRolJicoKSo0NTY3ODk6Q0RFRkdISUpTVFVWV1hZWmNkZWZnaGlqc3R1dnd4eXqDhIWGh4iJipKTlJWWl5iZmqKjpKWmp6ipqrKztLW2t7i5usLDxMXGx8jJytLT1NXW19jZ2uHi4+Tl5ufo6erx8vP09fb3+Pn6/9oACAEBAAA/APn+v//Z",
+                    "image_url": "https://cdn.bannerbear.com/test-image.jpg",
                 }
             },
             test_credentials=TEST_CREDENTIALS,
@@ -182,12 +177,7 @@ class BannerbearTextOverlayBlock(Block):
             raise Exception(error_msg)
 
     async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
+        self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs
     ) -> BlockOutput:
         # Build the modifications array
         modifications = []
@@ -244,18 +234,6 @@ class BannerbearTextOverlayBlock(Block):
 
         # Synchronous request - image should be ready
         yield "success", True
-
-        # Store the generated image to workspace for persistence
-        image_url = data.get("image_url", "")
-        if image_url:
-            stored_url = await store_media_file(
-                file=MediaFileType(image_url),
-                execution_context=execution_context,
-                return_format="for_block_output",
-            )
-            yield "image_url", stored_url
-        else:
-            yield "image_url", ""
-
+        yield "image_url", data.get("image_url", "")
         yield "uid", data.get("uid", "")
         yield "status", data.get("status", "completed")

autogpt_platform/backend/backend/blocks/basic.py

@@ -9,7 +9,6 @@ from backend.data.block import (
     BlockSchemaOutput,
     BlockType,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import SchemaField
 from backend.util.file import store_media_file
 from backend.util.type import MediaFileType, convert
@@ -18,10 +17,10 @@ from backend.util.type import MediaFileType, convert
 class FileStoreBlock(Block):
     class Input(BlockSchemaInput):
         file_in: MediaFileType = SchemaField(
-            description="The file to download and store. Can be a URL (https://...), data URI, or local path."
+            description="The file to store in the temporary directory, it can be a URL, data URI, or local path."
         )
         base_64: bool = SchemaField(
-            description="Whether to produce output in base64 format (not recommended, you can pass the file reference across blocks).",
+            description="Whether produce an output in base64 format (not recommended, you can pass the string path just fine accross blocks).",
             default=False,
             advanced=True,
             title="Produce Base64 Output",
@@ -29,18 +28,13 @@ class FileStoreBlock(Block):
 
     class Output(BlockSchemaOutput):
         file_out: MediaFileType = SchemaField(
-            description="Reference to the stored file. In CoPilot: workspace:// URI (visible in list_workspace_files). In graphs: data URI for passing to other blocks."
+            description="The relative path to the stored file in the temporary directory."
         )
 
     def __init__(self):
         super().__init__(
             id="cbb50872-625b-42f0-8203-a2ae78242d8a",
-            description=(
-                "Downloads and stores a file from a URL, data URI, or local path. "
-                "Use this to fetch images, documents, or other files for processing. "
-                "In CoPilot: saves to workspace (use list_workspace_files to see it). "
-                "In graphs: outputs a data URI to pass to other blocks."
-            ),
+            description="Stores the input file in the temporary directory.",
             categories={BlockCategory.BASIC, BlockCategory.MULTIMEDIA},
             input_schema=FileStoreBlock.Input,
             output_schema=FileStoreBlock.Output,
@@ -51,18 +45,15 @@ class FileStoreBlock(Block):
         self,
         input_data: Input,
         *,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
-        # Determine return format based on user preference
-        # for_external_api: always returns data URI (base64) - honors "Produce Base64 Output"
-        # for_block_output: smart format - workspace:// in CoPilot, data URI in graphs
-        return_format = "for_external_api" if input_data.base_64 else "for_block_output"
-
         yield "file_out", await store_media_file(
+            graph_exec_id=graph_exec_id,
             file=input_data.file_in,
-            execution_context=execution_context,
-            return_format=return_format,
+            user_id=user_id,
+            return_content=input_data.base_64,
         )
 
 

autogpt_platform/backend/backend/blocks/discord/bot_blocks.py

@@ -15,7 +15,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import APIKeyCredentials, SchemaField
 from backend.util.file import store_media_file
 from backend.util.request import Requests
@@ -667,7 +666,8 @@ class SendDiscordFileBlock(Block):
         file: MediaFileType,
         filename: str,
         message_content: str,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
     ) -> dict:
         intents = discord.Intents.default()
         intents.guilds = True
@@ -731,9 +731,10 @@ class SendDiscordFileBlock(Block):
                     # Local file path - read from stored media file
                     # This would be a path from a previous block's output
                     stored_file = await store_media_file(
+                        graph_exec_id=graph_exec_id,
                         file=file,
-                        execution_context=execution_context,
-                        return_format="for_external_api",  # Get content to send to Discord
+                        user_id=user_id,
+                        return_content=True,  # Get as data URI
                     )
                     # Now process as data URI
                     header, encoded = stored_file.split(",", 1)
@@ -780,7 +781,8 @@ class SendDiscordFileBlock(Block):
         input_data: Input,
         *,
         credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         try:
@@ -791,7 +793,8 @@ class SendDiscordFileBlock(Block):
                 file=input_data.file,
                 filename=input_data.filename,
                 message_content=input_data.message_content,
-                execution_context=execution_context,
+                graph_exec_id=graph_exec_id,
+                user_id=user_id,
             )
 
             yield "status", result.get("status", "Unknown error")

autogpt_platform/backend/backend/blocks/fal/ai_video_generator.py

@@ -17,11 +17,8 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import SchemaField
-from backend.util.file import store_media_file
 from backend.util.request import ClientResponseError, Requests
-from backend.util.type import MediaFileType
 
 logger = logging.getLogger(__name__)
 
@@ -67,13 +64,9 @@ class AIVideoGeneratorBlock(Block):
                 "credentials": TEST_CREDENTIALS_INPUT,
             },
             test_credentials=TEST_CREDENTIALS,
-            test_output=[
-                # Output will be a workspace ref or data URI depending on context
-                ("video_url", lambda x: x.startswith(("workspace://", "data:"))),
-            ],
+            test_output=[("video_url", "https://fal.media/files/example/video.mp4")],
             test_mock={
-                # Use data URI to avoid HTTP requests during tests
-                "generate_video": lambda *args, **kwargs: "data:video/mp4;base64,AAAA"
+                "generate_video": lambda *args, **kwargs: "https://fal.media/files/example/video.mp4"
             },
         )
 
@@ -215,22 +208,11 @@ class AIVideoGeneratorBlock(Block):
             raise RuntimeError(f"API request failed: {str(e)}")
 
     async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: FalCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
+        self, input_data: Input, *, credentials: FalCredentials, **kwargs
     ) -> BlockOutput:
         try:
             video_url = await self.generate_video(input_data, credentials)
-            # Store the generated video to the user's workspace for persistence
-            stored_url = await store_media_file(
-                file=MediaFileType(video_url),
-                execution_context=execution_context,
-                return_format="for_block_output",
-            )
-            yield "video_url", stored_url
+            yield "video_url", video_url
         except Exception as e:
             error_message = str(e)
             yield "error", error_message

autogpt_platform/backend/backend/blocks/flux_kontext.py

@@ -12,7 +12,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import (
     APIKeyCredentials,
     CredentialsField,
@@ -122,12 +121,10 @@ class AIImageEditorBlock(Block):
                 "credentials": TEST_CREDENTIALS_INPUT,
             },
             test_output=[
-                # Output will be a workspace ref or data URI depending on context
-                ("output_image", lambda x: x.startswith(("workspace://", "data:"))),
+                ("output_image", "https://replicate.com/output/edited-image.png"),
             ],
             test_mock={
-                # Use data URI to avoid HTTP requests during tests
-                "run_model": lambda *args, **kwargs: "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==",
+                "run_model": lambda *args, **kwargs: "https://replicate.com/output/edited-image.png",
             },
             test_credentials=TEST_CREDENTIALS,
         )
@@ -137,7 +134,8 @@ class AIImageEditorBlock(Block):
         input_data: Input,
         *,
         credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         result = await self.run_model(
@@ -146,25 +144,20 @@ class AIImageEditorBlock(Block):
             prompt=input_data.prompt,
             input_image_b64=(
                 await store_media_file(
+                    graph_exec_id=graph_exec_id,
                     file=input_data.input_image,
-                    execution_context=execution_context,
-                    return_format="for_external_api",  # Get content for Replicate API
+                    user_id=user_id,
+                    return_content=True,
                 )
                 if input_data.input_image
                 else None
             ),
             aspect_ratio=input_data.aspect_ratio.value,
             seed=input_data.seed,
-            user_id=execution_context.user_id or "",
-            graph_exec_id=execution_context.graph_exec_id or "",
+            user_id=user_id,
+            graph_exec_id=graph_exec_id,
         )
-        # Store the generated image to the user's workspace for persistence
-        stored_url = await store_media_file(
-            file=result,
-            execution_context=execution_context,
-            return_format="for_block_output",
-        )
-        yield "output_image", stored_url
+        yield "output_image", result
 
     async def run_model(
         self,

autogpt_platform/backend/backend/blocks/google/gmail.py

@@ -21,7 +21,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import SchemaField
 from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
 from backend.util.settings import Settings
@@ -96,7 +95,8 @@ def _make_mime_text(
 
 async def create_mime_message(
     input_data,
-    execution_context: ExecutionContext,
+    graph_exec_id: str,
+    user_id: str,
 ) -> str:
     """Create a MIME message with attachments and return base64-encoded raw message."""
 
@@ -117,12 +117,12 @@ async def create_mime_message(
     if input_data.attachments:
         for attach in input_data.attachments:
             local_path = await store_media_file(
+                user_id=user_id,
+                graph_exec_id=graph_exec_id,
                 file=attach,
-                execution_context=execution_context,
-                return_format="for_local_processing",
+                return_content=False,
             )
-            assert execution_context.graph_exec_id  # Validated by store_media_file
-            abs_path = get_exec_file_path(execution_context.graph_exec_id, local_path)
+            abs_path = get_exec_file_path(graph_exec_id, local_path)
             part = MIMEBase("application", "octet-stream")
             with open(abs_path, "rb") as f:
                 part.set_payload(f.read())
@@ -582,25 +582,27 @@ class GmailSendBlock(GmailBase):
         input_data: Input,
         *,
         credentials: GoogleCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         service = self._build_service(credentials, **kwargs)
         result = await self._send_email(
             service,
             input_data,
-            execution_context,
+            graph_exec_id,
+            user_id,
         )
         yield "result", result
 
     async def _send_email(
-        self, service, input_data: Input, execution_context: ExecutionContext
+        self, service, input_data: Input, graph_exec_id: str, user_id: str
     ) -> dict:
         if not input_data.to or not input_data.subject or not input_data.body:
             raise ValueError(
                 "At least one recipient, subject, and body are required for sending an email"
             )
-        raw_message = await create_mime_message(input_data, execution_context)
+        raw_message = await create_mime_message(input_data, graph_exec_id, user_id)
         sent_message = await asyncio.to_thread(
             lambda: service.users()
             .messages()
@@ -690,28 +692,30 @@ class GmailCreateDraftBlock(GmailBase):
         input_data: Input,
         *,
         credentials: GoogleCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         service = self._build_service(credentials, **kwargs)
         result = await self._create_draft(
             service,
             input_data,
-            execution_context,
+            graph_exec_id,
+            user_id,
         )
         yield "result", GmailDraftResult(
             id=result["id"], message_id=result["message"]["id"], status="draft_created"
         )
 
     async def _create_draft(
-        self, service, input_data: Input, execution_context: ExecutionContext
+        self, service, input_data: Input, graph_exec_id: str, user_id: str
     ) -> dict:
         if not input_data.to or not input_data.subject:
             raise ValueError(
                 "At least one recipient and subject are required for creating a draft"
             )
 
-        raw_message = await create_mime_message(input_data, execution_context)
+        raw_message = await create_mime_message(input_data, graph_exec_id, user_id)
         draft = await asyncio.to_thread(
             lambda: service.users()
             .drafts()
@@ -1096,7 +1100,7 @@ class GmailGetThreadBlock(GmailBase):
 
 
 async def _build_reply_message(
-    service, input_data, execution_context: ExecutionContext
+    service, input_data, graph_exec_id: str, user_id: str
 ) -> tuple[str, str]:
     """
     Builds a reply MIME message for Gmail threads.
@@ -1186,12 +1190,12 @@ async def _build_reply_message(
     # Handle attachments
     for attach in input_data.attachments:
         local_path = await store_media_file(
+            user_id=user_id,
+            graph_exec_id=graph_exec_id,
             file=attach,
-            execution_context=execution_context,
-            return_format="for_local_processing",
+            return_content=False,
         )
-        assert execution_context.graph_exec_id  # Validated by store_media_file
-        abs_path = get_exec_file_path(execution_context.graph_exec_id, local_path)
+        abs_path = get_exec_file_path(graph_exec_id, local_path)
         part = MIMEBase("application", "octet-stream")
         with open(abs_path, "rb") as f:
             part.set_payload(f.read())
@@ -1307,14 +1311,16 @@ class GmailReplyBlock(GmailBase):
         input_data: Input,
         *,
         credentials: GoogleCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         service = self._build_service(credentials, **kwargs)
         message = await self._reply(
             service,
             input_data,
-            execution_context,
+            graph_exec_id,
+            user_id,
         )
         yield "messageId", message["id"]
         yield "threadId", message.get("threadId", input_data.threadId)
@@ -1337,11 +1343,11 @@ class GmailReplyBlock(GmailBase):
         yield "email", email
 
     async def _reply(
-        self, service, input_data: Input, execution_context: ExecutionContext
+        self, service, input_data: Input, graph_exec_id: str, user_id: str
     ) -> dict:
         # Build the reply message using the shared helper
         raw, thread_id = await _build_reply_message(
-            service, input_data, execution_context
+            service, input_data, graph_exec_id, user_id
         )
 
         # Send the message
@@ -1435,14 +1441,16 @@ class GmailDraftReplyBlock(GmailBase):
         input_data: Input,
         *,
         credentials: GoogleCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         service = self._build_service(credentials, **kwargs)
         draft = await self._create_draft_reply(
             service,
             input_data,
-            execution_context,
+            graph_exec_id,
+            user_id,
         )
         yield "draftId", draft["id"]
         yield "messageId", draft["message"]["id"]
@@ -1450,11 +1458,11 @@ class GmailDraftReplyBlock(GmailBase):
         yield "status", "draft_created"
 
     async def _create_draft_reply(
-        self, service, input_data: Input, execution_context: ExecutionContext
+        self, service, input_data: Input, graph_exec_id: str, user_id: str
     ) -> dict:
         # Build the reply message using the shared helper
         raw, thread_id = await _build_reply_message(
-            service, input_data, execution_context
+            service, input_data, graph_exec_id, user_id
         )
 
         # Create draft with proper thread association
@@ -1621,21 +1629,23 @@ class GmailForwardBlock(GmailBase):
         input_data: Input,
         *,
         credentials: GoogleCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         service = self._build_service(credentials, **kwargs)
         result = await self._forward_message(
             service,
             input_data,
-            execution_context,
+            graph_exec_id,
+            user_id,
         )
         yield "messageId", result["id"]
         yield "threadId", result.get("threadId", "")
         yield "status", "forwarded"
 
     async def _forward_message(
-        self, service, input_data: Input, execution_context: ExecutionContext
+        self, service, input_data: Input, graph_exec_id: str, user_id: str
     ) -> dict:
         if not input_data.to:
             raise ValueError("At least one recipient is required for forwarding")
@@ -1717,12 +1727,12 @@ To: {original_to}
         # Add any additional attachments
         for attach in input_data.additionalAttachments:
             local_path = await store_media_file(
+                user_id=user_id,
+                graph_exec_id=graph_exec_id,
                 file=attach,
-                execution_context=execution_context,
-                return_format="for_local_processing",
+                return_content=False,
             )
-            assert execution_context.graph_exec_id  # Validated by store_media_file
-            abs_path = get_exec_file_path(execution_context.graph_exec_id, local_path)
+            abs_path = get_exec_file_path(graph_exec_id, local_path)
             part = MIMEBase("application", "octet-stream")
             with open(abs_path, "rb") as f:
                 part.set_payload(f.read())

autogpt_platform/backend/backend/blocks/http.py

@@ -15,7 +15,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import (
     CredentialsField,
     CredentialsMetaInput,
@@ -117,9 +116,10 @@ class SendWebRequestBlock(Block):
 
     @staticmethod
     async def _prepare_files(
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
         files_name: str,
         files: list[MediaFileType],
+        user_id: str,
     ) -> list[tuple[str, tuple[str, BytesIO, str]]]:
         """
         Prepare files for the request by storing them and reading their content.
@@ -127,16 +127,11 @@ class SendWebRequestBlock(Block):
         (files_name, (filename, BytesIO, mime_type))
         """
         files_payload: list[tuple[str, tuple[str, BytesIO, str]]] = []
-        graph_exec_id = execution_context.graph_exec_id
-        if graph_exec_id is None:
-            raise ValueError("graph_exec_id is required for file operations")
 
         for media in files:
             # Normalise to a list so we can repeat the same key
             rel_path = await store_media_file(
-                file=media,
-                execution_context=execution_context,
-                return_format="for_local_processing",
+                graph_exec_id, media, user_id, return_content=False
             )
             abs_path = get_exec_file_path(graph_exec_id, rel_path)
             async with aiofiles.open(abs_path, "rb") as f:
@@ -148,7 +143,7 @@ class SendWebRequestBlock(Block):
         return files_payload
 
     async def run(
-        self, input_data: Input, *, execution_context: ExecutionContext, **kwargs
+        self, input_data: Input, *, graph_exec_id: str, user_id: str, **kwargs
     ) -> BlockOutput:
         # ─── Parse/normalise body ────────────────────────────────────
         body = input_data.body
@@ -179,7 +174,7 @@ class SendWebRequestBlock(Block):
         files_payload: list[tuple[str, tuple[str, BytesIO, str]]] = []
         if use_files:
             files_payload = await self._prepare_files(
-                execution_context, input_data.files_name, input_data.files
+                graph_exec_id, input_data.files_name, input_data.files, user_id
             )
 
         # Enforce body format rules
@@ -243,8 +238,9 @@ class SendAuthenticatedWebRequestBlock(SendWebRequestBlock):
         self,
         input_data: Input,
         *,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
         credentials: HostScopedCredentials,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         # Create SendWebRequestBlock.Input from our input (removing credentials field)
@@ -275,6 +271,6 @@ class SendAuthenticatedWebRequestBlock(SendWebRequestBlock):
 
         # Use parent class run method
         async for output_name, output_data in super().run(
-            base_input, execution_context=execution_context, **kwargs
+            base_input, graph_exec_id=graph_exec_id, user_id=user_id, **kwargs
         ):
             yield output_name, output_data

autogpt_platform/backend/backend/blocks/io.py

@@ -12,7 +12,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockType,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import SchemaField
 from backend.util.file import store_media_file
 from backend.util.mock import MockObject
@@ -463,21 +462,18 @@ class AgentFileInputBlock(AgentInputBlock):
         self,
         input_data: Input,
         *,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         if not input_data.value:
             return
 
-        # Determine return format based on user preference
-        # for_external_api: always returns data URI (base64) - honors "Produce Base64 Output"
-        # for_block_output: smart format - workspace:// in CoPilot, data URI in graphs
-        return_format = "for_external_api" if input_data.base_64 else "for_block_output"
-
         yield "result", await store_media_file(
+            graph_exec_id=graph_exec_id,
             file=input_data.value,
-            execution_context=execution_context,
-            return_format=return_format,
+            user_id=user_id,
+            return_content=input_data.base_64,
         )
 
 

autogpt_platform/backend/backend/blocks/llm.py

@@ -115,6 +115,7 @@ class LlmModel(str, Enum, metaclass=LlmModelMeta):
     CLAUDE_4_5_OPUS = "claude-opus-4-5-20251101"
     CLAUDE_4_5_SONNET = "claude-sonnet-4-5-20250929"
     CLAUDE_4_5_HAIKU = "claude-haiku-4-5-20251001"
+    CLAUDE_3_7_SONNET = "claude-3-7-sonnet-20250219"
     CLAUDE_3_HAIKU = "claude-3-haiku-20240307"
     # AI/ML API models
     AIML_API_QWEN2_5_72B = "Qwen/Qwen2.5-72B-Instruct-Turbo"
@@ -279,6 +280,9 @@ MODEL_METADATA = {
     LlmModel.CLAUDE_4_5_HAIKU: ModelMetadata(
         "anthropic", 200000, 64000, "Claude Haiku 4.5", "Anthropic", "Anthropic", 2
     ),  # claude-haiku-4-5-20251001
+    LlmModel.CLAUDE_3_7_SONNET: ModelMetadata(
+        "anthropic", 200000, 64000, "Claude 3.7 Sonnet", "Anthropic", "Anthropic", 2
+    ),  # claude-3-7-sonnet-20250219
     LlmModel.CLAUDE_3_HAIKU: ModelMetadata(
         "anthropic", 200000, 4096, "Claude 3 Haiku", "Anthropic", "Anthropic", 1
     ),  # claude-3-haiku-20240307

autogpt_platform/backend/backend/blocks/media.py

@@ -1,6 +1,6 @@
 import os
 import tempfile
-from typing import Optional
+from typing import Literal, Optional
 
 from moviepy.audio.io.AudioFileClip import AudioFileClip
 from moviepy.video.fx.Loop import Loop
@@ -13,7 +13,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import SchemaField
 from backend.util.file import MediaFileType, get_exec_file_path, store_media_file
 
@@ -47,19 +46,18 @@ class MediaDurationBlock(Block):
         self,
         input_data: Input,
         *,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         # 1) Store the input media locally
         local_media_path = await store_media_file(
+            graph_exec_id=graph_exec_id,
             file=input_data.media_in,
-            execution_context=execution_context,
-            return_format="for_local_processing",
-        )
-        assert execution_context.graph_exec_id is not None
-        media_abspath = get_exec_file_path(
-            execution_context.graph_exec_id, local_media_path
+            user_id=user_id,
+            return_content=False,
         )
+        media_abspath = get_exec_file_path(graph_exec_id, local_media_path)
 
         # 2) Load the clip
         if input_data.is_video:
@@ -90,6 +88,10 @@ class LoopVideoBlock(Block):
             default=None,
             ge=1,
         )
+        output_return_type: Literal["file_path", "data_uri"] = SchemaField(
+            description="How to return the output video. Either a relative path or base64 data URI.",
+            default="file_path",
+        )
 
     class Output(BlockSchemaOutput):
         video_out: str = SchemaField(
@@ -109,19 +111,17 @@ class LoopVideoBlock(Block):
         self,
         input_data: Input,
         *,
-        execution_context: ExecutionContext,
+        node_exec_id: str,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
-        assert execution_context.graph_exec_id is not None
-        assert execution_context.node_exec_id is not None
-        graph_exec_id = execution_context.graph_exec_id
-        node_exec_id = execution_context.node_exec_id
-
         # 1) Store the input video locally
         local_video_path = await store_media_file(
+            graph_exec_id=graph_exec_id,
             file=input_data.video_in,
-            execution_context=execution_context,
-            return_format="for_local_processing",
+            user_id=user_id,
+            return_content=False,
         )
         input_abspath = get_exec_file_path(graph_exec_id, local_video_path)
 
@@ -149,11 +149,12 @@ class LoopVideoBlock(Block):
         looped_clip = looped_clip.with_audio(clip.audio)
         looped_clip.write_videofile(output_abspath, codec="libx264", audio_codec="aac")
 
-        # Return output - for_block_output returns workspace:// if available, else data URI
+        # Return as data URI
         video_out = await store_media_file(
+            graph_exec_id=graph_exec_id,
             file=output_filename,
-            execution_context=execution_context,
-            return_format="for_block_output",
+            user_id=user_id,
+            return_content=input_data.output_return_type == "data_uri",
         )
 
         yield "video_out", video_out
@@ -176,6 +177,10 @@ class AddAudioToVideoBlock(Block):
             description="Volume scale for the newly attached audio track (1.0 = original).",
             default=1.0,
         )
+        output_return_type: Literal["file_path", "data_uri"] = SchemaField(
+            description="Return the final output as a relative path or base64 data URI.",
+            default="file_path",
+        )
 
     class Output(BlockSchemaOutput):
         video_out: MediaFileType = SchemaField(
@@ -195,24 +200,23 @@ class AddAudioToVideoBlock(Block):
         self,
         input_data: Input,
         *,
-        execution_context: ExecutionContext,
+        node_exec_id: str,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
-        assert execution_context.graph_exec_id is not None
-        assert execution_context.node_exec_id is not None
-        graph_exec_id = execution_context.graph_exec_id
-        node_exec_id = execution_context.node_exec_id
-
         # 1) Store the inputs locally
         local_video_path = await store_media_file(
+            graph_exec_id=graph_exec_id,
             file=input_data.video_in,
-            execution_context=execution_context,
-            return_format="for_local_processing",
+            user_id=user_id,
+            return_content=False,
         )
         local_audio_path = await store_media_file(
+            graph_exec_id=graph_exec_id,
             file=input_data.audio_in,
-            execution_context=execution_context,
-            return_format="for_local_processing",
+            user_id=user_id,
+            return_content=False,
         )
 
         abs_temp_dir = os.path.join(tempfile.gettempdir(), "exec_file", graph_exec_id)
@@ -236,11 +240,12 @@ class AddAudioToVideoBlock(Block):
         output_abspath = os.path.join(abs_temp_dir, output_filename)
         final_clip.write_videofile(output_abspath, codec="libx264", audio_codec="aac")
 
-        # 5) Return output - for_block_output returns workspace:// if available, else data URI
+        # 5) Return either path or data URI
         video_out = await store_media_file(
+            graph_exec_id=graph_exec_id,
             file=output_filename,
-            execution_context=execution_context,
-            return_format="for_block_output",
+            user_id=user_id,
+            return_content=input_data.output_return_type == "data_uri",
         )
 
         yield "video_out", video_out

autogpt_platform/backend/backend/blocks/screenshotone.py

@@ -11,7 +11,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import (
     APIKeyCredentials,
     CredentialsField,
@@ -113,7 +112,8 @@ class ScreenshotWebPageBlock(Block):
     @staticmethod
     async def take_screenshot(
         credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         url: str,
         viewport_width: int,
         viewport_height: int,
@@ -155,11 +155,12 @@ class ScreenshotWebPageBlock(Block):
 
         return {
             "image": await store_media_file(
+                graph_exec_id=graph_exec_id,
                 file=MediaFileType(
                     f"data:image/{format.value};base64,{b64encode(content).decode('utf-8')}"
                 ),
-                execution_context=execution_context,
-                return_format="for_block_output",
+                user_id=user_id,
+                return_content=True,
             )
         }
 
@@ -168,13 +169,15 @@ class ScreenshotWebPageBlock(Block):
         input_data: Input,
         *,
         credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
+        graph_exec_id: str,
+        user_id: str,
         **kwargs,
     ) -> BlockOutput:
         try:
             screenshot_data = await self.take_screenshot(
                 credentials=credentials,
-                execution_context=execution_context,
+                graph_exec_id=graph_exec_id,
+                user_id=user_id,
                 url=input_data.url,
                 viewport_width=input_data.viewport_width,
                 viewport_height=input_data.viewport_height,

autogpt_platform/backend/backend/blocks/spreadsheet.py

@@ -7,7 +7,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import ContributorDetails, SchemaField
 from backend.util.file import get_exec_file_path, store_media_file
 from backend.util.type import MediaFileType
@@ -99,7 +98,7 @@ class ReadSpreadsheetBlock(Block):
         )
 
     async def run(
-        self, input_data: Input, *, execution_context: ExecutionContext, **_kwargs
+        self, input_data: Input, *, graph_exec_id: str, user_id: str, **_kwargs
     ) -> BlockOutput:
         import csv
         from io import StringIO
@@ -107,16 +106,14 @@ class ReadSpreadsheetBlock(Block):
         # Determine data source - prefer file_input if provided, otherwise use contents
         if input_data.file_input:
             stored_file_path = await store_media_file(
+                user_id=user_id,
+                graph_exec_id=graph_exec_id,
                 file=input_data.file_input,
-                execution_context=execution_context,
-                return_format="for_local_processing",
+                return_content=False,
             )
 
             # Get full file path
-            assert execution_context.graph_exec_id  # Validated by store_media_file
-            file_path = get_exec_file_path(
-                execution_context.graph_exec_id, stored_file_path
-            )
+            file_path = get_exec_file_path(graph_exec_id, stored_file_path)
             if not Path(file_path).exists():
                 raise ValueError(f"File does not exist: {file_path}")
 

autogpt_platform/backend/backend/blocks/stagehand/blocks.py

@@ -83,7 +83,7 @@ class StagehandRecommendedLlmModel(str, Enum):
     GPT41_MINI = "gpt-4.1-mini-2025-04-14"
 
     # Anthropic
-    CLAUDE_4_5_SONNET = "claude-sonnet-4-5-20250929"
+    CLAUDE_3_7_SONNET = "claude-3-7-sonnet-20250219"
 
     @property
     def provider_name(self) -> str:
@@ -137,7 +137,7 @@ class StagehandObserveBlock(Block):
         model: StagehandRecommendedLlmModel = SchemaField(
             title="LLM Model",
             description="LLM to use for Stagehand (provider is inferred)",
-            default=StagehandRecommendedLlmModel.CLAUDE_4_5_SONNET,
+            default=StagehandRecommendedLlmModel.CLAUDE_3_7_SONNET,
             advanced=False,
         )
         model_credentials: AICredentials = AICredentialsField()
@@ -230,7 +230,7 @@ class StagehandActBlock(Block):
         model: StagehandRecommendedLlmModel = SchemaField(
             title="LLM Model",
             description="LLM to use for Stagehand (provider is inferred)",
-            default=StagehandRecommendedLlmModel.CLAUDE_4_5_SONNET,
+            default=StagehandRecommendedLlmModel.CLAUDE_3_7_SONNET,
             advanced=False,
         )
         model_credentials: AICredentials = AICredentialsField()
@@ -330,7 +330,7 @@ class StagehandExtractBlock(Block):
         model: StagehandRecommendedLlmModel = SchemaField(
             title="LLM Model",
             description="LLM to use for Stagehand (provider is inferred)",
-            default=StagehandRecommendedLlmModel.CLAUDE_4_5_SONNET,
+            default=StagehandRecommendedLlmModel.CLAUDE_3_7_SONNET,
             advanced=False,
         )
         model_credentials: AICredentials = AICredentialsField()

autogpt_platform/backend/backend/blocks/talking_head.py

@@ -10,7 +10,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import (
     APIKeyCredentials,
     CredentialsField,
@@ -18,9 +17,7 @@ from backend.data.model import (
     SchemaField,
 )
 from backend.integrations.providers import ProviderName
-from backend.util.file import store_media_file
 from backend.util.request import Requests
-from backend.util.type import MediaFileType
 
 TEST_CREDENTIALS = APIKeyCredentials(
     id="01234567-89ab-cdef-0123-456789abcdef",
@@ -105,7 +102,7 @@ class CreateTalkingAvatarVideoBlock(Block):
             test_output=[
                 (
                     "video_url",
-                    lambda x: x.startswith(("workspace://", "data:")),
+                    "https://d-id.com/api/clips/abcd1234-5678-efgh-ijkl-mnopqrstuvwx/video",
                 ),
             ],
             test_mock={
@@ -113,10 +110,9 @@ class CreateTalkingAvatarVideoBlock(Block):
                     "id": "abcd1234-5678-efgh-ijkl-mnopqrstuvwx",
                     "status": "created",
                 },
-                # Use data URI to avoid HTTP requests during tests
                 "get_clip_status": lambda *args, **kwargs: {
                     "status": "done",
-                    "result_url": "data:video/mp4;base64,AAAA",
+                    "result_url": "https://d-id.com/api/clips/abcd1234-5678-efgh-ijkl-mnopqrstuvwx/video",
                 },
             },
             test_credentials=TEST_CREDENTIALS,
@@ -142,12 +138,7 @@ class CreateTalkingAvatarVideoBlock(Block):
         return response.json()
 
     async def run(
-        self,
-        input_data: Input,
-        *,
-        credentials: APIKeyCredentials,
-        execution_context: ExecutionContext,
-        **kwargs,
+        self, input_data: Input, *, credentials: APIKeyCredentials, **kwargs
     ) -> BlockOutput:
         # Create the clip
         payload = {
@@ -174,14 +165,7 @@ class CreateTalkingAvatarVideoBlock(Block):
         for _ in range(input_data.max_polling_attempts):
             status_response = await self.get_clip_status(credentials.api_key, clip_id)
             if status_response["status"] == "done":
-                # Store the generated video to the user's workspace for persistence
-                video_url = status_response["result_url"]
-                stored_url = await store_media_file(
-                    file=MediaFileType(video_url),
-                    execution_context=execution_context,
-                    return_format="for_block_output",
-                )
-                yield "video_url", stored_url
+                yield "video_url", status_response["result_url"]
                 return
             elif status_response["status"] == "error":
                 raise RuntimeError(

autogpt_platform/backend/backend/blocks/test/test_blocks_dos_vulnerability.py

@@ -12,7 +12,6 @@ from backend.blocks.iteration import StepThroughItemsBlock
 from backend.blocks.llm import AITextSummarizerBlock
 from backend.blocks.text import ExtractTextInformationBlock
 from backend.blocks.xml_parser import XMLParserBlock
-from backend.data.execution import ExecutionContext
 from backend.util.file import store_media_file
 from backend.util.type import MediaFileType
 
@@ -234,12 +233,9 @@ class TestStoreMediaFileSecurity:
 
         with pytest.raises(ValueError, match="File too large"):
             await store_media_file(
+                graph_exec_id="test",
                 file=MediaFileType(large_data_uri),
-                execution_context=ExecutionContext(
-                    user_id="test_user",
-                    graph_exec_id="test",
-                ),
-                return_format="for_local_processing",
+                user_id="test_user",
             )
 
     @patch("backend.util.file.Path")
@@ -274,12 +270,9 @@ class TestStoreMediaFileSecurity:
         # Should raise an error when directory size exceeds limit
         with pytest.raises(ValueError, match="Disk usage limit exceeded"):
             await store_media_file(
+                graph_exec_id="test",
                 file=MediaFileType(
                     "data:text/plain;base64,dGVzdA=="
                 ),  # Small test file
-                execution_context=ExecutionContext(
-                    user_id="test_user",
-                    graph_exec_id="test",
-                ),
-                return_format="for_local_processing",
+                user_id="test_user",
             )

autogpt_platform/backend/backend/blocks/test/test_http.py

@@ -11,22 +11,10 @@ from backend.blocks.http import (
     HttpMethod,
     SendAuthenticatedWebRequestBlock,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import HostScopedCredentials
 from backend.util.request import Response
 
 
-def make_test_context(
-    graph_exec_id: str = "test-exec-id",
-    user_id: str = "test-user-id",
-) -> ExecutionContext:
-    """Helper to create test ExecutionContext."""
-    return ExecutionContext(
-        user_id=user_id,
-        graph_exec_id=graph_exec_id,
-    )
-
-
 class TestHttpBlockWithHostScopedCredentials:
     """Test suite for HTTP block integration with HostScopedCredentials."""
 
@@ -117,7 +105,8 @@ class TestHttpBlockWithHostScopedCredentials:
         async for output_name, output_data in http_block.run(
             input_data,
             credentials=exact_match_credentials,
-            execution_context=make_test_context(),
+            graph_exec_id="test-exec-id",
+            user_id="test-user-id",
         ):
             result.append((output_name, output_data))
 
@@ -172,7 +161,8 @@ class TestHttpBlockWithHostScopedCredentials:
         async for output_name, output_data in http_block.run(
             input_data,
             credentials=wildcard_credentials,
-            execution_context=make_test_context(),
+            graph_exec_id="test-exec-id",
+            user_id="test-user-id",
         ):
             result.append((output_name, output_data))
 
@@ -218,7 +208,8 @@ class TestHttpBlockWithHostScopedCredentials:
         async for output_name, output_data in http_block.run(
             input_data,
             credentials=non_matching_credentials,
-            execution_context=make_test_context(),
+            graph_exec_id="test-exec-id",
+            user_id="test-user-id",
         ):
             result.append((output_name, output_data))
 
@@ -267,7 +258,8 @@ class TestHttpBlockWithHostScopedCredentials:
         async for output_name, output_data in http_block.run(
             input_data,
             credentials=exact_match_credentials,
-            execution_context=make_test_context(),
+            graph_exec_id="test-exec-id",
+            user_id="test-user-id",
         ):
             result.append((output_name, output_data))
 
@@ -326,7 +318,8 @@ class TestHttpBlockWithHostScopedCredentials:
         async for output_name, output_data in http_block.run(
             input_data,
             credentials=auto_discovered_creds,  # Execution manager found these
-            execution_context=make_test_context(),
+            graph_exec_id="test-exec-id",
+            user_id="test-user-id",
         ):
             result.append((output_name, output_data))
 
@@ -389,7 +382,8 @@ class TestHttpBlockWithHostScopedCredentials:
         async for output_name, output_data in http_block.run(
             input_data,
             credentials=multi_header_creds,
-            execution_context=make_test_context(),
+            graph_exec_id="test-exec-id",
+            user_id="test-user-id",
         ):
             result.append((output_name, output_data))
 
@@ -477,7 +471,8 @@ class TestHttpBlockWithHostScopedCredentials:
             async for output_name, output_data in http_block.run(
                 input_data,
                 credentials=test_creds,
-                execution_context=make_test_context(),
+                graph_exec_id="test-exec-id",
+                user_id="test-user-id",
             ):
                 result.append((output_name, output_data))
 

autogpt_platform/backend/backend/blocks/text.py

@@ -11,7 +11,6 @@ from backend.data.block import (
     BlockSchemaInput,
     BlockSchemaOutput,
 )
-from backend.data.execution import ExecutionContext
 from backend.data.model import SchemaField
 from backend.util import json, text
 from backend.util.file import get_exec_file_path, store_media_file
@@ -445,21 +444,18 @@ class FileReadBlock(Block):
         )
 
     async def run(
-        self, input_data: Input, *, execution_context: ExecutionContext, **_kwargs
+        self, input_data: Input, *, graph_exec_id: str, user_id: str, **_kwargs
     ) -> BlockOutput:
         # Store the media file properly (handles URLs, data URIs, etc.)
         stored_file_path = await store_media_file(
+            user_id=user_id,
+            graph_exec_id=graph_exec_id,
             file=input_data.file_input,
-            execution_context=execution_context,
-            return_format="for_local_processing",
+            return_content=False,
         )
 
-        # Get full file path (graph_exec_id validated by store_media_file above)
-        if not execution_context.graph_exec_id:
-            raise ValueError("execution_context.graph_exec_id is required")
-        file_path = get_exec_file_path(
-            execution_context.graph_exec_id, stored_file_path
-        )
+        # Get full file path
+        file_path = get_exec_file_path(graph_exec_id, stored_file_path)
 
         if not Path(file_path).exists():
             raise ValueError(f"File does not exist: {file_path}")

autogpt_platform/backend/backend/data/block_cost_config.py

@@ -81,6 +81,7 @@ MODEL_COST: dict[LlmModel, int] = {
     LlmModel.CLAUDE_4_5_HAIKU: 4,
     LlmModel.CLAUDE_4_5_OPUS: 14,
     LlmModel.CLAUDE_4_5_SONNET: 9,
+    LlmModel.CLAUDE_3_7_SONNET: 5,
     LlmModel.CLAUDE_3_HAIKU: 1,
     LlmModel.AIML_API_QWEN2_5_72B: 1,
     LlmModel.AIML_API_LLAMA3_1_70B: 1,

autogpt_platform/backend/backend/data/event_bus.py

@@ -133,23 +133,10 @@ class RedisEventBus(BaseRedisEventBus[M], ABC):
 
 
 class AsyncRedisEventBus(BaseRedisEventBus[M], ABC):
-    def __init__(self):
-        self._pubsub: AsyncPubSub | None = None
-
     @property
     async def connection(self) -> redis.AsyncRedis:
         return await redis.get_redis_async()
 
-    async def close(self) -> None:
-        """Close the PubSub connection if it exists."""
-        if self._pubsub is not None:
-            try:
-                await self._pubsub.close()
-            except Exception:
-                logger.warning("Failed to close PubSub connection", exc_info=True)
-            finally:
-                self._pubsub = None
-
     async def publish_event(self, event: M, channel_key: str):
         """
         Publish an event to Redis. Gracefully handles connection failures
@@ -170,7 +157,6 @@ class AsyncRedisEventBus(BaseRedisEventBus[M], ABC):
             await self.connection, channel_key
         )
         assert isinstance(pubsub, AsyncPubSub)
-        self._pubsub = pubsub
 
         if "*" in channel_key:
             await pubsub.psubscribe(full_channel_name)

autogpt_platform/backend/backend/data/execution.py

@@ -83,29 +83,12 @@ class ExecutionContext(BaseModel):
 
     model_config = {"extra": "ignore"}
 
-    # Execution identity
-    user_id: Optional[str] = None
-    graph_id: Optional[str] = None
-    graph_exec_id: Optional[str] = None
-    graph_version: Optional[int] = None
-    node_id: Optional[str] = None
-    node_exec_id: Optional[str] = None
-
-    # Safety settings
     human_in_the_loop_safe_mode: bool = True
     sensitive_action_safe_mode: bool = False
-
-    # User settings
     user_timezone: str = "UTC"
-
-    # Execution hierarchy
     root_execution_id: Optional[str] = None
     parent_execution_id: Optional[str] = None
 
-    # Workspace
-    workspace_id: Optional[str] = None
-    session_id: Optional[str] = None
-
 
 # -------------------------- Models -------------------------- #
 

autogpt_platform/backend/backend/data/graph.py

@@ -1028,39 +1028,6 @@ async def get_graph(
     return GraphModel.from_db(graph, for_export)
 
 
-async def get_store_listed_graphs(*graph_ids: str) -> dict[str, GraphModel]:
-    """Batch-fetch multiple store-listed graphs by their IDs.
-
-    Only returns graphs that have approved store listings (publicly available).
-    Does not require permission checks since store-listed graphs are public.
-
-    Args:
-        *graph_ids: Variable number of graph IDs to fetch
-
-    Returns:
-        Dict mapping graph_id to GraphModel for graphs with approved store listings
-    """
-    if not graph_ids:
-        return {}
-
-    store_listings = await StoreListingVersion.prisma().find_many(
-        where={
-            "agentGraphId": {"in": list(graph_ids)},
-            "submissionStatus": SubmissionStatus.APPROVED,
-            "isDeleted": False,
-        },
-        include={"AgentGraph": {"include": AGENT_GRAPH_INCLUDE}},
-        distinct=["agentGraphId"],
-        order={"agentGraphVersion": "desc"},
-    )
-
-    return {
-        listing.agentGraphId: GraphModel.from_db(listing.AgentGraph)
-        for listing in store_listings
-        if listing.AgentGraph
-    }
-
-
 async def get_graph_as_admin(
     graph_id: str,
     version: int | None = None,

autogpt_platform/backend/backend/data/human_review.py

@@ -263,14 +263,11 @@ async def get_pending_review_by_node_exec_id(
     return PendingHumanReviewModel.from_db(review, node_id=node_id)
 
 
-async def get_reviews_by_node_exec_ids(
+async def get_pending_reviews_by_node_exec_ids(
     node_exec_ids: list[str], user_id: str
 ) -> dict[str, "PendingHumanReviewModel"]:
     """
-    Get multiple reviews by their node execution IDs regardless of status.
-
-    Unlike get_pending_reviews_by_node_exec_ids, this returns reviews in any status
-    (WAITING, APPROVED, REJECTED). Used for validation in idempotent operations.
+    Get multiple pending reviews by their node execution IDs in a single batch query.
 
     Args:
         node_exec_ids: List of node execution IDs to look up
@@ -286,6 +283,7 @@ async def get_reviews_by_node_exec_ids(
         where={
             "nodeExecId": {"in": node_exec_ids},
             "userId": user_id,
+            "status": ReviewStatus.WAITING,
         }
     )
 
@@ -409,68 +407,38 @@ async def process_all_reviews_for_execution(
 ) -> dict[str, PendingHumanReviewModel]:
     """Process all pending reviews for an execution with approve/reject decisions.
 
-    Handles race conditions gracefully: if a review was already processed with the
-    same decision by a concurrent request, it's treated as success rather than error.
-
     Args:
         user_id: User ID for ownership validation
         review_decisions: Map of node_exec_id -> (status, reviewed_data, message)
 
     Returns:
-        Dict of node_exec_id -> updated review model (includes already-processed reviews)
+        Dict of node_exec_id -> updated review model
     """
     if not review_decisions:
         return {}
 
     node_exec_ids = list(review_decisions.keys())
 
-    # Get all reviews (both WAITING and already processed) for the user
-    all_reviews = await PendingHumanReview.prisma().find_many(
+    # Get all reviews for validation
+    reviews = await PendingHumanReview.prisma().find_many(
         where={
             "nodeExecId": {"in": node_exec_ids},
             "userId": user_id,
+            "status": ReviewStatus.WAITING,
         },
     )
 
-    # Separate into pending and already-processed reviews
-    reviews_to_process = []
-    already_processed = []
-    for review in all_reviews:
-        if review.status == ReviewStatus.WAITING:
-            reviews_to_process.append(review)
-        else:
-            already_processed.append(review)
-
-    # Check for truly missing reviews (not found at all)
-    found_ids = {review.nodeExecId for review in all_reviews}
-    missing_ids = set(node_exec_ids) - found_ids
-    if missing_ids:
+    # Validate all reviews can be processed
+    if len(reviews) != len(node_exec_ids):
+        missing_ids = set(node_exec_ids) - {review.nodeExecId for review in reviews}
         raise ValueError(
-            f"Reviews not found or access denied: {', '.join(missing_ids)}"
+            f"Reviews not found, access denied, or not in WAITING status: {', '.join(missing_ids)}"
         )
 
-    # Validate already-processed reviews have compatible status (same decision)
-    # This handles race conditions where another request processed the same reviews
-    for review in already_processed:
-        requested_status = review_decisions[review.nodeExecId][0]
-        if review.status != requested_status:
-            raise ValueError(
-                f"Review {review.nodeExecId} was already processed with status "
-                f"{review.status}, cannot change to {requested_status}"
-            )
-
-    # Log if we're handling a race condition (some reviews already processed)
-    if already_processed:
-        already_processed_ids = [r.nodeExecId for r in already_processed]
-        logger.info(
-            f"Race condition handled: {len(already_processed)} review(s) already "
-            f"processed by concurrent request: {already_processed_ids}"
-        )
-
-    # Create parallel update tasks for reviews that still need processing
+    # Create parallel update tasks
     update_tasks = []
 
-    for review in reviews_to_process:
+    for review in reviews:
         new_status, reviewed_data, message = review_decisions[review.nodeExecId]
         has_data_changes = reviewed_data is not None and reviewed_data != review.payload
 
@@ -495,7 +463,7 @@ async def process_all_reviews_for_execution(
         update_tasks.append(task)
 
     # Execute all updates in parallel and get updated reviews
-    updated_reviews = await asyncio.gather(*update_tasks) if update_tasks else []
+    updated_reviews = await asyncio.gather(*update_tasks)
 
     # Note: Execution resumption is now handled at the API layer after ALL reviews
     # for an execution are processed (both approved and rejected)
@@ -504,11 +472,8 @@ async def process_all_reviews_for_execution(
     # Local import to avoid event loop conflicts in tests
     from backend.data.execution import get_node_execution
 
-    # Combine updated reviews with already-processed ones (for idempotent response)
-    all_result_reviews = list(updated_reviews) + already_processed
-
     result = {}
-    for review in all_result_reviews:
+    for review in updated_reviews:
         node_exec = await get_node_execution(review.nodeExecId)
         node_id = node_exec.node_id if node_exec else review.nodeExecId
         result[review.nodeExecId] = PendingHumanReviewModel.from_db(

autogpt_platform/backend/backend/data/model.py

@@ -666,16 +666,10 @@ class CredentialsFieldInfo(BaseModel, Generic[CP, CT]):
         if not (self.discriminator and self.discriminator_mapping):
             return self
 
-        try:
-            provider = self.discriminator_mapping[discriminator_value]
-        except KeyError:
-            raise ValueError(
-                f"Model '{discriminator_value}' is not supported. "
-                "It may have been deprecated. Please update your agent configuration."
-            )
-
         return CredentialsFieldInfo(
-            credentials_provider=frozenset([provider]),
+            credentials_provider=frozenset(
+                [self.discriminator_mapping[discriminator_value]]
+            ),
             credentials_types=self.supported_types,
             credentials_scopes=self.required_scopes,
             discriminator=self.discriminator,

autogpt_platform/backend/backend/data/onboarding.py

@@ -41,7 +41,6 @@ FrontendOnboardingStep = Literal[
     OnboardingStep.AGENT_NEW_RUN,
     OnboardingStep.AGENT_INPUT,
     OnboardingStep.CONGRATS,
-    OnboardingStep.VISIT_COPILOT,
     OnboardingStep.MARKETPLACE_VISIT,
     OnboardingStep.BUILDER_OPEN,
 ]
@@ -123,9 +122,6 @@ async def update_user_onboarding(user_id: str, data: UserOnboardingUpdate):
 async def _reward_user(user_id: str, onboarding: UserOnboarding, step: OnboardingStep):
     reward = 0
     match step:
-        # Welcome bonus for visiting copilot ($5 = 500 credits)
-        case OnboardingStep.VISIT_COPILOT:
-            reward = 500
         # Reward user when they clicked New Run during onboarding
         # This is because they need credits before scheduling a run (next step)
         # This is seen as a reward for the GET_RESULTS step in the wallet

autogpt_platform/backend/backend/data/workspace.py

@@ -1,276 +0,0 @@
-"""
-Database CRUD operations for User Workspace.
-
-This module provides functions for managing user workspaces and workspace files.
-"""
-
-import logging
-from datetime import datetime, timezone
-from typing import Optional
-
-from prisma.models import UserWorkspace, UserWorkspaceFile
-from prisma.types import UserWorkspaceFileWhereInput
-
-from backend.util.json import SafeJson
-
-logger = logging.getLogger(__name__)
-
-
-async def get_or_create_workspace(user_id: str) -> UserWorkspace:
-    """
-    Get user's workspace, creating one if it doesn't exist.
-
-    Uses upsert to handle race conditions when multiple concurrent requests
-    attempt to create a workspace for the same user.
-
-    Args:
-        user_id: The user's ID
-
-    Returns:
-        UserWorkspace instance
-    """
-    workspace = await UserWorkspace.prisma().upsert(
-        where={"userId": user_id},
-        data={
-            "create": {"userId": user_id},
-            "update": {},  # No updates needed if exists
-        },
-    )
-
-    return workspace
-
-
-async def get_workspace(user_id: str) -> Optional[UserWorkspace]:
-    """
-    Get user's workspace if it exists.
-
-    Args:
-        user_id: The user's ID
-
-    Returns:
-        UserWorkspace instance or None
-    """
-    return await UserWorkspace.prisma().find_unique(where={"userId": user_id})
-
-
-async def create_workspace_file(
-    workspace_id: str,
-    file_id: str,
-    name: str,
-    path: str,
-    storage_path: str,
-    mime_type: str,
-    size_bytes: int,
-    checksum: Optional[str] = None,
-    metadata: Optional[dict] = None,
-) -> UserWorkspaceFile:
-    """
-    Create a new workspace file record.
-
-    Args:
-        workspace_id: The workspace ID
-        file_id: The file ID (same as used in storage path for consistency)
-        name: User-visible filename
-        path: Virtual path (e.g., "/documents/report.pdf")
-        storage_path: Actual storage path (GCS or local)
-        mime_type: MIME type of the file
-        size_bytes: File size in bytes
-        checksum: Optional SHA256 checksum
-        metadata: Optional additional metadata
-
-    Returns:
-        Created UserWorkspaceFile instance
-    """
-    # Normalize path to start with /
-    if not path.startswith("/"):
-        path = f"/{path}"
-
-    file = await UserWorkspaceFile.prisma().create(
-        data={
-            "id": file_id,
-            "workspaceId": workspace_id,
-            "name": name,
-            "path": path,
-            "storagePath": storage_path,
-            "mimeType": mime_type,
-            "sizeBytes": size_bytes,
-            "checksum": checksum,
-            "metadata": SafeJson(metadata or {}),
-        }
-    )
-
-    logger.info(
-        f"Created workspace file {file.id} at path {path} "
-        f"in workspace {workspace_id}"
-    )
-    return file
-
-
-async def get_workspace_file(
-    file_id: str,
-    workspace_id: Optional[str] = None,
-) -> Optional[UserWorkspaceFile]:
-    """
-    Get a workspace file by ID.
-
-    Args:
-        file_id: The file ID
-        workspace_id: Optional workspace ID for validation
-
-    Returns:
-        UserWorkspaceFile instance or None
-    """
-    where_clause: dict = {"id": file_id, "isDeleted": False}
-    if workspace_id:
-        where_clause["workspaceId"] = workspace_id
-
-    return await UserWorkspaceFile.prisma().find_first(where=where_clause)
-
-
-async def get_workspace_file_by_path(
-    workspace_id: str,
-    path: str,
-) -> Optional[UserWorkspaceFile]:
-    """
-    Get a workspace file by its virtual path.
-
-    Args:
-        workspace_id: The workspace ID
-        path: Virtual path
-
-    Returns:
-        UserWorkspaceFile instance or None
-    """
-    # Normalize path
-    if not path.startswith("/"):
-        path = f"/{path}"
-
-    return await UserWorkspaceFile.prisma().find_first(
-        where={
-            "workspaceId": workspace_id,
-            "path": path,
-            "isDeleted": False,
-        }
-    )
-
-
-async def list_workspace_files(
-    workspace_id: str,
-    path_prefix: Optional[str] = None,
-    include_deleted: bool = False,
-    limit: Optional[int] = None,
-    offset: int = 0,
-) -> list[UserWorkspaceFile]:
-    """
-    List files in a workspace.
-
-    Args:
-        workspace_id: The workspace ID
-        path_prefix: Optional path prefix to filter (e.g., "/documents/")
-        include_deleted: Whether to include soft-deleted files
-        limit: Maximum number of files to return
-        offset: Number of files to skip
-
-    Returns:
-        List of UserWorkspaceFile instances
-    """
-    where_clause: UserWorkspaceFileWhereInput = {"workspaceId": workspace_id}
-
-    if not include_deleted:
-        where_clause["isDeleted"] = False
-
-    if path_prefix:
-        # Normalize prefix
-        if not path_prefix.startswith("/"):
-            path_prefix = f"/{path_prefix}"
-        where_clause["path"] = {"startswith": path_prefix}
-
-    return await UserWorkspaceFile.prisma().find_many(
-        where=where_clause,
-        order={"createdAt": "desc"},
-        take=limit,
-        skip=offset,
-    )
-
-
-async def count_workspace_files(
-    workspace_id: str,
-    path_prefix: Optional[str] = None,
-    include_deleted: bool = False,
-) -> int:
-    """
-    Count files in a workspace.
-
-    Args:
-        workspace_id: The workspace ID
-        path_prefix: Optional path prefix to filter (e.g., "/sessions/abc123/")
-        include_deleted: Whether to include soft-deleted files
-
-    Returns:
-        Number of files
-    """
-    where_clause: dict = {"workspaceId": workspace_id}
-    if not include_deleted:
-        where_clause["isDeleted"] = False
-
-    if path_prefix:
-        # Normalize prefix
-        if not path_prefix.startswith("/"):
-            path_prefix = f"/{path_prefix}"
-        where_clause["path"] = {"startswith": path_prefix}
-
-    return await UserWorkspaceFile.prisma().count(where=where_clause)
-
-
-async def soft_delete_workspace_file(
-    file_id: str,
-    workspace_id: Optional[str] = None,
-) -> Optional[UserWorkspaceFile]:
-    """
-    Soft-delete a workspace file.
-
-    The path is modified to include a deletion timestamp to free up the original
-    path for new files while preserving the record for potential recovery.
-
-    Args:
-        file_id: The file ID
-        workspace_id: Optional workspace ID for validation
-
-    Returns:
-        Updated UserWorkspaceFile instance or None if not found
-    """
-    # First verify the file exists and belongs to workspace
-    file = await get_workspace_file(file_id, workspace_id)
-    if file is None:
-        return None
-
-    deleted_at = datetime.now(timezone.utc)
-    # Modify path to free up the unique constraint for new files at original path
-    # Format: {original_path}__deleted__{timestamp}
-    deleted_path = f"{file.path}__deleted__{int(deleted_at.timestamp())}"
-
-    updated = await UserWorkspaceFile.prisma().update(
-        where={"id": file_id},
-        data={
-            "isDeleted": True,
-            "deletedAt": deleted_at,
-            "path": deleted_path,
-        },
-    )
-
-    logger.info(f"Soft-deleted workspace file {file_id}")
-    return updated
-
-
-async def get_workspace_total_size(workspace_id: str) -> int:
-    """
-    Get the total size of all files in a workspace.
-
-    Args:
-        workspace_id: The workspace ID
-
-    Returns:
-        Total size in bytes
-    """
-    files = await list_workspace_files(workspace_id)
-    return sum(file.sizeBytes for file in files)

autogpt_platform/backend/backend/executor/manager.py

@@ -236,14 +236,7 @@ async def execute_node(
     input_size = len(input_data_str)
     log_metadata.debug("Executed node with input", input=input_data_str)
 
-    # Create node-specific execution context to avoid race conditions
-    # (multiple nodes can execute concurrently and would otherwise mutate shared state)
-    execution_context = execution_context.model_copy(
-        update={"node_id": node_id, "node_exec_id": node_exec_id}
-    )
-
     # Inject extra execution arguments for the blocks via kwargs
-    # Keep individual kwargs for backwards compatibility with existing blocks
     extra_exec_kwargs: dict = {
         "graph_id": graph_id,
         "graph_version": graph_version,

autogpt_platform/backend/backend/executor/utils.py

@@ -892,19 +892,11 @@ async def add_graph_execution(
         settings = await gdb.get_graph_settings(user_id=user_id, graph_id=graph_id)
 
         execution_context = ExecutionContext(
-            # Execution identity
-            user_id=user_id,
-            graph_id=graph_id,
-            graph_exec_id=graph_exec.id,
-            graph_version=graph_exec.graph_version,
-            # Safety settings
             human_in_the_loop_safe_mode=settings.human_in_the_loop_safe_mode,
             sensitive_action_safe_mode=settings.sensitive_action_safe_mode,
-            # User settings
             user_timezone=(
                 user.timezone if user.timezone != USER_TIMEZONE_NOT_SET else "UTC"
             ),
-            # Execution hierarchy
             root_execution_id=graph_exec.id,
         )
 

autogpt_platform/backend/backend/executor/utils_test.py

@@ -348,7 +348,6 @@ async def test_add_graph_execution_is_repeatable(mocker: MockerFixture):
     mock_graph_exec.id = "execution-id-123"
     mock_graph_exec.node_executions = []  # Add this to avoid AttributeError
     mock_graph_exec.status = ExecutionStatus.QUEUED  # Required for race condition check
-    mock_graph_exec.graph_version = graph_version
     mock_graph_exec.to_graph_execution_entry.return_value = mocker.MagicMock()
 
     # Mock the queue and event bus
@@ -435,9 +434,6 @@ async def test_add_graph_execution_is_repeatable(mocker: MockerFixture):
     # Create a second mock execution for the sanity check
     mock_graph_exec_2 = mocker.MagicMock(spec=GraphExecutionWithNodes)
     mock_graph_exec_2.id = "execution-id-456"
-    mock_graph_exec_2.node_executions = []
-    mock_graph_exec_2.status = ExecutionStatus.QUEUED
-    mock_graph_exec_2.graph_version = graph_version
     mock_graph_exec_2.to_graph_execution_entry.return_value = mocker.MagicMock()
 
     # Reset mocks and set up for second call
@@ -618,7 +614,6 @@ async def test_add_graph_execution_with_nodes_to_skip(mocker: MockerFixture):
     mock_graph_exec.id = "execution-id-123"
     mock_graph_exec.node_executions = []
     mock_graph_exec.status = ExecutionStatus.QUEUED  # Required for race condition check
-    mock_graph_exec.graph_version = graph_version
 
     # Track what's passed to to_graph_execution_entry
     captured_kwargs = {}

autogpt_platform/backend/backend/integrations/webhooks/utils_test.py

@@ -1,39 +0,0 @@
-from urllib.parse import urlparse
-
-import fastapi
-from fastapi.routing import APIRoute
-
-from backend.api.features.integrations.router import router as integrations_router
-from backend.integrations.providers import ProviderName
-from backend.integrations.webhooks import utils as webhooks_utils
-
-
-def test_webhook_ingress_url_matches_route(monkeypatch) -> None:
-    app = fastapi.FastAPI()
-    app.include_router(integrations_router, prefix="/api/integrations")
-
-    provider = ProviderName.GITHUB
-    webhook_id = "webhook_123"
-    base_url = "https://example.com"
-
-    monkeypatch.setattr(webhooks_utils.app_config, "platform_base_url", base_url)
-
-    route = next(
-        route
-        for route in integrations_router.routes
-        if isinstance(route, APIRoute)
-        and route.path == "/{provider}/webhooks/{webhook_id}/ingress"
-        and "POST" in route.methods
-    )
-    expected_path = f"/api/integrations{route.path}".format(
-        provider=provider.value,
-        webhook_id=webhook_id,
-    )
-    actual_url = urlparse(webhooks_utils.webhook_ingress_url(provider, webhook_id))
-    expected_base = urlparse(base_url)
-
-    assert (actual_url.scheme, actual_url.netloc) == (
-        expected_base.scheme,
-        expected_base.netloc,
-    )
-    assert actual_url.path == expected_path

autogpt_platform/backend/backend/util/cloud_storage.py

@@ -13,7 +13,6 @@ import aiohttp
 from gcloud.aio import storage as async_gcs_storage
 from google.cloud import storage as gcs_storage
 
-from backend.util.gcs_utils import download_with_fresh_session, generate_signed_url
 from backend.util.settings import Config
 
 logger = logging.getLogger(__name__)
@@ -252,7 +251,7 @@ class CloudStorageHandler:
             f"in_task: {current_task is not None}"
         )
 
-        # Parse bucket and blob name from path (path already has gcs:// prefix removed)
+        # Parse bucket and blob name from path
         parts = path.split("/", 1)
         if len(parts) != 2:
             raise ValueError(f"Invalid GCS path: {path}")
@@ -262,19 +261,50 @@ class CloudStorageHandler:
         # Authorization check
         self._validate_file_access(blob_name, user_id, graph_exec_id)
 
-        logger.info(
-            f"[CloudStorage] About to download from GCS - bucket: {bucket_name}, blob: {blob_name}"
+        # Use a fresh client for each download to avoid session issues
+        # This is less efficient but more reliable with the executor's event loop
+        logger.info("[CloudStorage] Creating fresh GCS client for download")
+
+        # Create a new session specifically for this download
+        session = aiohttp.ClientSession(
+            connector=aiohttp.TCPConnector(limit=10, force_close=True)
         )
 
+        async_client = None
         try:
-            content = await download_with_fresh_session(bucket_name, blob_name)
+            # Create a new GCS client with the fresh session
+            async_client = async_gcs_storage.Storage(session=session)
+
+            logger.info(
+                f"[CloudStorage] About to download from GCS - bucket: {bucket_name}, blob: {blob_name}"
+            )
+
+            # Download content using the fresh client
+            content = await async_client.download(bucket_name, blob_name)
             logger.info(
                 f"[CloudStorage] GCS download successful - size: {len(content)} bytes"
             )
+
+            # Clean up
+            await async_client.close()
+            await session.close()
+
             return content
-        except FileNotFoundError:
-            raise
+
         except Exception as e:
+            # Always try to clean up
+            if async_client is not None:
+                try:
+                    await async_client.close()
+                except Exception as cleanup_error:
+                    logger.warning(
+                        f"[CloudStorage] Error closing GCS client: {cleanup_error}"
+                    )
+            try:
+                await session.close()
+            except Exception as cleanup_error:
+                logger.warning(f"[CloudStorage] Error closing session: {cleanup_error}")
+
             # Log the specific error for debugging
             logger.error(
                 f"[CloudStorage] GCS download failed - error: {str(e)}, "
@@ -289,6 +319,10 @@ class CloudStorageHandler:
                     f"current_task: {current_task}, "
                     f"bucket: {bucket_name}, blob: redacted for privacy"
                 )
+
+            # Convert gcloud-aio exceptions to standard ones
+            if "404" in str(e) or "Not Found" in str(e):
+                raise FileNotFoundError(f"File not found: gcs://{path}")
             raise
 
     def _validate_file_access(
@@ -411,7 +445,8 @@ class CloudStorageHandler:
         graph_exec_id: str | None = None,
     ) -> str:
         """Generate signed URL for GCS with authorization."""
-        # Parse bucket and blob name from path (path already has gcs:// prefix removed)
+
+        # Parse bucket and blob name from path
         parts = path.split("/", 1)
         if len(parts) != 2:
             raise ValueError(f"Invalid GCS path: {path}")
@@ -421,11 +456,21 @@ class CloudStorageHandler:
         # Authorization check
         self._validate_file_access(blob_name, user_id, graph_exec_id)
 
+        # Use sync client for signed URLs since gcloud-aio doesn't support them
         sync_client = self._get_sync_gcs_client()
-        return await generate_signed_url(
-            sync_client, bucket_name, blob_name, expiration_hours * 3600
+        bucket = sync_client.bucket(bucket_name)
+        blob = bucket.blob(blob_name)
+
+        # Generate signed URL asynchronously using sync client
+        url = await asyncio.to_thread(
+            blob.generate_signed_url,
+            version="v4",
+            expiration=datetime.now(timezone.utc) + timedelta(hours=expiration_hours),
+            method="GET",
         )
 
+        return url
+
     async def delete_expired_files(self, provider: str = "gcs") -> int:
         """
         Delete files that have passed their expiration time.

autogpt_platform/backend/backend/util/exceptions.py

@@ -135,12 +135,6 @@ class GraphValidationError(ValueError):
         )
 
 
-class InvalidInputError(ValueError):
-    """Raised when user input validation fails (e.g., search term too long)"""
-
-    pass
-
-
 class DatabaseError(Exception):
     """Raised when there is an error interacting with the database"""
 

autogpt_platform/backend/backend/util/file.py

@@ -5,26 +5,13 @@ import shutil
 import tempfile
 import uuid
 from pathlib import Path
-from typing import TYPE_CHECKING, Literal
 from urllib.parse import urlparse
 
 from backend.util.cloud_storage import get_cloud_storage_handler
 from backend.util.request import Requests
-from backend.util.settings import Config
 from backend.util.type import MediaFileType
 from backend.util.virus_scanner import scan_content_safe
 
-if TYPE_CHECKING:
-    from backend.data.execution import ExecutionContext
-
-# Return format options for store_media_file
-# - "for_local_processing": Returns local file path - use with ffmpeg, MoviePy, PIL, etc.
-# - "for_external_api": Returns data URI (base64) - use when sending content to external APIs
-# - "for_block_output": Returns best format for output - workspace:// in CoPilot, data URI in graphs
-MediaReturnFormat = Literal[
-    "for_local_processing", "for_external_api", "for_block_output"
-]
-
 TEMP_DIR = Path(tempfile.gettempdir()).resolve()
 
 # Maximum filename length (conservative limit for most filesystems)
@@ -80,56 +67,42 @@ def clean_exec_files(graph_exec_id: str, file: str = "") -> None:
 
 
 async def store_media_file(
+    graph_exec_id: str,
     file: MediaFileType,
-    execution_context: "ExecutionContext",
-    *,
-    return_format: MediaReturnFormat,
+    user_id: str,
+    return_content: bool = False,
 ) -> MediaFileType:
     """
-    Safely handle 'file' (a data URI, a URL, a workspace:// reference, or a local path
-    relative to {temp}/exec_file/{exec_id}), placing or verifying it under:
+    Safely handle 'file' (a data URI, a URL, or a local path relative to {temp}/exec_file/{exec_id}),
+    placing or verifying it under:
         {tempdir}/exec_file/{exec_id}/...
 
-    For each MediaFileType input:
-    - Data URI: decode and store locally
-    - URL: download and store locally
-    - workspace:// reference: read from workspace, store locally
-    - Local path: verify it exists in exec_file directory
+    If 'return_content=True', return a data URI (data:<mime>;base64,<content>).
+    Otherwise, returns the file media path relative to the exec_id folder.
 
-    Return format options:
-    - "for_local_processing": Returns local file path - use with ffmpeg, MoviePy, PIL, etc.
-    - "for_external_api": Returns data URI (base64) - use when sending to external APIs
-    - "for_block_output": Returns best format for output - workspace:// in CoPilot, data URI in graphs
+    For each MediaFileType type:
+    - Data URI:
+      -> decode and store in a new random file in that folder
+    - URL:
+      -> download and store in that folder
+    - Local path:
+      -> interpret as relative to that folder; verify it exists
+         (no copying, as it's presumably already there).
+         We realpath-check so no symlink or '..' can escape the folder.
 
-    :param file:               Data URI, URL, workspace://, or local (relative) path.
-    :param execution_context:  ExecutionContext with user_id, graph_exec_id, workspace_id.
-    :param return_format:      What to return: "for_local_processing", "for_external_api", or "for_block_output".
-    :return:                   The requested result based on return_format.
+
+    :param graph_exec_id:  The unique ID of the graph execution.
+    :param file:           Data URI, URL, or local (relative) path.
+    :param return_content: If True, return a data URI of the file content.
+                           If False, return the *relative* path inside the exec_id folder.
+    :return:               The requested result: data URI or relative path of the media.
     """
-    # Extract values from execution_context
-    graph_exec_id = execution_context.graph_exec_id
-    user_id = execution_context.user_id
-
-    if not graph_exec_id:
-        raise ValueError("execution_context.graph_exec_id is required")
-    if not user_id:
-        raise ValueError("execution_context.user_id is required")
-
-    # Create workspace_manager if we have workspace_id (with session scoping)
-    # Import here to avoid circular import (file.py → workspace.py → data → blocks → file.py)
-    from backend.util.workspace import WorkspaceManager
-
-    workspace_manager: WorkspaceManager | None = None
-    if execution_context.workspace_id:
-        workspace_manager = WorkspaceManager(
-            user_id, execution_context.workspace_id, execution_context.session_id
-        )
     # Build base path
     base_path = Path(get_exec_file_path(graph_exec_id, ""))
     base_path.mkdir(parents=True, exist_ok=True)
 
     # Security fix: Add disk space limits to prevent DoS
-    MAX_FILE_SIZE_BYTES = Config().max_file_size_mb * 1024 * 1024
+    MAX_FILE_SIZE = 100 * 1024 * 1024  # 100MB per file
     MAX_TOTAL_DISK_USAGE = 1024 * 1024 * 1024  # 1GB total per execution directory
 
     # Check total disk usage in base_path
@@ -169,57 +142,9 @@ async def store_media_file(
         """
         return str(absolute_path.relative_to(base))
 
-    # Get cloud storage handler for checking cloud paths
-    cloud_storage = await get_cloud_storage_handler()
-
-    # Track if the input came from workspace (don't re-save it)
-    is_from_workspace = file.startswith("workspace://")
-
-    # Check if this is a workspace file reference
-    if is_from_workspace:
-        if workspace_manager is None:
-            raise ValueError(
-                "Workspace file reference requires workspace context. "
-                "This file type is only available in CoPilot sessions."
-            )
-
-        # Parse workspace reference
-        # workspace://abc123 - by file ID
-        # workspace:///path/to/file.txt - by virtual path
-        file_ref = file[12:]  # Remove "workspace://"
-
-        if file_ref.startswith("/"):
-            # Path reference
-            workspace_content = await workspace_manager.read_file(file_ref)
-            file_info = await workspace_manager.get_file_info_by_path(file_ref)
-            filename = sanitize_filename(
-                file_info.name if file_info else f"{uuid.uuid4()}.bin"
-            )
-        else:
-            # ID reference
-            workspace_content = await workspace_manager.read_file_by_id(file_ref)
-            file_info = await workspace_manager.get_file_info(file_ref)
-            filename = sanitize_filename(
-                file_info.name if file_info else f"{uuid.uuid4()}.bin"
-            )
-
-        try:
-            target_path = _ensure_inside_base(base_path / filename, base_path)
-        except OSError as e:
-            raise ValueError(f"Invalid file path '{filename}': {e}") from e
-
-        # Check file size limit
-        if len(workspace_content) > MAX_FILE_SIZE_BYTES:
-            raise ValueError(
-                f"File too large: {len(workspace_content)} bytes > {MAX_FILE_SIZE_BYTES} bytes"
-            )
-
-        # Virus scan the workspace content before writing locally
-        await scan_content_safe(workspace_content, filename=filename)
-        target_path.write_bytes(workspace_content)
-
     # Check if this is a cloud storage path
-    elif cloud_storage.is_cloud_path(file):
+    cloud_storage = await get_cloud_storage_handler()
+    if cloud_storage.is_cloud_path(file):
         # Download from cloud storage and store locally
         cloud_content = await cloud_storage.retrieve_file(
             file, user_id=user_id, graph_exec_id=graph_exec_id
@@ -234,9 +159,9 @@ async def store_media_file(
             raise ValueError(f"Invalid file path '{filename}': {e}") from e
 
         # Check file size limit
-        if len(cloud_content) > MAX_FILE_SIZE_BYTES:
+        if len(cloud_content) > MAX_FILE_SIZE:
             raise ValueError(
-                f"File too large: {len(cloud_content)} bytes > {MAX_FILE_SIZE_BYTES} bytes"
+                f"File too large: {len(cloud_content)} bytes > {MAX_FILE_SIZE} bytes"
             )
 
         # Virus scan the cloud content before writing locally
@@ -264,9 +189,9 @@ async def store_media_file(
         content = base64.b64decode(b64_content)
 
         # Check file size limit
-        if len(content) > MAX_FILE_SIZE_BYTES:
+        if len(content) > MAX_FILE_SIZE:
             raise ValueError(
-                f"File too large: {len(content)} bytes > {MAX_FILE_SIZE_BYTES} bytes"
+                f"File too large: {len(content)} bytes > {MAX_FILE_SIZE} bytes"
             )
 
         # Virus scan the base64 content before writing
@@ -274,31 +199,23 @@ async def store_media_file(
         target_path.write_bytes(content)
 
     elif file.startswith(("http://", "https://")):
-        # URL - download first to get Content-Type header
-        resp = await Requests().get(file)
-
-        # Check file size limit
-        if len(resp.content) > MAX_FILE_SIZE_BYTES:
-            raise ValueError(
-                f"File too large: {len(resp.content)} bytes > {MAX_FILE_SIZE_BYTES} bytes"
-            )
-
-        # Extract filename from URL path
+        # URL
         parsed_url = urlparse(file)
         filename = sanitize_filename(Path(parsed_url.path).name or f"{uuid.uuid4()}")
-
-        # If filename lacks extension, add one from Content-Type header
-        if "." not in filename:
-            content_type = resp.headers.get("Content-Type", "").split(";")[0].strip()
-            if content_type:
-                ext = _extension_from_mime(content_type)
-                filename = f"{filename}{ext}"
-
         try:
             target_path = _ensure_inside_base(base_path / filename, base_path)
         except OSError as e:
             raise ValueError(f"Invalid file path '{filename}': {e}") from e
 
+        # Download and save
+        resp = await Requests().get(file)
+
+        # Check file size limit
+        if len(resp.content) > MAX_FILE_SIZE:
+            raise ValueError(
+                f"File too large: {len(resp.content)} bytes > {MAX_FILE_SIZE} bytes"
+            )
+
         # Virus scan the downloaded content before writing
         await scan_content_safe(resp.content, filename=filename)
         target_path.write_bytes(resp.content)
@@ -313,43 +230,11 @@ async def store_media_file(
         if not target_path.is_file():
             raise ValueError(f"Local file does not exist: {target_path}")
 
-    # Return based on requested format
-    if return_format == "for_local_processing":
-        # Use when processing files locally with tools like ffmpeg, MoviePy, PIL
-        # Returns: relative path in exec_file directory (e.g., "image.png")
-        return MediaFileType(_strip_base_prefix(target_path, base_path))
-
-    elif return_format == "for_external_api":
-        # Use when sending content to external APIs that need base64
-        # Returns: data URI (e.g., "data:image/png;base64,iVBORw0...")
+    # Return result
+    if return_content:
         return MediaFileType(_file_to_data_uri(target_path))
-
-    elif return_format == "for_block_output":
-        # Use when returning output from a block to user/next block
-        # Returns: workspace:// ref (CoPilot) or data URI (graph execution)
-        if workspace_manager is None:
-            # No workspace available (graph execution without CoPilot)
-            # Fallback to data URI so the content can still be used/displayed
-            return MediaFileType(_file_to_data_uri(target_path))
-
-        # Don't re-save if input was already from workspace
-        if is_from_workspace:
-            # Return original workspace reference
-            return MediaFileType(file)
-
-        # Save new content to workspace
-        content = target_path.read_bytes()
-        filename = target_path.name
-
-        file_record = await workspace_manager.write_file(
-            content=content,
-            filename=filename,
-            overwrite=True,
-        )
-        return MediaFileType(f"workspace://{file_record.id}")
-
     else:
-        raise ValueError(f"Invalid return_format: {return_format}")
+        return MediaFileType(_strip_base_prefix(target_path, base_path))
 
 
 def get_dir_size(path: Path) -> int:

autogpt_platform/backend/backend/util/file_test.py

@@ -7,22 +7,10 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from backend.data.execution import ExecutionContext
 from backend.util.file import store_media_file
 from backend.util.type import MediaFileType
 
 
-def make_test_context(
-    graph_exec_id: str = "test-exec-123",
-    user_id: str = "test-user-123",
-) -> ExecutionContext:
-    """Helper to create test ExecutionContext."""
-    return ExecutionContext(
-        user_id=user_id,
-        graph_exec_id=graph_exec_id,
-    )
-
-
 class TestFileCloudIntegration:
     """Test cases for cloud storage integration in file utilities."""
 
@@ -82,9 +70,10 @@ class TestFileCloudIntegration:
             mock_path_class.side_effect = path_constructor
 
             result = await store_media_file(
-                file=MediaFileType(cloud_path),
-                execution_context=make_test_context(graph_exec_id=graph_exec_id),
-                return_format="for_local_processing",
+                graph_exec_id,
+                MediaFileType(cloud_path),
+                "test-user-123",
+                return_content=False,
             )
 
             # Verify cloud storage operations
@@ -155,9 +144,10 @@ class TestFileCloudIntegration:
             mock_path_obj.name = "image.png"
             with patch("backend.util.file.Path", return_value=mock_path_obj):
                 result = await store_media_file(
-                    file=MediaFileType(cloud_path),
-                    execution_context=make_test_context(graph_exec_id=graph_exec_id),
-                    return_format="for_external_api",
+                    graph_exec_id,
+                    MediaFileType(cloud_path),
+                    "test-user-123",
+                    return_content=True,
                 )
 
             # Verify result is a data URI
@@ -208,9 +198,10 @@ class TestFileCloudIntegration:
             mock_resolved_path.relative_to.return_value = Path("test-uuid-789.txt")
 
             await store_media_file(
-                file=MediaFileType(data_uri),
-                execution_context=make_test_context(graph_exec_id=graph_exec_id),
-                return_format="for_local_processing",
+                graph_exec_id,
+                MediaFileType(data_uri),
+                "test-user-123",
+                return_content=False,
             )
 
             # Verify cloud handler was checked but not used for retrieval
@@ -243,7 +234,5 @@ class TestFileCloudIntegration:
                 FileNotFoundError, match="File not found in cloud storage"
             ):
                 await store_media_file(
-                    file=MediaFileType(cloud_path),
-                    execution_context=make_test_context(graph_exec_id=graph_exec_id),
-                    return_format="for_local_processing",
+                    graph_exec_id, MediaFileType(cloud_path), "test-user-123"
                 )

autogpt_platform/backend/backend/util/gcs_utils.py

@@ -1,108 +0,0 @@
-"""
-Shared GCS utilities for workspace and cloud storage backends.
-
-This module provides common functionality for working with Google Cloud Storage,
-including path parsing, client management, and signed URL generation.
-"""
-
-import asyncio
-import logging
-from datetime import datetime, timedelta, timezone
-
-import aiohttp
-from gcloud.aio import storage as async_gcs_storage
-from google.cloud import storage as gcs_storage
-
-logger = logging.getLogger(__name__)
-
-
-def parse_gcs_path(path: str) -> tuple[str, str]:
-    """
-    Parse a GCS path in the format 'gcs://bucket/blob' to (bucket, blob).
-
-    Args:
-        path: GCS path string (e.g., "gcs://my-bucket/path/to/file")
-
-    Returns:
-        Tuple of (bucket_name, blob_name)
-
-    Raises:
-        ValueError: If the path format is invalid
-    """
-    if not path.startswith("gcs://"):
-        raise ValueError(f"Invalid GCS path: {path}")
-
-    path_without_prefix = path[6:]  # Remove "gcs://"
-    parts = path_without_prefix.split("/", 1)
-    if len(parts) != 2:
-        raise ValueError(f"Invalid GCS path format: {path}")
-
-    return parts[0], parts[1]
-
-
-async def download_with_fresh_session(bucket: str, blob: str) -> bytes:
-    """
-    Download file content using a fresh session.
-
-    This approach avoids event loop issues that can occur when reusing
-    sessions across different async contexts (e.g., in executors).
-
-    Args:
-        bucket: GCS bucket name
-        blob: Blob path within the bucket
-
-    Returns:
-        File content as bytes
-
-    Raises:
-        FileNotFoundError: If the file doesn't exist
-    """
-    session = aiohttp.ClientSession(
-        connector=aiohttp.TCPConnector(limit=10, force_close=True)
-    )
-    client: async_gcs_storage.Storage | None = None
-    try:
-        client = async_gcs_storage.Storage(session=session)
-        content = await client.download(bucket, blob)
-        return content
-    except Exception as e:
-        if "404" in str(e) or "Not Found" in str(e):
-            raise FileNotFoundError(f"File not found: gcs://{bucket}/{blob}")
-        raise
-    finally:
-        if client:
-            try:
-                await client.close()
-            except Exception:
-                pass  # Best-effort cleanup
-        await session.close()
-
-
-async def generate_signed_url(
-    sync_client: gcs_storage.Client,
-    bucket_name: str,
-    blob_name: str,
-    expires_in: int,
-) -> str:
-    """
-    Generate a signed URL for temporary access to a GCS file.
-
-    Uses asyncio.to_thread() to run the sync operation without blocking.
-
-    Args:
-        sync_client: Sync GCS client with service account credentials
-        bucket_name: GCS bucket name
-        blob_name: Blob path within the bucket
-        expires_in: URL expiration time in seconds
-
-    Returns:
-        Signed URL string
-    """
-    bucket = sync_client.bucket(bucket_name)
-    blob = bucket.blob(blob_name)
-    return await asyncio.to_thread(
-        blob.generate_signed_url,
-        version="v4",
-        expiration=datetime.now(timezone.utc) + timedelta(seconds=expires_in),
-        method="GET",
-    )

autogpt_platform/backend/backend/util/settings.py

@@ -263,12 +263,6 @@ class Config(UpdateTrackingModel["Config"], BaseSettings):
         description="The name of the Google Cloud Storage bucket for media files",
     )
 
-    workspace_storage_dir: str = Field(
-        default="",
-        description="Local directory for workspace file storage when GCS is not configured. "
-        "If empty, defaults to {app_data}/workspaces. Used for self-hosted deployments.",
-    )
-
     reddit_user_agent: str = Field(
         default="web:AutoGPT:v0.6.0 (by /u/autogpt)",
         description="The user agent for the Reddit API",
@@ -365,8 +359,8 @@ class Config(UpdateTrackingModel["Config"], BaseSettings):
         description="The port for the Agent Generator service",
     )
     agentgenerator_timeout: int = Field(
-        default=600,
-        description="The timeout in seconds for Agent Generator service requests (includes retries for rate limits)",
+        default=120,
+        description="The timeout in seconds for Agent Generator service requests",
     )
 
     enable_example_blocks: bool = Field(
@@ -395,13 +389,6 @@ class Config(UpdateTrackingModel["Config"], BaseSettings):
         description="Maximum file size in MB for file uploads (1-1024 MB)",
     )
 
-    max_file_size_mb: int = Field(
-        default=100,
-        ge=1,
-        le=1024,
-        description="Maximum file size in MB for workspace files (1-1024 MB)",
-    )
-
     # AutoMod configuration
     automod_enabled: bool = Field(
         default=False,
@@ -692,12 +679,6 @@ class Secrets(UpdateTrackingModel["Secrets"], BaseSettings):
         default="https://cloud.langfuse.com", description="Langfuse host URL"
     )
 
-    # PostHog analytics
-    posthog_api_key: str = Field(default="", description="PostHog API key")
-    posthog_host: str = Field(
-        default="https://eu.i.posthog.com", description="PostHog host URL"
-    )
-
     # Add more secret fields as needed
     model_config = SettingsConfigDict(
         env_file=".env",

autogpt_platform/backend/backend/util/test.py

@@ -140,29 +140,14 @@ async def execute_block_test(block: Block):
             setattr(block, mock_name, mock_obj)
 
     # Populate credentials argument(s)
-    # Generate IDs for execution context
-    graph_id = str(uuid.uuid4())
-    node_id = str(uuid.uuid4())
-    graph_exec_id = str(uuid.uuid4())
-    node_exec_id = str(uuid.uuid4())
-    user_id = str(uuid.uuid4())
-    graph_version = 1  # Default version for tests
-
     extra_exec_kwargs: dict = {
-        "graph_id": graph_id,
-        "node_id": node_id,
-        "graph_exec_id": graph_exec_id,
-        "node_exec_id": node_exec_id,
-        "user_id": user_id,
-        "graph_version": graph_version,
-        "execution_context": ExecutionContext(
-            user_id=user_id,
-            graph_id=graph_id,
-            graph_exec_id=graph_exec_id,
-            graph_version=graph_version,
-            node_id=node_id,
-            node_exec_id=node_exec_id,
-        ),
+        "graph_id": str(uuid.uuid4()),
+        "node_id": str(uuid.uuid4()),
+        "graph_exec_id": str(uuid.uuid4()),
+        "node_exec_id": str(uuid.uuid4()),
+        "user_id": str(uuid.uuid4()),
+        "graph_version": 1,  # Default version for tests
+        "execution_context": ExecutionContext(),
     }
     input_model = cast(type[BlockSchema], block.input_schema)
 

autogpt_platform/backend/backend/util/workspace.py

@@ -1,419 +0,0 @@
-"""
-WorkspaceManager for managing user workspace file operations.
-
-This module provides a high-level interface for workspace file operations,
-combining the storage backend and database layer.
-"""
-
-import logging
-import mimetypes
-import uuid
-from typing import Optional
-
-from prisma.errors import UniqueViolationError
-from prisma.models import UserWorkspaceFile
-
-from backend.data.workspace import (
-    count_workspace_files,
-    create_workspace_file,
-    get_workspace_file,
-    get_workspace_file_by_path,
-    list_workspace_files,
-    soft_delete_workspace_file,
-)
-from backend.util.settings import Config
-from backend.util.workspace_storage import compute_file_checksum, get_workspace_storage
-
-logger = logging.getLogger(__name__)
-
-
-class WorkspaceManager:
-    """
-    Manages workspace file operations.
-
-    Combines storage backend operations with database record management.
-    Supports session-scoped file segmentation where files are stored in
-    session-specific virtual paths: /sessions/{session_id}/{filename}
-    """
-
-    def __init__(
-        self, user_id: str, workspace_id: str, session_id: Optional[str] = None
-    ):
-        """
-        Initialize WorkspaceManager.
-
-        Args:
-            user_id: The user's ID
-            workspace_id: The workspace ID
-            session_id: Optional session ID for session-scoped file access
-        """
-        self.user_id = user_id
-        self.workspace_id = workspace_id
-        self.session_id = session_id
-        # Session path prefix for file isolation
-        self.session_path = f"/sessions/{session_id}" if session_id else ""
-
-    def _resolve_path(self, path: str) -> str:
-        """
-        Resolve a path, defaulting to session folder if session_id is set.
-
-        Cross-session access is allowed by explicitly using /sessions/other-session-id/...
-
-        Args:
-            path: Virtual path (e.g., "/file.txt" or "/sessions/abc123/file.txt")
-
-        Returns:
-            Resolved path with session prefix if applicable
-        """
-        # If path explicitly references a session folder, use it as-is
-        if path.startswith("/sessions/"):
-            return path
-
-        # If we have a session context, prepend session path
-        if self.session_path:
-            # Normalize the path
-            if not path.startswith("/"):
-                path = f"/{path}"
-            return f"{self.session_path}{path}"
-
-        # No session context, use path as-is
-        return path if path.startswith("/") else f"/{path}"
-
-    def _get_effective_path(
-        self, path: Optional[str], include_all_sessions: bool
-    ) -> Optional[str]:
-        """
-        Get effective path for list/count operations based on session context.
-
-        Args:
-            path: Optional path prefix to filter
-            include_all_sessions: If True, don't apply session scoping
-
-        Returns:
-            Effective path prefix for database query
-        """
-        if include_all_sessions:
-            # Normalize path to ensure leading slash (stored paths are normalized)
-            if path is not None and not path.startswith("/"):
-                return f"/{path}"
-            return path
-        elif path is not None:
-            # Resolve the provided path with session scoping
-            return self._resolve_path(path)
-        elif self.session_path:
-            # Default to session folder with trailing slash to prevent prefix collisions
-            # e.g., "/sessions/abc" should not match "/sessions/abc123"
-            return self.session_path.rstrip("/") + "/"
-        else:
-            # No session context, use path as-is
-            return path
-
-    async def read_file(self, path: str) -> bytes:
-        """
-        Read file from workspace by virtual path.
-
-        When session_id is set, paths are resolved relative to the session folder
-        unless they explicitly reference /sessions/...
-
-        Args:
-            path: Virtual path (e.g., "/documents/report.pdf")
-
-        Returns:
-            File content as bytes
-
-        Raises:
-            FileNotFoundError: If file doesn't exist
-        """
-        resolved_path = self._resolve_path(path)
-        file = await get_workspace_file_by_path(self.workspace_id, resolved_path)
-        if file is None:
-            raise FileNotFoundError(f"File not found at path: {resolved_path}")
-
-        storage = await get_workspace_storage()
-        return await storage.retrieve(file.storagePath)
-
-    async def read_file_by_id(self, file_id: str) -> bytes:
-        """
-        Read file from workspace by file ID.
-
-        Args:
-            file_id: The file's ID
-
-        Returns:
-            File content as bytes
-
-        Raises:
-            FileNotFoundError: If file doesn't exist
-        """
-        file = await get_workspace_file(file_id, self.workspace_id)
-        if file is None:
-            raise FileNotFoundError(f"File not found: {file_id}")
-
-        storage = await get_workspace_storage()
-        return await storage.retrieve(file.storagePath)
-
-    async def write_file(
-        self,
-        content: bytes,
-        filename: str,
-        path: Optional[str] = None,
-        mime_type: Optional[str] = None,
-        overwrite: bool = False,
-    ) -> UserWorkspaceFile:
-        """
-        Write file to workspace.
-
-        When session_id is set, files are written to /sessions/{session_id}/...
-        by default. Use explicit /sessions/... paths for cross-session access.
-
-        Args:
-            content: File content as bytes
-            filename: Filename for the file
-            path: Virtual path (defaults to "/{filename}", session-scoped if session_id set)
-            mime_type: MIME type (auto-detected if not provided)
-            overwrite: Whether to overwrite existing file at path
-
-        Returns:
-            Created UserWorkspaceFile instance
-
-        Raises:
-            ValueError: If file exceeds size limit or path already exists
-        """
-        # Enforce file size limit
-        max_file_size = Config().max_file_size_mb * 1024 * 1024
-        if len(content) > max_file_size:
-            raise ValueError(
-                f"File too large: {len(content)} bytes exceeds "
-                f"{Config().max_file_size_mb}MB limit"
-            )
-
-        # Determine path with session scoping
-        if path is None:
-            path = f"/{filename}"
-        elif not path.startswith("/"):
-            path = f"/{path}"
-
-        # Resolve path with session prefix
-        path = self._resolve_path(path)
-
-        # Check if file exists at path (only error for non-overwrite case)
-        # For overwrite=True, we let the write proceed and handle via UniqueViolationError
-        # This ensures the new file is written to storage BEFORE the old one is deleted,
-        # preventing data loss if the new write fails
-        if not overwrite:
-            existing = await get_workspace_file_by_path(self.workspace_id, path)
-            if existing is not None:
-                raise ValueError(f"File already exists at path: {path}")
-
-        # Auto-detect MIME type if not provided
-        if mime_type is None:
-            mime_type, _ = mimetypes.guess_type(filename)
-            mime_type = mime_type or "application/octet-stream"
-
-        # Compute checksum
-        checksum = compute_file_checksum(content)
-
-        # Generate unique file ID for storage
-        file_id = str(uuid.uuid4())
-
-        # Store file in storage backend
-        storage = await get_workspace_storage()
-        storage_path = await storage.store(
-            workspace_id=self.workspace_id,
-            file_id=file_id,
-            filename=filename,
-            content=content,
-        )
-
-        # Create database record - handle race condition where another request
-        # created a file at the same path between our check and create
-        try:
-            file = await create_workspace_file(
-                workspace_id=self.workspace_id,
-                file_id=file_id,
-                name=filename,
-                path=path,
-                storage_path=storage_path,
-                mime_type=mime_type,
-                size_bytes=len(content),
-                checksum=checksum,
-            )
-        except UniqueViolationError:
-            # Race condition: another request created a file at this path
-            if overwrite:
-                # Re-fetch and delete the conflicting file, then retry
-                existing = await get_workspace_file_by_path(self.workspace_id, path)
-                if existing:
-                    await self.delete_file(existing.id)
-                # Retry the create - if this also fails, clean up storage file
-                try:
-                    file = await create_workspace_file(
-                        workspace_id=self.workspace_id,
-                        file_id=file_id,
-                        name=filename,
-                        path=path,
-                        storage_path=storage_path,
-                        mime_type=mime_type,
-                        size_bytes=len(content),
-                        checksum=checksum,
-                    )
-                except Exception:
-                    # Clean up orphaned storage file on retry failure
-                    try:
-                        await storage.delete(storage_path)
-                    except Exception as e:
-                        logger.warning(f"Failed to clean up orphaned storage file: {e}")
-                    raise
-            else:
-                # Clean up the orphaned storage file before raising
-                try:
-                    await storage.delete(storage_path)
-                except Exception as e:
-                    logger.warning(f"Failed to clean up orphaned storage file: {e}")
-                raise ValueError(f"File already exists at path: {path}")
-        except Exception:
-            # Any other database error (connection, validation, etc.) - clean up storage
-            try:
-                await storage.delete(storage_path)
-            except Exception as e:
-                logger.warning(f"Failed to clean up orphaned storage file: {e}")
-            raise
-
-        logger.info(
-            f"Wrote file {file.id} ({filename}) to workspace {self.workspace_id} "
-            f"at path {path}, size={len(content)} bytes"
-        )
-
-        return file
-
-    async def list_files(
-        self,
-        path: Optional[str] = None,
-        limit: Optional[int] = None,
-        offset: int = 0,
-        include_all_sessions: bool = False,
-    ) -> list[UserWorkspaceFile]:
-        """
-        List files in workspace.
-
-        When session_id is set and include_all_sessions is False (default),
-        only files in the current session's folder are listed.
-
-        Args:
-            path: Optional path prefix to filter (e.g., "/documents/")
-            limit: Maximum number of files to return
-            offset: Number of files to skip
-            include_all_sessions: If True, list files from all sessions.
-                                  If False (default), only list current session's files.
-
-        Returns:
-            List of UserWorkspaceFile instances
-        """
-        effective_path = self._get_effective_path(path, include_all_sessions)
-
-        return await list_workspace_files(
-            workspace_id=self.workspace_id,
-            path_prefix=effective_path,
-            limit=limit,
-            offset=offset,
-        )
-
-    async def delete_file(self, file_id: str) -> bool:
-        """
-        Delete a file (soft-delete).
-
-        Args:
-            file_id: The file's ID
-
-        Returns:
-            True if deleted, False if not found
-        """
-        file = await get_workspace_file(file_id, self.workspace_id)
-        if file is None:
-            return False
-
-        # Delete from storage
-        storage = await get_workspace_storage()
-        try:
-            await storage.delete(file.storagePath)
-        except Exception as e:
-            logger.warning(f"Failed to delete file from storage: {e}")
-            # Continue with database soft-delete even if storage delete fails
-
-        # Soft-delete database record
-        result = await soft_delete_workspace_file(file_id, self.workspace_id)
-        return result is not None
-
-    async def get_download_url(self, file_id: str, expires_in: int = 3600) -> str:
-        """
-        Get download URL for a file.
-
-        Args:
-            file_id: The file's ID
-            expires_in: URL expiration in seconds (default 1 hour)
-
-        Returns:
-            Download URL (signed URL for GCS, API endpoint for local)
-
-        Raises:
-            FileNotFoundError: If file doesn't exist
-        """
-        file = await get_workspace_file(file_id, self.workspace_id)
-        if file is None:
-            raise FileNotFoundError(f"File not found: {file_id}")
-
-        storage = await get_workspace_storage()
-        return await storage.get_download_url(file.storagePath, expires_in)
-
-    async def get_file_info(self, file_id: str) -> Optional[UserWorkspaceFile]:
-        """
-        Get file metadata.
-
-        Args:
-            file_id: The file's ID
-
-        Returns:
-            UserWorkspaceFile instance or None
-        """
-        return await get_workspace_file(file_id, self.workspace_id)
-
-    async def get_file_info_by_path(self, path: str) -> Optional[UserWorkspaceFile]:
-        """
-        Get file metadata by path.
-
-        When session_id is set, paths are resolved relative to the session folder
-        unless they explicitly reference /sessions/...
-
-        Args:
-            path: Virtual path
-
-        Returns:
-            UserWorkspaceFile instance or None
-        """
-        resolved_path = self._resolve_path(path)
-        return await get_workspace_file_by_path(self.workspace_id, resolved_path)
-
-    async def get_file_count(
-        self,
-        path: Optional[str] = None,
-        include_all_sessions: bool = False,
-    ) -> int:
-        """
-        Get number of files in workspace.
-
-        When session_id is set and include_all_sessions is False (default),
-        only counts files in the current session's folder.
-
-        Args:
-            path: Optional path prefix to filter (e.g., "/documents/")
-            include_all_sessions: If True, count all files in workspace.
-                                  If False (default), only count current session's files.
-
-        Returns:
-            Number of files
-        """
-        effective_path = self._get_effective_path(path, include_all_sessions)
-
-        return await count_workspace_files(
-            self.workspace_id, path_prefix=effective_path
-        )