mirror of
https://github.com/Significant-Gravitas/AutoGPT.git
synced 2026-02-24 03:00:28 -05:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ef42b17e3b | ||
|
|
a18ffd0b21 | ||
|
|
e40c8c70ce |
@@ -38,8 +38,10 @@ class ChatConfig(BaseSettings):
|
||||
|
||||
# Long-running operation configuration
|
||||
long_running_operation_ttl: int = Field(
|
||||
default=600,
|
||||
description="TTL in seconds for long-running operation tracking in Redis (safety net if pod dies)",
|
||||
default=3600,
|
||||
description="TTL in seconds for long-running operation deduplication lock "
|
||||
"(1 hour, matches stream_ttl). Prevents duplicate operations if pod dies. "
|
||||
"For longer operations, the stream_registry heartbeat keeps them alive.",
|
||||
)
|
||||
|
||||
# Stream registry configuration for SSE reconnection
|
||||
@@ -47,6 +49,11 @@ class ChatConfig(BaseSettings):
|
||||
default=3600,
|
||||
description="TTL in seconds for stream data in Redis (1 hour)",
|
||||
)
|
||||
stream_lock_ttl: int = Field(
|
||||
default=120,
|
||||
description="TTL in seconds for stream lock (2 minutes). Short timeout allows "
|
||||
"reconnection after refresh/crash without long waits.",
|
||||
)
|
||||
stream_max_length: int = Field(
|
||||
default=10000,
|
||||
description="Maximum number of messages to store per stream",
|
||||
|
||||
@@ -3,8 +3,9 @@
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, cast
|
||||
from typing import Any
|
||||
|
||||
from prisma.errors import UniqueViolationError
|
||||
from prisma.models import ChatMessage as PrismaChatMessage
|
||||
from prisma.models import ChatSession as PrismaChatSession
|
||||
from prisma.types import (
|
||||
@@ -92,10 +93,9 @@ async def add_chat_message(
|
||||
function_call: dict[str, Any] | None = None,
|
||||
) -> ChatMessage:
|
||||
"""Add a message to a chat session."""
|
||||
# Build input dict dynamically rather than using ChatMessageCreateInput directly
|
||||
# because Prisma's TypedDict validation rejects optional fields set to None.
|
||||
# We only include fields that have values, then cast at the end.
|
||||
data: dict[str, Any] = {
|
||||
# Build ChatMessageCreateInput with only non-None values
|
||||
# (Prisma TypedDict rejects optional fields set to None)
|
||||
data: ChatMessageCreateInput = {
|
||||
"Session": {"connect": {"id": session_id}},
|
||||
"role": role,
|
||||
"sequence": sequence,
|
||||
@@ -123,7 +123,7 @@ async def add_chat_message(
|
||||
where={"id": session_id},
|
||||
data={"updatedAt": datetime.now(UTC)},
|
||||
),
|
||||
PrismaChatMessage.prisma().create(data=cast(ChatMessageCreateInput, data)),
|
||||
PrismaChatMessage.prisma().create(data=data),
|
||||
)
|
||||
return ChatMessage.from_db(message)
|
||||
|
||||
@@ -132,58 +132,93 @@ async def add_chat_messages_batch(
|
||||
session_id: str,
|
||||
messages: list[dict[str, Any]],
|
||||
start_sequence: int,
|
||||
) -> list[ChatMessage]:
|
||||
) -> int:
|
||||
"""Add multiple messages to a chat session in a batch.
|
||||
|
||||
Uses a transaction for atomicity - if any message creation fails,
|
||||
the entire batch is rolled back.
|
||||
Uses collision detection with retry: tries to create messages starting
|
||||
at start_sequence. If a unique constraint violation occurs (e.g., the
|
||||
streaming loop and long-running callback race), queries the latest
|
||||
sequence and retries with the correct offset. This avoids unnecessary
|
||||
upserts and DB queries in the common case (no collision).
|
||||
|
||||
Returns:
|
||||
Next sequence number for the next message to be inserted. This equals
|
||||
start_sequence + len(messages) and allows callers to update their
|
||||
counters even when collision detection adjusts start_sequence.
|
||||
"""
|
||||
if not messages:
|
||||
return []
|
||||
# No messages to add - return current count
|
||||
return start_sequence
|
||||
|
||||
created_messages = []
|
||||
max_retries = 5
|
||||
for attempt in range(max_retries):
|
||||
try:
|
||||
# Single timestamp for all messages and session update
|
||||
now = datetime.now(UTC)
|
||||
|
||||
async with db.transaction() as tx:
|
||||
for i, msg in enumerate(messages):
|
||||
# Build input dict dynamically rather than using ChatMessageCreateInput
|
||||
# directly because Prisma's TypedDict validation rejects optional fields
|
||||
# set to None. We only include fields that have values, then cast.
|
||||
data: dict[str, Any] = {
|
||||
"Session": {"connect": {"id": session_id}},
|
||||
"role": msg["role"],
|
||||
"sequence": start_sequence + i,
|
||||
}
|
||||
async with db.transaction() as tx:
|
||||
# Build all message data
|
||||
messages_data = []
|
||||
for i, msg in enumerate(messages):
|
||||
# Build ChatMessageCreateInput with only non-None values
|
||||
# (Prisma TypedDict rejects optional fields set to None)
|
||||
# Note: create_many doesn't support nested creates, use sessionId directly
|
||||
data: ChatMessageCreateInput = {
|
||||
"sessionId": session_id,
|
||||
"role": msg["role"],
|
||||
"sequence": start_sequence + i,
|
||||
"createdAt": now,
|
||||
}
|
||||
|
||||
# Add optional string fields
|
||||
if msg.get("content") is not None:
|
||||
data["content"] = msg["content"]
|
||||
if msg.get("name") is not None:
|
||||
data["name"] = msg["name"]
|
||||
if msg.get("tool_call_id") is not None:
|
||||
data["toolCallId"] = msg["tool_call_id"]
|
||||
if msg.get("refusal") is not None:
|
||||
data["refusal"] = msg["refusal"]
|
||||
# Add optional string fields
|
||||
if msg.get("content") is not None:
|
||||
data["content"] = msg["content"]
|
||||
if msg.get("name") is not None:
|
||||
data["name"] = msg["name"]
|
||||
if msg.get("tool_call_id") is not None:
|
||||
data["toolCallId"] = msg["tool_call_id"]
|
||||
if msg.get("refusal") is not None:
|
||||
data["refusal"] = msg["refusal"]
|
||||
|
||||
# Add optional JSON fields only when they have values
|
||||
if msg.get("tool_calls") is not None:
|
||||
data["toolCalls"] = SafeJson(msg["tool_calls"])
|
||||
if msg.get("function_call") is not None:
|
||||
data["functionCall"] = SafeJson(msg["function_call"])
|
||||
# Add optional JSON fields only when they have values
|
||||
if msg.get("tool_calls") is not None:
|
||||
data["toolCalls"] = SafeJson(msg["tool_calls"])
|
||||
if msg.get("function_call") is not None:
|
||||
data["functionCall"] = SafeJson(msg["function_call"])
|
||||
|
||||
created = await PrismaChatMessage.prisma(tx).create(
|
||||
data=cast(ChatMessageCreateInput, data)
|
||||
)
|
||||
created_messages.append(created)
|
||||
messages_data.append(data)
|
||||
|
||||
# Update session's updatedAt timestamp within the same transaction.
|
||||
# Note: Token usage (total_prompt_tokens, total_completion_tokens) is updated
|
||||
# separately via update_chat_session() after streaming completes.
|
||||
await PrismaChatSession.prisma(tx).update(
|
||||
where={"id": session_id},
|
||||
data={"updatedAt": datetime.now(UTC)},
|
||||
)
|
||||
# Run create_many and session update in parallel within transaction
|
||||
# Both use the same timestamp for consistency
|
||||
await asyncio.gather(
|
||||
PrismaChatMessage.prisma(tx).create_many(data=messages_data),
|
||||
PrismaChatSession.prisma(tx).update(
|
||||
where={"id": session_id},
|
||||
data={"updatedAt": now},
|
||||
),
|
||||
)
|
||||
|
||||
return [ChatMessage.from_db(m) for m in created_messages]
|
||||
# Return next sequence number for counter sync
|
||||
return start_sequence + len(messages)
|
||||
|
||||
except UniqueViolationError:
|
||||
if attempt < max_retries - 1:
|
||||
# Collision detected - query MAX(sequence)+1 and retry with correct offset
|
||||
logger.info(
|
||||
f"Collision detected for session {session_id} at sequence "
|
||||
f"{start_sequence}, querying DB for latest sequence"
|
||||
)
|
||||
start_sequence = await get_next_sequence(session_id)
|
||||
logger.info(
|
||||
f"Retrying batch insert with start_sequence={start_sequence}"
|
||||
)
|
||||
continue
|
||||
else:
|
||||
# Max retries exceeded - propagate error
|
||||
raise
|
||||
|
||||
# Should never reach here due to raise in exception handler
|
||||
raise RuntimeError(f"Failed to insert messages after {max_retries} attempts")
|
||||
|
||||
|
||||
async def get_user_chat_sessions(
|
||||
@@ -237,10 +272,20 @@ async def delete_chat_session(session_id: str, user_id: str | None = None) -> bo
|
||||
return False
|
||||
|
||||
|
||||
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 get_next_sequence(session_id: str) -> int:
|
||||
"""Get the next sequence number for a new message in this session.
|
||||
|
||||
Uses MAX(sequence) + 1 for robustness. Returns 0 if no messages exist.
|
||||
More robust than COUNT(*) because it's immune to deleted messages.
|
||||
|
||||
Optimized to select only the sequence column using raw SQL.
|
||||
The unique index on (sessionId, sequence) makes this query fast.
|
||||
"""
|
||||
results = await db.query_raw_with_schema(
|
||||
'SELECT "sequence" FROM {schema_prefix}"ChatMessage" WHERE "sessionId" = $1 ORDER BY "sequence" DESC LIMIT 1',
|
||||
session_id,
|
||||
)
|
||||
return 0 if not results else results[0]["sequence"] + 1
|
||||
|
||||
|
||||
async def update_tool_message_content(
|
||||
|
||||
@@ -434,8 +434,6 @@ async def _get_session_from_db(session_id: str) -> ChatSession | None:
|
||||
|
||||
async def upsert_chat_session(
|
||||
session: ChatSession,
|
||||
*,
|
||||
existing_message_count: int | None = None,
|
||||
) -> ChatSession:
|
||||
"""Update a chat session in both cache and database.
|
||||
|
||||
@@ -443,12 +441,6 @@ async def upsert_chat_session(
|
||||
operations (e.g., background title update and main stream handler)
|
||||
attempt to upsert the same session simultaneously.
|
||||
|
||||
Args:
|
||||
existing_message_count: If provided, skip the DB query to count
|
||||
existing messages. The caller is responsible for tracking this
|
||||
accurately. Useful for incremental saves in a streaming loop
|
||||
where the caller already knows how many messages are persisted.
|
||||
|
||||
Raises:
|
||||
DatabaseError: If the database write fails. The cache is still updated
|
||||
as a best-effort optimization, but the error is propagated to ensure
|
||||
@@ -459,11 +451,8 @@ async def upsert_chat_session(
|
||||
lock = await _get_session_lock(session.session_id)
|
||||
|
||||
async with lock:
|
||||
# Get existing message count from DB for incremental saves
|
||||
if existing_message_count is None:
|
||||
existing_message_count = await chat_db().get_chat_session_message_count(
|
||||
session.session_id
|
||||
)
|
||||
# Always query DB for existing message count to ensure consistency
|
||||
existing_message_count = await chat_db().get_next_sequence(session.session_id)
|
||||
|
||||
db_error: Exception | None = None
|
||||
|
||||
@@ -587,9 +576,7 @@ async def append_and_save_message(session_id: str, message: ChatMessage) -> Chat
|
||||
raise ValueError(f"Session {session_id} not found")
|
||||
|
||||
session.messages.append(message)
|
||||
existing_message_count = await chat_db().get_chat_session_message_count(
|
||||
session_id
|
||||
)
|
||||
existing_message_count = await chat_db().get_next_sequence(session_id)
|
||||
|
||||
try:
|
||||
await _save_session_to_db(session, existing_message_count)
|
||||
|
||||
@@ -331,3 +331,96 @@ def test_to_openai_messages_merges_split_assistants():
|
||||
tc_list = merged.get("tool_calls")
|
||||
assert tc_list is not None and len(list(tc_list)) == 1
|
||||
assert list(tc_list)[0]["id"] == "tc1"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Concurrent save collision detection #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
async def test_concurrent_saves_collision_detection(setup_test_user, test_user_id):
|
||||
"""Test that concurrent saves from streaming loop and callback handle collisions correctly.
|
||||
|
||||
Simulates the race condition where:
|
||||
1. Streaming loop starts with saved_msg_count=5
|
||||
2. Long-running callback appends message #5 and saves
|
||||
3. Streaming loop tries to save with stale count=5
|
||||
|
||||
The collision detection should handle this gracefully.
|
||||
"""
|
||||
import asyncio
|
||||
|
||||
# Create a session with initial messages
|
||||
session = ChatSession.new(user_id=test_user_id)
|
||||
for i in range(3):
|
||||
session.messages.append(
|
||||
ChatMessage(
|
||||
role="user" if i % 2 == 0 else "assistant", content=f"Message {i}"
|
||||
)
|
||||
)
|
||||
|
||||
# Save initial messages
|
||||
session = await upsert_chat_session(session)
|
||||
|
||||
# Simulate streaming loop and callback saving concurrently
|
||||
async def streaming_loop_save():
|
||||
"""Simulates streaming loop saving messages."""
|
||||
# Add 2 messages
|
||||
session.messages.append(ChatMessage(role="user", content="Streaming message 1"))
|
||||
session.messages.append(
|
||||
ChatMessage(role="assistant", content="Streaming message 2")
|
||||
)
|
||||
|
||||
# Wait a bit to let callback potentially save first
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
# Save (will query DB for existing count)
|
||||
return await upsert_chat_session(session)
|
||||
|
||||
async def callback_save():
|
||||
"""Simulates long-running callback saving a message."""
|
||||
# Add 1 message
|
||||
session.messages.append(
|
||||
ChatMessage(role="tool", content="Callback result", tool_call_id="tc1")
|
||||
)
|
||||
|
||||
# Save immediately (will query DB for existing count)
|
||||
return await upsert_chat_session(session)
|
||||
|
||||
# Run both saves concurrently - one will hit collision detection
|
||||
results = await asyncio.gather(streaming_loop_save(), callback_save())
|
||||
|
||||
# Both should succeed
|
||||
assert all(r is not None for r in results)
|
||||
|
||||
# Reload session from DB to verify
|
||||
from backend.data.redis_client import get_redis_async
|
||||
|
||||
redis_key = f"chat:session:{session.session_id}"
|
||||
async_redis = await get_redis_async()
|
||||
await async_redis.delete(redis_key) # Clear cache to force DB load
|
||||
|
||||
loaded_session = await get_chat_session(session.session_id, test_user_id)
|
||||
assert loaded_session is not None
|
||||
|
||||
# Should have all 6 messages (3 initial + 2 streaming + 1 callback)
|
||||
assert len(loaded_session.messages) == 6
|
||||
|
||||
# Verify no duplicate sequences
|
||||
sequences = []
|
||||
for i, msg in enumerate(loaded_session.messages):
|
||||
# Messages should have sequential sequence numbers starting from 0
|
||||
sequences.append(i)
|
||||
|
||||
# All sequences should be unique and sequential
|
||||
assert sequences == list(range(6))
|
||||
|
||||
# Verify message content is preserved
|
||||
contents = [m.content for m in loaded_session.messages]
|
||||
assert "Message 0" in contents
|
||||
assert "Message 1" in contents
|
||||
assert "Message 2" in contents
|
||||
assert "Streaming message 1" in contents
|
||||
assert "Streaming message 2" in contents
|
||||
assert "Callback result" in contents
|
||||
|
||||
@@ -7,8 +7,10 @@ import os
|
||||
import uuid
|
||||
from collections.abc import AsyncGenerator
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from typing import Any, cast
|
||||
|
||||
from backend.data.redis_client import get_redis_async
|
||||
from backend.executor.cluster_lock import AsyncClusterLock
|
||||
from backend.util.exceptions import NotFoundError
|
||||
|
||||
from .. import stream_registry
|
||||
@@ -61,6 +63,7 @@ from .transcript import (
|
||||
logger = logging.getLogger(__name__)
|
||||
config = ChatConfig()
|
||||
|
||||
|
||||
# Set to hold background tasks to prevent garbage collection
|
||||
_background_tasks: set[asyncio.Task[Any]] = set()
|
||||
|
||||
@@ -132,8 +135,12 @@ is delivered to the user via a background stream.
|
||||
All tasks must run in the foreground.
|
||||
"""
|
||||
|
||||
STREAM_LOCK_PREFIX = "copilot:stream:lock:"
|
||||
|
||||
def _build_long_running_callback(user_id: str | None) -> LongRunningCallback:
|
||||
|
||||
def _build_long_running_callback(
|
||||
user_id: str | None,
|
||||
) -> LongRunningCallback:
|
||||
"""Build a callback that delegates long-running tools to the non-SDK infrastructure.
|
||||
|
||||
Long-running tools (create_agent, edit_agent, etc.) are delegated to the
|
||||
@@ -142,6 +149,9 @@ def _build_long_running_callback(user_id: str | None) -> LongRunningCallback:
|
||||
page refreshes / pod restarts, and the frontend shows the proper loading
|
||||
widget with progress updates.
|
||||
|
||||
Args:
|
||||
user_id: User ID for the session
|
||||
|
||||
The returned callback matches the ``LongRunningCallback`` signature:
|
||||
``(tool_name, args, session) -> MCP response dict``.
|
||||
"""
|
||||
@@ -207,7 +217,8 @@ def _build_long_running_callback(user_id: str | None) -> LongRunningCallback:
|
||||
tool_call_id=tool_call_id,
|
||||
)
|
||||
session.messages.append(pending_message)
|
||||
await upsert_chat_session(session)
|
||||
# Collision detection happens in add_chat_messages_batch (db.py)
|
||||
session = await upsert_chat_session(session)
|
||||
|
||||
# --- Spawn background task (reuses non-SDK infrastructure) ---
|
||||
bg_task = asyncio.create_task(
|
||||
@@ -527,6 +538,9 @@ async def stream_chat_completion_sdk(
|
||||
f"Session {session_id} not found. Please create a new session first."
|
||||
)
|
||||
|
||||
# Type narrowing: session is guaranteed ChatSession after the check above
|
||||
session = cast(ChatSession, session)
|
||||
|
||||
# Append the new message to the session if it's not already there
|
||||
new_message_role = "user" if is_user_message else "assistant"
|
||||
if message and (
|
||||
@@ -564,6 +578,29 @@ async def stream_chat_completion_sdk(
|
||||
system_prompt += _SDK_TOOL_SUPPLEMENT
|
||||
message_id = str(uuid.uuid4())
|
||||
task_id = str(uuid.uuid4())
|
||||
stream_id = task_id # Use task_id as unique stream identifier
|
||||
|
||||
# Acquire stream lock to prevent concurrent streams to the same session
|
||||
lock = AsyncClusterLock(
|
||||
redis=await get_redis_async(),
|
||||
key=f"{STREAM_LOCK_PREFIX}{session_id}",
|
||||
owner_id=stream_id,
|
||||
timeout=config.stream_lock_ttl,
|
||||
)
|
||||
|
||||
lock_owner = await lock.try_acquire()
|
||||
if lock_owner != stream_id:
|
||||
# Another stream is active
|
||||
logger.warning(
|
||||
f"[SDK] Session {session_id} already has an active stream: {lock_owner}"
|
||||
)
|
||||
yield StreamError(
|
||||
errorText="Another stream is already active for this session. "
|
||||
"Please wait or stop it.",
|
||||
code="stream_already_active",
|
||||
)
|
||||
yield StreamFinish()
|
||||
return
|
||||
|
||||
yield StreamStart(messageId=message_id, taskId=task_id)
|
||||
|
||||
@@ -715,9 +752,6 @@ async def stream_chat_completion_sdk(
|
||||
accumulated_tool_calls: list[dict[str, Any]] = []
|
||||
has_appended_assistant = False
|
||||
has_tool_results = False
|
||||
# Track persisted message count to skip DB count queries
|
||||
# on incremental saves. Initial save happened at line 545.
|
||||
saved_msg_count = len(session.messages)
|
||||
|
||||
# Use an explicit async iterator with non-cancelling heartbeats.
|
||||
# CRITICAL: we must NOT cancel __anext__() mid-flight — doing so
|
||||
@@ -744,6 +778,8 @@ async def stream_chat_completion_sdk(
|
||||
|
||||
if not done:
|
||||
# Timeout — emit heartbeat but keep the task alive
|
||||
# Also refresh lock TTL to keep it alive
|
||||
await lock.refresh()
|
||||
yield StreamHeartbeat()
|
||||
continue
|
||||
|
||||
@@ -893,13 +929,10 @@ async def stream_chat_completion_sdk(
|
||||
has_appended_assistant = True
|
||||
# Save before tool execution starts so the
|
||||
# pending tool call is visible on refresh /
|
||||
# other devices.
|
||||
# other devices. Collision detection happens
|
||||
# in add_chat_messages_batch (db.py).
|
||||
try:
|
||||
await upsert_chat_session(
|
||||
session,
|
||||
existing_message_count=saved_msg_count,
|
||||
)
|
||||
saved_msg_count = len(session.messages)
|
||||
session = await upsert_chat_session(session)
|
||||
except Exception as save_err:
|
||||
logger.warning(
|
||||
"[SDK] [%s] Incremental save " "failed: %s",
|
||||
@@ -922,12 +955,9 @@ async def stream_chat_completion_sdk(
|
||||
has_tool_results = True
|
||||
# Save after tool completes so the result is
|
||||
# visible on refresh / other devices.
|
||||
# Collision detection happens in add_chat_messages_batch (db.py).
|
||||
try:
|
||||
await upsert_chat_session(
|
||||
session,
|
||||
existing_message_count=saved_msg_count,
|
||||
)
|
||||
saved_msg_count = len(session.messages)
|
||||
session = await upsert_chat_session(session)
|
||||
except Exception as save_err:
|
||||
logger.warning(
|
||||
"[SDK] [%s] Incremental save " "failed: %s",
|
||||
@@ -1059,7 +1089,7 @@ async def stream_chat_completion_sdk(
|
||||
"to use the OpenAI-compatible fallback."
|
||||
)
|
||||
|
||||
await asyncio.shield(upsert_chat_session(session))
|
||||
session = cast(ChatSession, await asyncio.shield(upsert_chat_session(session)))
|
||||
logger.info(
|
||||
"[SDK] [%s] Session saved with %d messages",
|
||||
session_id[:12],
|
||||
@@ -1076,10 +1106,11 @@ async def stream_chat_completion_sdk(
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"[SDK] Error: {e}", exc_info=True)
|
||||
try:
|
||||
await asyncio.shield(upsert_chat_session(session))
|
||||
except Exception as save_err:
|
||||
logger.error(f"[SDK] Failed to save session on error: {save_err}")
|
||||
if session:
|
||||
try:
|
||||
await asyncio.shield(upsert_chat_session(session))
|
||||
except Exception as save_err:
|
||||
logger.error(f"[SDK] Failed to save session on error: {save_err}")
|
||||
yield StreamError(
|
||||
errorText="An error occurred. Please try again.",
|
||||
code="sdk_error",
|
||||
@@ -1101,7 +1132,7 @@ async def stream_chat_completion_sdk(
|
||||
if not raw_transcript and use_resume and resume_file:
|
||||
raw_transcript = read_transcript_file(resume_file)
|
||||
|
||||
if raw_transcript:
|
||||
if raw_transcript and session is not None:
|
||||
await asyncio.shield(
|
||||
_try_upload_transcript(
|
||||
user_id,
|
||||
@@ -1121,6 +1152,9 @@ async def stream_chat_completion_sdk(
|
||||
if sdk_cwd:
|
||||
_cleanup_sdk_tool_results(sdk_cwd)
|
||||
|
||||
# Release stream lock to allow new streams for this session
|
||||
await lock.release()
|
||||
|
||||
|
||||
async def _try_upload_transcript(
|
||||
user_id: str,
|
||||
|
||||
@@ -352,7 +352,8 @@ async def assign_user_to_session(
|
||||
if not session:
|
||||
raise NotFoundError(f"Session {session_id} not found")
|
||||
session.user_id = user_id
|
||||
return await upsert_chat_session(session)
|
||||
session = await upsert_chat_session(session)
|
||||
return session
|
||||
|
||||
|
||||
async def stream_chat_completion(
|
||||
|
||||
@@ -303,7 +303,7 @@ class DatabaseManager(AppService):
|
||||
get_user_chat_sessions = _(chat_db.get_user_chat_sessions)
|
||||
get_user_session_count = _(chat_db.get_user_session_count)
|
||||
delete_chat_session = _(chat_db.delete_chat_session)
|
||||
get_chat_session_message_count = _(chat_db.get_chat_session_message_count)
|
||||
get_next_sequence = _(chat_db.get_next_sequence)
|
||||
update_tool_message_content = _(chat_db.update_tool_message_content)
|
||||
|
||||
|
||||
@@ -473,5 +473,5 @@ class DatabaseManagerAsyncClient(AppServiceClient):
|
||||
get_user_chat_sessions = d.get_user_chat_sessions
|
||||
get_user_session_count = d.get_user_session_count
|
||||
delete_chat_session = d.delete_chat_session
|
||||
get_chat_session_message_count = d.get_chat_session_message_count
|
||||
get_next_sequence = d.get_next_sequence
|
||||
update_tool_message_content = d.update_tool_message_content
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Redis-based distributed locking for cluster coordination."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
@@ -7,6 +8,7 @@ from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from redis import Redis
|
||||
from redis.asyncio import Redis as AsyncRedis
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -126,3 +128,124 @@ class ClusterLock:
|
||||
|
||||
with self._refresh_lock:
|
||||
self._last_refresh = 0.0
|
||||
|
||||
|
||||
class AsyncClusterLock:
|
||||
"""Async Redis-based distributed lock for preventing duplicate execution."""
|
||||
|
||||
def __init__(
|
||||
self, redis: "AsyncRedis", key: str, owner_id: str, timeout: int = 300
|
||||
):
|
||||
self.redis = redis
|
||||
self.key = key
|
||||
self.owner_id = owner_id
|
||||
self.timeout = timeout
|
||||
self._last_refresh = 0.0
|
||||
self._refresh_lock = asyncio.Lock()
|
||||
|
||||
async def try_acquire(self) -> str | None:
|
||||
"""Try to acquire the lock.
|
||||
|
||||
Returns:
|
||||
- owner_id (self.owner_id) if successfully acquired
|
||||
- different owner_id if someone else holds the lock
|
||||
- None if Redis is unavailable or other error
|
||||
"""
|
||||
try:
|
||||
success = await self.redis.set(
|
||||
self.key, self.owner_id, nx=True, ex=self.timeout
|
||||
)
|
||||
if success:
|
||||
async with self._refresh_lock:
|
||||
self._last_refresh = time.time()
|
||||
return self.owner_id # Successfully acquired
|
||||
|
||||
# Failed to acquire, get current owner
|
||||
current_value = await self.redis.get(self.key)
|
||||
if current_value:
|
||||
current_owner = (
|
||||
current_value.decode("utf-8")
|
||||
if isinstance(current_value, bytes)
|
||||
else str(current_value)
|
||||
)
|
||||
return current_owner
|
||||
|
||||
# Key doesn't exist but we failed to set it - race condition or Redis issue
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"AsyncClusterLock.try_acquire failed for key {self.key}: {e}")
|
||||
return None
|
||||
|
||||
async def refresh(self) -> bool:
|
||||
"""Refresh lock TTL if we still own it.
|
||||
|
||||
Rate limited to at most once every timeout/10 seconds (minimum 1 second).
|
||||
During rate limiting, still verifies lock existence but skips TTL extension.
|
||||
Setting _last_refresh to 0 bypasses rate limiting for testing.
|
||||
|
||||
Async-safe: uses asyncio.Lock to protect _last_refresh access.
|
||||
"""
|
||||
# Calculate refresh interval: max(timeout // 10, 1)
|
||||
refresh_interval = max(self.timeout // 10, 1)
|
||||
current_time = time.time()
|
||||
|
||||
# Check if we're within the rate limit period (async-safe read)
|
||||
# _last_refresh == 0 forces a refresh (bypasses rate limiting for testing)
|
||||
async with self._refresh_lock:
|
||||
last_refresh = self._last_refresh
|
||||
is_rate_limited = (
|
||||
last_refresh > 0 and (current_time - last_refresh) < refresh_interval
|
||||
)
|
||||
|
||||
try:
|
||||
# Always verify lock existence, even during rate limiting
|
||||
current_value = await self.redis.get(self.key)
|
||||
if not current_value:
|
||||
async with self._refresh_lock:
|
||||
self._last_refresh = 0
|
||||
return False
|
||||
|
||||
stored_owner = (
|
||||
current_value.decode("utf-8")
|
||||
if isinstance(current_value, bytes)
|
||||
else str(current_value)
|
||||
)
|
||||
if stored_owner != self.owner_id:
|
||||
async with self._refresh_lock:
|
||||
self._last_refresh = 0
|
||||
return False
|
||||
|
||||
# If rate limited, return True but don't update TTL or timestamp
|
||||
if is_rate_limited:
|
||||
return True
|
||||
|
||||
# Perform actual refresh
|
||||
if await self.redis.expire(self.key, self.timeout):
|
||||
async with self._refresh_lock:
|
||||
self._last_refresh = current_time
|
||||
return True
|
||||
|
||||
async with self._refresh_lock:
|
||||
self._last_refresh = 0
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"AsyncClusterLock.refresh failed for key {self.key}: {e}")
|
||||
async with self._refresh_lock:
|
||||
self._last_refresh = 0
|
||||
return False
|
||||
|
||||
async def release(self):
|
||||
"""Release the lock."""
|
||||
async with self._refresh_lock:
|
||||
if self._last_refresh == 0:
|
||||
return
|
||||
|
||||
try:
|
||||
await self.redis.delete(self.key)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
async with self._refresh_lock:
|
||||
self._last_refresh = 0.0
|
||||
|
||||
|
Before Width: | Height: | Size: 8.0 KiB After Width: | Height: | Size: 8.0 KiB |
@@ -11,6 +11,11 @@ import {
|
||||
MessageResponse,
|
||||
} from "@/components/ai-elements/message";
|
||||
import { Text } from "@/components/atoms/Text/Text";
|
||||
import {
|
||||
CredentialsProvidersContext,
|
||||
type CredentialsProviderData,
|
||||
type CredentialsProvidersContextType,
|
||||
} from "@/providers/agent-credentials/credentials-provider";
|
||||
import { CopilotChatActionsProvider } from "../components/CopilotChatActionsProvider/CopilotChatActionsProvider";
|
||||
import { CreateAgentTool } from "../tools/CreateAgent/CreateAgent";
|
||||
import { EditAgentTool } from "../tools/EditAgent/EditAgent";
|
||||
@@ -97,6 +102,65 @@ function uid() {
|
||||
return `sg-${++_id}`;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Mock credential providers for setup-requirements demos
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const noop = () => Promise.reject(new Error("Styleguide mock"));
|
||||
|
||||
function makeMockProvider(
|
||||
provider: string,
|
||||
providerName: string,
|
||||
savedCredentials: CredentialsProviderData["savedCredentials"] = [],
|
||||
): CredentialsProviderData {
|
||||
return {
|
||||
provider,
|
||||
providerName,
|
||||
savedCredentials,
|
||||
isSystemProvider: false,
|
||||
oAuthCallback: noop as CredentialsProviderData["oAuthCallback"],
|
||||
mcpOAuthCallback: noop as CredentialsProviderData["mcpOAuthCallback"],
|
||||
createAPIKeyCredentials:
|
||||
noop as CredentialsProviderData["createAPIKeyCredentials"],
|
||||
createUserPasswordCredentials:
|
||||
noop as CredentialsProviderData["createUserPasswordCredentials"],
|
||||
createHostScopedCredentials:
|
||||
noop as CredentialsProviderData["createHostScopedCredentials"],
|
||||
deleteCredentials: noop as CredentialsProviderData["deleteCredentials"],
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Provider context where the user already has saved credentials
|
||||
* so the credential picker shows a selection list.
|
||||
*/
|
||||
const MOCK_PROVIDERS_WITH_CREDENTIALS: CredentialsProvidersContextType = {
|
||||
google: makeMockProvider("google", "Google", [
|
||||
{
|
||||
id: "cred-google-1",
|
||||
provider: "google",
|
||||
type: "oauth2",
|
||||
title: "work@company.com",
|
||||
scopes: ["email", "calendar"],
|
||||
},
|
||||
{
|
||||
id: "cred-google-2",
|
||||
provider: "google",
|
||||
type: "oauth2",
|
||||
title: "personal@gmail.com",
|
||||
scopes: ["email", "calendar"],
|
||||
},
|
||||
]),
|
||||
};
|
||||
|
||||
/**
|
||||
* Provider context where the user has NO saved credentials,
|
||||
* so the credential picker shows an "add new" flow.
|
||||
*/
|
||||
const MOCK_PROVIDERS_WITHOUT_CREDENTIALS: CredentialsProvidersContextType = {
|
||||
openweathermap: makeMockProvider("openweathermap", "OpenWeatherMap"),
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Page
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -554,45 +618,80 @@ export default function StyleguidePage() {
|
||||
/>
|
||||
</SubSection>
|
||||
|
||||
<SubSection label="Output available (setup requirements)">
|
||||
<RunBlockTool
|
||||
part={{
|
||||
type: "tool-run_block",
|
||||
toolCallId: uid(),
|
||||
state: "output-available",
|
||||
input: { block_id: "weather-block-123" },
|
||||
output: {
|
||||
type: ResponseType.setup_requirements,
|
||||
message:
|
||||
"This block requires API credentials to run. Please configure them below.",
|
||||
setup_info: {
|
||||
agent_name: "Weather Agent",
|
||||
requirements: {
|
||||
inputs: [
|
||||
{
|
||||
name: "city",
|
||||
title: "City",
|
||||
type: "string",
|
||||
required: true,
|
||||
description: "The city to get weather for",
|
||||
},
|
||||
],
|
||||
},
|
||||
user_readiness: {
|
||||
missing_credentials: {
|
||||
openweathermap: {
|
||||
provider: "openweathermap",
|
||||
credentials_type: "api_key",
|
||||
title: "OpenWeatherMap API Key",
|
||||
description:
|
||||
"Required to access weather data. Get your key at openweathermap.org",
|
||||
<SubSection label="Setup requirements — no credentials (add new)">
|
||||
<CredentialsProvidersContext.Provider
|
||||
value={MOCK_PROVIDERS_WITHOUT_CREDENTIALS}
|
||||
>
|
||||
<RunBlockTool
|
||||
part={{
|
||||
type: "tool-run_block",
|
||||
toolCallId: uid(),
|
||||
state: "output-available",
|
||||
input: { block_id: "weather-block-123" },
|
||||
output: {
|
||||
type: ResponseType.setup_requirements,
|
||||
message:
|
||||
"This block requires API credentials to run. Please configure them below.",
|
||||
setup_info: {
|
||||
agent_id: "agent-weather-1",
|
||||
agent_name: "Weather Agent",
|
||||
requirements: {
|
||||
inputs: [
|
||||
{
|
||||
name: "city",
|
||||
title: "City",
|
||||
type: "string",
|
||||
required: true,
|
||||
description: "The city to get weather for",
|
||||
},
|
||||
],
|
||||
},
|
||||
user_readiness: {
|
||||
missing_credentials: {
|
||||
openweathermap_key: {
|
||||
provider: "openweathermap",
|
||||
types: ["api_key"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}}
|
||||
/>
|
||||
}}
|
||||
/>
|
||||
</CredentialsProvidersContext.Provider>
|
||||
</SubSection>
|
||||
|
||||
<SubSection label="Setup requirements — has credentials (pick from list)">
|
||||
<CredentialsProvidersContext.Provider
|
||||
value={MOCK_PROVIDERS_WITH_CREDENTIALS}
|
||||
>
|
||||
<RunBlockTool
|
||||
part={{
|
||||
type: "tool-run_block",
|
||||
toolCallId: uid(),
|
||||
state: "output-available",
|
||||
input: { block_id: "calendar-block-456" },
|
||||
output: {
|
||||
type: ResponseType.setup_requirements,
|
||||
message:
|
||||
"This block requires Google credentials. Pick an account below or connect a new one.",
|
||||
setup_info: {
|
||||
agent_id: "agent-calendar-1",
|
||||
agent_name: "Calendar Agent",
|
||||
user_readiness: {
|
||||
missing_credentials: {
|
||||
google_oauth: {
|
||||
provider: "google",
|
||||
types: ["oauth2"],
|
||||
scopes: ["email", "calendar"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}}
|
||||
/>
|
||||
</CredentialsProvidersContext.Provider>
|
||||
</SubSection>
|
||||
|
||||
<SubSection label="Output available (error)">
|
||||
@@ -849,34 +948,71 @@ export default function StyleguidePage() {
|
||||
/>
|
||||
</SubSection>
|
||||
|
||||
<SubSection label="Output available (setup requirements)">
|
||||
<RunAgentTool
|
||||
part={{
|
||||
type: "tool-run_agent",
|
||||
toolCallId: uid(),
|
||||
state: "output-available",
|
||||
input: { username_agent_slug: "creator/my-agent" },
|
||||
output: {
|
||||
type: ResponseType.setup_requirements,
|
||||
message: "This agent requires additional setup.",
|
||||
setup_info: {
|
||||
agent_name: "YouTube Summarizer",
|
||||
requirements: {},
|
||||
user_readiness: {
|
||||
missing_credentials: {
|
||||
youtube_api: {
|
||||
provider: "youtube",
|
||||
credentials_type: "api_key",
|
||||
title: "YouTube Data API Key",
|
||||
description:
|
||||
"Required to access YouTube video data.",
|
||||
<SubSection label="Setup requirements — no credentials (add new)">
|
||||
<CredentialsProvidersContext.Provider
|
||||
value={MOCK_PROVIDERS_WITHOUT_CREDENTIALS}
|
||||
>
|
||||
<RunAgentTool
|
||||
part={{
|
||||
type: "tool-run_agent",
|
||||
toolCallId: uid(),
|
||||
state: "output-available",
|
||||
input: { username_agent_slug: "creator/weather-agent" },
|
||||
output: {
|
||||
type: ResponseType.setup_requirements,
|
||||
message:
|
||||
"This agent requires an API key. Add your credentials below.",
|
||||
setup_info: {
|
||||
agent_id: "agent-weather-1",
|
||||
agent_name: "Weather Agent",
|
||||
requirements: {},
|
||||
user_readiness: {
|
||||
missing_credentials: {
|
||||
openweathermap_key: {
|
||||
provider: "openweathermap",
|
||||
types: ["api_key"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}}
|
||||
/>
|
||||
}}
|
||||
/>
|
||||
</CredentialsProvidersContext.Provider>
|
||||
</SubSection>
|
||||
|
||||
<SubSection label="Setup requirements — has credentials (pick from list)">
|
||||
<CredentialsProvidersContext.Provider
|
||||
value={MOCK_PROVIDERS_WITH_CREDENTIALS}
|
||||
>
|
||||
<RunAgentTool
|
||||
part={{
|
||||
type: "tool-run_agent",
|
||||
toolCallId: uid(),
|
||||
state: "output-available",
|
||||
input: { username_agent_slug: "creator/calendar-agent" },
|
||||
output: {
|
||||
type: ResponseType.setup_requirements,
|
||||
message:
|
||||
"This agent needs Google credentials. Pick an account or connect a new one.",
|
||||
setup_info: {
|
||||
agent_id: "agent-calendar-1",
|
||||
agent_name: "Google Calendar Agent",
|
||||
requirements: {},
|
||||
user_readiness: {
|
||||
missing_credentials: {
|
||||
google_oauth: {
|
||||
provider: "google",
|
||||
types: ["oauth2"],
|
||||
scopes: ["email", "calendar"],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}}
|
||||
/>
|
||||
</CredentialsProvidersContext.Provider>
|
||||
</SubSection>
|
||||
|
||||
<SubSection label="Output available (need login)">
|
||||
|
||||
@@ -16,7 +16,6 @@ import {
|
||||
ContentCardDescription,
|
||||
ContentCodeBlock,
|
||||
ContentGrid,
|
||||
ContentHint,
|
||||
ContentMessage,
|
||||
} from "../../components/ToolAccordion/AccordionContent";
|
||||
import { ToolAccordion } from "../../components/ToolAccordion/ToolAccordion";
|
||||
@@ -24,8 +23,8 @@ import {
|
||||
ClarificationQuestionsCard,
|
||||
ClarifyingQuestion,
|
||||
} from "./components/ClarificationQuestionsCard";
|
||||
import sparklesImg from "./components/MiniGame/assets/sparkles.png";
|
||||
import { MiniGame } from "./components/MiniGame/MiniGame";
|
||||
import sparklesImg from "../../components/MiniGame/assets/sparkles.png";
|
||||
import { MiniGame } from "../../components/MiniGame/MiniGame";
|
||||
import { SuggestedGoalCard } from "./components/SuggestedGoalCard";
|
||||
import {
|
||||
AccordionIcon,
|
||||
@@ -93,9 +92,7 @@ function getAccordionMeta(output: CreateAgentToolOutput) {
|
||||
) {
|
||||
return {
|
||||
icon,
|
||||
title:
|
||||
"Creating agent, this may take a few minutes. Play while you wait.",
|
||||
expanded: true,
|
||||
title: output.message || "Agent creation started",
|
||||
};
|
||||
}
|
||||
return {
|
||||
@@ -169,15 +166,22 @@ export function CreateAgentTool({ part }: Props) {
|
||||
/>
|
||||
</div>
|
||||
|
||||
{isStreaming && (
|
||||
<ToolAccordion
|
||||
icon={<AccordionIcon />}
|
||||
title="Creating agent, this may take a few minutes. Play while you wait."
|
||||
expanded
|
||||
>
|
||||
<ContentGrid>
|
||||
<MiniGame />
|
||||
</ContentGrid>
|
||||
</ToolAccordion>
|
||||
)}
|
||||
|
||||
{hasExpandableContent && output && (
|
||||
<ToolAccordion {...getAccordionMeta(output)}>
|
||||
{isOperating && (
|
||||
<ContentGrid>
|
||||
<MiniGame />
|
||||
<ContentHint>
|
||||
This could take a few minutes — play while you wait!
|
||||
</ContentHint>
|
||||
</ContentGrid>
|
||||
{isOperating && output.message && (
|
||||
<ContentMessage>{output.message}</ContentMessage>
|
||||
)}
|
||||
|
||||
{isAgentSavedOutput(output) && (
|
||||
|
||||
@@ -4,17 +4,15 @@ import { WarningDiamondIcon } from "@phosphor-icons/react";
|
||||
import type { ToolUIPart } from "ai";
|
||||
import { useCopilotChatActions } from "../../components/CopilotChatActionsProvider/useCopilotChatActions";
|
||||
import { MorphingTextAnimation } from "../../components/MorphingTextAnimation/MorphingTextAnimation";
|
||||
import { OrbitLoader } from "../../components/OrbitLoader/OrbitLoader";
|
||||
import {
|
||||
ContentCardDescription,
|
||||
ContentCodeBlock,
|
||||
ContentGrid,
|
||||
ContentHint,
|
||||
ContentLink,
|
||||
ContentMessage,
|
||||
} from "../../components/ToolAccordion/AccordionContent";
|
||||
import { ToolAccordion } from "../../components/ToolAccordion/ToolAccordion";
|
||||
import { MiniGame } from "../CreateAgent/components/MiniGame/MiniGame";
|
||||
import { MiniGame } from "../../components/MiniGame/MiniGame";
|
||||
import {
|
||||
ClarificationQuestionsCard,
|
||||
ClarifyingQuestion,
|
||||
@@ -81,9 +79,8 @@ function getAccordionMeta(output: EditAgentToolOutput): {
|
||||
isOperationInProgressOutput(output)
|
||||
) {
|
||||
return {
|
||||
icon: <OrbitLoader size={32} />,
|
||||
title: "Editing agent, this may take a few minutes. Play while you wait.",
|
||||
expanded: true,
|
||||
icon,
|
||||
title: output.message || "Agent editing started",
|
||||
};
|
||||
}
|
||||
return {
|
||||
@@ -148,15 +145,22 @@ export function EditAgentTool({ part }: Props) {
|
||||
/>
|
||||
</div>
|
||||
|
||||
{isStreaming && (
|
||||
<ToolAccordion
|
||||
icon={<AccordionIcon />}
|
||||
title="Editing agent, this may take a few minutes. Play while you wait."
|
||||
expanded
|
||||
>
|
||||
<ContentGrid>
|
||||
<MiniGame />
|
||||
</ContentGrid>
|
||||
</ToolAccordion>
|
||||
)}
|
||||
|
||||
{hasExpandableContent && output && (
|
||||
<ToolAccordion {...getAccordionMeta(output)}>
|
||||
{isOperating && (
|
||||
<ContentGrid>
|
||||
<MiniGame />
|
||||
<ContentHint>
|
||||
This could take a few minutes — play while you wait!
|
||||
</ContentHint>
|
||||
</ContentGrid>
|
||||
{isOperating && output.message && (
|
||||
<ContentMessage>{output.message}</ContentMessage>
|
||||
)}
|
||||
|
||||
{isAgentSavedOutput(output) && (
|
||||
|
||||
@@ -9,7 +9,7 @@ import {
|
||||
ContentHint,
|
||||
ContentMessage,
|
||||
} from "../../components/ToolAccordion/AccordionContent";
|
||||
import { MiniGame } from "../CreateAgent/components/MiniGame/MiniGame";
|
||||
import { MiniGame } from "../../components/MiniGame/MiniGame";
|
||||
import {
|
||||
getAccordionMeta,
|
||||
getAnimationText,
|
||||
@@ -47,14 +47,25 @@ export function RunAgentTool({ part }: Props) {
|
||||
const isError =
|
||||
part.state === "output-error" ||
|
||||
(!!output && isRunAgentErrorOutput(output));
|
||||
const isOutputAvailable = part.state === "output-available" && !!output;
|
||||
|
||||
const setupRequirementsOutput =
|
||||
isOutputAvailable && isRunAgentSetupRequirementsOutput(output)
|
||||
? output
|
||||
: null;
|
||||
|
||||
const agentDetailsOutput =
|
||||
isOutputAvailable && isRunAgentAgentDetailsOutput(output) ? output : null;
|
||||
|
||||
const needLoginOutput =
|
||||
isOutputAvailable && isRunAgentNeedLoginOutput(output) ? output : null;
|
||||
|
||||
const hasExpandableContent =
|
||||
part.state === "output-available" &&
|
||||
!!output &&
|
||||
(isRunAgentExecutionStartedOutput(output) ||
|
||||
isRunAgentAgentDetailsOutput(output) ||
|
||||
isRunAgentSetupRequirementsOutput(output) ||
|
||||
isRunAgentNeedLoginOutput(output) ||
|
||||
isRunAgentErrorOutput(output));
|
||||
isOutputAvailable &&
|
||||
!setupRequirementsOutput &&
|
||||
!agentDetailsOutput &&
|
||||
!needLoginOutput &&
|
||||
(isRunAgentExecutionStartedOutput(output) || isRunAgentErrorOutput(output));
|
||||
|
||||
return (
|
||||
<div className="py-2">
|
||||
@@ -81,24 +92,30 @@ export function RunAgentTool({ part }: Props) {
|
||||
</ToolAccordion>
|
||||
)}
|
||||
|
||||
{setupRequirementsOutput && (
|
||||
<div className="mt-2">
|
||||
<SetupRequirementsCard output={setupRequirementsOutput} />
|
||||
</div>
|
||||
)}
|
||||
|
||||
{agentDetailsOutput && (
|
||||
<div className="mt-2">
|
||||
<AgentDetailsCard output={agentDetailsOutput} />
|
||||
</div>
|
||||
)}
|
||||
|
||||
{needLoginOutput && (
|
||||
<div className="mt-2">
|
||||
<ContentMessage>{needLoginOutput.message}</ContentMessage>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{hasExpandableContent && output && (
|
||||
<ToolAccordion {...getAccordionMeta(output)}>
|
||||
{isRunAgentExecutionStartedOutput(output) && (
|
||||
<ExecutionStartedCard output={output} />
|
||||
)}
|
||||
|
||||
{isRunAgentAgentDetailsOutput(output) && (
|
||||
<AgentDetailsCard output={output} />
|
||||
)}
|
||||
|
||||
{isRunAgentSetupRequirementsOutput(output) && (
|
||||
<SetupRequirementsCard output={output} />
|
||||
)}
|
||||
|
||||
{isRunAgentNeedLoginOutput(output) && (
|
||||
<ContentMessage>{output.message}</ContentMessage>
|
||||
)}
|
||||
|
||||
{isRunAgentErrorOutput(output) && <ErrorCard output={output} />}
|
||||
</ToolAccordion>
|
||||
)}
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
"use client";
|
||||
|
||||
import { useState } from "react";
|
||||
import { CredentialsGroupedView } from "@/components/contextual/CredentialsInput/components/CredentialsGroupedView/CredentialsGroupedView";
|
||||
import { Button } from "@/components/atoms/Button/Button";
|
||||
import type { CredentialsMetaInput } from "@/lib/autogpt-server-api/types";
|
||||
import type { SetupRequirementsResponse } from "@/app/api/__generated__/models/setupRequirementsResponse";
|
||||
import { Button } from "@/components/atoms/Button/Button";
|
||||
import { Text } from "@/components/atoms/Text/Text";
|
||||
import { CredentialsGroupedView } from "@/components/contextual/CredentialsInput/components/CredentialsGroupedView/CredentialsGroupedView";
|
||||
import type { CredentialsMetaInput } from "@/lib/autogpt-server-api/types";
|
||||
import { useState } from "react";
|
||||
import { useCopilotChatActions } from "../../../../components/CopilotChatActionsProvider/useCopilotChatActions";
|
||||
import {
|
||||
ContentBadge,
|
||||
@@ -38,40 +39,40 @@ export function SetupRequirementsCard({ output }: Props) {
|
||||
setInputCredentials((prev) => ({ ...prev, [key]: value }));
|
||||
}
|
||||
|
||||
const isAllComplete =
|
||||
credentialFields.length > 0 &&
|
||||
const needsCredentials = credentialFields.length > 0;
|
||||
const isAllCredentialsComplete =
|
||||
needsCredentials &&
|
||||
[...requiredCredentials].every((key) => !!inputCredentials[key]);
|
||||
|
||||
const canProceed =
|
||||
!hasSent && (!needsCredentials || isAllCredentialsComplete);
|
||||
|
||||
function handleProceed() {
|
||||
setHasSent(true);
|
||||
onSend(
|
||||
"I've configured the required credentials. Please check if everything is ready and proceed with running the agent.",
|
||||
);
|
||||
const message = needsCredentials
|
||||
? "I've configured the required credentials. Please check if everything is ready and proceed with running the agent."
|
||||
: "Please proceed with running the agent.";
|
||||
onSend(message);
|
||||
}
|
||||
|
||||
return (
|
||||
<div className="grid gap-2">
|
||||
<ContentMessage>{output.message}</ContentMessage>
|
||||
|
||||
{credentialFields.length > 0 && (
|
||||
{needsCredentials && (
|
||||
<div className="rounded-2xl border bg-background p-3">
|
||||
<CredentialsGroupedView
|
||||
credentialFields={credentialFields}
|
||||
requiredCredentials={requiredCredentials}
|
||||
inputCredentials={inputCredentials}
|
||||
inputValues={{}}
|
||||
onCredentialChange={handleCredentialChange}
|
||||
/>
|
||||
{isAllComplete && !hasSent && (
|
||||
<Button
|
||||
variant="primary"
|
||||
size="small"
|
||||
className="mt-3 w-full"
|
||||
onClick={handleProceed}
|
||||
>
|
||||
Proceed
|
||||
</Button>
|
||||
)}
|
||||
<Text variant="small" className="w-fit border-b text-zinc-500">
|
||||
Agent credentials
|
||||
</Text>
|
||||
<div className="mt-6">
|
||||
<CredentialsGroupedView
|
||||
credentialFields={credentialFields}
|
||||
requiredCredentials={requiredCredentials}
|
||||
inputCredentials={inputCredentials}
|
||||
inputValues={{}}
|
||||
onCredentialChange={handleCredentialChange}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
|
||||
@@ -100,6 +101,18 @@ export function SetupRequirementsCard({ output }: Props) {
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{(needsCredentials || expectedInputs.length > 0) && (
|
||||
<Button
|
||||
variant="primary"
|
||||
size="small"
|
||||
className="mt-4 w-fit"
|
||||
disabled={!canProceed}
|
||||
onClick={handleProceed}
|
||||
>
|
||||
Proceed
|
||||
</Button>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -39,12 +39,19 @@ export function RunBlockTool({ part }: Props) {
|
||||
const isError =
|
||||
part.state === "output-error" ||
|
||||
(!!output && isRunBlockErrorOutput(output));
|
||||
const setupRequirementsOutput =
|
||||
part.state === "output-available" &&
|
||||
output &&
|
||||
isRunBlockSetupRequirementsOutput(output)
|
||||
? output
|
||||
: null;
|
||||
|
||||
const hasExpandableContent =
|
||||
part.state === "output-available" &&
|
||||
!!output &&
|
||||
!setupRequirementsOutput &&
|
||||
(isRunBlockBlockOutput(output) ||
|
||||
isRunBlockDetailsOutput(output) ||
|
||||
isRunBlockSetupRequirementsOutput(output) ||
|
||||
isRunBlockErrorOutput(output));
|
||||
|
||||
return (
|
||||
@@ -57,6 +64,12 @@ export function RunBlockTool({ part }: Props) {
|
||||
/>
|
||||
</div>
|
||||
|
||||
{setupRequirementsOutput && (
|
||||
<div className="mt-2">
|
||||
<SetupRequirementsCard output={setupRequirementsOutput} />
|
||||
</div>
|
||||
)}
|
||||
|
||||
{hasExpandableContent && output && (
|
||||
<ToolAccordion {...getAccordionMeta(output)}>
|
||||
{isRunBlockBlockOutput(output) && <BlockOutputCard output={output} />}
|
||||
@@ -65,10 +78,6 @@ export function RunBlockTool({ part }: Props) {
|
||||
<BlockDetailsCard output={output} />
|
||||
)}
|
||||
|
||||
{isRunBlockSetupRequirementsOutput(output) && (
|
||||
<SetupRequirementsCard output={output} />
|
||||
)}
|
||||
|
||||
{isRunBlockErrorOutput(output) && <ErrorCard output={output} />}
|
||||
</ToolAccordion>
|
||||
)}
|
||||
|
||||
@@ -6,15 +6,9 @@ import { Text } from "@/components/atoms/Text/Text";
|
||||
import { CredentialsGroupedView } from "@/components/contextual/CredentialsInput/components/CredentialsGroupedView/CredentialsGroupedView";
|
||||
import { FormRenderer } from "@/components/renderers/InputRenderer/FormRenderer";
|
||||
import type { CredentialsMetaInput } from "@/lib/autogpt-server-api/types";
|
||||
import { AnimatePresence, motion } from "framer-motion";
|
||||
import { useState } from "react";
|
||||
import { useCopilotChatActions } from "../../../../components/CopilotChatActionsProvider/useCopilotChatActions";
|
||||
import {
|
||||
ContentBadge,
|
||||
ContentCardDescription,
|
||||
ContentCardTitle,
|
||||
ContentMessage,
|
||||
} from "../../../../components/ToolAccordion/AccordionContent";
|
||||
import { ContentMessage } from "../../../../components/ToolAccordion/AccordionContent";
|
||||
import {
|
||||
buildExpectedInputsSchema,
|
||||
coerceCredentialFields,
|
||||
@@ -31,10 +25,8 @@ export function SetupRequirementsCard({ output }: Props) {
|
||||
const [inputCredentials, setInputCredentials] = useState<
|
||||
Record<string, CredentialsMetaInput | undefined>
|
||||
>({});
|
||||
const [hasSentCredentials, setHasSentCredentials] = useState(false);
|
||||
|
||||
const [showInputForm, setShowInputForm] = useState(false);
|
||||
const [inputValues, setInputValues] = useState<Record<string, unknown>>({});
|
||||
const [hasSent, setHasSent] = useState(false);
|
||||
|
||||
const { credentialFields, requiredCredentials } = coerceCredentialFields(
|
||||
output.setup_info.user_readiness?.missing_credentials,
|
||||
@@ -50,27 +42,49 @@ export function SetupRequirementsCard({ output }: Props) {
|
||||
setInputCredentials((prev) => ({ ...prev, [key]: value }));
|
||||
}
|
||||
|
||||
const needsCredentials = credentialFields.length > 0;
|
||||
const isAllCredentialsComplete =
|
||||
credentialFields.length > 0 &&
|
||||
needsCredentials &&
|
||||
[...requiredCredentials].every((key) => !!inputCredentials[key]);
|
||||
|
||||
function handleProceedCredentials() {
|
||||
setHasSentCredentials(true);
|
||||
onSend(
|
||||
"I've configured the required credentials. Please re-run the block now.",
|
||||
);
|
||||
}
|
||||
const needsInputs = inputSchema !== null;
|
||||
const requiredInputNames = expectedInputs
|
||||
.filter((i) => i.required)
|
||||
.map((i) => i.name);
|
||||
const isAllInputsComplete =
|
||||
needsInputs &&
|
||||
requiredInputNames.every((name) => {
|
||||
const v = inputValues[name];
|
||||
return v !== undefined && v !== null && v !== "";
|
||||
});
|
||||
|
||||
function handleRunWithInputs() {
|
||||
const nonEmpty = Object.fromEntries(
|
||||
Object.entries(inputValues).filter(
|
||||
([, v]) => v !== undefined && v !== null && v !== "",
|
||||
),
|
||||
);
|
||||
onSend(
|
||||
`Run the block with these inputs: ${JSON.stringify(nonEmpty, null, 2)}`,
|
||||
);
|
||||
setShowInputForm(false);
|
||||
const canRun =
|
||||
!hasSent &&
|
||||
(!needsCredentials || isAllCredentialsComplete) &&
|
||||
(!needsInputs || isAllInputsComplete);
|
||||
|
||||
function handleRun() {
|
||||
setHasSent(true);
|
||||
|
||||
const parts: string[] = [];
|
||||
if (needsCredentials) {
|
||||
parts.push("I've configured the required credentials.");
|
||||
}
|
||||
|
||||
if (needsInputs) {
|
||||
const nonEmpty = Object.fromEntries(
|
||||
Object.entries(inputValues).filter(
|
||||
([, v]) => v !== undefined && v !== null && v !== "",
|
||||
),
|
||||
);
|
||||
parts.push(
|
||||
`Run the block with these inputs: ${JSON.stringify(nonEmpty, null, 2)}`,
|
||||
);
|
||||
} else {
|
||||
parts.push("Please re-run the block now.");
|
||||
}
|
||||
|
||||
onSend(parts.join(" "));
|
||||
setInputValues({});
|
||||
}
|
||||
|
||||
@@ -78,119 +92,54 @@ export function SetupRequirementsCard({ output }: Props) {
|
||||
<div className="grid gap-2">
|
||||
<ContentMessage>{output.message}</ContentMessage>
|
||||
|
||||
{credentialFields.length > 0 && (
|
||||
{needsCredentials && (
|
||||
<div className="rounded-2xl border bg-background p-3">
|
||||
<CredentialsGroupedView
|
||||
credentialFields={credentialFields}
|
||||
requiredCredentials={requiredCredentials}
|
||||
inputCredentials={inputCredentials}
|
||||
inputValues={{}}
|
||||
onCredentialChange={handleCredentialChange}
|
||||
/>
|
||||
{isAllCredentialsComplete && !hasSentCredentials && (
|
||||
<Button
|
||||
variant="primary"
|
||||
size="small"
|
||||
className="mt-3 w-full"
|
||||
onClick={handleProceedCredentials}
|
||||
>
|
||||
Proceed
|
||||
</Button>
|
||||
)}
|
||||
<Text variant="small" className="w-fit border-b text-zinc-500">
|
||||
Block credentials
|
||||
</Text>
|
||||
<div className="mt-6">
|
||||
<CredentialsGroupedView
|
||||
credentialFields={credentialFields}
|
||||
requiredCredentials={requiredCredentials}
|
||||
inputCredentials={inputCredentials}
|
||||
inputValues={{}}
|
||||
onCredentialChange={handleCredentialChange}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{inputSchema && (
|
||||
<div className="flex gap-2 pt-2">
|
||||
<Button
|
||||
variant="outline"
|
||||
size="small"
|
||||
className="w-fit"
|
||||
onClick={() => setShowInputForm((prev) => !prev)}
|
||||
>
|
||||
{showInputForm ? "Hide inputs" : "Fill in inputs"}
|
||||
</Button>
|
||||
<div className="rounded-2xl border bg-background p-3 pt-4">
|
||||
<Text variant="small" className="w-fit border-b text-zinc-500">
|
||||
Block inputs
|
||||
</Text>
|
||||
<FormRenderer
|
||||
jsonSchema={inputSchema}
|
||||
className="mb-3 mt-3"
|
||||
handleChange={(v) => setInputValues(v.formData ?? {})}
|
||||
uiSchema={{
|
||||
"ui:submitButtonOptions": { norender: true },
|
||||
}}
|
||||
initialValues={inputValues}
|
||||
formContext={{
|
||||
showHandles: false,
|
||||
size: "small",
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
|
||||
<AnimatePresence initial={false}>
|
||||
{showInputForm && inputSchema && (
|
||||
<motion.div
|
||||
initial={{ height: 0, opacity: 0, filter: "blur(6px)" }}
|
||||
animate={{ height: "auto", opacity: 1, filter: "blur(0px)" }}
|
||||
exit={{ height: 0, opacity: 0, filter: "blur(6px)" }}
|
||||
transition={{
|
||||
height: { type: "spring", bounce: 0.15, duration: 0.5 },
|
||||
opacity: { duration: 0.25 },
|
||||
filter: { duration: 0.2 },
|
||||
}}
|
||||
className="overflow-hidden"
|
||||
style={{ willChange: "height, opacity, filter" }}
|
||||
>
|
||||
<div className="rounded-2xl border bg-background p-3 pt-4">
|
||||
<Text variant="body-medium">Block inputs</Text>
|
||||
<FormRenderer
|
||||
jsonSchema={inputSchema}
|
||||
handleChange={(v) => setInputValues(v.formData ?? {})}
|
||||
uiSchema={{
|
||||
"ui:submitButtonOptions": { norender: true },
|
||||
}}
|
||||
initialValues={inputValues}
|
||||
formContext={{
|
||||
showHandles: false,
|
||||
size: "small",
|
||||
}}
|
||||
/>
|
||||
<div className="-mt-8 flex gap-2">
|
||||
<Button
|
||||
variant="primary"
|
||||
size="small"
|
||||
className="w-fit"
|
||||
onClick={handleRunWithInputs}
|
||||
>
|
||||
Run
|
||||
</Button>
|
||||
<Button
|
||||
variant="secondary"
|
||||
size="small"
|
||||
className="w-fit"
|
||||
onClick={() => {
|
||||
setShowInputForm(false);
|
||||
setInputValues({});
|
||||
}}
|
||||
>
|
||||
Cancel
|
||||
</Button>
|
||||
</div>
|
||||
</div>
|
||||
</motion.div>
|
||||
)}
|
||||
</AnimatePresence>
|
||||
|
||||
{expectedInputs.length > 0 && !inputSchema && (
|
||||
<div className="rounded-2xl border bg-background p-3">
|
||||
<ContentCardTitle className="text-xs">
|
||||
Expected inputs
|
||||
</ContentCardTitle>
|
||||
<div className="mt-2 grid gap-2">
|
||||
{expectedInputs.map((input) => (
|
||||
<div key={input.name} className="rounded-xl border p-2">
|
||||
<div className="flex items-center justify-between gap-2">
|
||||
<ContentCardTitle className="text-xs">
|
||||
{input.title}
|
||||
</ContentCardTitle>
|
||||
<ContentBadge>
|
||||
{input.required ? "Required" : "Optional"}
|
||||
</ContentBadge>
|
||||
</div>
|
||||
<ContentCardDescription className="mt-1">
|
||||
{input.name} • {input.type}
|
||||
{input.description ? ` \u2022 ${input.description}` : ""}
|
||||
</ContentCardDescription>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
</div>
|
||||
{(needsCredentials || needsInputs) && (
|
||||
<Button
|
||||
variant="primary"
|
||||
size="small"
|
||||
className="w-fit"
|
||||
disabled={!canRun}
|
||||
onClick={handleRun}
|
||||
>
|
||||
Proceed
|
||||
</Button>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
|
||||
@@ -119,7 +119,7 @@ export function CredentialsFlatView({
|
||||
) : (
|
||||
!readOnly && (
|
||||
<Button
|
||||
variant="secondary"
|
||||
variant="primary"
|
||||
size="small"
|
||||
onClick={onAddCredential}
|
||||
className="w-fit"
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
import { cn } from "@/lib/utils";
|
||||
import { RJSFSchema } from "@rjsf/utils";
|
||||
import { preprocessInputSchema } from "./utils/input-schema-pre-processor";
|
||||
import { useMemo } from "react";
|
||||
import { customValidator } from "./utils/custom-validator";
|
||||
import Form from "./registry";
|
||||
import { ExtendedFormContextType } from "./types";
|
||||
import { customValidator } from "./utils/custom-validator";
|
||||
import { generateUiSchemaForCustomFields } from "./utils/generate-ui-schema";
|
||||
import { preprocessInputSchema } from "./utils/input-schema-pre-processor";
|
||||
|
||||
type FormRendererProps = {
|
||||
jsonSchema: RJSFSchema;
|
||||
@@ -12,15 +13,17 @@ type FormRendererProps = {
|
||||
uiSchema: any;
|
||||
initialValues: any;
|
||||
formContext: ExtendedFormContextType;
|
||||
className?: string;
|
||||
};
|
||||
|
||||
export const FormRenderer = ({
|
||||
export function FormRenderer({
|
||||
jsonSchema,
|
||||
handleChange,
|
||||
uiSchema,
|
||||
initialValues,
|
||||
formContext,
|
||||
}: FormRendererProps) => {
|
||||
className,
|
||||
}: FormRendererProps) {
|
||||
const preprocessedSchema = useMemo(() => {
|
||||
return preprocessInputSchema(jsonSchema);
|
||||
}, [jsonSchema]);
|
||||
@@ -31,7 +34,10 @@ export const FormRenderer = ({
|
||||
}, [preprocessedSchema, uiSchema]);
|
||||
|
||||
return (
|
||||
<div className={"mb-6 mt-4"} data-tutorial-id="input-handles">
|
||||
<div
|
||||
className={cn("mb-6 mt-4", className)}
|
||||
data-tutorial-id="input-handles"
|
||||
>
|
||||
<Form
|
||||
formContext={formContext}
|
||||
idPrefix="agpt"
|
||||
@@ -45,4 +51,4 @@ export const FormRenderer = ({
|
||||
/>
|
||||
</div>
|
||||
);
|
||||
};
|
||||
}
|
||||
|
||||
@@ -218,6 +218,17 @@ If you initially installed Docker with Hyper-V, you **don’t need to reinstall*
|
||||
|
||||
For more details, refer to [Docker's official documentation](https://docs.docker.com/desktop/windows/wsl/).
|
||||
|
||||
### ⚠️ Podman Not Supported
|
||||
|
||||
AutoGPT requires **Docker** (Docker Desktop or Docker Engine). **Podman and podman-compose are not supported** and may cause path resolution issues, particularly on Windows.
|
||||
|
||||
If you see errors like:
|
||||
```text
|
||||
Error: the specified Containerfile or Dockerfile does not exist, ..\..\autogpt_platform\backend\Dockerfile
|
||||
```
|
||||
|
||||
This indicates you're using Podman instead of Docker. Please install [Docker Desktop](https://docs.docker.com/desktop/) and use `docker compose` instead of `podman-compose`.
|
||||
|
||||
|
||||
## Development
|
||||
|
||||
|
||||
Reference in New Issue
Block a user