diff --git a/api/services/dify_builder/service.py b/api/services/dify_builder/service.py index 1c4bd54731c..da061baee90 100644 --- a/api/services/dify_builder/service.py +++ b/api/services/dify_builder/service.py @@ -1001,11 +1001,16 @@ class DifyBuilderService: view, expect_advance = self._prepare_action(session_id, actor, action) if not expect_advance: + # ``message`` (replay of a settled turn) and ``update_model`` both + # settle synchronously while changing the version, so they must emit + # a terminal ``state`` frame carrying the new version -- else a FE + # tracking its held version off ``state`` frames stays stale and its + # next action 409s. return stream_advance_frames( asdict(view), None, expect_advance=False, - emit_state_when_settled=action.kind == "message", + emit_state_when_settled=action.kind in {"message", "update_model"}, ) subscription = self._subscribe_fn(session_id) # BEFORE dispatch try: diff --git a/api/tasks/dify_builder_advance_task.py b/api/tasks/dify_builder_advance_task.py index 1200733cbef..f0cdaf7d19f 100644 --- a/api/tasks/dify_builder_advance_task.py +++ b/api/tasks/dify_builder_advance_task.py @@ -137,11 +137,24 @@ def advance_session(session_id: str, action_dict: dict, actor_dict: dict, token: if terminal_error is not None: progress_bus.publish(session_id, terminal_error) elif completed is not None: - repo, dify, actor = completed - view = DifyBuilderService( - repo, - session_lock, - lambda *a, **k: None, - get_app_revision_fn=lambda app_id, owner: dify.read_graph(app_id, owner)[1], - ).get_session_view(session_id, actor) - progress_bus.publish(session_id, {"kind": "state", **asdict(view)}) + # The advance already succeeded and the lock is released; the session is + # durable. But the view projection (read_graph) or the publish itself can + # still raise -- and if this block let that escape, NO terminal frame + # would reach the SSE client, hanging its stream to the deadline with a + # stale held version (-> 409 on its next action). Fall back to a terminal + # error frame so the client stream always ends. + try: + repo, dify, actor = completed + view = DifyBuilderService( + repo, + session_lock, + lambda *a, **k: None, + get_app_revision_fn=lambda app_id, owner: dify.read_graph(app_id, owner)[1], + ).get_session_view(session_id, actor) + progress_bus.publish(session_id, {"kind": "state", **asdict(view)}) + except Exception: + logger.exception("dify_builder terminal state projection/publish failed for session %s", session_id) + try: + progress_bus.publish(session_id, {"kind": "error", "error": "step failed"}) + except Exception: + logger.exception("dify_builder terminal error frame also failed to publish for session %s", session_id) diff --git a/api/tests/unit_tests/services/dify_builder/test_advance_task.py b/api/tests/unit_tests/services/dify_builder/test_advance_task.py index 00ba6cd8017..4ee49a62c48 100644 --- a/api/tests/unit_tests/services/dify_builder/test_advance_task.py +++ b/api/tests/unit_tests/services/dify_builder/test_advance_task.py @@ -220,6 +220,44 @@ def test_terminal_state_is_published_after_the_advance_lock_is_released( assert timeline[-2:] == ["release", "publish:state"] +def test_terminal_state_projection_failure_publishes_error_frame( + monkeypatch: pytest.MonkeyPatch, + repo: SqlDifyBuilderRepository, +) -> None: + """If the post-advance view projection (get_session_view -> read_graph) or + the final state publish raises after a SUCCESSFUL advance, the task must not + let the exception escape with no terminal frame -- that leaves the SSE client + hanging to the stream deadline with a stale held version (-> 409 on its next + action). It must fall back to publishing a terminal error frame.""" + monkeypatch.setattr(mod, "_build_repo", lambda: repo) + monkeypatch.setattr(mod, "WorkflowServiceDifyPort", FakeDifyPort) + timeline: list[str] = [] + monkeypatch.setattr( + mod.progress_bus, + "publish", + lambda _sid, event: timeline.append(f"publish:{event['kind']}"), + ) + monkeypatch.setattr( + mod.session_lock, + "release", + lambda _sid, _token: timeline.append("release"), + ) + + class _BoomService: + def __init__(self, *_a, **_k) -> None: ... + + def get_session_view(self, *_a, **_k): + raise RuntimeError("projection boom") + + monkeypatch.setattr(mod, "DifyBuilderService", _BoomService) + s = _seed_fix_session(repo) + + # Must NOT raise, and must still emit a terminal frame after releasing the lock. + mod.advance_session(s.id, _act("request_fix", 1), _ACTOR_DICT, "tok-boom") + + assert timeline[-2:] == ["release", "publish:error"] + + def test_message_advance_publishes_assistant_delta_before_durable_reply( repo: SqlDifyBuilderRepository, wired, diff --git a/api/tests/unit_tests/services/dify_builder/test_service.py b/api/tests/unit_tests/services/dify_builder/test_service.py index 1edd9e05247..5d471e707ed 100644 --- a/api/tests/unit_tests/services/dify_builder/test_service.py +++ b/api/tests/unit_tests/services/dify_builder/test_service.py @@ -854,6 +854,27 @@ def test_submit_action_update_model_rejects_missing_config( ) +def test_submit_action_stream_update_model_emits_terminal_state_frame( + service: DifyBuilderService, + repo: SqlDifyBuilderRepository, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """``update_model`` settles synchronously and bumps the version. The stream + MUST emit a terminal ``state`` frame carrying the new version -- otherwise a + FE that tracks its held version off ``state`` frames keeps the stale value + and its next action 409s (regression class from the 409 review).""" + s = _seed_session_at(repo, PcState.FIX_AWAIT_VERIFY) + model_config = {"provider": "openai", "name": "gpt-4o", "mode": "chat", "completion_params": {}} + monkeypatch.setattr(service_module, "validate_model_config", lambda _tenant_id, _config: None) + + action = Action(kind="update_model", payload={"model_config": model_config}, base_version=s.version) + frames = list(service.submit_action_stream(s.id, _actor(), action)) + + assert _business_event(frames[0]) == "snapshot" + state_frames = [f for f in frames if _business_event(f) == "state"] + assert state_frames, "update_model must emit a terminal state frame with the new version" + + def test_submit_action_stream_subscribes_before_dispatch_and_streams(repo: SqlDifyBuilderRepository) -> None: lock = FakeSessionLock() order: list[str] = []