fix(dify-builder): never report a truncated test run as a failed one

Streaming introduced a correctness regression: if the event stream ended
without a terminal frame (idle timeout, dead publishing worker), run_draft
synthesised status="failed" with an empty dify_run_id. A run that had in fact
SUCCEEDED would then be reported as a failed build and sent into the repair
loop. The blocking path would have waited and returned the truth.

The stream is the fast path, not the authority. Every non-ping frame is
stamped with workflow_run_id by convert_stream_full_response, so the run id
is known from the first frame onwards -- long before the terminal frame that
may never arrive. run_draft now captures it from any frame and, when no
terminal frame arrives, reads the run's real status from the workflow-run row
instead of inventing one.

Note that data["id"] is the node-execution id on node frames, not the run id,
so it is only read as a run id on workflow-level frames.

When even the row cannot settle it -- no run id, no row, or a row still
marked running because the worker died mid-run -- the result is reported as
status="running" with an explicit Run.error, which is what Run.status already
documents for "not over / not known". A truncated stream is now either the
truth or a visible unknown, never a fabricated failure.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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:

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.

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 == ""
 
 

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}