chore(api): upgrade graphon to v0.6.0, migrate HITL logic back to Dify (#38247)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

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

@@ -74,9 +74,9 @@ from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories.human_input_repository import HumanInputFormRepositoryImpl
 from core.workflow.file_reference import resolve_file_record_id
+from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
 from core.workflow.system_variables import build_system_variables
 from extensions.ext_database import db
-from graphon.entities.pause_reason import HumanInputRequired
 from graphon.enums import WorkflowExecutionStatus
 from graphon.model_runtime.entities.llm_entities import LLMUsage
 from graphon.model_runtime.utils.encoders import jsonable_encoder

api/core/app/apps/common/workflow_response_converter.py

@@ -60,11 +60,11 @@ from core.workflow.human_input_policy import (
     enrich_human_input_pause_reasons,
     resolve_human_input_pause_reason_inputs,
 )
+from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
 from core.workflow.system_variables import SystemVariableKey, system_variables_to_mapping
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from graphon.entities import WorkflowStartReason
-from graphon.entities.pause_reason import HumanInputRequired
 from graphon.enums import (
     BuiltinNodeTypes,
     WorkflowExecutionStatus,

api/core/app/apps/workflow_app_runner.py

@@ -34,12 +34,15 @@ from core.app.entities.queue_entities import (
     QueueWorkflowSucceededEvent,
 )
 from core.rag.entities import RetrievalSourceMetadata
+from core.repositories.human_input_repository import HumanInputFormSubmissionRepository
 from core.workflow.node_factory import (
     DifyGraphInitContext,
     DifyNodeFactory,
     get_default_root_node_id,
     resolve_workflow_node_class,
 )
+from core.workflow.nodes.human_input.boundary import enrich_graph_pause_reasons
+from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
 from core.workflow.system_variables import (
     build_bootstrap_variables,
     default_system_variables,
@@ -51,7 +54,6 @@ from core.workflow.variable_pool_initializer import add_variables_to_pool
 from core.workflow.workflow_entry import WorkflowEntry
 from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
 from graphon.entities.graph_config import NodeConfigDictAdapter
-from graphon.entities.pause_reason import HumanInputRequired
 from graphon.graph import Graph
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import (
@@ -426,10 +428,15 @@ class WorkflowBasedAppRunner:
             case GraphRunPausedEvent():
                 runtime_state = workflow_entry.graph_engine.graph_runtime_state
                 paused_nodes = runtime_state.get_paused_nodes()
-                self._enqueue_human_input_notifications(event.reasons)
+                enriched_reasons = enrich_graph_pause_reasons(
+                    reasons=event.reasons,
+                    form_repository=HumanInputFormSubmissionRepository(),
+                    variable_pool=runtime_state.variable_pool,
+                )
+                self._enqueue_human_input_notifications(enriched_reasons)
                 self._publish_event(
                     QueueWorkflowPausedEvent(
-                        reasons=event.reasons,
+                        reasons=enriched_reasons,
                         outputs=event.outputs,
                         paused_nodes=paused_nodes,
                     )

api/core/app/entities/queue_entities.py

@@ -7,8 +7,8 @@ from pydantic import BaseModel, ConfigDict, Field
 
 from core.app.entities.agent_strategy import AgentStrategyInfo
 from core.rag.entities import RetrievalSourceMetadata
+from core.workflow.nodes.human_input.pause_reason import PauseReason
 from graphon.entities import WorkflowStartReason
-from graphon.entities.pause_reason import PauseReason
 from graphon.enums import NodeType, WorkflowNodeExecutionMetadataKey
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk
 from graphon.variables.segments import Segment

api/core/app/entities/task_entities.py

@@ -6,11 +6,11 @@ from pydantic import BaseModel, ConfigDict, Field, JsonValue
 
 from core.app.entities.agent_strategy import AgentStrategyInfo
 from core.rag.entities import RetrievalSourceMetadata
+from core.workflow.nodes.human_input.entities import FormInputConfig, UserActionConfig
+from core.workflow.nodes.human_input.pause_reason import DifyHITLEventType
 from graphon.entities import WorkflowStartReason
-from graphon.entities.pause_reason import PauseReasonType
 from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
-from graphon.nodes.human_input.entities import FormInputConfig, UserActionConfig
 
 
 class AnnotationReplyAccount(BaseModel):
@@ -307,7 +307,7 @@ class HumanInputRequiredPauseReasonPayload(BaseModel):
     ``human_input_required`` events are available.
     """
 
-    TYPE: Literal[PauseReasonType.HUMAN_INPUT_REQUIRED] = PauseReasonType.HUMAN_INPUT_REQUIRED
+    TYPE: Literal[DifyHITLEventType.HUMAN_INPUT_REQUIRED] = DifyHITLEventType.HUMAN_INPUT_REQUIRED
     form_id: str
     node_id: str
     node_title: str

api/core/app/layers/pause_state_persist_layer.py

@@ -6,6 +6,8 @@ from sqlalchemy import Engine
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
+from core.repositories.human_input_repository import HumanInputFormSubmissionRepository
+from core.workflow.nodes.human_input.boundary import enrich_graph_pause_reasons
 from core.workflow.system_variables import SystemVariableKey, get_system_text
 from graphon.graph_engine.layers import GraphEngineLayer
 from graphon.graph_events import GraphEngineEvent, GraphRunPausedEvent
@@ -126,12 +128,20 @@ class PauseStatePersistenceLayer(GraphEngineLayer):
             SystemVariableKey.WORKFLOW_EXECUTION_ID,
         )
         assert workflow_run_id is not None
+        # NOTE(QuantumGhost): Dify owns the pause-reason semantics that cross the
+        # persistence boundary. Graphon session ids are translated back to form ids
+        # here so repository/model layers only handle Dify-owned pause reasons.
+        pause_reasons = enrich_graph_pause_reasons(
+            reasons=event.reasons,
+            form_repository=HumanInputFormSubmissionRepository(),
+            variable_pool=self.graph_runtime_state.variable_pool,
+        )
         repo = self._get_repo()
         repo.create_workflow_pause(
             workflow_run_id=workflow_run_id,
             state_owner_user_id=self._state_owner_user_id,
             state=state.dumps(),
-            pause_reasons=event.reasons,
+            pause_reasons=pause_reasons,
         )
 
     @override

api/core/entities/execution_extra_content.py

@@ -5,7 +5,7 @@ from typing import Any, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, Field, JsonValue
 
-from graphon.nodes.human_input.entities import FormInputConfig, UserActionConfig
+from core.workflow.nodes.human_input.entities import FormInputConfig, UserActionConfig
 from models.execution_extra_content import ExecutionContentType
 
 

api/core/plugin/impl/model_runtime.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import hashlib
 import logging
-from collections.abc import Generator, Iterable, Sequence
+from collections.abc import Generator, Iterable, Mapping, Sequence
 from typing import IO, Any, Literal, cast, overload, override
 
 from pydantic import ValidationError
@@ -285,6 +285,7 @@ class PluginModelRuntime(ModelRuntime):
         tools: list[PromptMessageTool] | None,
         stop: Sequence[str] | None,
         stream: Literal[False],
+        request_metadata: Mapping[str, object] | None = None,
     ) -> LLMResult: ...
 
     @overload
@@ -299,6 +300,7 @@ class PluginModelRuntime(ModelRuntime):
         tools: list[PromptMessageTool] | None,
         stop: Sequence[str] | None,
         stream: Literal[True],
+        request_metadata: Mapping[str, object] | None = None,
     ) -> Generator[LLMResultChunk, None, None]: ...
 
     @override
@@ -313,7 +315,9 @@ class PluginModelRuntime(ModelRuntime):
         tools: list[PromptMessageTool] | None,
         stop: Sequence[str] | None,
         stream: bool,
+        request_metadata: Mapping[str, object] | None = None,
     ) -> LLMResult | Generator[LLMResultChunk, None, None]:
+        del request_metadata
         plugin_id, provider_name = self._split_provider(provider)
         result = self.client.invoke_llm(
             tenant_id=self.tenant_id,
@@ -489,7 +493,9 @@ class PluginModelRuntime(ModelRuntime):
         credentials: dict[str, Any],
         texts: list[str],
         input_type: EmbeddingInputType,
+        request_metadata: Mapping[str, object] | None = None,
     ) -> EmbeddingResult:
+        del request_metadata
         plugin_id, provider_name = self._split_provider(provider)
         return self.client.invoke_text_embedding(
             tenant_id=self.tenant_id,
@@ -511,7 +517,9 @@ class PluginModelRuntime(ModelRuntime):
         credentials: dict[str, Any],
         documents: list[dict[str, Any]],
         input_type: EmbeddingInputType,
+        request_metadata: Mapping[str, object] | None = None,
     ) -> EmbeddingResult:
+        del request_metadata
         plugin_id, provider_name = self._split_provider(provider)
         return self.client.invoke_multimodal_embedding(
             tenant_id=self.tenant_id,
@@ -555,7 +563,9 @@ class PluginModelRuntime(ModelRuntime):
         docs: list[str],
         score_threshold: float | None,
         top_n: int | None,
+        request_metadata: Mapping[str, object] | None = None,
     ) -> RerankResult:
+        del request_metadata
         plugin_id, provider_name = self._split_provider(provider)
         return self.client.invoke_rerank(
             tenant_id=self.tenant_id,
@@ -581,7 +591,9 @@ class PluginModelRuntime(ModelRuntime):
         docs: list[MultimodalRerankInput],
         score_threshold: float | None,
         top_n: int | None,
+        request_metadata: Mapping[str, object] | None = None,
     ) -> RerankResult:
+        del request_metadata
         plugin_id, provider_name = self._split_provider(provider)
         return self.client.invoke_multimodal_rerank(
             tenant_id=self.tenant_id,
@@ -605,7 +617,9 @@ class PluginModelRuntime(ModelRuntime):
         credentials: dict[str, Any],
         content_text: str,
         voice: str,
+        request_metadata: Mapping[str, object] | None = None,
     ) -> Iterable[bytes]:
+        del request_metadata
         plugin_id, provider_name = self._split_provider(provider)
         return self.client.invoke_tts(
             tenant_id=self.tenant_id,
@@ -646,7 +660,9 @@ class PluginModelRuntime(ModelRuntime):
         model: str,
         credentials: dict[str, Any],
         file: IO[bytes],
+        request_metadata: Mapping[str, object] | None = None,
     ) -> str:
+        del request_metadata
         plugin_id, provider_name = self._split_provider(provider)
         return self.client.invoke_speech_to_text(
             tenant_id=self.tenant_id,
@@ -666,7 +682,9 @@ class PluginModelRuntime(ModelRuntime):
         model: str,
         credentials: dict[str, Any],
         text: str,
+        request_metadata: Mapping[str, object] | None = None,
     ) -> bool:
+        del request_metadata
         plugin_id, provider_name = self._split_provider(provider)
         return self.client.invoke_moderation(
             tenant_id=self.tenant_id,

api/core/repositories/human_input_repository.py

@@ -17,8 +17,8 @@ from core.workflow.human_input_adapter import (
     InteractiveSurfaceDeliveryMethod,
     is_human_input_webapp_enabled,
 )
-from graphon.nodes.human_input.entities import FormDefinition, HumanInputNodeData
-from graphon.nodes.human_input.enums import HumanInputFormKind, HumanInputFormStatus
+from core.workflow.nodes.human_input.entities import FormDefinition, HumanInputNodeData
+from core.workflow.nodes.human_input.enums import HumanInputFormKind, HumanInputFormStatus
 from libs.datetime_utils import naive_utc_now
 from libs.uuid_utils import uuidv7
 from models.account import Account, TenantAccountJoin
@@ -93,6 +93,9 @@ class HumanInputFormEntity(Protocol):
     @property
     def selected_action_id(self) -> str | None: ...
 
+    @property
+    def created_at(self) -> datetime: ...
+
     @property
     def submitted_data(self) -> Mapping[str, Any] | None: ...
 
@@ -178,6 +181,11 @@ class _HumanInputFormEntityImpl(HumanInputFormEntity):
     def selected_action_id(self) -> str | None:
         return self._form_model.selected_action_id
 
+    @property
+    @override
+    def created_at(self) -> datetime:
+        return self._form_model.created_at
+
     @property
     @override
     def submitted_data(self) -> Mapping[str, Any] | None:
@@ -572,6 +580,13 @@ class HumanInputFormSubmissionRepository:
                 return None
             return HumanInputFormRecord.from_models(recipient_model.form, recipient_model)
 
+    def get_by_form_id(self, form_id: str) -> HumanInputFormRecord | None:
+        with session_factory.create_session() as session:
+            form_model = session.get(HumanInputForm, form_id)
+            if form_model is None:
+                return None
+            return HumanInputFormRecord.from_models(form_model, None)
+
     def get_by_form_id_and_recipient_type(
         self,
         form_id: str,

api/core/workflow/human_input_policy.py

@@ -4,9 +4,13 @@ from collections.abc import Mapping, Sequence
 from enum import StrEnum
 from typing import Any, NamedTuple
 
-from graphon.entities.pause_reason import HumanInputRequired, PauseReason, PauseReasonType
-from graphon.nodes.human_input.entities import FormInputConfig, SelectInputConfig
-from graphon.nodes.human_input.enums import ValueSourceType
+from core.workflow.nodes.human_input.entities import FormInputConfig, SelectInputConfig
+from core.workflow.nodes.human_input.enums import ValueSourceType
+from core.workflow.nodes.human_input.pause_reason import (
+    DifyHITLEventType,
+    HumanInputRequired,
+    PauseReason,
+)
 from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool
 from graphon.variables import ArrayStringSegment
 from models.human_input import ApprovalChannel, RecipientType
@@ -97,7 +101,7 @@ def enrich_human_input_pause_reasons(
     enriched: list[dict[str, Any]] = []
     for reason in reasons:
         updated = dict(reason)
-        if updated.get("TYPE") == PauseReasonType.HUMAN_INPUT_REQUIRED:
+        if updated.get("TYPE") == DifyHITLEventType.HUMAN_INPUT_REQUIRED:
             form_id = updated.get("form_id")
             if isinstance(form_id, str):
                 disposition = dispositions_by_form_id.get(form_id)

api/core/workflow/node_factory.py

@@ -43,6 +43,8 @@ from core.workflow.nodes.agent_v2 import DifyAgentNode
 from core.workflow.nodes.agent_v2.binding_resolver import WorkflowAgentBindingResolver
 from core.workflow.nodes.agent_v2.output_adapter import WorkflowAgentOutputAdapter
 from core.workflow.nodes.agent_v2.runtime_request_builder import WorkflowAgentRuntimeRequestBuilder
+from core.workflow.nodes.human_input.callback import DifyHITLCallback
+from core.workflow.nodes.human_input.entities import HumanInputNodeData as DifyHumanInputNodeData
 from core.workflow.system_variables import SystemVariableKey, get_system_text, system_variable_selector
 from core.workflow.template_rendering import CodeExecutorJinja2TemplateRenderer
 from graphon.entities.base_node_data import BaseNodeData
@@ -409,9 +411,9 @@ class DifyNodeFactory(NodeFactory):
                 "file_reference_factory": self._file_reference_factory,
             },
             BuiltinNodeTypes.HUMAN_INPUT: lambda: {
-                "runtime": self._human_input_runtime,
-                "file_reference_factory": self._file_reference_factory,
-                "form_repository": self._human_input_runtime.build_form_repository(),
+                "hitl_callback": self._build_human_input_callback(
+                    node_data=DifyHumanInputNodeData.model_validate(adapted_node_config["data"])
+                ),
             },
             BuiltinNodeTypes.LLM: lambda: self._build_llm_compatible_node_init_kwargs(
                 node_class=node_class,
@@ -515,6 +517,20 @@ class DifyNodeFactory(NodeFactory):
             "message_transformer": self._agent_message_transformer,
         }
 
+    def _build_human_input_callback(
+        self,
+        *,
+        node_data: DifyHumanInputNodeData,
+    ) -> DifyHITLCallback:
+        return DifyHITLCallback(
+            form_repository=self._human_input_runtime.build_form_repository(),
+            node_data=node_data,
+            conversation_id=self._conversation_id(),
+            delivery_methods=self._human_input_runtime._resolve_delivery_methods(node_data=node_data),
+            display_in_ui=self._human_input_runtime._display_in_ui(node_data=node_data),
+            file_reference_factory=self._file_reference_factory,
+        )
+
     def _build_llm_compatible_node_init_kwargs(
         self,
         *,

api/core/workflow/node_runtime.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Callable, Generator, Mapping, Sequence
 from dataclasses import dataclass
+from enum import Enum
 from typing import TYPE_CHECKING, Any, Literal, cast, overload, override
 
 from pydantic import JsonValue
@@ -25,6 +26,7 @@ from core.plugin.impl.plugin import PluginInstaller
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.repositories.human_input_repository import (
     FormCreateParams,
+    HumanInputFormEntity,
     HumanInputFormRepository,
     HumanInputFormRepositoryImpl,
 )
@@ -35,6 +37,12 @@ from core.tools.tool_file_manager import ToolFileManager
 from core.tools.tool_manager import ToolManager
 from core.tools.utils.message_transformer import ToolFileMessageTransformer
 from core.workflow.file_reference import build_file_reference
+from core.workflow.nodes.human_input.entities import (
+    FileInputConfig,
+    FileListInputConfig,
+    FormInputConfig,
+    HumanInputNodeData,
+)
 from extensions.ext_database import db
 from factories import file_factory
 from graphon.file import File, FileTransferMethod, FileType
@@ -50,12 +58,6 @@ from graphon.model_runtime.entities.llm_entities import (
 from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageTool
 from graphon.model_runtime.entities.model_entities import AIModelEntity
 from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
-from graphon.nodes.human_input.entities import (
-    FileInputConfig,
-    FileListInputConfig,
-    FormInputConfig,
-    HumanInputNodeData,
-)
 from graphon.nodes.llm.runtime_protocols import (
     LLMPollingCapableProtocol,
     LLMProtocol,
@@ -63,11 +65,7 @@ from graphon.nodes.llm.runtime_protocols import (
     RetrieverAttachmentLoaderProtocol,
 )
 from graphon.nodes.protocols import FileReferenceFactoryProtocol, HttpClientProtocol, ToolFileManagerProtocol
-from graphon.nodes.runtime import (
-    HumanInputFormStateProtocol,
-    HumanInputNodeRuntimeProtocol,
-    ToolNodeRuntimeProtocol,
-)
+from graphon.nodes.runtime import ToolNodeRuntimeProtocol
 from graphon.nodes.tool.exc import ToolNodeError, ToolRuntimeInvocationError, ToolRuntimeResolutionError
 from graphon.nodes.tool_runtime_entities import (
     ToolRuntimeHandle,
@@ -766,7 +764,7 @@ class DifyToolNodeRuntime(ToolNodeRuntimeProtocol):
                 return ToolRuntimeInvocationError(str(exc))
 
 
-class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):
+class DifyHumanInputNodeRuntime:
     def __init__(
         self,
         run_context: Mapping[str, Any] | DifyRunContext,
@@ -785,7 +783,9 @@ class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):
         invoke_from = self._run_context.invoke_from
         if isinstance(invoke_from, str):
             return invoke_from
-        return str(getattr(invoke_from, "value", invoke_from))
+        if isinstance(invoke_from, Enum):
+            return str(invoke_from.value)
+        return str(invoke_from)
 
     def _resolve_delivery_methods(self, *, node_data: HumanInputNodeData) -> Sequence[DeliveryChannelConfig]:
         invoke_source = self._invoke_source()
@@ -830,8 +830,7 @@ class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):
             form_repository=form_repository,
         )
 
-    @override
-    def get_form(self, *, node_id: str) -> HumanInputFormStateProtocol | None:
+    def get_form(self, *, node_id: str) -> HumanInputFormEntity | None:
         repo = self.build_form_repository()
         return repo.get_form(node_id)
 
@@ -852,7 +851,6 @@ class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):
             )
         return restored_data
 
-    @override
     def create_form(
         self,
         *,
@@ -860,7 +858,7 @@ class DifyHumanInputNodeRuntime(HumanInputNodeRuntimeProtocol):
         node_data: HumanInputNodeData,
         rendered_content: str,
         resolved_default_values: Mapping[str, Any],
-    ) -> HumanInputFormStateProtocol:
+    ) -> HumanInputFormEntity:
         repo = self.build_form_repository()
         params = FormCreateParams(
             workflow_execution_id=self._workflow_execution_id_getter() if self._workflow_execution_id_getter else None,

api/core/workflow/nodes/agent_v2/agent_node.py

@@ -25,8 +25,10 @@ from clients.agent_backend import (
 )
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
 from core.repositories.human_input_repository import HumanInputFormRepository, HumanInputFormRepositoryImpl
+from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
+from core.workflow.nodes.human_input.session_binding import default_session_binding
 from core.workflow.system_variables import SystemVariableKey, get_system_text
-from graphon.entities.pause_reason import HumanInputRequired, SchedulingPause
+from graphon.entities.pause_reason import HitlRequired, SchedulingPause
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.node_events import NodeEventBase, NodeRunResult, PauseRequestedEvent, StreamCompletedEvent
 from graphon.nodes.base.node import Node
@@ -113,6 +115,16 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
     def populate_start_event(self, event) -> None:
         event.extras["agent_node"] = {"version": "2", "agent_node_kind": self.node_data.agent_node_kind}
 
+    @staticmethod
+    def _to_graph_pause_reason(reason: HumanInputRequired | SchedulingPause) -> HitlRequired | SchedulingPause:
+        if isinstance(reason, HumanInputRequired):
+            return HitlRequired(
+                session_id=default_session_binding.issue_session_id_for_form(form_id=reason.form_id),
+                node_id=reason.node_id,
+                node_title=reason.node_title,
+            )
+        return reason
+
     @override
     def _run(self) -> Generator[NodeEventBase, None, None]:
         dify_ctx = DifyRunContext.model_validate(self.require_run_context_value(DIFY_RUN_CONTEXT_KEY))
@@ -193,7 +205,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                     node_id=self._node_id,
                 )
                 if resume_outcome is not None and resume_outcome.repause is not None:
-                    yield PauseRequestedEvent(reason=resume_outcome.repause)
+                    yield PauseRequestedEvent(reason=self._to_graph_pause_reason(resume_outcome.repause))
                     return
                 if (
                     resume_outcome is not None
@@ -295,7 +307,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                 # form is built *before* the snapshot is saved so its id can be
                 # persisted as the pause correlation (ENG-637).
                 try:
-                    pause_reason: HumanInputRequired | SchedulingPause | None = build_ask_human_pause_reason(
+                    pause_request = build_ask_human_pause_reason(
                         deferred_tool_call=terminal_event.deferred_tool_call,
                         node_id=self._node_id,
                         default_node_title=bundle.agent.name or self._node_id,
@@ -321,9 +333,11 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                 # deferred_tool_results from the submitted form.
                 pending_form_id: str | None = None
                 pending_tool_call_id: str | None = None
-                if isinstance(pause_reason, HumanInputRequired):
-                    pending_form_id = pause_reason.form_id
+                pause_reason: HumanInputRequired | SchedulingPause
+                if pause_request is not None:
+                    pending_form_id = pause_request.form_id
                     pending_tool_call_id = terminal_event.deferred_tool_call.tool_call_id
+                    pause_reason = pause_request
                 else:
                     pause_reason = SchedulingPause(
                         message=terminal_event.message
@@ -338,7 +352,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                     pending_form_id=pending_form_id,
                     pending_tool_call_id=pending_tool_call_id,
                 )
-                yield PauseRequestedEvent(reason=pause_reason)
+                yield PauseRequestedEvent(reason=self._to_graph_pause_reason(pause_reason))
                 return
 
             # Non-success terminal (failed / cancelled) skips per-output

api/core/workflow/nodes/agent_v2/ask_human_hitl.py

@@ -2,11 +2,12 @@
 
 ENG-636. When an Agent backend run ends with a deferred ``dify.ask_human`` tool
 call, the workflow Agent node pauses the *outer* workflow through the very same
-``HumanInputRequired`` form mechanism the Human Input node uses — reusing the
-form repository, delivery channels, and submission endpoints unchanged. This
-module is a pure translation layer: model-facing ask_human tool args become
-graphon form entities (``HumanInputNodeData`` / ``FormInputConfig`` /
-``UserActionConfig``) plus Dify delivery configs. It adds no new HITL behavior.
+HITL form path the Human Input node uses — reusing the form repository,
+delivery channels, and submission endpoints unchanged. This module is a pure
+translation layer: model-facing ask_human tool args become graphon form
+entities (``HumanInputNodeData`` / ``FormInputConfig`` / ``UserActionConfig``)
+plus Dify delivery configs and the enriched Dify-side pause reason payload.
+The Graphon ``HitlRequired`` conversion stays at the node boundary.
 
 The agent-side ``dify.ask_human`` contract is richer than the workflow form
 schema in two places, handled here without widening the form vocabulary:
@@ -48,8 +49,7 @@ from core.workflow.human_input_adapter import (
     ExternalRecipient,
     InteractiveSurfaceDeliveryMethod,
 )
-from graphon.entities.pause_reason import HumanInputRequired
-from graphon.nodes.human_input.entities import (
+from core.workflow.nodes.human_input.entities import (
     FileInputConfig,
     FileListInputConfig,
     FormInputConfig,
@@ -60,7 +60,8 @@ from graphon.nodes.human_input.entities import (
     StringSource,
     UserActionConfig,
 )
-from graphon.nodes.human_input.enums import ButtonStyle, TimeoutUnit, ValueSourceType
+from core.workflow.nodes.human_input.enums import ButtonStyle, TimeoutUnit, ValueSourceType
+from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
 from models.agent_config_entities import AgentHumanContactConfig
 
 # Default ask_human tool name (see ``DifyAskHumanLayerConfig.tool_name``). The

api/core/workflow/nodes/agent_v2/ask_human_resume.py

@@ -7,9 +7,10 @@ agent-side ``dify.ask_human`` contract so the node can start a *second* Agent ru
 that carries the human's answer:
 
 * submitted  -> AskHumanToolResult(status="submitted", action, values)
-* timeout / expired -> AskHumanToolResult(status="timeout")
+* timeout -> AskHumanToolResult(status="timeout")
+* expired (global timeout) -> invalid resume state; the workflow must not resume
 * still waiting (defensive: the host resumed us early) -> re-emit the same
-  HumanInputRequired pause rebuilt from the stored form definition.
+  Dify-side ``HumanInputRequired`` pause rebuilt from the stored form definition.
 
 It only *reads* existing HITL form state — it never mutates the form or the HITL
 submission flow. The DB read (``resolve_ask_human_form``) is kept thin so the
@@ -31,18 +32,17 @@ from pydantic import JsonValue
 from sqlalchemy import select
 
 from core.db.session_factory import session_factory
-from graphon.entities.pause_reason import HumanInputRequired
-from graphon.nodes.human_input.entities import FormDefinition
-from graphon.nodes.human_input.enums import HumanInputFormStatus
+from core.workflow.nodes.human_input.entities import FormDefinition
+from core.workflow.nodes.human_input.enums import HumanInputFormStatus
+from core.workflow.nodes.human_input.pause_reason import HumanInputRequired
 from models.human_input import HumanInputForm
 
-# A WAITING form has not been answered yet; the other terminal states map onto
-# the agent-facing result status. EXPIRED (global timeout) and TIMEOUT
-# (node-level) both surface to the model as "timeout" so it can react.
+# A WAITING form has not been answered yet. TIMEOUT is resumable through the
+# agent-facing "timeout" result, but EXPIRED is a global timeout and therefore
+# an invalid resume state, matching the ordinary Human Input callback boundary.
 _FORM_STATUS_TO_RESULT: dict[HumanInputFormStatus, AskHumanResultStatus] = {
     HumanInputFormStatus.SUBMITTED: "submitted",
     HumanInputFormStatus.TIMEOUT: "timeout",
-    HumanInputFormStatus.EXPIRED: "timeout",
 }
 
 
@@ -99,6 +99,9 @@ def map_form_to_outcome(
     definition = FormDefinition.model_validate_json(form_definition)
     if status == HumanInputFormStatus.WAITING:
         return AskHumanResumeOutcome(repause=_rebuild_pause(definition=definition, form_id=form_id, node_id=node_id))
+    if status == HumanInputFormStatus.EXPIRED:
+        msg = f"cannot resume globally expired ask_human form, form_id={form_id}"
+        raise AssertionError(msg)
 
     result_status = _FORM_STATUS_TO_RESULT.get(status, "unavailable")
     if result_status != "submitted":

api/core/workflow/nodes/human_input/__init__.py

@@ -0,0 +1,50 @@
+from .entities import (
+    FileInputConfig,
+    FileListInputConfig,
+    FormDefinition,
+    FormInputConfig,
+    HumanInputNodeData,
+    HumanInputSubmissionValidationError,
+    ParagraphInputConfig,
+    SelectInputConfig,
+    StringListSource,
+    StringSource,
+    UserActionConfig,
+    validate_human_input_submission,
+)
+from .enums import (
+    ButtonStyle,
+    FormInputType,
+    HumanInputFormKind,
+    HumanInputFormStatus,
+    TimeoutUnit,
+    ValueSourceType,
+)
+from .pause_reason import DifyHITLEventType, HumanInputRequired, PauseReason
+from .session_binding import SessionBinding, default_session_binding
+
+__all__ = [
+    "ButtonStyle",
+    "DifyHITLEventType",
+    "FileInputConfig",
+    "FileListInputConfig",
+    "FormDefinition",
+    "FormInputConfig",
+    "FormInputType",
+    "HumanInputFormKind",
+    "HumanInputFormStatus",
+    "HumanInputNodeData",
+    "HumanInputRequired",
+    "HumanInputSubmissionValidationError",
+    "ParagraphInputConfig",
+    "PauseReason",
+    "SelectInputConfig",
+    "SessionBinding",
+    "StringListSource",
+    "StringSource",
+    "TimeoutUnit",
+    "UserActionConfig",
+    "ValueSourceType",
+    "default_session_binding",
+    "validate_human_input_submission",
+]

api/core/workflow/nodes/human_input/_exc.py

@@ -0,0 +1,22 @@
+from graphon.file.enums import FileTransferMethod
+
+
+class InvalidConfigError(Exception):
+    pass
+
+
+class InvalidSubmittedDataError(Exception):
+    pass
+
+
+class InvalidTransferMethodError(InvalidConfigError):
+    transfer_method: FileTransferMethod
+
+    def __init__(self, transfer_method: FileTransferMethod) -> None:
+        self.transfer_method = transfer_method
+        super().__init__(f"invalid file transfer method: {transfer_method}")
+
+
+class ExtensionsNotSetErrorValueError(InvalidConfigError):
+    def __init__(self) -> None:
+        super().__init__("allowed_file_extensions not set")

api/core/workflow/nodes/human_input/boundary.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from core.repositories.human_input_repository import HumanInputFormSubmissionRepository
+from core.workflow.human_input_policy import resolve_variable_select_input_options
+from graphon.entities.pause_reason import HitlRequired, SchedulingPause
+from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool
+
+from .pause_reason import HumanInputRequired, PauseReason
+from .session_binding import default_session_binding
+
+
+class HumanInputPauseReasonResolutionError(LookupError):
+    """Raised when a graph pause reason cannot be resolved into Dify-owned form state."""
+
+
+def enrich_graph_pause_reasons(
+    *,
+    reasons: Sequence[HitlRequired | PauseReason],
+    form_repository: HumanInputFormSubmissionRepository,
+    variable_pool: ReadOnlyVariablePool | None,
+) -> list[PauseReason]:
+    enriched: list[PauseReason] = []
+    for reason in reasons:
+        if isinstance(reason, HitlRequired):
+            enriched_reason = _enrich_hitl_required(
+                reason=reason,
+                form_repository=form_repository,
+                variable_pool=variable_pool,
+            )
+            if enriched_reason is not None:
+                enriched.append(enriched_reason)
+            continue
+        if isinstance(reason, HumanInputRequired | SchedulingPause):
+            enriched.append(reason)
+    return enriched
+
+
+def _enrich_hitl_required(
+    *,
+    reason: HitlRequired,
+    form_repository: HumanInputFormSubmissionRepository,
+    variable_pool: ReadOnlyVariablePool | None,
+) -> HumanInputRequired:
+    form_id = default_session_binding.resolve_form_id_from_session_id(session_id=reason.session_id)
+    record = form_repository.get_by_form_id(form_id)
+    if record is None:
+        raise HumanInputPauseReasonResolutionError(
+            f"missing human input form while enriching pause reason: form_id={form_id}, session_id={reason.session_id}"
+        )
+
+    return HumanInputRequired(
+        form_id=record.form_id,
+        form_content=record.rendered_content,
+        inputs=resolve_variable_select_input_options(record.definition.inputs, variable_pool=variable_pool),
+        actions=list(record.definition.user_actions),
+        node_id=reason.node_id,
+        node_title=reason.node_title or record.definition.node_title or record.node_id,
+        resolved_default_values=dict(record.definition.default_values),
+    )

api/core/workflow/nodes/human_input/callback.py

@@ -0,0 +1,329 @@
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import Mapping, Sequence
+from datetime import datetime, timedelta
+from typing import Any
+
+from configs import dify_config
+from core.repositories.human_input_repository import FormCreateParams, HumanInputFormEntity, HumanInputFormRepository
+from core.workflow.human_input_adapter import DeliveryChannelConfig
+from core.workflow.node_runtime import DifyFileReferenceFactory
+from graphon.nodes.human_input.entities import Completed, Expired, HITLContext, HITLDecision, PauseRequested
+from graphon.runtime import VariablePool
+from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool
+from graphon.variables.factory import build_segment
+from graphon.variables.segments import Segment
+from libs.datetime_utils import ensure_naive_utc, naive_utc_now
+
+from .entities import (
+    FileInputConfig,
+    FileListInputConfig,
+    FormInputConfig,
+    HumanInputNodeData,
+    ParagraphInputConfig,
+    SelectInputConfig,
+)
+from .enums import HumanInputFormStatus
+from .session_binding import default_session_binding
+
+logger = logging.getLogger(__name__)
+
+
+def _require_template_variable_pool(pool: ReadOnlyVariablePool) -> VariablePool:
+    """Return the concrete graphon pool required for template expansion."""
+    if isinstance(pool, VariablePool):
+        return pool
+
+    msg = "human input rendering requires graphon.runtime.VariablePool for template expansion"
+    raise TypeError(msg)
+
+
+def render_form_content_before_submission(
+    node_data: HumanInputNodeData,
+    *,
+    variable_pool: ReadOnlyVariablePool,
+) -> str:
+    """Process form content by substituting runtime variables before pause."""
+    # NOTE(QuantumGhost): This is not ideal, we should expose
+    # VariablePool method in Graphon.
+    rendered_form_content = _require_template_variable_pool(variable_pool).convert_template(node_data.form_content)
+    return rendered_form_content.markdown
+
+
+def resolve_default_values(
+    node_data: HumanInputNodeData,
+    *,
+    variable_pool: ReadOnlyVariablePool,
+) -> Mapping[str, Any]:
+    resolved_defaults: dict[str, Any] = {}
+    for form_input in node_data.inputs:
+        resolved_default = form_input.resolve_default_value(variable_pool)
+        if resolved_default is None:
+            continue
+        resolved_defaults[form_input.output_variable_name] = resolved_default.to_object()
+    return resolved_defaults
+
+
+class DifyHITLCallback:
+    """Bridge Dify human-input semantics onto graphon HITL decisions.
+
+    The boundary preserves Dify's split between node timeout and global
+    expiration: only explicit node timeout resumes along the timeout handle,
+    while global expiration is treated as an invalid resume state.
+    """
+
+    pause_requested_type = PauseRequested
+
+    _OUTPUT_FIELD_ACTION_ID = "__action_id"
+    _OUTPUT_FIELD_ACTION_VALUE = "__action_value"
+    _OUTPUT_FIELD_RENDERED_CONTENT = "__rendered_content"
+    _TIMEOUT_HANDLE = "__timeout__"
+
+    def __init__(
+        self,
+        *,
+        form_repository: HumanInputFormRepository,
+        node_data: HumanInputNodeData,
+        workflow_execution_id: str | None = None,
+        conversation_id: str | None = None,
+        delivery_methods: Sequence[DeliveryChannelConfig] = (),
+        display_in_ui: bool = False,
+        file_reference_factory: DifyFileReferenceFactory | None = None,
+    ) -> None:
+        self._form_repository = form_repository
+        self._session_binding = default_session_binding
+        self._node_data = node_data
+        self._workflow_execution_id = workflow_execution_id
+        self._conversation_id = conversation_id
+        self._delivery_methods = tuple(delivery_methods)
+        self._display_in_ui = display_in_ui
+        self._file_reference_factory = file_reference_factory
+
+    def __call__(self, ctx: HITLContext) -> HITLDecision:
+        form = self._form_repository.get_form(ctx.node_id)
+        if form is None:
+            created = self._create_form(ctx)
+            return PauseRequested(session_id=self._session_binding.issue_session_id_for_form(form_id=created.id))
+
+        status = self._normalize_status(form.status)
+        if status == HumanInputFormStatus.TIMEOUT.value:
+            return Expired(
+                selected_handle=self._TIMEOUT_HANDLE,
+                outputs=self._build_special_outputs(
+                    action_id="",
+                    action_value="",
+                    rendered_content=form.rendered_content,
+                ),
+            )
+
+        if status == HumanInputFormStatus.EXPIRED.value:
+            msg = f"cannot resume globally expired human input form, form_id={form.id}"
+            raise AssertionError(msg)
+
+        if not form.submitted:
+            if status == HumanInputFormStatus.WAITING.value and self._is_past_global_deadline(form.created_at):
+                msg = f"cannot resume waiting human input form after global timeout, form_id={form.id}"
+                raise AssertionError(msg)
+            if self._is_past_node_deadline(form.expiration_time):
+                return Expired(
+                    selected_handle=self._TIMEOUT_HANDLE,
+                    outputs=self._build_special_outputs(
+                        action_id="",
+                        action_value="",
+                        rendered_content=form.rendered_content,
+                    ),
+                )
+            return PauseRequested(session_id=self._session_binding.issue_session_id_for_form(form_id=form.id))
+
+        selected_action_id = form.selected_action_id
+        if selected_action_id is None:
+            msg = f"selected_action_id should not be None when form submitted, form_id={form.id}"
+            raise AssertionError(msg)
+
+        submitted_data = self._restore_submitted_data(submitted_data=form.submitted_data or {})
+        rendered_content = self.render_form_content_with_outputs(
+            form.rendered_content,
+            submitted_data,
+            self._node_data.outputs_field_names(),
+            self._node_data.inputs,
+        )
+        outputs = dict(submitted_data)
+        outputs.update(
+            self._build_special_outputs(
+                action_id=selected_action_id,
+                action_value=self._node_data.must_resolve_action_value(selected_action_id),
+                rendered_content=rendered_content,
+            )
+        )
+        return Completed(
+            selected_handle=selected_action_id,
+            inputs=submitted_data,
+            outputs=outputs,
+        )
+
+    def _create_form(self, ctx: HITLContext) -> HumanInputFormEntity:
+        params = FormCreateParams(
+            workflow_execution_id=self._workflow_execution_id or ctx.workflow_execution_id,
+            conversation_id=self._conversation_id,
+            node_id=ctx.node_id,
+            form_config=self._node_data,
+            rendered_content=render_form_content_before_submission(
+                self._node_data,
+                variable_pool=ctx.variable_pool,
+            ),
+            delivery_methods=self._delivery_methods,
+            display_in_ui=self._display_in_ui,
+            resolved_default_values=dict(
+                resolve_default_values(
+                    self._node_data,
+                    variable_pool=ctx.variable_pool,
+                )
+            ),
+        )
+        return self._form_repository.create_form(params)
+
+    @staticmethod
+    def _normalize_status(status: HumanInputFormStatus | str) -> str:
+        if isinstance(status, HumanInputFormStatus):
+            return status.value
+        return status
+
+    @staticmethod
+    def _is_past_node_deadline(expiration_time: datetime) -> bool:
+        expiration_time = ensure_naive_utc(expiration_time)
+        return expiration_time <= naive_utc_now()
+
+    @staticmethod
+    def _is_past_global_deadline(created_at: datetime) -> bool:
+        global_timeout_seconds = dify_config.HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS
+        if global_timeout_seconds <= 0:
+            return False
+        global_deadline = ensure_naive_utc(created_at) + timedelta(seconds=global_timeout_seconds)
+        return global_deadline <= naive_utc_now()
+
+    @staticmethod
+    def render_form_content_with_outputs(
+        form_content: str,
+        outputs: Mapping[str, Segment],
+        field_names: Sequence[str],
+        form_inputs: Sequence[FormInputConfig] | None = None,
+    ) -> str:
+        """Replace {{#$output.xxx#}} placeholders with submitted values.
+
+        Text inputs render their submitted value directly. File inputs render as
+        stable placeholders so the final content stays readable and does not
+        inline transport metadata.
+
+        Returns:
+            the interplated form content
+        """
+        inputs_by_name: dict[str, FormInputConfig] = {}
+        if form_inputs is not None:
+            inputs_by_name = {form_input.output_variable_name: form_input for form_input in form_inputs}
+
+        rendered_content = form_content
+        for field_name in field_names:
+            placeholder = "{{#$output." + field_name + "#}}"
+            replacement = DifyHITLCallback._render_output_placeholder_value(
+                value=outputs.get(field_name),
+                form_input=inputs_by_name.get(field_name),
+            )
+            rendered_content = rendered_content.replace(placeholder, replacement)
+        return rendered_content
+
+    @staticmethod
+    def _render_output_placeholder_value(
+        *,
+        value: Any,
+        form_input: FormInputConfig | None,
+    ) -> str:
+        if isinstance(value, Segment):
+            value = value.to_object()
+
+        if value is None:
+            return ""
+
+        if isinstance(form_input, FileInputConfig):
+            return "[file]"
+
+        if isinstance(form_input, FileListInputConfig):
+            file_count = 0
+            if isinstance(value, Sequence) and not isinstance(value, str | bytes):
+                file_count = len(value)
+            return f"[{file_count} files]"
+
+        if isinstance(form_input, ParagraphInputConfig | SelectInputConfig):
+            return str(value)
+
+        if isinstance(value, dict | list):
+            return json.dumps(value, ensure_ascii=False)
+
+        return str(value)
+
+    def _restore_submitted_data(
+        self,
+        *,
+        submitted_data: Mapping[str, Any],
+    ) -> dict[str, Segment]:
+        """Reconstruct graphon runtime values from validated submission payloads."""
+        restored_data: dict[str, Segment] = {}
+        inputs_by_name = {form_input.output_variable_name: form_input for form_input in self._node_data.inputs}
+
+        for name, value in submitted_data.items():
+            form_input = inputs_by_name.get(name)
+            if form_input is None:
+                logger.error("unexpected form data in submitted data, key=%s", name)
+                continue
+            restored_data[name] = build_segment(self._restore_submitted_value(input_config=form_input, value=value))
+
+        return restored_data
+
+    def _restore_submitted_value(
+        self,
+        *,
+        input_config: FormInputConfig,
+        value: Any,
+    ) -> Any:
+        if isinstance(input_config, FileInputConfig):
+            if not isinstance(value, Mapping):
+                raise ValueError(
+                    "HumanInput file submission must be persisted as a mapping, "
+                    f"output_variable_name={input_config.output_variable_name}"
+                )
+            return self._build_file_reference(mapping=value)
+
+        if isinstance(input_config, FileListInputConfig):
+            if not isinstance(value, list):
+                raise ValueError(
+                    "HumanInput file-list submission must be persisted as a list, "
+                    f"output_variable_name={input_config.output_variable_name}"
+                )
+            if any(not isinstance(item, Mapping) for item in value):
+                raise ValueError(
+                    "HumanInput file-list submission must contain mappings, "
+                    f"output_variable_name={input_config.output_variable_name}"
+                )
+            return [self._build_file_reference(mapping=item) for item in value]
+
+        return value
+
+    def _build_file_reference(self, *, mapping: Mapping[str, Any]) -> Any:
+        if self._file_reference_factory is None:
+            raise ValueError("file_reference_factory is required to restore file submissions")
+        return self._file_reference_factory.build_from_mapping(mapping=mapping)
+
+    @classmethod
+    def _build_special_outputs(
+        cls,
+        *,
+        action_id: str,
+        action_value: str,
+        rendered_content: str,
+    ) -> dict[str, Segment]:
+        return {
+            cls._OUTPUT_FIELD_ACTION_ID: build_segment(action_id),
+            cls._OUTPUT_FIELD_ACTION_VALUE: build_segment(action_value),
+            cls._OUTPUT_FIELD_RENDERED_CONTENT: build_segment(rendered_content),
+        }

api/core/workflow/nodes/human_input/entities.py

@@ -0,0 +1,374 @@
+"""Dify-owned Human Input entities.
+
+Graphon v0.6.0 keeps only the minimal HITL callback contract. Dify owns the
+workflow-facing form schema, validation rules, and compatibility payload shapes
+that used to live in graphon.
+"""
+
+import abc
+import re
+from collections.abc import Mapping, Sequence
+from datetime import datetime, timedelta
+from typing import Annotated, Any, Literal, Self, assert_never, override
+
+from pydantic import BaseModel, Field, NonNegativeInt, field_validator, model_validator
+
+from graphon.entities.base_node_data import BaseNodeData
+from graphon.enums import BuiltinNodeTypes, NodeType
+from graphon.file.enums import FileTransferMethod, FileType
+from graphon.nodes.base.variable_template_parser import VariableTemplateParser
+from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool
+from graphon.variables.consts import SELECTORS_LENGTH
+from graphon.variables.segments import Segment
+
+from . import _exc as exc
+from .enums import ButtonStyle, FormInputType, TimeoutUnit, ValueSourceType
+
+_OUTPUT_VARIABLE_PATTERN = re.compile(
+    r"\{\{#\$output\.(?P<field_name>[a-zA-Z_][a-zA-Z0-9_]{0,29})#\}\}",
+)
+
+
+class StringSource(BaseModel):
+    """Default configuration for form inputs."""
+
+    # NOTE: Ideally, a discriminated union would be used to model
+    # FormInputDefault. However, the UI requires preserving the previous
+    # value when switching between `VARIABLE` and `CONSTANT` types. This
+    # necessitates retaining all fields, making a discriminated union unsuitable.
+
+    # NOTE: This class is renamed from FormInputDefault.
+
+    type: ValueSourceType
+
+    # The selector of default variable, used when `type` is `VARIABLE`.
+    selector: Sequence[str] = Field(default_factory=tuple)
+
+    # Constant defaults are stored as strings because current form inputs are
+    # text-based (`TEXT_INPUT` and `PARAGRAPH`).
+    value: str = ""
+
+    @model_validator(mode="after")
+    def _validate_selector(self) -> Self:
+        if self.type == ValueSourceType.CONSTANT:
+            return self
+        if len(self.selector) < SELECTORS_LENGTH:
+            msg = f"the length of selector should be at least {SELECTORS_LENGTH}, selector={self.selector}"
+            raise ValueError(msg)
+        return self
+
+
+class StringListSource(BaseModel):
+    type: ValueSourceType
+
+    # The selector of default variable, used when `type` is `VARIABLE`.
+    selector: Sequence[str] = Field(default_factory=tuple)
+
+    # The value of the default, used when `type` is `CONSTANT`.
+    value: list[str] = Field(default_factory=list)
+
+
+class BaseInputConfig(BaseModel):
+    """BaseInputConfig is the base class for all input field definitions.
+    One input corresponds to one output variable during form submission.
+    """
+
+    output_variable_name: str
+
+    @abc.abstractmethod
+    def extract_variable_selectors(self) -> Sequence[Sequence[str]]:
+        """`extract_variable_selectors` extracts variable selectors
+        used by this input field.
+        """
+
+    @abc.abstractmethod
+    def resolve_default_value(self, pool: ReadOnlyVariablePool) -> Segment | None:
+        """`resolve_default_value` resolves the default value for form submission.
+
+        If the form input does not specify a default value, or the default value does
+        not depend on the runtime variable, this method should return `None`.
+        """
+
+
+class ParagraphInputConfig(BaseInputConfig):
+    """Form input definition."""
+
+    # NOTE: This class is renamed from FormInput.
+    type: Literal[FormInputType.PARAGRAPH] = FormInputType.PARAGRAPH
+    default: StringSource | None = None
+
+    @override
+    def extract_variable_selectors(self) -> Sequence[Sequence[str]]:
+        default = self.default
+        if default is None:
+            return []
+        if default.type == ValueSourceType.CONSTANT:
+            return []
+        return [default.selector]
+
+    @override
+    def resolve_default_value(self, pool: ReadOnlyVariablePool) -> Segment | None:
+        default = self.default
+        if default is None:
+            return None
+
+        if default.type == ValueSourceType.CONSTANT:
+            return None
+
+        return pool.get(default.selector)
+
+
+class SelectInputConfig(BaseInputConfig):
+    type: Literal[FormInputType.SELECT] = FormInputType.SELECT
+    option_source: StringListSource
+
+    @override
+    def extract_variable_selectors(self) -> Sequence[Sequence[NodeType]]:
+        if self.option_source.type == ValueSourceType.CONSTANT:
+            return []
+        return [self.option_source.selector]
+
+    @override
+    def resolve_default_value(self, pool: ReadOnlyVariablePool) -> Segment | None:
+        _ = pool
+        return None
+
+
+_ALLOWED_TRANSFER_METHOD = frozenset(
+    [
+        FileTransferMethod.LOCAL_FILE,
+        FileTransferMethod.REMOTE_URL,
+    ]
+)
+
+
+class _FileInputCommonConfig(BaseModel):
+    allowed_file_types: Sequence[FileType] = Field(default_factory=list[FileType])
+    allowed_file_extensions: Sequence[str] = Field(default_factory=list)
+    allowed_file_upload_methods: Sequence[FileTransferMethod] = Field(default_factory=list[FileTransferMethod])
+
+    @field_validator("allowed_file_upload_methods", mode="after")
+    @classmethod
+    def _validate_upload_methods(cls, transfer_methods: Sequence[FileTransferMethod]) -> Sequence[FileTransferMethod]:
+        validated_values: list[FileTransferMethod] = []
+        for value in transfer_methods:
+            if value not in _ALLOWED_TRANSFER_METHOD:
+                raise exc.InvalidTransferMethodError(value)
+            validated_values.append(value)
+
+        return validated_values
+
+    @model_validator(mode="after")
+    def _validate_extensions(self) -> Self:
+        if FileType.CUSTOM not in self.allowed_file_types:
+            return self
+        if not self.allowed_file_extensions:
+            raise exc.ExtensionsNotSetErrorValueError
+        return self
+
+
+class FileInputConfig(_FileInputCommonConfig, BaseInputConfig):
+    type: Literal[FormInputType.FILE] = FormInputType.FILE
+
+    @override
+    def extract_variable_selectors(self) -> Sequence[Sequence[NodeType]]:
+        return []
+
+    @override
+    def resolve_default_value(self, pool: ReadOnlyVariablePool) -> Segment | None:
+        _ = pool
+        return None
+
+
+class FileListInputConfig(_FileInputCommonConfig, BaseInputConfig):
+    type: Literal[FormInputType.FILE_LIST] = FormInputType.FILE_LIST
+    number_limits: NonNegativeInt = 0
+
+    @override
+    def extract_variable_selectors(self) -> Sequence[Sequence[NodeType]]:
+        return []
+
+    @override
+    def resolve_default_value(self, pool: ReadOnlyVariablePool) -> Segment | None:
+        _ = pool
+        return None
+
+
+type FormInputConfig = Annotated[
+    ParagraphInputConfig | SelectInputConfig | FileInputConfig | FileListInputConfig,
+    Field(discriminator="type"),
+]
+
+
+_IDENTIFIER_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+
+class UserActionConfig(BaseModel):
+    """User action configuration."""
+
+    # id is the identifier for this action.
+    # It also serves as the identifiers of output handle.
+    #
+    # The id must be a valid identifier (satisfy the _IDENTIFIER_PATTERN above.)
+    id: str = Field(max_length=20)
+    title: str = Field(max_length=100)
+    button_style: ButtonStyle = ButtonStyle.DEFAULT
+
+    @field_validator("id")
+    @classmethod
+    def _validate_id(cls, value: str) -> str:
+        if not _IDENTIFIER_PATTERN.match(value):
+            msg = (
+                f"'{value}' is not a valid identifier. It must start with "
+                f"a letter or underscore, and contain only letters, "
+                f"numbers, or underscores."
+            )
+            raise ValueError(msg)
+        return value
+
+
+class HumanInputNodeData(BaseNodeData):
+    """Human Input node data."""
+
+    type: NodeType = BuiltinNodeTypes.HUMAN_INPUT
+    form_content: str = ""
+    inputs: list[FormInputConfig] = Field(default_factory=list[FormInputConfig])
+    user_actions: list[UserActionConfig] = Field(default_factory=list[UserActionConfig])
+    timeout: int = 36
+    timeout_unit: TimeoutUnit = TimeoutUnit.HOUR
+
+    @field_validator("inputs")
+    @classmethod
+    def _validate_inputs(cls, inputs: list[FormInputConfig]) -> list[FormInputConfig]:
+        seen_names: set[str] = set()
+        for form_input in inputs:
+            name = form_input.output_variable_name
+            if name in seen_names:
+                msg = f"duplicated output_variable_name '{name}' in inputs"
+                raise ValueError(msg)
+            seen_names.add(name)
+        return inputs
+
+    @field_validator("user_actions")
+    @classmethod
+    def _validate_user_actions(cls, user_actions: list[UserActionConfig]) -> list[UserActionConfig]:
+        seen_ids: set[str] = set()
+        for action in user_actions:
+            action_id = action.id
+            if action_id in seen_ids:
+                msg = f"duplicated user action id '{action_id}'"
+                raise ValueError(msg)
+            seen_ids.add(action_id)
+        return user_actions
+
+    def expiration_time(self, start_time: datetime) -> datetime:
+        match self.timeout_unit:
+            case TimeoutUnit.HOUR:
+                return start_time + timedelta(hours=self.timeout)
+            case TimeoutUnit.DAY:
+                return start_time + timedelta(days=self.timeout)
+            case _:
+                assert_never(self.timeout_unit)
+
+    def outputs_field_names(self) -> Sequence[str]:
+        return [match.group("field_name") for match in _OUTPUT_VARIABLE_PATTERN.finditer(self.form_content)]
+
+    def extract_variable_selector_to_variable_mapping(
+        self,
+        node_id: str,
+    ) -> Mapping[str, Sequence[str]]:
+        variable_mappings: dict[str, Sequence[str]] = {}
+
+        def _add_variable_selectors(selectors: Sequence[Sequence[str]]) -> None:
+            for selector in selectors:
+                if len(selector) < SELECTORS_LENGTH:
+                    continue
+                qualified_variable_mapping_key = f"{node_id}.#{'.'.join(selector[:SELECTORS_LENGTH])}#"
+                variable_mappings[qualified_variable_mapping_key] = list(
+                    selector[:SELECTORS_LENGTH],
+                )
+
+        form_template_parser = VariableTemplateParser(template=self.form_content)
+        _add_variable_selectors(
+            [selector.value_selector for selector in form_template_parser.extract_variable_selectors()]
+        )
+
+        for form_input in self.inputs:
+            selectors = form_input.extract_variable_selectors()
+            for selector in selectors:
+                value_key = ".".join(selector)
+                qualified_variable_mapping_key = f"{node_id}.#{value_key}#"
+                variable_mappings[qualified_variable_mapping_key] = selector
+
+        return variable_mappings
+
+    def find_action_text(self, action_id: str) -> str:
+        """Resolve action display text by id."""
+        for action in self.user_actions:
+            if action.id == action_id:
+                return action.title
+        return action_id
+
+    def must_resolve_action_value(self, action_id: str) -> str:
+        """Resolve the selected action's workflow-facing value by id.
+
+        This method should only be called with action ids that have already been
+        validated against the node configuration.
+
+        Returns:
+            The configured workflow-facing value for the selected action id.
+
+        Raises:
+            AssertionError: If the action id is not present in the node config.
+        """
+        for action in self.user_actions:
+            if action.id == action_id:
+                return action.title
+        msg = f"Invalid action: {action_id}"
+        raise AssertionError(msg)
+
+
+class FormDefinition(BaseModel):
+    form_content: str
+    inputs: list[FormInputConfig] = Field(default_factory=list[FormInputConfig])
+    user_actions: list[UserActionConfig] = Field(default_factory=list[UserActionConfig])
+    rendered_content: str
+    expiration_time: datetime
+
+    # this is used to store the resolved default values
+    default_values: dict[str, Any] = Field(default_factory=dict)
+
+    # node_title records the title of the HumanInput node.
+    node_title: str | None = None
+
+    # display_in_ui controls whether the form should be displayed in UI surfaces.
+    display_in_ui: bool | None = None
+
+
+class HumanInputSubmissionValidationError(ValueError):
+    pass
+
+
+def validate_human_input_submission(
+    *,
+    inputs: Sequence[FormInputConfig],
+    user_actions: Sequence[UserActionConfig],
+    selected_action_id: str,
+    form_data: Mapping[str, Any],
+) -> None:
+    available_actions = {action.id for action in user_actions}
+    if selected_action_id not in available_actions:
+        msg = f"Invalid action: {selected_action_id}"
+        raise HumanInputSubmissionValidationError(msg)
+
+    provided_inputs = set(form_data.keys())
+    missing_inputs = [
+        form_input.output_variable_name
+        for form_input in inputs
+        if form_input.output_variable_name not in provided_inputs
+    ]
+
+    if missing_inputs:
+        missing_list = ", ".join(missing_inputs)
+        msg = f"Missing required inputs: {missing_list}"
+        raise HumanInputSubmissionValidationError(msg)

api/core/workflow/nodes/human_input/enums.py

@@ -0,0 +1,76 @@
+import enum
+
+
+class HumanInputFormStatus(enum.StrEnum):
+    """Status of a human input form."""
+
+    # Awaiting submission from any recipient. Forms stay in this state until
+    # submitted or a timeout rule applies.
+    WAITING = enum.auto()
+    # Global timeout reached. The workflow run is stopped and will not resume.
+    # This is distinct from node-level timeout.
+    EXPIRED = enum.auto()
+    # Submitted by a recipient; form data is available and execution resumes
+    # along the selected action edge.
+    SUBMITTED = enum.auto()
+    # Node-level timeout reached. The human input node should emit a timeout
+    # event and the workflow should resume along the timeout edge.
+    TIMEOUT = enum.auto()
+
+
+class HumanInputFormKind(enum.StrEnum):
+    """Kind of a human input form."""
+
+    RUNTIME = enum.auto()  # Form created during workflow execution.
+    DELIVERY_TEST = enum.auto()  # Form created for delivery tests.
+
+
+class ButtonStyle(enum.StrEnum):
+    """Button styles for user actions."""
+
+    PRIMARY = enum.auto()
+    DEFAULT = enum.auto()
+    ACCENT = enum.auto()
+    GHOST = enum.auto()
+
+
+class TimeoutUnit(enum.StrEnum):
+    """Timeout unit for form expiration."""
+
+    HOUR = enum.auto()
+    DAY = enum.auto()
+
+
+class FormInputType(enum.StrEnum):
+    """Form input types.
+
+    Name for this enumeration are intentionally keep the same as those for
+    `VariableEntityType`.
+    """
+
+    # Both `TEXT_INPUT` and `PARAGRAPH` represent string input fields.
+    # The corresponding generated variable type is `SegmentType.STRING`.
+    PARAGRAPH = "paragraph"
+
+    # A single-select input field (e.g., a dropdown or radio buttons).
+    # The corresponding generated variable type is `SegmentType.STRING`.
+    SELECT = "select"
+
+    # A file input field that accepts a single file.
+    #  The corresponding generated variable type is `SegmentType.FILE`.
+    FILE = "file"
+
+    # A file input field that accepts zero or more files.
+    # The corresponding generated variable type is `SegmentType.ARRAY_FILE`.
+    FILE_LIST = "file-list"
+
+
+class ValueSourceType(enum.StrEnum):
+    """ValueSourceType records whether the value comes from a static setting
+    in form definiton, or a variable while the workflow is running.
+    """
+
+    # `VARIABLE` means that the value comes from a variable in workflow execution
+    VARIABLE = enum.auto()
+    # `CONSTANT` measn that the value comes from a static setting in form definition.
+    CONSTANT = enum.auto()

api/core/workflow/nodes/human_input/pause_reason.py

@@ -0,0 +1,49 @@
+from collections.abc import Mapping
+from enum import StrEnum
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+from graphon.entities.pause_reason import PauseReasonType, SchedulingPause
+
+from .entities import FormInputConfig, UserActionConfig
+
+
+class DifyHITLEventType(StrEnum):
+    """Dify HITL event type.
+    only used for discriminated union tag.
+
+    """
+
+    # Ideally this should be a string constaint. However, we cannot put
+    # string constant into Literal type cosntructor. We have to warp it as a
+    # string enumeration.
+    HUMAN_INPUT_REQUIRED = PauseReasonType.LEGACY_HUMAN_INPUT_REQUIRED.value
+
+
+class HumanInputRequired(BaseModel):
+    TYPE: Literal[DifyHITLEventType.HUMAN_INPUT_REQUIRED] = DifyHITLEventType.HUMAN_INPUT_REQUIRED
+    form_id: str
+    form_content: str
+    inputs: list[FormInputConfig] = Field(default_factory=list[FormInputConfig])
+    actions: list[UserActionConfig] = Field(default_factory=list[UserActionConfig])
+    node_id: str
+    node_title: str
+
+    # The `resolved_default_values` stores the resolved values of variable
+    # defaults. It's a mapping from `output_variable_name` to their
+    # resolved values.
+    #
+    # For example, the form contains an input with output variable name `name`
+    # and placeholder type `VARIABLE`, its selector is ["start", "name"].
+    # When the HumanInputNode is executed, the corresponding value of
+    # variable `start.name` in the variable pool is `John`.
+    # Thus, the resolved value of the output variable `name` is `John`. The
+    # `resolved_default_values` is `{"name": "John"}`.
+    #
+    # Only form inputs with default value type `VARIABLE` will be resolved
+    # and stored in `resolved_default_values`.
+    resolved_default_values: Mapping[str, Any] = Field(default_factory=dict)
+
+
+type PauseReason = HumanInputRequired | SchedulingPause

api/core/workflow/nodes/human_input/session_binding.py

@@ -0,0 +1,16 @@
+class SessionBinding:
+    """Translate between graphon session ids and Dify form ids.
+
+    Phase 1 keeps the public graphon `session_id` contract while Dify continues
+    to persist and query `form_id`. The identity mapping lives here so later
+    phases can change the translation without scattering equality assumptions.
+    """
+
+    def issue_session_id_for_form(self, *, form_id: str) -> str:
+        return form_id
+
+    def resolve_form_id_from_session_id(self, *, session_id: str) -> str:
+        return session_id
+
+
+default_session_binding = SessionBinding()