dify/api/tests/unit_tests/models/test_workflow.py

import dataclasses
import json
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest import mock
from unittest.mock import Mock
from uuid import NAMESPACE_URL, UUID, uuid4, uuid5

import pytest
from pydantic import JsonValue
from sqlalchemy import inspect, select
from sqlalchemy.orm import Session, configure_mappers

from constants import HIDDEN_VALUE
from core.helper import encrypter
from core.workflow.file_reference import build_file_reference
from core.workflow.llm_environment_variable import LLMEnvironmentVariable
from extensions.storage.storage_type import StorageType
from factories import variable_factory
from factories.variable_factory import build_segment
from fields.workflow_run_fields import (
    node_execution_response_source,
    workflow_run_pagination_response_source,
    workflow_run_response_source,
)
from graphon.enums import (
    BuiltinNodeTypes,
    WorkflowExecutionStatus,
    WorkflowNodeExecutionStatus,
)
from graphon.file import File, FileTransferMethod, FileType
from graphon.variables import FloatVariable, IntegerVariable, SecretVariable, SegmentType, StringVariable
from graphon.variables.segments import IntegerSegment, Segment
from models import ConversationVariable
from models import Workflow as _workflow_feature_compatibility_Workflow
from models.account import Account
from models.base import Base, TypeBase
from models.enums import CreatorUserRole, EndUserType, ExecutionOffLoadType, WorkflowRunTriggeredFrom
from models.model import App, AppMode, EndUser, UploadFile
from models.tools import WorkflowToolProvider
from models.workflow import (
    Workflow as _workflow_Workflow,
)
from models.workflow import (
    Workflow as _workflow_models_Workflow,
)
from models.workflow import (
    WorkflowAppLog,
    WorkflowAppLogCreatedFrom,
    WorkflowDraftVariable,
    WorkflowNodeExecutionModel,
    WorkflowNodeExecutionOffload,
    WorkflowNodeExecutionTriggeredFrom,
    WorkflowRun,
    WorkflowType,
    WorkflowVersionCounter,
    _resolve_workflow_app_tenant_id,
    is_system_variable_editable,
)
from models.workflow import WorkflowType as ModelWorkflowType


class TestConversationVariableCases:
    @staticmethod
    def test_from_variable_coerces_non_uuid_primary_key() -> None:
        variable = variable_factory.build_conversation_variable_from_mapping(
            {
                "id": "opt-comp-prompt-var",
                "name": "optimization_comparison_prompt",
                "value_type": SegmentType.STRING,
                "value": "-",
            }
        )

        row = ConversationVariable.from_variable(app_id="app_id", conversation_id="conversation_id", variable=variable)

        expected = str(uuid5(NAMESPACE_URL, "dify:conversation-variable:optimization_comparison_prompt"))
        UUID(row.id)
        assert row.id == expected
        assert variable.id == "opt-comp-prompt-var"
        assert row.to_variable().id == "opt-comp-prompt-var"

    @staticmethod
    def test_from_variable_and_to_variable() -> None:
        variable = variable_factory.build_conversation_variable_from_mapping(
            {
                "id": str(uuid4()),
                "name": "name",
                "value_type": SegmentType.OBJECT,
                "value": {
                    "key": {
                        "key": "value",
                    }
                },
            }
        )

        conversation_variable = ConversationVariable.from_variable(
            app_id="app_id", conversation_id="conversation_id", variable=variable
        )

        assert conversation_variable.to_variable() == variable


class TestWorkflowCases:
    @staticmethod
    def test_environment_variables() -> None:
        # tenant_id context variable removed - using current_user.current_tenant_id directly

        # Create a Workflow instance
        workflow = _workflow_Workflow(
            tenant_id="tenant_id",
            app_id="app_id",
            type=ModelWorkflowType("workflow"),
            version="draft",
            graph="{}",
            _features="{}",
            created_by="account_id",
            _environment_variables=[],
            _conversation_variables=[],
        )

        # Create some EnvironmentVariable instances
        variable1 = StringVariable(name="var1", value="value1", id=str(uuid4()), selector=["env", "var1"])
        variable2 = IntegerVariable(name="var2", value=123, id=str(uuid4()), selector=["env", "var2"])
        variable3 = SecretVariable(name="var3", value="secret", id=str(uuid4()), selector=["env", "var3"])
        variable4 = FloatVariable(name="var4", value=float("3.14"), id=str(uuid4()), selector=["env", "var4"])

        with (
            mock.patch("core.helper.encrypter.encrypt_token", return_value="encrypted_token"),
            mock.patch("core.helper.encrypter.decrypt_token", return_value="secret"),
        ):
            # Set the environment_variables property of the Workflow instance
            variables = [variable1, variable2, variable3, variable4]
            workflow.environment_variables = variables

            # Get the environment_variables property and assert its value
            assert workflow.environment_variables == variables

    @staticmethod
    def test_llm_environment_variable_round_trip() -> None:
        workflow = _workflow_Workflow(
            tenant_id="tenant_id",
            app_id="app_id",
            type=ModelWorkflowType("workflow"),
            version="draft",
            graph="{}",
            _features="{}",
            created_by="account_id",
            _environment_variables=[],
            _conversation_variables=[],
        )
        variable = LLMEnvironmentVariable(
            name="for_research",
            value={"provider": "langgenius/anthropic/anthropic", "name": "claude-sonnet", "mode": "chat"},
            id=str(uuid4()),
            selector=["env", "for_research"],
        )

        workflow.environment_variables = [variable]

        assert workflow.environment_variables == [variable]
        assert json.loads(workflow._environment_variables)["for_research"]["value_type"] == "llm"
        assert workflow.to_dict()["environment_variables"][0]["value_type"] == "llm"

    @staticmethod
    def test_update_environment_variables() -> None:
        # tenant_id context variable removed - using current_user.current_tenant_id directly

        # Create a Workflow instance
        workflow = _workflow_Workflow(
            tenant_id="tenant_id",
            app_id="app_id",
            type=ModelWorkflowType("workflow"),
            version="draft",
            graph="{}",
            _features="{}",
            created_by="account_id",
            _environment_variables=[],
            _conversation_variables=[],
        )

        # Create some EnvironmentVariable instances
        variable1 = StringVariable(name="var1", value="value1", id=str(uuid4()), selector=["env", "var1"])
        variable2 = IntegerVariable(name="var2", value=123, id=str(uuid4()), selector=["env", "var2"])
        variable3 = SecretVariable(
            name="var3",
            value="secret",
            id=str(uuid4()),
            selector=["env", "var3"],
            description="old description",
        )
        variable4 = FloatVariable(name="var4", value=float("3.14"), id=str(uuid4()), selector=["env", "var4"])

        with (
            mock.patch("core.helper.encrypter.encrypt_token", return_value="encrypted_token"),
            mock.patch("core.helper.encrypter.decrypt_token", return_value="secret"),
        ):
            variables = [variable1, variable2, variable3, variable4]

            # Set the environment_variables property of the Workflow instance
            workflow.environment_variables = variables
            assert workflow.environment_variables == [variable1, variable2, variable3, variable4]

            # Update the name and description of variable3 and keep the value as it is
            variables[2] = variable3.model_copy(
                update={
                    "name": "new name",
                    "description": "new description",
                    "value": HIDDEN_VALUE,
                }
            )

            workflow.environment_variables = variables
            assert workflow.environment_variables[2].name == "new name"
            assert workflow.environment_variables[2].description == "new description"
            assert workflow.environment_variables[2].value == variable3.value

    @staticmethod
    def test_to_dict() -> None:
        # tenant_id context variable removed - using current_user.current_tenant_id directly

        # Create a Workflow instance
        workflow = _workflow_Workflow(
            tenant_id="tenant_id",
            app_id="app_id",
            type=ModelWorkflowType("workflow"),
            version="draft",
            graph="{}",
            _features="{}",
            created_by="account_id",
            _environment_variables=[],
            _conversation_variables=[],
        )

        # Create some EnvironmentVariable instances

        with (
            mock.patch("core.helper.encrypter.encrypt_token", return_value="encrypted_token"),
            mock.patch("core.helper.encrypter.decrypt_token", return_value="secret"),
        ):
            # Set the environment_variables property of the Workflow instance
            workflow.environment_variables = [
                SecretVariable(name="secret", value="secret", id=str(uuid4())),
                StringVariable(name="text", value="text", id=str(uuid4())),
            ]

            workflow_dict = workflow.to_dict()
            assert workflow_dict["environment_variables"][0]["value"] == ""
            assert workflow_dict["environment_variables"][1]["value"] == "text"

            workflow_dict = workflow.to_dict(include_secret=True)
            assert workflow_dict["environment_variables"][0]["value"] == "secret"
            assert workflow_dict["environment_variables"][1]["value"] == "text"

    @staticmethod
    @pytest.mark.parametrize("sqlite_session", [(_workflow_Workflow, Account)], indirect=True)
    def test_workflow_account_accessors_use_caller_session(sqlite_session: Session) -> None:
        created_account = Account(name="Created Account", email="created@example.com")
        created_account.id = "created-account-id"
        updated_account = Account(name="Updated Account", email="updated@example.com")
        updated_account.id = "updated-account-id"
        decoy_account = Account(name="Decoy Account", email="decoy@example.com")
        decoy_account.id = "decoy-account-id"
        workflow = _workflow_Workflow(
            tenant_id="tenant_id",
            app_id="app_id",
            type=ModelWorkflowType("workflow"),
            version="draft",
            graph="{}",
            _features="{}",
            created_by="created-account-id",
            _environment_variables=[],
            _conversation_variables=[],
            updated_by="updated-account-id",
        )
        sqlite_session.add_all([decoy_account, updated_account, workflow, created_account])
        sqlite_session.flush()

        assert workflow.created_by_account(sqlite_session) is created_account
        assert workflow.updated_by_account(sqlite_session) is updated_account

    @staticmethod
    @pytest.mark.parametrize("sqlite_session", [(_workflow_Workflow, WorkflowToolProvider)], indirect=True)
    def test_workflow_tool_published_accessor_uses_caller_session(sqlite_session: Session) -> None:
        workflow = _workflow_Workflow(
            tenant_id="tenant_id",
            app_id="app_id",
            type=ModelWorkflowType("workflow"),
            version="draft",
            graph="{}",
            _features="{}",
            created_by="account_id",
            _environment_variables=[],
            _conversation_variables=[],
        )
        matching_provider = WorkflowToolProvider(
            name="matching-provider",
            label="Matching provider",
            icon="tool",
            app_id=workflow.app_id,
            version="1",
            user_id="account-id",
            tenant_id=workflow.tenant_id,
            description="Matching workflow tool",
        )
        decoy_provider = WorkflowToolProvider(
            name="decoy-provider",
            label="Decoy provider",
            icon="tool",
            app_id="other-app",
            version="1",
            user_id="account-id",
            tenant_id=workflow.tenant_id,
            description="Different app",
        )
        sqlite_session.add_all([decoy_provider, workflow, matching_provider])
        sqlite_session.flush()

        with pytest.warns(DeprecationWarning, match="not accurate"):
            # The deprecated accessor is the compatibility surface under test.
            assert workflow.tool_published(sqlite_session) is True  # pyrefly: ignore[deprecated]

    @staticmethod
    def test_normalize_environment_variable_mappings_converts_full_mask_to_hidden_value() -> None:
        normalized = _workflow_Workflow.normalize_environment_variable_mappings(
            [
                {
                    "id": str(uuid4()),
                    "name": "secret",
                    "value": encrypter.full_mask_token(),
                    "value_type": "secret",
                }
            ]
        )

        assert normalized[0]["value"] == HIDDEN_VALUE

    @staticmethod
    def test_normalize_environment_variable_mappings_keeps_hidden_value() -> None:
        normalized = _workflow_Workflow.normalize_environment_variable_mappings(
            [
                {
                    "id": str(uuid4()),
                    "name": "secret",
                    "value": HIDDEN_VALUE,
                    "value_type": "secret",
                }
            ]
        )

        assert normalized[0]["value"] == HIDDEN_VALUE

    class TestWorkflowNodeExecution:
        def test_execution_metadata_dict(self) -> None:
            node_exec = WorkflowNodeExecutionModel(
                execution_metadata=None,
            )
            assert node_exec.execution_metadata_dict == {}

            original = {"a": 1, "b": ["2"]}
            node_exec.execution_metadata = json.dumps(original)
            assert node_exec.execution_metadata_dict == original

    class TestIsSystemVariableEditable:
        def test_is_system_variable(self) -> None:
            cases = [
                ("query", True),
                ("files", True),
                ("dialogue_count", False),
                ("conversation_id", False),
                ("user_id", False),
                ("app_id", False),
                ("workflow_id", False),
                ("workflow_run_id", False),
            ]
            for name, editable in cases:
                assert editable == is_system_variable_editable(name)

            assert is_system_variable_editable("invalid_or_new_system_variable") is False

    class TestWorkflowDraftVariableGetValue:
        def test_get_value_by_case(self, sqlite_session: Session) -> None:
            @dataclasses.dataclass
            class TestCase:
                name: str
                value: Segment

            test_file = File(
                file_type=FileType.IMAGE,
                transfer_method=FileTransferMethod.REMOTE_URL,
                remote_url="https://example.com/example.jpg",
                filename="example.jpg",
                extension=".jpg",
                mime_type="image/jpeg",
                size=100,
            )
            cases: list[TestCase] = [
                TestCase(
                    name="number/int",
                    value=build_segment(1),
                ),
                TestCase(
                    name="number/float",
                    value=build_segment(1.0),
                ),
                TestCase(
                    name="string",
                    value=build_segment("a"),
                ),
                TestCase(
                    name="object",
                    value=build_segment({}),
                ),
                TestCase(
                    name="file",
                    value=build_segment(test_file),
                ),
                TestCase(
                    name="array[any]",
                    value=build_segment([1, "a"]),
                ),
                TestCase(
                    name="array[string]",
                    value=build_segment(["a", "b"]),
                ),
                TestCase(
                    name="array[number]/int",
                    value=build_segment([1, 2]),
                ),
                TestCase(
                    name="array[number]/float",
                    value=build_segment([1.0, 2.0]),
                ),
                TestCase(
                    name="array[number]/mixed",
                    value=build_segment([1, 2.0]),
                ),
                TestCase(
                    name="array[object]",
                    value=build_segment([{}, {"a": 1}]),
                ),
                TestCase(
                    name="none",
                    value=build_segment(None),
                ),
            ]

            for idx, c in enumerate(cases, 1):
                fail_msg = f"test case {c.name} failed, index={idx}"
                draft_var = WorkflowDraftVariable()
                draft_var.set_value(c.value)
                assert c.value == draft_var.get_value(session=sqlite_session), fail_msg

        def test_file_variable_preserves_all_fields(self, sqlite_session: Session) -> None:
            """Test that File type variables preserve all fields during encoding/decoding."""

            # Create a File with specific field values
            test_file = File(
                file_id="test_file_id",
                file_type=FileType.IMAGE,
                transfer_method=FileTransferMethod.REMOTE_URL,
                remote_url="https://example.com/test.jpg",
                filename="test.jpg",
                extension=".jpg",
                mime_type="image/jpeg",
                size=12345,  # Specific size to test preservation
                storage_key="test_storage_key",
            )

            # Create a FileSegment and WorkflowDraftVariable
            file_segment = build_segment(test_file)
            draft_var = WorkflowDraftVariable()
            draft_var.set_value(file_segment)

            # Retrieve the value and verify all fields are preserved
            retrieved_segment = draft_var.get_value(session=sqlite_session)
            retrieved_file = retrieved_segment.value

            # Verify all important fields are preserved
            assert retrieved_file.id == test_file.id
            assert retrieved_file.type == test_file.type
            assert retrieved_file.transfer_method == test_file.transfer_method
            assert retrieved_file.remote_url == test_file.remote_url
            assert retrieved_file.filename == test_file.filename
            assert retrieved_file.extension == test_file.extension
            assert retrieved_file.mime_type == test_file.mime_type
            assert retrieved_file.size == test_file.size  # This was the main issue being fixed
            # Note: storage_key is not serialized in model_dump() so it won't be preserved

            # Verify the segments have the same type and the important fields match
            assert file_segment.value_type == retrieved_segment.value_type

        def test_file_variable_rebuilds_storage_backed_payloads_with_app_tenant(self, sqlite_session: Session) -> None:
            persisted_file = File(
                file_id="test_file_id",
                file_type=FileType.DOCUMENT,
                transfer_method=FileTransferMethod.LOCAL_FILE,
                reference=build_file_reference(record_id="upload-1", storage_key="legacy-storage-key"),
                filename="test.txt",
                extension=".txt",
                mime_type="text/plain",
                size=12,
            )
            rebuilt_file = File(
                file_id="test_file_id",
                file_type=FileType.DOCUMENT,
                transfer_method=FileTransferMethod.LOCAL_FILE,
                reference=build_file_reference(record_id="upload-1"),
                filename="test.txt",
                extension=".txt",
                mime_type="text/plain",
                size=12,
                storage_key="canonical-storage-key",
            )
            draft_var = WorkflowDraftVariable(
                app_id="app-1",
            )
            draft_var.set_value(build_segment(persisted_file))
            draft_var._WorkflowDraftVariable__value = None  # pyrefly: ignore[missing-attribute]

            with (
                mock.patch("models.workflow._resolve_workflow_app_tenant_id", return_value="tenant-1"),
                mock.patch("models.workflow.build_file_from_stored_mapping", return_value=rebuilt_file) as rebuild_file,
            ):
                retrieved_segment = draft_var.get_value(session=sqlite_session)

            assert retrieved_segment.value == rebuilt_file
            rebuild_file.assert_called_once()
            assert rebuild_file.call_args.kwargs["tenant_id"] == "tenant-1"

        def test_array_file_variable_rebuilds_each_stored_payload_with_app_tenant(
            self, sqlite_session: Session
        ) -> None:
            files = [
                File(
                    file_id=f"test_file_id_{index}",
                    file_type=FileType.DOCUMENT,
                    transfer_method=FileTransferMethod.LOCAL_FILE,
                    reference=build_file_reference(record_id=f"upload-{index}", storage_key="legacy-storage-key"),
                    filename=f"test_{index}.txt",
                    extension=".txt",
                    mime_type="text/plain",
                    size=12,
                )
                for index in range(2)
            ]
            rebuilt_files = [
                file.model_copy(update={"storage_key": "canonical-storage-key"}, deep=True) for file in files
            ]
            encoded = WorkflowDraftVariable(
                app_id="app-1",
            )
            encoded.set_value(build_segment(files))
            # A fresh instance holding only the persisted columns, so get_value decodes from scratch.
            draft_var = WorkflowDraftVariable(
                app_id="app-1",
            )
            draft_var.value = encoded.value
            draft_var.value_type = encoded.value_type

            with (
                mock.patch("models.workflow._resolve_workflow_app_tenant_id", return_value="tenant-1") as resolve,
                mock.patch("models.workflow.build_file_from_stored_mapping", side_effect=rebuilt_files) as rebuild_file,
            ):
                retrieved_segment = draft_var.get_value(session=sqlite_session)

            assert retrieved_segment.value == rebuilt_files
            resolve.assert_called_once_with(sqlite_session, "app-1")
            assert rebuild_file.call_count == 2
            assert all(call.kwargs["tenant_id"] == "tenant-1" for call in rebuild_file.call_args_list)

        def test_get_and_set_value(self, sqlite_session: Session) -> None:
            draft_var = WorkflowDraftVariable()
            int_var = IntegerSegment(value=1)
            draft_var.set_value(int_var)
            value = draft_var.get_value(session=sqlite_session)
            assert value == int_var


"""Regression coverage for ``models.workflow.WorkflowAppLog`` account accessors.

Ensures the ``@property``→session-parameter refactor preserves the role-based dispatch:
``created_by_account`` looks up an Account only when role is ACCOUNT; ``created_by_end_user``
looks up an EndUser only when role is END_USER.

Both accessors are exercised against the real ``sqlite_session`` fixture (a genuine
SQLAlchemy ``Session`` bound to a pristine full-schema SQLite database) so the assertions
cover actual query behaviour rather than a mock's recorded call.
"""


def _log(role: CreatorUserRole, created_by: str) -> WorkflowAppLog:
    """Construct a WorkflowAppLog without touching the database."""
    return WorkflowAppLog(
        tenant_id="00000000-0000-0000-0000-000000000001",
        app_id="00000000-0000-0000-0000-000000000002",
        workflow_id="00000000-0000-0000-0000-000000000003",
        workflow_run_id="00000000-0000-0000-0000-000000000004",
        created_from=WorkflowAppLogCreatedFrom.WEB_APP,
        created_by_role=role,
        created_by=created_by,
    )


class TestWorkflowAppLogModelsCases:
    class TestCreatedByAccount:
        def test_returns_account_lookup_when_role_is_account(self, sqlite_session: Session) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            log = _log(CreatorUserRole.ACCOUNT, created_by=account.id)

            result = log.created_by_account(session=sqlite_session)

            assert result is not None
            assert result.id == account.id

        def test_returns_none_when_role_is_end_user(self, sqlite_session: Session) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            log = _log(CreatorUserRole.END_USER, created_by=account.id)

            assert log.created_by_account(session=sqlite_session) is None

    class TestCreatedByEndUser:
        def test_returns_end_user_lookup_when_role_is_end_user(self, sqlite_session: Session) -> None:
            end_user = EndUser(
                tenant_id="00000000-0000-0000-0000-000000000001",
                type=EndUserType.BROWSER,
                session_id="session-1",
            )
            sqlite_session.add(end_user)
            sqlite_session.flush()
            log = _log(CreatorUserRole.END_USER, created_by=end_user.id)

            result = log.created_by_end_user(session=sqlite_session)

            assert result is not None
            assert result.id == end_user.id

        def test_returns_none_when_role_is_account(self, sqlite_session: Session) -> None:
            end_user = EndUser(
                tenant_id="00000000-0000-0000-0000-000000000001",
                type=EndUserType.BROWSER,
                session_id="session-1",
            )
            sqlite_session.add(end_user)
            sqlite_session.flush()
            log = _log(CreatorUserRole.ACCOUNT, created_by=end_user.id)

            assert log.created_by_end_user(session=sqlite_session) is None


"""
This unit test verifies Workflow compatibility before and after supporting multiple file types.
"""


OLD_VERSION_WORKFLOW_FEATURES: dict[str, JsonValue] = {
    "file_upload": {
        "image": {
            "enabled": True,
            "number_limits": 6,
            "transfer_methods": ["remote_url", "local_file"],
        }
    },
    "opening_statement": "",
    "retriever_resource": {"enabled": True},
    "sensitive_word_avoidance": {"enabled": False},
    "speech_to_text": {"enabled": False},
    "suggested_questions": [],
    "suggested_questions_after_answer": {"enabled": False},
    "text_to_speech": {"enabled": False, "language": "", "voice": ""},
}

NEW_VERSION_WORKFLOW_FEATURES: dict[str, JsonValue] = {
    "file_upload": {
        "enabled": True,
        "allowed_file_types": ["image"],
        "allowed_file_extensions": [],
        "allowed_file_upload_methods": ["remote_url", "local_file"],
        "number_limits": 6,
    },
    "opening_statement": "",
    "retriever_resource": {"enabled": True},
    "sensitive_word_avoidance": {"enabled": False},
    "speech_to_text": {"enabled": False},
    "suggested_questions": [],
    "suggested_questions_after_answer": {"enabled": False},
    "text_to_speech": {"enabled": False, "language": "", "voice": ""},
}


class TestWorkflowFeatureCompatibilityCases:
    @staticmethod
    def test_workflow_features() -> None:
        workflow = _workflow_feature_compatibility_Workflow(
            tenant_id="",
            app_id="",
            type=WorkflowType.WORKFLOW,
            version="",
            graph="",
            _features=json.dumps(OLD_VERSION_WORKFLOW_FEATURES),
            created_by="",
            _environment_variables=[],
            _conversation_variables=[],
        )

        assert workflow.features_dict == NEW_VERSION_WORKFLOW_FEATURES


"""
Comprehensive unit tests for Workflow models.

This test suite covers:
- Workflow model validation
- WorkflowRun state transitions
- NodeExecution relationships
- Graph configuration validation
"""


class TestWorkflowModelsCases:
    class TestWorkflowModelValidation:
        """Test suite for Workflow model validation and basic operations."""

        def test_workflow_creation_with_required_fields(self) -> None:
            """Test creating a workflow with all required fields."""
            # Arrange
            tenant_id = str(uuid4())
            app_id = str(uuid4())
            created_by = str(uuid4())
            graph = json.dumps({"nodes": [], "edges": []})
            features = json.dumps({"file_upload": {"enabled": True}})

            # Act
            workflow = _workflow_models_Workflow.new(
                tenant_id=tenant_id,
                app_id=app_id,
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph=graph,
                features=features,
                created_by=created_by,
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Assert
            assert workflow.tenant_id == tenant_id
            assert workflow.app_id == app_id
            assert workflow.type == WorkflowType.WORKFLOW
            assert workflow.version == "draft"
            assert workflow.graph == graph
            assert workflow.created_by == created_by
            assert workflow.created_at is not None
            assert workflow.updated_at is not None

        def test_workflow_type_enum_values(self) -> None:
            """Test WorkflowType enum values."""
            # Assert
            assert WorkflowType.WORKFLOW == "workflow"
            assert WorkflowType.CHAT.value == "chat"
            assert WorkflowType.RAG_PIPELINE.value == "rag-pipeline"

        def test_workflow_type_value_of(self) -> None:
            """Test WorkflowType.value_of method."""
            # Act & Assert
            assert WorkflowType.value_of("workflow") == WorkflowType.WORKFLOW
            assert WorkflowType.value_of("chat") == WorkflowType.CHAT
            assert WorkflowType.value_of("rag-pipeline") == WorkflowType.RAG_PIPELINE

            with pytest.raises(ValueError, match="invalid workflow type value"):
                WorkflowType.value_of("invalid_type")

        def test_workflow_graph_dict_property(self) -> None:
            """Test graph_dict property parses JSON correctly."""
            # Arrange
            graph_data = {"nodes": [{"id": "start", "type": "start"}], "edges": list[object]()}
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph=json.dumps(graph_data),
                features="{}",
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Act
            result = workflow.graph_dict

            # Assert
            assert result == graph_data
            assert "nodes" in result
            assert len(result["nodes"]) == 1

        def test_workflow_features_dict_property(self) -> None:
            """Test features_dict property parses JSON correctly."""
            # Arrange
            features_data = {"file_upload": {"enabled": True, "max_files": 5}}
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph="{}",
                features=json.dumps(features_data),
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Act
            result = workflow.features_dict

            # Assert
            assert result == features_data
            assert result["file_upload"]["enabled"] is True
            assert result["file_upload"]["max_files"] == 5

        def test_workflow_with_marked_name_and_comment(self) -> None:
            """Test workflow creation with marked name and comment."""
            # Arrange & Act
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="v1.0",
                graph="{}",
                features="{}",
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
                marked_name="Production Release",
                marked_comment="Initial production version",
            )

            # Assert
            assert workflow.marked_name == "Production Release"
            assert workflow.marked_comment == "Initial production version"

        def test_workflow_version_draft_constant(self) -> None:
            """Test VERSION_DRAFT constant."""
            # Assert
            assert _workflow_models_Workflow.VERSION_DRAFT == "draft"

    class TestWorkflowRunStateTransitions:
        """Test suite for WorkflowRun state transitions and lifecycle."""

        def test_workflow_run_creation_with_required_fields(self) -> None:
            """Test creating a workflow run with required fields."""
            # Arrange
            tenant_id = str(uuid4())
            app_id = str(uuid4())
            workflow_id = str(uuid4())
            created_by = str(uuid4())

            # Act
            workflow_run = WorkflowRun(
                tenant_id=tenant_id,
                app_id=app_id,
                workflow_id=workflow_id,
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
                version="draft",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=created_by,
            )

            # Assert
            assert workflow_run.tenant_id == tenant_id
            assert workflow_run.app_id == app_id
            assert workflow_run.workflow_id == workflow_id
            assert workflow_run.type == WorkflowType.WORKFLOW
            assert workflow_run.triggered_from == WorkflowRunTriggeredFrom.DEBUGGING
            assert workflow_run.status == WorkflowExecutionStatus.RUNNING
            assert workflow_run.created_by == created_by

        def test_workflow_run_state_transition_running_to_succeeded(self) -> None:
            """Test state transition from running to succeeded."""
            # Arrange
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.END_USER,
                created_by=str(uuid4()),
            )

            # Act
            workflow_run.status = WorkflowExecutionStatus.SUCCEEDED
            workflow_run.finished_at = datetime.now(UTC)
            workflow_run.elapsed_time = 2.5

            # Assert
            assert workflow_run.status == WorkflowExecutionStatus.SUCCEEDED
            assert workflow_run.finished_at is not None
            assert workflow_run.elapsed_time == 2.5

        def test_workflow_run_state_transition_running_to_failed(self) -> None:
            """Test state transition from running to failed with error."""
            # Arrange
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
            )

            # Act
            workflow_run.status = WorkflowExecutionStatus.FAILED
            workflow_run.error = "Node execution failed: Invalid input"
            workflow_run.finished_at = datetime.now(UTC)

            # Assert
            assert workflow_run.status == WorkflowExecutionStatus.FAILED
            assert workflow_run.error == "Node execution failed: Invalid input"
            assert workflow_run.finished_at is not None

        def test_workflow_run_state_transition_running_to_stopped(self) -> None:
            """Test state transition from running to stopped."""
            # Arrange
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
                version="draft",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
            )

            # Act
            workflow_run.status = WorkflowExecutionStatus.STOPPED
            workflow_run.finished_at = datetime.now(UTC)

            # Assert
            assert workflow_run.status == WorkflowExecutionStatus.STOPPED
            assert workflow_run.finished_at is not None

        def test_workflow_run_state_transition_running_to_paused(self) -> None:
            """Test state transition from running to paused."""
            # Arrange
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.END_USER,
                created_by=str(uuid4()),
            )

            # Act
            workflow_run.status = WorkflowExecutionStatus.PAUSED

            # Assert
            assert workflow_run.status == WorkflowExecutionStatus.PAUSED
            assert workflow_run.finished_at is None  # Not finished when paused

        def test_workflow_run_state_transition_paused_to_running(self) -> None:
            """Test state transition from paused back to running."""
            # Arrange
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.PAUSED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
            )

            # Act
            workflow_run.status = WorkflowExecutionStatus.RUNNING

            # Assert
            assert workflow_run.status == WorkflowExecutionStatus.RUNNING

        def test_workflow_run_with_partial_succeeded_status(self) -> None:
            """Test workflow run with partial-succeeded status."""
            # Arrange & Act
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.PARTIAL_SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                exceptions_count=2,
            )

            # Assert
            assert workflow_run.status == WorkflowExecutionStatus.PARTIAL_SUCCEEDED
            assert workflow_run.exceptions_count == 2

        def test_workflow_run_with_inputs_and_outputs(self) -> None:
            """Test workflow run with inputs and outputs as JSON."""
            # Arrange
            inputs = {"query": "What is AI?", "context": "technology"}
            outputs = {"answer": "AI is Artificial Intelligence", "confidence": 0.95}

            # Act
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.END_USER,
                created_by=str(uuid4()),
                inputs=json.dumps(inputs),
                outputs=json.dumps(outputs),
            )

            # Assert
            assert workflow_run.inputs_dict == inputs
            assert workflow_run.outputs_dict == outputs

        def test_workflow_run_graph_dict_property(self) -> None:
            """Test graph_dict property for workflow run."""
            # Arrange
            graph = {"nodes": [{"id": "start", "type": "start"}], "edges": list[object]()}
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
                version="draft",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                graph=json.dumps(graph),
            )

            # Act
            result = workflow_run.graph_dict

            # Assert
            assert result == graph
            assert "nodes" in result

        def test_workflow_run_to_dict_serialization(self) -> None:
            """Test WorkflowRun to_dict method."""
            # Arrange
            workflow_run_id = str(uuid4())
            tenant_id = str(uuid4())
            app_id = str(uuid4())
            workflow_id = str(uuid4())
            created_by = str(uuid4())

            workflow_run = WorkflowRun(
                tenant_id=tenant_id,
                app_id=app_id,
                workflow_id=workflow_id,
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=created_by,
                total_tokens=1500,
                total_steps=5,
            )
            workflow_run.id = workflow_run_id

            # Act
            result = workflow_run.to_dict()

            # Assert
            assert result["id"] == workflow_run_id
            assert result["tenant_id"] == tenant_id
            assert result["app_id"] == app_id
            assert result["workflow_id"] == workflow_id
            assert result["status"] == WorkflowExecutionStatus.SUCCEEDED
            assert result["total_tokens"] == 1500
            assert result["total_steps"] == 5

        def test_workflow_run_from_dict_deserialization(self) -> None:
            """Test WorkflowRun from_dict method."""
            # Arrange
            data = {
                "id": str(uuid4()),
                "tenant_id": str(uuid4()),
                "app_id": str(uuid4()),
                "workflow_id": str(uuid4()),
                "type": WorkflowType.WORKFLOW,
                "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
                "version": "v1.0",
                "graph": {"nodes": list[object](), "edges": list[object]()},
                "inputs": {"query": "test"},
                "status": WorkflowExecutionStatus.SUCCEEDED,
                "outputs": {"result": "success"},
                "error": None,
                "elapsed_time": 3.5,
                "total_tokens": 2000,
                "total_steps": 10,
                "created_by_role": CreatorUserRole.ACCOUNT,
                "created_by": str(uuid4()),
                "created_at": datetime.now(UTC),
                "finished_at": datetime.now(UTC),
                "exceptions_count": 0,
            }

            # Act
            workflow_run = WorkflowRun.from_dict(data)

            # Assert
            assert workflow_run.id == data["id"]
            assert workflow_run.workflow_id == data["workflow_id"]
            assert workflow_run.status == WorkflowExecutionStatus.SUCCEEDED
            assert workflow_run.total_tokens == 2000

    class TestNodeExecutionRelationships:
        """Test suite for WorkflowNodeExecutionModel relationships and data."""

        def test_node_execution_creation_with_required_fields(self) -> None:
            """Test creating a node execution with required fields."""
            # Arrange
            tenant_id = str(uuid4())
            app_id = str(uuid4())
            workflow_id = str(uuid4())
            workflow_run_id = str(uuid4())
            created_by = str(uuid4())

            # Act
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=tenant_id,
                app_id=app_id,
                workflow_id=workflow_id,
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=workflow_run_id,
                index=1,
                node_id="start",
                node_type=BuiltinNodeTypes.START,
                title="Start Node",
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=created_by,
            )

            # Assert
            assert node_execution.tenant_id == tenant_id
            assert node_execution.app_id == app_id
            assert node_execution.workflow_id == workflow_id
            assert node_execution.workflow_run_id == workflow_run_id
            assert node_execution.node_id == "start"
            assert node_execution.node_type == BuiltinNodeTypes.START
            assert node_execution.index == 1

        def test_node_execution_with_predecessor_relationship(self) -> None:
            """Test node execution with predecessor node relationship."""
            # Arrange
            predecessor_node_id = "start"
            current_node_id = "llm_1"

            # Act
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=str(uuid4()),
                index=2,
                predecessor_node_id=predecessor_node_id,
                node_id=current_node_id,
                node_type=BuiltinNodeTypes.LLM,
                title="LLM Node",
                status=WorkflowNodeExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
            )

            # Assert
            assert node_execution.predecessor_node_id == predecessor_node_id
            assert node_execution.node_id == current_node_id
            assert node_execution.index == 2

        def test_node_execution_single_step_debugging(self) -> None:
            """Test node execution for single-step debugging (no workflow_run_id)."""
            # Arrange & Act
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
                workflow_run_id=None,  # Single-step has no workflow run
                index=1,
                node_id="llm_test",
                node_type=BuiltinNodeTypes.LLM,
                title="Test LLM",
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
            )

            # Assert
            assert node_execution.triggered_from == WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP.value
            assert node_execution.workflow_run_id is None

        def test_node_execution_with_inputs_outputs_process_data(self) -> None:
            """Test node execution with inputs, outputs, and process_data."""
            # Arrange
            inputs = {"query": "What is AI?", "temperature": 0.7}
            outputs = {"answer": "AI is Artificial Intelligence"}
            process_data = {"tokens_used": 150, "model": "gpt-4"}

            # Act
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=str(uuid4()),
                index=1,
                node_id="llm_1",
                node_type=BuiltinNodeTypes.LLM,
                title="LLM Node",
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                inputs=json.dumps(inputs),
                outputs=json.dumps(outputs),
                process_data=json.dumps(process_data),
            )

            # Assert
            assert node_execution.inputs_dict == inputs
            assert node_execution.outputs_dict == outputs
            assert node_execution.process_data_dict == process_data

        def test_node_execution_status_transitions(self) -> None:
            """Test node execution status transitions."""
            # Arrange
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=str(uuid4()),
                index=1,
                node_id="code_1",
                node_type=BuiltinNodeTypes.CODE,
                title="Code Node",
                status=WorkflowNodeExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
            )

            # Act - transition to succeeded
            node_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
            node_execution.elapsed_time = 1.2
            node_execution.finished_at = datetime.now(UTC)

            # Assert
            assert node_execution.status == WorkflowNodeExecutionStatus.SUCCEEDED
            assert node_execution.elapsed_time == 1.2
            assert node_execution.finished_at is not None

        def test_node_execution_with_error(self) -> None:
            """Test node execution with error status."""
            # Arrange
            error_message = "Code execution failed: SyntaxError on line 5"

            # Act
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=str(uuid4()),
                index=3,
                node_id="code_1",
                node_type=BuiltinNodeTypes.CODE,
                title="Code Node",
                status=WorkflowNodeExecutionStatus.FAILED,
                error=error_message,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
            )

            # Assert
            assert node_execution.status == WorkflowNodeExecutionStatus.FAILED
            assert node_execution.error == error_message

        def test_node_execution_with_metadata(self) -> None:
            """Test node execution with execution metadata."""
            # Arrange
            metadata = {
                "total_tokens": 500,
                "total_price": 0.01,
                "currency": "USD",
                "tool_info": {"provider": "openai", "tool": "gpt-4"},
            }

            # Act
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=str(uuid4()),
                index=1,
                node_id="llm_1",
                node_type=BuiltinNodeTypes.LLM,
                title="LLM Node",
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                execution_metadata=json.dumps(metadata),
            )

            # Assert
            assert node_execution.execution_metadata_dict == metadata
            assert node_execution.execution_metadata_dict["total_tokens"] == 500

        def test_node_execution_metadata_dict_empty(self) -> None:
            """Test execution_metadata_dict returns empty dict when metadata is None."""
            # Arrange
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=str(uuid4()),
                index=1,
                node_id="start",
                node_type=BuiltinNodeTypes.START,
                title="Start",
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                execution_metadata=None,
            )

            # Act
            result = node_execution.execution_metadata_dict

            # Assert
            assert result == {}

        def test_node_execution_different_node_types(self) -> None:
            """Test node execution with different node types."""
            # Test various node types
            node_types = [
                (BuiltinNodeTypes.START, "Start Node"),
                (BuiltinNodeTypes.LLM, "LLM Node"),
                (BuiltinNodeTypes.CODE, "Code Node"),
                (BuiltinNodeTypes.TOOL, "Tool Node"),
                (BuiltinNodeTypes.IF_ELSE, "Conditional Node"),
                (BuiltinNodeTypes.END, "End Node"),
            ]

            for node_type, title in node_types:
                # Act
                node_execution = WorkflowNodeExecutionModel(
                    tenant_id=str(uuid4()),
                    app_id=str(uuid4()),
                    workflow_id=str(uuid4()),
                    triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                    workflow_run_id=str(uuid4()),
                    index=1,
                    node_id=f"{node_type}_1",
                    node_type=node_type,
                    title=title,
                    status=WorkflowNodeExecutionStatus.SUCCEEDED,
                    created_by_role=CreatorUserRole.ACCOUNT,
                    created_by=str(uuid4()),
                )

                # Assert
                assert node_execution.node_type == node_type
                assert node_execution.title == title

    class TestGraphConfigurationValidation:
        """Test suite for graph configuration validation."""

        def test_workflow_graph_with_nodes_and_edges(self) -> None:
            """Test workflow graph configuration with nodes and edges."""
            # Arrange
            graph_config = {
                "nodes": [
                    {"id": "start", "type": "start", "data": {"title": "Start"}},
                    {"id": "llm_1", "type": "llm", "data": {"title": "LLM Node", "model": "gpt-4"}},
                    {"id": "end", "type": "end", "data": {"title": "End"}},
                ],
                "edges": [
                    {"source": "start", "target": "llm_1"},
                    {"source": "llm_1", "target": "end"},
                ],
            }

            # Act
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph=json.dumps(graph_config),
                features="{}",
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Assert
            graph_dict = workflow.graph_dict
            assert len(graph_dict["nodes"]) == 3
            assert len(graph_dict["edges"]) == 2
            assert graph_dict["nodes"][0]["id"] == "start"
            assert graph_dict["edges"][0]["source"] == "start"
            assert graph_dict["edges"][0]["target"] == "llm_1"

        def test_workflow_graph_empty_configuration(self) -> None:
            """Test workflow with empty graph configuration."""
            # Arrange
            graph_config = {"nodes": list[object](), "edges": list[object]()}

            # Act
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph=json.dumps(graph_config),
                features="{}",
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Assert
            graph_dict = workflow.graph_dict
            assert graph_dict["nodes"] == []
            assert graph_dict["edges"] == []

        def test_workflow_graph_complex_node_data(self) -> None:
            """Test workflow graph with complex node data structures."""
            # Arrange
            graph_config = {
                "nodes": [
                    {
                        "id": "llm_1",
                        "type": "llm",
                        "data": {
                            "title": "Advanced LLM",
                            "model": {"provider": "openai", "name": "gpt-4", "mode": "chat"},
                            "prompt_template": [
                                {"role": "system", "text": "You are a helpful assistant"},
                                {"role": "user", "text": "{{query}}"},
                            ],
                            "model_parameters": {"temperature": 0.7, "max_tokens": 2000},
                        },
                    }
                ],
                "edges": list[object](),
            }

            # Act
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph=json.dumps(graph_config),
                features="{}",
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Assert
            graph_dict = workflow.graph_dict
            node_data = graph_dict["nodes"][0]["data"]
            assert node_data["model"]["provider"] == "openai"
            assert node_data["model_parameters"]["temperature"] == 0.7
            assert len(node_data["prompt_template"]) == 2

        def test_workflow_run_graph_preservation(self) -> None:
            """Test that WorkflowRun preserves graph configuration from Workflow."""
            # Arrange
            original_graph = {
                "nodes": [
                    {"id": "start", "type": "start"},
                    {"id": "end", "type": "end"},
                ],
                "edges": [{"source": "start", "target": "end"}],
            }

            # Act
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                graph=json.dumps(original_graph),
            )

            # Assert
            assert workflow_run.graph_dict == original_graph
            assert len(workflow_run.graph_dict["nodes"]) == 2

        def test_workflow_graph_with_conditional_branches(self) -> None:
            """Test workflow graph with conditional branching (if-else)."""
            # Arrange
            graph_config = {
                "nodes": [
                    {"id": "start", "type": "start"},
                    {"id": "if_else_1", "type": "if-else", "data": {"conditions": list[object]()}},
                    {"id": "branch_true", "type": "llm"},
                    {"id": "branch_false", "type": "code"},
                    {"id": "end", "type": "end"},
                ],
                "edges": [
                    {"source": "start", "target": "if_else_1"},
                    {"source": "if_else_1", "sourceHandle": "true", "target": "branch_true"},
                    {"source": "if_else_1", "sourceHandle": "false", "target": "branch_false"},
                    {"source": "branch_true", "target": "end"},
                    {"source": "branch_false", "target": "end"},
                ],
            }

            # Act
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph=json.dumps(graph_config),
                features="{}",
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Assert
            graph_dict = workflow.graph_dict
            assert len(graph_dict["nodes"]) == 5
            assert len(graph_dict["edges"]) == 5
            # Verify conditional edges
            conditional_edges = [e for e in graph_dict["edges"] if "sourceHandle" in e]
            assert len(conditional_edges) == 2

        def test_workflow_graph_with_loop_structure(self) -> None:
            """Test workflow graph with loop/iteration structure."""
            # Arrange
            graph_config = {
                "nodes": [
                    {"id": "start", "type": "start"},
                    {"id": "iteration_1", "type": "iteration", "data": {"iterator": "items"}},
                    {"id": "loop_body", "type": "llm"},
                    {"id": "end", "type": "end"},
                ],
                "edges": [
                    {"source": "start", "target": "iteration_1"},
                    {"source": "iteration_1", "target": "loop_body"},
                    {"source": "loop_body", "target": "iteration_1"},
                    {"source": "iteration_1", "target": "end"},
                ],
            }

            # Act
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph=json.dumps(graph_config),
                features="{}",
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Assert
            graph_dict = workflow.graph_dict
            iteration_node = next(n for n in graph_dict["nodes"] if n["type"] == "iteration")
            assert iteration_node["data"]["iterator"] == "items"

        def test_workflow_graph_dict_with_null_graph(self) -> None:
            """Test graph_dict property when graph is None."""
            # Arrange
            workflow = _workflow_models_Workflow.new(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                version="draft",
                graph=None,  # pyrefly: ignore[bad-argument-type]
                features="{}",
                created_by=str(uuid4()),
                environment_variables=[],
                conversation_variables=[],
                rag_pipeline_variables=[],
            )

            # Act
            result = workflow.graph_dict

            # Assert
            assert result == {}

        def test_workflow_run_inputs_dict_with_null_inputs(self) -> None:
            """Test inputs_dict property when inputs is None."""
            # Arrange
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                inputs=None,
            )

            # Act
            result = workflow_run.inputs_dict

            # Assert
            assert result == {}

        def test_workflow_run_outputs_dict_with_null_outputs(self) -> None:
            """Test outputs_dict property when outputs is None."""
            # Arrange
            workflow_run = WorkflowRun(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                type=WorkflowType.WORKFLOW,
                triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
                version="v1.0",
                status=WorkflowExecutionStatus.RUNNING,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                outputs=None,
            )

            # Act
            result = workflow_run.outputs_dict

            # Assert
            assert result == {}

        def test_node_execution_inputs_dict_with_null_inputs(self) -> None:
            """Test node execution inputs_dict when inputs is None."""
            # Arrange
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=str(uuid4()),
                index=1,
                node_id="start",
                node_type=BuiltinNodeTypes.START,
                title="Start",
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                inputs=None,
            )

            # Act
            result = node_execution.inputs_dict

            # Assert
            assert result is None

        def test_node_execution_outputs_dict_with_null_outputs(self) -> None:
            """Test node execution outputs_dict when outputs is None."""
            # Arrange
            node_execution = WorkflowNodeExecutionModel(
                tenant_id=str(uuid4()),
                app_id=str(uuid4()),
                workflow_id=str(uuid4()),
                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
                workflow_run_id=str(uuid4()),
                index=1,
                node_id="start",
                node_type=BuiltinNodeTypes.START,
                title="Start",
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                created_by_role=CreatorUserRole.ACCOUNT,
                created_by=str(uuid4()),
                outputs=None,
            )

            # Act
            result = node_execution.outputs_dict

            # Assert
            assert result is None


"""Regression coverage for ``WorkflowNodeExecutionModel`` account accessors.

Ensures the ``@property``→session-parameter refactor preserves the role-based dispatch:
``created_by_account`` looks up an Account only when role is ACCOUNT; ``created_by_end_user``
looks up an EndUser only when role is END_USER. Also covers the session-carrying response
source used by the console endpoints that validate ``WorkflowRunNodeExecutionResponse``
with ``from_attributes=True``.
"""


def _execution(role: CreatorUserRole, created_by: str) -> WorkflowNodeExecutionModel:
    """Construct a WorkflowNodeExecutionModel without touching the database."""
    return WorkflowNodeExecutionModel(
        tenant_id="00000000-0000-0000-0000-000000000001",
        app_id="00000000-0000-0000-0000-000000000002",
        workflow_id=str(uuid4()),
        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
        workflow_run_id=None,
        index=1,
        predecessor_node_id=None,
        node_execution_id=None,
        node_id="n1",
        node_type="start",
        title="Start",
        inputs=None,
        process_data=None,
        outputs=None,
        status=WorkflowNodeExecutionStatus.SUCCEEDED,
        error=None,
        elapsed_time=0.0,
        execution_metadata=None,
        created_by_role=role,
        created_by=created_by,
    )


class TestWorkflowNodeExecutionAccessorsCases:
    class TestCreatedByAccount:
        def test_returns_account_lookup_when_role_is_account(self, sqlite_session: Session) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            execution = _execution(CreatorUserRole.ACCOUNT, created_by=account.id)

            result = execution.created_by_account(session=sqlite_session)

            assert result is not None
            assert result.id == account.id

        def test_returns_none_when_role_is_end_user(self, sqlite_session: Session) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            execution = _execution(CreatorUserRole.END_USER, created_by=account.id)

            assert execution.created_by_account(session=sqlite_session) is None

    class TestCreatedByEndUser:
        def test_returns_end_user_lookup_when_role_is_end_user(self, sqlite_session: Session) -> None:
            end_user = EndUser(
                tenant_id="00000000-0000-0000-0000-000000000001",
                type=EndUserType.BROWSER,
                session_id="session-1",
            )
            sqlite_session.add(end_user)
            sqlite_session.flush()
            execution = _execution(CreatorUserRole.END_USER, created_by=end_user.id)

            result = execution.created_by_end_user(session=sqlite_session)

            assert result is not None
            assert result.id == end_user.id

        def test_returns_none_when_role_is_account(self, sqlite_session: Session) -> None:
            end_user = EndUser(
                tenant_id="00000000-0000-0000-0000-000000000001",
                type=EndUserType.BROWSER,
                session_id="session-1",
            )
            sqlite_session.add(end_user)
            sqlite_session.flush()
            execution = _execution(CreatorUserRole.ACCOUNT, created_by=end_user.id)

            assert execution.created_by_end_user(session=sqlite_session) is None

    class TestNodeExecutionResponseSource:
        def test_accessors_resolve_via_wrapped_session_and_other_attributes_proxy(
            self, sqlite_session: Session
        ) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            execution = _execution(CreatorUserRole.ACCOUNT, created_by=account.id)

            source = node_execution_response_source(execution, session=sqlite_session)

            resolved = source.created_by_account
            assert resolved is not None
            assert resolved.id == account.id
            assert source.created_by_end_user is None
            assert source.node_id == "n1"


"""SQLite-backed tests for workflow node execution offload behavior."""


TABLES = (WorkflowNodeExecutionModel, WorkflowNodeExecutionOffload, UploadFile)


def _persist_upload_file(session: Session, *, key: str = "offload/process-data.json") -> UploadFile:
    upload_file = UploadFile(
        tenant_id="tenant-1",
        storage_type=StorageType.LOCAL,
        key=key,
        name="process-data.json",
        size=18,
        extension="json",
        mime_type="application/json",
        created_by_role=CreatorUserRole.ACCOUNT,
        created_by="account-1",
        created_at=datetime.now(UTC),
        used=True,
    )
    session.add(upload_file)
    session.flush()
    return upload_file


def _persist_execution(
    session: Session,
    *,
    process_data: str = '{"test": "data"}',
    offloads: tuple[tuple[ExecutionOffLoadType, str], ...] = (),
) -> WorkflowNodeExecutionModel:
    execution = WorkflowNodeExecutionModel(
        tenant_id="tenant-1",
        app_id="app-1",
        workflow_id="workflow-1",
        triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
        workflow_run_id="run-1",
        index=1,
        predecessor_node_id=None,
        node_execution_id="node-execution-1",
        node_id="node-1",
        node_type="code",
        title="Code",
        inputs='{"input": "value"}',
        process_data=process_data,
        outputs='{"output": "value"}',
        status=WorkflowNodeExecutionStatus.SUCCEEDED,
        error=None,
        elapsed_time=0,
        execution_metadata=None,
        created_by_role=CreatorUserRole.ACCOUNT,
        created_by="account-1",
        finished_at=None,
    )
    session.add(execution)
    session.flush()

    session.add_all(
        [
            WorkflowNodeExecutionOffload(
                tenant_id=execution.tenant_id,
                app_id=execution.app_id,
                node_execution_id=execution.id,
                type_=type_,
                file_id=file_id,
            )
            for type_, file_id in offloads
        ]
    )
    session.commit()
    session.expunge_all()

    stmt = WorkflowNodeExecutionModel.preload_offload_data(
        select(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id == execution.id)
    )
    loaded_execution = session.scalar(stmt)
    assert loaded_execution is not None
    return loaded_execution


class TestWorkflowNodeExecutionOffloadCases:
    @pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
    class TestWorkflowNodeExecutionModel:
        """Exercise truncation flags and content restoration using persisted offload rows."""

        def test_process_data_truncated_property_false_when_no_offload_data(self, sqlite_session: Session) -> None:
            execution = _persist_execution(sqlite_session)

            assert execution.process_data_truncated is False

        def test_process_data_truncated_property_false_when_no_process_data_file(self, sqlite_session: Session) -> None:
            execution = _persist_execution(
                sqlite_session,
                offloads=(
                    (ExecutionOffLoadType.INPUTS, "inputs-file"),
                    (ExecutionOffLoadType.OUTPUTS, "outputs-file"),
                ),
            )

            assert execution.process_data_truncated is False

        def test_process_data_truncated_property_true_when_process_data_file_exists(
            self, sqlite_session: Session
        ) -> None:
            execution = _persist_execution(
                sqlite_session,
                offloads=((ExecutionOffLoadType.PROCESS_DATA, "process-data-file-id"),),
            )

            assert execution.process_data_truncated is True

        def test_load_full_process_data_with_no_offload_data(self, sqlite_session: Session) -> None:
            execution = _persist_execution(sqlite_session)
            storage = Mock()

            result = execution.load_full_process_data(sqlite_session, storage)

            assert result == {"test": "data"}
            storage.load.assert_not_called()

        def test_load_full_process_data_with_no_file(self, sqlite_session: Session) -> None:
            execution = _persist_execution(
                sqlite_session,
                offloads=((ExecutionOffLoadType.INPUTS, "inputs-file"),),
            )
            storage = Mock()

            result = execution.load_full_process_data(sqlite_session, storage)

            assert result == {"test": "data"}
            storage.load.assert_not_called()

        def test_load_full_process_data_with_file(self, sqlite_session: Session) -> None:
            upload_file = _persist_upload_file(sqlite_session)
            execution = _persist_execution(
                sqlite_session,
                process_data='{"truncated": true}',
                offloads=((ExecutionOffLoadType.PROCESS_DATA, upload_file.id),),
            )
            full_process_data = {"full": "data", "large_field": "x" * 10000}
            storage = Mock()
            storage.load.return_value = json.dumps(full_process_data)

            result = execution.load_full_process_data(sqlite_session, storage)

            assert result == full_process_data
            storage.load.assert_called_once_with(upload_file.key)

        def test_consistency_with_inputs_outputs_truncation(self, sqlite_session: Session) -> None:
            execution = _persist_execution(
                sqlite_session,
                offloads=(
                    (ExecutionOffLoadType.INPUTS, "inputs-file"),
                    (ExecutionOffLoadType.OUTPUTS, "outputs-file"),
                    (ExecutionOffLoadType.PROCESS_DATA, "process-data-file"),
                ),
            )

            assert execution.inputs_truncated is True
            assert execution.outputs_truncated is True
            assert execution.process_data_truncated is True

        def test_mixed_truncation_states(self, sqlite_session: Session) -> None:
            execution = _persist_execution(
                sqlite_session,
                offloads=((ExecutionOffLoadType.PROCESS_DATA, "process-data-file"),),
            )

            assert execution.inputs_truncated is False
            assert execution.outputs_truncated is False
            assert execution.process_data_truncated is True

        def test_preload_offload_data_and_files_loads_process_data_file(self, sqlite_session: Session) -> None:
            upload_file = _persist_upload_file(sqlite_session)
            execution = _persist_execution(
                sqlite_session,
                offloads=((ExecutionOffLoadType.PROCESS_DATA, upload_file.id),),
            )
            execution_id = execution.id
            sqlite_session.expunge_all()

            stmt = WorkflowNodeExecutionModel.preload_offload_data_and_files(
                select(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id == execution_id)
            )
            preloaded_execution = sqlite_session.scalar(stmt)

            assert preloaded_execution is not None
            assert len(preloaded_execution.offload_data) == 1
            assert preloaded_execution.offload_data[0].file is not None
            assert preloaded_execution.offload_data[0].file.key == upload_file.key


"""Regression coverage for ``WorkflowRun`` account accessors.

Ensures the ``@property``→session-parameter refactor preserves the role-based dispatch:
``created_by_account`` looks up an Account only when role is ACCOUNT; ``created_by_end_user``
looks up an EndUser only when role is END_USER. Also covers the session-carrying response
sources used by the console endpoints that validate the run detail / pagination responses
with ``from_attributes=True``.
"""


def _run(role: CreatorUserRole, created_by: str) -> WorkflowRun:
    """Construct a WorkflowRun without touching the database."""
    return WorkflowRun(
        id=str(uuid4()),
        tenant_id="00000000-0000-0000-0000-000000000001",
        app_id="00000000-0000-0000-0000-000000000002",
        workflow_id=str(uuid4()),
        type=ModelWorkflowType.WORKFLOW,
        triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
        version="1.0",
        status=WorkflowExecutionStatus.SUCCEEDED,
        created_by_role=role,
        created_by=created_by,
    )


class TestWorkflowRunAccessorsCases:
    class TestCreatedByAccount:
        def test_returns_account_lookup_when_role_is_account(self, sqlite_session: Session) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            run = _run(CreatorUserRole.ACCOUNT, created_by=account.id)

            result = run.created_by_account(session=sqlite_session)

            assert result is not None
            assert result.id == account.id

        def test_returns_none_when_role_is_end_user(self, sqlite_session: Session) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            run = _run(CreatorUserRole.END_USER, created_by=account.id)

            assert run.created_by_account(session=sqlite_session) is None

    class TestCreatedByEndUser:
        def test_returns_end_user_lookup_when_role_is_end_user(self, sqlite_session: Session) -> None:
            end_user = EndUser(
                tenant_id="00000000-0000-0000-0000-000000000001",
                type=EndUserType.BROWSER,
                session_id="session-1",
            )
            sqlite_session.add(end_user)
            sqlite_session.flush()
            run = _run(CreatorUserRole.END_USER, created_by=end_user.id)

            result = run.created_by_end_user(session=sqlite_session)

            assert result is not None
            assert result.id == end_user.id

        def test_returns_none_when_role_is_account(self, sqlite_session: Session) -> None:
            end_user = EndUser(
                tenant_id="00000000-0000-0000-0000-000000000001",
                type=EndUserType.BROWSER,
                session_id="session-1",
            )
            sqlite_session.add(end_user)
            sqlite_session.flush()
            run = _run(CreatorUserRole.ACCOUNT, created_by=end_user.id)

            assert run.created_by_end_user(session=sqlite_session) is None

    class TestWorkflowRunResponseSources:
        def test_detail_source_resolves_accessors_and_proxies_other_attributes(self, sqlite_session: Session) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            run = _run(CreatorUserRole.ACCOUNT, created_by=account.id)

            source = workflow_run_response_source(run, session=sqlite_session)

            resolved = source.created_by_account
            assert resolved is not None
            assert resolved.id == account.id
            assert source.created_by_end_user is None
            assert source.version == "1.0"

        def test_pagination_source_wraps_each_run(self, sqlite_session: Session) -> None:
            account = Account(name="Test Account", email="test@example.com")
            sqlite_session.add(account)
            sqlite_session.flush()
            runs = [_run(CreatorUserRole.ACCOUNT, created_by=account.id) for _ in range(2)]
            pagination = SimpleNamespace(data=runs, limit=20, has_more=False)

            payload = workflow_run_pagination_response_source(pagination, session=sqlite_session)

            assert payload["limit"] == 20
            assert payload["has_more"] is False
            assert [item.created_by_account.id for item in payload["data"]] == [account.id, account.id]


@pytest.mark.parametrize(
    ("features", "expected"),
    [
        pytest.param({}, {}, id="missing-upload"),
        pytest.param({"file_upload": None}, {"file_upload": None}, id="null-upload"),
        pytest.param({"file_upload": False}, {"file_upload": False}, id="non-object-upload"),
        pytest.param({"file_upload": {}}, {"file_upload": {}}, id="missing-image"),
        pytest.param({"file_upload": {"image": None}}, {"file_upload": {"image": None}}, id="null-image"),
        pytest.param({"file_upload": {"image": False}}, {"file_upload": {"image": False}}, id="non-object-image"),
        pytest.param(
            {"file_upload": {"enabled": False, "image": {"enabled": False}}},
            {"file_upload": {"enabled": False, "image": {"enabled": False}}},
            id="disabled-image",
        ),
        pytest.param(OLD_VERSION_WORKFLOW_FEATURES, NEW_VERSION_WORKFLOW_FEATURES, id="legacy-image"),
        pytest.param(NEW_VERSION_WORKFLOW_FEATURES, NEW_VERSION_WORKFLOW_FEATURES, id="modern-upload"),
    ],
)
def test_normalized_features_preserves_stored_payload_without_dirtying_workflow(
    sqlite_session: Session,
    features: dict[str, JsonValue],
    expected: dict[str, JsonValue],
) -> None:
    stored_features = json.dumps(features)
    workflow = _workflow_feature_compatibility_Workflow(
        tenant_id=str(uuid4()),
        app_id=str(uuid4()),
        type=WorkflowType.CHAT,
        version="published",
        graph="{}",
        _features=stored_features,
        created_by=str(uuid4()),
    )
    sqlite_session.add(workflow)
    sqlite_session.commit()

    assert workflow.normalized_features_dict == expected
    assert workflow.serialized_features == stored_features
    assert workflow not in sqlite_session.dirty

    sqlite_session.expire(workflow)
    assert workflow.serialized_features == stored_features


class TestWorkflowTypeBaseCases:
    @staticmethod
    def test_workflow_models_use_typebase_registry_and_generated_defaults() -> None:
        models = (
            _workflow_Workflow,
            WorkflowVersionCounter,
            WorkflowRun,
            WorkflowNodeExecutionModel,
            WorkflowDraftVariable,
        )

        typebase_models = {mapper.class_ for mapper in TypeBase.registry.mappers}
        base_models = {mapper.class_ for mapper in Base.registry.mappers}

        assert all(model in typebase_models for model in models)
        assert all(model not in base_models for model in models)

        instances = (_workflow_Workflow(), WorkflowRun(), WorkflowNodeExecutionModel(), WorkflowDraftVariable())
        assert all(instance.id for instance in instances)
        assert len({instance.id for instance in instances}) == len(instances)
        assert isinstance(instances[0].created_at, datetime)
        assert isinstance(instances[-1].created_at, datetime)

    @staticmethod
    def test_workflow_constructor_serializes_variable_collections_and_configures_offload_relationship() -> None:
        workflow = _workflow_Workflow(
            _features="{}",
            _environment_variables=[],
            _conversation_variables=[],
            _rag_pipeline_variables=[],
        )

        assert workflow._features == "{}"
        assert json.loads(workflow._environment_variables) == {}
        assert json.loads(workflow._conversation_variables) == {}
        assert json.loads(workflow._rag_pipeline_variables) == {}

        configure_mappers()
        relationship = inspect(WorkflowNodeExecutionModel).relationships["offload_data"]
        assert relationship.mapper.class_ is WorkflowNodeExecutionOffload
        assert relationship.back_populates == "execution"


@pytest.mark.parametrize("sqlite_session", [(App,)], indirect=True)
def test_resolve_workflow_app_tenant_id_reads_through_caller_session(sqlite_session: Session) -> None:
    app = App(
        id="app-1",
        tenant_id="tenant-1",
        name="Workflow",
        mode=AppMode.WORKFLOW,
        enable_site=True,
        enable_api=True,
    )
    sqlite_session.add(app)
    sqlite_session.flush()

    assert _resolve_workflow_app_tenant_id(sqlite_session, "app-1") == "tenant-1"

    with pytest.raises(ValueError, match="Unable to resolve tenant_id"):
        _resolve_workflow_app_tenant_id(sqlite_session, "missing-app")