From 478e8bf1fedee7183e3e9b161030eacf50e2ebaf Mon Sep 17 00:00:00 2001 From: Charles Yao Date: Sun, 23 Aug 2026 10:42:53 -0700 Subject: [PATCH] fix(copilot): drive recovery_restart through working FIX/FIX_CHECKLIST entry states The runner's check_recovery/recovery_continue/recovery_restart short-circuit committed the transition and returned immediately, without entering the advance loop. recovery_restart in FIX/FIX_CHECKLIST mode (the forced path for structural_invalidating drift) lands on the *working* fix.diagnose / checklist.diagnose entry state, which the runner must drive by running its handler -- the early return stranded the session there with no UI actions. Fix: rest only when the landed state is waiting/terminal (covers stop/resume, check_recovery, recovery_continue, and BUILD/EDIT restarts unchanged); otherwise consume the action and fall through to the existing advance loop. Also: parity-test recovery.entry_state_for against the states service.create_*_session actually assigns; add classify()-through-runner coverage for the three remaining recovery classes and target_node_ids guard branches; pin recovery_ref_for's message for config_only/structural_compatible; and qualify dify_port.py's graph_ops delegations (module import) so structural_fingerprint/node_ids no longer read as unqualified self-recursion. Co-Authored-By: Claude Opus 4.8 --- api/core/workflow_copilot/runner.py | 12 +- api/services/workflow_copilot/dify_port.py | 34 ++-- .../core/workflow_copilot/test_recovery.py | 172 +++++++++++++++++- .../services/workflow_copilot/test_service.py | 13 +- 4 files changed, 205 insertions(+), 26 deletions(-) diff --git a/api/core/workflow_copilot/runner.py b/api/core/workflow_copilot/runner.py index 51fc51d34fa..19246dff6a5 100644 --- a/api/core/workflow_copilot/runner.py +++ b/api/core/workflow_copilot/runner.py @@ -105,7 +105,17 @@ class Runner: next_state, items = recovery.apply_recovery_action(self._env.dify, turn, s, fc) s.version = self._env.repo.compare_and_advance(s.id, s.version, next_state, fc, items) s.current_state = next_state - return s + # check_recovery / recovery_continue stay at the current waiting state, + # and recovery_restart into BUILD/EDIT lands on the waiting + # capability_check — all rest here. recovery_restart into FIX / + # FIX_CHECKLIST lands on a *working* entry state (fix.diagnose / + # checklist.diagnose) that the runner must drive, exactly like any + # transition into a working state — fall through to the advance loop + # with the recovery action consumed. + if is_waiting(s.current_state) or is_terminal(s.current_state): + return s + turn = Turn(actor=turn.actor) # action consumed + # fall through to the advance loop below first = True while True: diff --git a/api/services/workflow_copilot/dify_port.py b/api/services/workflow_copilot/dify_port.py index f7f2bef8166..7ddb4dd69bc 100644 --- a/api/services/workflow_copilot/dify_port.py +++ b/api/services/workflow_copilot/dify_port.py @@ -60,18 +60,8 @@ from models.workflow import Workflow from repositories.factory import DifyAPIRepositoryFactory from services.app_generate_service import AppGenerateService from services.errors.app import WorkflowHashNotEqualError +from services.workflow_copilot import graph_ops from services.workflow_copilot.errors import HashMismatchError, WorkflowNotInitializedError -from services.workflow_copilot.graph_ops import ( - apply_connect, - apply_create_node, - apply_delete_node, - apply_insert_between, - apply_set_node_config, - diff_graphs, - node_ids, - structural_fingerprint, - validate_intent_args, -) from services.workflow_copilot.identity import load_app, resolve_account from services.workflow_copilot.run_mapping import map_run_result, to_node_event from services.workflow_service import WorkflowService @@ -80,11 +70,11 @@ __all__ = ["WorkflowServiceDifyPort"] _APPLY_FNS: dict[str, Callable[..., tuple[Graph, list[str]]]] = { - "set_node_config": apply_set_node_config, - "create_node": apply_create_node, - "delete_node": apply_delete_node, - "connect": apply_connect, - "insert_between": apply_insert_between, + "set_node_config": graph_ops.apply_set_node_config, + "create_node": graph_ops.apply_create_node, + "delete_node": graph_ops.apply_delete_node, + "connect": graph_ops.apply_connect, + "insert_between": graph_ops.apply_insert_between, } _CREATE_NODE_CANVAS_EVENTS: dict[str, CanvasEvent] = { @@ -193,7 +183,7 @@ class WorkflowServiceDifyPort: apply_fn = _APPLY_FNS.get(intent.op) if apply_fn is None: continue - validate_intent_args(intent) + graph_ops.validate_intent_args(intent) graph, changed = apply_fn(graph, **intent.args) changed_nodes.extend(changed) if on_canvas is not None: @@ -205,10 +195,10 @@ class WorkflowServiceDifyPort: new_hash=unique_hash, changes=[], scope="", - structure_fingerprint=structural_fingerprint(before_graph), + structure_fingerprint=graph_ops.structural_fingerprint(before_graph), ) - changes, scope = diff_graphs(before_graph, graph) + changes, scope = graph_ops.diff_graphs(before_graph, graph) try: updated = WorkflowService().sync_draft_workflow( @@ -233,14 +223,14 @@ class WorkflowServiceDifyPort: new_hash=updated.unique_hash, changes=changes, scope=scope, - structure_fingerprint=structural_fingerprint(graph), + structure_fingerprint=graph_ops.structural_fingerprint(graph), ) def structural_fingerprint(self, graph: Graph) -> str: - return structural_fingerprint(graph) + return graph_ops.structural_fingerprint(graph) def graph_node_ids(self, graph: Graph) -> list[str]: - return node_ids(graph) + return graph_ops.node_ids(graph) def restore_graph(self, app_id: str, actor: Actor, graph: Graph) -> str: """Restore the draft to a prior snapshot graph (Slice 4 revert): write diff --git a/api/tests/unit_tests/core/workflow_copilot/test_recovery.py b/api/tests/unit_tests/core/workflow_copilot/test_recovery.py index 11a1cf3e82e..8e9c61872f9 100644 --- a/api/tests/unit_tests/core/workflow_copilot/test_recovery.py +++ b/api/tests/unit_tests/core/workflow_copilot/test_recovery.py @@ -1,11 +1,21 @@ from datetime import datetime from core.workflow_copilot.contract import RecoveryClass +from core.workflow_copilot.handlers_fix import fix_registry from core.workflow_copilot.models import ( - Action, ChecklistError, ConversationItem, CopilotContext, Diagnosis, EntryMode, Session, Turn, + Action, + ChecklistError, + ConversationItem, + CopilotContext, + Diagnosis, + EntryMode, + Run, + Session, + Turn, ) +from core.workflow_copilot.placeholder_agent import PlaceholderAgent from core.workflow_copilot.runner import Env, Runner -from core.workflow_copilot.state import PcState +from core.workflow_copilot.state import PcState, is_waiting, is_working from tests.unit_tests.core.workflow_copilot.fakes import FakeDifyPort, InMemoryRepository, StubAgent @@ -17,6 +27,20 @@ def _actor(): # ---- entry_state_for ------------------------------------------------------- def test_entry_state_for_all_modes(): + """Parity test for entry_state_for's docstring claim ("mirrors + service.create_*_session; a parity test guards drift"). The expected + mapping below is exactly what services.workflow_copilot.service assigns + as the initial ``Session.current_state`` for each mode: + - create_fix_session: EntryMode.FIX -> PcState.FIX_DIAGNOSE, + EntryMode.FIX_CHECKLIST -> PcState.CHECKLIST_DIAGNOSE (branches on + whether checklist_errors were passed). + - create_build_session: EntryMode.BUILD -> PcState.BUILD_CAPABILITY_CHECK. + - create_edit_session: EntryMode.EDIT -> PcState.EDIT_CAPABILITY_CHECK. + We assert the known mapping directly (no need to call the service + functions) so this stays a light core-layer test; if either side drifts + a fresh session would start somewhere entry_state_for doesn't expect, + which is exactly the drift this test guards against. + """ from core.workflow_copilot import recovery assert recovery.entry_state_for(EntryMode.FIX) == PcState.FIX_DIAGNOSE @@ -47,6 +71,17 @@ def test_target_node_ids_per_mode(): assert recovery.target_node_ids(checklist_fc, EntryMode.FIX_CHECKLIST) == ["k1", "k2"] +def test_target_node_ids_fix_guard_branches(): + from core.workflow_copilot import recovery + + # no diagnosis yet (e.g. recovery checked before diagnose ever ran). + assert recovery.target_node_ids(CopilotContext(), EntryMode.FIX) == [] + + # a diagnosis with no resolved culprit node. + fc = CopilotContext(diagnosis=Diagnosis(culprit_node_id="")) + assert recovery.target_node_ids(fc, EntryMode.FIX) == [] + + # ---- classify -------------------------------------------------------------- def test_classify_unchanged(): @@ -115,6 +150,59 @@ def test_check_recovery_sets_class_and_appends_notice(): assert any(item.kind == "notice" for item in convo[1:]) # a notice was appended +def test_check_recovery_classifies_unchanged_via_runner(): + """The remaining classify() classes, exercised end-to-end through the + runner's check_recovery path (test_check_recovery_sets_class_and_appends_notice + above already covers CONFIG_ONLY).""" + dify = FakeDifyPort() # graph {"nodes": []}, hash "h0" (untouched) + repo = InMemoryRepository() + # last-known hash matches the current one -> unchanged, regardless of fingerprint. + fc = CopilotContext(last_snapshot_hash=dify.hash, last_structure_fingerprint="stale-fp", next_seq=1) + s = _seed(repo, EntryMode.BUILD, PcState.BUILD_REVIEW, fc) + runner = Runner(Env(dify=dify, agent=StubAgent(), repo=repo, now=lambda: datetime.min), {}) + + runner.advance(s.id, Turn(action=Action(kind="check_recovery", base_version=1), actor=_actor())) + + _s, fc2 = repo.get_session(s.id) + assert fc2.recovery_class == "unchanged" + + +def test_check_recovery_classifies_structural_compatible_via_runner(): + dify = FakeDifyPort() + dify.graph = {"nodes": [{"id": "n1", "data": {"type": "llm"}}, {"id": "n2", "data": {"type": "llm"}}], "edges": []} + dify.hash = "h-new" + repo = InMemoryRepository() + # hash + fingerprint both changed, but the build target n1 is still present. + fc = CopilotContext( + last_snapshot_hash="h-old", last_structure_fingerprint="stale-fp", built_node_ids=["n1"], next_seq=1, + ) + s = _seed(repo, EntryMode.BUILD, PcState.BUILD_REVIEW, fc) + runner = Runner(Env(dify=dify, agent=StubAgent(), repo=repo, now=lambda: datetime.min), {}) + + runner.advance(s.id, Turn(action=Action(kind="check_recovery", base_version=1), actor=_actor())) + + _s, fc2 = repo.get_session(s.id) + assert fc2.recovery_class == "structural_compatible" + + +def test_check_recovery_classifies_structural_invalidating_via_runner(): + dify = FakeDifyPort() + dify.graph = {"nodes": [{"id": "n2", "data": {"type": "llm"}}], "edges": []} + dify.hash = "h-new" + repo = InMemoryRepository() + # hash + fingerprint both changed, and the build target n1 is gone from the graph. + fc = CopilotContext( + last_snapshot_hash="h-old", last_structure_fingerprint="stale-fp", built_node_ids=["n1"], next_seq=1, + ) + s = _seed(repo, EntryMode.BUILD, PcState.BUILD_REVIEW, fc) + runner = Runner(Env(dify=dify, agent=StubAgent(), repo=repo, now=lambda: datetime.min), {}) + + runner.advance(s.id, Turn(action=Action(kind="check_recovery", base_version=1), actor=_actor())) + + _s, fc2 = repo.get_session(s.id) + assert fc2.recovery_class == "structural_invalidating" + + def test_recovery_continue_clears_class(): dify = FakeDifyPort() repo = InMemoryRepository() @@ -153,3 +241,83 @@ def test_recovery_restart_resets_to_entry_state_and_preserves_conversation(): assert fc2.checkpoint_id == "" assert fc2.goal_text == "build me a thing" # preserved assert len(repo.list_conversation(s.id)) >= 1 # conversation preserved (+ restart notice) + + +# ---- recovery_restart into a WORKING entry state must be driven, not stranded (FINDING 1) -- + + +def _engine_env() -> tuple[Env, InMemoryRepository, FakeDifyPort]: + """A real engine env (PlaceholderAgent + FakeDifyPort + InMemoryRepository), + mirroring test_fix_flow.py's _new_env -- needed here because, unlike the + BUILD/EDIT recovery_restart tests above (registry={}), a FIX/FIX_CHECKLIST + restart lands on a WORKING state (fix.diagnose / checklist.diagnose) that + the runner must actually drive through real handlers.""" + repo = InMemoryRepository() + dify = FakeDifyPort() + env = Env(dify=dify, agent=PlaceholderAgent(), repo=repo, now=lambda: datetime.min) + return env, repo, dify + + +def test_recovery_restart_in_fix_mode_drives_working_entry_state_to_waiting(): + """FINDING 1 regression: before the runner fix, the check_recovery / + recovery_continue / recovery_restart short-circuit committed the + transition into fix.diagnose (a WORKING state) and then returned + immediately, without entering the advance loop that drives working + states through their handlers. The session was left resting at + fix.diagnose with no UI actions -- get_session_view reports + interrupted=True and nothing ever re-drives it. This is the FORCED path + for structural_invalidating drift (can_continue=False), so it must not + strand the session. Asserts the post-fix behavior: recovery_restart + drives fix.diagnose -> fix.propose -> fix.apply (PlaceholderAgent's + canned repair is low-risk, so no approval gate) and rests at the next + genuine waiting state, exactly like any other transition into a working + state (e.g. request_fix in test_fix_flow.py). This test fails against + the old runner (out.current_state == PcState.FIX_DIAGNOSE, a working + state) and passes after the fix. + """ + env, repo, dify = _engine_env() + fc = CopilotContext(failed_run_id="TR-1", recovery_class="structural_invalidating", next_seq=1) + s = _seed(repo, EntryMode.FIX, PcState.FIX_AWAIT_DECISION, fc) + repo.save_run(s.id, Run(id="TR-1", kind="original-failed", status="failed", immutable=True)) + + runner = Runner(env, fix_registry()) + out = runner.advance(s.id, Turn(action=Action(kind="recovery_restart", base_version=1), actor=_actor())) + + # not stranded at the working entry state ... + assert out.current_state != PcState.FIX_DIAGNOSE + # ... it was driven forward to a genuine waiting state with actions. + assert is_waiting(out.current_state) + assert not is_working(out.current_state) + + _s, fc2 = repo.get_session(s.id) + assert fc2.recovery_class == "" + assert fc2.diagnosis is not None, "fix.diagnose's handler must have actually run" + assert dify.applied, "the drive must have reached fix.apply and applied the repair" + + +def test_recovery_restart_in_fix_checklist_mode_drives_working_entry_state_to_waiting(): + """FINDING 1 regression, checklist entry mode: recovery_restart must not + strand the session at checklist.diagnose (also a WORKING state). Mirrors + the FIX test above; the checklist flow instead rests at + checklist.await_recheck once handle_apply routes on fc.source == + 'checklist' (see test_fix_flow.py's checklist-entry smoke test).""" + env, repo, dify = _engine_env() + errors = [ChecklistError(node_id="n1", node_type="llm", title="LLM", messages=["missing prompt"])] + fc = CopilotContext( + source="checklist", checklist_errors=errors, recovery_class="structural_invalidating", next_seq=1, + ) + s = _seed(repo, EntryMode.FIX_CHECKLIST, PcState.CHECKLIST_AWAIT_RECHECK, fc) + + runner = Runner(env, fix_registry()) + out = runner.advance(s.id, Turn(action=Action(kind="recovery_restart", base_version=1), actor=_actor())) + + # not stranded at the working entry state ... + assert out.current_state != PcState.CHECKLIST_DIAGNOSE + # ... it was driven forward to a genuine waiting state with actions. + assert is_waiting(out.current_state) + assert not is_working(out.current_state) + + _s, fc2 = repo.get_session(s.id) + assert fc2.recovery_class == "" + assert fc2.diagnosis is not None, "checklist.diagnose's handler must have actually run" + assert dify.applied, "the drive must have reached fix.apply and applied the repair" diff --git a/api/tests/unit_tests/services/workflow_copilot/test_service.py b/api/tests/unit_tests/services/workflow_copilot/test_service.py index 3ebd5557cff..bda89871d8d 100644 --- a/api/tests/unit_tests/services/workflow_copilot/test_service.py +++ b/api/tests/unit_tests/services/workflow_copilot/test_service.py @@ -649,8 +649,8 @@ def test_get_session_view_run_status_paused( def test_recovery_ref_for(): # A session whose fc has a recovery_class surfaces a RecoveryRef; # can_continue/can_restart follow the class. - from core.workflow_copilot.contract import RecoveryClass from core.workflow_copilot import recovery + from core.workflow_copilot.contract import RecoveryClass ref = recovery.recovery_ref_for(str(RecoveryClass.STRUCTURAL_INVALIDATING)) assert ref is not None @@ -665,6 +665,17 @@ def test_recovery_ref_for(): assert recovery.recovery_ref_for("") is None + # the two "keep going either way" classes: both continuing and restarting + # are offered, and .message comes from the dict-lookup path (recovery. + # _RECOVERY_MESSAGE), not the RecoveryRef.get(..., "") fallback. + for cls in (RecoveryClass.CONFIG_ONLY, RecoveryClass.STRUCTURAL_COMPATIBLE): + ref3 = recovery.recovery_ref_for(str(cls)) + assert ref3 is not None + assert ref3.can_continue is True + assert ref3.can_restart is True + assert ref3.message + assert ref3.message == recovery._RECOVERY_MESSAGE[cls] + def test_get_session_view_surfaces_recovery_when_set( service: WorkflowCopilotService, repo: SqlCopilotRepository