Merge branch 'main' into feat/support-agent-sandbox

api/core/app/app_config/entities.py

@@ -1,4 +1,3 @@
-import json
 from collections.abc import Sequence
 from enum import StrEnum, auto
 from typing import Any, Literal
@@ -121,7 +120,7 @@ class VariableEntity(BaseModel):
     allowed_file_types: Sequence[FileType] | None = Field(default_factory=list)
     allowed_file_extensions: Sequence[str] | None = Field(default_factory=list)
     allowed_file_upload_methods: Sequence[FileTransferMethod] | None = Field(default_factory=list)
-    json_schema: str | None = Field(default=None)
+    json_schema: dict | None = Field(default=None)
 
     @field_validator("description", mode="before")
     @classmethod
@@ -135,17 +134,11 @@ class VariableEntity(BaseModel):
 
     @field_validator("json_schema")
     @classmethod
-    def validate_json_schema(cls, schema: str | None) -> str | None:
+    def validate_json_schema(cls, schema: dict | None) -> dict | None:
         if schema is None:
             return None
-
         try:
-            json_schema = json.loads(schema)
-        except json.JSONDecodeError:
-            raise ValueError(f"invalid json_schema value {schema}")
-
-        try:
-            Draft7Validator.check_schema(json_schema)
+            Draft7Validator.check_schema(schema)
         except SchemaError as e:
             raise ValueError(f"Invalid JSON schema: {e.message}")
         return schema

api/core/app/apps/advanced_chat/app_config_manager.py

@@ -26,7 +26,6 @@ class AdvancedChatAppConfigManager(BaseAppConfigManager):
     @classmethod
     def get_app_config(cls, app_model: App, workflow: Workflow) -> AdvancedChatAppConfig:
         features_dict = workflow.features_dict
-
         app_mode = AppMode.value_of(app_model.mode)
         app_config = AdvancedChatAppConfig(
             tenant_id=app_model.tenant_id,

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -471,25 +471,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if node_finish_resp:
             yield node_finish_resp
 
-        # For ANSWER nodes, check if we need to send a message_replace event
-        # Only send if the final output differs from the accumulated task_state.answer
-        # This happens when variables were updated by variable_assigner during workflow execution
-        if event.node_type == NodeType.ANSWER and event.outputs:
-            final_answer = event.outputs.get("answer")
-            if final_answer is not None and final_answer != self._task_state.answer:
-                logger.info(
-                    "ANSWER node final output '%s' differs from accumulated answer '%s', sending message_replace event",
-                    final_answer,
-                    self._task_state.answer,
-                )
-                # Update the task state answer
-                self._task_state.answer = str(final_answer)
-                # Send message_replace event to update the UI
-                yield self._message_cycle_manager.message_replace_to_stream_response(
-                    answer=str(final_answer),
-                    reason="variable_update",
-                )
-
     def _handle_node_failed_events(
         self,
         event: Union[QueueNodeFailedEvent, QueueNodeExceptionEvent],

api/core/app/apps/base_app_generator.py

@@ -1,4 +1,3 @@
-import json
 from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Union, final
 
@@ -76,12 +75,24 @@ class BaseAppGenerator:
         user_inputs = {**user_inputs, **files_inputs, **file_list_inputs}
 
         # Check if all files are converted to File
-        if any(filter(lambda v: isinstance(v, dict), user_inputs.values())):
-            raise ValueError("Invalid input type")
-        if any(
-            filter(lambda v: isinstance(v, dict), filter(lambda item: isinstance(item, list), user_inputs.values()))
-        ):
-            raise ValueError("Invalid input type")
+        invalid_dict_keys = [
+            k
+            for k, v in user_inputs.items()
+            if isinstance(v, dict)
+            and entity_dictionary[k].type not in {VariableEntityType.FILE, VariableEntityType.JSON_OBJECT}
+        ]
+        if invalid_dict_keys:
+            raise ValueError(f"Invalid input type for {invalid_dict_keys}")
+
+        invalid_list_dict_keys = [
+            k
+            for k, v in user_inputs.items()
+            if isinstance(v, list)
+            and any(isinstance(item, dict) for item in v)
+            and entity_dictionary[k].type != VariableEntityType.FILE_LIST
+        ]
+        if invalid_list_dict_keys:
+            raise ValueError(f"Invalid input type for {invalid_list_dict_keys}")
 
         return user_inputs
 
@@ -178,12 +189,8 @@ class BaseAppGenerator:
                     elif value == 0:
                         value = False
             case VariableEntityType.JSON_OBJECT:
-                if not isinstance(value, str):
-                    raise ValueError(f"{variable_entity.variable} in input form must be a string")
-                try:
-                    json.loads(value)
-                except json.JSONDecodeError:
-                    raise ValueError(f"{variable_entity.variable} in input form must be a valid JSON object")
+                if not isinstance(value, dict):
+                    raise ValueError(f"{variable_entity.variable} in input form must be a dict")
             case _:
                 raise AssertionError("this statement should be unreachable.")
 

api/core/workflow/nodes/start/start_node.py

@@ -1,4 +1,3 @@
-import json
 from typing import Any
 
 from jsonschema import Draft7Validator, ValidationError
@@ -43,25 +42,22 @@ class StartNode(Node[StartNodeData]):
             if value is None and variable.required:
                 raise ValueError(f"{key} is required in input form")
 
+            # If no value provided, skip further processing for this key
+            if not value:
+                continue
+
+            if not isinstance(value, dict):
+                raise ValueError(f"JSON object for '{key}' must be an object")
+
+            # Overwrite with normalized dict to ensure downstream consistency
+            node_inputs[key] = value
+
+            # If schema exists, then validate against it
             schema = variable.json_schema
             if not schema:
                 continue
 
-            if not value:
-                continue
-
             try:
-                json_schema = json.loads(schema)
-            except json.JSONDecodeError as e:
-                raise ValueError(f"{schema} must be a valid JSON object")
-
-            try:
-                json_value = json.loads(value)
-            except json.JSONDecodeError as e:
-                raise ValueError(f"{value} must be a valid JSON object")
-
-            try:
-                Draft7Validator(json_schema).validate(json_value)
+                Draft7Validator(schema).validate(value)
             except ValidationError as e:
                 raise ValueError(f"JSON object for '{key}' does not match schema: {e.message}")
-            node_inputs[key] = json_value

api/tests/unit_tests/core/app/apps/advanced_chat/test_generate_task_pipeline_answer_node.py

@@ -1,390 +0,0 @@
-"""
-Tests for AdvancedChatAppGenerateTaskPipeline._handle_node_succeeded_event method,
-specifically testing the ANSWER node message_replace logic.
-"""
-
-from datetime import datetime
-from types import SimpleNamespace
-from unittest.mock import MagicMock, Mock, patch
-
-import pytest
-
-from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity
-from core.app.entities.queue_entities import QueueNodeSucceededEvent
-from core.workflow.enums import NodeType
-from models import EndUser
-from models.model import AppMode
-
-
-class TestAnswerNodeMessageReplace:
-    """Test cases for ANSWER node message_replace event logic."""
-
-    @pytest.fixture
-    def mock_application_generate_entity(self):
-        """Create a mock application generate entity."""
-        entity = Mock(spec=AdvancedChatAppGenerateEntity)
-        entity.task_id = "test-task-id"
-        entity.app_id = "test-app-id"
-        entity.workflow_run_id = "test-workflow-run-id"
-        # minimal app_config used by pipeline internals
-        entity.app_config = SimpleNamespace(
-            tenant_id="test-tenant-id",
-            app_id="test-app-id",
-            app_mode=AppMode.ADVANCED_CHAT,
-            app_model_config_dict={},
-            additional_features=None,
-            sensitive_word_avoidance=None,
-        )
-        entity.query = "test query"
-        entity.files = []
-        entity.extras = {}
-        entity.trace_manager = None
-        entity.inputs = {}
-        entity.invoke_from = "debugger"
-        return entity
-
-    @pytest.fixture
-    def mock_workflow(self):
-        """Create a mock workflow."""
-        workflow = Mock()
-        workflow.id = "test-workflow-id"
-        workflow.features_dict = {}
-        return workflow
-
-    @pytest.fixture
-    def mock_queue_manager(self):
-        """Create a mock queue manager."""
-        manager = Mock()
-        manager.listen.return_value = []
-        manager.graph_runtime_state = None
-        return manager
-
-    @pytest.fixture
-    def mock_conversation(self):
-        """Create a mock conversation."""
-        conversation = Mock()
-        conversation.id = "test-conversation-id"
-        conversation.mode = "advanced_chat"
-        return conversation
-
-    @pytest.fixture
-    def mock_message(self):
-        """Create a mock message."""
-        message = Mock()
-        message.id = "test-message-id"
-        message.query = "test query"
-        message.created_at = Mock()
-        message.created_at.timestamp.return_value = 1234567890
-        return message
-
-    @pytest.fixture
-    def mock_user(self):
-        """Create a mock end user."""
-        user = MagicMock(spec=EndUser)
-        user.id = "test-user-id"
-        user.session_id = "test-session-id"
-        return user
-
-    @pytest.fixture
-    def mock_draft_var_saver_factory(self):
-        """Create a mock draft variable saver factory."""
-        return Mock()
-
-    @pytest.fixture
-    def pipeline(
-        self,
-        mock_application_generate_entity,
-        mock_workflow,
-        mock_queue_manager,
-        mock_conversation,
-        mock_message,
-        mock_user,
-        mock_draft_var_saver_factory,
-    ):
-        """Create an AdvancedChatAppGenerateTaskPipeline instance with mocked dependencies."""
-        from core.app.apps.advanced_chat.generate_task_pipeline import AdvancedChatAppGenerateTaskPipeline
-
-        with patch("core.app.apps.advanced_chat.generate_task_pipeline.db"):
-            pipeline = AdvancedChatAppGenerateTaskPipeline(
-                application_generate_entity=mock_application_generate_entity,
-                workflow=mock_workflow,
-                queue_manager=mock_queue_manager,
-                conversation=mock_conversation,
-                message=mock_message,
-                user=mock_user,
-                stream=True,
-                dialogue_count=1,
-                draft_var_saver_factory=mock_draft_var_saver_factory,
-            )
-            # Initialize workflow run id to avoid validation errors
-            pipeline._workflow_run_id = "test-workflow-run-id"
-            # Mock the message cycle manager methods we need to track
-            pipeline._message_cycle_manager.message_replace_to_stream_response = Mock()
-            return pipeline
-
-    def test_answer_node_with_different_output_sends_message_replace(self, pipeline, mock_application_generate_entity):
-        """
-        Test that when an ANSWER node's final output differs from accumulated answer,
-        a message_replace event is sent.
-        """
-        # Arrange: Set initial accumulated answer
-        pipeline._task_state.answer = "initial answer"
-
-        # Create ANSWER node succeeded event with different final output
-        event = QueueNodeSucceededEvent(
-            node_execution_id="test-node-execution-id",
-            node_id="test-answer-node",
-            node_type=NodeType.ANSWER,
-            start_at=datetime.now(),
-            outputs={"answer": "updated final answer"},
-        )
-
-        # Mock the workflow response converter to avoid extra processing
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        responses = list(pipeline._handle_node_succeeded_event(event))
-
-        # Assert
-        assert pipeline._task_state.answer == "updated final answer"
-        # Verify message_replace was called
-        pipeline._message_cycle_manager.message_replace_to_stream_response.assert_called_once_with(
-            answer="updated final answer", reason="variable_update"
-        )
-
-    def test_answer_node_with_same_output_does_not_send_message_replace(self, pipeline):
-        """
-        Test that when an ANSWER node's final output is the same as accumulated answer,
-        no message_replace event is sent.
-        """
-        # Arrange: Set initial accumulated answer
-        pipeline._task_state.answer = "same answer"
-
-        # Create ANSWER node succeeded event with same output
-        event = QueueNodeSucceededEvent(
-            node_execution_id="test-node-execution-id",
-            node_id="test-answer-node",
-            node_type=NodeType.ANSWER,
-            start_at=datetime.now(),
-            outputs={"answer": "same answer"},
-        )
-
-        # Mock the workflow response converter
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        list(pipeline._handle_node_succeeded_event(event))
-
-        # Assert: answer should remain unchanged
-        assert pipeline._task_state.answer == "same answer"
-        # Verify message_replace was NOT called
-        pipeline._message_cycle_manager.message_replace_to_stream_response.assert_not_called()
-
-    def test_answer_node_with_none_output_does_not_send_message_replace(self, pipeline):
-        """
-        Test that when an ANSWER node's output is None or missing 'answer' key,
-        no message_replace event is sent.
-        """
-        # Arrange: Set initial accumulated answer
-        pipeline._task_state.answer = "existing answer"
-
-        # Create ANSWER node succeeded event with None output
-        event = QueueNodeSucceededEvent(
-            node_execution_id="test-node-execution-id",
-            node_id="test-answer-node",
-            node_type=NodeType.ANSWER,
-            start_at=datetime.now(),
-            outputs={"answer": None},
-        )
-
-        # Mock the workflow response converter
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        list(pipeline._handle_node_succeeded_event(event))
-
-        # Assert: answer should remain unchanged
-        assert pipeline._task_state.answer == "existing answer"
-        # Verify message_replace was NOT called
-        pipeline._message_cycle_manager.message_replace_to_stream_response.assert_not_called()
-
-    def test_answer_node_with_empty_outputs_does_not_send_message_replace(self, pipeline):
-        """
-        Test that when an ANSWER node has empty outputs dict,
-        no message_replace event is sent.
-        """
-        # Arrange: Set initial accumulated answer
-        pipeline._task_state.answer = "existing answer"
-
-        # Create ANSWER node succeeded event with empty outputs
-        event = QueueNodeSucceededEvent(
-            node_execution_id="test-node-execution-id",
-            node_id="test-answer-node",
-            node_type=NodeType.ANSWER,
-            start_at=datetime.now(),
-            outputs={},
-        )
-
-        # Mock the workflow response converter
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        list(pipeline._handle_node_succeeded_event(event))
-
-        # Assert: answer should remain unchanged
-        assert pipeline._task_state.answer == "existing answer"
-        # Verify message_replace was NOT called
-        pipeline._message_cycle_manager.message_replace_to_stream_response.assert_not_called()
-
-    def test_answer_node_with_no_answer_key_in_outputs(self, pipeline):
-        """
-        Test that when an ANSWER node's outputs don't contain 'answer' key,
-        no message_replace event is sent.
-        """
-        # Arrange: Set initial accumulated answer
-        pipeline._task_state.answer = "existing answer"
-
-        # Create ANSWER node succeeded event without 'answer' key in outputs
-        event = QueueNodeSucceededEvent(
-            node_execution_id="test-node-execution-id",
-            node_id="test-answer-node",
-            node_type=NodeType.ANSWER,
-            start_at=datetime.now(),
-            outputs={"other_key": "some value"},
-        )
-
-        # Mock the workflow response converter
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        list(pipeline._handle_node_succeeded_event(event))
-
-        # Assert: answer should remain unchanged
-        assert pipeline._task_state.answer == "existing answer"
-        # Verify message_replace was NOT called
-        pipeline._message_cycle_manager.message_replace_to_stream_response.assert_not_called()
-
-    def test_non_answer_node_does_not_send_message_replace(self, pipeline):
-        """
-        Test that non-ANSWER nodes (e.g., LLM, END) don't trigger message_replace events.
-        """
-        # Arrange: Set initial accumulated answer
-        pipeline._task_state.answer = "existing answer"
-
-        # Test with LLM node
-        llm_event = QueueNodeSucceededEvent(
-            node_execution_id="test-llm-execution-id",
-            node_id="test-llm-node",
-            node_type=NodeType.LLM,
-            start_at=datetime.now(),
-            outputs={"answer": "different answer"},
-        )
-
-        # Mock the workflow response converter
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        list(pipeline._handle_node_succeeded_event(llm_event))
-
-        # Assert: answer should remain unchanged
-        assert pipeline._task_state.answer == "existing answer"
-        # Verify message_replace was NOT called
-        pipeline._message_cycle_manager.message_replace_to_stream_response.assert_not_called()
-
-    def test_end_node_does_not_send_message_replace(self, pipeline):
-        """
-        Test that END nodes don't trigger message_replace events even with 'answer' output.
-        """
-        # Arrange: Set initial accumulated answer
-        pipeline._task_state.answer = "existing answer"
-
-        # Create END node succeeded event with answer output
-        event = QueueNodeSucceededEvent(
-            node_execution_id="test-end-execution-id",
-            node_id="test-end-node",
-            node_type=NodeType.END,
-            start_at=datetime.now(),
-            outputs={"answer": "different answer"},
-        )
-
-        # Mock the workflow response converter
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        list(pipeline._handle_node_succeeded_event(event))
-
-        # Assert: answer should remain unchanged
-        assert pipeline._task_state.answer == "existing answer"
-        # Verify message_replace was NOT called
-        pipeline._message_cycle_manager.message_replace_to_stream_response.assert_not_called()
-
-    def test_answer_node_with_numeric_output_converts_to_string(self, pipeline):
-        """
-        Test that when an ANSWER node's final output is numeric,
-        it gets converted to string properly.
-        """
-        # Arrange: Set initial accumulated answer
-        pipeline._task_state.answer = "text answer"
-
-        # Create ANSWER node succeeded event with numeric output
-        event = QueueNodeSucceededEvent(
-            node_execution_id="test-node-execution-id",
-            node_id="test-answer-node",
-            node_type=NodeType.ANSWER,
-            start_at=datetime.now(),
-            outputs={"answer": 12345},
-        )
-
-        # Mock the workflow response converter
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        list(pipeline._handle_node_succeeded_event(event))
-
-        # Assert: answer should be converted to string
-        assert pipeline._task_state.answer == "12345"
-        # Verify message_replace was called with string
-        pipeline._message_cycle_manager.message_replace_to_stream_response.assert_called_once_with(
-            answer="12345", reason="variable_update"
-        )
-
-    def test_answer_node_files_are_recorded(self, pipeline):
-        """
-        Test that ANSWER nodes properly record files from outputs.
-        """
-        # Arrange
-        pipeline._task_state.answer = "existing answer"
-
-        # Create ANSWER node succeeded event with files
-        event = QueueNodeSucceededEvent(
-            node_execution_id="test-node-execution-id",
-            node_id="test-answer-node",
-            node_type=NodeType.ANSWER,
-            start_at=datetime.now(),
-            outputs={
-                "answer": "same answer",
-                "files": [
-                    {"type": "image", "transfer_method": "remote_url", "remote_url": "http://example.com/img.png"}
-                ],
-            },
-        )
-
-        # Mock the workflow response converter
-        pipeline._workflow_response_converter.fetch_files_from_node_outputs = Mock(return_value=event.outputs["files"])
-        pipeline._workflow_response_converter.workflow_node_finish_to_stream_response = Mock(return_value=None)
-        pipeline._save_output_for_event = Mock()
-
-        # Act
-        list(pipeline._handle_node_succeeded_event(event))
-
-        # Assert: files should be recorded
-        assert len(pipeline._recorded_files) == 1
-        assert pipeline._recorded_files[0] == event.outputs["files"][0]

api/tests/unit_tests/core/workflow/nodes/test_start_node_json_object.py

@@ -58,6 +58,8 @@ def test_json_object_valid_schema():
         }
     )
 
+    schema = json.loads(schema)
+
     variables = [
         VariableEntity(
             variable="profile",
@@ -68,7 +70,7 @@ def test_json_object_valid_schema():
         )
     ]
 
-    user_inputs = {"profile": json.dumps({"age": 20, "name": "Tom"})}
+    user_inputs = {"profile": {"age": 20, "name": "Tom"}}
 
     node = make_start_node(user_inputs, variables)
     result = node._run()
@@ -87,6 +89,8 @@ def test_json_object_invalid_json_string():
             "required": ["age", "name"],
         }
     )
+
+    schema = json.loads(schema)
     variables = [
         VariableEntity(
             variable="profile",
@@ -97,12 +101,12 @@ def test_json_object_invalid_json_string():
         )
     ]
 
-    # Missing closing brace makes this invalid JSON
+    # Providing a string instead of an object should raise a type error
     user_inputs = {"profile": '{"age": 20, "name": "Tom"'}
 
     node = make_start_node(user_inputs, variables)
 
-    with pytest.raises(ValueError, match='{"age": 20, "name": "Tom" must be a valid JSON object'):
+    with pytest.raises(ValueError, match="JSON object for 'profile' must be an object"):
         node._run()
 
 
@@ -118,6 +122,8 @@ def test_json_object_does_not_match_schema():
         }
     )
 
+    schema = json.loads(schema)
+
     variables = [
         VariableEntity(
             variable="profile",
@@ -129,7 +135,7 @@ def test_json_object_does_not_match_schema():
     ]
 
     # age is a string, which violates the schema (expects number)
-    user_inputs = {"profile": json.dumps({"age": "twenty", "name": "Tom"})}
+    user_inputs = {"profile": {"age": "twenty", "name": "Tom"}}
 
     node = make_start_node(user_inputs, variables)
 
@@ -149,6 +155,8 @@ def test_json_object_missing_required_schema_field():
         }
     )
 
+    schema = json.loads(schema)
+
     variables = [
         VariableEntity(
             variable="profile",
@@ -160,7 +168,7 @@ def test_json_object_missing_required_schema_field():
     ]
 
     # Missing required field "name"
-    user_inputs = {"profile": json.dumps({"age": 20})}
+    user_inputs = {"profile": {"age": 20}}
 
     node = make_start_node(user_inputs, variables)