diff --git a/api/services/dify_builder/dify_port.py b/api/services/dify_builder/dify_port.py index 3011bc1d782..775d8e6e3e6 100644 --- a/api/services/dify_builder/dify_port.py +++ b/api/services/dify_builder/dify_port.py @@ -28,7 +28,11 @@ the P2 plan's Global Constraints): ``AppGenerateService`` hand the execution to the ``workflow_based_app_execution`` Celery queue and hand *us* a Redis-backed event stream; the Builder's own task runs on the separate ``dify_builder`` - queue, so the two never contend for the same worker slots. + queue, so the two never contend for the same worker slots. If the stream + ends WITHOUT a terminal frame (idle timeout, dead publishing worker), the + outcome is recovered from the workflow-run row rather than synthesised as a + failure -- reporting a run that actually succeeded as failed would feed the + repair loop a lie. - ``publish`` MUST update ``app.workflow_id`` in the same transaction as ``publish_workflow``: ``publish_workflow`` only creates the new published ``Workflow`` row: it does not repoint the app at it. Skipping that update @@ -68,8 +72,12 @@ from services.dify_builder import graph_ops from services.dify_builder.errors import HashMismatchError, WorkflowNotInitializedError from services.dify_builder.identity import load_app, resolve_account from services.dify_builder.run_mapping import ( + is_unfinished_run_status, map_run_result, + map_unknown_run_outcome, node_event_from_stream_chunk, + run_id_from_stream_chunk, + run_result_data_from_run_row, run_result_data_from_terminal_chunk, stream_chunk_as_mapping, ) @@ -322,7 +330,8 @@ class WorkflowServiceDifyPort: streaming=True, ) - final: dict[str, Any] = {"id": ""} + final: dict[str, Any] = {} + stream_run_id = "" try: for chunk in response: # Streaming yields SSE-formatted strings ("data: {...}" events @@ -330,6 +339,9 @@ class WorkflowServiceDifyPort: payload = stream_chunk_as_mapping(chunk) if payload is None: continue + # Every non-ping frame stamps the run id, so it is known long + # before the terminal frame -- which may never arrive. + stream_run_id = stream_run_id or run_id_from_stream_chunk(payload) node_event = node_event_from_stream_chunk(payload) if node_event is not None: on_event(node_event) @@ -346,7 +358,7 @@ class WorkflowServiceDifyPort: if callable(close): close() - run_id = str(final.get("id") or "") + run_id = str(final.get("id") or stream_run_id or "") # The stream carries progress, not outputs: ``map_run_result`` still # needs the node-execution rows for ``per_node[].outputs``. @@ -358,7 +370,36 @@ class WorkflowServiceDifyPort: else [] ) - return map_run_result(final, node_execs) + if final: + return map_run_result(final, node_execs) + + # The stream ended without a terminal frame -- the subscription hit its + # idle timeout, or the publishing worker died. The stream is no longer + # the authority on how the run went; the database is. Synthesising + # "failed" here would report a run that may well have SUCCEEDED as a + # failed build and feed the repair loop a lie. + return self._run_result_without_terminal_frame(tenant_id, app_id, run_id, node_execs) + + @staticmethod + def _run_result_without_terminal_frame(tenant_id: str, app_id: str, run_id: str, node_execs: list[Any]) -> Run: + """Recover a truncated stream's outcome from the workflow-run row.""" + run_row = ( + DifyAPIRepositoryFactory.create_api_workflow_run_repository( + sessionmaker(bind=db.engine) + ).get_workflow_run_by_id(tenant_id=tenant_id, app_id=app_id, run_id=run_id) + if run_id + else None + ) + if run_row is None: + # No id, or no row: nothing can say how this went. + return map_unknown_run_outcome({"id": run_id}, node_execs) + + recovered = run_result_data_from_run_row(run_row) + if is_unfinished_run_status(recovered["status"]): + # The row says the run is still going, so its outcome is genuinely + # unknown rather than bad. + return map_unknown_run_outcome(recovered, node_execs) + return map_run_result(recovered, node_execs) def publish(self, app_id: str, actor: Actor) -> None: with _session_factory()() as session: diff --git a/api/services/dify_builder/run_mapping.py b/api/services/dify_builder/run_mapping.py index 6cdc7e3ad36..597bb20b13d 100644 --- a/api/services/dify_builder/run_mapping.py +++ b/api/services/dify_builder/run_mapping.py @@ -22,12 +22,21 @@ read as green to the user, even if it's technically "still running" import json from collections.abc import Mapping, Sequence +from dataclasses import replace from typing import Any from core.dify_builder.models import NodeEvent, NodeOutput, Run _FAILED_NODE_STATUSES = {"failed", "exception"} +# Workflow-run statuses that mean "not over yet" (``WorkflowExecutionStatus``); +# "" covers a row we could not read at all. +_UNFINISHED_RUN_STATUSES = {"", "scheduled", "running"} + +# What a truncated stream reports instead of a fabricated failure. Kept in the +# domain ``Run.error`` so the reason reaches the user rather than a bare status. +TRUNCATED_STREAM_ERROR = "the workflow run's progress stream ended before the run did, so its outcome is unknown" + # ``BaseAppGenerator.convert_to_event_stream`` (the tail of every streaming # ``AppGenerateService.generate``) emits ``f"data: {json}\n\n"`` for an event # mapping and ``f"event: {msg}\n\n"`` for a bare keep-alive such as ``"ping"``. @@ -40,6 +49,10 @@ _NODE_STREAM_EVENTS = {"node_started": "running", "node_finished": ""} # The two events that end a stream (``streaming_utils._normalize_terminal_events``). _TERMINAL_STREAM_EVENTS = {"workflow_finished", "workflow_paused"} +# Events whose ``data["id"]`` is the RUN id. On a node frame ``data["id"]`` is +# the node-execution id instead, so it must never be read as a run id. +_RUN_ID_DATA_EVENTS = {"workflow_started", "workflow_finished", "workflow_paused"} + def _status_value(status: Any) -> Any: """Unwrap a ``StrEnum``'s ``.value`` if present, else pass through as-is. @@ -98,6 +111,62 @@ def node_event_from_stream_chunk(chunk: Mapping[str, Any]) -> NodeEvent | None: ) +def run_id_from_stream_chunk(chunk: Mapping[str, Any]) -> str: + """The workflow run id a chunk carries, or ``""``. + + ``convert_stream_full_response`` stamps ``workflow_run_id`` on *every* + non-ping frame, so the run id is known from the first frame onwards -- + long before the terminal frame that may never arrive. That is what lets a + truncated stream still ask the database how the run actually went. + """ + direct = chunk.get("workflow_run_id") + if direct: + return str(direct) + data = chunk.get("data") + if not isinstance(data, Mapping): + return "" + if data.get("workflow_run_id"): + return str(data["workflow_run_id"]) + if str(chunk.get("event") or "") in _RUN_ID_DATA_EVENTS: + return str(data.get("id") or "") + return "" + + +def run_result_data_from_run_row(run_row: Any) -> dict[str, Any]: + """A ``WorkflowRun`` row shaped the way ``map_run_result`` expects it. + + The stream is the fast path, not the authority: when it ends without a + terminal frame the row is what actually says how the run went. Duck-typed + like the node-execution rows, so tests can pass stand-ins. + """ + return { + "id": str(getattr(run_row, "id", "") or ""), + "status": _status_value(getattr(run_row, "status", "")) or "", + "error": getattr(run_row, "error", "") or "", + "elapsed_time": getattr(run_row, "elapsed_time", 0) or 0, + "total_tokens": getattr(run_row, "total_tokens", 0) or 0, + } + + +def is_unfinished_run_status(status: Any) -> bool: + """True while a workflow run has not reached any final state.""" + return str(_status_value(status) or "") in _UNFINISHED_RUN_STATUSES + + +def map_unknown_run_outcome(data: Mapping[str, Any], node_execs: Sequence[Any]) -> Run: + """A run whose outcome nothing could establish -- NOT a failed run. + + ``map_run_result`` is deliberately conservative: anything that is not a + clean ``succeeded`` reads as ``failed``. That is right for a run we watched + end, and wrong for one we merely lost sight of -- a fabricated ``failed`` + sends a build that may well have succeeded straight into the repair loop. + ``Run.status`` documents ``running`` alongside ``succeeded``/``failed`` + exactly for this "not over / not known" case, and ``Run.error`` carries the + reason so it is visible instead of silent. + """ + return replace(map_run_result(data, node_execs), status="running", error=TRUNCATED_STREAM_ERROR) + + def run_result_data_from_terminal_chunk(chunk: Mapping[str, Any]) -> dict[str, Any] | None: """The ``data`` a terminal streaming chunk carries, shaped the way ``map_run_result`` expects it, or ``None`` for a non-terminal chunk. diff --git a/api/tests/unit_tests/services/dify_builder/test_dify_port.py b/api/tests/unit_tests/services/dify_builder/test_dify_port.py index 9c013d796b4..042e65b04b6 100644 --- a/api/tests/unit_tests/services/dify_builder/test_dify_port.py +++ b/api/tests/unit_tests/services/dify_builder/test_dify_port.py @@ -13,7 +13,9 @@ The goal is to pin the adapter's ORCHESTRATION and the P2 plan's OSS gotchas: ``invoke_from=InvokeFrom.DEBUGGER`` and ``streaming=True``, emits an ``on_event`` per node frame *as the run streams*, consumes that stream outside the ``Session`` block, and still reads the node-execution rows once - at the end for the ``run_mapping``-mapped ``Run``. + at the end for the ``run_mapping``-mapped ``Run``. A stream that ends without + a terminal frame never fabricates a failure -- it recovers the run's real + status from the database, or reports the outcome as unknown. - ``publish`` calls ``publish_workflow`` AND sets ``app.workflow_id`` AND commits -- omitting the ``workflow_id`` update makes publish a silent no-op, so all three are asserted together. @@ -33,6 +35,7 @@ from core.dify_builder.models import Actor, ChangedNode, MutationIntent, NodeEve from core.dify_builder.ports import DifyPort from models.account import Account from models.model import App +from services.dify_builder import run_mapping from services.dify_builder.dify_port import WorkflowServiceDifyPort from services.dify_builder.errors import HashMismatchError, WorkflowNotInitializedError from services.errors.app import WorkflowHashNotEqualError @@ -711,8 +714,126 @@ def test_run_draft_maps_a_paused_stream_to_a_failed_run(mock_session: MagicMock) assert run.dify_run_id == "run-9" -def test_run_draft_survives_a_stream_that_never_reaches_a_terminal_event(mock_session: MagicMock): - """No terminal frame -> no run id -> skip the node-execution read, report failed.""" +# ---- run_draft: truncated streams must never fabricate a failure ------------- + + +def _truncated_stream() -> list[str]: + """A stream that stamps the run id and then just stops -- no terminal frame. + + This is what the subscription's idle timeout (or a dead publishing worker) + leaves behind. + """ + return [ + _sse({"event": "workflow_started", "workflow_run_id": "run-1", "data": {"id": "run-1"}}), + _sse( + { + "event": "node_started", + "workflow_run_id": "run-1", + "data": {"id": "node-exec-1", "node_id": "node-1", "title": "Code"}, + } + ), + ] + + +def test_a_truncated_stream_reports_the_runs_real_status_from_the_database(mock_session: MagicMock): + """THE regression this guards: a slow run that SUCCEEDED must not read as failed. + + The stream can end early while the run is still fine. Synthesising "failed" + would send a perfectly good build into the repair loop. + """ + _configure_default_identity(mock_session) + + with ( + patch("services.dify_builder.dify_port.AppGenerateService") as mock_ags, + patch("services.dify_builder.dify_port.DifyAPIRepositoryFactory") as mock_repo_factory, + ): + mock_ags.generate.return_value = iter(_truncated_stream()) + node_repo = mock_repo_factory.create_api_workflow_node_execution_repository.return_value + node_repo.get_executions_by_workflow_run.return_value = [_node_exec()] + run_repo = mock_repo_factory.create_api_workflow_run_repository.return_value + run_repo.get_workflow_run_by_id.return_value = SimpleNamespace( + id="run-1", status="succeeded", error=None, elapsed_time=361.0, total_tokens=99 + ) + + run = WorkflowServiceDifyPort().run_draft("app-1", _actor(), {}, lambda _e: None) + + # The run id came off a non-terminal frame, so both reads could still happen. + node_repo.get_executions_by_workflow_run.assert_called_once_with("tenant-1", "app-1", "run-1") + run_repo.get_workflow_run_by_id.assert_called_once_with(tenant_id="tenant-1", app_id="app-1", run_id="run-1") + assert run.status == "succeeded" + assert run.error == "" + assert run.dify_run_id == "run-1" + assert run.tokens == 99 + assert run.per_node[0].node_id == "node-1" + + +def test_a_truncated_stream_reports_a_real_database_failure_as_failed(mock_session: MagicMock): + """The flip side: a genuine failure recovered from the row still reads failed.""" + _configure_default_identity(mock_session) + + with ( + patch("services.dify_builder.dify_port.AppGenerateService") as mock_ags, + patch("services.dify_builder.dify_port.DifyAPIRepositoryFactory") as mock_repo_factory, + ): + mock_ags.generate.return_value = iter(_truncated_stream()) + node_repo = mock_repo_factory.create_api_workflow_node_execution_repository.return_value + node_repo.get_executions_by_workflow_run.return_value = [] + run_repo = mock_repo_factory.create_api_workflow_run_repository.return_value + run_repo.get_workflow_run_by_id.return_value = SimpleNamespace( + id="run-1", status="failed", error="node blew up", elapsed_time=1.0, total_tokens=1 + ) + + run = WorkflowServiceDifyPort().run_draft("app-1", _actor(), {}, lambda _e: None) + + assert run.status == "failed" + assert run.dify_run_id == "run-1" + + +def test_a_truncated_stream_over_a_still_running_run_is_unknown_not_failed(mock_session: MagicMock): + """Outcome genuinely unknown -> "running" + a visible reason, never "failed".""" + _configure_default_identity(mock_session) + + with ( + patch("services.dify_builder.dify_port.AppGenerateService") as mock_ags, + patch("services.dify_builder.dify_port.DifyAPIRepositoryFactory") as mock_repo_factory, + ): + mock_ags.generate.return_value = iter(_truncated_stream()) + node_repo = mock_repo_factory.create_api_workflow_node_execution_repository.return_value + node_repo.get_executions_by_workflow_run.return_value = [] + run_repo = mock_repo_factory.create_api_workflow_run_repository.return_value + run_repo.get_workflow_run_by_id.return_value = SimpleNamespace( + id="run-1", status="running", error=None, elapsed_time=300.0, total_tokens=5 + ) + + run = WorkflowServiceDifyPort().run_draft("app-1", _actor(), {}, lambda _e: None) + + assert run.status == "running" + assert run.status != "failed" + assert run.error == run_mapping.TRUNCATED_STREAM_ERROR + assert run.dify_run_id == "run-1" + + +def test_a_truncated_stream_with_no_run_row_is_unknown_not_failed(mock_session: MagicMock): + _configure_default_identity(mock_session) + + with ( + patch("services.dify_builder.dify_port.AppGenerateService") as mock_ags, + patch("services.dify_builder.dify_port.DifyAPIRepositoryFactory") as mock_repo_factory, + ): + mock_ags.generate.return_value = iter(_truncated_stream()) + node_repo = mock_repo_factory.create_api_workflow_node_execution_repository.return_value + node_repo.get_executions_by_workflow_run.return_value = [] + run_repo = mock_repo_factory.create_api_workflow_run_repository.return_value + run_repo.get_workflow_run_by_id.return_value = None + + run = WorkflowServiceDifyPort().run_draft("app-1", _actor(), {}, lambda _e: None) + + assert run.status == "running" + assert run.error == run_mapping.TRUNCATED_STREAM_ERROR + + +def test_a_stream_of_nothing_but_keep_alives_is_unknown_not_failed(mock_session: MagicMock): + """No frame ever carried a run id, so there is nothing to look up.""" _configure_default_identity(mock_session) with ( @@ -720,12 +841,16 @@ def test_run_draft_survives_a_stream_that_never_reaches_a_terminal_event(mock_se patch("services.dify_builder.dify_port.DifyAPIRepositoryFactory") as mock_repo_factory, ): mock_ags.generate.return_value = iter(["event: ping\n\n"]) - repo = mock_repo_factory.create_api_workflow_node_execution_repository.return_value + node_repo = mock_repo_factory.create_api_workflow_node_execution_repository.return_value + run_repo = mock_repo_factory.create_api_workflow_run_repository.return_value run = WorkflowServiceDifyPort().run_draft("app-1", _actor(), {}, lambda _e: None) - repo.get_executions_by_workflow_run.assert_not_called() - assert run.status == "failed" + node_repo.get_executions_by_workflow_run.assert_not_called() + run_repo.get_workflow_run_by_id.assert_not_called() + assert run.status == "running" + assert run.status != "failed" + assert run.error == run_mapping.TRUNCATED_STREAM_ERROR assert run.dify_run_id == "" diff --git a/api/tests/unit_tests/services/dify_builder/test_run_mapping.py b/api/tests/unit_tests/services/dify_builder/test_run_mapping.py index 41c9ee18358..98669c4305b 100644 --- a/api/tests/unit_tests/services/dify_builder/test_run_mapping.py +++ b/api/tests/unit_tests/services/dify_builder/test_run_mapping.py @@ -13,8 +13,13 @@ from types import SimpleNamespace from core.dify_builder.models import NodeEvent, NodeOutput, Run from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus from services.dify_builder.run_mapping import ( + TRUNCATED_STREAM_ERROR, + is_unfinished_run_status, map_run_result, + map_unknown_run_outcome, node_event_from_stream_chunk, + run_id_from_stream_chunk, + run_result_data_from_run_row, run_result_data_from_terminal_chunk, stream_chunk_as_mapping, to_node_event, @@ -325,3 +330,85 @@ def test_workflow_paused_backfills_the_run_id_map_run_result_requires(): # A paused run is never green. assert run.status == "failed" assert run.dify_run_id == "run-9" + + +# ---- truncated streams: the run id, and the database as the authority -------- + + +def test_every_non_ping_frame_carries_the_run_id_at_the_top_level(): + """``convert_stream_full_response`` stamps ``workflow_run_id`` on each frame.""" + chunk = {"event": "node_started", "workflow_run_id": "run-1", "data": {"node_id": "n1"}} + + assert run_id_from_stream_chunk(chunk) == "run-1" + + +def test_a_node_frames_data_id_is_never_read_as_the_run_id(): + """On a node frame ``data["id"]`` is the NODE-EXECUTION id, not the run id.""" + chunk = {"event": "node_started", "data": {"id": "node-exec-1", "node_id": "n1"}} + + assert run_id_from_stream_chunk(chunk) == "" + + +def test_workflow_level_frames_fall_back_to_their_data_id(): + assert run_id_from_stream_chunk({"event": "workflow_started", "data": {"id": "run-1"}}) == "run-1" + assert run_id_from_stream_chunk({"event": "workflow_finished", "data": {"id": "run-2"}}) == "run-2" + + +def test_paused_frames_fall_back_to_their_data_workflow_run_id(): + chunk = {"event": "workflow_paused", "data": {"workflow_run_id": "run-9"}} + + assert run_id_from_stream_chunk(chunk) == "run-9" + + +def test_a_frame_with_no_run_id_yields_empty_string(): + assert run_id_from_stream_chunk({"event": "ping"}) == "" + assert run_id_from_stream_chunk({"event": "node_started", "data": "not-a-mapping"}) == "" + + +def test_run_row_maps_to_the_shape_map_run_result_expects(): + row = SimpleNamespace( + id="run-1", + status=WorkflowExecutionStatus.SUCCEEDED, + error=None, + elapsed_time=361.0, + total_tokens=99, + ) + + data = run_result_data_from_run_row(row) + + assert data == {"id": "run-1", "status": "succeeded", "error": "", "elapsed_time": 361.0, "total_tokens": 99} + run = map_run_result(data, []) + assert run.status == "succeeded" + assert run.dify_run_id == "run-1" + assert run.elapsed_ms == 361000 + + +def test_unfinished_statuses_are_the_ones_that_mean_not_over_yet(): + assert is_unfinished_run_status("running") is True + assert is_unfinished_run_status("scheduled") is True + assert is_unfinished_run_status(WorkflowExecutionStatus.RUNNING) is True + assert is_unfinished_run_status("") is True + assert is_unfinished_run_status(None) is True + assert is_unfinished_run_status("succeeded") is False + assert is_unfinished_run_status("failed") is False + assert is_unfinished_run_status(WorkflowExecutionStatus.PARTIAL_SUCCEEDED) is False + + +def test_an_unknown_outcome_is_running_with_a_visible_reason_not_failed(): + """A run we merely lost sight of must not read as a failed run.""" + run = map_unknown_run_outcome({"id": "run-1", "status": "running"}, []) + + assert run.status == "running" + assert run.status != "failed" + assert run.error == TRUNCATED_STREAM_ERROR + assert run.dify_run_id == "run-1" + + +def test_an_unknown_outcome_still_carries_the_per_node_rows_it_has(): + node = _node_exec("node-1", "code", "Code", WorkflowNodeExecutionStatus.SUCCEEDED, outputs_dict={"x": 1}) + + run = map_unknown_run_outcome({"id": "run-1"}, [node]) + + assert run.status == "running" + assert [n.node_id for n in run.per_node] == ["node-1"] + assert run.per_node[0].outputs == {"x": 1}