test: fix Human Input event test repository after main sync

Restore the concrete pre-seeded form repository, implement timeout state retention, and remove the undefined protocol annotation and spec-based mocks.

api/controllers/console/human_input_form.py

@@ -4,7 +4,7 @@ Console/Studio Human Input Form APIs.
 
 import json
 import logging
-from collections.abc import Generator
+from collections.abc import Generator, Sequence
 from typing import Any
 
 from flask import Response, jsonify, request
@@ -31,6 +31,7 @@ from core.app.apps.common.workflow_response_converter import WorkflowResponseCon
 from core.app.apps.message_generator import MessageGenerator
 from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.workflow.human_input_policy import HumanInputSurface, is_recipient_type_allowed_for_surface
+from core.workflow.nodes.human_input.entities import FormInputConfig
 from extensions.ext_database import db
 from libs.login import login_required
 from models import Account, App
@@ -61,8 +62,10 @@ register_response_schema_models(
 )
 
 
-def _jsonify_form_definition(form: Form) -> Response:
+def _jsonify_form_definition(form: Form, *, inputs: Sequence[FormInputConfig] | None = None) -> Response:
     payload = form.get_definition().model_dump()
+    if inputs is not None:
+        payload["inputs"] = [form_input.model_dump(mode="json") for form_input in inputs]
     payload["expiration_time"] = int(form.expiration_time.timestamp())
     return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
 
@@ -100,7 +103,7 @@ class ConsoleHumanInputFormApi(Resource):
 
         self._ensure_console_access(form, current_tenant_id)
 
-        return _jsonify_form_definition(form)
+        return _jsonify_form_definition(form, inputs=service.resolve_form_inputs(form))
 
     @account_initialization_required
     @login_required

api/core/repositories/human_input_repository.py

@@ -4,7 +4,7 @@ from collections.abc import Mapping, Sequence
 from datetime import datetime
 from typing import Any, Protocol, override
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.orm import Session, selectinload
 
 from core.db.session_factory import session_factory
@@ -115,6 +115,10 @@ class HumanInputFormRepository(Protocol):
 
     def create_form(self, params: FormCreateParams) -> HumanInputFormEntity: ...
 
+    def mark_timeout(self, node_id: str, *, form_id: str) -> HumanInputFormEntity:
+        """Time out an overdue waiting form and return its current state."""
+        ...
+
 
 class _HumanInputFormRecipientEntityImpl(HumanInputFormRecipientEntity):
     def __init__(self, recipient_model: HumanInputFormRecipient):
@@ -557,6 +561,8 @@ class HumanInputFormRepositoryImpl:
         )
         if form_id is not None:
             form_query = form_query.where(HumanInputForm.id == form_id)
+        if self._app_id is not None:
+            form_query = form_query.where(HumanInputForm.app_id == self._app_id)
         with session_factory.create_session() as session:
             form_model: HumanInputForm | None = session.scalars(form_query).first()
             if form_model is None:
@@ -566,6 +572,34 @@ class HumanInputFormRepositoryImpl:
             recipient_models = session.scalars(recipient_query).all()
         return _HumanInputFormEntityImpl(form_model=form_model, recipient_models=recipient_models)
 
+    def mark_timeout(self, node_id: str, *, form_id: str) -> HumanInputFormEntity:
+        if self._workflow_execution_id is None:
+            raise ValueError("workflow_execution_id is required to time out runtime human input forms")
+
+        statement = (
+            update(HumanInputForm)
+            .where(
+                HumanInputForm.id == form_id,
+                HumanInputForm.tenant_id == self._tenant_id,
+                HumanInputForm.workflow_run_id == self._workflow_execution_id,
+                HumanInputForm.node_id == node_id,
+                HumanInputForm.status == HumanInputFormStatus.WAITING,
+                HumanInputForm.submitted_at.is_(None),
+                HumanInputForm.expiration_time <= naive_utc_now(),
+            )
+            .values(status=HumanInputFormStatus.TIMEOUT)
+        )
+        if self._app_id is not None:
+            statement = statement.where(HumanInputForm.app_id == self._app_id)
+        with session_factory.create_session() as session, session.begin():
+            session.execute(statement)
+
+        # A submission or global expiration may have won the race to finish the form.
+        form = self.get_form(node_id, form_id=form_id)
+        if form is None:
+            raise FormNotFoundError(f"form not found, id={form_id}")
+        return form
+
 
 class HumanInputFormSubmissionRepository:
     """Repository for fetching and submitting human input forms."""

api/core/workflow/human_input_policy.py

@@ -121,11 +121,8 @@ def resolve_variable_select_input_options(
 ) -> list[FormInputConfig]:
     """Resolve variable-backed select options to runtime values."""
 
-    # This function replace the SelectInputConfig.option_source.value
-    # field with acutial runtime values when option_source.type is VARIABLE.
-    #
-    # This is a dirty hacks. However it does reduces the logic leaked to callers of
-    # the api.
+    # Convert variable-backed options into a self-contained constant source so
+    # downstream callers do not need the runtime variable pool or selector.
     resolved_inputs: list[FormInputConfig] = []
 
     if variable_pool is None:
@@ -148,8 +145,13 @@ def resolve_variable_select_input_options(
         if not isinstance(option_values, ArrayStringSegment):
             raise TypeError(f"expected ArrayStringSegment, got {type(option_values)}")
 
-        updated_option_source = option_source.model_copy(update={"value": option_values.value})
-        # Ensure frontend receives concrete select options instead of unresolved selectors.
+        updated_option_source = option_source.model_copy(
+            update={
+                "type": ValueSourceType.CONSTANT,
+                "selector": (),
+                "value": option_values.value,
+            }
+        )
         resolved_inputs.append(
             form_input.model_copy(
                 update={"option_source": updated_option_source},
@@ -171,7 +173,6 @@ def resolve_human_input_pause_reason_inputs(
         if not isinstance(reason, HumanInputRequired):
             resolved_reasons.append(reason)
             continue
-
         resolved_inputs = resolve_variable_select_input_options(
             reason.inputs,
             variable_pool=variable_pool,

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

@@ -9,6 +9,7 @@ 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.human_input_policy import resolve_variable_select_input_options
 from core.workflow.node_runtime import DifyFileReferenceFactory
 from graphon.nodes.human_input.entities import Completed, Expired, HITLContext, HITLDecision, PauseRequested
 from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool
@@ -100,6 +101,14 @@ class DifyHITLCallback:
             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.WAITING.value and not form.submitted:
+            if 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):
+                form = self._form_repository.mark_timeout(ctx.node_id, form_id=form.id)
+                status = self._normalize_status(form.status)
+
         if status == HumanInputFormStatus.TIMEOUT.value:
             return Expired(
                 selected_handle=TIMEOUT_HANDLE,
@@ -115,18 +124,6 @@ class DifyHITLCallback:
             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=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
@@ -156,11 +153,19 @@ class DifyHITLCallback:
         )
 
     def _create_form(self, ctx: HITLContext, *, form_id: str | None = None) -> HumanInputFormEntity:
+        form_config = self._node_data.model_copy(
+            update={
+                "inputs": resolve_variable_select_input_options(
+                    self._node_data.inputs,
+                    variable_pool=ctx.variable_pool,
+                )
+            }
+        )
         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,
+            form_config=form_config,
             rendered_content=render_form_content_before_submission(
                 self._node_data,
                 variable_pool=ctx.variable_pool,

api/services/human_input_service.py

@@ -186,11 +186,28 @@ class HumanInputService:
         return form
 
     def resolve_form_inputs(self, form: Form) -> Sequence[FormInputConfig]:
-        variable_pool = self._load_variable_pool_for_form(form)
-        return resolve_variable_select_input_options(
-            form.get_definition().inputs,
-            variable_pool=variable_pool,
-        )
+        definition = form.get_definition()
+        inputs = list(definition.inputs)
+        if any(
+            isinstance(form_input, SelectInputConfig) and form_input.option_source.type == ValueSourceType.VARIABLE
+            for form_input in inputs
+        ):
+            inputs = resolve_variable_select_input_options(
+                inputs,
+                variable_pool=self._load_variable_pool_for_form(form),
+            )
+        return [
+            form_input.model_copy(
+                update={
+                    "option_source": form_input.option_source.model_copy(
+                        update={"type": ValueSourceType.CONSTANT, "selector": ()}
+                    )
+                }
+            )
+            if isinstance(form_input, SelectInputConfig) and form_input.option_source.type == ValueSourceType.VARIABLE
+            else form_input
+            for form_input in inputs
+        ]
 
     def submit_form_by_token(
         self,
@@ -252,6 +269,7 @@ class HumanInputService:
         form_data: Mapping[str, Any],
     ) -> dict[str, JsonValue]:
         definition = form.get_definition()
+        definition = definition.model_copy(update={"inputs": self.resolve_form_inputs(form)})
         try:
             return self.validate_and_normalize_submission(
                 tenant_id=form.tenant_id,

api/tests/test_containers_integration_tests/controllers/web/test_human_input_form.py

@@ -231,6 +231,6 @@ def test_get_human_input_form_resolves_runtime_select_options(
 
     assert response.status_code == 200, response.get_data(as_text=True)
     body = json.loads(response.get_data(as_text=True))
-    assert body["inputs"][0]["option_source"]["type"] == "variable"
-    assert body["inputs"][0]["option_source"]["selector"] == ["start", "options"]
+    assert body["inputs"][0]["option_source"]["type"] == "constant"
+    assert body["inputs"][0]["option_source"]["selector"] == []
     assert body["inputs"][0]["option_source"]["value"] == ["approve", "reject"]

api/tests/unit_tests/controllers/console/test_human_input_form.py

@@ -98,6 +98,7 @@ def test_get_form_definition_success(app: Flask, monkeypatch: pytest.MonkeyPatch
     expiration = datetime(2024, 1, 1, tzinfo=UTC)
     definition = SimpleNamespace(model_dump=lambda: {"fields": ["a"]})
     form = SimpleNamespace(tenant_id="tenant-1", get_definition=lambda: definition, expiration_time=expiration)
+    resolved_input = SimpleNamespace(model_dump=lambda **_kwargs: {"id": "choice", "options": ["a"]})
 
     class _ServiceStub:
         def __init__(self, *_args, **_kwargs):
@@ -106,6 +107,10 @@ def test_get_form_definition_success(app: Flask, monkeypatch: pytest.MonkeyPatch
         def get_form_definition_by_token_for_console(self, _token):
             return form
 
+        def resolve_form_inputs(self, resolved_form):
+            assert resolved_form is form
+            return [resolved_input]
+
     monkeypatch.setattr("controllers.console.human_input_form.HumanInputService", _ServiceStub)
 
     api = ConsoleHumanInputFormApi()
@@ -116,6 +121,7 @@ def test_get_form_definition_success(app: Flask, monkeypatch: pytest.MonkeyPatch
 
     payload = json.loads(response.get_data(as_text=True))
     assert payload["fields"] == ["a"]
+    assert payload["inputs"] == [{"id": "choice", "options": ["a"]}]
 
 
 def test_get_form_definition_not_found(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:

api/tests/unit_tests/core/repositories/test_human_input_form_repository_impl.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timedelta
 
 import pytest
 from sqlalchemy import Engine, event
@@ -10,6 +10,7 @@ from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 
 from core.repositories.human_input_repository import (
+    FormNotFoundError,
     HumanInputFormRecord,
     HumanInputFormRepositoryImpl,
     HumanInputFormSubmissionRepository,
@@ -276,6 +277,39 @@ def _persist_form(
 
 
 class TestHumanInputFormRepositoryImplPublicMethods:
+    @pytest.mark.parametrize("mismatched_scope", ["tenant_id", "app_id", "workflow_run_id", "node_id", "id"])
+    def test_mark_timeout_rejects_form_outside_execution_scope(
+        self, sqlite_session: Session, mismatched_scope: str
+    ) -> None:
+        form = _make_form(expiration_time=naive_utc_now() - timedelta(minutes=1))
+        setattr(form, mismatched_scope, "other-value")
+        _persist_form(sqlite_session, form)
+        repo = HumanInputFormRepositoryImpl(tenant_id="tenant-id", app_id="app-id", workflow_execution_id="run-1")
+
+        with pytest.raises(FormNotFoundError):
+            repo.mark_timeout("node-1", form_id="form-1")
+
+        sqlite_session.refresh(form)
+        assert form.status == HumanInputFormStatus.WAITING
+
+    @pytest.mark.parametrize("already_submitted", [False, True])
+    def test_mark_timeout_preserves_waiting_form_that_is_not_eligible(
+        self, sqlite_session: Session, already_submitted: bool
+    ) -> None:
+        form = _make_form(
+            expiration_time=naive_utc_now() + timedelta(minutes=-1 if already_submitted else 1),
+            submitted_at=naive_utc_now() if already_submitted else None,
+        )
+        _persist_form(sqlite_session, form)
+        repo = HumanInputFormRepositoryImpl(tenant_id="tenant-id", app_id="app-id", workflow_execution_id="run-1")
+
+        entity = repo.mark_timeout("node-1", form_id="form-1")
+
+        assert entity.status == HumanInputFormStatus.WAITING
+        assert entity.submitted == already_submitted
+        sqlite_session.refresh(form)
+        assert form.status == HumanInputFormStatus.WAITING
+
     def test_get_form_returns_entity_and_recipients(self, sqlite_session: Session):
         form = _make_form()
         recipient = _make_recipient(form.id)

api/tests/unit_tests/core/workflow/nodes/human_input/test_human_input_form_filled_event.py

@@ -113,12 +113,20 @@ def _persist_form(
 
 
 class _FakeFormRepository:
+    """Serve pre-seeded forms and retain their terminal state on timeout."""
+
     def __init__(self, form):
         self._form = form
 
     def get_form(self, *_args, **_kwargs):
         return self._form
 
+    def mark_timeout(self, _node_id: str, *, form_id: str):
+        assert form_id == self._form.id
+        if self._form.status == HumanInputFormStatus.WAITING and not self._form.submitted:
+            self._form.status = HumanInputFormStatus.TIMEOUT
+        return self._form
+
 
 class _TestFileReferenceFactory(FileReferenceFactoryProtocol):
     def build_from_mapping(self, *, mapping: Mapping[str, Any]):

api/tests/unit_tests/core/workflow/test_human_input_callback.py

@@ -8,10 +8,21 @@ import pytest
 from pytest_mock import MockerFixture
 from sqlalchemy.orm import Session, sessionmaker
 
-from core.repositories.human_input_repository import FormCreateParams, HumanInputFormRepositoryImpl
+from core.repositories.human_input_repository import (
+    FormCreateParams,
+    HumanInputFormEntity,
+    HumanInputFormRepositoryImpl,
+    HumanInputFormSubmissionRepository,
+)
 from core.workflow.nodes.human_input.callback import DifyHITLCallback
-from core.workflow.nodes.human_input.entities import HumanInputNodeData, ParagraphInputConfig, UserActionConfig
-from core.workflow.nodes.human_input.enums import HumanInputFormStatus
+from core.workflow.nodes.human_input.entities import (
+    HumanInputNodeData,
+    ParagraphInputConfig,
+    SelectInputConfig,
+    StringListSource,
+    UserActionConfig,
+)
+from core.workflow.nodes.human_input.enums import HumanInputFormStatus, ValueSourceType
 from core.workflow.nodes.human_input.session_binding import SessionBinding
 from graphon.runtime import VariablePool
 from graphon.variables.factory import build_segment
@@ -75,6 +86,42 @@ def test_dify_hitl_callback_creates_pause_requested_for_new_form(
     assert params.node_id == "node-1"
 
 
+def test_dify_hitl_callback_persists_variable_select_options(
+    repository: HumanInputFormRepositoryImpl, mocker: MockerFixture, sqlite_session: Session
+) -> None:
+    create = mocker.spy(repository, "create_form")
+    variable_pool = VariablePool()
+    variable_pool.add(("source", "options"), ["approve", "reject"])
+    callback = DifyHITLCallback(
+        form_repository=repository,
+        node_data=HumanInputNodeData(
+            title="Approval",
+            inputs=[
+                SelectInputConfig(
+                    output_variable_name="decision",
+                    option_source=StringListSource(
+                        type=ValueSourceType.VARIABLE,
+                        selector=("source", "options"),
+                    ),
+                )
+            ],
+        ),
+    )
+
+    callback(_Context(workflow_execution_id="run-1", node_id="node-1", variable_pool=variable_pool))
+
+    params: FormCreateParams = create.call_args.args[0]
+    select_input = params.form_config.inputs[0]
+    assert isinstance(select_input, SelectInputConfig)
+    assert select_input.option_source.type == ValueSourceType.CONSTANT
+    assert select_input.option_source.selector == ()
+    assert select_input.option_source.value == ["approve", "reject"]
+    form = sqlite_session.get(HumanInputForm, create.spy_return.id)
+    assert form is not None
+    option_source = json.loads(form.form_definition)["inputs"][0]["option_source"]
+    assert option_source == {"type": "constant", "selector": [], "value": ["approve", "reject"]}
+
+
 def test_dify_hitl_callback_scopes_form_to_node_execution(
     repository: HumanInputFormRepositoryImpl, mocker: MockerFixture
 ) -> None:
@@ -177,33 +224,39 @@ def test_dify_hitl_callback_returns_timeout_for_explicit_timeout_form(
     }
 
 
-def test_dify_hitl_callback_returns_timeout_for_waiting_form_past_node_deadline(
-    repository: HumanInputFormRepositoryImpl, sqlite_session: Session
-) -> None:
-    form = HumanInputForm(
-        id="form-1",
-        rendered_content="<p>Please approve</p>",
-        selected_action_id=None,
-        submitted_data=None,
-        status=HumanInputFormStatus.WAITING,
-        created_at=naive_utc_now(),
-        expiration_time=naive_utc_now() - timedelta(minutes=1),
-        submitted_at=None,
-        tenant_id="tenant-1",
-        app_id="app-1",
-        workflow_run_id="run-1",
-        node_id="node-1",
-        form_definition="{}",
-    )
-    sqlite_session.add(form)
-    sqlite_session.commit()
+def _past_deadline_callback(
+    session: Session,
+    repository: HumanInputFormRepositoryImpl,
+) -> tuple[DifyHITLCallback, HumanInputFormRepositoryImpl]:
     callback = DifyHITLCallback(
         form_repository=repository,
-        node_data=HumanInputNodeData(title="Approval", form_content="Please approve"),
+        node_data=HumanInputNodeData(
+            title="Approval",
+            form_content="<p>Please approve</p>",
+            inputs=[ParagraphInputConfig(output_variable_name="answer")],
+            user_actions=[UserActionConfig(id="approve", title="Approve")],
+        ),
+        execution_id_getter=lambda: "form-1",
     )
+    callback(_ctx("run-1", "node-1"))
+    form = session.get(HumanInputForm, "form-1")
+    assert form is not None
+    form.expiration_time = naive_utc_now() - timedelta(minutes=1)
+    session.commit()
+    return callback, repository
+
+
+def test_dify_hitl_callback_persists_timeout_for_waiting_form_past_node_deadline(
+    repository: HumanInputFormRepositoryImpl, sqlite_session: Session
+) -> None:
+    callback, _ = _past_deadline_callback(sqlite_session, repository)
 
     decision = callback(_ctx("run-1", "node-1"))
 
+    sqlite_session.expire_all()
+    form = sqlite_session.get(HumanInputForm, "form-1")
+    assert form is not None
+    assert form.status == HumanInputFormStatus.TIMEOUT
     assert decision.selected_handle == "__timeout"
     assert decision.outputs == {
         "__action_id": build_segment(""),
@@ -212,6 +265,50 @@ def test_dify_hitl_callback_returns_timeout_for_waiting_form_past_node_deadline(
     }
 
 
+@pytest.mark.parametrize("terminal_status", [HumanInputFormStatus.SUBMITTED, HumanInputFormStatus.EXPIRED])
+def test_dify_hitl_callback_timeout_uses_concurrent_terminal_state(
+    monkeypatch: pytest.MonkeyPatch,
+    repository: HumanInputFormRepositoryImpl,
+    sqlite_session: Session,
+    terminal_status: HumanInputFormStatus,
+) -> None:
+    callback, repository = _past_deadline_callback(sqlite_session, repository)
+    get_form = repository.get_form
+
+    def finish_after_read(node_id: str, *, form_id: str | None = None) -> HumanInputFormEntity | None:
+        waiting_form = get_form(node_id, form_id=form_id)
+        monkeypatch.setattr(repository, "get_form", get_form)
+        if terminal_status == HumanInputFormStatus.SUBMITTED:
+            HumanInputFormSubmissionRepository().mark_submitted(
+                form_id="form-1",
+                recipient_id=None,
+                selected_action_id="approve",
+                form_data={"answer": "accepted before timeout"},
+                submission_user_id="user-1",
+                submission_end_user_id=None,
+            )
+        else:
+            form = sqlite_session.get(HumanInputForm, "form-1")
+            assert form is not None
+            form.status = HumanInputFormStatus.EXPIRED
+            sqlite_session.commit()
+        return waiting_form
+
+    monkeypatch.setattr(repository, "get_form", finish_after_read)
+    if terminal_status == HumanInputFormStatus.SUBMITTED:
+        decision = callback(_ctx("run-1", "node-1"))
+        assert decision.selected_handle == "approve"
+        assert decision.inputs == {"answer": build_segment("accepted before timeout")}
+    else:
+        with pytest.raises(AssertionError, match="globally expired human input form"):
+            callback(_ctx("run-1", "node-1"))
+
+    sqlite_session.expire_all()
+    form = sqlite_session.get(HumanInputForm, "form-1")
+    assert form is not None
+    assert form.status == terminal_status
+
+
 def test_dify_hitl_callback_rejects_expired_form_as_invalid_resume_state(
     repository: HumanInputFormRepositoryImpl, sqlite_session: Session
 ) -> None:

api/tests/unit_tests/core/workflow/test_human_input_policy.py

@@ -9,6 +9,7 @@ from core.workflow.human_input_policy import (
 from core.workflow.nodes.human_input.entities import SelectInputConfig, StringListSource
 from core.workflow.nodes.human_input.enums import ValueSourceType
 from graphon.runtime import VariablePool
+from graphon.variables import ArrayStringSegment
 from models.human_input import RecipientType
 
 
@@ -57,9 +58,10 @@ def test_preferred_form_token_skips_prioritized_type_with_empty_token() -> None:
     assert get_preferred_form_token(recipients) == "console-token"
 
 
-def test_resolve_variable_select_input_options_uses_runtime_values() -> None:
+@pytest.mark.parametrize("child_options", [["approve", "reject"], []])
+def test_resolve_variable_select_input_options_uses_runtime_values(child_options: list[str]) -> None:
     variable_pool = VariablePool()
-    variable_pool.add(("start", "options"), ["approve", "reject"])
+    variable_pool.add(("start", "options"), ArrayStringSegment(value=child_options))
     inputs: list[SelectInputConfig] = [
         SelectInputConfig(
             output_variable_name="decision",
@@ -73,7 +75,9 @@ def test_resolve_variable_select_input_options_uses_runtime_values() -> None:
 
     resolved = resolve_variable_select_input_options(inputs, variable_pool=variable_pool)
     assert isinstance(resolved[0], SelectInputConfig)
-    assert resolved[0].option_source.value == ["approve", "reject"]
+    assert resolved[0].option_source.type == ValueSourceType.CONSTANT
+    assert resolved[0].option_source.selector == ()
+    assert resolved[0].option_source.value == child_options
 
 
 def test_resolve_variable_select_input_options_keeps_original_when_value_not_string_list() -> None:

api/tests/unit_tests/services/test_human_input_service.py

@@ -332,10 +332,33 @@ def test_resolve_form_inputs_uses_runtime_select_options(
     assert len(resolved_inputs) == 1
     resolved_input = resolved_inputs[0]
     assert isinstance(resolved_input, SelectInputConfig)
+    assert resolved_input.option_source.type == ValueSourceType.CONSTANT
+    assert resolved_input.option_source.selector == ()
     assert resolved_input.option_source.value == ["approve", "reject"]
     workflow_run_repo.get_workflow_pause.assert_called_once_with(record.workflow_run_id)
 
 
+def test_constant_select_rejects_unlisted_submission(
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
+    configured_input = SelectInputConfig(
+        output_variable_name="decision",
+        option_source=StringListSource(
+            type=ValueSourceType.CONSTANT,
+            value=["approve", "reject"],
+        ),
+    )
+    record = dataclasses.replace(
+        sample_form_record,
+        definition=sample_form_record.definition.model_copy(update={"inputs": [configured_input]}),
+    )
+    service = HumanInputService(unbound_session_factory)
+
+    with pytest.raises(InvalidFormDataError, match="Invalid value for select input 'decision': forged"):
+        service._validate_submission(Form(record), selected_action_id="submit", form_data={"decision": "forged"})
+
+
 def test_submit_form_by_token_calls_repository_and_enqueue(
     form_repository: Callable[[HumanInputFormRecord | None], HumanInputFormSubmissionRepository],
     sample_form_record: HumanInputFormRecord,