From 0dd4b6dd77802a5093644fe1b9461e2445830b20 Mon Sep 17 00:00:00 2001 From: Eapp1e <163811520+Eapp1e@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:02:48 +0800 Subject: [PATCH] fix(api): read the START node sys.timestamp from the UTC clock in the workflow app runner `WorkflowAppRunner.prepare()` builds the START node system variables with `timestamp=int(naive_utc_now().timestamp())`. That value is a naive UTC wall clock, and `datetime.timestamp()` interprets naive values in the process-local timezone, so every workflow run started from the app runner reports an epoch that is shifted by the host's UTC offset (`sys.timestamp` is 8 hours behind on a `TZ=Asia/Shanghai` host). `_setup_variable_pool()` in `services/workflow_service.py` had the same call and was fixed the same way, so this brings the second START node bootstrap path in line with the `to_utc_timestamp()` contract in `libs/datetime_utils.py`. Signed-off-by: Eapp1e <163811520+Eapp1e@users.noreply.github.com> --- api/core/app/apps/workflow/app_runner.py | 4 +- .../core/app/apps/workflow/test_app_runner.py | 100 ++++++++++++++++++ 2 files changed, 102 insertions(+), 2 deletions(-) create mode 100644 api/tests/unit_tests/core/app/apps/workflow/test_app_runner.py diff --git a/api/core/app/apps/workflow/app_runner.py b/api/core/app/apps/workflow/app_runner.py index b2390ea19b156c..0bf68d634391bc 100644 --- a/api/core/app/apps/workflow/app_runner.py +++ b/api/core/app/apps/workflow/app_runner.py @@ -31,7 +31,7 @@ from graphon.enums import WorkflowType from graphon.runtime import RuntimeState, VariablePool from graphon.variable_loader import VariableLoader -from libs.datetime_utils import naive_utc_now +from libs.datetime_utils import naive_utc_now, to_utc_timestamp from models.workflow import Workflow logger = logging.getLogger(__name__) @@ -116,7 +116,7 @@ def prepare(self) -> PreparedWorkflowRun | None: files=self.application_generate_entity.files, user_id=self._sys_user_id, app_id=app_config.app_id, - timestamp=int(naive_utc_now().timestamp()), + timestamp=to_utc_timestamp(naive_utc_now()), workflow_id=app_config.workflow_id, workflow_execution_id=self.application_generate_entity.workflow_execution_id, ) diff --git a/api/tests/unit_tests/core/app/apps/workflow/test_app_runner.py b/api/tests/unit_tests/core/app/apps/workflow/test_app_runner.py new file mode 100644 index 00000000000000..38cdd5350b14ae --- /dev/null +++ b/api/tests/unit_tests/core/app/apps/workflow/test_app_runner.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import json +import os +import time +from datetime import UTC, datetime +from unittest.mock import MagicMock, patch + +import pytest + +from core.app.app_config.entities import WorkflowUIBasedAppConfig +from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager +from core.app.apps.workflow.app_runner import WorkflowAppRunner +from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom, WorkflowAppGenerateEntity +from graphon.variable_loader import DUMMY_VARIABLE_LOADER +from models.model import AppMode +from models.workflow import Workflow, WorkflowKind +from models.workflow import WorkflowType as ModelWorkflowType + + +def _runner() -> WorkflowAppRunner: + """Build a runner over real Dify objects so prepare() runs its production path.""" + app_generate_entity = WorkflowAppGenerateEntity( + app_config=WorkflowUIBasedAppConfig( + app_id="app", tenant_id="tenant", workflow_id="workflow", app_mode=AppMode.WORKFLOW + ), + inputs={"question": "hello"}, + files=[], + user_id="user", + user_from=UserFrom.END_USER, + invoke_from=InvokeFrom.SERVICE_API, + workflow_execution_id="execution-id", + task_id="task-id", + stream=True, + call_depth=1, + ) + workflow = Workflow( + tenant_id="tenant", + app_id="app", + id="workflow", + type=ModelWorkflowType.WORKFLOW, + version="v1", + graph=json.dumps({"nodes": [], "edges": []}), + kind=WorkflowKind.SNIPPET, + ) + workflow.environment_variables = [] + with patch("core.app.apps.base_app_queue_manager.redis_client"): + queue_manager = WorkflowAppQueueManager( + task_id="task-id", + user_id="user", + invoke_from=InvokeFrom.SERVICE_API, + app_mode="workflow", + ) + return WorkflowAppRunner( + application_generate_entity=app_generate_entity, + queue_manager=queue_manager, + variable_loader=DUMMY_VARIABLE_LOADER, + workflow=workflow, + system_user_id="system-user", + ) + + +def _system_variables_built_by_prepare() -> MagicMock: + """Run prepare() with the Redis-backed collaborators replaced and return the build_system_variables mock.""" + runner = _runner() + with ( + patch("core.app.apps.workflow.app_runner.RedisChannel"), + patch("core.app.apps.workflow.app_runner.redis_client"), + patch("core.app.apps.workflow.app_runner.WorkflowEntry"), + patch("core.app.apps.workflow.app_runner.build_system_variables", return_value=[]) as system_variables, + patch("core.app.apps.workflow.app_runner.build_bootstrap_variables", return_value=[]), + patch("core.app.apps.workflow.app_runner.add_variables_to_pool"), + patch("core.app.apps.workflow.app_runner.get_default_root_node_id", return_value="root-node"), + patch("core.app.apps.workflow.app_runner.get_compatible_start_aliases", return_value=()), + patch("core.app.apps.workflow.app_runner.add_node_inputs_to_pool"), + patch.object(runner, "_init_graph"), + ): + runner.prepare() + return system_variables + + +def test_prepare_should_read_the_system_timestamp_from_the_epoch(monkeypatch: pytest.MonkeyPatch) -> None: + """sys.timestamp is a Unix timestamp, so it must not follow the host timezone.""" + if not hasattr(time, "tzset"): + pytest.skip("tzset is only available on POSIX platforms") + recorded_at = datetime(2024, 5, 1, tzinfo=UTC) + monkeypatch.setattr("core.app.apps.workflow.app_runner.naive_utc_now", lambda: recorded_at.replace(tzinfo=None)) + previous_tz = os.environ.get("TZ") + monkeypatch.setenv("TZ", "Asia/Shanghai") + time.tzset() + try: + system_variables = _system_variables_built_by_prepare() + finally: + if previous_tz is None: + os.environ.pop("TZ", None) + else: + os.environ["TZ"] = previous_tz + time.tzset() + + assert system_variables.call_args.kwargs["timestamp"] == int(recorded_at.timestamp())