Fix: Preserve complete History edit cascades

Editing a complete History session could bypass the global in-place recompute
planner, while multi-output or legacy actions could lose downstream dependencies
after runtime mappings were pruned. All edit selections now use one deduplicated
plan with consistent output UUID resolution.

* [FIX] : Expand selected sessions into actions and process overlapping
  selections once through the global edit planner
* [FIX] : Resolve dependency closure and failed-branch blocking from durable,
  runtime, then legacy output UUIDs
* [CHG] : Cover Qt tree selection, pruned multi-output mappings, and legacy
  failed branches on Python 3.9 and 3.14

Author: dappham-CODRA

datalab/gui/panel/history/chain.py

@@ -219,6 +219,19 @@ def action_output_uuid(panel: HistoryPanel, action: HistoryAction) -> str | None
     return find_output_object_uuid(panel, panel_data, action)
 
 
+def recorded_action_output_uuids(
+    panel: HistoryPanel, action: HistoryAction
+) -> list[str]:
+    """Return output UUIDs recorded for ``action``, preferring durable history."""
+    if action.output_uuids:
+        return list(action.output_uuids)
+    runtime_outputs = panel.runtime.objects.action_output_uuids.get(action.uuid, [])
+    if runtime_outputs:
+        return list(runtime_outputs)
+    legacy_output = action_output_uuid(panel, action)
+    return [legacy_output] if legacy_output is not None else []
+
+
 def action_consumes_any(action: HistoryAction, uuids: set[str]) -> bool:
     """Return True if ``action``'s input UUIDs intersect ``uuids``."""
     if action.kind != HistoryAction.KIND_COMPUTE:
@@ -235,10 +248,10 @@ def get_downstream_actions(
     current = get_session_of(panel, action)
     if current is None:
         return []
-    root_out = action_output_uuid(panel, action)
-    if root_out is None:
+    root_outputs = recorded_action_output_uuids(panel, action)
+    if not root_outputs:
         return []
-    closure: set[str] = {root_out}
+    closure: set[str] = set(root_outputs)
     downstream: list[HistoryAction] = []
     idx = current.actions.index(action)
     for candidate in current.actions[idx + 1 :]:
@@ -247,9 +260,7 @@ def get_downstream_actions(
         if not action_consumes_any(candidate, closure):
             continue
         downstream.append(candidate)
-        out_uuid = action_output_uuid(panel, candidate)
-        if out_uuid is not None:
-            closure.add(out_uuid)
+        closure.update(recorded_action_output_uuids(panel, candidate))
     return downstream
 
 

datalab/gui/panel/history/interactive_replay.py

@@ -43,9 +43,15 @@ def replay_restore_actions(
         if not panel.history_sessions:
             return
         selected = [panel.history_sessions[-1]]
+    edit_mode = panel.runtime.execution.edit_mode
     edit_actions: list[HistoryAction] = []
     for session_or_action in selected:
-        if isinstance(session_or_action, HistoryAction) and session_or_action.is_stale:
+        if (
+            replay
+            and not edit_mode
+            and isinstance(session_or_action, HistoryAction)
+            and session_or_action.is_stale
+        ):
             hrec.recompute_cascade(panel, session_or_action)
             continue
         if not session_or_action.is_current_state_compatible(
@@ -59,22 +65,20 @@ def replay_restore_actions(
                 )
             return
         if replay:
-            if panel.runtime.execution.edit_mode and isinstance(
-                session_or_action, HistoryAction
-            ):
-                # Defer: edit only the selected actions, no automatic cascade
-                edit_actions.append(session_or_action)
+            if edit_mode:
+                if isinstance(session_or_action, HistorySession):
+                    edit_actions.extend(session_or_action.actions)
+                else:
+                    edit_actions.append(session_or_action)
             else:
-                # Scope decision: clicking a session in edit mode now replays it
-                # WITH parameter dialogs (view-only session replay disabled).
                 with panel.replaying(), panel.output_suppressed():
                     session_or_action.replay(
                         panel.mainwindow,
                         restore_selection=restore_selection,
-                        edit=panel.runtime.execution.edit_mode,
+                        edit=edit_mode,
                     )
         elif restore_selection:
-            if panel.runtime.execution.edit_mode or any(
+            if edit_mode or any(
                 action.has_pending_edits
                 for session in panel.history_sessions
                 for action in session.actions
@@ -196,15 +200,15 @@ def edit_mode_replay_actions(panel: HistoryPanel, actions: list[HistoryAction])
                     continue
                 if hchain.action_consumes_any(action, blocked_outputs):
                     blocked_outputs.update(
-                        panel.runtime.objects.action_output_uuids.get(action.uuid, [])
+                        hchain.recorded_action_output_uuids(panel, action)
                     )
                     continue
                 success = hrec.recompute_action_in_place(panel, action)
                 action.is_stale = not success
                 panel.tree.refresh_action_item(action)
                 if not success:
                     blocked_outputs.update(
-                        panel.runtime.objects.action_output_uuids.get(action.uuid, [])
+                        hchain.recorded_action_output_uuids(panel, action)
                     )
         finally:
             hrec.flush_cascade_warnings(panel)

datalab/tests/features/common/history_workflow_test.py

@@ -17,6 +17,7 @@
 from datalab.adapters_metadata.common import ResultData
 from datalab.gui import historytools_ops as htools
 from datalab.gui.panel.history import HistoryAction
+from datalab.gui.panel.history import chain as hchain
 from datalab.gui.panel.history import interactive_replay as hireplay
 from datalab.gui.panel.history import recompute as hrec
 from datalab.gui.panel.history.chainmodel import (
@@ -36,6 +37,7 @@
 from datalab.tests.features.common.history_test_helpers import (
     add_paracetamol_signals,
     build_signal_chain,
+    get_tree_item,
     read_history_sessions,
     select_tree_entry,
     select_tree_session,
@@ -493,6 +495,87 @@ def test_multi_action_edit_recomputes_selected_descendants_once() -> None:
         assert all(action.is_stale is False for action in expected)
 
 
+def test_edit_mode_selected_session_uses_global_replay_planner() -> None:
+    """Plan a selected session and duplicate stale action exactly once."""
+    with datalab_test_app_context(history=True) as win:
+        history, panel = win.historypanel, win.signalpanel
+        history.toggle_record_mode(True)
+        history.toggle_edit_mode(True)
+        build_signal_chain(panel, history)
+        session = history.history_sessions[-1]
+        expected = list(session.actions)
+        stale_action = expected[1]
+        stale_action.is_stale = True
+        select_tree_session(history, session)
+        get_tree_item(history, stale_action.uuid).setSelected(True)
+        selected = history.tree.get_selected_actions_or_sessions(
+            history.history_sessions
+        )
+        assert selected == [session, stale_action]
+
+        with (
+            patch.object(type(session), "replay") as direct_replay,
+            patch.object(hrec, "recompute_cascade") as direct_cascade,
+            patch.object(
+                hireplay,
+                "edit_mode_replay_actions",
+                wraps=hireplay.edit_mode_replay_actions,
+            ) as edit_planner,
+            patch.object(
+                hireplay, "prompt_edit_action_params", return_value=True
+            ) as prompt,
+            patch.object(
+                hrec, "recompute_action_in_place", return_value=True
+            ) as recompute,
+        ):
+            hireplay.replay_restore_actions(history)
+
+        direct_replay.assert_not_called()
+        direct_cascade.assert_not_called()
+        edit_planner.assert_called_once_with(history, [*expected, stale_action])
+        assert [call.args[1] for call in prompt.call_args_list] == expected
+        assert [call.args[1] for call in recompute.call_args_list] == expected
+        assert all(action.is_stale is False for action in expected)
+
+
+def test_downstream_actions_follow_every_registered_output() -> None:
+    """Follow second registered outputs through transitive dependencies."""
+    with datalab_test_app_context(history=True) as win:
+        history, panel = win.historypanel, win.signalpanel
+        history.toggle_record_mode(True)
+        producer, consumer, descendant = build_signal_chain(panel, history).actions
+        producer_second_output = "producer-second-output"
+        consumer_second_output = "consumer-second-output"
+        producer.output_uuids.append(producer_second_output)
+        consumer.output_uuids.append(consumer_second_output)
+        history.runtime.objects.action_output_uuids[producer.uuid] = list(
+            producer.output_uuids
+        )
+        history.runtime.objects.action_output_uuids[consumer.uuid] = list(
+            consumer.output_uuids
+        )
+        history.runtime.objects.output_to_action[producer_second_output] = producer.uuid
+        history.runtime.objects.output_to_action[consumer_second_output] = consumer.uuid
+        hchain.prune_output_mapping(history)
+        assert producer_second_output in producer.output_uuids
+        assert consumer_second_output in consumer.output_uuids
+        assert (
+            producer_second_output
+            not in (history.runtime.objects.action_output_uuids[producer.uuid])
+        )
+        assert (
+            consumer_second_output
+            not in (history.runtime.objects.action_output_uuids[consumer.uuid])
+        )
+        consumer.state.selection["signal"] = [producer_second_output]
+        descendant.state.selection["signal"] = [consumer_second_output]
+
+        assert hchain.get_downstream_actions(history, producer) == [
+            consumer,
+            descendant,
+        ]
+
+
 def test_multi_action_edit_cascades_across_independent_sessions() -> None:
     """Recompute edited branches from multiple sessions in global order."""
     with datalab_test_app_context(history=True) as win:
@@ -527,6 +610,20 @@ def test_multi_action_edit_failure_skips_dependents_and_continues() -> None:
         history.create_new_session(panel_str="signal")
         successful_chain = build_independent_signal_branch(panel, history)
         failed_root = failed_chain[0]
+        failed_output_uuid = failed_root.output_uuids[0]
+        failed_root.output_uuids.clear()
+        history.runtime.objects.action_output_uuids.pop(failed_root.uuid)
+        history.runtime.objects.output_to_action.pop(failed_output_uuid)
+        failed_output = panel.objmodel[failed_output_uuid]
+        processing_parameters = extract_processing_parameters(failed_output)
+        assert not failed_root.output_uuids
+        assert failed_root.uuid not in history.runtime.objects.action_output_uuids
+        assert failed_output_uuid not in history.runtime.objects.output_to_action
+        assert processing_parameters is not None
+        assert processing_parameters.func_name == failed_root.func_name
+        assert hchain.recorded_action_output_uuids(history, failed_root) == [
+            failed_output_uuid
+        ]
         recomputed: list[HistoryAction] = []
 
         def recompute_action(_panel, action):