refactor(api): add Workflow API gateways and credential readers

api/extensions/application_services/human_input.py

@@ -0,0 +1,32 @@
+"""Compose human-input debug dependencies."""
+
+from functools import partial
+
+from sqlalchemy.orm import Session, sessionmaker
+
+from core.app.file_access import DatabaseFileAccessController
+from extensions.application_services.workflow_variables import build_workflow_variable_service
+from repositories.human_input.form_repository import HumanInputFormRepositoryImpl, HumanInputFormSubmissionRepository
+from repositories.workflow.definition_repository import WorkflowDefinitionRepository
+from services.human_input.debug_service import HumanInputDebugService
+from services.human_input.file_gateway import HumanInputDebugFileGateway
+from services.human_input_delivery_test_service import (
+    DeliveryTestRegistry,
+    EmailDeliveryTestHandler,
+    HumanInputDeliveryTestService,
+)
+from services.human_input_service import HumanInputService
+
+
+def build_human_input_debug_service(*, database_client: sessionmaker[Session]) -> HumanInputDebugService:
+    return HumanInputDebugService(
+        definitions=WorkflowDefinitionRepository(session_factory=database_client),
+        variables=build_workflow_variable_service(database_client=database_client),
+        files=HumanInputDebugFileGateway(sessions=database_client, access=DatabaseFileAccessController()),
+        submissions=HumanInputService(
+            session_factory=database_client,
+            form_repository=HumanInputFormSubmissionRepository(sessions=database_client),
+        ),
+        forms=partial(HumanInputFormRepositoryImpl, sessions=database_client),
+        delivery=HumanInputDeliveryTestService(DeliveryTestRegistry([EmailDeliveryTestHandler(database_client)])),
+    )

api/fields/workflow_fields.py

@@ -0,0 +1,179 @@
+"""Shared workflow response schemas for Console, snippets and RAG pipelines."""
+
+from datetime import datetime
+from typing import Any, NotRequired, TypedDict
+
+from pydantic import AliasChoices, Field, RootModel, field_validator
+
+from core.helper import encrypter
+from core.workflow.llm_environment_variable import LLM_ENVIRONMENT_VARIABLE_VALUE_TYPE, LLMEnvironmentVariable
+from fields.base import ResponseModel
+from fields.conversation_variable_fields import WorkflowConversationVariableResponse
+from fields.member_fields import SimpleAccount
+from graphon.variables import SecretVariable, SegmentType, VariableBase
+from libs.helper import to_timestamp
+
+ENVIRONMENT_VARIABLE_SUPPORTED_TYPES = (SegmentType.STRING, SegmentType.NUMBER, SegmentType.SECRET)
+
+
+class EnvironmentVariableResponseDict(TypedDict):
+    value_type: str
+    id: NotRequired[str]
+    name: NotRequired[str]
+    value: NotRequired[Any]
+    description: NotRequired[str | None]
+
+
+class PipelineVariableResponse(ResponseModel):
+    label: str
+    variable: str
+    type: str
+    belong_to_node_id: str
+    max_length: int | None = None
+    required: bool
+    unit: str | None = None
+    default_value: Any = Field(default=None)
+    options: list[str] | None = None
+    placeholder: str | None = None
+    tooltips: str | None = None
+    allowed_file_types: list[str] | None = None
+    allowed_file_extensions: list[str] | None = Field(
+        default=None, validation_alias=AliasChoices("allowed_file_extensions", "allow_file_extension")
+    )
+    allowed_file_upload_methods: list[str] | None = Field(
+        default=None, validation_alias=AliasChoices("allowed_file_upload_methods", "allow_file_upload_methods")
+    )
+
+
+class WorkflowEnvironmentVariableResponse(ResponseModel):
+    value_type: str
+    id: str
+    name: str
+    value: Any
+    description: str
+
+
+class WorkflowResponse(ResponseModel):
+    id: str
+    graph: dict[str, Any] = Field(
+        validation_alias=AliasChoices("graph_dict", "graph"),
+    )
+    features: dict[str, Any] = Field(
+        validation_alias=AliasChoices("features_dict", "features"),
+    )
+    hash: str = Field(validation_alias=AliasChoices("unique_hash", "hash"))
+    version: str
+    # NULL for drafts and for versions published before numbering was introduced; those
+    # render as "Untitled Version" instead of `#N`. Never 0, so clients must test for null.
+    version_number: int | None = None
+    marked_name: str
+    marked_comment: str
+    created_by: SimpleAccount | None = Field(
+        default=None, validation_alias=AliasChoices("created_by_account", "created_by")
+    )
+    created_at: int
+    updated_by: SimpleAccount | None = Field(
+        default=None, validation_alias=AliasChoices("updated_by_account", "updated_by")
+    )
+    updated_at: int
+    tool_published: bool
+    environment_variables: list[WorkflowEnvironmentVariableResponse]
+    conversation_variables: list[WorkflowConversationVariableResponse]
+    rag_pipeline_variables: list[PipelineVariableResponse]
+
+    @field_validator("created_at", "updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int | None) -> int:
+        timestamp = to_timestamp(value)
+        if timestamp is None:
+            raise ValueError("timestamp is required")
+        return timestamp
+
+    @field_validator("environment_variables", mode="before")
+    @classmethod
+    def _serialize_environment_variables(cls, value: Any) -> list[Any]:
+        if value is None:
+            return []
+
+        return [_serialize_environment_variable(item) for item in value]
+
+
+class WorkflowPaginationResponse(ResponseModel):
+    items: list[WorkflowResponse]
+    page: int
+    limit: int
+    has_more: bool
+
+
+def _serialize_environment_variable(value: Any) -> EnvironmentVariableResponseDict | Any:
+    match value:
+        case LLMEnvironmentVariable():
+            return {
+                "id": value.id,
+                "name": value.name,
+                "value": value.value,
+                "value_type": LLM_ENVIRONMENT_VARIABLE_VALUE_TYPE,
+                "description": value.description,
+            }
+
+        case SecretVariable():
+            return {
+                "id": value.id,
+                "name": value.name,
+                "value": encrypter.full_mask_token(),
+                "value_type": value.value_type.value,
+                "description": value.description,
+            }
+
+        case VariableBase():
+            return {
+                "id": value.id,
+                "name": value.name,
+                "value": value.value,
+                "value_type": str(value.value_type.exposed_type()),
+                "description": value.description,
+            }
+
+        case dict():
+            value_type_str = value.get("value_type")
+            if not isinstance(value_type_str, str):
+                raise TypeError(
+                    f"unexpected type for value_type field, value={value_type_str}, type={type(value_type_str)}"
+                )
+            if value_type_str == LLM_ENVIRONMENT_VARIABLE_VALUE_TYPE:
+                return value
+            value_type = SegmentType(value_type_str).exposed_type()
+            if value_type not in ENVIRONMENT_VARIABLE_SUPPORTED_TYPES:
+                raise ValueError(f"Unsupported environment variable value type: {value_type}")
+            return value
+
+        case _:
+            return value
+
+
+class WorkflowPublishResponse(ResponseModel):
+    result: str
+    created_at: int
+    warning: str | None = Field(
+        default=None,
+        description="Advisory warning for variable references that can read a skipped branch. Publish still succeeds.",
+    )
+
+
+class WorkflowRestoreResponse(ResponseModel):
+    result: str
+    hash: str
+    updated_at: int
+
+
+class DefaultBlockConfigItemResponse(ResponseModel):
+    type: str
+    config: dict[str, Any]
+
+
+class DefaultBlockConfigsResponse(RootModel[list[DefaultBlockConfigItemResponse]]):
+    pass
+
+
+class DefaultBlockConfigResponse(RootModel[dict[str, Any]]):
+    root: dict[str, Any]

api/models/human_input_contracts.py

@@ -106,6 +106,66 @@ def submitted(self) -> bool:
         return self.submitted_at is not None
 
 
+class HumanInputFormRecordView(Protocol):
+    """Read-only submission state consumed by the Human Input application service."""
+
+    @property
+    def form_id(self) -> str: ...
+
+    @property
+    def workflow_run_id(self) -> str | None: ...
+
+    @property
+    def conversation_id(self) -> str | None: ...
+
+    @property
+    def tenant_id(self) -> str: ...
+
+    @property
+    def app_id(self) -> str: ...
+
+    @property
+    def form_kind(self) -> HumanInputFormKind: ...
+
+    @property
+    def definition(self) -> FormDefinition: ...
+
+    @property
+    def created_at(self) -> datetime: ...
+
+    @property
+    def expiration_time(self) -> datetime: ...
+
+    @property
+    def status(self) -> HumanInputFormStatus: ...
+
+    @property
+    def recipient_id(self) -> str | None: ...
+
+    @property
+    def recipient_type(self) -> RecipientType | None: ...
+
+    @property
+    def submitted(self) -> bool: ...
+
+
+class HumanInputFormSubmissionStore(Protocol):
+    """Persistence port shared by legacy and session-bound Human Input repositories."""
+
+    def get_by_token(self, form_token: str) -> HumanInputFormRecordView | None: ...
+
+    def mark_submitted(
+        self,
+        *,
+        form_id: str,
+        recipient_id: str | None,
+        selected_action_id: str,
+        form_data: Mapping[str, Any],
+        submission_user_id: str | None,
+        submission_end_user_id: str | None,
+    ) -> HumanInputFormRecordView: ...
+
+
 @dataclasses.dataclass(frozen=True)
 class PauseFormSnapshot:
     id: str

api/repositories/app/agent_app_repository.py

@@ -18,6 +18,11 @@ class AgentAppRepository:
     def __init__(self, *, session_factory: sessionmaker[Session]) -> None:
         self._session_factory = session_factory
 
+    def app_tenant_id(self, app_id: str) -> str | None:
+        """Read ownership only, allowing the trusted inner API to distinguish 403 from 404."""
+        with self._session_factory() as session:
+            return session.scalar(select(App.tenant_id).where(App.id == app_id))
+
     def resolve_roster_app_id(self, *, tenant_id: str, agent_id: str) -> str:
         """Resolve a roster Agent's configuration owner without materializing inline Apps."""
         with self._session_factory() as session:

api/repositories/credentials/query_repository.py

@@ -11,10 +11,11 @@
 from models.credential_permission import CredentialPermission, CredentialType
 from models.enums import PermissionEnum
 from models.oauth import DatasourceProvider
-from models.provider import ProviderCredential
+from models.provider import LoadBalancingModelConfig, ProviderCredential
 from models.provider_ids import ModelProviderID
 from models.tools import BuiltinToolProvider
 from models.trigger import TriggerSubscription, WorkflowPluginTrigger
+from repositories.tools.provider_repository import select_builtin_credentials
 from services.credentials.query import (
     CredentialQuery,
     DatasourceCredentialListItem,
@@ -118,6 +119,34 @@ def update_tool_credential_if_unchanged(
             assert isinstance(result, CursorResult)
             return result.rowcount == 1
 
+    @override
+    def list_model_load_balancing_credential_ids(
+        self, *, workspace_id: str, provider: str, model: str
+    ) -> Sequence[str]:
+        # Both predefined and custom model credentials need policy validation.
+        # Only IDs are needed; publication must not fetch/decrypt credential forms.
+        names = [provider]
+        provider_id = ModelProviderID(provider)
+        if provider_id.is_langgenius():
+            names.append(provider_id.provider_name if "/" in provider else str(provider_id))
+        with self._session_factory() as session:
+            return [
+                credential_id
+                for credential_id in session.scalars(
+                    select(LoadBalancingModelConfig.credential_id)
+                    .where(
+                        LoadBalancingModelConfig.tenant_id == workspace_id,
+                        LoadBalancingModelConfig.provider_name.in_(names),
+                        LoadBalancingModelConfig.model_type == "llm",
+                        LoadBalancingModelConfig.model_name == model,
+                        LoadBalancingModelConfig.credential_id.is_not(None),
+                        LoadBalancingModelConfig.credential_id != "",
+                    )
+                    .distinct()
+                )
+                if credential_id is not None
+            ]
+
     @override
     def list_models(self, *, workspace_id: str, provider: str, actor_id: str | None) -> Sequence[ModelCredentialRecord]:
         names = [provider]
@@ -185,6 +214,15 @@ def list_datasources(
                 for row in rows
             ]
 
+    @override
+    def default_tool_credential_id(self, *, workspace_id: str, provider: str) -> str | None:
+        with self._session_factory() as session:
+            return session.scalar(
+                select_builtin_credentials(tenant_id=workspace_id, provider_names=[provider])
+                .with_only_columns(BuiltinToolProvider.id)
+                .limit(1)
+            )
+
     @override
     def list_tools(
         self,