github/AutoGPT
1
1
Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-02-02 02:45:18 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' into kpczerwinski/secrt-1778-update-new-builder-search

Browse Source
This commit is contained in:
Krzysztof Czerwinski
2026-01-28 14:41:46 +09:00
parent 8f7429e3fd 0058cd3ba6
commit ca1df77c20
72 changed files with 3943 additions and 1914 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
autogpt_platform/backend/backend/api/features/chat/config.py
View File

@@ -33,9 +33,15 @@ 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=3, description="Maximum number of agent runs")
max_agent_runs: int = Field(default=30, description="Maximum number of agent runs")
max_agent_schedules: int = Field(
default=3, description="Maximum number of agent schedules"
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)",
)
# Langfuse Prompt Management Configuration

42
autogpt_platform/backend/backend/api/features/chat/db.py
View File

@@ -247,3 +247,45 @@ 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

15
autogpt_platform/backend/backend/api/features/chat/model.py
View File

@@ -295,6 +295,21 @@ 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)

18
autogpt_platform/backend/backend/api/features/chat/response_model.py
View File

@@ -31,6 +31,7 @@ class ResponseType(str, Enum):
# Other
ERROR = "error"
USAGE = "usage"
HEARTBEAT = "heartbeat"
class StreamBaseResponse(BaseModel):
@@ -142,3 +143,20 @@ 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"

1732
autogpt_platform/backend/backend/api/features/chat/service.py
View File

File diff suppressed because it is too large Load Diff

7
autogpt_platform/backend/backend/api/features/chat/tools/__init__.py
View File

@@ -49,6 +49,11 @@ 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],
@@ -57,7 +62,7 @@ async def execute_tool(
tool_call_id: str,
) -> "StreamToolOutputAvailable":
"""Execute a tool by name."""
tool = TOOL_REGISTRY.get(tool_name)
tool = get_tool(tool_name)
if not tool:
raise ValueError(f"Tool {tool_name} not found")

3
autogpt_platform/backend/backend/api/features/chat/tools/add_understanding.py
View File

@@ -3,8 +3,6 @@
import logging
from typing import Any
from langfuse import observe
from backend.api.features.chat.model import ChatSession
from backend.data.understanding import (
BusinessUnderstandingInput,
@@ -61,7 +59,6 @@ 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,

2
autogpt_platform/backend/backend/api/features/chat/tools/agent_output.py
View File

@@ -5,7 +5,6 @@ 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
@@ -329,7 +328,6 @@ 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,

10
autogpt_platform/backend/backend/api/features/chat/tools/base.py
View File

@@ -36,6 +36,16 @@ 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(

21
autogpt_platform/backend/backend/api/features/chat/tools/create_agent.py
View File

@@ -3,8 +3,6 @@
import logging
from typing import Any
from langfuse import observe
from backend.api.features.chat.model import ChatSession
from .agent_generator import (
@@ -44,6 +42,10 @@ 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 {
@@ -75,7 +77,6 @@ class CreateAgentTool(BaseTool):
"required": ["description"],
}
@observe(as_type="tool", name="create_agent")
async def _execute(
self,
user_id: str | None,
@@ -116,8 +117,11 @@ class CreateAgentTool(BaseTool):
if decomposition_result is None:
return ErrorResponse(
message="Failed to analyze the goal. Please try rephrasing.",
error="Decomposition failed",
message="Failed to analyze the goal. The agent generation service may be unavailable or timed out. Please try again.",
error="decomposition_failed",
details={
"description": description[:100]
}, # Include context for debugging
session_id=session_id,
)
@@ -182,8 +186,11 @@ class CreateAgentTool(BaseTool):
if agent_json is None:
return ErrorResponse(
message="Failed to generate the agent. Please try again.",
error="Generation failed",
message="Failed to generate the agent. The agent generation service may be unavailable or timed out. Please try again.",
error="generation_failed",
details={
"description": description[:100]
}, # Include context for debugging
session_id=session_id,
)

12
autogpt_platform/backend/backend/api/features/chat/tools/edit_agent.py
View File

@@ -3,8 +3,6 @@
import logging
from typing import Any
from langfuse import observe
from backend.api.features.chat.model import ChatSession
from .agent_generator import (
@@ -44,6 +42,10 @@ 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 {
@@ -81,7 +83,6 @@ class EditAgentTool(BaseTool):
"required": ["agent_id", "changes"],
}
@observe(as_type="tool", name="edit_agent")
async def _execute(
self,
user_id: str | None,
@@ -145,8 +146,9 @@ class EditAgentTool(BaseTool):
if result is None:
return ErrorResponse(
message="Failed to generate changes. Please try rephrasing.",
error="Update generation failed",
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,
)

3
autogpt_platform/backend/backend/api/features/chat/tools/find_agent.py
View File

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

5
autogpt_platform/backend/backend/api/features/chat/tools/find_block.py
View File

@@ -1,7 +1,6 @@
import logging
from typing import Any
from langfuse import observe
from prisma.enums import ContentType
from backend.api.features.chat.model import ChatSession
@@ -56,7 +55,6 @@ 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,
@@ -109,7 +107,8 @@ class FindBlockTool(BaseTool):
block_id = result["content_id"]
block = get_block(block_id)
if block:
# Skip disabled blocks
if block and not block.disabled:
# Get input/output schemas
input_schema = {}
output_schema = {}

3
autogpt_platform/backend/backend/api/features/chat/tools/find_library_agent.py
View File

@@ -2,8 +2,6 @@
from typing import Any
from langfuse import observe
from backend.api.features.chat.model import ChatSession
from .agent_search import search_agents
@@ -43,7 +41,6 @@ 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:

3
autogpt_platform/backend/backend/api/features/chat/tools/get_doc_page.py
View File

@@ -4,8 +4,6 @@ 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 (
@@ -73,7 +71,6 @@ 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,

40
autogpt_platform/backend/backend/api/features/chat/tools/models.py
View File

@@ -28,6 +28,10 @@ class ResponseType(str, Enum):
BLOCK_OUTPUT = "block_output"
DOC_SEARCH_RESULTS = "doc_search_results"
DOC_PAGE = "doc_page"
# Long-running operation types
OPERATION_STARTED = "operation_started"
OPERATION_PENDING = "operation_pending"
OPERATION_IN_PROGRESS = "operation_in_progress"
# Base response model
@@ -334,3 +338,39 @@ 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

2
autogpt_platform/backend/backend/api/features/chat/tools/run_agent.py
View File

@@ -3,7 +3,6 @@
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
@@ -159,7 +158,6 @@ class RunAgentTool(BaseTool):
"""All operations require authentication."""
return True
@observe(as_type="tool", name="run_agent")
async def _execute(
self,
user_id: str | None,

3
autogpt_platform/backend/backend/api/features/chat/tools/run_block.py
View File

@@ -4,8 +4,6 @@ import logging
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
@@ -130,7 +128,6 @@ class RunBlockTool(BaseTool):
return matched_credentials, missing_credentials
@observe(as_type="tool", name="run_block")
async def _execute(
self,
user_id: str | None,

2
autogpt_platform/backend/backend/api/features/chat/tools/search_docs.py
View File

@@ -3,7 +3,6 @@
import logging
from typing import Any
from langfuse import observe
from prisma.enums import ContentType
from backend.api.features.chat.model import ChatSession
@@ -88,7 +87,6 @@ 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,

44
autogpt_platform/backend/backend/api/features/executions/review/review_routes_test.py
View File

@@ -164,9 +164,9 @@ async def test_process_review_action_approve_success(
"""Test successful review approval"""
# Mock the route functions
# Mock get_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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 "No pending review found" in response.json()["detail"]
assert "Review(s) not 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_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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 "No pending review found" in response.json()["detail"]
assert "Review(s) not 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_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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_pending_reviews_by_node_exec_ids (called to find the graph_exec_id)
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_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_pending_reviews_by_node_exec_ids - return different reviews based on node_exec_id
# Mock get_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_pending_reviews_by_node_exec_ids"
"backend.api.features.executions.review.routes.get_reviews_by_node_exec_ids"
)
# Create a mapping of node_exec_id to review

10
autogpt_platform/backend/backend/api/features/executions/review/routes.py
View File

@@ -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
reviews_map = await get_pending_reviews_by_node_exec_ids(
# Batch fetch all requested reviews (regardless of status for idempotent handling)
reviews_map = await get_reviews_by_node_exec_ids(
list(all_request_node_ids), user_id
)
# Validate all reviews were found
# Validate all reviews were found (must exist, any status is OK for now)
missing_ids = all_request_node_ids - set(reviews_map.keys())
if missing_ids:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"No pending review found for node execution(s): {', '.join(missing_ids)}",
detail=f"Review(s) not found: {', '.join(missing_ids)}",
)
# Validate all reviews belong to the same execution

15
autogpt_platform/backend/backend/api/features/store/content_handlers.py
View File

@@ -188,6 +188,10 @@ 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:
@@ -273,12 +277,19 @@ class BlockHandler(ContentHandler):
from backend.data.block import get_blocks
all_blocks = get_blocks()
total_blocks = len(all_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)
if total_blocks == 0:
return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}
block_ids = list(all_blocks.keys())
block_ids = enabled_block_ids
placeholders = ",".join([f"${i+1}" for i in range(len(block_ids))])
embedded_result = await query_raw_with_schema(

18
autogpt_platform/backend/backend/api/features/store/content_handlers_test.py
View File

@@ -81,6 +81,7 @@ 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"}}
}
@@ -116,11 +117,18 @@ async def test_block_handler_get_stats(mocker):
"""Test BlockHandler returns correct stats."""
handler = BlockHandler()
# Mock get_blocks
# 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_blocks = {
"block-1": MagicMock(),
"block-2": MagicMock(),
"block-3": MagicMock(),
"block-1": make_mock_block_class(),
"block-2": make_mock_block_class(),
"block-3": make_mock_block_class(),
}
# Mock embedded count query (2 blocks have embeddings)
@@ -309,6 +317,7 @@ 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
@@ -342,6 +351,7 @@ 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()

65
autogpt_platform/backend/backend/data/human_review.py
View File

@@ -263,11 +263,14 @@ async def get_pending_review_by_node_exec_id(
return PendingHumanReviewModel.from_db(review, node_id=node_id)
async def get_pending_reviews_by_node_exec_ids(
async def get_reviews_by_node_exec_ids(
node_exec_ids: list[str], user_id: str
) -> dict[str, "PendingHumanReviewModel"]:
"""
Get multiple pending reviews by their node execution IDs in a single batch query.
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.
Args:
node_exec_ids: List of node execution IDs to look up
@@ -283,7 +286,6 @@ async def get_pending_reviews_by_node_exec_ids(
where={
"nodeExecId": {"in": node_exec_ids},
"userId": user_id,
"status": ReviewStatus.WAITING,
}
)
@@ -407,38 +409,68 @@ 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
Dict of node_exec_id -> updated review model (includes already-processed reviews)
"""
if not review_decisions:
return {}
node_exec_ids = list(review_decisions.keys())
# Get all reviews for validation
reviews = await PendingHumanReview.prisma().find_many(
# Get all reviews (both WAITING and already processed) for the user
all_reviews = await PendingHumanReview.prisma().find_many(
where={
"nodeExecId": {"in": node_exec_ids},
"userId": user_id,
"status": ReviewStatus.WAITING,
},
)
# 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}
# 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:
raise ValueError(
f"Reviews not found, access denied, or not in WAITING status: {', '.join(missing_ids)}"
f"Reviews not found or access denied: {', '.join(missing_ids)}"
)
# Create parallel update tasks
# 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
update_tasks = []
for review in reviews:
for review in reviews_to_process:
new_status, reviewed_data, message = review_decisions[review.nodeExecId]
has_data_changes = reviewed_data is not None and reviewed_data != review.payload
@@ -463,7 +495,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)
updated_reviews = await asyncio.gather(*update_tasks) if update_tasks else []
# Note: Execution resumption is now handled at the API layer after ALL reviews
# for an execution are processed (both approved and rejected)
@@ -472,8 +504,11 @@ 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 updated_reviews:
for review in all_result_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(

4
autogpt_platform/backend/backend/util/settings.py
View File

@@ -359,8 +359,8 @@ class Config(UpdateTrackingModel["Config"], BaseSettings):
description="The port for the Agent Generator service",
)
agentgenerator_timeout: int = Field(
default=120,
description="The timeout in seconds for Agent Generator service requests",
default=600,
description="The timeout in seconds for Agent Generator service requests (includes retries for rate limits)",
)
enable_example_blocks: bool = Field(

41
autogpt_platform/frontend/src/app/(platform)/copilot/NewChatContext.tsx
View File

@@ -1,41 +0,0 @@
"use client";
import { createContext, useContext, useRef, type ReactNode } from "react";
interface NewChatContextValue {
onNewChatClick: () => void;
setOnNewChatClick: (handler?: () => void) => void;
performNewChat?: () => void;
setPerformNewChat: (handler?: () => void) => void;
}
const NewChatContext = createContext<NewChatContextValue | null>(null);
export function NewChatProvider({ children }: { children: ReactNode }) {
const onNewChatRef = useRef<(() => void) | undefined>();
const performNewChatRef = useRef<(() => void) | undefined>();
const contextValueRef = useRef<NewChatContextValue>({
onNewChatClick() {
onNewChatRef.current?.();
},
setOnNewChatClick(handler?: () => void) {
onNewChatRef.current = handler;
},
performNewChat() {
performNewChatRef.current?.();
},
setPerformNewChat(handler?: () => void) {
performNewChatRef.current = handler;
},
});
return (
<NewChatContext.Provider value={contextValueRef.current}>
{children}
</NewChatContext.Provider>
);
}
export function useNewChat() {
return useContext(NewChatContext);
}

46
autogpt_platform/frontend/src/app/(platform)/copilot/components/CopilotShell/CopilotShell.tsx
View File

@@ -1,12 +1,10 @@
"use client";
import { ChatLoader } from "@/components/contextual/Chat/components/ChatLoader/ChatLoader";
import { Text } from "@/components/atoms/Text/Text";
import { NAVBAR_HEIGHT_PX } from "@/lib/constants";
import type { ReactNode } from "react";
import { useEffect } from "react";
import { useNewChat } from "../../NewChatContext";
import { DesktopSidebar } from "./components/DesktopSidebar/DesktopSidebar";
import { LoadingState } from "./components/LoadingState/LoadingState";
import { MobileDrawer } from "./components/MobileDrawer/MobileDrawer";
import { MobileHeader } from "./components/MobileHeader/MobileHeader";
import { useCopilotShell } from "./useCopilotShell";
@@ -20,36 +18,21 @@ export function CopilotShell({ children }: Props) {
isMobile,
isDrawerOpen,
isLoading,
isCreatingSession,
isLoggedIn,
hasActiveSession,
sessions,
currentSessionId,
handleSelectSession,
handleOpenDrawer,
handleCloseDrawer,
handleDrawerOpenChange,
handleNewChat,
handleNewChatClick,
handleSessionClick,
hasNextPage,
isFetchingNextPage,
fetchNextPage,
isReadyToShowContent,
} = useCopilotShell();
const newChatContext = useNewChat();
const handleNewChatClickWrapper =
newChatContext?.onNewChatClick || handleNewChat;
useEffect(
function registerNewChatHandler() {
if (!newChatContext) return;
newChatContext.setPerformNewChat(handleNewChat);
return function cleanup() {
newChatContext.setPerformNewChat(undefined);
};
},
[newChatContext, handleNewChat],
);
if (!isLoggedIn) {
return (
<div className="flex h-full items-center justify-center">
@@ -70,9 +53,9 @@ export function CopilotShell({ children }: Props) {
isLoading={isLoading}
hasNextPage={hasNextPage}
isFetchingNextPage={isFetchingNextPage}
onSelectSession={handleSelectSession}
onSelectSession={handleSessionClick}
onFetchNextPage={fetchNextPage}
onNewChat={handleNewChatClickWrapper}
onNewChat={handleNewChatClick}
hasActiveSession={Boolean(hasActiveSession)}
/>
)}
@@ -80,7 +63,18 @@ export function CopilotShell({ children }: Props) {
<div className="relative flex min-h-0 flex-1 flex-col">
{isMobile && <MobileHeader onOpenDrawer={handleOpenDrawer} />}
<div className="flex min-h-0 flex-1 flex-col">
{isReadyToShowContent ? children : <LoadingState />}
{isCreatingSession ? (
<div className="flex h-full flex-1 flex-col items-center justify-center bg-[#f8f8f9]">
<div className="flex flex-col items-center gap-4">
<ChatLoader />
<Text variant="body" className="text-zinc-500">
Creating your chat...
</Text>
</div>
</div>
) : (
children
)}
</div>
</div>
@@ -92,9 +86,9 @@ export function CopilotShell({ children }: Props) {
isLoading={isLoading}
hasNextPage={hasNextPage}
isFetchingNextPage={isFetchingNextPage}
onSelectSession={handleSelectSession}
onSelectSession={handleSessionClick}
onFetchNextPage={fetchNextPage}
onNewChat={handleNewChatClickWrapper}
onNewChat={handleNewChatClick}
onClose={handleCloseDrawer}
onOpenChange={handleDrawerOpenChange}
hasActiveSession={Boolean(hasActiveSession)}

15
autogpt_platform/frontend/src/app/(platform)/copilot/components/CopilotShell/components/LoadingState/LoadingState.tsx
View File

@@ -1,15 +0,0 @@
import { Text } from "@/components/atoms/Text/Text";
import { ChatLoader } from "@/components/contextual/Chat/components/ChatLoader/ChatLoader";
export function LoadingState() {
return (
<div className="flex flex-1 items-center justify-center">
<div className="flex flex-col items-center gap-4">
<ChatLoader />
<Text variant="body" className="text-zinc-500">
Loading your chats...
</Text>
</div>
</div>
);
}

12
autogpt_platform/frontend/src/app/(platform)/copilot/components/CopilotShell/components/MobileDrawer/useMobileDrawer.ts
View File

@@ -3,17 +3,17 @@ import { useState } from "react";
export function useMobileDrawer() {
const [isDrawerOpen, setIsDrawerOpen] = useState(false);
function handleOpenDrawer() {
const handleOpenDrawer = () => {
setIsDrawerOpen(true);
}
};
function handleCloseDrawer() {
const handleCloseDrawer = () => {
setIsDrawerOpen(false);
}
};
function handleDrawerOpenChange(open: boolean) {
const handleDrawerOpenChange = (open: boolean) => {
setIsDrawerOpen(open);
}
};
return {
isDrawerOpen,

29
autogpt_platform/frontend/src/app/(platform)/copilot/components/CopilotShell/components/SessionsList/useSessionsPagination.ts
View File

@@ -1,7 +1,7 @@
import { useGetV2ListSessions } from "@/app/api/__generated__/endpoints/chat/chat";
import type { SessionSummaryResponse } from "@/app/api/__generated__/models/sessionSummaryResponse";
import { okData } from "@/app/api/helpers";
import { useEffect, useMemo, useState } from "react";
import { useEffect, useState } from "react";
const PAGE_SIZE = 50;
@@ -11,9 +11,11 @@ export interface UseSessionsPaginationArgs {
export function useSessionsPagination({ enabled }: UseSessionsPaginationArgs) {
const [offset, setOffset] = useState(0);
const [accumulatedSessions, setAccumulatedSessions] = useState<
SessionSummaryResponse[]
>([]);
const [totalCount, setTotalCount] = useState<number | null>(null);
const { data, isLoading, isFetching, isError } = useGetV2ListSessions(
@@ -43,17 +45,14 @@ export function useSessionsPagination({ enabled }: UseSessionsPaginationArgs) {
}
}, [data, offset, enabled]);
const hasNextPage = useMemo(() => {
if (totalCount === null) return false;
return accumulatedSessions.length < totalCount;
}, [accumulatedSessions.length, totalCount]);
const hasNextPage =
totalCount !== null && accumulatedSessions.length < totalCount;
const areAllSessionsLoaded = useMemo(() => {
if (totalCount === null) return false;
return (
accumulatedSessions.length >= totalCount && !isFetching && !isLoading
);
}, [accumulatedSessions.length, totalCount, isFetching, isLoading]);
const areAllSessionsLoaded =
totalCount !== null &&
accumulatedSessions.length >= totalCount &&
!isFetching &&
!isLoading;
useEffect(() => {
if (
@@ -67,17 +66,17 @@ export function useSessionsPagination({ enabled }: UseSessionsPaginationArgs) {
}
}, [hasNextPage, isFetching, isLoading, isError, totalCount]);
function fetchNextPage() {
const fetchNextPage = () => {
if (hasNextPage && !isFetching) {
setOffset((prev) => prev + PAGE_SIZE);
}
}
};
function reset() {
const reset = () => {
setOffset(0);
setAccumulatedSessions([]);
setTotalCount(null);
}
};
return {
sessions: accumulatedSessions,

139
autogpt_platform/frontend/src/app/(platform)/copilot/components/CopilotShell/helpers.ts
View File

@@ -2,9 +2,7 @@ import type { SessionDetailResponse } from "@/app/api/__generated__/models/sessi
import type { SessionSummaryResponse } from "@/app/api/__generated__/models/sessionSummaryResponse";
import { format, formatDistanceToNow, isToday } from "date-fns";
export function convertSessionDetailToSummary(
session: SessionDetailResponse,
): SessionSummaryResponse {
export function convertSessionDetailToSummary(session: SessionDetailResponse) {
return {
id: session.id,
created_at: session.created_at,
@@ -13,17 +11,25 @@ export function convertSessionDetailToSummary(
};
}
export function filterVisibleSessions(
sessions: SessionSummaryResponse[],
): SessionSummaryResponse[] {
return sessions.filter(
(session) => session.updated_at !== session.created_at,
);
export function filterVisibleSessions(sessions: SessionSummaryResponse[]) {
const fiveMinutesAgo = Date.now() - 5 * 60 * 1000;
return sessions.filter((session) => {
const hasBeenUpdated = session.updated_at !== session.created_at;
if (hasBeenUpdated) return true;
const isRecentlyCreated =
new Date(session.created_at).getTime() > fiveMinutesAgo;
return isRecentlyCreated;
});
}
export function getSessionTitle(session: SessionSummaryResponse): string {
export function getSessionTitle(session: SessionSummaryResponse) {
if (session.title) return session.title;
const isNewSession = session.updated_at === session.created_at;
if (isNewSession) {
const createdDate = new Date(session.created_at);
if (isToday(createdDate)) {
@@ -31,12 +37,11 @@ export function getSessionTitle(session: SessionSummaryResponse): string {
}
return format(createdDate, "MMM d, yyyy");
}
return "Untitled Chat";
}
export function getSessionUpdatedLabel(
session: SessionSummaryResponse,
): string {
export function getSessionUpdatedLabel(session: SessionSummaryResponse) {
if (!session.updated_at) return "";
return formatDistanceToNow(new Date(session.updated_at), { addSuffix: true });
}
@@ -45,8 +50,10 @@ export function mergeCurrentSessionIntoList(
accumulatedSessions: SessionSummaryResponse[],
currentSessionId: string | null,
currentSessionData: SessionDetailResponse | null | undefined,
): SessionSummaryResponse[] {
recentlyCreatedSessions?: Map<string, SessionSummaryResponse>,
) {
const filteredSessions: SessionSummaryResponse[] = [];
const addedIds = new Set<string>();
if (accumulatedSessions.length > 0) {
const visibleSessions = filterVisibleSessions(accumulatedSessions);
@@ -61,105 +68,39 @@ export function mergeCurrentSessionIntoList(
);
if (!isInVisible) {
filteredSessions.push(currentInAll);
addedIds.add(currentInAll.id);
}
}
}
filteredSessions.push(...visibleSessions);
for (const session of visibleSessions) {
if (!addedIds.has(session.id)) {
filteredSessions.push(session);
addedIds.add(session.id);
}
}
}
if (currentSessionId && currentSessionData) {
const isCurrentInList = filteredSessions.some(
(s) => s.id === currentSessionId,
);
if (!isCurrentInList) {
if (!addedIds.has(currentSessionId)) {
const summarySession = convertSessionDetailToSummary(currentSessionData);
filteredSessions.unshift(summarySession);
addedIds.add(currentSessionId);
}
}
if (recentlyCreatedSessions) {
for (const [sessionId, sessionData] of recentlyCreatedSessions) {
if (!addedIds.has(sessionId)) {
filteredSessions.unshift(sessionData);
addedIds.add(sessionId);
}
}
}
return filteredSessions;
}
export function getCurrentSessionId(
searchParams: URLSearchParams,
): string | null {
export function getCurrentSessionId(searchParams: URLSearchParams) {
return searchParams.get("sessionId");
}
export function shouldAutoSelectSession(
areAllSessionsLoaded: boolean,
hasAutoSelectedSession: boolean,
paramSessionId: string | null,
visibleSessions: SessionSummaryResponse[],
accumulatedSessions: SessionSummaryResponse[],
isLoading: boolean,
totalCount: number | null,
): {
shouldSelect: boolean;
sessionIdToSelect: string | null;
shouldCreate: boolean;
} {
if (!areAllSessionsLoaded || hasAutoSelectedSession) {
return {
shouldSelect: false,
sessionIdToSelect: null,
shouldCreate: false,
};
}
if (paramSessionId) {
return {
shouldSelect: false,
sessionIdToSelect: null,
shouldCreate: false,
};
}
if (visibleSessions.length > 0) {
return {
shouldSelect: true,
sessionIdToSelect: visibleSessions[0].id,
shouldCreate: false,
};
}
if (accumulatedSessions.length === 0 && !isLoading && totalCount === 0) {
return { shouldSelect: false, sessionIdToSelect: null, shouldCreate: true };
}
if (totalCount === 0) {
return {
shouldSelect: false,
sessionIdToSelect: null,
shouldCreate: false,
};
}
return { shouldSelect: false, sessionIdToSelect: null, shouldCreate: false };
}
export function checkReadyToShowContent(
areAllSessionsLoaded: boolean,
paramSessionId: string | null,
accumulatedSessions: SessionSummaryResponse[],
isCurrentSessionLoading: boolean,
currentSessionData: SessionDetailResponse | null | undefined,
hasAutoSelectedSession: boolean,
): boolean {
if (!areAllSessionsLoaded) return false;
if (paramSessionId) {
const sessionFound = accumulatedSessions.some(
(s) => s.id === paramSessionId,
);
return (
sessionFound ||
(!isCurrentSessionLoading &&
currentSessionData !== undefined &&
currentSessionData !== null)
);
}
return hasAutoSelectedSession;
}

201
autogpt_platform/frontend/src/app/(platform)/copilot/components/CopilotShell/useCopilotShell.ts
View File

@@ -1,26 +1,24 @@
"use client";
import {
getGetV2GetSessionQueryKey,
getGetV2ListSessionsQueryKey,
useGetV2GetSession,
} from "@/app/api/__generated__/endpoints/chat/chat";
import { okData } from "@/app/api/helpers";
import { useChatStore } from "@/components/contextual/Chat/chat-store";
import { useBreakpoint } from "@/lib/hooks/useBreakpoint";
import { useSupabase } from "@/lib/supabase/hooks/useSupabase";
import { useQueryClient } from "@tanstack/react-query";
import { usePathname, useRouter, useSearchParams } from "next/navigation";
import { useEffect, useRef, useState } from "react";
import { usePathname, useSearchParams } from "next/navigation";
import { useRef } from "react";
import { useCopilotStore } from "../../copilot-page-store";
import { useCopilotSessionId } from "../../useCopilotSessionId";
import { useMobileDrawer } from "./components/MobileDrawer/useMobileDrawer";
import { useSessionsPagination } from "./components/SessionsList/useSessionsPagination";
import {
checkReadyToShowContent,
filterVisibleSessions,
getCurrentSessionId,
mergeCurrentSessionIntoList,
} from "./helpers";
import { getCurrentSessionId } from "./helpers";
import { useShellSessionList } from "./useShellSessionList";
export function useCopilotShell() {
const router = useRouter();
const pathname = usePathname();
const searchParams = useSearchParams();
const queryClient = useQueryClient();
@@ -29,6 +27,8 @@ export function useCopilotShell() {
const isMobile =
breakpoint === "base" || breakpoint === "sm" || breakpoint === "md";
const { urlSessionId, setUrlSessionId } = useCopilotSessionId();
const isOnHomepage = pathname === "/copilot";
const paramSessionId = searchParams.get("sessionId");
@@ -41,114 +41,113 @@ export function useCopilotShell() {
const paginationEnabled = !isMobile || isDrawerOpen || !!paramSessionId;
const {
sessions: accumulatedSessions,
isLoading: isSessionsLoading,
isFetching: isSessionsFetching,
hasNextPage,
areAllSessionsLoaded,
fetchNextPage,
reset: resetPagination,
} = useSessionsPagination({
enabled: paginationEnabled,
});
const currentSessionId = getCurrentSessionId(searchParams);
const { data: currentSessionData, isLoading: isCurrentSessionLoading } =
useGetV2GetSession(currentSessionId || "", {
const { data: currentSessionData } = useGetV2GetSession(
currentSessionId || "",
{
query: {
enabled: !!currentSessionId,
select: okData,
},
});
const [hasAutoSelectedSession, setHasAutoSelectedSession] = useState(false);
const hasAutoSelectedRef = useRef(false);
// Mark as auto-selected when sessionId is in URL
useEffect(() => {
if (paramSessionId && !hasAutoSelectedRef.current) {
hasAutoSelectedRef.current = true;
setHasAutoSelectedSession(true);
}
}, [paramSessionId]);
// On homepage without sessionId, mark as ready immediately
useEffect(() => {
if (isOnHomepage && !paramSessionId && !hasAutoSelectedRef.current) {
hasAutoSelectedRef.current = true;
setHasAutoSelectedSession(true);
}
}, [isOnHomepage, paramSessionId]);
// Invalidate sessions list when navigating to homepage (to show newly created sessions)
useEffect(() => {
if (isOnHomepage && !paramSessionId) {
queryClient.invalidateQueries({
queryKey: getGetV2ListSessionsQueryKey(),
});
}
}, [isOnHomepage, paramSessionId, queryClient]);
// Reset pagination when query becomes disabled
const prevPaginationEnabledRef = useRef(paginationEnabled);
useEffect(() => {
if (prevPaginationEnabledRef.current && !paginationEnabled) {
resetPagination();
resetAutoSelect();
}
prevPaginationEnabledRef.current = paginationEnabled;
}, [paginationEnabled, resetPagination]);
const sessions = mergeCurrentSessionIntoList(
accumulatedSessions,
currentSessionId,
currentSessionData,
},
);
const visibleSessions = filterVisibleSessions(sessions);
const {
sessions,
isLoading,
isSessionsFetching,
hasNextPage,
fetchNextPage,
resetPagination,
recentlyCreatedSessionsRef,
} = useShellSessionList({
paginationEnabled,
currentSessionId,
currentSessionData,
isOnHomepage,
paramSessionId,
});
const sidebarSelectedSessionId =
isOnHomepage && !paramSessionId ? null : currentSessionId;
const stopStream = useChatStore((s) => s.stopStream);
const onStreamComplete = useChatStore((s) => s.onStreamComplete);
const isStreaming = useCopilotStore((s) => s.isStreaming);
const isCreatingSession = useCopilotStore((s) => s.isCreatingSession);
const setIsSwitchingSession = useCopilotStore((s) => s.setIsSwitchingSession);
const openInterruptModal = useCopilotStore((s) => s.openInterruptModal);
const isReadyToShowContent = isOnHomepage
? true
: checkReadyToShowContent(
areAllSessionsLoaded,
paramSessionId,
accumulatedSessions,
isCurrentSessionLoading,
currentSessionData,
hasAutoSelectedSession,
);
const pendingActionRef = useRef<(() => void) | null>(null);
function handleSelectSession(sessionId: string) {
// Navigate using replaceState to avoid full page reload
window.history.replaceState(null, "", `/copilot?sessionId=${sessionId}`);
// Force a re-render by updating the URL through router
router.replace(`/copilot?sessionId=${sessionId}`);
async function stopCurrentStream() {
if (!currentSessionId) return;
setIsSwitchingSession(true);
await new Promise<void>((resolve) => {
const unsubscribe = onStreamComplete((completedId) => {
if (completedId === currentSessionId) {
clearTimeout(timeout);
unsubscribe();
resolve();
}
});
const timeout = setTimeout(() => {
unsubscribe();
resolve();
}, 3000);
stopStream(currentSessionId);
});
queryClient.invalidateQueries({
queryKey: getGetV2GetSessionQueryKey(currentSessionId),
});
setIsSwitchingSession(false);
}
function selectSession(sessionId: string) {
if (sessionId === currentSessionId) return;
if (recentlyCreatedSessionsRef.current.has(sessionId)) {
queryClient.invalidateQueries({
queryKey: getGetV2GetSessionQueryKey(sessionId),
});
}
setUrlSessionId(sessionId, { shallow: false });
if (isMobile) handleCloseDrawer();
}
function handleNewChat() {
resetAutoSelect();
function startNewChat() {
resetPagination();
// Invalidate and refetch sessions list to ensure newly created sessions appear
queryClient.invalidateQueries({
queryKey: getGetV2ListSessionsQueryKey(),
});
window.history.replaceState(null, "", "/copilot");
router.replace("/copilot");
setUrlSessionId(null, { shallow: false });
if (isMobile) handleCloseDrawer();
}
function resetAutoSelect() {
hasAutoSelectedRef.current = false;
setHasAutoSelectedSession(false);
function handleSessionClick(sessionId: string) {
if (sessionId === currentSessionId) return;
if (isStreaming) {
pendingActionRef.current = async () => {
await stopCurrentStream();
selectSession(sessionId);
};
openInterruptModal(pendingActionRef.current);
} else {
selectSession(sessionId);
}
}
const isLoading = isSessionsLoading && accumulatedSessions.length === 0;
function handleNewChatClick() {
if (isStreaming) {
pendingActionRef.current = async () => {
await stopCurrentStream();
startNewChat();
};
openInterruptModal(pendingActionRef.current);
} else {
startNewChat();
}
}
return {
isMobile,
@@ -156,17 +155,17 @@ export function useCopilotShell() {
isLoggedIn,
hasActiveSession:
Boolean(currentSessionId) && (!isOnHomepage || Boolean(paramSessionId)),
isLoading,
sessions: visibleSessions,
currentSessionId: sidebarSelectedSessionId,
handleSelectSession,
isLoading: isLoading || isCreatingSession,
isCreatingSession,
sessions,
currentSessionId: urlSessionId,
handleOpenDrawer,
handleCloseDrawer,
handleDrawerOpenChange,
handleNewChat,
handleNewChatClick,
handleSessionClick,
hasNextPage,
isFetchingNextPage: isSessionsFetching,
fetchNextPage,
isReadyToShowContent,
};
}

113
autogpt_platform/frontend/src/app/(platform)/copilot/components/CopilotShell/useShellSessionList.ts Normal file
View File

@@ -0,0 +1,113 @@
import { getGetV2ListSessionsQueryKey } from "@/app/api/__generated__/endpoints/chat/chat";
import type { SessionDetailResponse } from "@/app/api/__generated__/models/sessionDetailResponse";
import type { SessionSummaryResponse } from "@/app/api/__generated__/models/sessionSummaryResponse";
import { useChatStore } from "@/components/contextual/Chat/chat-store";
import { useQueryClient } from "@tanstack/react-query";
import { useEffect, useMemo, useRef } from "react";
import { useSessionsPagination } from "./components/SessionsList/useSessionsPagination";
import {
convertSessionDetailToSummary,
filterVisibleSessions,
mergeCurrentSessionIntoList,
} from "./helpers";
interface UseShellSessionListArgs {
paginationEnabled: boolean;
currentSessionId: string | null;
currentSessionData: SessionDetailResponse | null | undefined;
isOnHomepage: boolean;
paramSessionId: string | null;
}
export function useShellSessionList({
paginationEnabled,
currentSessionId,
currentSessionData,
isOnHomepage,
paramSessionId,
}: UseShellSessionListArgs) {
const queryClient = useQueryClient();
const onStreamComplete = useChatStore((s) => s.onStreamComplete);
const {
sessions: accumulatedSessions,
isLoading: isSessionsLoading,
isFetching: isSessionsFetching,
hasNextPage,
fetchNextPage,
reset: resetPagination,
} = useSessionsPagination({
enabled: paginationEnabled,
});
const recentlyCreatedSessionsRef = useRef<
Map<string, SessionSummaryResponse>
>(new Map());
useEffect(() => {
if (isOnHomepage && !paramSessionId) {
queryClient.invalidateQueries({
queryKey: getGetV2ListSessionsQueryKey(),
});
}
}, [isOnHomepage, paramSessionId, queryClient]);
useEffect(() => {
if (currentSessionId && currentSessionData) {
const isNewSession =
currentSessionData.updated_at === currentSessionData.created_at;
const isNotInAccumulated = !accumulatedSessions.some(
(s) => s.id === currentSessionId,
);
if (isNewSession || isNotInAccumulated) {
const summary = convertSessionDetailToSummary(currentSessionData);
recentlyCreatedSessionsRef.current.set(currentSessionId, summary);
}
}
}, [currentSessionId, currentSessionData, accumulatedSessions]);
useEffect(() => {
for (const sessionId of recentlyCreatedSessionsRef.current.keys()) {
if (accumulatedSessions.some((s) => s.id === sessionId)) {
recentlyCreatedSessionsRef.current.delete(sessionId);
}
}
}, [accumulatedSessions]);
useEffect(() => {
const unsubscribe = onStreamComplete(() => {
queryClient.invalidateQueries({
queryKey: getGetV2ListSessionsQueryKey(),
});
});
return unsubscribe;
}, [onStreamComplete, queryClient]);
const sessions = useMemo(
() =>
mergeCurrentSessionIntoList(
accumulatedSessions,
currentSessionId,
currentSessionData,
recentlyCreatedSessionsRef.current,
),
[accumulatedSessions, currentSessionId, currentSessionData],
);
const visibleSessions = useMemo(
() => filterVisibleSessions(sessions),
[sessions],
);
const isLoading = isSessionsLoading && accumulatedSessions.length === 0;
return {
sessions: visibleSessions,
isLoading,
isSessionsFetching,
hasNextPage,
fetchNextPage,
resetPagination,
recentlyCreatedSessionsRef,
};
}

56
autogpt_platform/frontend/src/app/(platform)/copilot/copilot-page-store.ts Normal file
View File

@@ -0,0 +1,56 @@
"use client";
import { create } from "zustand";
interface CopilotStoreState {
isStreaming: boolean;
isSwitchingSession: boolean;
isCreatingSession: boolean;
isInterruptModalOpen: boolean;
pendingAction: (() => void) | null;
}
interface CopilotStoreActions {
setIsStreaming: (isStreaming: boolean) => void;
setIsSwitchingSession: (isSwitchingSession: boolean) => void;
setIsCreatingSession: (isCreating: boolean) => void;
openInterruptModal: (onConfirm: () => void) => void;
confirmInterrupt: () => void;
cancelInterrupt: () => void;
}
type CopilotStore = CopilotStoreState & CopilotStoreActions;
export const useCopilotStore = create<CopilotStore>((set, get) => ({
isStreaming: false,
isSwitchingSession: false,
isCreatingSession: false,
isInterruptModalOpen: false,
pendingAction: null,
setIsStreaming(isStreaming) {
set({ isStreaming });
},
setIsSwitchingSession(isSwitchingSession) {
set({ isSwitchingSession });
},
setIsCreatingSession(isCreatingSession) {
set({ isCreatingSession });
},
openInterruptModal(onConfirm) {
set({ isInterruptModalOpen: true, pendingAction: onConfirm });
},
confirmInterrupt() {
const { pendingAction } = get();
set({ isInterruptModalOpen: false, pendingAction: null });
if (pendingAction) pendingAction();
},
cancelInterrupt() {
set({ isInterruptModalOpen: false, pendingAction: null });
},
}));

23
autogpt_platform/frontend/src/app/(platform)/copilot/helpers.ts
View File

@@ -1,28 +1,5 @@
import type { User } from "@supabase/supabase-js";
export type PageState =
| { type: "welcome" }
| { type: "newChat" }
| { type: "creating"; prompt: string }
| { type: "chat"; sessionId: string; initialPrompt?: string };
export function getInitialPromptFromState(
pageState: PageState,
storedInitialPrompt: string | undefined,
) {
if (storedInitialPrompt) return storedInitialPrompt;
if (pageState.type === "creating") return pageState.prompt;
if (pageState.type === "chat") return pageState.initialPrompt;
}
export function shouldResetToWelcome(pageState: PageState) {
return (
pageState.type !== "newChat" &&
pageState.type !== "creating" &&
pageState.type !== "welcome"
);
}
export function getGreetingName(user?: User | null): string {
if (!user) return "there";
const metadata = user.user_metadata as Record<string, unknown> | undefined;

7
autogpt_platform/frontend/src/app/(platform)/copilot/layout.tsx
View File

@@ -1,11 +1,6 @@
import type { ReactNode } from "react";
import { NewChatProvider } from "./NewChatContext";
import { CopilotShell } from "./components/CopilotShell/CopilotShell";
export default function CopilotLayout({ children }: { children: ReactNode }) {
return (
<NewChatProvider>
<CopilotShell>{children}</CopilotShell>
</NewChatProvider>
);
return <CopilotShell>{children}</CopilotShell>;
}

69
autogpt_platform/frontend/src/app/(platform)/copilot/page.tsx
View File

@@ -1,22 +1,25 @@
"use client";
import { Skeleton } from "@/components/__legacy__/ui/skeleton";
import { Button } from "@/components/atoms/Button/Button";
import { Skeleton } from "@/components/atoms/Skeleton/Skeleton";
import { Text } from "@/components/atoms/Text/Text";
import { Chat } from "@/components/contextual/Chat/Chat";
import { ChatInput } from "@/components/contextual/Chat/components/ChatInput/ChatInput";
import { ChatLoader } from "@/components/contextual/Chat/components/ChatLoader/ChatLoader";
import { Dialog } from "@/components/molecules/Dialog/Dialog";
import { useCopilotStore } from "./copilot-page-store";
import { useCopilotPage } from "./useCopilotPage";
export default function CopilotPage() {
const { state, handlers } = useCopilotPage();
const isInterruptModalOpen = useCopilotStore((s) => s.isInterruptModalOpen);
const confirmInterrupt = useCopilotStore((s) => s.confirmInterrupt);
const cancelInterrupt = useCopilotStore((s) => s.cancelInterrupt);
const {
greetingName,
quickActions,
isLoading,
pageState,
isNewChatModalOpen,
hasSession,
initialPrompt,
isReady,
} = state;
const {
@@ -24,24 +27,16 @@ export default function CopilotPage() {
startChatWithPrompt,
handleSessionNotFound,
handleStreamingChange,
handleCancelNewChat,
proceedWithNewChat,
handleNewChatModalOpen,
} = handlers;
if (!isReady) {
return null;
}
if (!isReady) return null;
// Show Chat when we have an active session
if (pageState.type === "chat") {
if (hasSession) {
return (
<div className="flex h-full flex-col">
<Chat
key={pageState.sessionId ?? "welcome"}
className="flex-1"
urlSessionId={pageState.sessionId}
initialPrompt={pageState.initialPrompt}
initialPrompt={initialPrompt}
onSessionNotFound={handleSessionNotFound}
onStreamingChange={handleStreamingChange}
/>
@@ -49,31 +44,33 @@ export default function CopilotPage() {
title="Interrupt current chat?"
styling={{ maxWidth: 300, width: "100%" }}
controlled={{
isOpen: isNewChatModalOpen,
set: handleNewChatModalOpen,
isOpen: isInterruptModalOpen,
set: (open) => {
if (!open) cancelInterrupt();
},
}}
onClose={handleCancelNewChat}
onClose={cancelInterrupt}
>
<Dialog.Content>
<div className="flex flex-col gap-4">
<Text variant="body">
The current chat response will be interrupted. Are you sure you
want to start a new chat?
want to continue?
</Text>
<Dialog.Footer>
<Button
type="button"
variant="outline"
onClick={handleCancelNewChat}
onClick={cancelInterrupt}
>
Cancel
</Button>
<Button
type="button"
variant="primary"
onClick={proceedWithNewChat}
onClick={confirmInterrupt}
>
Start new chat
Continue
</Button>
</Dialog.Footer>
</div>
@@ -83,34 +80,6 @@ export default function CopilotPage() {
);
}
if (pageState.type === "newChat") {
return (
<div className="flex h-full flex-1 flex-col items-center justify-center bg-[#f8f8f9]">
<div className="flex flex-col items-center gap-4">
<ChatLoader />
<Text variant="body" className="text-zinc-500">
Loading your chats...
</Text>
</div>
</div>
);
}
// Show loading state while creating session and sending first message
if (pageState.type === "creating") {
return (
<div className="flex h-full flex-1 flex-col items-center justify-center bg-[#f8f8f9]">
<div className="flex flex-col items-center gap-4">
<ChatLoader />
<Text variant="body" className="text-zinc-500">
Loading your chats...
</Text>
</div>
</div>
);
}
// Show Welcome screen
return (
<div className="flex h-full flex-1 items-center justify-center overflow-y-auto bg-[#f8f8f9] px-6 py-10">
<div className="w-full text-center">

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

@@ -1,4 +1,7 @@
import { postV2CreateSession } from "@/app/api/__generated__/endpoints/chat/chat";
import {
getGetV2ListSessionsQueryKey,
postV2CreateSession,
} from "@/app/api/__generated__/endpoints/chat/chat";
import { useToast } from "@/components/molecules/Toast/use-toast";
import { getHomepageRoute } from "@/lib/constants";
import { useSupabase } from "@/lib/supabase/hooks/useSupabase";
@@ -7,81 +10,27 @@ import {
type FlagValues,
useGetFlag,
} from "@/services/feature-flags/use-get-flag";
import { SessionKey, sessionStorage } from "@/services/storage/session-storage";
import * as Sentry from "@sentry/nextjs";
import { useQueryClient } from "@tanstack/react-query";
import { useFlags } from "launchdarkly-react-client-sdk";
import { useRouter } from "next/navigation";
import { useEffect, useReducer } from "react";
import { useNewChat } from "./NewChatContext";
import { getGreetingName, getQuickActions, type PageState } from "./helpers";
import { useCopilotURLState } from "./useCopilotURLState";
type CopilotState = {
pageState: PageState;
isStreaming: boolean;
isNewChatModalOpen: boolean;
initialPrompts: Record<string, string>;
previousSessionId: string | null;
};
type CopilotAction =
| { type: "setPageState"; pageState: PageState }
| { type: "setStreaming"; isStreaming: boolean }
| { type: "setNewChatModalOpen"; isOpen: boolean }
| { type: "setInitialPrompt"; sessionId: string; prompt: string }
| { type: "setPreviousSessionId"; sessionId: string | null };
function isSamePageState(next: PageState, current: PageState) {
if (next.type !== current.type) return false;
if (next.type === "creating" && current.type === "creating") {
return next.prompt === current.prompt;
}
if (next.type === "chat" && current.type === "chat") {
return (
next.sessionId === current.sessionId &&
next.initialPrompt === current.initialPrompt
);
}
return true;
}
function copilotReducer(
state: CopilotState,
action: CopilotAction,
): CopilotState {
if (action.type === "setPageState") {
if (isSamePageState(action.pageState, state.pageState)) return state;
return { ...state, pageState: action.pageState };
}
if (action.type === "setStreaming") {
if (action.isStreaming === state.isStreaming) return state;
return { ...state, isStreaming: action.isStreaming };
}
if (action.type === "setNewChatModalOpen") {
if (action.isOpen === state.isNewChatModalOpen) return state;
return { ...state, isNewChatModalOpen: action.isOpen };
}
if (action.type === "setInitialPrompt") {
if (state.initialPrompts[action.sessionId] === action.prompt) return state;
return {
...state,
initialPrompts: {
...state.initialPrompts,
[action.sessionId]: action.prompt,
},
};
}
if (action.type === "setPreviousSessionId") {
if (state.previousSessionId === action.sessionId) return state;
return { ...state, previousSessionId: action.sessionId };
}
return state;
}
import { useEffect } from "react";
import { useCopilotStore } from "./copilot-page-store";
import { getGreetingName, getQuickActions } from "./helpers";
import { useCopilotSessionId } from "./useCopilotSessionId";
export function useCopilotPage() {
const router = useRouter();
const queryClient = useQueryClient();
const { user, isLoggedIn, isUserLoading } = useSupabase();
const { toast } = useToast();
const { urlSessionId, setUrlSessionId } = useCopilotSessionId();
const setIsStreaming = useCopilotStore((s) => s.setIsStreaming);
const isCreating = useCopilotStore((s) => s.isCreatingSession);
const setIsCreating = useCopilotStore((s) => s.setIsCreatingSession);
const isChatEnabled = useGetFlag(Flag.CHAT);
const flags = useFlags<FlagValues>();
const homepageRoute = getHomepageRoute(isChatEnabled);
@@ -91,86 +40,27 @@ export function useCopilotPage() {
const isFlagReady =
!isLaunchDarklyConfigured || flags[Flag.CHAT] !== undefined;
const [state, dispatch] = useReducer(copilotReducer, {
pageState: { type: "welcome" },
isStreaming: false,
isNewChatModalOpen: false,
initialPrompts: {},
previousSessionId: null,
});
const newChatContext = useNewChat();
const greetingName = getGreetingName(user);
const quickActions = getQuickActions();
function setPageState(pageState: PageState) {
dispatch({ type: "setPageState", pageState });
}
const hasSession = Boolean(urlSessionId);
const initialPrompt = urlSessionId
? getInitialPrompt(urlSessionId)
: undefined;
function setInitialPrompt(sessionId: string, prompt: string) {
dispatch({ type: "setInitialPrompt", sessionId, prompt });
}
function setPreviousSessionId(sessionId: string | null) {
dispatch({ type: "setPreviousSessionId", sessionId });
}
const { setUrlSessionId } = useCopilotURLState({
pageState: state.pageState,
initialPrompts: state.initialPrompts,
previousSessionId: state.previousSessionId,
setPageState,
setInitialPrompt,
setPreviousSessionId,
});
useEffect(
function registerNewChatHandler() {
if (!newChatContext) return;
newChatContext.setOnNewChatClick(handleNewChatClick);
return function cleanup() {
newChatContext.setOnNewChatClick(undefined);
};
},
[newChatContext, handleNewChatClick],
);
useEffect(
function transitionNewChatToWelcome() {
if (state.pageState.type === "newChat") {
function setWelcomeState() {
dispatch({ type: "setPageState", pageState: { type: "welcome" } });
}
const timer = setTimeout(setWelcomeState, 300);
return function cleanup() {
clearTimeout(timer);
};
}
},
[state.pageState.type],
);
useEffect(
function ensureAccess() {
if (!isFlagReady) return;
if (isChatEnabled === false) {
router.replace(homepageRoute);
}
},
[homepageRoute, isChatEnabled, isFlagReady, router],
);
useEffect(() => {
if (!isFlagReady) return;
if (isChatEnabled === false) {
router.replace(homepageRoute);
}
}, [homepageRoute, isChatEnabled, isFlagReady, router]);
async function startChatWithPrompt(prompt: string) {
if (!prompt?.trim()) return;
if (state.pageState.type === "creating") return;
if (isCreating) return;
const trimmedPrompt = prompt.trim();
dispatch({
type: "setPageState",
pageState: { type: "creating", prompt: trimmedPrompt },
});
setIsCreating(true);
try {
const sessionResponse = await postV2CreateSession({
@@ -182,23 +72,19 @@ export function useCopilotPage() {
}
const sessionId = sessionResponse.data.id;
setInitialPrompt(sessionId, trimmedPrompt);
dispatch({
type: "setInitialPrompt",
sessionId,
prompt: trimmedPrompt,
await queryClient.invalidateQueries({
queryKey: getGetV2ListSessionsQueryKey(),
});
await setUrlSessionId(sessionId, { shallow: false });
dispatch({
type: "setPageState",
pageState: { type: "chat", sessionId, initialPrompt: trimmedPrompt },
});
await setUrlSessionId(sessionId, { shallow: true });
} catch (error) {
console.error("[CopilotPage] Failed to start chat:", error);
toast({ title: "Failed to start chat", variant: "destructive" });
Sentry.captureException(error);
dispatch({ type: "setPageState", pageState: { type: "welcome" } });
} finally {
setIsCreating(false);
}
}
@@ -211,37 +97,7 @@ export function useCopilotPage() {
}
function handleStreamingChange(isStreamingValue: boolean) {
dispatch({ type: "setStreaming", isStreaming: isStreamingValue });
}
async function proceedWithNewChat() {
dispatch({ type: "setNewChatModalOpen", isOpen: false });
if (newChatContext?.performNewChat) {
newChatContext.performNewChat();
return;
}
try {
await setUrlSessionId(null, { shallow: false });
} catch (error) {
console.error("[CopilotPage] Failed to clear session:", error);
}
router.replace("/copilot");
}
function handleCancelNewChat() {
dispatch({ type: "setNewChatModalOpen", isOpen: false });
}
function handleNewChatModalOpen(isOpen: boolean) {
dispatch({ type: "setNewChatModalOpen", isOpen });
}
function handleNewChatClick() {
if (state.isStreaming) {
dispatch({ type: "setNewChatModalOpen", isOpen: true });
} else {
proceedWithNewChat();
}
setIsStreaming(isStreamingValue);
}
return {
@@ -249,8 +105,8 @@ export function useCopilotPage() {
greetingName,
quickActions,
isLoading: isUserLoading,
pageState: state.pageState,
isNewChatModalOpen: state.isNewChatModalOpen,
hasSession,
initialPrompt,
isReady: isFlagReady && isChatEnabled !== false && isLoggedIn,
},
handlers: {
@@ -258,9 +114,32 @@ export function useCopilotPage() {
startChatWithPrompt,
handleSessionNotFound,
handleStreamingChange,
handleCancelNewChat,
proceedWithNewChat,
handleNewChatModalOpen,
},
};
}
function getInitialPrompt(sessionId: string): string | undefined {
try {
const prompts = JSON.parse(
sessionStorage.get(SessionKey.CHAT_INITIAL_PROMPTS) || "{}",
);
return prompts[sessionId];
} catch {
return undefined;
}
}
function setInitialPrompt(sessionId: string, prompt: string): void {
try {
const prompts = JSON.parse(
sessionStorage.get(SessionKey.CHAT_INITIAL_PROMPTS) || "{}",
);
prompts[sessionId] = prompt;
sessionStorage.set(
SessionKey.CHAT_INITIAL_PROMPTS,
JSON.stringify(prompts),
);
} catch {
// Ignore storage errors
}
}

10
autogpt_platform/frontend/src/app/(platform)/copilot/useCopilotSessionId.ts Normal file
View File

@@ -0,0 +1,10 @@
import { parseAsString, useQueryState } from "nuqs";
export function useCopilotSessionId() {
const [urlSessionId, setUrlSessionId] = useQueryState(
"sessionId",
parseAsString,
);
return { urlSessionId, setUrlSessionId };
}

80
autogpt_platform/frontend/src/app/(platform)/copilot/useCopilotURLState.ts
View File

@@ -1,80 +0,0 @@
import { parseAsString, useQueryState } from "nuqs";
import { useLayoutEffect } from "react";
import {
getInitialPromptFromState,
type PageState,
shouldResetToWelcome,
} from "./helpers";
interface UseCopilotUrlStateArgs {
pageState: PageState;
initialPrompts: Record<string, string>;
previousSessionId: string | null;
setPageState: (pageState: PageState) => void;
setInitialPrompt: (sessionId: string, prompt: string) => void;
setPreviousSessionId: (sessionId: string | null) => void;
}
export function useCopilotURLState({
pageState,
initialPrompts,
previousSessionId,
setPageState,
setInitialPrompt,
setPreviousSessionId,
}: UseCopilotUrlStateArgs) {
const [urlSessionId, setUrlSessionId] = useQueryState(
"sessionId",
parseAsString,
);
function syncSessionFromUrl() {
if (urlSessionId) {
if (pageState.type === "chat" && pageState.sessionId === urlSessionId) {
setPreviousSessionId(urlSessionId);
return;
}
const storedInitialPrompt = initialPrompts[urlSessionId];
const currentInitialPrompt = getInitialPromptFromState(
pageState,
storedInitialPrompt,
);
if (currentInitialPrompt) {
setInitialPrompt(urlSessionId, currentInitialPrompt);
}
setPageState({
type: "chat",
sessionId: urlSessionId,
initialPrompt: currentInitialPrompt,
});
setPreviousSessionId(urlSessionId);
return;
}
const wasInChat = previousSessionId !== null && pageState.type === "chat";
setPreviousSessionId(null);
if (wasInChat) {
setPageState({ type: "newChat" });
return;
}
if (shouldResetToWelcome(pageState)) {
setPageState({ type: "welcome" });
}
}
useLayoutEffect(syncSessionFromUrl, [
urlSessionId,
pageState.type,
previousSessionId,
initialPrompts,
]);
return {
urlSessionId,
setUrlSessionId,
};
}

2
autogpt_platform/frontend/src/app/(platform)/layout.tsx
View File

@@ -1,10 +1,12 @@
import { Navbar } from "@/components/layout/Navbar/Navbar";
import { NetworkStatusMonitor } from "@/services/network-status/NetworkStatusMonitor";
import { ReactNode } from "react";
import { AdminImpersonationBanner } from "./admin/components/AdminImpersonationBanner";
export default function PlatformLayout({ children }: { children: ReactNode }) {
return (
<main className="flex h-screen w-full flex-col">
<NetworkStatusMonitor />
<Navbar />
<AdminImpersonationBanner />
<section className="flex-1">{children}</section>

14
autogpt_platform/frontend/src/components/atoms/Skeleton/Skeleton.tsx Normal file
View File

@@ -0,0 +1,14 @@
import { cn } from "@/lib/utils";
interface Props extends React.HTMLAttributes<HTMLDivElement> {
className?: string;
}
export function Skeleton({ className, ...props }: Props) {
return (
<div
className={cn("animate-pulse rounded-md bg-zinc-100", className)}
{...props}
/>
);
}

2
autogpt_platform/frontend/src/components/atoms/Skeleton/skeleton.stories.tsx
View File

@@ -1,4 +1,4 @@
import { Skeleton } from "@/components/__legacy__/ui/skeleton";
import { Skeleton } from "./Skeleton";
import type { Meta, StoryObj } from "@storybook/nextjs";
const meta: Meta<typeof Skeleton> = {

52
autogpt_platform/frontend/src/components/contextual/Chat/Chat.tsx
View File

@@ -1,16 +1,17 @@
"use client";
import { useCopilotSessionId } from "@/app/(platform)/copilot/useCopilotSessionId";
import { useCopilotStore } from "@/app/(platform)/copilot/copilot-page-store";
import { LoadingSpinner } from "@/components/atoms/LoadingSpinner/LoadingSpinner";
import { Text } from "@/components/atoms/Text/Text";
import { cn } from "@/lib/utils";
import { useEffect, useRef } from "react";
import { ChatContainer } from "./components/ChatContainer/ChatContainer";
import { ChatErrorState } from "./components/ChatErrorState/ChatErrorState";
import { ChatLoader } from "./components/ChatLoader/ChatLoader";
import { useChat } from "./useChat";
export interface ChatProps {
className?: string;
urlSessionId?: string | null;
initialPrompt?: string;
onSessionNotFound?: () => void;
onStreamingChange?: (isStreaming: boolean) => void;
@@ -18,12 +19,13 @@ export interface ChatProps {
export function Chat({
className,
urlSessionId,
initialPrompt,
onSessionNotFound,
onStreamingChange,
}: ChatProps) {
const { urlSessionId } = useCopilotSessionId();
const hasHandledNotFoundRef = useRef(false);
const isSwitchingSession = useCopilotStore((s) => s.isSwitchingSession);
const {
messages,
isLoading,
@@ -33,49 +35,59 @@ export function Chat({
sessionId,
createSession,
showLoader,
startPollingForOperation,
} = useChat({ urlSessionId });
useEffect(
function handleMissingSession() {
if (!onSessionNotFound) return;
if (!urlSessionId) return;
if (!isSessionNotFound || isLoading || isCreating) return;
if (hasHandledNotFoundRef.current) return;
hasHandledNotFoundRef.current = true;
onSessionNotFound();
},
[onSessionNotFound, urlSessionId, isSessionNotFound, isLoading, isCreating],
);
useEffect(() => {
if (!onSessionNotFound) return;
if (!urlSessionId) return;
if (!isSessionNotFound || isLoading || isCreating) return;
if (hasHandledNotFoundRef.current) return;
hasHandledNotFoundRef.current = true;
onSessionNotFound();
}, [
onSessionNotFound,
urlSessionId,
isSessionNotFound,
isLoading,
isCreating,
]);
const shouldShowLoader =
(showLoader && (isLoading || isCreating)) || isSwitchingSession;
return (
<div className={cn("flex h-full flex-col", className)}>
{/* Main Content */}
<main className="flex min-h-0 w-full flex-1 flex-col overflow-hidden bg-[#f8f8f9]">
{/* Loading State */}
{showLoader && (isLoading || isCreating) && (
{shouldShowLoader && (
<div className="flex flex-1 items-center justify-center">
<div className="flex flex-col items-center gap-4">
<ChatLoader />
<div className="flex flex-col items-center gap-3">
<LoadingSpinner size="large" className="text-neutral-400" />
<Text variant="body" className="text-zinc-500">
Loading your chats...
{isSwitchingSession
? "Switching chat..."
: "Loading your chat..."}
</Text>
</div>
</div>
)}
{/* Error State */}
{error && !isLoading && (
{error && !isLoading && !isSwitchingSession && (
<ChatErrorState error={error} onRetry={createSession} />
)}
{/* Session Content */}
{sessionId && !isLoading && !error && (
{sessionId && !isLoading && !error && !isSwitchingSession && (
<ChatContainer
sessionId={sessionId}
initialMessages={messages}
initialPrompt={initialPrompt}
className="flex-1"
onStreamingChange={onStreamingChange}
onOperationStarted={startPollingForOperation}
/>
)}
</main>

289
autogpt_platform/frontend/src/components/contextual/Chat/chat-store.ts Normal file
View File

@@ -0,0 +1,289 @@
"use client";
import { create } from "zustand";
import type {
ActiveStream,
StreamChunk,
StreamCompleteCallback,
StreamResult,
StreamStatus,
} from "./chat-types";
import { executeStream } from "./stream-executor";
const COMPLETED_STREAM_TTL = 5 * 60 * 1000; // 5 minutes
interface ChatStoreState {
activeStreams: Map<string, ActiveStream>;
completedStreams: Map<string, StreamResult>;
activeSessions: Set<string>;
streamCompleteCallbacks: Set<StreamCompleteCallback>;
}
interface ChatStoreActions {
startStream: (
sessionId: string,
message: string,
isUserMessage: boolean,
context?: { url: string; content: string },
onChunk?: (chunk: StreamChunk) => void,
) => Promise<void>;
stopStream: (sessionId: string) => void;
subscribeToStream: (
sessionId: string,
onChunk: (chunk: StreamChunk) => void,
skipReplay?: boolean,
) => () => void;
getStreamStatus: (sessionId: string) => StreamStatus;
getCompletedStream: (sessionId: string) => StreamResult | undefined;
clearCompletedStream: (sessionId: string) => void;
isStreaming: (sessionId: string) => boolean;
registerActiveSession: (sessionId: string) => void;
unregisterActiveSession: (sessionId: string) => void;
isSessionActive: (sessionId: string) => boolean;
onStreamComplete: (callback: StreamCompleteCallback) => () => void;
}
type ChatStore = ChatStoreState & ChatStoreActions;
function notifyStreamComplete(
callbacks: Set<StreamCompleteCallback>,
sessionId: string,
) {
for (const callback of callbacks) {
try {
callback(sessionId);
} catch (err) {
console.warn("[ChatStore] Stream complete callback error:", err);
}
}
}
function cleanupExpiredStreams(
completedStreams: Map<string, StreamResult>,
): Map<string, StreamResult> {
const now = Date.now();
const cleaned = new Map(completedStreams);
for (const [sessionId, result] of cleaned) {
if (now - result.completedAt > COMPLETED_STREAM_TTL) {
cleaned.delete(sessionId);
}
}
return cleaned;
}
export const useChatStore = create<ChatStore>((set, get) => ({
activeStreams: new Map(),
completedStreams: new Map(),
activeSessions: new Set(),
streamCompleteCallbacks: new Set(),
startStream: async function startStream(
sessionId,
message,
isUserMessage,
context,
onChunk,
) {
const state = get();
const newActiveStreams = new Map(state.activeStreams);
let newCompletedStreams = new Map(state.completedStreams);
const callbacks = state.streamCompleteCallbacks;
const existingStream = newActiveStreams.get(sessionId);
if (existingStream) {
existingStream.abortController.abort();
const normalizedStatus =
existingStream.status === "streaming"
? "completed"
: existingStream.status;
const result: StreamResult = {
sessionId,
status: normalizedStatus,
chunks: existingStream.chunks,
completedAt: Date.now(),
error: existingStream.error,
};
newCompletedStreams.set(sessionId, result);
newActiveStreams.delete(sessionId);
newCompletedStreams = cleanupExpiredStreams(newCompletedStreams);
if (normalizedStatus === "completed" || normalizedStatus === "error") {
notifyStreamComplete(callbacks, sessionId);
}
}
const abortController = new AbortController();
const initialCallbacks = new Set<(chunk: StreamChunk) => void>();
if (onChunk) initialCallbacks.add(onChunk);
const stream: ActiveStream = {
sessionId,
abortController,
status: "streaming",
startedAt: Date.now(),
chunks: [],
onChunkCallbacks: initialCallbacks,
};
newActiveStreams.set(sessionId, stream);
set({
activeStreams: newActiveStreams,
completedStreams: newCompletedStreams,
});
try {
await executeStream(stream, message, isUserMessage, context);
} finally {
if (onChunk) stream.onChunkCallbacks.delete(onChunk);
if (stream.status !== "streaming") {
const currentState = get();
const finalActiveStreams = new Map(currentState.activeStreams);
let finalCompletedStreams = new Map(currentState.completedStreams);
const storedStream = finalActiveStreams.get(sessionId);
if (storedStream === stream) {
const result: StreamResult = {
sessionId,
status: stream.status,
chunks: stream.chunks,
completedAt: Date.now(),
error: stream.error,
};
finalCompletedStreams.set(sessionId, result);
finalActiveStreams.delete(sessionId);
finalCompletedStreams = cleanupExpiredStreams(finalCompletedStreams);
set({
activeStreams: finalActiveStreams,
completedStreams: finalCompletedStreams,
});
if (stream.status === "completed" || stream.status === "error") {
notifyStreamComplete(
currentState.streamCompleteCallbacks,
sessionId,
);
}
}
}
}
},
stopStream: function stopStream(sessionId) {
const state = get();
const stream = state.activeStreams.get(sessionId);
if (!stream) return;
stream.abortController.abort();
stream.status = "completed";
const newActiveStreams = new Map(state.activeStreams);
let newCompletedStreams = new Map(state.completedStreams);
const result: StreamResult = {
sessionId,
status: stream.status,
chunks: stream.chunks,
completedAt: Date.now(),
error: stream.error,
};
newCompletedStreams.set(sessionId, result);
newActiveStreams.delete(sessionId);
newCompletedStreams = cleanupExpiredStreams(newCompletedStreams);
set({
activeStreams: newActiveStreams,
completedStreams: newCompletedStreams,
});
notifyStreamComplete(state.streamCompleteCallbacks, sessionId);
},
subscribeToStream: function subscribeToStream(
sessionId,
onChunk,
skipReplay = false,
) {
const state = get();
const stream = state.activeStreams.get(sessionId);
if (stream) {
if (!skipReplay) {
for (const chunk of stream.chunks) {
onChunk(chunk);
}
}
stream.onChunkCallbacks.add(onChunk);
return function unsubscribe() {
stream.onChunkCallbacks.delete(onChunk);
};
}
return function noop() {};
},
getStreamStatus: function getStreamStatus(sessionId) {
const { activeStreams, completedStreams } = get();
const active = activeStreams.get(sessionId);
if (active) return active.status;
const completed = completedStreams.get(sessionId);
if (completed) return completed.status;
return "idle";
},
getCompletedStream: function getCompletedStream(sessionId) {
return get().completedStreams.get(sessionId);
},
clearCompletedStream: function clearCompletedStream(sessionId) {
const state = get();
if (!state.completedStreams.has(sessionId)) return;
const newCompletedStreams = new Map(state.completedStreams);
newCompletedStreams.delete(sessionId);
set({ completedStreams: newCompletedStreams });
},
isStreaming: function isStreaming(sessionId) {
const stream = get().activeStreams.get(sessionId);
return stream?.status === "streaming";
},
registerActiveSession: function registerActiveSession(sessionId) {
const state = get();
if (state.activeSessions.has(sessionId)) return;
const newActiveSessions = new Set(state.activeSessions);
newActiveSessions.add(sessionId);
set({ activeSessions: newActiveSessions });
},
unregisterActiveSession: function unregisterActiveSession(sessionId) {
const state = get();
if (!state.activeSessions.has(sessionId)) return;
const newActiveSessions = new Set(state.activeSessions);
newActiveSessions.delete(sessionId);
set({ activeSessions: newActiveSessions });
},
isSessionActive: function isSessionActive(sessionId) {
return get().activeSessions.has(sessionId);
},
onStreamComplete: function onStreamComplete(callback) {
const state = get();
const newCallbacks = new Set(state.streamCompleteCallbacks);
newCallbacks.add(callback);
set({ streamCompleteCallbacks: newCallbacks });
return function unsubscribe() {
const currentState = get();
const cleanedCallbacks = new Set(currentState.streamCompleteCallbacks);
cleanedCallbacks.delete(callback);
set({ streamCompleteCallbacks: cleanedCallbacks });
};
},
}));

94
autogpt_platform/frontend/src/components/contextual/Chat/chat-types.ts Normal file
View File

@@ -0,0 +1,94 @@
import type { ToolArguments, ToolResult } from "@/types/chat";
export type StreamStatus = "idle" | "streaming" | "completed" | "error";
export interface StreamChunk {
type:
| "text_chunk"
| "text_ended"
| "tool_call"
| "tool_call_start"
| "tool_response"
| "login_needed"
| "need_login"
| "credentials_needed"
| "error"
| "usage"
| "stream_end";
timestamp?: string;
content?: string;
message?: string;
code?: string;
details?: Record<string, unknown>;
tool_id?: string;
tool_name?: string;
arguments?: ToolArguments;
result?: ToolResult;
success?: boolean;
idx?: number;
session_id?: string;
agent_info?: {
graph_id: string;
name: string;
trigger_type: string;
};
provider?: string;
provider_name?: string;
credential_type?: string;
scopes?: string[];
title?: string;
[key: string]: unknown;
}
export type VercelStreamChunk =
| { type: "start"; messageId: string }
| { type: "finish" }
| { type: "text-start"; id: string }
| { type: "text-delta"; id: string; delta: string }
| { type: "text-end"; id: string }
| { type: "tool-input-start"; toolCallId: string; toolName: string }
| {
type: "tool-input-available";
toolCallId: string;
toolName: string;
input: Record<string, unknown>;
}
| {
type: "tool-output-available";
toolCallId: string;
toolName?: string;
output: unknown;
success?: boolean;
}
| {
type: "usage";
promptTokens: number;
completionTokens: number;
totalTokens: number;
}
| {
type: "error";
errorText: string;
code?: string;
details?: Record<string, unknown>;
};
export interface ActiveStream {
sessionId: string;
abortController: AbortController;
status: StreamStatus;
startedAt: number;
chunks: StreamChunk[];
error?: Error;
onChunkCallbacks: Set<(chunk: StreamChunk) => void>;
}
export interface StreamResult {
sessionId: string;
status: StreamStatus;
chunks: StreamChunk[];
completedAt: number;
error?: Error;
}
export type StreamCompleteCallback = (sessionId: string) => void;

27
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/ChatContainer.tsx
View File

@@ -4,6 +4,7 @@ import { Text } from "@/components/atoms/Text/Text";
import { Dialog } from "@/components/molecules/Dialog/Dialog";
import { useBreakpoint } from "@/lib/hooks/useBreakpoint";
import { cn } from "@/lib/utils";
import { GlobeHemisphereEastIcon } from "@phosphor-icons/react";
import { useEffect } from "react";
import { ChatInput } from "../ChatInput/ChatInput";
import { MessageList } from "../MessageList/MessageList";
@@ -15,6 +16,7 @@ export interface ChatContainerProps {
initialPrompt?: string;
className?: string;
onStreamingChange?: (isStreaming: boolean) => void;
onOperationStarted?: () => void;
}
export function ChatContainer({
@@ -23,6 +25,7 @@ export function ChatContainer({
initialPrompt,
className,
onStreamingChange,
onOperationStarted,
}: ChatContainerProps) {
const {
messages,
@@ -37,6 +40,7 @@ export function ChatContainer({
sessionId,
initialMessages,
initialPrompt,
onOperationStarted,
});
useEffect(() => {
@@ -55,24 +59,37 @@ export function ChatContainer({
)}
>
<Dialog
title="Service unavailable"
title={
<div className="flex items-center gap-2">
<GlobeHemisphereEastIcon className="size-6" />
<Text
variant="body"
className="text-md font-poppins leading-none md:text-lg"
>
Service unavailable
</Text>
</div>
}
controlled={{
isOpen: isRegionBlockedModalOpen,
set: handleRegionModalOpenChange,
}}
onClose={handleRegionModalClose}
styling={{ maxWidth: 550, width: "100%", minWidth: "auto" }}
>
<Dialog.Content>
<div className="flex flex-col gap-4">
<div className="flex flex-col gap-8">
<Text variant="body">
This model is not available in your region. Please connect via VPN
and try again.
The Autogpt AI model is not available in your region or your
connection is blocking it. Please try again with a different
connection.
</Text>
<div className="flex justify-end">
<div className="flex justify-center">
<Button
type="button"
variant="primary"
onClick={handleRegionModalClose}
className="w-full"
>
Got it
</Button>

2
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/createStreamEventDispatcher.ts
View File

@@ -1,5 +1,5 @@
import { toast } from "sonner";
import { StreamChunk } from "../../useChatStream";
import type { StreamChunk } from "../../chat-types";
import type { HandlerDependencies } from "./handlers";
import {
handleError,

40
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/handlers.ts
View File

@@ -22,6 +22,7 @@ export interface HandlerDependencies {
setIsStreamingInitiated: Dispatch<SetStateAction<boolean>>;
setIsRegionBlockedModalOpen: Dispatch<SetStateAction<boolean>>;
sessionId: string;
onOperationStarted?: () => void;
}
export function isRegionBlockedError(chunk: StreamChunk): boolean {
@@ -48,6 +49,15 @@ export function handleTextEnded(
const completedText = deps.streamingChunksRef.current.join("");
if (completedText.trim()) {
deps.setMessages((prev) => {
// Check if this exact message already exists to prevent duplicates
const exists = prev.some(
(msg) =>
msg.type === "message" &&
msg.role === "assistant" &&
msg.content === completedText,
);
if (exists) return prev;
const assistantMessage: ChatMessageData = {
type: "message",
role: "assistant",
@@ -154,6 +164,11 @@ export function handleToolResponse(
}
return;
}
// Trigger polling when operation_started is received
if (responseMessage.type === "operation_started") {
deps.onOperationStarted?.();
}
deps.setMessages((prev) => {
const toolCallIndex = prev.findIndex(
(msg) => msg.type === "tool_call" && msg.toolId === chunk.tool_id,
@@ -203,13 +218,24 @@ export function handleStreamEnd(
]);
}
if (completedContent.trim()) {
const assistantMessage: ChatMessageData = {
type: "message",
role: "assistant",
content: completedContent,
timestamp: new Date(),
};
deps.setMessages((prev) => [...prev, assistantMessage]);
deps.setMessages((prev) => {
// Check if this exact message already exists to prevent duplicates
const exists = prev.some(
(msg) =>
msg.type === "message" &&
msg.role === "assistant" &&
msg.content === completedContent,
);
if (exists) return prev;
const assistantMessage: ChatMessageData = {
type: "message",
role: "assistant",
content: completedContent,
timestamp: new Date(),
};
return [...prev, assistantMessage];
});
}
deps.setStreamingChunks([]);
deps.streamingChunksRef.current = [];

165
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/helpers.ts
View File

@@ -1,7 +1,118 @@
import type { SessionDetailResponse } from "@/app/api/__generated__/models/sessionDetailResponse";
import { SessionKey, sessionStorage } from "@/services/storage/session-storage";
import type { ToolResult } from "@/types/chat";
import type { ChatMessageData } from "../ChatMessage/useChatMessage";
export function processInitialMessages(
initialMessages: SessionDetailResponse["messages"],
): ChatMessageData[] {
const processedMessages: ChatMessageData[] = [];
const toolCallMap = new Map<string, string>();
for (const msg of initialMessages) {
if (!isValidMessage(msg)) {
console.warn("Invalid message structure from backend:", msg);
continue;
}
let content = String(msg.content || "");
const role = String(msg.role || "assistant").toLowerCase();
const toolCalls = msg.tool_calls;
const timestamp = msg.timestamp
? new Date(msg.timestamp as string)
: undefined;
if (role === "user") {
content = removePageContext(content);
if (!content.trim()) continue;
processedMessages.push({
type: "message",
role: "user",
content,
timestamp,
});
continue;
}
if (role === "assistant") {
content = content
.replace(/<thinking>[\s\S]*?<\/thinking>/gi, "")
.replace(/<internal_reasoning>[\s\S]*?<\/internal_reasoning>/gi, "")
.trim();
if (toolCalls && isToolCallArray(toolCalls) && toolCalls.length > 0) {
for (const toolCall of toolCalls) {
const toolName = toolCall.function.name;
const toolId = toolCall.id;
toolCallMap.set(toolId, toolName);
try {
const args = JSON.parse(toolCall.function.arguments || "{}");
processedMessages.push({
type: "tool_call",
toolId,
toolName,
arguments: args,
timestamp,
});
} catch (err) {
console.warn("Failed to parse tool call arguments:", err);
processedMessages.push({
type: "tool_call",
toolId,
toolName,
arguments: {},
timestamp,
});
}
}
if (content.trim()) {
processedMessages.push({
type: "message",
role: "assistant",
content,
timestamp,
});
}
} else if (content.trim()) {
processedMessages.push({
type: "message",
role: "assistant",
content,
timestamp,
});
}
continue;
}
if (role === "tool") {
const toolCallId = (msg.tool_call_id as string) || "";
const toolName = toolCallMap.get(toolCallId) || "unknown";
const toolResponse = parseToolResponse(
content,
toolCallId,
toolName,
timestamp,
);
if (toolResponse) {
processedMessages.push(toolResponse);
}
continue;
}
if (content.trim()) {
processedMessages.push({
type: "message",
role: role as "user" | "assistant" | "system",
content,
timestamp,
});
}
}
return processedMessages;
}
export function hasSentInitialPrompt(sessionId: string): boolean {
try {
const sent = JSON.parse(
@@ -193,6 +304,7 @@ export function parseToolResponse(
if (isAgentArray(agentsData)) {
return {
type: "agent_carousel",
toolId,
toolName: "agent_carousel",
agents: agentsData,
totalCount: parsedResult.total_count as number | undefined,
@@ -205,6 +317,7 @@ export function parseToolResponse(
if (responseType === "execution_started") {
return {
type: "execution_started",
toolId,
toolName: "execution_started",
executionId: (parsedResult.execution_id as string) || "",
agentName: (parsedResult.graph_name as string) || undefined,
@@ -213,6 +326,58 @@ export function parseToolResponse(
timestamp: timestamp || new Date(),
};
}
if (responseType === "clarification_needed") {
return {
type: "clarification_needed",
toolName,
questions:
(parsedResult.questions as Array<{
question: string;
keyword: string;
example?: string;
}>) || [],
message:
(parsedResult.message as string) ||
"I need more information to proceed.",
sessionId: (parsedResult.session_id as string) || "",
timestamp: timestamp || new Date(),
};
}
if (responseType === "operation_started") {
return {
type: "operation_started",
toolName: (parsedResult.tool_name as string) || toolName,
toolId,
operationId: (parsedResult.operation_id as string) || "",
message:
(parsedResult.message as string) ||
"Operation started. You can close this tab.",
timestamp: timestamp || new Date(),
};
}
if (responseType === "operation_pending") {
return {
type: "operation_pending",
toolName: (parsedResult.tool_name as string) || toolName,
toolId,
operationId: (parsedResult.operation_id as string) || "",
message:
(parsedResult.message as string) ||
"Operation in progress. Please wait...",
timestamp: timestamp || new Date(),
};
}
if (responseType === "operation_in_progress") {
return {
type: "operation_in_progress",
toolName: (parsedResult.tool_name as string) || toolName,
toolCallId: (parsedResult.tool_call_id as string) || toolId,
message:
(parsedResult.message as string) ||
"Operation already in progress. Please wait...",
timestamp: timestamp || new Date(),
};
}
if (responseType === "need_login") {
return {
type: "login_needed",

364
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/useChatContainer.ts
View File

@@ -1,5 +1,6 @@
import type { SessionDetailResponse } from "@/app/api/__generated__/models/sessionDetailResponse";
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { useEffect, useMemo, useRef, useState } from "react";
import { useChatStore } from "../../chat-store";
import { toast } from "sonner";
import { useChatStream } from "../../useChatStream";
import { usePageContext } from "../../usePageContext";
@@ -9,23 +10,44 @@ import {
createUserMessage,
filterAuthMessages,
hasSentInitialPrompt,
isToolCallArray,
isValidMessage,
markInitialPromptSent,
parseToolResponse,
removePageContext,
processInitialMessages,
} from "./helpers";
// Helper to generate deduplication key for a message
function getMessageKey(msg: ChatMessageData): string {
if (msg.type === "message") {
// Don't include timestamp - dedupe by role + content only
// This handles the case where local and server timestamps differ
// Server messages are authoritative, so duplicates from local state are filtered
return `msg:${msg.role}:${msg.content}`;
} else if (msg.type === "tool_call") {
return `toolcall:${msg.toolId}`;
} else if (msg.type === "tool_response") {
return `toolresponse:${(msg as any).toolId}`;
} else if (
msg.type === "operation_started" ||
msg.type === "operation_pending" ||
msg.type === "operation_in_progress"
) {
return `op:${(msg as any).toolId || (msg as any).operationId || (msg as any).toolCallId || ""}:${msg.toolName}`;
} else {
return `${msg.type}:${JSON.stringify(msg).slice(0, 100)}`;
}
}
interface Args {
sessionId: string | null;
initialMessages: SessionDetailResponse["messages"];
initialPrompt?: string;
onOperationStarted?: () => void;
}
export function useChatContainer({
sessionId,
initialMessages,
initialPrompt,
onOperationStarted,
}: Args) {
const [messages, setMessages] = useState<ChatMessageData[]>([]);
const [streamingChunks, setStreamingChunks] = useState<string[]>([]);
@@ -41,11 +63,18 @@ export function useChatContainer({
sendMessage: sendStreamMessage,
stopStreaming,
} = useChatStream();
const activeStreams = useChatStore((s) => s.activeStreams);
const subscribeToStream = useChatStore((s) => s.subscribeToStream);
const isStreaming = isStreamingInitiated || hasTextChunks;
useEffect(() => {
if (sessionId !== previousSessionIdRef.current) {
stopStreaming(previousSessionIdRef.current ?? undefined, true);
useEffect(
function handleSessionChange() {
if (sessionId === previousSessionIdRef.current) return;
const prevSession = previousSessionIdRef.current;
if (prevSession) {
stopStreaming(prevSession);
}
previousSessionIdRef.current = sessionId;
setMessages([]);
setStreamingChunks([]);
@@ -53,138 +82,11 @@ export function useChatContainer({
setHasTextChunks(false);
setIsStreamingInitiated(false);
hasResponseRef.current = false;
}
}, [sessionId, stopStreaming]);
const allMessages = useMemo(() => {
const processedInitialMessages: ChatMessageData[] = [];
const toolCallMap = new Map<string, string>();
if (!sessionId) return;
for (const msg of initialMessages) {
if (!isValidMessage(msg)) {
console.warn("Invalid message structure from backend:", msg);
continue;
}
let content = String(msg.content || "");
const role = String(msg.role || "assistant").toLowerCase();
const toolCalls = msg.tool_calls;
const timestamp = msg.timestamp
? new Date(msg.timestamp as string)
: undefined;
if (role === "user") {
content = removePageContext(content);
if (!content.trim()) continue;
processedInitialMessages.push({
type: "message",
role: "user",
content,
timestamp,
});
continue;
}
if (role === "assistant") {
content = content
.replace(/<thinking>[\s\S]*?<\/thinking>/gi, "")
.trim();
if (toolCalls && isToolCallArray(toolCalls) && toolCalls.length > 0) {
for (const toolCall of toolCalls) {
const toolName = toolCall.function.name;
const toolId = toolCall.id;
toolCallMap.set(toolId, toolName);
try {
const args = JSON.parse(toolCall.function.arguments || "{}");
processedInitialMessages.push({
type: "tool_call",
toolId,
toolName,
arguments: args,
timestamp,
});
} catch (err) {
console.warn("Failed to parse tool call arguments:", err);
processedInitialMessages.push({
type: "tool_call",
toolId,
toolName,
arguments: {},
timestamp,
});
}
}
if (content.trim()) {
processedInitialMessages.push({
type: "message",
role: "assistant",
content,
timestamp,
});
}
} else if (content.trim()) {
processedInitialMessages.push({
type: "message",
role: "assistant",
content,
timestamp,
});
}
continue;
}
if (role === "tool") {
const toolCallId = (msg.tool_call_id as string) || "";
const toolName = toolCallMap.get(toolCallId) || "unknown";
const toolResponse = parseToolResponse(
content,
toolCallId,
toolName,
timestamp,
);
if (toolResponse) {
processedInitialMessages.push(toolResponse);
}
continue;
}
if (content.trim()) {
processedInitialMessages.push({
type: "message",
role: role as "user" | "assistant" | "system",
content,
timestamp,
});
}
}
return [...processedInitialMessages, ...messages];
}, [initialMessages, messages]);
const sendMessage = useCallback(
async function sendMessage(
content: string,
isUserMessage: boolean = true,
context?: { url: string; content: string },
) {
if (!sessionId) {
console.error("[useChatContainer] Cannot send message: no session ID");
return;
}
setIsRegionBlockedModalOpen(false);
if (isUserMessage) {
const userMessage = createUserMessage(content);
setMessages((prev) => [...filterAuthMessages(prev), userMessage]);
} else {
setMessages((prev) => filterAuthMessages(prev));
}
setStreamingChunks([]);
streamingChunksRef.current = [];
setHasTextChunks(false);
setIsStreamingInitiated(true);
hasResponseRef.current = false;
const activeStream = activeStreams.get(sessionId);
if (!activeStream || activeStream.status !== "streaming") return;
const dispatcher = createStreamEventDispatcher({
setHasTextChunks,
@@ -195,44 +97,170 @@ export function useChatContainer({
setIsRegionBlockedModalOpen,
sessionId,
setIsStreamingInitiated,
onOperationStarted,
});
try {
await sendStreamMessage(
sessionId,
content,
dispatcher,
isUserMessage,
context,
);
} catch (err) {
console.error("[useChatContainer] Failed to send message:", err);
setIsStreamingInitiated(false);
// Don't show error toast for AbortError (expected during cleanup)
if (err instanceof Error && err.name === "AbortError") return;
const errorMessage =
err instanceof Error ? err.message : "Failed to send message";
toast.error("Failed to send message", {
description: errorMessage,
});
}
setIsStreamingInitiated(true);
const skipReplay = initialMessages.length > 0;
return subscribeToStream(sessionId, dispatcher, skipReplay);
},
[sessionId, sendStreamMessage],
[
sessionId,
stopStreaming,
activeStreams,
subscribeToStream,
onOperationStarted,
],
);
const handleStopStreaming = useCallback(() => {
// Collect toolIds from completed tool results in initialMessages
// Used to filter out operation messages when their results arrive
const completedToolIds = useMemo(() => {
const processedInitial = processInitialMessages(initialMessages);
const ids = new Set<string>();
for (const msg of processedInitial) {
if (
msg.type === "tool_response" ||
msg.type === "agent_carousel" ||
msg.type === "execution_started"
) {
const toolId = (msg as any).toolId;
if (toolId) {
ids.add(toolId);
}
}
}
return ids;
}, [initialMessages]);
// Clean up local operation messages when their completed results arrive from polling
// This effect runs when completedToolIds changes (i.e., when polling brings new results)
useEffect(
function cleanupCompletedOperations() {
if (completedToolIds.size === 0) return;
setMessages((prev) => {
const filtered = prev.filter((msg) => {
if (
msg.type === "operation_started" ||
msg.type === "operation_pending" ||
msg.type === "operation_in_progress"
) {
const toolId = (msg as any).toolId || (msg as any).toolCallId;
if (toolId && completedToolIds.has(toolId)) {
return false; // Remove - operation completed
}
}
return true;
});
// Only update state if something was actually filtered
return filtered.length === prev.length ? prev : filtered;
});
},
[completedToolIds],
);
// Combine initial messages from backend with local streaming messages,
// Server messages maintain correct order; only append truly new local messages
const allMessages = useMemo(() => {
const processedInitial = processInitialMessages(initialMessages);
// Build a set of keys from server messages for deduplication
const serverKeys = new Set<string>();
for (const msg of processedInitial) {
serverKeys.add(getMessageKey(msg));
}
// Filter local messages: remove duplicates and completed operation messages
const newLocalMessages = messages.filter((msg) => {
// Remove operation messages for completed tools
if (
msg.type === "operation_started" ||
msg.type === "operation_pending" ||
msg.type === "operation_in_progress"
) {
const toolId = (msg as any).toolId || (msg as any).toolCallId;
if (toolId && completedToolIds.has(toolId)) {
return false;
}
}
// Remove messages that already exist in server data
const key = getMessageKey(msg);
return !serverKeys.has(key);
});
// Server messages first (correct order), then new local messages
return [...processedInitial, ...newLocalMessages];
}, [initialMessages, messages, completedToolIds]);
async function sendMessage(
content: string,
isUserMessage: boolean = true,
context?: { url: string; content: string },
) {
if (!sessionId) {
console.error("[useChatContainer] Cannot send message: no session ID");
return;
}
setIsRegionBlockedModalOpen(false);
if (isUserMessage) {
const userMessage = createUserMessage(content);
setMessages((prev) => [...filterAuthMessages(prev), userMessage]);
} else {
setMessages((prev) => filterAuthMessages(prev));
}
setStreamingChunks([]);
streamingChunksRef.current = [];
setHasTextChunks(false);
setIsStreamingInitiated(true);
hasResponseRef.current = false;
const dispatcher = createStreamEventDispatcher({
setHasTextChunks,
setStreamingChunks,
streamingChunksRef,
hasResponseRef,
setMessages,
setIsRegionBlockedModalOpen,
sessionId,
setIsStreamingInitiated,
onOperationStarted,
});
try {
await sendStreamMessage(
sessionId,
content,
dispatcher,
isUserMessage,
context,
);
} catch (err) {
console.error("[useChatContainer] Failed to send message:", err);
setIsStreamingInitiated(false);
if (err instanceof Error && err.name === "AbortError") return;
const errorMessage =
err instanceof Error ? err.message : "Failed to send message";
toast.error("Failed to send message", {
description: errorMessage,
});
}
}
function handleStopStreaming() {
stopStreaming();
setStreamingChunks([]);
streamingChunksRef.current = [];
setHasTextChunks(false);
setIsStreamingInitiated(false);
}, [stopStreaming]);
}
const { capturePageContext } = usePageContext();
const sendMessageRef = useRef(sendMessage);
sendMessageRef.current = sendMessage;
// Send initial prompt if provided (for new sessions from homepage)
useEffect(
function handleInitialPrompt() {
if (!initialPrompt || !sessionId) return;
@@ -241,15 +269,9 @@ export function useChatContainer({
markInitialPromptSent(sessionId);
const context = capturePageContext();
sendMessage(initialPrompt, true, context);
sendMessageRef.current(initialPrompt, true, context);
},
[
initialPrompt,
sessionId,
initialMessages.length,
sendMessage,
capturePageContext,
],
[initialPrompt, sessionId, initialMessages.length, capturePageContext],
);
async function sendMessageWithContext(

11
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatInput/ChatInput.tsx
View File

@@ -21,7 +21,7 @@ export function ChatInput({
className,
}: Props) {
const inputId = "chat-input";
const { value, setValue, handleKeyDown, handleSend, hasMultipleLines } =
const { value, handleKeyDown, handleSubmit, handleChange, hasMultipleLines } =
useChatInput({
onSend,
disabled: disabled || isStreaming,
@@ -29,15 +29,6 @@ export function ChatInput({
inputId,
});
function handleSubmit(e: React.FormEvent<HTMLFormElement>) {
e.preventDefault();
handleSend();
}
function handleChange(e: React.ChangeEvent<HTMLTextAreaElement>) {
setValue(e.target.value);
}
return (
<form onSubmit={handleSubmit} className={cn("relative flex-1", className)}>
<div className="relative">

55
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatInput/useChatInput.ts
View File

@@ -1,4 +1,10 @@
import { KeyboardEvent, useCallback, useEffect, useState } from "react";
import {
ChangeEvent,
FormEvent,
KeyboardEvent,
useEffect,
useState,
} from "react";
interface UseChatInputArgs {
onSend: (message: string) => void;
@@ -16,6 +22,23 @@ export function useChatInput({
const [value, setValue] = useState("");
const [hasMultipleLines, setHasMultipleLines] = useState(false);
useEffect(
function focusOnMount() {
const textarea = document.getElementById(inputId) as HTMLTextAreaElement;
if (textarea) textarea.focus();
},
[inputId],
);
useEffect(
function focusWhenEnabled() {
if (disabled) return;
const textarea = document.getElementById(inputId) as HTMLTextAreaElement;
if (textarea) textarea.focus();
},
[disabled, inputId],
);
useEffect(() => {
const textarea = document.getElementById(inputId) as HTMLTextAreaElement;
const wrapper = document.getElementById(
@@ -77,7 +100,7 @@ export function useChatInput({
}
}, [value, maxRows, inputId]);
const handleSend = useCallback(() => {
const handleSend = () => {
if (disabled || !value.trim()) return;
onSend(value.trim());
setValue("");
@@ -93,23 +116,31 @@ export function useChatInput({
wrapper.style.height = "";
wrapper.style.maxHeight = "";
}
}, [value, onSend, disabled, inputId]);
};
const handleKeyDown = useCallback(
(event: KeyboardEvent<HTMLTextAreaElement>) => {
if (event.key === "Enter" && !event.shiftKey) {
event.preventDefault();
handleSend();
}
},
[handleSend],
);
function handleKeyDown(event: KeyboardEvent<HTMLTextAreaElement>) {
if (event.key === "Enter" && !event.shiftKey) {
event.preventDefault();
handleSend();
}
}
function handleSubmit(e: FormEvent<HTMLFormElement>) {
e.preventDefault();
handleSend();
}
function handleChange(e: ChangeEvent<HTMLTextAreaElement>) {
setValue(e.target.value);
}
return {
value,
setValue,
handleKeyDown,
handleSend,
handleSubmit,
handleChange,
hasMultipleLines,
};
}

71
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatMessage/ChatMessage.tsx
View File

@@ -14,7 +14,9 @@ import { AgentCarouselMessage } from "../AgentCarouselMessage/AgentCarouselMessa
import { AIChatBubble } from "../AIChatBubble/AIChatBubble";
import { AuthPromptWidget } from "../AuthPromptWidget/AuthPromptWidget";
import { ChatCredentialsSetup } from "../ChatCredentialsSetup/ChatCredentialsSetup";
import { ClarificationQuestionsWidget } from "../ClarificationQuestionsWidget/ClarificationQuestionsWidget";
import { ExecutionStartedMessage } from "../ExecutionStartedMessage/ExecutionStartedMessage";
import { PendingOperationWidget } from "../PendingOperationWidget/PendingOperationWidget";
import { MarkdownContent } from "../MarkdownContent/MarkdownContent";
import { NoResultsMessage } from "../NoResultsMessage/NoResultsMessage";
import { ToolCallMessage } from "../ToolCallMessage/ToolCallMessage";
@@ -69,6 +71,10 @@ export function ChatMessage({
isToolResponse,
isLoginNeeded,
isCredentialsNeeded,
isClarificationNeeded,
isOperationStarted,
isOperationPending,
isOperationInProgress,
} = useChatMessage(message);
const displayContent = getDisplayContent(message, isUser);
@@ -96,6 +102,18 @@ export function ChatMessage({
}
}
function handleClarificationAnswers(answers: Record<string, string>) {
if (onSendMessage) {
const contextMessage = Object.entries(answers)
.map(([keyword, answer]) => `${keyword}: ${answer}`)
.join("\n");
onSendMessage(
`I have the answers to your questions:\n\n${contextMessage}\n\nPlease proceed with creating the agent.`,
);
}
}
const handleCopy = useCallback(
async function handleCopy() {
if (message.type !== "message") return;
@@ -112,10 +130,6 @@ export function ChatMessage({
[displayContent, message],
);
function isLongResponse(content: string): boolean {
return content.split("\n").length > 5;
}
const handleTryAgain = useCallback(() => {
if (message.type !== "message" || !onSendMessage) return;
onSendMessage(message.content, message.role === "user");
@@ -141,6 +155,17 @@ export function ChatMessage({
);
}
if (isClarificationNeeded && message.type === "clarification_needed") {
return (
<ClarificationQuestionsWidget
questions={message.questions}
message={message.message}
onSubmitAnswers={handleClarificationAnswers}
className={className}
/>
);
}
// Render login needed messages
if (isLoginNeeded && message.type === "login_needed") {
// If user is already logged in, show success message instead of auth prompt
@@ -269,6 +294,42 @@ export function ChatMessage({
);
}
// Render operation_started messages (long-running background operations)
if (isOperationStarted && message.type === "operation_started") {
return (
<PendingOperationWidget
status="started"
message={message.message}
toolName={message.toolName}
className={className}
/>
);
}
// Render operation_pending messages (operations in progress when refreshing)
if (isOperationPending && message.type === "operation_pending") {
return (
<PendingOperationWidget
status="pending"
message={message.message}
toolName={message.toolName}
className={className}
/>
);
}
// Render operation_in_progress messages (duplicate request while operation running)
if (isOperationInProgress && message.type === "operation_in_progress") {
return (
<PendingOperationWidget
status="in_progress"
message={message.message}
toolName={message.toolName}
className={className}
/>
);
}
// Render tool response messages (but skip agent_output if it's being rendered inside assistant message)
if (isToolResponse && message.type === "tool_response") {
return (
@@ -333,7 +394,7 @@ export function ChatMessage({
<ArrowsClockwiseIcon className="size-4 text-zinc-600" />
</Button>
)}
{!isUser && isFinalMessage && isLongResponse(displayContent) && (
{!isUser && isFinalMessage && !isStreaming && (
<Button
variant="ghost"
size="icon"

41
autogpt_platform/frontend/src/components/contextual/Chat/components/ChatMessage/useChatMessage.ts
View File

@@ -61,6 +61,7 @@ export type ChatMessageData =
}
| {
type: "agent_carousel";
toolId: string;
toolName: string;
agents: Array<{
id: string;
@@ -74,6 +75,7 @@ export type ChatMessageData =
}
| {
type: "execution_started";
toolId: string;
toolName: string;
executionId: string;
agentName?: string;
@@ -91,6 +93,41 @@ export type ChatMessageData =
credentialsSchema?: Record<string, any>;
message: string;
timestamp?: string | Date;
}
| {
type: "clarification_needed";
toolName: string;
questions: Array<{
question: string;
keyword: string;
example?: string;
}>;
message: string;
sessionId: string;
timestamp?: string | Date;
}
| {
type: "operation_started";
toolName: string;
toolId: string;
operationId: string;
message: string;
timestamp?: string | Date;
}
| {
type: "operation_pending";
toolName: string;
toolId: string;
operationId: string;
message: string;
timestamp?: string | Date;
}
| {
type: "operation_in_progress";
toolName: string;
toolCallId: string;
message: string;
timestamp?: string | Date;
};
export function useChatMessage(message: ChatMessageData) {
@@ -111,5 +148,9 @@ export function useChatMessage(message: ChatMessageData) {
isAgentCarousel: message.type === "agent_carousel",
isExecutionStarted: message.type === "execution_started",
isInputsNeeded: message.type === "inputs_needed",
isClarificationNeeded: message.type === "clarification_needed",
isOperationStarted: message.type === "operation_started",
isOperationPending: message.type === "operation_pending",
isOperationInProgress: message.type === "operation_in_progress",
};
}

186
autogpt_platform/frontend/src/components/contextual/Chat/components/ClarificationQuestionsWidget/ClarificationQuestionsWidget.tsx Normal file
View File

@@ -0,0 +1,186 @@
"use client";
import { Button } from "@/components/atoms/Button/Button";
import { Card } from "@/components/atoms/Card/Card";
import { Input } from "@/components/atoms/Input/Input";
import { Text } from "@/components/atoms/Text/Text";
import { cn } from "@/lib/utils";
import { CheckCircleIcon, QuestionIcon } from "@phosphor-icons/react";
import { useState } from "react";
export interface ClarifyingQuestion {
question: string;
keyword: string;
example?: string;
}
interface Props {
questions: ClarifyingQuestion[];
message: string;
onSubmitAnswers: (answers: Record<string, string>) => void;
onCancel?: () => void;
className?: string;
}
export function ClarificationQuestionsWidget({
questions,
message,
onSubmitAnswers,
onCancel,
className,
}: Props) {
const [answers, setAnswers] = useState<Record<string, string>>({});
const [isSubmitted, setIsSubmitted] = useState(false);
function handleAnswerChange(keyword: string, value: string) {
setAnswers((prev) => ({ ...prev, [keyword]: value }));
}
function handleSubmit() {
// Check if all questions are answered
const allAnswered = questions.every((q) => answers[q.keyword]?.trim());
if (!allAnswered) {
return;
}
setIsSubmitted(true);
onSubmitAnswers(answers);
}
const allAnswered = questions.every((q) => answers[q.keyword]?.trim());
// Show submitted state after answers are submitted
if (isSubmitted) {
return (
<div
className={cn(
"group relative flex w-full justify-start gap-3 px-4 py-3",
className,
)}
>
<div className="flex w-full max-w-3xl gap-3">
<div className="flex-shrink-0">
<div className="flex h-7 w-7 items-center justify-center rounded-lg bg-green-500">
<CheckCircleIcon className="h-4 w-4 text-white" weight="bold" />
</div>
</div>
<div className="flex min-w-0 flex-1 flex-col">
<Card className="p-4">
<Text variant="h4" className="mb-1 text-slate-900">
Answers submitted
</Text>
<Text variant="small" className="text-slate-600">
Processing your responses...
</Text>
</Card>
</div>
</div>
</div>
);
}
return (
<div
className={cn(
"group relative flex w-full justify-start gap-3 px-4 py-3",
className,
)}
>
<div className="flex w-full max-w-3xl gap-3">
<div className="flex-shrink-0">
<div className="flex h-7 w-7 items-center justify-center rounded-lg bg-indigo-500">
<QuestionIcon className="h-4 w-4 text-indigo-50" weight="bold" />
</div>
</div>
<div className="flex min-w-0 flex-1 flex-col">
<Card className="space-y-4 p-4">
<div>
<Text variant="h4" className="mb-1 text-slate-900">
I need more information
</Text>
<Text variant="small" className="text-slate-600">
{message}
</Text>
</div>
<div className="space-y-3">
{questions.map((q, index) => {
const isAnswered = !!answers[q.keyword]?.trim();
return (
<div
key={`${q.keyword}-${index}`}
className={cn(
"relative rounded-lg border p-3",
isAnswered
? "border-green-500 bg-green-50/50"
: "border-slate-200 bg-white/50",
)}
>
<div className="mb-2 flex items-start gap-2">
{isAnswered ? (
<CheckCircleIcon
size={16}
className="mt-0.5 text-green-500"
weight="bold"
/>
) : (
<div className="mt-0.5 flex h-4 w-4 items-center justify-center rounded-full border border-slate-300 bg-white text-xs text-slate-500">
{index + 1}
</div>
)}
<div className="flex-1">
<Text
variant="small"
className="mb-2 font-semibold text-slate-900"
>
{q.question}
</Text>
{q.example && (
<Text
variant="small"
className="mb-2 italic text-slate-500"
>
Example: {q.example}
</Text>
)}
<Input
type="textarea"
id={`clarification-${q.keyword}-${index}`}
label={q.question}
hideLabel
placeholder="Your answer..."
rows={2}
value={answers[q.keyword] || ""}
onChange={(e) =>
handleAnswerChange(q.keyword, e.target.value)
}
/>
</div>
</div>
</div>
);
})}
</div>
<div className="flex gap-2">
<Button
onClick={handleSubmit}
disabled={!allAnswered}
className="flex-1"
variant="primary"
>
Submit Answers
</Button>
{onCancel && (
<Button onClick={onCancel} variant="outline">
Cancel
</Button>
)}
</div>
</Card>
</div>
</div>
</div>
);
}

15
autogpt_platform/frontend/src/components/contextual/Chat/components/MessageList/components/LastToolResponse/LastToolResponse.tsx
View File

@@ -1,7 +1,5 @@
import { AIChatBubble } from "../../../AIChatBubble/AIChatBubble";
import type { ChatMessageData } from "../../../ChatMessage/useChatMessage";
import { MarkdownContent } from "../../../MarkdownContent/MarkdownContent";
import { formatToolResponse } from "../../../ToolResponseMessage/helpers";
import { ToolResponseMessage } from "../../../ToolResponseMessage/ToolResponseMessage";
import { shouldSkipAgentOutput } from "../../helpers";
export interface LastToolResponseProps {
@@ -15,16 +13,15 @@ export function LastToolResponse({
}: LastToolResponseProps) {
if (message.type !== "tool_response") return null;
// Skip if this is an agent_output that should be rendered inside assistant message
if (shouldSkipAgentOutput(message, prevMessage)) return null;
const formattedText = formatToolResponse(message.result, message.toolName);
return (
<div className="min-w-0 overflow-x-hidden hyphens-auto break-words px-4 py-2">
<AIChatBubble>
<MarkdownContent content={formattedText} />
</AIChatBubble>
<ToolResponseMessage
toolId={message.toolId}
toolName={message.toolName}
result={message.result}
/>
</div>
);
}

109
autogpt_platform/frontend/src/components/contextual/Chat/components/PendingOperationWidget/PendingOperationWidget.tsx Normal file
View File

@@ -0,0 +1,109 @@
"use client";
import { Card } from "@/components/atoms/Card/Card";
import { Text } from "@/components/atoms/Text/Text";
import { cn } from "@/lib/utils";
import { CircleNotch, CheckCircle, XCircle } from "@phosphor-icons/react";
type OperationStatus =
| "pending"
| "started"
| "in_progress"
| "completed"
| "error";
interface Props {
status: OperationStatus;
message: string;
toolName?: string;
className?: string;
}
function getOperationTitle(toolName?: string): string {
if (!toolName) return "Operation";
// Convert tool name to human-readable format
// e.g., "create_agent" -> "Creating Agent", "edit_agent" -> "Editing Agent"
if (toolName === "create_agent") return "Creating Agent";
if (toolName === "edit_agent") return "Editing Agent";
// Default: capitalize and format tool name
return toolName
.split("_")
.map((word) => word.charAt(0).toUpperCase() + word.slice(1))
.join(" ");
}
export function PendingOperationWidget({
status,
message,
toolName,
className,
}: Props) {
const isPending =
status === "pending" || status === "started" || status === "in_progress";
const isCompleted = status === "completed";
const isError = status === "error";
const operationTitle = getOperationTitle(toolName);
return (
<div
className={cn(
"group relative flex w-full justify-start gap-3 px-4 py-3",
className,
)}
>
<div className="flex w-full max-w-3xl gap-3">
<div className="flex-shrink-0">
<div
className={cn(
"flex h-7 w-7 items-center justify-center rounded-lg",
isPending && "bg-blue-500",
isCompleted && "bg-green-500",
isError && "bg-red-500",
)}
>
{isPending && (
<CircleNotch
className="h-4 w-4 animate-spin text-white"
weight="bold"
/>
)}
{isCompleted && (
<CheckCircle className="h-4 w-4 text-white" weight="bold" />
)}
{isError && (
<XCircle className="h-4 w-4 text-white" weight="bold" />
)}
</div>
</div>
<div className="flex min-w-0 flex-1 flex-col">
<Card className="space-y-2 p-4">
<div>
<Text variant="h4" className="mb-1 text-slate-900">
{isPending && operationTitle}
{isCompleted && `${operationTitle} Complete`}
{isError && `${operationTitle} Failed`}
</Text>
<Text variant="small" className="text-slate-600">
{message}
</Text>
</div>
{isPending && (
<Text variant="small" className="italic text-slate-500">
Check your library in a few minutes.
</Text>
)}
{toolName && (
<Text variant="small" className="text-slate-400">
Tool: {toolName}
</Text>
)}
</Card>
</div>
</div>
</div>
);
}

27
autogpt_platform/frontend/src/components/contextual/Chat/components/ToolResponseMessage/ToolResponseMessage.tsx
View File

@@ -1,7 +1,14 @@
import { Text } from "@/components/atoms/Text/Text";
import { cn } from "@/lib/utils";
import type { ToolResult } from "@/types/chat";
import { WarningCircleIcon } from "@phosphor-icons/react";
import { AIChatBubble } from "../AIChatBubble/AIChatBubble";
import { MarkdownContent } from "../MarkdownContent/MarkdownContent";
import { formatToolResponse } from "./helpers";
import {
formatToolResponse,
getErrorMessage,
isErrorResponse,
} from "./helpers";
export interface ToolResponseMessageProps {
toolId?: string;
@@ -18,6 +25,24 @@ export function ToolResponseMessage({
success: _success,
className,
}: ToolResponseMessageProps) {
if (isErrorResponse(result)) {
const errorMessage = getErrorMessage(result);
return (
<AIChatBubble className={className}>
<div className="flex items-center gap-2">
<WarningCircleIcon
size={14}
weight="regular"
className="shrink-0 text-neutral-400"
/>
<Text variant="small" className={cn("text-xs text-neutral-500")}>
{errorMessage}
</Text>
</div>
</AIChatBubble>
);
}
const formattedText = formatToolResponse(result, toolName);
return (

43
autogpt_platform/frontend/src/components/contextual/Chat/components/ToolResponseMessage/helpers.ts
View File

@@ -1,3 +1,42 @@
function stripInternalReasoning(content: string): string {
return content
.replace(/<internal_reasoning>[\s\S]*?<\/internal_reasoning>/gi, "")
.replace(/<thinking>[\s\S]*?<\/thinking>/gi, "")
.replace(/\n{3,}/g, "\n\n")
.trim();
}
export function isErrorResponse(result: unknown): boolean {
if (typeof result === "string") {
const lower = result.toLowerCase();
return (
lower.startsWith("error:") ||
lower.includes("not found") ||
lower.includes("does not exist") ||
lower.includes("failed to") ||
lower.includes("unable to")
);
}
if (typeof result === "object" && result !== null) {
const response = result as Record<string, unknown>;
return response.type === "error" || response.error !== undefined;
}
return false;
}
export function getErrorMessage(result: unknown): string {
if (typeof result === "string") {
return stripInternalReasoning(result.replace(/^error:\s*/i, ""));
}
if (typeof result === "object" && result !== null) {
const response = result as Record<string, unknown>;
if (response.error) return stripInternalReasoning(String(response.error));
if (response.message)
return stripInternalReasoning(String(response.message));
}
return "An error occurred";
}
function getToolCompletionPhrase(toolName: string): string {
const toolCompletionPhrases: Record<string, string> = {
add_understanding: "Updated your business information",
@@ -28,10 +67,10 @@ export function formatToolResponse(result: unknown, toolName: string): string {
const parsed = JSON.parse(trimmed);
return formatToolResponse(parsed, toolName);
} catch {
return trimmed;
return stripInternalReasoning(trimmed);
}
}
return result;
return stripInternalReasoning(result);
}
if (typeof result !== "object" || result === null) {

2
autogpt_platform/frontend/src/components/contextual/Chat/components/UserChatBubble/UserChatBubble.tsx
View File

@@ -10,7 +10,7 @@ export function UserChatBubble({ children, className }: UserChatBubbleProps) {
return (
<div
className={cn(
"group relative min-w-20 overflow-hidden rounded-xl bg-purple-100 px-3 text-right text-[1rem] leading-relaxed transition-all duration-500 ease-in-out",
"group relative min-w-20 overflow-hidden rounded-xl bg-purple-100 px-3 text-left text-[1rem] leading-relaxed transition-all duration-500 ease-in-out",
className,
)}
style={{

142
autogpt_platform/frontend/src/components/contextual/Chat/stream-executor.ts Normal file
View File

@@ -0,0 +1,142 @@
import type {
ActiveStream,
StreamChunk,
VercelStreamChunk,
} from "./chat-types";
import {
INITIAL_RETRY_DELAY,
MAX_RETRIES,
normalizeStreamChunk,
parseSSELine,
} from "./stream-utils";
function notifySubscribers(stream: ActiveStream, chunk: StreamChunk) {
stream.chunks.push(chunk);
for (const callback of stream.onChunkCallbacks) {
try {
callback(chunk);
} catch (err) {
console.warn("[StreamExecutor] Subscriber callback error:", err);
}
}
}
export async function executeStream(
stream: ActiveStream,
message: string,
isUserMessage: boolean,
context?: { url: string; content: string },
retryCount: number = 0,
): Promise<void> {
const { sessionId, abortController } = stream;
try {
const url = `/api/chat/sessions/${sessionId}/stream`;
const body = JSON.stringify({
message,
is_user_message: isUserMessage,
context: context || null,
});
const response = await fetch(url, {
method: "POST",
headers: {
"Content-Type": "application/json",
Accept: "text/event-stream",
},
body,
signal: abortController.signal,
});
if (!response.ok) {
const errorText = await response.text();
throw new Error(errorText || `HTTP ${response.status}`);
}
if (!response.body) {
throw new Error("Response body is null");
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) {
notifySubscribers(stream, { type: "stream_end" });
stream.status = "completed";
return;
}
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() || "";
for (const line of lines) {
const data = parseSSELine(line);
if (data !== null) {
if (data === "[DONE]") {
notifySubscribers(stream, { type: "stream_end" });
stream.status = "completed";
return;
}
try {
const rawChunk = JSON.parse(data) as
| StreamChunk
| VercelStreamChunk;
const chunk = normalizeStreamChunk(rawChunk);
if (!chunk) continue;
notifySubscribers(stream, chunk);
if (chunk.type === "stream_end") {
stream.status = "completed";
return;
}
if (chunk.type === "error") {
stream.status = "error";
stream.error = new Error(
chunk.message || chunk.content || "Stream error",
);
return;
}
} catch (err) {
console.warn("[StreamExecutor] Failed to parse SSE chunk:", err);
}
}
}
}
} catch (err) {
if (err instanceof Error && err.name === "AbortError") {
notifySubscribers(stream, { type: "stream_end" });
stream.status = "completed";
return;
}
if (retryCount < MAX_RETRIES) {
const retryDelay = INITIAL_RETRY_DELAY * Math.pow(2, retryCount);
console.log(
`[StreamExecutor] Retrying in ${retryDelay}ms (attempt ${retryCount + 1}/${MAX_RETRIES})`,
);
await new Promise((resolve) => setTimeout(resolve, retryDelay));
return executeStream(
stream,
message,
isUserMessage,
context,
retryCount + 1,
);
}
stream.status = "error";
stream.error = err instanceof Error ? err : new Error("Stream failed");
notifySubscribers(stream, {
type: "error",
message: stream.error.message,
});
}
}

84
autogpt_platform/frontend/src/components/contextual/Chat/stream-utils.ts Normal file
View File

@@ -0,0 +1,84 @@
import type { ToolArguments, ToolResult } from "@/types/chat";
import type { StreamChunk, VercelStreamChunk } from "./chat-types";
const LEGACY_STREAM_TYPES = new Set<StreamChunk["type"]>([
"text_chunk",
"text_ended",
"tool_call",
"tool_call_start",
"tool_response",
"login_needed",
"need_login",
"credentials_needed",
"error",
"usage",
"stream_end",
]);
export function isLegacyStreamChunk(
chunk: StreamChunk | VercelStreamChunk,
): chunk is StreamChunk {
return LEGACY_STREAM_TYPES.has(chunk.type as StreamChunk["type"]);
}
export function normalizeStreamChunk(
chunk: StreamChunk | VercelStreamChunk,
): StreamChunk | null {
if (isLegacyStreamChunk(chunk)) return chunk;
switch (chunk.type) {
case "text-delta":
return { type: "text_chunk", content: chunk.delta };
case "text-end":
return { type: "text_ended" };
case "tool-input-available":
return {
type: "tool_call_start",
tool_id: chunk.toolCallId,
tool_name: chunk.toolName,
arguments: chunk.input as ToolArguments,
};
case "tool-output-available":
return {
type: "tool_response",
tool_id: chunk.toolCallId,
tool_name: chunk.toolName,
result: chunk.output as ToolResult,
success: chunk.success ?? true,
};
case "usage":
return {
type: "usage",
promptTokens: chunk.promptTokens,
completionTokens: chunk.completionTokens,
totalTokens: chunk.totalTokens,
};
case "error":
return {
type: "error",
message: chunk.errorText,
code: chunk.code,
details: chunk.details,
};
case "finish":
return { type: "stream_end" };
case "start":
case "text-start":
return null;
case "tool-input-start":
return {
type: "tool_call_start",
tool_id: chunk.toolCallId,
tool_name: chunk.toolName,
arguments: {},
};
}
}
export const MAX_RETRIES = 3;
export const INITIAL_RETRY_DELAY = 1000;
export function parseSSELine(line: string): string | null {
if (line.startsWith("data: ")) return line.slice(6);
return null;
}

43
autogpt_platform/frontend/src/components/contextual/Chat/useChat.ts
View File

@@ -2,7 +2,6 @@
import { useSupabase } from "@/lib/supabase/hooks/useSupabase";
import { useEffect, useRef, useState } from "react";
import { toast } from "sonner";
import { useChatSession } from "./useChatSession";
import { useChatStream } from "./useChatStream";
@@ -27,6 +26,7 @@ export function useChat({ urlSessionId }: UseChatArgs = {}) {
claimSession,
clearSession: clearSessionBase,
loadSession,
startPollingForOperation,
} = useChatSession({
urlSessionId,
autoCreate: false,
@@ -67,38 +67,16 @@ export function useChat({ urlSessionId }: UseChatArgs = {}) {
],
);
useEffect(() => {
if (isLoading || isCreating) {
const timer = setTimeout(() => {
setShowLoader(true);
}, 300);
return () => clearTimeout(timer);
} else {
useEffect(
function showLoaderWithDelay() {
if (isLoading || isCreating) {
const timer = setTimeout(() => setShowLoader(true), 300);
return () => clearTimeout(timer);
}
setShowLoader(false);
}
}, [isLoading, isCreating]);
useEffect(function monitorNetworkStatus() {
function handleOnline() {
toast.success("Connection restored", {
description: "You're back online",
});
}
function handleOffline() {
toast.error("You're offline", {
description: "Check your internet connection",
});
}
window.addEventListener("online", handleOnline);
window.addEventListener("offline", handleOffline);
return () => {
window.removeEventListener("online", handleOnline);
window.removeEventListener("offline", handleOffline);
};
}, []);
},
[isLoading, isCreating],
);
function clearSession() {
clearSessionBase();
@@ -117,5 +95,6 @@ export function useChat({ urlSessionId }: UseChatArgs = {}) {
loadSession,
sessionId: sessionIdFromHook,
showLoader,
startPollingForOperation,
};
}

17
autogpt_platform/frontend/src/components/contextual/Chat/useChatDrawer.ts
View File

@@ -1,17 +0,0 @@
"use client";
import { create } from "zustand";
interface ChatDrawerState {
isOpen: boolean;
open: () => void;
close: () => void;
toggle: () => void;
}
export const useChatDrawer = create<ChatDrawerState>((set) => ({
isOpen: false,
open: () => set({ isOpen: true }),
close: () => set({ isOpen: false }),
toggle: () => set((state) => ({ isOpen: !state.isOpen })),
}));

123
autogpt_platform/frontend/src/components/contextual/Chat/useChatSession.ts
View File

@@ -1,6 +1,7 @@
import {
getGetV2GetSessionQueryKey,
getGetV2GetSessionQueryOptions,
getGetV2ListSessionsQueryKey,
postV2CreateSession,
useGetV2GetSession,
usePatchV2SessionAssignUser,
@@ -58,6 +59,7 @@ export function useChatSession({
query: {
enabled: !!sessionId,
select: okData,
staleTime: 0,
retry: shouldRetrySessionLoad,
retryDelay: getSessionRetryDelay,
},
@@ -101,6 +103,125 @@ export function useChatSession({
}
}, [createError, loadError]);
// Track if we should be polling (set by external callers when they receive operation_started via SSE)
const [forcePolling, setForcePolling] = useState(false);
// Track if we've seen server acknowledge the pending operation (to avoid clearing forcePolling prematurely)
const hasSeenServerPendingRef = useRef(false);
// Check if there are any pending operations in the messages
// Must check all operation types: operation_pending, operation_started, operation_in_progress
const hasPendingOperationsFromServer = useMemo(() => {
if (!messages || messages.length === 0) return false;
const pendingTypes = new Set([
"operation_pending",
"operation_in_progress",
"operation_started",
]);
return messages.some((msg) => {
if (msg.role !== "tool" || !msg.content) return false;
try {
const content =
typeof msg.content === "string"
? JSON.parse(msg.content)
: msg.content;
return pendingTypes.has(content?.type);
} catch {
return false;
}
});
}, [messages]);
// Track when server has acknowledged the pending operation
useEffect(() => {
if (hasPendingOperationsFromServer) {
hasSeenServerPendingRef.current = true;
}
}, [hasPendingOperationsFromServer]);
// Combined: poll if server has pending ops OR if we received operation_started via SSE
const hasPendingOperations = hasPendingOperationsFromServer || forcePolling;
// Clear forcePolling only after server has acknowledged AND completed the operation
useEffect(() => {
if (
forcePolling &&
!hasPendingOperationsFromServer &&
hasSeenServerPendingRef.current
) {
// Server acknowledged the operation and it's now complete
setForcePolling(false);
hasSeenServerPendingRef.current = false;
}
}, [forcePolling, hasPendingOperationsFromServer]);
// Function to trigger polling (called when operation_started is received via SSE)
function startPollingForOperation() {
setForcePolling(true);
hasSeenServerPendingRef.current = false; // Reset for new operation
}
// Refresh sessions list when a pending operation completes
// (hasPendingOperations transitions from true to false)
const prevHasPendingOperationsRef = useRef(hasPendingOperations);
useEffect(
function refreshSessionsListOnOperationComplete() {
const wasHasPending = prevHasPendingOperationsRef.current;
prevHasPendingOperationsRef.current = hasPendingOperations;
// Only invalidate when transitioning from pending to not pending
if (wasHasPending && !hasPendingOperations && sessionId) {
queryClient.invalidateQueries({
queryKey: getGetV2ListSessionsQueryKey(),
});
}
},
[hasPendingOperations, sessionId, queryClient],
);
// Poll for updates when there are pending operations
// Backoff: 2s, 4s, 6s, 8s, 10s, ... up to 30s max
const pollAttemptRef = useRef(0);
const hasPendingOperationsRef = useRef(hasPendingOperations);
hasPendingOperationsRef.current = hasPendingOperations;
useEffect(
function pollForPendingOperations() {
if (!sessionId || !hasPendingOperations) {
pollAttemptRef.current = 0;
return;
}
let cancelled = false;
let timeoutId: ReturnType<typeof setTimeout> | null = null;
function schedule() {
// 2s, 4s, 6s, 8s, 10s, ... 30s (max)
const delay = Math.min((pollAttemptRef.current + 1) * 2000, 30000);
timeoutId = setTimeout(async () => {
if (cancelled) return;
pollAttemptRef.current += 1;
try {
await refetch();
} catch (err) {
console.error("[useChatSession] Poll failed:", err);
} finally {
if (!cancelled && hasPendingOperationsRef.current) {
schedule();
}
}
}, delay);
}
schedule();
return () => {
cancelled = true;
if (timeoutId) clearTimeout(timeoutId);
};
},
[sessionId, hasPendingOperations, refetch],
);
async function createSession() {
try {
setError(null);
@@ -227,11 +348,13 @@ export function useChatSession({
isCreating,
error,
isSessionNotFound: isNotFoundError(loadError),
hasPendingOperations,
createSession,
loadSession,
refreshSession,
claimSession,
clearSession,
startPollingForOperation,
};
}

615
autogpt_platform/frontend/src/components/contextual/Chat/useChatStream.ts
View File

@@ -1,543 +1,110 @@
import type { ToolArguments, ToolResult } from "@/types/chat";
import { useCallback, useEffect, useRef, useState } from "react";
"use client";
import { useEffect, useRef, useState } from "react";
import { toast } from "sonner";
import { useChatStore } from "./chat-store";
import type { StreamChunk } from "./chat-types";
const MAX_RETRIES = 3;
const INITIAL_RETRY_DELAY = 1000;
export interface StreamChunk {
type:
| "text_chunk"
| "text_ended"
| "tool_call"
| "tool_call_start"
| "tool_response"
| "login_needed"
| "need_login"
| "credentials_needed"
| "error"
| "usage"
| "stream_end";
timestamp?: string;
content?: string;
message?: string;
code?: string;
details?: Record<string, unknown>;
tool_id?: string;
tool_name?: string;
arguments?: ToolArguments;
result?: ToolResult;
success?: boolean;
idx?: number;
session_id?: string;
agent_info?: {
graph_id: string;
name: string;
trigger_type: string;
};
provider?: string;
provider_name?: string;
credential_type?: string;
scopes?: string[];
title?: string;
[key: string]: unknown;
}
type VercelStreamChunk =
| { type: "start"; messageId: string }
| { type: "finish" }
| { type: "text-start"; id: string }
| { type: "text-delta"; id: string; delta: string }
| { type: "text-end"; id: string }
| { type: "tool-input-start"; toolCallId: string; toolName: string }
| {
type: "tool-input-available";
toolCallId: string;
toolName: string;
input: ToolArguments;
}
| {
type: "tool-output-available";
toolCallId: string;
toolName?: string;
output: ToolResult;
success?: boolean;
}
| {
type: "usage";
promptTokens: number;
completionTokens: number;
totalTokens: number;
}
| {
type: "error";
errorText: string;
code?: string;
details?: Record<string, unknown>;
};
const LEGACY_STREAM_TYPES = new Set<StreamChunk["type"]>([
"text_chunk",
"text_ended",
"tool_call",
"tool_call_start",
"tool_response",
"login_needed",
"need_login",
"credentials_needed",
"error",
"usage",
"stream_end",
]);
function isLegacyStreamChunk(
chunk: StreamChunk | VercelStreamChunk,
): chunk is StreamChunk {
return LEGACY_STREAM_TYPES.has(chunk.type as StreamChunk["type"]);
}
function normalizeStreamChunk(
chunk: StreamChunk | VercelStreamChunk,
): StreamChunk | null {
if (isLegacyStreamChunk(chunk)) {
return chunk;
}
switch (chunk.type) {
case "text-delta":
return { type: "text_chunk", content: chunk.delta };
case "text-end":
return { type: "text_ended" };
case "tool-input-available":
return {
type: "tool_call_start",
tool_id: chunk.toolCallId,
tool_name: chunk.toolName,
arguments: chunk.input,
};
case "tool-output-available":
return {
type: "tool_response",
tool_id: chunk.toolCallId,
tool_name: chunk.toolName,
result: chunk.output,
success: chunk.success ?? true,
};
case "usage":
return {
type: "usage",
promptTokens: chunk.promptTokens,
completionTokens: chunk.completionTokens,
totalTokens: chunk.totalTokens,
};
case "error":
return {
type: "error",
message: chunk.errorText,
code: chunk.code,
details: chunk.details,
};
case "finish":
return { type: "stream_end" };
case "start":
case "text-start":
return null;
case "tool-input-start":
const toolInputStart = chunk as Extract<
VercelStreamChunk,
{ type: "tool-input-start" }
>;
return {
type: "tool_call_start",
tool_id: toolInputStart.toolCallId,
tool_name: toolInputStart.toolName,
arguments: {},
};
}
}
export type { StreamChunk } from "./chat-types";
export function useChatStream() {
const [isStreaming, setIsStreaming] = useState(false);
const [error, setError] = useState<Error | null>(null);
const retryCountRef = useRef<number>(0);
const retryTimeoutRef = useRef<NodeJS.Timeout | null>(null);
const abortControllerRef = useRef<AbortController | null>(null);
const currentSessionIdRef = useRef<string | null>(null);
const requestStartTimeRef = useRef<number | null>(null);
const stopStreaming = useCallback(
(sessionId?: string, force: boolean = false) => {
console.log("[useChatStream] stopStreaming called", {
hasAbortController: !!abortControllerRef.current,
isAborted: abortControllerRef.current?.signal.aborted,
currentSessionId: currentSessionIdRef.current,
requestedSessionId: sessionId,
requestStartTime: requestStartTimeRef.current,
timeSinceStart: requestStartTimeRef.current
? Date.now() - requestStartTimeRef.current
: null,
force,
stack: new Error().stack,
});
if (
sessionId &&
currentSessionIdRef.current &&
currentSessionIdRef.current !== sessionId
) {
console.log(
"[useChatStream] Session changed, aborting previous stream",
{
oldSessionId: currentSessionIdRef.current,
newSessionId: sessionId,
},
);
}
const controller = abortControllerRef.current;
if (controller) {
const timeSinceStart = requestStartTimeRef.current
? Date.now() - requestStartTimeRef.current
: null;
if (!force && timeSinceStart !== null && timeSinceStart < 100) {
console.log(
"[useChatStream] Request just started (<100ms), skipping abort to prevent race condition",
{
timeSinceStart,
},
);
return;
}
try {
const signal = controller.signal;
if (
signal &&
typeof signal.aborted === "boolean" &&
!signal.aborted
) {
console.log("[useChatStream] Aborting stream");
controller.abort();
} else {
console.log(
"[useChatStream] Stream already aborted or signal invalid",
);
}
} catch (error) {
if (error instanceof Error && error.name === "AbortError") {
console.log(
"[useChatStream] AbortError caught (expected during cleanup)",
);
} else {
console.warn("[useChatStream] Error aborting stream:", error);
}
} finally {
abortControllerRef.current = null;
requestStartTimeRef.current = null;
}
}
if (retryTimeoutRef.current) {
clearTimeout(retryTimeoutRef.current);
retryTimeoutRef.current = null;
}
setIsStreaming(false);
},
[],
const onChunkCallbackRef = useRef<((chunk: StreamChunk) => void) | null>(
null,
);
const stopStream = useChatStore((s) => s.stopStream);
const unregisterActiveSession = useChatStore(
(s) => s.unregisterActiveSession,
);
const isSessionActive = useChatStore((s) => s.isSessionActive);
const onStreamComplete = useChatStore((s) => s.onStreamComplete);
const getCompletedStream = useChatStore((s) => s.getCompletedStream);
const registerActiveSession = useChatStore((s) => s.registerActiveSession);
const startStream = useChatStore((s) => s.startStream);
const getStreamStatus = useChatStore((s) => s.getStreamStatus);
function stopStreaming(sessionId?: string) {
const targetSession = sessionId || currentSessionIdRef.current;
if (targetSession) {
stopStream(targetSession);
unregisterActiveSession(targetSession);
}
setIsStreaming(false);
}
useEffect(() => {
console.log("[useChatStream] Component mounted");
return () => {
const sessionIdAtUnmount = currentSessionIdRef.current;
console.log(
"[useChatStream] Component unmounting, calling stopStreaming",
{
sessionIdAtUnmount,
},
);
stopStreaming(undefined, false);
return function cleanup() {
const sessionId = currentSessionIdRef.current;
if (sessionId && !isSessionActive(sessionId)) {
stopStream(sessionId);
}
currentSessionIdRef.current = null;
onChunkCallbackRef.current = null;
};
}, [stopStreaming]);
}, []);
const sendMessage = useCallback(
async (
sessionId: string,
message: string,
onChunk: (chunk: StreamChunk) => void,
isUserMessage: boolean = true,
context?: { url: string; content: string },
isRetry: boolean = false,
) => {
console.log("[useChatStream] sendMessage called", {
sessionId,
message: message.substring(0, 50),
isUserMessage,
isRetry,
stack: new Error().stack,
});
useEffect(() => {
const unsubscribe = onStreamComplete(
function handleStreamComplete(completedSessionId) {
if (completedSessionId !== currentSessionIdRef.current) return;
const previousSessionId = currentSessionIdRef.current;
stopStreaming(sessionId, true);
currentSessionIdRef.current = sessionId;
const abortController = new AbortController();
abortControllerRef.current = abortController;
requestStartTimeRef.current = Date.now();
console.log("[useChatStream] Created new AbortController", {
sessionId,
previousSessionId,
requestStartTime: requestStartTimeRef.current,
});
if (abortController.signal.aborted) {
console.warn(
"[useChatStream] AbortController was aborted before request started",
);
requestStartTimeRef.current = null;
return Promise.reject(new Error("Request aborted"));
}
if (!isRetry) {
retryCountRef.current = 0;
}
setIsStreaming(true);
setError(null);
try {
const url = `/api/chat/sessions/${sessionId}/stream`;
const body = JSON.stringify({
message,
is_user_message: isUserMessage,
context: context || null,
});
const response = await fetch(url, {
method: "POST",
headers: {
"Content-Type": "application/json",
Accept: "text/event-stream",
},
body,
signal: abortController.signal,
});
console.info("[useChatStream] Stream response", {
sessionId,
status: response.status,
ok: response.ok,
contentType: response.headers.get("content-type"),
});
if (!response.ok) {
const errorText = await response.text();
console.warn("[useChatStream] Stream response error", {
sessionId,
status: response.status,
errorText,
});
throw new Error(errorText || `HTTP ${response.status}`);
}
if (!response.body) {
console.warn("[useChatStream] Response body is null", { sessionId });
throw new Error("Response body is null");
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let receivedChunkCount = 0;
let firstChunkAt: number | null = null;
let loggedLineCount = 0;
return new Promise<void>((resolve, reject) => {
let didDispatchStreamEnd = false;
function dispatchStreamEnd() {
if (didDispatchStreamEnd) return;
didDispatchStreamEnd = true;
onChunk({ type: "stream_end" });
}
const cleanup = () => {
reader.cancel().catch(() => {
// Ignore cancel errors
});
};
async function readStream() {
try {
while (true) {
const { done, value } = await reader.read();
if (done) {
cleanup();
console.info("[useChatStream] Stream closed", {
sessionId,
receivedChunkCount,
timeSinceStart: requestStartTimeRef.current
? Date.now() - requestStartTimeRef.current
: null,
});
dispatchStreamEnd();
retryCountRef.current = 0;
stopStreaming();
resolve();
return;
}
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() || "";
for (const line of lines) {
if (line.startsWith("data: ")) {
const data = line.slice(6);
if (loggedLineCount < 3) {
console.info("[useChatStream] Raw stream line", {
sessionId,
data:
data.length > 300 ? `${data.slice(0, 300)}...` : data,
});
loggedLineCount += 1;
}
if (data === "[DONE]") {
cleanup();
console.info("[useChatStream] Stream done marker", {
sessionId,
receivedChunkCount,
timeSinceStart: requestStartTimeRef.current
? Date.now() - requestStartTimeRef.current
: null,
});
dispatchStreamEnd();
retryCountRef.current = 0;
stopStreaming();
resolve();
return;
}
try {
const rawChunk = JSON.parse(data) as
| StreamChunk
| VercelStreamChunk;
const chunk = normalizeStreamChunk(rawChunk);
if (!chunk) {
continue;
}
if (!firstChunkAt) {
firstChunkAt = Date.now();
console.info("[useChatStream] First stream chunk", {
sessionId,
chunkType: chunk.type,
timeSinceStart: requestStartTimeRef.current
? firstChunkAt - requestStartTimeRef.current
: null,
});
}
receivedChunkCount += 1;
// Call the chunk handler
onChunk(chunk);
// Handle stream lifecycle
if (chunk.type === "stream_end") {
didDispatchStreamEnd = true;
cleanup();
console.info("[useChatStream] Stream end chunk", {
sessionId,
receivedChunkCount,
timeSinceStart: requestStartTimeRef.current
? Date.now() - requestStartTimeRef.current
: null,
});
retryCountRef.current = 0;
stopStreaming();
resolve();
return;
} else if (chunk.type === "error") {
cleanup();
reject(
new Error(
chunk.message || chunk.content || "Stream error",
),
);
return;
}
} catch (err) {
// Skip invalid JSON lines
console.warn("Failed to parse SSE chunk:", err, data);
}
}
}
}
} catch (err) {
if (err instanceof Error && err.name === "AbortError") {
cleanup();
dispatchStreamEnd();
stopStreaming();
resolve();
return;
}
const streamError =
err instanceof Error ? err : new Error("Failed to read stream");
if (retryCountRef.current < MAX_RETRIES) {
retryCountRef.current += 1;
const retryDelay =
INITIAL_RETRY_DELAY * Math.pow(2, retryCountRef.current - 1);
toast.info("Connection interrupted", {
description: `Retrying in ${retryDelay / 1000} seconds...`,
});
retryTimeoutRef.current = setTimeout(() => {
sendMessage(
sessionId,
message,
onChunk,
isUserMessage,
context,
true,
).catch((_err) => {
// Retry failed
});
}, retryDelay);
} else {
setError(streamError);
toast.error("Connection Failed", {
description:
"Unable to connect to chat service. Please try again.",
});
cleanup();
dispatchStreamEnd();
retryCountRef.current = 0;
stopStreaming();
reject(streamError);
}
}
}
readStream();
});
} catch (err) {
if (err instanceof Error && err.name === "AbortError") {
setIsStreaming(false);
return Promise.resolve();
}
const streamError =
err instanceof Error ? err : new Error("Failed to start stream");
setError(streamError);
setIsStreaming(false);
throw streamError;
const completed = getCompletedStream(completedSessionId);
if (completed?.error) {
setError(completed.error);
}
unregisterActiveSession(completedSessionId);
},
);
return unsubscribe;
}, []);
async function sendMessage(
sessionId: string,
message: string,
onChunk: (chunk: StreamChunk) => void,
isUserMessage: boolean = true,
context?: { url: string; content: string },
) {
const previousSessionId = currentSessionIdRef.current;
if (previousSessionId && previousSessionId !== sessionId) {
stopStreaming(previousSessionId);
}
currentSessionIdRef.current = sessionId;
onChunkCallbackRef.current = onChunk;
setIsStreaming(true);
setError(null);
registerActiveSession(sessionId);
try {
await startStream(sessionId, message, isUserMessage, context, onChunk);
const status = getStreamStatus(sessionId);
if (status === "error") {
const completed = getCompletedStream(sessionId);
if (completed?.error) {
setError(completed.error);
toast.error("Connection Failed", {
description: "Unable to connect to chat service. Please try again.",
});
throw completed.error;
}
}
},
[stopStreaming],
);
} catch (err) {
const streamError =
err instanceof Error ? err : new Error("Failed to start stream");
setError(streamError);
throw streamError;
} finally {
setIsStreaming(false);
}
}
return {
isStreaming,

7
autogpt_platform/frontend/src/providers/posthog/posthog-provider.tsx
View File

@@ -9,11 +9,12 @@ import { ReactNode, useEffect, useRef } from "react";
export function PostHogProvider({ children }: { children: ReactNode }) {
const isPostHogEnabled = environment.isPostHogEnabled();
const postHogCredentials = environment.getPostHogCredentials();
useEffect(() => {
if (process.env.NEXT_PUBLIC_POSTHOG_KEY) {
posthog.init(process.env.NEXT_PUBLIC_POSTHOG_KEY, {
api_host: process.env.NEXT_PUBLIC_POSTHOG_HOST,
if (postHogCredentials.key) {
posthog.init(postHogCredentials.key, {
api_host: postHogCredentials.host,
defaults: "2025-11-30",
capture_pageview: false,
capture_pageleave: true,

8
autogpt_platform/frontend/src/services/network-status/NetworkStatusMonitor.tsx Normal file
View File

@@ -0,0 +1,8 @@
"use client";
import { useNetworkStatus } from "./useNetworkStatus";
export function NetworkStatusMonitor() {
useNetworkStatus();
return null;
}

28
autogpt_platform/frontend/src/services/network-status/useNetworkStatus.ts Normal file
View File

@@ -0,0 +1,28 @@
"use client";
import { useEffect } from "react";
import { toast } from "sonner";
export function useNetworkStatus() {
useEffect(function monitorNetworkStatus() {
function handleOnline() {
toast.success("Connection restored", {
description: "You're back online",
});
}
function handleOffline() {
toast.error("You're offline", {
description: "Check your internet connection",
});
}
window.addEventListener("online", handleOnline);
window.addEventListener("offline", handleOffline);
return function cleanup() {
window.removeEventListener("online", handleOnline);
window.removeEventListener("offline", handleOffline);
};
}, []);
}

1
autogpt_platform/frontend/src/services/storage/session-storage.ts
View File

@@ -3,6 +3,7 @@ import { environment } from "../environment";
export enum SessionKey {
CHAT_SENT_INITIAL_PROMPTS = "chat_sent_initial_prompts",
CHAT_INITIAL_PROMPTS = "chat_initial_prompts",
}
function get(key: SessionKey) {

1
backend/blocks/video/__init__.py
View File

@@ -1 +0,0 @@
# Video editing blocks