From f623dc3fe1eed2ece90acb87cb84ff93485d919f Mon Sep 17 00:00:00 2001 From: Charles Yao Date: Fri, 4 Sep 2026 00:07:14 -0700 Subject: [PATCH] feat(dify_builder): recover an interrupted working step via Retry/Start over When a session crashes mid working-step (e.g. build.publish; the async advance raised and released the lock), it was a locked dead-end: working states surface no UI actions and the composer locked with no way forward. - Surface a RecoveryRef offer (can_continue=Retry, can_restart=Start over, recovery_class="") on an interrupted working-state view -- built directly so a real drift class still wins. - Allow recovery_continue / recovery_restart at an interrupted working state in _internal_action_allowed. recovery_continue returns the working state (non-settled) so the runner re-runs the stalled handler; recovery_restart resets working fields and re-enters the entry state -- both reuse existing recovery machinery, no new action kinds / no runner change. Adds unit tests (offer surfaced, allowance) + regression tests (recovery_continue re-runs handle_publish and advances; recovery_restart resets to entry state; repeat-failure re-surfaces the offer with no wedge). This reconciler was reviewed as a standalone plan on the pre-merge base; here it is re-applied onto feat/dify-builder after PR #41677 merged in the reworked service.py + _internal_action_allowed gate (which makes the allowance necessary). The apply_repair idempotency + delete-guard commits (safe fix.apply Retry) land alongside it. Co-Authored-By: Claude Opus 4.8 --- api/services/dify_builder/service.py | 20 ++++++ .../core/dify_builder/test_handlers_build.py | 37 +++++++++++ .../services/dify_builder/test_service.py | 61 +++++++++++++++++++ 3 files changed, 118 insertions(+) diff --git a/api/services/dify_builder/service.py b/api/services/dify_builder/service.py index da061baee90..56ad7d18160 100644 --- a/api/services/dify_builder/service.py +++ b/api/services/dify_builder/service.py @@ -22,6 +22,7 @@ from core.dify_builder.contract import ( CheckpointRef, NoticeItem, Phase, + RecoveryRef, RunContextCard, RunStatus, SessionModel, @@ -417,6 +418,11 @@ def _internal_action_allowed(state: PcState, fc: DifyBuilderContext, kind: str) transitions must change the current pause flag, and recovery choices must be present in the current RecoveryRef. """ + # Interrupted working step: allow Retry/Start over via the existing recovery + # actions. (The lock is handled downstream -- recovery_continue's dispatch + # acquires the advance lock and raises BusyError if one is already held.) + if is_working(state) and kind in ("recovery_continue", "recovery_restart"): + return True if not is_waiting(state): return False if kind in {"check_recovery", "message", "update_model"}: @@ -852,6 +858,20 @@ class DifyBuilderService: else None ) recovery_ref = recovery.recovery_ref_for(fc.recovery_class) + # Interrupted working step (crash mid auto-advance, lock released): offer + # Retry (recovery_continue) / Start over (recovery_restart). Built directly + # with recovery_class="" so recovery_ref_for/classify never see a fabricated + # drift class. Retry re-runs the working handler (recovery_continue returns + # the working state -> non-settled -> runner re-drives it); Start over resets + # + re-enters the flow's entry state. A real drift class still wins (only fires + # when recovery_ref is None). + if recovery_ref is None and is_working(st) and not lock_held: + recovery_ref = RecoveryRef( + recovery_class="", + can_continue=True, + can_restart=True, + message="The last step didn't finish. Retry it, or start over.", + ) return SessionView( session_id=s.id, app_id=s.app_id, diff --git a/api/tests/unit_tests/core/dify_builder/test_handlers_build.py b/api/tests/unit_tests/core/dify_builder/test_handlers_build.py index c007ea02606..c8d05c646dc 100644 --- a/api/tests/unit_tests/core/dify_builder/test_handlers_build.py +++ b/api/tests/unit_tests/core/dify_builder/test_handlers_build.py @@ -1183,3 +1183,40 @@ def test_execution_revert_then_retry_after_revert_reapprove_is_idempotent(): assert out.current_state == PcState.BUILD_EXECUTION assert len(dify.graph["nodes"]) == 4 assert len(dify.graph["edges"]) == 3 + + +def test_recovery_continue_reruns_interrupted_publish_step(): + """An interrupted build.publish recovered via recovery_continue must re-run + handle_publish (the runner falls through because build.publish is a working + state) and advance to build.governance_feedback -- not stay wedged.""" + from core.dify_builder.handlers_build import build_registry + from core.dify_builder.runner import Runner + from tests.unit_tests.core.dify_builder.fakes import FakeBuildDifyPort + + dify = FakeBuildDifyPort() + env, repo = _new_env(dify=dify) + s = _seed_build_session(repo, PcState.BUILD_PUBLISH, plan_items=["x"], built_node_ids=["start_1"]) + runner = Runner(env, build_registry()) + + runner.advance(s.id, Turn(action=Action(kind="recovery_continue", base_version=s.version), actor=_actor())) + + reloaded, _fc = repo.get_session(s.id) + assert reloaded.current_state != PcState.BUILD_PUBLISH # advanced, not wedged + assert dify.published is True # handle_publish re-ran (FakeBuildDifyPort records publish) + + +def test_recovery_restart_resets_interrupted_step_to_entry_state(): + from core.dify_builder.handlers_build import build_registry + from core.dify_builder.models import EntryMode + from core.dify_builder.recovery import entry_state_for + from core.dify_builder.runner import Runner + from tests.unit_tests.core.dify_builder.fakes import FakeBuildDifyPort + + env, repo = _new_env(dify=FakeBuildDifyPort()) + s = _seed_build_session(repo, PcState.BUILD_PUBLISH, plan_items=["x"], staged_repair=[object()]) + Runner(env, build_registry()).advance( + s.id, Turn(action=Action(kind="recovery_restart", base_version=s.version), actor=_actor()) + ) + reloaded, fc = repo.get_session(s.id) + assert reloaded.current_state == entry_state_for(EntryMode.BUILD) # back at the flow entry + assert fc.staged_repair == [] # working fields reset 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 5d471e707ed..da7888c69e9 100644 --- a/api/tests/unit_tests/services/dify_builder/test_service.py +++ b/api/tests/unit_tests/services/dify_builder/test_service.py @@ -1415,3 +1415,64 @@ def test_get_session_view_no_recovery_when_unset(service: DifyBuilderService, re repo.create_session(s, DifyBuilderContext(), [ConversationItem(kind="user", seq=0)]) view = service.get_session_view(s.id, _actor()) assert view.recovery is None + + +def test_interrupted_working_state_surfaces_recovery_offer( + service: DifyBuilderService, repo: SqlDifyBuilderRepository +) -> None: + s = _seed_session_at(repo, PcState.BUILD_PUBLISH) # working state, lock unheld -> interrupted + view = service.get_session_view(s.id, _actor()) + assert view.interrupted is True + assert view.recovery is not None + assert view.recovery.can_continue is True # Retry + assert view.recovery.can_restart is True # Start over + assert view.recovery.recovery_class == "" # not a drift class + + +def test_recovery_actions_allowed_at_interrupted_working_state() -> None: + from core.dify_builder.models import DifyBuilderContext + from services.dify_builder.service import _internal_action_allowed + + fc = DifyBuilderContext() + assert _internal_action_allowed(PcState.BUILD_PUBLISH, fc, "recovery_continue") is True + assert _internal_action_allowed(PcState.BUILD_PUBLISH, fc, "recovery_restart") is True + # an unrelated action is still rejected at a working state + assert _internal_action_allowed(PcState.BUILD_PUBLISH, fc, "publish_workflow") is False + + +def test_repeat_failure_resurfaces_offer_no_wedge( + service: DifyBuilderService, repo: SqlDifyBuilderRepository, lock: FakeSessionLock, monkeypatch +) -> None: + """A Retry (recovery_continue) whose handler raises again must not wedge the + session: the task's generic except publishes a "step failed" event and releases + the lock in `finally`, and afterward the session still reads as interrupted with + the recovery offer present -- never a dead end.""" + import tasks.dify_builder_advance_task as advance_mod + from tests.unit_tests.core.dify_builder.fakes import FakeBuildDifyPort + + class RaisingDify(FakeBuildDifyPort): + def publish(self, _app_id, _actor) -> None: + raise RuntimeError("boom: publish still broken") + + monkeypatch.setattr(advance_mod, "_build_repo", lambda: repo) + monkeypatch.setattr(advance_mod, "WorkflowServiceDifyPort", RaisingDify) + monkeypatch.setattr(advance_mod, "session_lock", lock) + events: list[tuple[str, dict]] = [] + monkeypatch.setattr(advance_mod.progress_bus, "publish", lambda sid, ev: events.append((sid, ev))) + + s = _seed_session_at(repo, PcState.BUILD_PUBLISH) # already interrupted: working, lock unheld + actor = _actor() + token = lock.acquire(s.id) + assert token is not None + + action = {"kind": "recovery_continue", "payload": {}, "base_version": s.version} + advance_mod.advance_session(s.id, action, {"account_id": actor.account_id, "tenant_id": actor.tenant_id}, token) + + assert not lock.exists(s.id) # lock released -- a stuck lock would wedge the session + assert ("error", "step failed") in [(ev["kind"], ev.get("error")) for _sid, ev in events] + + view = service.get_session_view(s.id, actor) + assert view.interrupted is True + assert view.recovery is not None # offer resurfaces -- no dead end + assert view.recovery.can_continue is True + assert view.recovery.can_restart is True