feat(copilot): add dummy agent generator for testing

Add optional dummy agent generator that returns mock responses matching
the expected format from the external Agent Generator service.

Enable via AGENTGENERATOR_USE_DUMMY=true in settings.

Changes:
- Add agentgenerator_use_dummy setting to settings.py (default: false)
- Create dummy.py with mock implementations of all service functions
- Modify service.py to route to dummy when enabled
- Add startup warning when dummy mode is active

This enables local CoPilot testing without needing the external
Agent Generator service running.

.github/workflows/platform-frontend-ci.yml

@@ -142,6 +142,9 @@ jobs:
 
   e2e_test:
     runs-on: big-boi
+    needs: setup
+    strategy:
+      fail-fast: false
 
     steps:
       - name: Checkout repository
@@ -171,29 +174,29 @@ jobs:
 
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
+
+      - name: Cache Docker layers
+        uses: actions/cache@v5
         with:
-          driver-opts: network=host
-
-      - name: Build Docker images with cache
-        working-directory: autogpt_platform
-        run: |
-          pip install pyyaml
-          python ../.github/workflows/scripts/generate-docker-ci-compose.py \
-            --source docker-compose.platform.yml \
-            --output docker-compose.ci.yml \
-            --cache-from "type=gha" \
-            --cache-to "type=gha,mode=max" \
-            --backend-scope "platform-backend-${{ hashFiles('autogpt_platform/backend/Dockerfile', 'autogpt_platform/backend/poetry.lock', 'autogpt_platform/backend/backend') }}" \
-            --frontend-scope "platform-frontend-${{ hashFiles('autogpt_platform/frontend/Dockerfile', 'autogpt_platform/frontend/pnpm-lock.yaml', 'autogpt_platform/frontend/src') }}"
-
-          docker buildx bake --allow=fs.read=.. -f docker-compose.yml -f docker-compose.ci.yml --load
-        env:
-          NEXT_PUBLIC_PW_TEST: true
+          path: /tmp/.buildx-cache
+          key: ${{ runner.os }}-buildx-frontend-test-${{ hashFiles('autogpt_platform/docker-compose.yml', 'autogpt_platform/backend/Dockerfile', 'autogpt_platform/backend/pyproject.toml', 'autogpt_platform/backend/poetry.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-buildx-frontend-test-
 
       - name: Run docker compose
-        run: docker compose -f ../docker-compose.yml up -d --no-build
+        run: |
+          NEXT_PUBLIC_PW_TEST=true docker compose -f ../docker-compose.yml up -d
         env:
-          NEXT_PUBLIC_PW_TEST: true
+          DOCKER_BUILDKIT: 1
+          BUILDX_CACHE_FROM: type=local,src=/tmp/.buildx-cache
+          BUILDX_CACHE_TO: type=local,dest=/tmp/.buildx-cache-new,mode=max
+
+      - name: Move cache
+        run: |
+          rm -rf /tmp/.buildx-cache
+          if [ -d "/tmp/.buildx-cache-new" ]; then
+            mv /tmp/.buildx-cache-new /tmp/.buildx-cache
+          fi
 
       - name: Wait for services to be ready
         run: |
@@ -227,14 +230,14 @@ jobs:
             }
           fi
 
-      - name: Cache pnpm store
+      - name: Restore dependencies cache
         uses: actions/cache@v5
         with:
           path: ~/.pnpm-store
-          # Use separate cache key for big-boi runner since it doesn't share cache with ubuntu-latest
-          key: big-boi-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+          key: ${{ needs.setup.outputs.cache-key }}
           restore-keys: |
-            big-boi-pnpm-
+            ${{ runner.os }}-pnpm-${{ hashFiles('autogpt_platform/frontend/pnpm-lock.yaml') }}
+            ${{ runner.os }}-pnpm-
 
       - name: Install dependencies
         run: pnpm install --frozen-lockfile

.github/workflows/scripts/generate-docker-ci-compose.py

@@ -1,93 +0,0 @@
-#!/usr/bin/env python3
-"""
-Generate a docker-compose.ci.yml with cache configuration for all services
-that have a build key in the source compose file.
-"""
-
-import argparse
-
-import yaml
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Generate docker-compose cache override file"
-    )
-    parser.add_argument(
-        "--source",
-        default="docker-compose.platform.yml",
-        help="Source compose file to read (default: docker-compose.platform.yml)",
-    )
-    parser.add_argument(
-        "--output",
-        default="docker-compose.ci.yml",
-        help="Output compose file to write (default: docker-compose.ci.yml)",
-    )
-    parser.add_argument(
-        "--cache-from",
-        default="type=local,src=/tmp/.buildx-cache",
-        help="Cache source configuration",
-    )
-    parser.add_argument(
-        "--cache-to",
-        default="type=local,dest=/tmp/.buildx-cache-new,mode=max",
-        help="Cache destination configuration",
-    )
-    parser.add_argument(
-        "--backend-scope",
-        default="",
-        help="GHA cache scope for backend services (e.g., platform-backend-{hash})",
-    )
-    parser.add_argument(
-        "--frontend-scope",
-        default="",
-        help="GHA cache scope for frontend service (e.g., platform-frontend-{hash})",
-    )
-    args = parser.parse_args()
-
-    with open(args.source, "r") as f:
-        compose = yaml.safe_load(f)
-
-    ci_compose = {"services": {}}
-    for service_name, service_config in compose.get("services", {}).items():
-        if "build" not in service_config:
-            continue
-
-        cache_from = args.cache_from
-        cache_to = args.cache_to
-
-        # Determine scope based on Dockerfile path
-        if "type=gha" in args.cache_from or "type=gha" in args.cache_to:
-            dockerfile = service_config["build"].get("dockerfile", "Dockerfile")
-            if "frontend" in dockerfile:
-                scope = args.frontend_scope
-            elif "backend" in dockerfile:
-                scope = args.backend_scope
-            else:
-                # Skip services that don't clearly match frontend/backend
-                continue
-
-            if scope:
-                if "type=gha" in args.cache_from:
-                    cache_from = f"{args.cache_from},scope={scope}"
-                if "type=gha" in args.cache_to:
-                    cache_to = f"{args.cache_to},scope={scope}"
-
-        ci_compose["services"][service_name] = {
-            "build": {
-                "cache_from": [cache_from],
-                "cache_to": [cache_to],
-            }
-        }
-
-    with open(args.output, "w") as f:
-        yaml.dump(ci_compose, f, default_flow_style=False)
-
-    services = list(ci_compose["services"].keys())
-    print(f"Generated {args.output} with cache config for {len(services)} services:")
-    for svc in services:
-        print(f"  - {svc}")
-
-
-if __name__ == "__main__":
-    main()

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

@@ -2,7 +2,7 @@ import asyncio
 import logging
 import uuid
 from datetime import UTC, datetime
-from typing import Any, cast
+from typing import Any
 from weakref import WeakValueDictionary
 
 from openai.types.chat import (
@@ -104,26 +104,6 @@ class ChatSession(BaseModel):
     successful_agent_runs: dict[str, int] = {}
     successful_agent_schedules: dict[str, int] = {}
 
-    def add_tool_call_to_current_turn(self, tool_call: dict) -> None:
-        """Attach a tool_call to the current turn's assistant message.
-
-        Searches backwards for the most recent assistant message (stopping at
-        any user message boundary). If found, appends the tool_call to it.
-        Otherwise creates a new assistant message with the tool_call.
-        """
-        for msg in reversed(self.messages):
-            if msg.role == "user":
-                break
-            if msg.role == "assistant":
-                if not msg.tool_calls:
-                    msg.tool_calls = []
-                msg.tool_calls.append(tool_call)
-                return
-
-        self.messages.append(
-            ChatMessage(role="assistant", content="", tool_calls=[tool_call])
-        )
-
     @staticmethod
     def new(user_id: str) -> "ChatSession":
         return ChatSession(
@@ -192,47 +172,6 @@ class ChatSession(BaseModel):
             successful_agent_schedules=successful_agent_schedules,
         )
 
-    @staticmethod
-    def _merge_consecutive_assistant_messages(
-        messages: list[ChatCompletionMessageParam],
-    ) -> list[ChatCompletionMessageParam]:
-        """Merge consecutive assistant messages into single messages.
-
-        Long-running tool flows can create split assistant messages: one with
-        text content and another with tool_calls. Anthropic's API requires
-        tool_result blocks to reference a tool_use in the immediately preceding
-        assistant message, so these splits cause 400 errors via OpenRouter.
-        """
-        if len(messages) < 2:
-            return messages
-
-        result: list[ChatCompletionMessageParam] = [messages[0]]
-        for msg in messages[1:]:
-            prev = result[-1]
-            if prev.get("role") != "assistant" or msg.get("role") != "assistant":
-                result.append(msg)
-                continue
-
-            prev = cast(ChatCompletionAssistantMessageParam, prev)
-            curr = cast(ChatCompletionAssistantMessageParam, msg)
-
-            curr_content = curr.get("content") or ""
-            if curr_content:
-                prev_content = prev.get("content") or ""
-                prev["content"] = (
-                    f"{prev_content}\n{curr_content}" if prev_content else curr_content
-                )
-
-            curr_tool_calls = curr.get("tool_calls")
-            if curr_tool_calls:
-                prev_tool_calls = prev.get("tool_calls")
-                prev["tool_calls"] = (
-                    list(prev_tool_calls) + list(curr_tool_calls)
-                    if prev_tool_calls
-                    else list(curr_tool_calls)
-                )
-        return result
-
     def to_openai_messages(self) -> list[ChatCompletionMessageParam]:
         messages = []
         for message in self.messages:
@@ -319,7 +258,7 @@ class ChatSession(BaseModel):
                         name=message.name or "",
                     )
                 )
-        return self._merge_consecutive_assistant_messages(messages)
+        return messages
 
 
 async def _get_session_from_cache(session_id: str) -> ChatSession | None:

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

@@ -1,16 +1,4 @@
-from typing import cast
-
 import pytest
-from openai.types.chat import (
-    ChatCompletionAssistantMessageParam,
-    ChatCompletionMessageParam,
-    ChatCompletionToolMessageParam,
-    ChatCompletionUserMessageParam,
-)
-from openai.types.chat.chat_completion_message_tool_call_param import (
-    ChatCompletionMessageToolCallParam,
-    Function,
-)
 
 from .model import (
     ChatMessage,
@@ -129,205 +117,3 @@ async def test_chatsession_db_storage(setup_test_user, test_user_id):
                 loaded.tool_calls is not None
             ), f"Tool calls missing for {orig.role} message"
             assert len(orig.tool_calls) == len(loaded.tool_calls)
-
-
-# --------------------------------------------------------------------------- #
-#  _merge_consecutive_assistant_messages                                       #
-# --------------------------------------------------------------------------- #
-
-_tc = ChatCompletionMessageToolCallParam(
-    id="tc1", type="function", function=Function(name="do_stuff", arguments="{}")
-)
-_tc2 = ChatCompletionMessageToolCallParam(
-    id="tc2", type="function", function=Function(name="other", arguments="{}")
-)
-
-
-def test_merge_noop_when_no_consecutive_assistants():
-    """Messages without consecutive assistants are returned unchanged."""
-    msgs = [
-        ChatCompletionUserMessageParam(role="user", content="hi"),
-        ChatCompletionAssistantMessageParam(role="assistant", content="hello"),
-        ChatCompletionUserMessageParam(role="user", content="bye"),
-    ]
-    merged = ChatSession._merge_consecutive_assistant_messages(msgs)
-    assert len(merged) == 3
-    assert [m["role"] for m in merged] == ["user", "assistant", "user"]
-
-
-def test_merge_splits_text_and_tool_calls():
-    """The exact bug scenario: text-only assistant followed by tool_calls-only assistant."""
-    msgs = [
-        ChatCompletionUserMessageParam(role="user", content="build agent"),
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="Let me build that"
-        ),
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="", tool_calls=[_tc]
-        ),
-        ChatCompletionToolMessageParam(role="tool", content="ok", tool_call_id="tc1"),
-    ]
-    merged = ChatSession._merge_consecutive_assistant_messages(msgs)
-
-    assert len(merged) == 3
-    assert merged[0]["role"] == "user"
-    assert merged[2]["role"] == "tool"
-    a = cast(ChatCompletionAssistantMessageParam, merged[1])
-    assert a["role"] == "assistant"
-    assert a.get("content") == "Let me build that"
-    assert a.get("tool_calls") == [_tc]
-
-
-def test_merge_combines_tool_calls_from_both():
-    """Both consecutive assistants have tool_calls — they get merged."""
-    msgs: list[ChatCompletionAssistantMessageParam] = [
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="text", tool_calls=[_tc]
-        ),
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="", tool_calls=[_tc2]
-        ),
-    ]
-    merged = ChatSession._merge_consecutive_assistant_messages(msgs)  # type: ignore[arg-type]
-
-    assert len(merged) == 1
-    a = cast(ChatCompletionAssistantMessageParam, merged[0])
-    assert a.get("tool_calls") == [_tc, _tc2]
-    assert a.get("content") == "text"
-
-
-def test_merge_three_consecutive_assistants():
-    """Three consecutive assistants collapse into one."""
-    msgs: list[ChatCompletionAssistantMessageParam] = [
-        ChatCompletionAssistantMessageParam(role="assistant", content="a"),
-        ChatCompletionAssistantMessageParam(role="assistant", content="b"),
-        ChatCompletionAssistantMessageParam(
-            role="assistant", content="", tool_calls=[_tc]
-        ),
-    ]
-    merged = ChatSession._merge_consecutive_assistant_messages(msgs)  # type: ignore[arg-type]
-
-    assert len(merged) == 1
-    a = cast(ChatCompletionAssistantMessageParam, merged[0])
-    assert a.get("content") == "a\nb"
-    assert a.get("tool_calls") == [_tc]
-
-
-def test_merge_empty_and_single_message():
-    """Edge cases: empty list and single message."""
-    assert ChatSession._merge_consecutive_assistant_messages([]) == []
-
-    single: list[ChatCompletionMessageParam] = [
-        ChatCompletionUserMessageParam(role="user", content="hi")
-    ]
-    assert ChatSession._merge_consecutive_assistant_messages(single) == single
-
-
-# --------------------------------------------------------------------------- #
-#  add_tool_call_to_current_turn                                               #
-# --------------------------------------------------------------------------- #
-
-_raw_tc = {
-    "id": "tc1",
-    "type": "function",
-    "function": {"name": "f", "arguments": "{}"},
-}
-_raw_tc2 = {
-    "id": "tc2",
-    "type": "function",
-    "function": {"name": "g", "arguments": "{}"},
-}
-
-
-def test_add_tool_call_appends_to_existing_assistant():
-    """When the last assistant is from the current turn, tool_call is added to it."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="user", content="hi"),
-        ChatMessage(role="assistant", content="working on it"),
-    ]
-    session.add_tool_call_to_current_turn(_raw_tc)
-
-    assert len(session.messages) == 2  # no new message created
-    assert session.messages[1].tool_calls == [_raw_tc]
-
-
-def test_add_tool_call_creates_assistant_when_none_exists():
-    """When there's no current-turn assistant, a new one is created."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="user", content="hi"),
-    ]
-    session.add_tool_call_to_current_turn(_raw_tc)
-
-    assert len(session.messages) == 2
-    assert session.messages[1].role == "assistant"
-    assert session.messages[1].tool_calls == [_raw_tc]
-
-
-def test_add_tool_call_does_not_cross_user_boundary():
-    """A user message acts as a boundary — previous assistant is not modified."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="assistant", content="old turn"),
-        ChatMessage(role="user", content="new message"),
-    ]
-    session.add_tool_call_to_current_turn(_raw_tc)
-
-    assert len(session.messages) == 3  # new assistant was created
-    assert session.messages[0].tool_calls is None  # old assistant untouched
-    assert session.messages[2].role == "assistant"
-    assert session.messages[2].tool_calls == [_raw_tc]
-
-
-def test_add_tool_call_multiple_times():
-    """Multiple long-running tool calls accumulate on the same assistant."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="user", content="hi"),
-        ChatMessage(role="assistant", content="doing stuff"),
-    ]
-    session.add_tool_call_to_current_turn(_raw_tc)
-    # Simulate a pending tool result in between (like _yield_tool_call does)
-    session.messages.append(
-        ChatMessage(role="tool", content="pending", tool_call_id="tc1")
-    )
-    session.add_tool_call_to_current_turn(_raw_tc2)
-
-    assert len(session.messages) == 3  # user, assistant, tool — no extra assistant
-    assert session.messages[1].tool_calls == [_raw_tc, _raw_tc2]
-
-
-def test_to_openai_messages_merges_split_assistants():
-    """End-to-end: session with split assistants produces valid OpenAI messages."""
-    session = ChatSession.new(user_id="u")
-    session.messages = [
-        ChatMessage(role="user", content="build agent"),
-        ChatMessage(role="assistant", content="Let me build that"),
-        ChatMessage(
-            role="assistant",
-            content="",
-            tool_calls=[
-                {
-                    "id": "tc1",
-                    "type": "function",
-                    "function": {"name": "create_agent", "arguments": "{}"},
-                }
-            ],
-        ),
-        ChatMessage(role="tool", content="done", tool_call_id="tc1"),
-        ChatMessage(role="assistant", content="Saved!"),
-        ChatMessage(role="user", content="show me an example run"),
-    ]
-    openai_msgs = session.to_openai_messages()
-
-    # The two consecutive assistants at index 1,2 should be merged
-    roles = [m["role"] for m in openai_msgs]
-    assert roles == ["user", "assistant", "tool", "assistant", "user"]
-
-    # The merged assistant should have both content and tool_calls
-    merged = cast(ChatCompletionAssistantMessageParam, openai_msgs[1])
-    assert merged.get("content") == "Let me build that"
-    tc_list = merged.get("tool_calls")
-    assert tc_list is not None and len(list(tc_list)) == 1
-    assert list(tc_list)[0]["id"] == "tc1"

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

@@ -800,13 +800,9 @@ async def stream_chat_completion(
         # Build the messages list in the correct order
         messages_to_save: list[ChatMessage] = []
 
-        # Add assistant message with tool_calls if any.
-        # Use extend (not assign) to preserve tool_calls already added by
-        # _yield_tool_call for long-running tools.
+        # Add assistant message with tool_calls if any
         if accumulated_tool_calls:
-            if not assistant_response.tool_calls:
-                assistant_response.tool_calls = []
-            assistant_response.tool_calls.extend(accumulated_tool_calls)
+            assistant_response.tool_calls = accumulated_tool_calls
             logger.info(
                 f"Added {len(accumulated_tool_calls)} tool calls to assistant message"
             )
@@ -1408,9 +1404,13 @@ async def _yield_tool_call(
                 operation_id=operation_id,
             )
 
-            # Attach the tool_call to the current turn's assistant message
-            # (or create one if this is a tool-only response with no text).
-            session.add_tool_call_to_current_turn(tool_calls[yield_idx])
+            # Save assistant message with tool_call FIRST (required by LLM)
+            assistant_message = ChatMessage(
+                role="assistant",
+                content="",
+                tool_calls=[tool_calls[yield_idx]],
+            )
+            session.messages.append(assistant_message)
 
             # Then save pending tool result
             pending_message = ChatMessage(

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

@@ -0,0 +1,152 @@
+"""Dummy Agent Generator for testing.
+
+Returns mock responses matching the format expected from the external service.
+Enable via AGENTGENERATOR_USE_DUMMY=true in settings.
+
+WARNING: This is for testing only. Do not use in production.
+"""
+
+import logging
+import uuid
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Dummy decomposition result (instructions type)
+DUMMY_DECOMPOSITION_RESULT: dict[str, Any] = {
+    "type": "instructions",
+    "steps": [
+        {
+            "description": "Get input from user",
+            "action": "input",
+            "block_name": "AgentInputBlock",
+        },
+        {
+            "description": "Process the input",
+            "action": "process",
+            "block_name": "TextFormatterBlock",
+        },
+        {
+            "description": "Return output to user",
+            "action": "output",
+            "block_name": "AgentOutputBlock",
+        },
+    ],
+}
+
+# Block IDs from backend/blocks/io.py
+AGENT_INPUT_BLOCK_ID = "c0a8e994-ebf1-4a9c-a4d8-89d09c86741b"
+AGENT_OUTPUT_BLOCK_ID = "363ae599-353e-4804-937e-b2ee3cef3da4"
+
+
+def _generate_dummy_agent_json() -> dict[str, Any]:
+    """Generate a minimal valid agent JSON for testing."""
+    input_node_id = str(uuid.uuid4())
+    output_node_id = str(uuid.uuid4())
+
+    return {
+        "id": str(uuid.uuid4()),
+        "version": 1,
+        "is_active": True,
+        "name": "Dummy Test Agent",
+        "description": "A dummy agent generated for testing purposes",
+        "nodes": [
+            {
+                "id": input_node_id,
+                "block_id": AGENT_INPUT_BLOCK_ID,
+                "input_default": {
+                    "name": "input",
+                    "title": "Input",
+                    "description": "Enter your input",
+                    "placeholder_values": [],
+                },
+                "metadata": {"position": {"x": 0, "y": 0}},
+            },
+            {
+                "id": output_node_id,
+                "block_id": AGENT_OUTPUT_BLOCK_ID,
+                "input_default": {
+                    "name": "output",
+                    "title": "Output",
+                    "description": "Agent output",
+                    "format": "{output}",
+                },
+                "metadata": {"position": {"x": 400, "y": 0}},
+            },
+        ],
+        "links": [
+            {
+                "id": str(uuid.uuid4()),
+                "source_id": input_node_id,
+                "sink_id": output_node_id,
+                "source_name": "result",
+                "sink_name": "value",
+                "is_static": False,
+            },
+        ],
+    }
+
+
+async def decompose_goal_dummy(
+    description: str,
+    context: str = "",
+    library_agents: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Return dummy decomposition result."""
+    logger.info("Using dummy agent generator for decompose_goal")
+    return DUMMY_DECOMPOSITION_RESULT.copy()
+
+
+async def generate_agent_dummy(
+    instructions: dict[str, Any],
+    library_agents: list[dict[str, Any]] | None = None,
+    operation_id: str | None = None,
+    task_id: str | None = None,
+) -> dict[str, Any]:
+    """Return dummy agent JSON."""
+    logger.info("Using dummy agent generator for generate_agent")
+    return _generate_dummy_agent_json()
+
+
+async def generate_agent_patch_dummy(
+    update_request: str,
+    current_agent: dict[str, Any],
+    library_agents: list[dict[str, Any]] | None = None,
+    operation_id: str | None = None,
+    task_id: str | None = None,
+) -> dict[str, Any]:
+    """Return dummy patched agent (returns the current agent with updated description)."""
+    logger.info("Using dummy agent generator for generate_agent_patch")
+    patched = current_agent.copy()
+    patched["description"] = (
+        f"{current_agent.get('description', '')} (updated: {update_request})"
+    )
+    return patched
+
+
+async def customize_template_dummy(
+    template_agent: dict[str, Any],
+    modification_request: str,
+    context: str = "",
+) -> dict[str, Any]:
+    """Return dummy customized template (returns template with updated description)."""
+    logger.info("Using dummy agent generator for customize_template")
+    customized = template_agent.copy()
+    customized["description"] = (
+        f"{template_agent.get('description', '')} (customized: {modification_request})"
+    )
+    return customized
+
+
+async def get_blocks_dummy() -> list[dict[str, Any]]:
+    """Return dummy blocks list."""
+    logger.info("Using dummy agent generator for get_blocks")
+    return [
+        {"id": AGENT_INPUT_BLOCK_ID, "name": "AgentInputBlock"},
+        {"id": AGENT_OUTPUT_BLOCK_ID, "name": "AgentOutputBlock"},
+    ]
+
+
+async def health_check_dummy() -> bool:
+    """Always returns healthy for dummy service."""
+    return True

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

@@ -12,8 +12,19 @@ import httpx
 
 from backend.util.settings import Settings
 
+from .dummy import (
+    customize_template_dummy,
+    decompose_goal_dummy,
+    generate_agent_dummy,
+    generate_agent_patch_dummy,
+    get_blocks_dummy,
+    health_check_dummy,
+)
+
 logger = logging.getLogger(__name__)
 
+_dummy_mode_warned = False
+
 
 def _create_error_response(
     error_message: str,
@@ -90,10 +101,26 @@ def _get_settings() -> Settings:
     return _settings
 
 
-def is_external_service_configured() -> bool:
-    """Check if external Agent Generator service is configured."""
+def _is_dummy_mode() -> bool:
+    """Check if dummy mode is enabled for testing."""
+    global _dummy_mode_warned
     settings = _get_settings()
-    return bool(settings.config.agentgenerator_host)
+    is_dummy = bool(settings.config.agentgenerator_use_dummy)
+    if is_dummy and not _dummy_mode_warned:
+        logger.warning(
+            "Agent Generator running in DUMMY MODE - returning mock responses. "
+            "Do not use in production!"
+        )
+        _dummy_mode_warned = True
+    return is_dummy
+
+
+def is_external_service_configured() -> bool:
+    """Check if external Agent Generator service is configured (or dummy mode)."""
+    settings = _get_settings()
+    return bool(settings.config.agentgenerator_host) or bool(
+        settings.config.agentgenerator_use_dummy
+    )
 
 
 def _get_base_url() -> str:
@@ -137,6 +164,9 @@ async def decompose_goal_external(
         - {"type": "error", "error": "...", "error_type": "..."} on error
         Or None on unexpected error
     """
+    if _is_dummy_mode():
+        return await decompose_goal_dummy(description, context, library_agents)
+
     client = _get_client()
 
     if context:
@@ -226,6 +256,11 @@ async def generate_agent_external(
     Returns:
         Agent JSON dict, {"status": "accepted"} for async, or error dict {"type": "error", ...} on error
     """
+    if _is_dummy_mode():
+        return await generate_agent_dummy(
+            instructions, library_agents, operation_id, task_id
+        )
+
     client = _get_client()
 
     # Build request payload
@@ -297,6 +332,11 @@ async def generate_agent_patch_external(
     Returns:
         Updated agent JSON, clarifying questions dict, {"status": "accepted"} for async, or error dict on error
     """
+    if _is_dummy_mode():
+        return await generate_agent_patch_dummy(
+            update_request, current_agent, library_agents, operation_id, task_id
+        )
+
     client = _get_client()
 
     # Build request payload
@@ -383,6 +423,11 @@ async def customize_template_external(
     Returns:
         Customized agent JSON, clarifying questions dict, or error dict on error
     """
+    if _is_dummy_mode():
+        return await customize_template_dummy(
+            template_agent, modification_request, context
+        )
+
     client = _get_client()
 
     request = modification_request
@@ -445,6 +490,9 @@ async def get_blocks_external() -> list[dict[str, Any]] | None:
     Returns:
         List of block info dicts or None on error
     """
+    if _is_dummy_mode():
+        return await get_blocks_dummy()
+
     client = _get_client()
 
     try:
@@ -478,6 +526,9 @@ async def health_check() -> bool:
     if not is_external_service_configured():
         return False
 
+    if _is_dummy_mode():
+        return await health_check_dummy()
+
     client = _get_client()
 
     try:

autogpt_platform/backend/backend/util/settings.py

@@ -368,6 +368,10 @@ class Config(UpdateTrackingModel["Config"], BaseSettings):
         default=600,
         description="The timeout in seconds for Agent Generator service requests (includes retries for rate limits)",
     )
+    agentgenerator_use_dummy: bool = Field(
+        default=False,
+        description="Use dummy agent generator responses for testing (bypasses external service)",
+    )
 
     enable_example_blocks: bool = Field(
         default=False,