WIP: test(api): tests for truncation logic

api/tests/integration_tests/services/test_workflow_draft_variable_service.py

@@ -3,16 +3,27 @@ import unittest
 import uuid
 
 import pytest
+from sqlalchemy import delete
 from sqlalchemy.orm import Session
 
+from core.variables.segments import StringSegment
+from core.variables.types import SegmentType
 from core.variables.variables import StringVariable
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from core.workflow.nodes import NodeType
+from extensions.ext_storage import storage
 from factories.variable_factory import build_segment
 from libs import datetime_utils
 from models import db
-from models.workflow import Workflow, WorkflowDraftVariable, WorkflowNodeExecutionModel
-from services.workflow_draft_variable_service import DraftVarLoader, VariableResetError, WorkflowDraftVariableService
+from models.enums import CreatorUserRole
+from models.model import UploadFile
+from models.workflow import Workflow, WorkflowDraftVariable, WorkflowDraftVariableFile, WorkflowNodeExecutionModel
+from services.workflow_draft_variable_service import (
+    DraftVariableSaver,
+    DraftVarLoader,
+    VariableResetError,
+    WorkflowDraftVariableService,
+)
 
 
 @pytest.mark.usefixtures("flask_req_ctx")
@@ -175,6 +186,23 @@ class TestDraftVariableLoader(unittest.TestCase):
     _node1_id = "test_loader_node_1"
     _node_exec_id = str(uuid.uuid4())
 
+    # @pytest.fixture
+    # def test_app_id(self):
+    #     return str(uuid.uuid4())
+
+    # @pytest.fixture
+    # def test_tenant_id(self):
+    #     return str(uuid.uuid4())
+
+    # @pytest.fixture
+    # def session(self):
+    #     with Session(bind=db.engine, expire_on_commit=False) as session:
+    #         yield session
+
+    # @pytest.fixture
+    # def node_var(self, session):
+    #     pass
+
     def setUp(self):
         self._test_app_id = str(uuid.uuid4())
         self._test_tenant_id = str(uuid.uuid4())
@@ -241,6 +269,246 @@ class TestDraftVariableLoader(unittest.TestCase):
         node1_var = next(v for v in variables if v.selector[0] == self._node1_id)
         assert node1_var.id == self._node_var_id
 
+    @pytest.mark.usefixtures("setup_account")
+    def test_load_offloaded_variable_string_type_integration(self, setup_account):
+        """Test _load_offloaded_variable with string type using DraftVariableSaver for data creation."""
+
+        # Create a large string that will be offloaded
+        test_content = "x" * 15000  # Create a string larger than LARGE_VARIABLE_THRESHOLD (10KB)
+        large_string_segment = StringSegment(value=test_content)
+
+        node_execution_id = str(uuid.uuid4())
+
+        try:
+            with Session(bind=db.engine, expire_on_commit=False) as session:
+                # Use DraftVariableSaver to create offloaded variable (this mimics production)
+                saver = DraftVariableSaver(
+                    session=session,
+                    app_id=self._test_app_id,
+                    node_id="test_offload_node",
+                    node_type=NodeType.LLM,  # Use a real node type
+                    node_execution_id=node_execution_id,
+                    user=setup_account,
+                )
+
+                # Save the variable - this will trigger offloading due to large size
+                saver.save(outputs={"offloaded_string_var": large_string_segment})
+                session.commit()
+
+                # Now test loading using DraftVarLoader
+                var_loader = DraftVarLoader(engine=db.engine, app_id=self._test_app_id, tenant_id=self._test_tenant_id)
+
+                # Load the variable using the standard workflow
+                variables = var_loader.load_variables([["test_offload_node", "offloaded_string_var"]])
+
+                # Verify results
+                assert len(variables) == 1
+                loaded_variable = variables[0]
+                assert loaded_variable.name == "offloaded_string_var"
+                assert loaded_variable.selector == ["test_offload_node", "offloaded_string_var"]
+                assert isinstance(loaded_variable.value, StringSegment)
+                assert loaded_variable.value.value == test_content
+
+        finally:
+            # Clean up - delete all draft variables for this app
+            with Session(bind=db.engine) as session:
+                service = WorkflowDraftVariableService(session)
+                service.delete_workflow_variables(self._test_app_id)
+                session.commit()
+
+    def test_load_offloaded_variable_object_type_integration(self):
+        """Test _load_offloaded_variable with object type using real storage and service."""
+
+        # Create a test object
+        test_object = {"key1": "value1", "key2": 42, "nested": {"inner": "data"}}
+        test_json = json.dumps(test_object, ensure_ascii=False, separators=(",", ":"))
+        content_bytes = test_json.encode()
+
+        # Create an upload file record
+        upload_file = UploadFile(
+            tenant_id=self._test_tenant_id,
+            storage_type="local",
+            key=f"test_offload_{uuid.uuid4()}.json",
+            name="test_offload.json",
+            size=len(content_bytes),
+            extension="json",
+            mime_type="application/json",
+            created_by_role=CreatorUserRole.ACCOUNT,
+            created_by=str(uuid.uuid4()),
+            created_at=datetime_utils.naive_utc_now(),
+            used=True,
+            used_by=str(uuid.uuid4()),
+            used_at=datetime_utils.naive_utc_now(),
+        )
+
+        # Store the content in storage
+        storage.save(upload_file.key, content_bytes)
+
+        # Create a variable file record
+        variable_file = WorkflowDraftVariableFile(
+            upload_file_id=upload_file.id,
+            value_type=SegmentType.OBJECT,
+            tenant_id=self._test_tenant_id,
+            app_id=self._test_app_id,
+            user_id=str(uuid.uuid4()),
+            size=len(content_bytes),
+            created_at=datetime_utils.naive_utc_now(),
+        )
+
+        try:
+            with Session(bind=db.engine, expire_on_commit=False) as session:
+                # Add upload file and variable file first to get their IDs
+                session.add_all([upload_file, variable_file])
+                session.flush()  # This generates the IDs
+
+                # Now create the offloaded draft variable with the correct file_id
+                offloaded_var = WorkflowDraftVariable.new_node_variable(
+                    app_id=self._test_app_id,
+                    node_id="test_offload_node",
+                    name="offloaded_object_var",
+                    value=build_segment({"truncated": True}),
+                    visible=True,
+                    node_execution_id=str(uuid.uuid4()),
+                )
+                offloaded_var.file_id = variable_file.id
+
+                session.add(offloaded_var)
+                session.flush()
+                session.commit()
+
+                # Use the service method that properly preloads relationships
+                service = WorkflowDraftVariableService(session)
+                draft_vars = service.get_draft_variables_by_selectors(
+                    self._test_app_id, [["test_offload_node", "offloaded_object_var"]]
+                )
+
+                assert len(draft_vars) == 1
+                loaded_var = draft_vars[0]
+                assert loaded_var.is_truncated()
+
+                # Create DraftVarLoader and test loading
+                var_loader = DraftVarLoader(engine=db.engine, app_id=self._test_app_id, tenant_id=self._test_tenant_id)
+
+                # Test the _load_offloaded_variable method
+                selector_tuple, variable = var_loader._load_offloaded_variable(loaded_var)
+
+                # Verify the results
+                assert selector_tuple == ("test_offload_node", "offloaded_object_var")
+                assert variable.id == loaded_var.id
+                assert variable.name == "offloaded_object_var"
+                assert variable.value.value == test_object
+
+        finally:
+            # Clean up
+            with Session(bind=db.engine) as session:
+                # Query and delete by ID to ensure they're tracked in this session
+                session.query(WorkflowDraftVariable).filter_by(id=offloaded_var.id).delete()
+                session.query(WorkflowDraftVariableFile).filter_by(id=variable_file.id).delete()
+                session.query(UploadFile).filter_by(id=upload_file.id).delete()
+                session.commit()
+            # Clean up storage
+            try:
+                storage.delete(upload_file.key)
+            except Exception:
+                pass  # Ignore cleanup failures
+
+    def test_load_variables_with_offloaded_variables_integration(self):
+        """Test load_variables method with mix of regular and offloaded variables using real storage."""
+        # Create a regular variable (already exists from setUp)
+        # Create offloaded variable content
+        test_content = "This is offloaded content for integration test"
+        content_bytes = test_content.encode()
+
+        # Create upload file record
+        upload_file = UploadFile(
+            tenant_id=self._test_tenant_id,
+            storage_type="local",
+            key=f"test_integration_{uuid.uuid4()}.txt",
+            name="test_integration.txt",
+            size=len(content_bytes),
+            extension="txt",
+            mime_type="text/plain",
+            created_by_role=CreatorUserRole.ACCOUNT,
+            created_by=str(uuid.uuid4()),
+            created_at=datetime_utils.naive_utc_now(),
+            used=True,
+            used_by=str(uuid.uuid4()),
+            used_at=datetime_utils.naive_utc_now(),
+        )
+
+        # Store the content
+        storage.save(upload_file.key, content_bytes)
+
+        # Create variable file
+        variable_file = WorkflowDraftVariableFile(
+            upload_file_id=upload_file.id,
+            value_type=SegmentType.STRING,
+            tenant_id=self._test_tenant_id,
+            app_id=self._test_app_id,
+            user_id=str(uuid.uuid4()),
+            size=len(content_bytes),
+            created_at=datetime_utils.naive_utc_now(),
+        )
+
+        try:
+            with Session(bind=db.engine, expire_on_commit=False) as session:
+                # Add upload file and variable file first to get their IDs
+                session.add_all([upload_file, variable_file])
+                session.flush()  # This generates the IDs
+
+                # Now create the offloaded draft variable with the correct file_id
+                offloaded_var = WorkflowDraftVariable.new_node_variable(
+                    app_id=self._test_app_id,
+                    node_id="test_integration_node",
+                    name="offloaded_integration_var",
+                    value=build_segment("truncated"),
+                    visible=True,
+                    node_execution_id=str(uuid.uuid4()),
+                )
+                offloaded_var.file_id = variable_file.id
+
+                session.add(offloaded_var)
+                session.flush()
+                session.commit()
+
+                # Test load_variables with both regular and offloaded variables
+                # This method should handle the relationship preloading internally
+                var_loader = DraftVarLoader(engine=db.engine, app_id=self._test_app_id, tenant_id=self._test_tenant_id)
+
+                variables = var_loader.load_variables(
+                    [
+                        [SYSTEM_VARIABLE_NODE_ID, "sys_var"],  # Regular variable from setUp
+                        ["test_integration_node", "offloaded_integration_var"],  # Offloaded variable
+                    ]
+                )
+
+                # Verify results
+                assert len(variables) == 2
+
+                # Find regular variable
+                regular_var = next(v for v in variables if v.selector[0] == SYSTEM_VARIABLE_NODE_ID)
+                assert regular_var.id == self._sys_var_id
+                assert regular_var.value == "sys_value"
+
+                # Find offloaded variable
+                offloaded_loaded_var = next(v for v in variables if v.selector[0] == "test_integration_node")
+                assert offloaded_loaded_var.id == offloaded_var.id
+                assert offloaded_loaded_var.value == test_content
+
+        finally:
+            # Clean up
+            with Session(bind=db.engine) as session:
+                # Query and delete by ID to ensure they're tracked in this session
+                session.query(WorkflowDraftVariable).filter_by(id=offloaded_var.id).delete()
+                session.query(WorkflowDraftVariableFile).filter_by(id=variable_file.id).delete()
+                session.query(UploadFile).filter_by(id=upload_file.id).delete()
+                session.commit()
+            # Clean up storage
+            try:
+                storage.delete(upload_file.key)
+            except Exception:
+                pass  # Ignore cleanup failures
+
 
 @pytest.mark.usefixtures("flask_req_ctx")
 class TestWorkflowDraftVariableServiceResetVariable(unittest.TestCase):
@@ -272,7 +540,7 @@ class TestWorkflowDraftVariableServiceResetVariable(unittest.TestCase):
             triggered_from="workflow-run",
             workflow_run_id=str(uuid.uuid4()),
             index=1,
-            node_execution_id=self._node_exec_id,
+            node_execution_id=str(uuid.uuid4()),
             node_id=self._node_id,
             node_type=NodeType.LLM.value,
             title="Test Node",
@@ -281,7 +549,7 @@ class TestWorkflowDraftVariableServiceResetVariable(unittest.TestCase):
             outputs='{"test_var": "output_value", "other_var": "other_output"}',
             status="succeeded",
             elapsed_time=1.5,
-            created_by_role="account",
+            created_by_role=CreatorUserRole.ACCOUNT,
             created_by=str(uuid.uuid4()),
         )
 
@@ -336,10 +604,14 @@ class TestWorkflowDraftVariableServiceResetVariable(unittest.TestCase):
         )
         self._conv_var.last_edited_at = datetime_utils.naive_utc_now()
 
+        with Session(db.engine, expire_on_commit=False) as persistent_session, persistent_session.begin():
+            persistent_session.add(
+                self._workflow_node_execution,
+            )
+
         # Add all to database
         db.session.add_all(
             [
-                self._workflow_node_execution,
                 self._node_var_with_exec,
                 self._node_var_without_exec,
                 self._node_var_missing_exec,
@@ -354,6 +626,14 @@ class TestWorkflowDraftVariableServiceResetVariable(unittest.TestCase):
         self._node_var_missing_exec_id = self._node_var_missing_exec.id
         self._conv_var_id = self._conv_var.id
 
+    def tearDown(self):
+        self._session.rollback()
+        with Session(db.engine) as session, session.begin():
+            stmt = delete(WorkflowNodeExecutionModel).where(
+                WorkflowNodeExecutionModel.id == self._workflow_node_execution.id
+            )
+            session.execute(stmt)
+
     def _get_test_srv(self) -> WorkflowDraftVariableService:
         return WorkflowDraftVariableService(session=self._session)
 
@@ -380,9 +660,6 @@ class TestWorkflowDraftVariableServiceResetVariable(unittest.TestCase):
         )
         return workflow
 
-    def tearDown(self):
-        self._session.rollback()
-
     def test_reset_node_variable_with_valid_execution_record(self):
         """Test resetting a node variable with valid execution record - should restore from execution"""
         srv = self._get_test_srv()

api/tests/integration_tests/tasks/test_remove_app_and_related_data_task.py

@@ -1,12 +1,14 @@
 import uuid
+from unittest.mock import patch
 
 import pytest
 from sqlalchemy import delete
 
 from core.variables.segments import StringSegment
 from models import Tenant, db
-from models.model import App
-from models.workflow import WorkflowDraftVariable
+from models.enums import CreatorUserRole
+from models.model import App, UploadFile
+from models.workflow import WorkflowDraftVariable, WorkflowDraftVariableFile
 from tasks.remove_app_and_related_data_task import _delete_draft_variables, delete_draft_variables_batch
 
 
@@ -212,3 +214,255 @@ class TestDeleteDraftVariablesIntegration:
                 .execution_options(synchronize_session=False)
             )
             db.session.execute(query)
+
+
+class TestDeleteDraftVariablesWithOffloadIntegration:
+    """Integration tests for draft variable deletion with Offload data."""
+
+    @pytest.fixture
+    def setup_offload_test_data(self, app_and_tenant):
+        """Create test data with draft variables that have associated Offload files."""
+        tenant, app = app_and_tenant
+
+        # Create UploadFile records
+        from libs.datetime_utils import naive_utc_now
+
+        upload_file1 = UploadFile(
+            tenant_id=tenant.id,
+            storage_type="local",
+            key="test/file1.json",
+            name="file1.json",
+            size=1024,
+            extension="json",
+            mime_type="application/json",
+            created_by_role=CreatorUserRole.ACCOUNT,
+            created_by=str(uuid.uuid4()),
+            created_at=naive_utc_now(),
+            used=False,
+        )
+        upload_file2 = UploadFile(
+            tenant_id=tenant.id,
+            storage_type="local",
+            key="test/file2.json",
+            name="file2.json",
+            size=2048,
+            extension="json",
+            mime_type="application/json",
+            created_by_role=CreatorUserRole.ACCOUNT,
+            created_by=str(uuid.uuid4()),
+            created_at=naive_utc_now(),
+            used=False,
+        )
+        db.session.add(upload_file1)
+        db.session.add(upload_file2)
+        db.session.flush()
+
+        # Create WorkflowDraftVariableFile records
+        from core.variables.types import SegmentType
+        var_file1 = WorkflowDraftVariableFile(
+            tenant_id=tenant.id,
+            app_id=app.id,
+            user_id=str(uuid.uuid4()),
+            upload_file_id=upload_file1.id,
+            size=1024,
+            length=10,
+            value_type=SegmentType.STRING,
+        )
+        var_file2 = WorkflowDraftVariableFile(
+            tenant_id=tenant.id,
+            app_id=app.id,
+            user_id=str(uuid.uuid4()),
+            upload_file_id=upload_file2.id,
+            size=2048,
+            length=20,
+            value_type=SegmentType.OBJECT,
+        )
+        db.session.add(var_file1)
+        db.session.add(var_file2)
+        db.session.flush()
+
+        # Create WorkflowDraftVariable records with file associations
+        draft_var1 = WorkflowDraftVariable.new_node_variable(
+            app_id=app.id,
+            node_id="node_1",
+            name="large_var_1",
+            value=StringSegment(value="truncated..."),
+            node_execution_id=str(uuid.uuid4()),
+            file_id=var_file1.id,
+        )
+        draft_var2 = WorkflowDraftVariable.new_node_variable(
+            app_id=app.id,
+            node_id="node_2",
+            name="large_var_2",
+            value=StringSegment(value="truncated..."),
+            node_execution_id=str(uuid.uuid4()),
+            file_id=var_file2.id,
+        )
+        # Create a regular variable without Offload data
+        draft_var3 = WorkflowDraftVariable.new_node_variable(
+            app_id=app.id,
+            node_id="node_3",
+            name="regular_var",
+            value=StringSegment(value="regular_value"),
+            node_execution_id=str(uuid.uuid4()),
+        )
+
+        db.session.add(draft_var1)
+        db.session.add(draft_var2)
+        db.session.add(draft_var3)
+        db.session.commit()
+
+        yield {
+            "app": app,
+            "tenant": tenant,
+            "upload_files": [upload_file1, upload_file2],
+            "variable_files": [var_file1, var_file2],
+            "draft_variables": [draft_var1, draft_var2, draft_var3],
+        }
+
+        # Cleanup
+        db.session.rollback()
+
+        # Clean up any remaining records
+        for table, ids in [
+            (WorkflowDraftVariable, [v.id for v in [draft_var1, draft_var2, draft_var3]]),
+            (WorkflowDraftVariableFile, [vf.id for vf in [var_file1, var_file2]]),
+            (UploadFile, [uf.id for uf in [upload_file1, upload_file2]]),
+        ]:
+            cleanup_query = delete(table).where(table.id.in_(ids)).execution_options(synchronize_session=False)
+            db.session.execute(cleanup_query)
+
+        db.session.commit()
+
+    @patch("extensions.ext_storage.storage")
+    def test_delete_draft_variables_with_offload_data(self, mock_storage, setup_offload_test_data):
+        """Test that deleting draft variables also cleans up associated Offload data."""
+        data = setup_offload_test_data
+        app_id = data["app"].id
+
+        # Mock storage deletion to succeed
+        mock_storage.delete.return_value = None
+
+        # Verify initial state
+        draft_vars_before = db.session.query(WorkflowDraftVariable).filter_by(app_id=app_id).count()
+        var_files_before = db.session.query(WorkflowDraftVariableFile).count()
+        upload_files_before = db.session.query(UploadFile).count()
+
+        assert draft_vars_before == 3  # 2 with files + 1 regular
+        assert var_files_before == 2
+        assert upload_files_before == 2
+
+        # Delete draft variables
+        deleted_count = delete_draft_variables_batch(app_id, batch_size=10)
+
+        # Verify results
+        assert deleted_count == 3
+
+        # Check that all draft variables are deleted
+        draft_vars_after = db.session.query(WorkflowDraftVariable).filter_by(app_id=app_id).count()
+        assert draft_vars_after == 0
+
+        # Check that associated Offload data is cleaned up
+        var_files_after = db.session.query(WorkflowDraftVariableFile).count()
+        upload_files_after = db.session.query(UploadFile).count()
+
+        assert var_files_after == 0  # All variable files should be deleted
+        assert upload_files_after == 0  # All upload files should be deleted
+
+        # Verify storage deletion was called for both files
+        assert mock_storage.delete.call_count == 2
+        storage_keys_deleted = [call.args[0] for call in mock_storage.delete.call_args_list]
+        assert "test/file1.json" in storage_keys_deleted
+        assert "test/file2.json" in storage_keys_deleted
+
+    @patch("extensions.ext_storage.storage")
+    def test_delete_draft_variables_storage_failure_continues_cleanup(self, mock_storage, setup_offload_test_data):
+        """Test that database cleanup continues even when storage deletion fails."""
+        data = setup_offload_test_data
+        app_id = data["app"].id
+
+        # Mock storage deletion to fail for first file, succeed for second
+        mock_storage.delete.side_effect = [Exception("Storage error"), None]
+
+        # Delete draft variables
+        deleted_count = delete_draft_variables_batch(app_id, batch_size=10)
+
+        # Verify that all draft variables are still deleted
+        assert deleted_count == 3
+
+        draft_vars_after = db.session.query(WorkflowDraftVariable).filter_by(app_id=app_id).count()
+        assert draft_vars_after == 0
+
+        # Database cleanup should still succeed even with storage errors
+        var_files_after = db.session.query(WorkflowDraftVariableFile).count()
+        upload_files_after = db.session.query(UploadFile).count()
+
+        assert var_files_after == 0
+        assert upload_files_after == 0
+
+        # Verify storage deletion was attempted for both files
+        assert mock_storage.delete.call_count == 2
+
+    @patch("extensions.ext_storage.storage")
+    def test_delete_draft_variables_partial_offload_data(self, mock_storage, setup_offload_test_data):
+        """Test deletion with mix of variables with and without Offload data."""
+        data = setup_offload_test_data
+        app_id = data["app"].id
+
+        # Create additional app with only regular variables (no offload data)
+        tenant = data["tenant"]
+        app2 = App(
+            tenant_id=tenant.id,
+            name="Test App 2",
+            mode="workflow",
+            enable_site=True,
+            enable_api=True,
+        )
+        db.session.add(app2)
+        db.session.flush()
+
+        # Add regular variables to app2
+        regular_vars = []
+        for i in range(3):
+            var = WorkflowDraftVariable.new_node_variable(
+                app_id=app2.id,
+                node_id=f"node_{i}",
+                name=f"var_{i}",
+                value=StringSegment(value="regular_value"),
+                node_execution_id=str(uuid.uuid4()),
+            )
+            db.session.add(var)
+            regular_vars.append(var)
+        db.session.commit()
+
+        try:
+            # Mock storage deletion
+            mock_storage.delete.return_value = None
+
+            # Delete variables for app2 (no offload data)
+            deleted_count_app2 = delete_draft_variables_batch(app2.id, batch_size=10)
+            assert deleted_count_app2 == 3
+
+            # Verify storage wasn't called for app2 (no offload files)
+            mock_storage.delete.assert_not_called()
+
+            # Delete variables for original app (with offload data)
+            deleted_count_app1 = delete_draft_variables_batch(app_id, batch_size=10)
+            assert deleted_count_app1 == 3
+
+            # Now storage should be called for the offload files
+            assert mock_storage.delete.call_count == 2
+
+        finally:
+            # Cleanup app2 and its variables
+            cleanup_vars_query = (
+                delete(WorkflowDraftVariable)
+                .where(WorkflowDraftVariable.app_id == app2.id)
+                .execution_options(synchronize_session=False)
+            )
+            db.session.execute(cleanup_vars_query)
+
+            app2_obj = db.session.get(App, app2.id)
+            if app2_obj:
+                db.session.delete(app2_obj)
+            db.session.commit()

api/tests/integration_tests/test_offload.py

@@ -0,0 +1,213 @@
+import uuid
+
+import pytest
+from sqlalchemy.orm import Session, joinedload, selectinload
+
+from libs.datetime_utils import naive_utc_now
+from libs.uuid_utils import uuidv7
+from models import db
+from models.enums import CreatorUserRole
+from models.model import UploadFile
+from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionOffload, WorkflowNodeExecutionTriggeredFrom
+
+
+@pytest.fixture
+def session(flask_req_ctx):
+    with Session(bind=db.engine, expire_on_commit=False) as session:
+        yield session
+
+
+def test_offload(session, setup_account):
+    tenant_id = str(uuid.uuid4())
+    app_id = str(uuid.uuid4())
+    # step 1: create a UploadFile
+    input_upload_file = UploadFile(
+        tenant_id=tenant_id,
+        storage_type="local",
+        key="fake_storage_key",
+        name="test_file.txt",
+        size=1024,
+        extension="txt",
+        mime_type="text/plain",
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by=setup_account.id,
+        created_at=naive_utc_now(),
+        used=False,
+    )
+    output_upload_file = UploadFile(
+        tenant_id=tenant_id,
+        storage_type="local",
+        key="fake_storage_key",
+        name="test_file.txt",
+        size=1024,
+        extension="txt",
+        mime_type="text/plain",
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by=setup_account.id,
+        created_at=naive_utc_now(),
+        used=False,
+    )
+    session.add(input_upload_file)
+    session.add(output_upload_file)
+    session.flush()
+
+    # step 2: create a WorkflowNodeExecutionModel
+    node_execution = WorkflowNodeExecutionModel(
+        id=str(uuid.uuid4()),
+        tenant_id=tenant_id,
+        app_id=app_id,
+        workflow_id=str(uuid.uuid4()),
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        index=1,
+        node_id="test_node_id",
+        node_type="test",
+        title="Test Node",
+        status="succeeded",
+        created_by_role=CreatorUserRole.ACCOUNT.value,
+        created_by=setup_account.id,
+    )
+    session.add(node_execution)
+    session.flush()
+
+    # step 3: create a WorkflowNodeExecutionOffload
+    offload = WorkflowNodeExecutionOffload(
+        id=uuidv7(),
+        tenant_id=tenant_id,
+        app_id=app_id,
+        node_execution_id=node_execution.id,
+        inputs_file_id=input_upload_file.id,
+        outputs_file_id=output_upload_file.id,
+    )
+    session.add(offload)
+    session.flush()
+
+    # Test preloading - this should work without raising LazyLoadError
+    result = (
+        session.query(WorkflowNodeExecutionModel)
+        .options(
+            selectinload(WorkflowNodeExecutionModel.offload_data).options(
+                joinedload(
+                    WorkflowNodeExecutionOffload.inputs_file,
+                ),
+                joinedload(
+                    WorkflowNodeExecutionOffload.outputs_file,
+                ),
+            )
+        )
+        .filter(WorkflowNodeExecutionModel.id == node_execution.id)
+        .first()
+    )
+
+    # Verify the relationships are properly loaded
+    assert result is not None
+    assert result.offload_data is not None
+    assert result.offload_data.inputs_file is not None
+    assert result.offload_data.inputs_file.id == input_upload_file.id
+    assert result.offload_data.inputs_file.name == "test_file.txt"
+
+    # Test the computed properties
+    assert result.inputs_truncated is True
+    assert result.outputs_truncated is False
+    assert False
+
+
+def _test_offload_save(session, setup_account):
+    tenant_id = str(uuid.uuid4())
+    app_id = str(uuid.uuid4())
+    # step 1: create a UploadFile
+    input_upload_file = UploadFile(
+        tenant_id=tenant_id,
+        storage_type="local",
+        key="fake_storage_key",
+        name="test_file.txt",
+        size=1024,
+        extension="txt",
+        mime_type="text/plain",
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by=setup_account.id,
+        created_at=naive_utc_now(),
+        used=False,
+    )
+    output_upload_file = UploadFile(
+        tenant_id=tenant_id,
+        storage_type="local",
+        key="fake_storage_key",
+        name="test_file.txt",
+        size=1024,
+        extension="txt",
+        mime_type="text/plain",
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by=setup_account.id,
+        created_at=naive_utc_now(),
+        used=False,
+    )
+
+    node_execution_id = id = str(uuid.uuid4())
+
+    # step 3: create a WorkflowNodeExecutionOffload
+    offload = WorkflowNodeExecutionOffload(
+        id=uuidv7(),
+        tenant_id=tenant_id,
+        app_id=app_id,
+        node_execution_id=node_execution_id,
+    )
+    offload.inputs_file = input_upload_file
+    offload.outputs_file = output_upload_file
+
+    # step 2: create a WorkflowNodeExecutionModel
+    node_execution = WorkflowNodeExecutionModel(
+        id=str(uuid.uuid4()),
+        tenant_id=tenant_id,
+        app_id=app_id,
+        workflow_id=str(uuid.uuid4()),
+        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        index=1,
+        node_id="test_node_id",
+        node_type="test",
+        title="Test Node",
+        status="succeeded",
+        created_by_role=CreatorUserRole.ACCOUNT.value,
+        created_by=setup_account.id,
+    )
+    node_execution.offload_data = offload
+    session.add(node_execution)
+    session.flush()
+
+    assert False
+
+
+"""
+2025-08-21 15:34:49,570 INFO sqlalchemy.engine.Engine BEGIN (implicit)
+2025-08-21 15:34:49,572 INFO sqlalchemy.engine.Engine INSERT INTO upload_files (id, tenant_id, storage_type, key, name, size, extension, mime_type, created_by_role, created_by, created_at, used, used_by, used_at, hash, source_url) VALUES (%(id__0)s::UUID, %(tenant_id__0)s::UUID, %(storage_type__0)s, %(k ... 410 characters truncated ... (created_at__1)s, %(used__1)s, %(used_by__1)s::UUID, %(used_at__1)s, %(hash__1)s, %(source_url__1)s)
+2025-08-21 15:34:49,572 INFO sqlalchemy.engine.Engine [generated in 0.00009s (insertmanyvalues) 1/1 (unordered)] {'created_at__0': datetime.datetime(2025, 8, 21, 15, 34, 49, 570482), 'id__0': '366621fa-4326-403e-8709-62e4d0de7367', 'storage_type__0': 'local', 'extension__0': 'txt', 'created_by__0': 'ccc7657c-fb48-46bd-8f42-c837b14eab18', 'used_at__0': None, 'used_by__0': None, 'source_url__0': '', 'mime_type__0': 'text/plain', 'created_by_role__0': 'account', 'used__0': False, 'size__0': 1024, 'tenant_id__0': '4c1bbfc9-a28b-4d93-8987-45db78e3269c', 'hash__0': None, 'key__0': 'fake_storage_key', 'name__0': 'test_file.txt', 'created_at__1': datetime.datetime(2025, 8, 21, 15, 34, 49, 570563), 'id__1': '3cdec641-a452-4df0-a9af-4a1a30c27ea5', 'storage_type__1': 'local', 'extension__1': 'txt', 'created_by__1': 'ccc7657c-fb48-46bd-8f42-c837b14eab18', 'used_at__1': None, 'used_by__1': None, 'source_url__1': '', 'mime_type__1': 'text/plain', 'created_by_role__1': 'account', 'used__1': False, 'size__1': 1024, 'tenant_id__1': '4c1bbfc9-a28b-4d93-8987-45db78e3269c', 'hash__1': None, 'key__1': 'fake_storage_key', 'name__1': 'test_file.txt'}
+2025-08-21 15:34:49,576 INFO sqlalchemy.engine.Engine INSERT INTO workflow_node_executions (id, tenant_id, app_id, workflow_id, triggered_from, workflow_run_id, index, predecessor_node_id, node_execution_id, node_id, node_type, title, inputs, process_data, outputs, status, error, execution_metadata, created_by_role, created_by, finished_at) VALUES (%(id)s::UUID, %(tenant_id)s::UUID, %(app_id)s::UUID, %(workflow_id)s::UUID, %(triggered_from)s, %(workflow_run_id)s::UUID, %(index)s, %(predecessor_node_id)s, %(node_execution_id)s, %(node_id)s, %(node_type)s, %(title)s, %(inputs)s, %(process_data)s, %(outputs)s, %(status)s, %(error)s, %(execution_metadata)s, %(created_by_role)s, %(created_by)s::UUID, %(finished_at)s) RETURNING workflow_node_executions.elapsed_time, workflow_node_executions.created_at
+2025-08-21 15:34:49,576 INFO sqlalchemy.engine.Engine [generated in 0.00019s] {'id': '9aac28b6-b6fc-4aea-abdf-21da3227e621', 'tenant_id': '4c1bbfc9-a28b-4d93-8987-45db78e3269c', 'app_id': '79fa81c7-2760-40db-af54-74cb2fea2ce7', 'workflow_id': '95d341e3-381c-4c54-a383-f685a9741053', 'triggered_from': <WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN: 'workflow-run'>, 'workflow_run_id': None, 'index': 1, 'predecessor_node_id': None, 'node_execution_id': None, 'node_id': 'test_node_id', 'node_type': 'test', 'title': 'Test Node', 'inputs': None, 'process_data': None, 'outputs': None, 'status': 'succeeded', 'error': None, 'execution_metadata': None, 'created_by_role': 'account', 'created_by': 'ccc7657c-fb48-46bd-8f42-c837b14eab18', 'finished_at': None}
+2025-08-21 15:34:49,579 INFO sqlalchemy.engine.Engine INSERT INTO workflow_node_execution_offload (id, created_at, tenant_id, app_id, node_execution_id, inputs_file_id, outputs_file_id) VALUES (%(id)s::UUID, %(created_at)s, %(tenant_id)s::UUID, %(app_id)s::UUID, %(node_execution_id)s::UUID, %(inputs_file_id)s::UUID, %(outputs_file_id)s::UUID)
+2025-08-21 15:34:49,579 INFO sqlalchemy.engine.Engine [generated in 0.00016s] {'id': '0198cd44-b7ea-724b-9e1b-5f062a2ef45b', 'created_at': datetime.datetime(2025, 8, 21, 15, 34, 49, 579072), 'tenant_id': '4c1bbfc9-a28b-4d93-8987-45db78e3269c', 'app_id': '79fa81c7-2760-40db-af54-74cb2fea2ce7', 'node_execution_id': '9aac28b6-b6fc-4aea-abdf-21da3227e621', 'inputs_file_id': '366621fa-4326-403e-8709-62e4d0de7367', 'outputs_file_id': '3cdec641-a452-4df0-a9af-4a1a30c27ea5'}
+2025-08-21 15:34:49,581 INFO sqlalchemy.engine.Engine SELECT workflow_node_executions.id AS workflow_node_executions_id, workflow_node_executions.tenant_id AS workflow_node_executions_tenant_id, workflow_node_executions.app_id AS workflow_node_executions_app_id, workflow_node_executions.workflow_id AS workflow_node_executions_workflow_id, workflow_node_executions.triggered_from AS workflow_node_executions_triggered_from, workflow_node_executions.workflow_run_id AS workflow_node_executions_workflow_run_id, workflow_node_executions.index AS workflow_node_executions_index, workflow_node_executions.predecessor_node_id AS workflow_node_executions_predecessor_node_id, workflow_node_executions.node_execution_id AS workflow_node_executions_node_execution_id, workflow_node_executions.node_id AS workflow_node_executions_node_id, workflow_node_executions.node_type AS workflow_node_executions_node_type, workflow_node_executions.title AS workflow_node_executions_title, workflow_node_executions.inputs AS workflow_node_executions_inputs, workflow_node_executions.process_data AS workflow_node_executions_process_data, workflow_node_executions.outputs AS workflow_node_executions_outputs, workflow_node_executions.status AS workflow_node_executions_status, workflow_node_executions.error AS workflow_node_executions_error, workflow_node_executions.elapsed_time AS workflow_node_executions_elapsed_time, workflow_node_executions.execution_metadata AS workflow_node_executions_execution_metadata, workflow_node_executions.created_at AS workflow_node_executions_created_at, workflow_node_executions.created_by_role AS workflow_node_executions_created_by_role, workflow_node_executions.created_by AS workflow_node_executions_created_by, workflow_node_executions.finished_at AS workflow_node_executions_finished_at
+FROM workflow_node_executions
+WHERE workflow_node_executions.id = %(id_1)s::UUID
+ LIMIT %(param_1)s
+2025-08-21 15:34:49,581 INFO sqlalchemy.engine.Engine [generated in 0.00009s] {'id_1': '9aac28b6-b6fc-4aea-abdf-21da3227e621', 'param_1': 1}
+2025-08-21 15:34:49,585 INFO sqlalchemy.engine.Engine SELECT workflow_node_execution_offload.node_execution_id AS workflow_node_execution_offload_node_execution_id, workflow_node_execution_offload.id AS workflow_node_execution_offload_id, workflow_node_execution_offload.created_at AS workflow_node_execution_offload_created_at, workflow_node_execution_offload.tenant_id AS workflow_node_execution_offload_tenant_id, workflow_node_execution_offload.app_id AS workflow_node_execution_offload_app_id, workflow_node_execution_offload.inputs_file_id AS workflow_node_execution_offload_inputs_file_id, workflow_node_execution_offload.outputs_file_id AS workflow_node_execution_offload_outputs_file_id
+FROM workflow_node_execution_offload
+WHERE workflow_node_execution_offload.node_execution_id IN (%(primary_keys_1)s::UUID)
+2025-08-21 15:34:49,585 INFO sqlalchemy.engine.Engine [generated in 0.00021s] {'primary_keys_1': '9aac28b6-b6fc-4aea-abdf-21da3227e621'}
+2025-08-21 15:34:49,587 INFO sqlalchemy.engine.Engine SELECT upload_files.id AS upload_files_id, upload_files.tenant_id AS upload_files_tenant_id, upload_files.storage_type AS upload_files_storage_type, upload_files.key AS upload_files_key, upload_files.name AS upload_files_name, upload_files.size AS upload_files_size, upload_files.extension AS upload_files_extension, upload_files.mime_type AS upload_files_mime_type, upload_files.created_by_role AS upload_files_created_by_role, upload_files.created_by AS upload_files_created_by, upload_files.created_at AS upload_files_created_at, upload_files.used AS upload_files_used, upload_files.used_by AS upload_files_used_by, upload_files.used_at AS upload_files_used_at, upload_files.hash AS upload_files_hash, upload_files.source_url AS upload_files_source_url
+FROM upload_files
+WHERE upload_files.id IN (%(primary_keys_1)s::UUID)
+2025-08-21 15:34:49,587 INFO sqlalchemy.engine.Engine [generated in 0.00012s] {'primary_keys_1': '3cdec641-a452-4df0-a9af-4a1a30c27ea5'}
+2025-08-21 15:34:49,588 INFO sqlalchemy.engine.Engine SELECT upload_files.id AS upload_files_id, upload_files.tenant_id AS upload_files_tenant_id, upload_files.storage_type AS upload_files_storage_type, upload_files.key AS upload_files_key, upload_files.name AS upload_files_name, upload_files.size AS upload_files_size, upload_files.extension AS upload_files_extension, upload_files.mime_type AS upload_files_mime_type, upload_files.created_by_role AS upload_files_created_by_role, upload_files.created_by AS upload_files_created_by, upload_files.created_at AS upload_files_created_at, upload_files.used AS upload_files_used, upload_files.used_by AS upload_files_used_by, upload_files.used_at AS upload_files_used_at, upload_files.hash AS upload_files_hash, upload_files.source_url AS upload_files_source_url
+FROM upload_files
+WHERE upload_files.id IN (%(primary_keys_1)s::UUID)
+2025-08-21 15:34:49,588 INFO sqlalchemy.engine.Engine [generated in 0.00010s] {'primary_keys_1': '366621fa-4326-403e-8709-62e4d0de7367'}
+"""
+
+
+"""
+upload_file_id: 366621fa-4326-403e-8709-62e4d0de7367 3cdec641-a452-4df0-a9af-4a1a30c27ea5
+
+workflow_node_executions_id: 9aac28b6-b6fc-4aea-abdf-21da3227e621
+
+offload_id: 0198cd44-b7ea-724b-9e1b-5f062a2ef45b
+"""

api/tests/integration_tests/workflow/test_process_data_truncation_integration.py

@@ -0,0 +1,421 @@
+"""
+Integration tests for process_data truncation functionality.
+
+These tests verify the end-to-end behavior of process_data truncation across
+the entire system, from database storage to API responses.
+"""
+
+import json
+from dataclasses import dataclass
+from datetime import datetime
+from unittest.mock import Mock, patch
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+from core.repositories.sqlalchemy_workflow_node_execution_repository import SQLAlchemyWorkflowNodeExecutionRepository
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecution, WorkflowNodeExecutionStatus
+from core.workflow.nodes.enums import NodeType
+from models import Account
+from models.workflow import WorkflowNodeExecutionTriggeredFrom
+
+
+@dataclass
+class TruncationTestData:
+    """Test data for truncation scenarios."""
+    name: str
+    process_data: dict[str, any]
+    should_truncate: bool
+    expected_storage_interaction: bool
+
+
+class TestProcessDataTruncationIntegration:
+    """Integration tests for process_data truncation functionality."""
+    
+    @pytest.fixture
+    def in_memory_db_engine(self):
+        """Create an in-memory SQLite database for testing."""
+        engine = create_engine("sqlite:///:memory:")
+        
+        # Create minimal table structure for testing
+        with engine.connect() as conn:
+            # Create workflow_node_executions table
+            conn.execute(text("""
+                CREATE TABLE workflow_node_executions (
+                    id TEXT PRIMARY KEY,
+                    tenant_id TEXT NOT NULL,
+                    app_id TEXT NOT NULL,
+                    workflow_id TEXT NOT NULL,
+                    triggered_from TEXT NOT NULL,
+                    workflow_run_id TEXT,
+                    index_ INTEGER NOT NULL,
+                    predecessor_node_id TEXT,
+                    node_execution_id TEXT,
+                    node_id TEXT NOT NULL,
+                    node_type TEXT NOT NULL,
+                    title TEXT NOT NULL,
+                    inputs TEXT,
+                    process_data TEXT,
+                    outputs TEXT,
+                    status TEXT NOT NULL,
+                    error TEXT,
+                    elapsed_time REAL DEFAULT 0,
+                    execution_metadata TEXT,
+                    created_at DATETIME NOT NULL,
+                    created_by_role TEXT NOT NULL,
+                    created_by TEXT NOT NULL,
+                    finished_at DATETIME
+                )
+            """))
+            
+            # Create workflow_node_execution_offload table
+            conn.execute(text("""
+                CREATE TABLE workflow_node_execution_offload (
+                    id TEXT PRIMARY KEY,
+                    created_at DATETIME NOT NULL,
+                    tenant_id TEXT NOT NULL,
+                    app_id TEXT NOT NULL,
+                    node_execution_id TEXT NOT NULL UNIQUE,
+                    inputs_file_id TEXT,
+                    outputs_file_id TEXT,
+                    process_data_file_id TEXT
+                )
+            """))
+            
+            # Create upload_files table (simplified)
+            conn.execute(text("""
+                CREATE TABLE upload_files (
+                    id TEXT PRIMARY KEY,
+                    tenant_id TEXT NOT NULL,
+                    storage_key TEXT NOT NULL,
+                    filename TEXT NOT NULL,
+                    size INTEGER NOT NULL,
+                    created_at DATETIME NOT NULL
+                )
+            """))
+            
+            conn.commit()
+        
+        return engine
+
+    @pytest.fixture
+    def mock_account(self):
+        """Create a mock account for testing."""
+        account = Mock(spec=Account)
+        account.id = "test-user-id"
+        account.tenant_id = "test-tenant-id"
+        return account
+
+    @pytest.fixture
+    def repository(self, in_memory_db_engine, mock_account):
+        """Create a repository instance for testing."""
+        session_factory = sessionmaker(bind=in_memory_db_engine)
+        
+        return SQLAlchemyWorkflowNodeExecutionRepository(
+            session_factory=session_factory,
+            user=mock_account,
+            app_id="test-app-id",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        )
+
+    def create_test_execution(
+        self, 
+        process_data: dict[str, any] | None = None,
+        execution_id: str = "test-execution-id"
+    ) -> WorkflowNodeExecution:
+        """Create a test execution with process_data."""
+        return WorkflowNodeExecution(
+            id=execution_id,
+            workflow_id="test-workflow-id",
+            workflow_execution_id="test-run-id",
+            index=1,
+            node_id="test-node-id",
+            node_type=NodeType.LLM,
+            title="Test Node",
+            process_data=process_data,
+            status=WorkflowNodeExecutionStatus.SUCCEEDED,
+            created_at=datetime.now(),
+            finished_at=datetime.now(),
+        )
+
+    def get_truncation_test_data(self) -> list[TruncationTestData]:
+        """Get test data for various truncation scenarios."""
+        return [
+            TruncationTestData(
+                name="small_process_data",
+                process_data={"small": "data", "count": 5},
+                should_truncate=False,
+                expected_storage_interaction=False,
+            ),
+            TruncationTestData(
+                name="large_process_data",
+                process_data={"large_field": "x" * 10000, "metadata": "info"},
+                should_truncate=True,
+                expected_storage_interaction=True,
+            ),
+            TruncationTestData(
+                name="complex_large_data",
+                process_data={
+                    "logs": ["log entry"] * 500,  # Large array
+                    "config": {"setting": "value"},
+                    "status": "processing",
+                    "details": {"description": "y" * 5000}  # Large string
+                },
+                should_truncate=True,
+                expected_storage_interaction=True,
+            ),
+        ]
+
+    @patch('core.repositories.sqlalchemy_workflow_node_execution_repository.dify_config')
+    @patch('services.file_service.FileService.upload_file')
+    @patch('extensions.ext_storage.storage')
+    def test_end_to_end_process_data_truncation(
+        self, 
+        mock_storage, 
+        mock_upload_file,
+        mock_config,
+        repository
+    ):
+        """Test end-to-end process_data truncation functionality."""
+        # Configure truncation limits
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE = 1000
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH = 100
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_STRING_LENGTH = 500
+        
+        # Create large process_data that should be truncated
+        large_process_data = {
+            "large_field": "x" * 10000,  # Exceeds string length limit
+            "metadata": {"type": "processing", "timestamp": 1234567890}
+        }
+        
+        # Mock file upload
+        mock_file = Mock()
+        mock_file.id = "mock-process-data-file-id"
+        mock_upload_file.return_value = mock_file
+        
+        # Create and save execution
+        execution = self.create_test_execution(process_data=large_process_data)
+        repository.save(execution)
+        
+        # Verify truncation occurred
+        assert execution.process_data_truncated is True
+        truncated_data = execution.get_truncated_process_data()
+        assert truncated_data is not None
+        assert truncated_data != large_process_data  # Should be different due to truncation
+        
+        # Verify file upload was called for process_data
+        assert mock_upload_file.called
+        upload_args = mock_upload_file.call_args
+        assert "_process_data" in upload_args[1]["filename"]
+
+    @patch('core.repositories.sqlalchemy_workflow_node_execution_repository.dify_config')
+    def test_small_process_data_no_truncation(self, mock_config, repository):
+        """Test that small process_data is not truncated."""
+        # Configure truncation limits
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE = 1000
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH = 100
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_STRING_LENGTH = 500
+        
+        # Create small process_data
+        small_process_data = {"small": "data", "count": 5}
+        
+        execution = self.create_test_execution(process_data=small_process_data)
+        repository.save(execution)
+        
+        # Verify no truncation occurred
+        assert execution.process_data_truncated is False
+        assert execution.get_truncated_process_data() is None
+        assert execution.get_response_process_data() == small_process_data
+
+    @pytest.mark.parametrize("test_data", [
+        data for data in get_truncation_test_data(None)
+    ], ids=[data.name for data in get_truncation_test_data(None)])
+    @patch('core.repositories.sqlalchemy_workflow_node_execution_repository.dify_config')
+    @patch('services.file_service.FileService.upload_file')
+    def test_various_truncation_scenarios(
+        self, 
+        mock_upload_file,
+        mock_config,
+        test_data: TruncationTestData,
+        repository
+    ):
+        """Test various process_data truncation scenarios."""
+        # Configure truncation limits
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE = 1000
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH = 100
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_STRING_LENGTH = 500
+        
+        if test_data.expected_storage_interaction:
+            # Mock file upload for truncation scenarios
+            mock_file = Mock()
+            mock_file.id = f"file-{test_data.name}"
+            mock_upload_file.return_value = mock_file
+        
+        execution = self.create_test_execution(process_data=test_data.process_data)
+        repository.save(execution)
+        
+        # Verify truncation behavior matches expectations
+        assert execution.process_data_truncated == test_data.should_truncate
+        
+        if test_data.should_truncate:
+            assert execution.get_truncated_process_data() is not None
+            assert execution.get_truncated_process_data() != test_data.process_data
+            assert mock_upload_file.called
+        else:
+            assert execution.get_truncated_process_data() is None
+            assert execution.get_response_process_data() == test_data.process_data
+
+    @patch('core.repositories.sqlalchemy_workflow_node_execution_repository.dify_config')
+    @patch('services.file_service.FileService.upload_file')
+    @patch('extensions.ext_storage.storage')
+    def test_load_truncated_execution_from_database(
+        self, 
+        mock_storage, 
+        mock_upload_file,
+        mock_config,
+        repository,
+        in_memory_db_engine
+    ):
+        """Test loading an execution with truncated process_data from database."""
+        # Configure truncation
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE = 1000
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_ARRAY_LENGTH = 100
+        mock_config.WORKFLOW_VARIABLE_TRUNCATION_STRING_LENGTH = 500
+        
+        # Create and save execution with large process_data
+        large_process_data = {
+            "large_field": "x" * 10000,
+            "metadata": "info"
+        }
+        
+        # Mock file upload
+        mock_file = Mock()
+        mock_file.id = "process-data-file-id"
+        mock_upload_file.return_value = mock_file
+        
+        execution = self.create_test_execution(process_data=large_process_data)
+        repository.save(execution)
+        
+        # Mock storage load for reconstruction
+        mock_storage.load.return_value = json.dumps(large_process_data).encode()
+        
+        # Create a new repository instance to simulate fresh load
+        session_factory = sessionmaker(bind=in_memory_db_engine)
+        new_repository = SQLAlchemyWorkflowNodeExecutionRepository(
+            session_factory=session_factory,
+            user=Mock(spec=Account, id="test-user", tenant_id="test-tenant"),
+            app_id="test-app-id",
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        )
+        
+        # Load executions from database
+        executions = new_repository.get_by_workflow_run("test-run-id")
+        
+        assert len(executions) == 1
+        loaded_execution = executions[0]
+        
+        # Verify that full data is loaded
+        assert loaded_execution.process_data == large_process_data
+        assert loaded_execution.process_data_truncated is True
+        
+        # Verify truncated data for responses
+        response_data = loaded_execution.get_response_process_data()
+        assert response_data != large_process_data  # Should be truncated version
+
+    def test_process_data_none_handling(self, repository):
+        """Test handling of None process_data."""
+        execution = self.create_test_execution(process_data=None)
+        repository.save(execution)
+        
+        # Should handle None gracefully
+        assert execution.process_data is None
+        assert execution.process_data_truncated is False
+        assert execution.get_response_process_data() is None
+
+    def test_empty_process_data_handling(self, repository):
+        """Test handling of empty process_data."""
+        execution = self.create_test_execution(process_data={})
+        repository.save(execution)
+        
+        # Should handle empty dict gracefully
+        assert execution.process_data == {}
+        assert execution.process_data_truncated is False
+        assert execution.get_response_process_data() == {}
+
+
+class TestProcessDataTruncationApiIntegration:
+    """Integration tests for API responses with process_data truncation."""
+    
+    def test_api_response_includes_truncated_flag(self):
+        """Test that API responses include the process_data_truncated flag."""
+        from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
+        from core.app.entities.app_invoke_entities import WorkflowAppGenerateEntity
+        from core.app.entities.queue_entities import QueueNodeSucceededEvent
+        
+        # Create execution with truncated process_data
+        execution = WorkflowNodeExecution(
+            id="test-execution-id",
+            workflow_id="test-workflow-id",
+            workflow_execution_id="test-run-id",
+            index=1,
+            node_id="test-node-id",
+            node_type=NodeType.LLM,
+            title="Test Node",
+            process_data={"large": "x" * 10000},
+            status=WorkflowNodeExecutionStatus.SUCCEEDED,
+            created_at=datetime.now(),
+            finished_at=datetime.now(),
+        )
+        
+        # Set truncated data
+        execution.set_truncated_process_data({"large": "[TRUNCATED]"})
+        
+        # Create converter and event
+        converter = WorkflowResponseConverter(
+            application_generate_entity=Mock(
+                spec=WorkflowAppGenerateEntity,
+                app_config=Mock(tenant_id="test-tenant")
+            )
+        )
+        
+        event = QueueNodeSucceededEvent(
+            node_id="test-node-id",
+            node_type=NodeType.LLM,
+            node_data=Mock(),
+            parallel_id=None,
+            parallel_start_node_id=None,
+            parent_parallel_id=None,
+            parent_parallel_start_node_id=None,
+            in_iteration_id=None,
+            in_loop_id=None,
+        )
+        
+        # Generate response
+        response = converter.workflow_node_finish_to_stream_response(
+            event=event,
+            task_id="test-task-id",
+            workflow_node_execution=execution,
+        )
+        
+        # Verify response includes truncated flag and data
+        assert response is not None
+        assert response.data.process_data_truncated is True
+        assert response.data.process_data == {"large": "[TRUNCATED]"}
+        
+        # Verify response can be serialized
+        response_dict = response.to_dict()
+        assert "process_data_truncated" in response_dict["data"]
+        assert response_dict["data"]["process_data_truncated"] is True
+
+    def test_workflow_run_fields_include_truncated_flag(self):
+        """Test that workflow run fields include process_data_truncated."""
+        from fields.workflow_run_fields import workflow_run_node_execution_fields
+        
+        # Verify the field is included in the definition
+        assert "process_data_truncated" in workflow_run_node_execution_fields
+        
+        # The field should be a Boolean field
+        field = workflow_run_node_execution_fields["process_data_truncated"]
+        from flask_restful import fields
+        assert isinstance(field, fields.Boolean)