mirror of
https://github.com/Significant-Gravitas/AutoGPT.git
synced 2026-02-12 07:45:14 -05:00
Compare commits
1 Commits
fix/claude
...
lluisagust
| Author | SHA1 | Date | |
|---|---|---|---|
|
4eff3fd581 |
@@ -2,7 +2,7 @@ import asyncio
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, cast
|
||||
from typing import Any
|
||||
from weakref import WeakValueDictionary
|
||||
|
||||
from openai.types.chat import (
|
||||
@@ -104,26 +104,6 @@ class ChatSession(BaseModel):
|
||||
successful_agent_runs: dict[str, int] = {}
|
||||
successful_agent_schedules: dict[str, int] = {}
|
||||
|
||||
def add_tool_call_to_current_turn(self, tool_call: dict) -> None:
|
||||
"""Attach a tool_call to the current turn's assistant message.
|
||||
|
||||
Searches backwards for the most recent assistant message (stopping at
|
||||
any user message boundary). If found, appends the tool_call to it.
|
||||
Otherwise creates a new assistant message with the tool_call.
|
||||
"""
|
||||
for msg in reversed(self.messages):
|
||||
if msg.role == "user":
|
||||
break
|
||||
if msg.role == "assistant":
|
||||
if not msg.tool_calls:
|
||||
msg.tool_calls = []
|
||||
msg.tool_calls.append(tool_call)
|
||||
return
|
||||
|
||||
self.messages.append(
|
||||
ChatMessage(role="assistant", content="", tool_calls=[tool_call])
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def new(user_id: str) -> "ChatSession":
|
||||
return ChatSession(
|
||||
@@ -192,47 +172,6 @@ class ChatSession(BaseModel):
|
||||
successful_agent_schedules=successful_agent_schedules,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _merge_consecutive_assistant_messages(
|
||||
messages: list[ChatCompletionMessageParam],
|
||||
) -> list[ChatCompletionMessageParam]:
|
||||
"""Merge consecutive assistant messages into single messages.
|
||||
|
||||
Long-running tool flows can create split assistant messages: one with
|
||||
text content and another with tool_calls. Anthropic's API requires
|
||||
tool_result blocks to reference a tool_use in the immediately preceding
|
||||
assistant message, so these splits cause 400 errors via OpenRouter.
|
||||
"""
|
||||
if len(messages) < 2:
|
||||
return messages
|
||||
|
||||
result: list[ChatCompletionMessageParam] = [messages[0]]
|
||||
for msg in messages[1:]:
|
||||
prev = result[-1]
|
||||
if prev.get("role") != "assistant" or msg.get("role") != "assistant":
|
||||
result.append(msg)
|
||||
continue
|
||||
|
||||
prev = cast(ChatCompletionAssistantMessageParam, prev)
|
||||
curr = cast(ChatCompletionAssistantMessageParam, msg)
|
||||
|
||||
curr_content = curr.get("content") or ""
|
||||
if curr_content:
|
||||
prev_content = prev.get("content") or ""
|
||||
prev["content"] = (
|
||||
f"{prev_content}\n{curr_content}" if prev_content else curr_content
|
||||
)
|
||||
|
||||
curr_tool_calls = curr.get("tool_calls")
|
||||
if curr_tool_calls:
|
||||
prev_tool_calls = prev.get("tool_calls")
|
||||
prev["tool_calls"] = (
|
||||
list(prev_tool_calls) + list(curr_tool_calls)
|
||||
if prev_tool_calls
|
||||
else list(curr_tool_calls)
|
||||
)
|
||||
return result
|
||||
|
||||
def to_openai_messages(self) -> list[ChatCompletionMessageParam]:
|
||||
messages = []
|
||||
for message in self.messages:
|
||||
@@ -319,7 +258,7 @@ class ChatSession(BaseModel):
|
||||
name=message.name or "",
|
||||
)
|
||||
)
|
||||
return self._merge_consecutive_assistant_messages(messages)
|
||||
return messages
|
||||
|
||||
|
||||
async def _get_session_from_cache(session_id: str) -> ChatSession | None:
|
||||
|
||||
@@ -1,16 +1,4 @@
|
||||
from typing import cast
|
||||
|
||||
import pytest
|
||||
from openai.types.chat import (
|
||||
ChatCompletionAssistantMessageParam,
|
||||
ChatCompletionMessageParam,
|
||||
ChatCompletionToolMessageParam,
|
||||
ChatCompletionUserMessageParam,
|
||||
)
|
||||
from openai.types.chat.chat_completion_message_tool_call_param import (
|
||||
ChatCompletionMessageToolCallParam,
|
||||
Function,
|
||||
)
|
||||
|
||||
from .model import (
|
||||
ChatMessage,
|
||||
@@ -129,205 +117,3 @@ async def test_chatsession_db_storage(setup_test_user, test_user_id):
|
||||
loaded.tool_calls is not None
|
||||
), f"Tool calls missing for {orig.role} message"
|
||||
assert len(orig.tool_calls) == len(loaded.tool_calls)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# _merge_consecutive_assistant_messages #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
_tc = ChatCompletionMessageToolCallParam(
|
||||
id="tc1", type="function", function=Function(name="do_stuff", arguments="{}")
|
||||
)
|
||||
_tc2 = ChatCompletionMessageToolCallParam(
|
||||
id="tc2", type="function", function=Function(name="other", arguments="{}")
|
||||
)
|
||||
|
||||
|
||||
def test_merge_noop_when_no_consecutive_assistants():
|
||||
"""Messages without consecutive assistants are returned unchanged."""
|
||||
msgs = [
|
||||
ChatCompletionUserMessageParam(role="user", content="hi"),
|
||||
ChatCompletionAssistantMessageParam(role="assistant", content="hello"),
|
||||
ChatCompletionUserMessageParam(role="user", content="bye"),
|
||||
]
|
||||
merged = ChatSession._merge_consecutive_assistant_messages(msgs)
|
||||
assert len(merged) == 3
|
||||
assert [m["role"] for m in merged] == ["user", "assistant", "user"]
|
||||
|
||||
|
||||
def test_merge_splits_text_and_tool_calls():
|
||||
"""The exact bug scenario: text-only assistant followed by tool_calls-only assistant."""
|
||||
msgs = [
|
||||
ChatCompletionUserMessageParam(role="user", content="build agent"),
|
||||
ChatCompletionAssistantMessageParam(
|
||||
role="assistant", content="Let me build that"
|
||||
),
|
||||
ChatCompletionAssistantMessageParam(
|
||||
role="assistant", content="", tool_calls=[_tc]
|
||||
),
|
||||
ChatCompletionToolMessageParam(role="tool", content="ok", tool_call_id="tc1"),
|
||||
]
|
||||
merged = ChatSession._merge_consecutive_assistant_messages(msgs)
|
||||
|
||||
assert len(merged) == 3
|
||||
assert merged[0]["role"] == "user"
|
||||
assert merged[2]["role"] == "tool"
|
||||
a = cast(ChatCompletionAssistantMessageParam, merged[1])
|
||||
assert a["role"] == "assistant"
|
||||
assert a.get("content") == "Let me build that"
|
||||
assert a.get("tool_calls") == [_tc]
|
||||
|
||||
|
||||
def test_merge_combines_tool_calls_from_both():
|
||||
"""Both consecutive assistants have tool_calls — they get merged."""
|
||||
msgs: list[ChatCompletionAssistantMessageParam] = [
|
||||
ChatCompletionAssistantMessageParam(
|
||||
role="assistant", content="text", tool_calls=[_tc]
|
||||
),
|
||||
ChatCompletionAssistantMessageParam(
|
||||
role="assistant", content="", tool_calls=[_tc2]
|
||||
),
|
||||
]
|
||||
merged = ChatSession._merge_consecutive_assistant_messages(msgs) # type: ignore[arg-type]
|
||||
|
||||
assert len(merged) == 1
|
||||
a = cast(ChatCompletionAssistantMessageParam, merged[0])
|
||||
assert a.get("tool_calls") == [_tc, _tc2]
|
||||
assert a.get("content") == "text"
|
||||
|
||||
|
||||
def test_merge_three_consecutive_assistants():
|
||||
"""Three consecutive assistants collapse into one."""
|
||||
msgs: list[ChatCompletionAssistantMessageParam] = [
|
||||
ChatCompletionAssistantMessageParam(role="assistant", content="a"),
|
||||
ChatCompletionAssistantMessageParam(role="assistant", content="b"),
|
||||
ChatCompletionAssistantMessageParam(
|
||||
role="assistant", content="", tool_calls=[_tc]
|
||||
),
|
||||
]
|
||||
merged = ChatSession._merge_consecutive_assistant_messages(msgs) # type: ignore[arg-type]
|
||||
|
||||
assert len(merged) == 1
|
||||
a = cast(ChatCompletionAssistantMessageParam, merged[0])
|
||||
assert a.get("content") == "a\nb"
|
||||
assert a.get("tool_calls") == [_tc]
|
||||
|
||||
|
||||
def test_merge_empty_and_single_message():
|
||||
"""Edge cases: empty list and single message."""
|
||||
assert ChatSession._merge_consecutive_assistant_messages([]) == []
|
||||
|
||||
single: list[ChatCompletionMessageParam] = [
|
||||
ChatCompletionUserMessageParam(role="user", content="hi")
|
||||
]
|
||||
assert ChatSession._merge_consecutive_assistant_messages(single) == single
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# add_tool_call_to_current_turn #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
_raw_tc = {
|
||||
"id": "tc1",
|
||||
"type": "function",
|
||||
"function": {"name": "f", "arguments": "{}"},
|
||||
}
|
||||
_raw_tc2 = {
|
||||
"id": "tc2",
|
||||
"type": "function",
|
||||
"function": {"name": "g", "arguments": "{}"},
|
||||
}
|
||||
|
||||
|
||||
def test_add_tool_call_appends_to_existing_assistant():
|
||||
"""When the last assistant is from the current turn, tool_call is added to it."""
|
||||
session = ChatSession.new(user_id="u")
|
||||
session.messages = [
|
||||
ChatMessage(role="user", content="hi"),
|
||||
ChatMessage(role="assistant", content="working on it"),
|
||||
]
|
||||
session.add_tool_call_to_current_turn(_raw_tc)
|
||||
|
||||
assert len(session.messages) == 2 # no new message created
|
||||
assert session.messages[1].tool_calls == [_raw_tc]
|
||||
|
||||
|
||||
def test_add_tool_call_creates_assistant_when_none_exists():
|
||||
"""When there's no current-turn assistant, a new one is created."""
|
||||
session = ChatSession.new(user_id="u")
|
||||
session.messages = [
|
||||
ChatMessage(role="user", content="hi"),
|
||||
]
|
||||
session.add_tool_call_to_current_turn(_raw_tc)
|
||||
|
||||
assert len(session.messages) == 2
|
||||
assert session.messages[1].role == "assistant"
|
||||
assert session.messages[1].tool_calls == [_raw_tc]
|
||||
|
||||
|
||||
def test_add_tool_call_does_not_cross_user_boundary():
|
||||
"""A user message acts as a boundary — previous assistant is not modified."""
|
||||
session = ChatSession.new(user_id="u")
|
||||
session.messages = [
|
||||
ChatMessage(role="assistant", content="old turn"),
|
||||
ChatMessage(role="user", content="new message"),
|
||||
]
|
||||
session.add_tool_call_to_current_turn(_raw_tc)
|
||||
|
||||
assert len(session.messages) == 3 # new assistant was created
|
||||
assert session.messages[0].tool_calls is None # old assistant untouched
|
||||
assert session.messages[2].role == "assistant"
|
||||
assert session.messages[2].tool_calls == [_raw_tc]
|
||||
|
||||
|
||||
def test_add_tool_call_multiple_times():
|
||||
"""Multiple long-running tool calls accumulate on the same assistant."""
|
||||
session = ChatSession.new(user_id="u")
|
||||
session.messages = [
|
||||
ChatMessage(role="user", content="hi"),
|
||||
ChatMessage(role="assistant", content="doing stuff"),
|
||||
]
|
||||
session.add_tool_call_to_current_turn(_raw_tc)
|
||||
# Simulate a pending tool result in between (like _yield_tool_call does)
|
||||
session.messages.append(
|
||||
ChatMessage(role="tool", content="pending", tool_call_id="tc1")
|
||||
)
|
||||
session.add_tool_call_to_current_turn(_raw_tc2)
|
||||
|
||||
assert len(session.messages) == 3 # user, assistant, tool — no extra assistant
|
||||
assert session.messages[1].tool_calls == [_raw_tc, _raw_tc2]
|
||||
|
||||
|
||||
def test_to_openai_messages_merges_split_assistants():
|
||||
"""End-to-end: session with split assistants produces valid OpenAI messages."""
|
||||
session = ChatSession.new(user_id="u")
|
||||
session.messages = [
|
||||
ChatMessage(role="user", content="build agent"),
|
||||
ChatMessage(role="assistant", content="Let me build that"),
|
||||
ChatMessage(
|
||||
role="assistant",
|
||||
content="",
|
||||
tool_calls=[
|
||||
{
|
||||
"id": "tc1",
|
||||
"type": "function",
|
||||
"function": {"name": "create_agent", "arguments": "{}"},
|
||||
}
|
||||
],
|
||||
),
|
||||
ChatMessage(role="tool", content="done", tool_call_id="tc1"),
|
||||
ChatMessage(role="assistant", content="Saved!"),
|
||||
ChatMessage(role="user", content="show me an example run"),
|
||||
]
|
||||
openai_msgs = session.to_openai_messages()
|
||||
|
||||
# The two consecutive assistants at index 1,2 should be merged
|
||||
roles = [m["role"] for m in openai_msgs]
|
||||
assert roles == ["user", "assistant", "tool", "assistant", "user"]
|
||||
|
||||
# The merged assistant should have both content and tool_calls
|
||||
merged = cast(ChatCompletionAssistantMessageParam, openai_msgs[1])
|
||||
assert merged.get("content") == "Let me build that"
|
||||
tc_list = merged.get("tool_calls")
|
||||
assert tc_list is not None and len(list(tc_list)) == 1
|
||||
assert list(tc_list)[0]["id"] == "tc1"
|
||||
|
||||
@@ -800,13 +800,9 @@ async def stream_chat_completion(
|
||||
# Build the messages list in the correct order
|
||||
messages_to_save: list[ChatMessage] = []
|
||||
|
||||
# Add assistant message with tool_calls if any.
|
||||
# Use extend (not assign) to preserve tool_calls already added by
|
||||
# _yield_tool_call for long-running tools.
|
||||
# Add assistant message with tool_calls if any
|
||||
if accumulated_tool_calls:
|
||||
if not assistant_response.tool_calls:
|
||||
assistant_response.tool_calls = []
|
||||
assistant_response.tool_calls.extend(accumulated_tool_calls)
|
||||
assistant_response.tool_calls = accumulated_tool_calls
|
||||
logger.info(
|
||||
f"Added {len(accumulated_tool_calls)} tool calls to assistant message"
|
||||
)
|
||||
@@ -1408,9 +1404,13 @@ async def _yield_tool_call(
|
||||
operation_id=operation_id,
|
||||
)
|
||||
|
||||
# Attach the tool_call to the current turn's assistant message
|
||||
# (or create one if this is a tool-only response with no text).
|
||||
session.add_tool_call_to_current_turn(tool_calls[yield_idx])
|
||||
# Save assistant message with tool_call FIRST (required by LLM)
|
||||
assistant_message = ChatMessage(
|
||||
role="assistant",
|
||||
content="",
|
||||
tool_calls=[tool_calls[yield_idx]],
|
||||
)
|
||||
session.messages.append(assistant_message)
|
||||
|
||||
# Then save pending tool result
|
||||
pending_message = ChatMessage(
|
||||
|
||||
@@ -1,6 +1,4 @@
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import shlex
|
||||
import uuid
|
||||
from typing import Literal, Optional
|
||||
@@ -23,11 +21,6 @@ from backend.data.model import (
|
||||
)
|
||||
from backend.integrations.providers import ProviderName
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Maximum size for binary files to extract (50MB)
|
||||
MAX_BINARY_FILE_SIZE = 50 * 1024 * 1024
|
||||
|
||||
|
||||
class ClaudeCodeExecutionError(Exception):
|
||||
"""Exception raised when Claude Code execution fails.
|
||||
@@ -187,9 +180,7 @@ class ClaudeCodeBlock(Block):
|
||||
path: str
|
||||
relative_path: str # Path relative to working directory (for GitHub, etc.)
|
||||
name: str
|
||||
content: str # Text content for text files, empty string for binary files
|
||||
is_binary: bool = False # True if this is a binary file
|
||||
content_base64: Optional[str] = None # Base64-encoded content for binary files
|
||||
content: str
|
||||
|
||||
class Output(BlockSchemaOutput):
|
||||
response: str = SchemaField(
|
||||
@@ -197,11 +188,8 @@ class ClaudeCodeBlock(Block):
|
||||
)
|
||||
files: list["ClaudeCodeBlock.FileOutput"] = SchemaField(
|
||||
description=(
|
||||
"List of files created/modified by Claude Code during this execution. "
|
||||
"Each file has 'path', 'relative_path', 'name', 'content', 'is_binary', "
|
||||
"and 'content_base64' fields. For text files, 'content' contains the text "
|
||||
"and 'is_binary' is False. For binary files (PDFs, images, etc.), "
|
||||
"'is_binary' is True and 'content_base64' contains the base64-encoded data."
|
||||
"List of text files created/modified by Claude Code during this execution. "
|
||||
"Each file has 'path', 'relative_path', 'name', and 'content' fields."
|
||||
)
|
||||
)
|
||||
conversation_history: str = SchemaField(
|
||||
@@ -264,8 +252,6 @@ class ClaudeCodeBlock(Block):
|
||||
"relative_path": "index.html",
|
||||
"name": "index.html",
|
||||
"content": "<html>Hello World</html>",
|
||||
"is_binary": False,
|
||||
"content_base64": None,
|
||||
}
|
||||
],
|
||||
),
|
||||
@@ -286,8 +272,6 @@ class ClaudeCodeBlock(Block):
|
||||
relative_path="index.html",
|
||||
name="index.html",
|
||||
content="<html>Hello World</html>",
|
||||
is_binary=False,
|
||||
content_base64=None,
|
||||
)
|
||||
], # files
|
||||
"User: Create a hello world HTML file\n"
|
||||
@@ -547,6 +531,7 @@ class ClaudeCodeBlock(Block):
|
||||
".env",
|
||||
".gitignore",
|
||||
".dockerfile",
|
||||
"Dockerfile",
|
||||
".vue",
|
||||
".svelte",
|
||||
".astro",
|
||||
@@ -555,44 +540,6 @@ class ClaudeCodeBlock(Block):
|
||||
".tex",
|
||||
".csv",
|
||||
".log",
|
||||
".svg", # SVG is XML-based text
|
||||
}
|
||||
|
||||
# Binary file extensions we can read and base64-encode
|
||||
binary_extensions = {
|
||||
# Images
|
||||
".png",
|
||||
".jpg",
|
||||
".jpeg",
|
||||
".gif",
|
||||
".webp",
|
||||
".ico",
|
||||
".bmp",
|
||||
".tiff",
|
||||
".tif",
|
||||
# Documents
|
||||
".pdf",
|
||||
# Archives (useful for downloads)
|
||||
".zip",
|
||||
".tar",
|
||||
".gz",
|
||||
".7z",
|
||||
# Audio/Video (if small enough)
|
||||
".mp3",
|
||||
".wav",
|
||||
".mp4",
|
||||
".webm",
|
||||
# Other binary formats
|
||||
".woff",
|
||||
".woff2",
|
||||
".ttf",
|
||||
".otf",
|
||||
".eot",
|
||||
".bin",
|
||||
".exe",
|
||||
".dll",
|
||||
".so",
|
||||
".dylib",
|
||||
}
|
||||
|
||||
try:
|
||||
@@ -617,26 +564,10 @@ class ClaudeCodeBlock(Block):
|
||||
if not file_path:
|
||||
continue
|
||||
|
||||
# Check if it's a text file we can read (case-insensitive)
|
||||
file_path_lower = file_path.lower()
|
||||
# Check if it's a text file we can read
|
||||
is_text = any(
|
||||
file_path_lower.endswith(ext) for ext in text_extensions
|
||||
) or file_path_lower.endswith("dockerfile")
|
||||
|
||||
# Check if it's a binary file we should extract
|
||||
is_binary = any(
|
||||
file_path_lower.endswith(ext) for ext in binary_extensions
|
||||
)
|
||||
|
||||
# Helper to extract filename and relative path
|
||||
def get_file_info(path: str, work_dir: str) -> tuple[str, str]:
|
||||
name = path.split("/")[-1]
|
||||
rel_path = path
|
||||
if path.startswith(work_dir):
|
||||
rel_path = path[len(work_dir) :]
|
||||
if rel_path.startswith("/"):
|
||||
rel_path = rel_path[1:]
|
||||
return name, rel_path
|
||||
file_path.endswith(ext) for ext in text_extensions
|
||||
) or file_path.endswith("Dockerfile")
|
||||
|
||||
if is_text:
|
||||
try:
|
||||
@@ -645,75 +576,32 @@ class ClaudeCodeBlock(Block):
|
||||
if isinstance(content, bytes):
|
||||
content = content.decode("utf-8", errors="replace")
|
||||
|
||||
file_name, relative_path = get_file_info(
|
||||
file_path, working_directory
|
||||
)
|
||||
# Extract filename from path
|
||||
file_name = file_path.split("/")[-1]
|
||||
|
||||
# Calculate relative path by stripping working directory
|
||||
relative_path = file_path
|
||||
if file_path.startswith(working_directory):
|
||||
relative_path = file_path[len(working_directory) :]
|
||||
# Remove leading slash if present
|
||||
if relative_path.startswith("/"):
|
||||
relative_path = relative_path[1:]
|
||||
|
||||
files.append(
|
||||
ClaudeCodeBlock.FileOutput(
|
||||
path=file_path,
|
||||
relative_path=relative_path,
|
||||
name=file_name,
|
||||
content=content,
|
||||
is_binary=False,
|
||||
content_base64=None,
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to read text file {file_path}: {e}")
|
||||
elif is_binary:
|
||||
try:
|
||||
# Check file size before reading to avoid OOM
|
||||
stat_result = await sandbox.commands.run(
|
||||
f"stat -c %s {shlex.quote(file_path)} 2>/dev/null"
|
||||
)
|
||||
if (
|
||||
stat_result.exit_code != 0
|
||||
or not stat_result.stdout
|
||||
):
|
||||
logger.warning(
|
||||
f"Skipping binary file {file_path}: "
|
||||
f"could not determine file size"
|
||||
)
|
||||
continue
|
||||
file_size = int(stat_result.stdout.strip())
|
||||
if file_size > MAX_BINARY_FILE_SIZE:
|
||||
logger.warning(
|
||||
f"Skipping binary file {file_path}: "
|
||||
f"size {file_size} exceeds limit "
|
||||
f"{MAX_BINARY_FILE_SIZE}"
|
||||
)
|
||||
continue
|
||||
except Exception:
|
||||
# Skip files that can't be read
|
||||
pass
|
||||
|
||||
# Read binary file as bytes using format="bytes"
|
||||
content_bytes = await sandbox.files.read(
|
||||
file_path, format="bytes"
|
||||
)
|
||||
|
||||
# Base64 encode the binary content
|
||||
content_b64 = base64.b64encode(content_bytes).decode(
|
||||
"ascii"
|
||||
)
|
||||
|
||||
file_name, relative_path = get_file_info(
|
||||
file_path, working_directory
|
||||
)
|
||||
files.append(
|
||||
ClaudeCodeBlock.FileOutput(
|
||||
path=file_path,
|
||||
relative_path=relative_path,
|
||||
name=file_name,
|
||||
content="", # Empty for binary files
|
||||
is_binary=True,
|
||||
content_base64=content_b64,
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Failed to read binary file {file_path}: {e}"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"File extraction failed: {e}")
|
||||
except Exception:
|
||||
# If file extraction fails, return empty results
|
||||
pass
|
||||
|
||||
return files
|
||||
|
||||
|
||||
@@ -21,71 +21,43 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
class HumanInTheLoopBlock(Block):
|
||||
"""
|
||||
Pauses execution and waits for human approval or rejection of the data.
|
||||
This block pauses execution and waits for human approval or modification of the data.
|
||||
|
||||
When executed, this block creates a pending review entry and sets the node execution
|
||||
status to REVIEW. The execution remains paused until a human user either approves
|
||||
or rejects the data.
|
||||
When executed, it creates a pending review entry and sets the node execution status
|
||||
to REVIEW. The execution will remain paused until a human user either:
|
||||
- Approves the data (with or without modifications)
|
||||
- Rejects the data
|
||||
|
||||
**How it works:**
|
||||
- The input data is presented to a human reviewer
|
||||
- The reviewer can approve or reject (and optionally modify the data if editable)
|
||||
- On approval: the data flows out through the `approved_data` output pin
|
||||
- On rejection: the data flows out through the `rejected_data` output pin
|
||||
|
||||
**Important:** The output pins yield the actual data itself, NOT status strings.
|
||||
The approval/rejection decision determines WHICH output pin fires, not the value.
|
||||
You do NOT need to compare the output to "APPROVED" or "REJECTED" - simply connect
|
||||
downstream blocks to the appropriate output pin for each case.
|
||||
|
||||
**Example usage:**
|
||||
- Connect `approved_data` → next step in your workflow (data was approved)
|
||||
- Connect `rejected_data` → error handling or notification (data was rejected)
|
||||
This is useful for workflows that require human validation or intervention before
|
||||
proceeding to the next steps.
|
||||
"""
|
||||
|
||||
class Input(BlockSchemaInput):
|
||||
data: Any = SchemaField(
|
||||
description="The data to be reviewed by a human user. "
|
||||
"This exact data will be passed through to either approved_data or "
|
||||
"rejected_data output based on the reviewer's decision."
|
||||
)
|
||||
data: Any = SchemaField(description="The data to be reviewed by a human user")
|
||||
name: str = SchemaField(
|
||||
description="A descriptive name for what this data represents. "
|
||||
"This helps the reviewer understand what they are reviewing.",
|
||||
description="A descriptive name for what this data represents",
|
||||
)
|
||||
editable: bool = SchemaField(
|
||||
description="Whether the human reviewer can edit the data before "
|
||||
"approving or rejecting it",
|
||||
description="Whether the human reviewer can edit the data",
|
||||
default=True,
|
||||
advanced=True,
|
||||
)
|
||||
|
||||
class Output(BlockSchemaOutput):
|
||||
approved_data: Any = SchemaField(
|
||||
description="Outputs the input data when the reviewer APPROVES it. "
|
||||
"The value is the actual data itself (not a status string like 'APPROVED'). "
|
||||
"If the reviewer edited the data, this contains the modified version. "
|
||||
"Connect downstream blocks here for the 'approved' workflow path."
|
||||
description="The data when approved (may be modified by reviewer)"
|
||||
)
|
||||
rejected_data: Any = SchemaField(
|
||||
description="Outputs the input data when the reviewer REJECTS it. "
|
||||
"The value is the actual data itself (not a status string like 'REJECTED'). "
|
||||
"If the reviewer edited the data, this contains the modified version. "
|
||||
"Connect downstream blocks here for the 'rejected' workflow path."
|
||||
description="The data when rejected (may be modified by reviewer)"
|
||||
)
|
||||
review_message: str = SchemaField(
|
||||
description="Optional message provided by the reviewer explaining their "
|
||||
"decision. Only outputs when the reviewer provides a message; "
|
||||
"this pin does not fire if no message was given.",
|
||||
default="",
|
||||
description="Any message provided by the reviewer", default=""
|
||||
)
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
id="8b2a7b3c-6e9d-4a5f-8c1b-2e3f4a5b6c7d",
|
||||
description="Pause execution for human review. Data flows through "
|
||||
"approved_data or rejected_data output based on the reviewer's decision. "
|
||||
"Outputs contain the actual data, not status strings.",
|
||||
description="Pause execution and wait for human approval or modification of data",
|
||||
categories={BlockCategory.BASIC},
|
||||
input_schema=HumanInTheLoopBlock.Input,
|
||||
output_schema=HumanInTheLoopBlock.Output,
|
||||
|
||||
@@ -1,10 +0,0 @@
|
||||
import { parseAsString, useQueryState } from "nuqs";
|
||||
|
||||
export function useCopilotSessionId() {
|
||||
const [urlSessionId, setUrlSessionId] = useQueryState(
|
||||
"sessionId",
|
||||
parseAsString,
|
||||
);
|
||||
|
||||
return { urlSessionId, setUrlSessionId };
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
import { getGetV2GetSessionQueryKey } from "@/app/api/__generated__/endpoints/chat/chat";
|
||||
import { useQueryClient } from "@tanstack/react-query";
|
||||
import type { UIDataTypes, UIMessage, UITools } from "ai";
|
||||
import { useCallback, useEffect, useRef } from "react";
|
||||
import { convertChatSessionMessagesToUiMessages } from "../helpers/convertChatSessionToUiMessages";
|
||||
|
||||
const OPERATING_TYPES = new Set([
|
||||
"operation_started",
|
||||
"operation_pending",
|
||||
"operation_in_progress",
|
||||
]);
|
||||
|
||||
const POLL_INTERVAL_MS = 5_000;
|
||||
|
||||
/**
|
||||
* Detects whether any message contains a tool part whose output indicates
|
||||
* a long-running operation is still in progress.
|
||||
*/
|
||||
function hasOperatingTool(
|
||||
messages: UIMessage<unknown, UIDataTypes, UITools>[],
|
||||
): boolean {
|
||||
for (const msg of messages) {
|
||||
for (const part of msg.parts) {
|
||||
if (!part.type.startsWith("tool-")) continue;
|
||||
const toolPart = part as { output?: unknown };
|
||||
if (!toolPart.output) continue;
|
||||
const output =
|
||||
typeof toolPart.output === "string"
|
||||
? safeParse(toolPart.output)
|
||||
: toolPart.output;
|
||||
if (
|
||||
output &&
|
||||
typeof output === "object" &&
|
||||
"type" in output &&
|
||||
OPERATING_TYPES.has((output as { type: string }).type)
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
function safeParse(value: string): unknown {
|
||||
try {
|
||||
return JSON.parse(value);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Polls the session endpoint while any tool is in an "operating" state
|
||||
* (operation_started / operation_pending / operation_in_progress).
|
||||
*
|
||||
* When the session data shows the tool output has changed (e.g. to
|
||||
* agent_saved), it calls `setMessages` with the updated messages.
|
||||
*/
|
||||
export function useLongRunningToolPolling(
|
||||
sessionId: string | null,
|
||||
messages: UIMessage<unknown, UIDataTypes, UITools>[],
|
||||
setMessages: (
|
||||
updater: (
|
||||
prev: UIMessage<unknown, UIDataTypes, UITools>[],
|
||||
) => UIMessage<unknown, UIDataTypes, UITools>[],
|
||||
) => void,
|
||||
) {
|
||||
const queryClient = useQueryClient();
|
||||
const isPollingRef = useRef(false);
|
||||
const intervalRef = useRef<ReturnType<typeof setInterval> | null>(null);
|
||||
|
||||
const stopPolling = useCallback(() => {
|
||||
if (intervalRef.current) {
|
||||
clearInterval(intervalRef.current);
|
||||
intervalRef.current = null;
|
||||
}
|
||||
isPollingRef.current = false;
|
||||
}, []);
|
||||
|
||||
const poll = useCallback(async () => {
|
||||
if (!sessionId) return;
|
||||
|
||||
// Invalidate the query cache so the next fetch gets fresh data
|
||||
await queryClient.invalidateQueries({
|
||||
queryKey: getGetV2GetSessionQueryKey(sessionId),
|
||||
});
|
||||
|
||||
// Fetch fresh session data
|
||||
const data = queryClient.getQueryData<{
|
||||
status: number;
|
||||
data: { messages?: unknown[] };
|
||||
}>(getGetV2GetSessionQueryKey(sessionId));
|
||||
|
||||
if (data?.status !== 200 || !data.data.messages) return;
|
||||
|
||||
const freshMessages = convertChatSessionMessagesToUiMessages(
|
||||
sessionId,
|
||||
data.data.messages,
|
||||
);
|
||||
|
||||
if (!freshMessages || freshMessages.length === 0) return;
|
||||
|
||||
// Only update if the fresh data no longer has operating tools
|
||||
// (meaning the long-running tool completed)
|
||||
if (!hasOperatingTool(freshMessages)) {
|
||||
setMessages(() => freshMessages);
|
||||
stopPolling();
|
||||
}
|
||||
}, [sessionId, queryClient, setMessages, stopPolling]);
|
||||
|
||||
useEffect(() => {
|
||||
const shouldPoll = hasOperatingTool(messages);
|
||||
|
||||
if (shouldPoll && !isPollingRef.current && sessionId) {
|
||||
isPollingRef.current = true;
|
||||
intervalRef.current = setInterval(() => {
|
||||
poll();
|
||||
}, POLL_INTERVAL_MS);
|
||||
} else if (!shouldPoll && isPollingRef.current) {
|
||||
stopPolling();
|
||||
}
|
||||
|
||||
return () => {
|
||||
// Cleanup on unmount or dependency change
|
||||
if (intervalRef.current) {
|
||||
clearInterval(intervalRef.current);
|
||||
intervalRef.current = null;
|
||||
}
|
||||
};
|
||||
}, [messages, sessionId, poll, stopPolling]);
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import { useChat } from "@ai-sdk/react";
|
||||
import { DefaultChatTransport } from "ai";
|
||||
import { useEffect, useMemo, useState } from "react";
|
||||
import { useChatSession } from "./useChatSession";
|
||||
import { useLongRunningToolPolling } from "./hooks/useLongRunningToolPolling";
|
||||
|
||||
export function useCopilotPage() {
|
||||
const { isUserLoading, isLoggedIn } = useSupabase();
|
||||
@@ -60,6 +61,11 @@ export function useCopilotPage() {
|
||||
});
|
||||
}, [hydratedMessages, setMessages]);
|
||||
|
||||
// Poll session endpoint when a long-running tool (create_agent, edit_agent)
|
||||
// is in progress. When the backend completes, the session data will contain
|
||||
// the final tool output — this hook detects the change and updates messages.
|
||||
useLongRunningToolPolling(sessionId, messages, setMessages);
|
||||
|
||||
// Clear messages when session is null
|
||||
useEffect(() => {
|
||||
if (!sessionId) setMessages([]);
|
||||
|
||||
@@ -61,7 +61,7 @@ Below is a comprehensive list of all available blocks, categorized by their prim
|
||||
| [Get List Item](block-integrations/basic.md#get-list-item) | Returns the element at the given index |
|
||||
| [Get Store Agent Details](block-integrations/system/store_operations.md#get-store-agent-details) | Get detailed information about an agent from the store |
|
||||
| [Get Weather Information](block-integrations/basic.md#get-weather-information) | Retrieves weather information for a specified location using OpenWeatherMap API |
|
||||
| [Human In The Loop](block-integrations/basic.md#human-in-the-loop) | Pause execution for human review |
|
||||
| [Human In The Loop](block-integrations/basic.md#human-in-the-loop) | Pause execution and wait for human approval or modification of data |
|
||||
| [List Is Empty](block-integrations/basic.md#list-is-empty) | Checks if a list is empty |
|
||||
| [List Library Agents](block-integrations/system/library_operations.md#list-library-agents) | List all agents in your personal library |
|
||||
| [Note](block-integrations/basic.md#note) | A visual annotation block that displays a sticky note in the workflow editor for documentation and organization purposes |
|
||||
|
||||
@@ -975,7 +975,7 @@ A travel planning application could use this block to provide users with current
|
||||
## Human In The Loop
|
||||
|
||||
### What it is
|
||||
Pause execution for human review. Data flows through approved_data or rejected_data output based on the reviewer's decision. Outputs contain the actual data, not status strings.
|
||||
Pause execution and wait for human approval or modification of data
|
||||
|
||||
### How it works
|
||||
<!-- MANUAL: how_it_works -->
|
||||
@@ -988,18 +988,18 @@ This enables human oversight at critical points in automated workflows, ensuring
|
||||
|
||||
| Input | Description | Type | Required |
|
||||
|-------|-------------|------|----------|
|
||||
| data | The data to be reviewed by a human user. This exact data will be passed through to either approved_data or rejected_data output based on the reviewer's decision. | Data | Yes |
|
||||
| name | A descriptive name for what this data represents. This helps the reviewer understand what they are reviewing. | str | Yes |
|
||||
| editable | Whether the human reviewer can edit the data before approving or rejecting it | bool | No |
|
||||
| data | The data to be reviewed by a human user | Data | Yes |
|
||||
| name | A descriptive name for what this data represents | str | Yes |
|
||||
| editable | Whether the human reviewer can edit the data | bool | No |
|
||||
|
||||
### Outputs
|
||||
|
||||
| Output | Description | Type |
|
||||
|--------|-------------|------|
|
||||
| error | Error message if the operation failed | str |
|
||||
| approved_data | Outputs the input data when the reviewer APPROVES it. The value is the actual data itself (not a status string like 'APPROVED'). If the reviewer edited the data, this contains the modified version. Connect downstream blocks here for the 'approved' workflow path. | Approved Data |
|
||||
| rejected_data | Outputs the input data when the reviewer REJECTS it. The value is the actual data itself (not a status string like 'REJECTED'). If the reviewer edited the data, this contains the modified version. Connect downstream blocks here for the 'rejected' workflow path. | Rejected Data |
|
||||
| review_message | Optional message provided by the reviewer explaining their decision. Only outputs when the reviewer provides a message; this pin does not fire if no message was given. | str |
|
||||
| approved_data | The data when approved (may be modified by reviewer) | Approved Data |
|
||||
| rejected_data | The data when rejected (may be modified by reviewer) | Rejected Data |
|
||||
| review_message | Any message provided by the reviewer | str |
|
||||
|
||||
### Possible use case
|
||||
<!-- MANUAL: use_case -->
|
||||
|
||||
@@ -16,7 +16,7 @@ When activated, the block:
|
||||
- Install dependencies (npm, pip, etc.)
|
||||
- Run terminal commands
|
||||
- Build and test applications
|
||||
5. Extracts all text and binary files created/modified during execution
|
||||
5. Extracts all text files created/modified during execution
|
||||
6. Returns the response and files, optionally keeping the sandbox alive for follow-up tasks
|
||||
|
||||
The block supports conversation continuation through three mechanisms:
|
||||
@@ -42,7 +42,7 @@ The block supports conversation continuation through three mechanisms:
|
||||
| Output | Description |
|
||||
|--------|-------------|
|
||||
| Response | The output/response from Claude Code execution |
|
||||
| Files | List of files created/modified during execution. Each file includes path, relative_path, name, content, is_binary, and content_base64 fields. For text files, content contains the text and is_binary is False. For binary files (PDFs, images, etc.), is_binary is True and content_base64 contains the base64-encoded data |
|
||||
| Files | List of text files created/modified during execution. Each file includes path, relative_path, name, and content fields |
|
||||
| Conversation History | Full conversation history including this turn. Use to restore context on a fresh sandbox |
|
||||
| Session ID | Session ID for this conversation. Pass back with sandbox_id to continue the conversation |
|
||||
| Sandbox ID | ID of the sandbox instance (null if disposed). Pass back with session_id to continue the conversation |
|
||||
|
||||
@@ -535,7 +535,7 @@ When activated, the block:
|
||||
2. Installs the latest version of Claude Code in the sandbox
|
||||
3. Optionally runs setup commands to prepare the environment
|
||||
4. Executes your prompt using Claude Code, which can create/edit files, install dependencies, run terminal commands, and build applications
|
||||
5. Extracts all text and binary files created/modified during execution
|
||||
5. Extracts all text files created/modified during execution
|
||||
6. Returns the response and files, optionally keeping the sandbox alive for follow-up tasks
|
||||
|
||||
The block supports conversation continuation through three mechanisms:
|
||||
@@ -563,7 +563,7 @@ The block supports conversation continuation through three mechanisms:
|
||||
|--------|-------------|------|
|
||||
| error | Error message if execution failed | str |
|
||||
| response | The output/response from Claude Code execution | str |
|
||||
| files | List of files created/modified by Claude Code during this execution. Each file has 'path', 'relative_path', 'name', 'content', 'is_binary', and 'content_base64' fields. For text files, 'content' contains the text and 'is_binary' is False. For binary files (PDFs, images, etc.), 'is_binary' is True and 'content_base64' contains the base64-encoded data. | List[FileOutput] |
|
||||
| files | List of text files created/modified by Claude Code during this execution. Each file has 'path', 'relative_path', 'name', and 'content' fields. | List[FileOutput] |
|
||||
| conversation_history | Full conversation history including this turn. Pass this to conversation_history input to continue on a fresh sandbox if the previous sandbox timed out. | str |
|
||||
| session_id | Session ID for this conversation. Pass this back along with sandbox_id to continue the conversation. | str |
|
||||
| sandbox_id | ID of the sandbox instance. Pass this back along with session_id to continue the conversation. This is None if dispose_sandbox was True (sandbox was disposed). | str |
|
||||
|
||||
Reference in New Issue
Block a user