feat: allow graceful stop for all run statuses.

apps/gust/lib/gust/dag/runner/dag_worker.ex

@@ -121,6 +121,18 @@ defmodule Gust.DAG.Runner.DAGWorker do
     end
   end
 
+  @impl true
+  def handle_call(:stop, _from, %State{} = state) do
+    case stop_active_tasks(state) do
+      :ok ->
+        teardown(state.dag_def, state.runtime_id)
+        {:stop, :normal, {:ok, state.run}, state}
+
+      {:error, reason} ->
+        {:reply, {:error, reason}, state}
+    end
+  end
+
   @impl true
   def handle_info(
         {:task_result, result, task_id, status},
@@ -331,6 +343,27 @@ defmodule Gust.DAG.Runner.DAGWorker do
     :ok
   end
 
+  defp stop_active_tasks(%State{} = state) do
+    Enum.reduce_while(state.current_task_ids, :ok, fn task_id, :ok ->
+      task_id
+      |> Flows.get_task()
+      |> stop_task(state)
+    end)
+  end
+
+  defp stop_task(nil, _state), do: {:cont, :ok}
+
+  defp stop_task(%Flows.Task{} = task, state) do
+    case maybe_cancel_execution(task, state) do
+      :ok -> {:cont, :ok}
+      {:error, reason} -> {:halt, {:error, reason}}
+    end
+  end
+
+  defp maybe_cancel_execution(%Flows.Task{status: status} = task, state) do
+    if TaskStatus.cancellable?(status), do: cancel_execution(task, state), else: :ok
+  end
+
   defp update_run_status(run, status) do
     {:ok, %Flows.Run{id: id, status: run_status} = run} =
       Flows.update_run_status(run, status)

apps/gust/lib/gust/dag/runner/run_gateway/default.ex

@@ -19,6 +19,7 @@ defmodule Gust.DAG.Runner.RunGateway.Default do
   @doc false
   def local_call(run_id, command) do
     case Registry.lookup(Gust.Registry, registry_key(run_id)) do
+      [{pid, _value}] when command == :stop -> safe_stop(pid)
       [{pid, _value}] -> safe_call(pid, command)
       [] -> {:error, :run_not_active}
     end
@@ -47,7 +48,35 @@ defmodule Gust.DAG.Runner.RunGateway.Default do
   end
 
   defp safe_call(pid, command) do
-    GenServer.call(pid, command, call_timeout())
+    call_safely(fn -> GenServer.call(pid, command, call_timeout()) end)
+  end
+
+  defp safe_stop(pid) do
+    monitor = Process.monitor(pid)
+
+    try do
+      call_safely(fn -> stop_and_wait(pid, monitor) end)
+    after
+      Process.demonitor(monitor, [:flush])
+    end
+  end
+
+  defp stop_and_wait(pid, monitor) do
+    case GenServer.call(pid, :stop, call_timeout()) do
+      {:ok, _value} = result ->
+        receive do
+          {:DOWN, ^monitor, :process, ^pid, _reason} -> result
+        after
+          call_timeout() -> {:error, :run_command_timeout}
+        end
+
+      {:error, _reason} = error ->
+        error
+    end
+  end
+
+  defp call_safely(callback) do
+    callback.()
   catch
     :exit, {:normal, _call} -> {:error, :run_not_active}
     :exit, {:shutdown, _call} -> {:error, :run_not_active}

apps/gust/lib/gust/dag/terminator.ex

@@ -3,11 +3,14 @@ defmodule Gust.DAG.Terminator do
 
   alias Gust.Flows
 
-  @type result :: {:ok, Flows.Task.t()} | {:error, term()}
+  @type cancel_result :: {:ok, Flows.Task.t()} | {:error, term()}
+  @type stop_run_result :: {:ok, Flows.Run.t()} | {:error, term()}
 
-  @callback cancel(task :: Flows.Task.t()) :: result()
+  @callback cancel(task :: Flows.Task.t()) :: cancel_result()
+  @callback stop_run(run :: Flows.Run.t()) :: stop_run_result()
 
   def cancel(%Flows.Task{} = task), do: impl().cancel(task)
+  def stop_run(%Flows.Run{} = run), do: impl().stop_run(run)
 
   def impl, do: Application.get_env(:gust, :dag_terminator, Gust.DAG.Terminator.Gateway)
 end

apps/gust/lib/gust/dag/terminator/gateway.ex

@@ -17,4 +17,12 @@ defmodule Gust.DAG.Terminator.Gateway do
       result -> result
     end
   end
+
+  @impl true
+  def stop_run(%Flows.Run{} = run) do
+    case RunGateway.call(run, :stop) do
+      {:error, :run_not_active} -> {:ok, run}
+      result -> result
+    end
+  end
 end

apps/gust/test/dag/runner/dag_worker_test.exs

@@ -226,6 +226,52 @@ defmodule Gust.DAG.Runner.DAGWorkerTest do
     assert Flows.get_run!(run.id).status == :failed
   end
 
+  test "stops active task workers, ignores missing tasks, and tears down", %{run: run} do
+    dag_def = definition([["running"]])
+    expect_task_starts(1)
+
+    Gust.RuntimeAdapterMock
+    |> expect(:kill, fn task_pid ->
+      Process.exit(task_pid, :kill)
+      :ok
+    end)
+    |> expect(:teardown, fn ^dag_def, _runtime_id -> :ok end)
+
+    runner = start_runner(run, dag_def)
+    runner_ref = Process.monitor(runner)
+
+    assert_receive {:task_started, %Flows.Task{name: "running"} = task, ^runner}
+    [{task_pid, _value}] = Registry.lookup(Gust.Registry, TaskWorker.registry_name(task))
+    task_ref = Process.monitor(task_pid)
+
+    :sys.replace_state(runner, fn state ->
+      %{state | current_task_ids: MapSet.put(state.current_task_ids, -1)}
+    end)
+
+    assert is_nil(Flows.get_task(-1))
+
+    assert {:ok, %Flows.Run{id: run_id}} = RunGateway.call(run, :stop)
+    assert run_id == run.id
+    assert_receive {:DOWN, ^task_ref, :process, ^task_pid, :killed}
+    assert_receive {:DOWN, ^runner_ref, :process, ^runner, :normal}
+  end
+
+  test "returns the task cancellation error and keeps the run alive", %{run: run} do
+    dag_def = definition([["running"]])
+    expect_task_starts(1)
+
+    Gust.RuntimeAdapterMock
+    |> expect(:kill, fn _task_pid -> {:error, :cannot_kill_task} end)
+
+    runner = start_runner(run, dag_def)
+    runner_ref = Process.monitor(runner)
+
+    assert_receive {:task_started, %Flows.Task{name: "running"}, ^runner}
+    assert {:error, :cannot_kill_task} = RunGateway.call(run, :stop)
+    assert Process.alive?(runner)
+    refute_receive {:DOWN, ^runner_ref, :process, ^runner, _reason}
+  end
+
   test "persists waiting state and pauses the run", %{run: run} do
     dag_def =
       definition(

apps/gust/test/dag/runner/run_gateway/default_test.exs

@@ -53,6 +53,46 @@ defmodule Gust.DAG.Runner.RunGateway.DefaultTest do
     assert {:error, :run_not_active} = Default.call(run, :stop)
   end
 
+  test "returns a stop error without waiting for the worker to exit" do
+    run = run()
+
+    worker =
+      spawn_registered(run.id, fn
+        {:"$gen_call", from, :stop} ->
+          GenServer.reply(from, {:error, :cannot_stop})
+
+          receive do
+            :stop -> :ok
+          end
+      end)
+
+    assert {:error, :cannot_stop} = Default.call(run, :stop)
+    assert Process.alive?(worker)
+    send(worker, :stop)
+  end
+
+  test "times out when the worker acknowledges stop but remains alive" do
+    run = run()
+
+    worker =
+      spawn_registered(run.id, fn
+        {:"$gen_call", from, :stop} ->
+          GenServer.reply(from, {:ok, :stopping})
+
+          receive do
+            :stop -> :ok
+          end
+      end)
+
+    on_exit(fn ->
+      if Process.alive?(worker), do: Process.exit(worker, :kill)
+    end)
+
+    assert {:error, :run_command_timeout} = Default.call(run, :stop)
+    assert Process.alive?(worker)
+    send(worker, :stop)
+  end
+
   test "returns run_command_failed when the worker crashes during the call" do
     run = run()
 

apps/gust/test/dag/terminator/gateway_test.exs

@@ -61,4 +61,18 @@ defmodule Gust.DAG.Terminator.GatewayTest do
     assert {:error, :run_not_active} = Gateway.cancel(task)
     assert Flows.get_task!(task.id).status == :running
   end
+
+  test "routes run stops through the run gateway", %{run: run} do
+    Gust.RunGatewayMock
+    |> expect(:call, fn ^run, :stop -> {:ok, run} end)
+
+    assert {:ok, ^run} = Gateway.stop_run(run)
+  end
+
+  test "treats a run without an active process as stopped", %{run: run} do
+    Gust.RunGatewayMock
+    |> expect(:call, fn ^run, :stop -> {:error, :run_not_active} end)
+
+    assert {:ok, ^run} = Gateway.stop_run(run)
+  end
 end

apps/gust/test/dag/terminator_test.exs

@@ -16,14 +16,21 @@ defmodule Gust.DAG.TerminatorTest do
     run = run_fixture(%{dag_id: dag.id})
     task = task_fixture(%{run_id: run.id, name: "task", status: :running})
 
-    %{task: task}
+    %{run: run, task: task}
   end
 
-  test "delegates to the configured implementation", %{task: task} do
+  test "delegates task cancellation to the configured implementation", %{task: task} do
     Gust.DAGTerminatorMock
     |> expect(:cancel, fn ^task -> {:error, :delegated} end)
 
     assert {:error, :delegated} = Terminator.cancel(task)
     assert Terminator.impl() == Gust.DAGTerminatorMock
   end
+
+  test "delegates run stops to the configured implementation", %{run: run} do
+    Gust.DAGTerminatorMock
+    |> expect(:stop_run, fn ^run -> {:error, :delegated} end)
+
+    assert {:error, :delegated} = Terminator.stop_run(run)
+  end
 end

apps/gust_web/lib/gust_web/live/run_live/index.ex

@@ -1,5 +1,6 @@
 defmodule GustWeb.RunLive.Index do
   alias Gust.DAG.Run.Trigger
+  alias Gust.DAG.Terminator
   alias Gust.Flows
   alias Gust.PubSub
   use GustWeb, :live_view
@@ -144,10 +145,10 @@ defmodule GustWeb.RunLive.Index do
   end
 
   def handle_event("batch_delete", _params, socket) do
-    {eligible_runs, skipped_runs} = socket |> selected_runs_on_dag() |> partition_batch_runs()
+    {stopped_runs, skipped_runs} = socket |> selected_runs_on_dag() |> partition_stoppable_runs()
 
     {:ok, deleted_runs} =
-      Flows.delete_runs_on_dag(socket.assigns.dag_id, Enum.map(eligible_runs, & &1.id))
+      Flows.delete_runs_on_dag(socket.assigns.dag_id, Enum.map(stopped_runs, & &1.id))
 
     {:noreply,
      socket
@@ -161,7 +162,9 @@ defmodule GustWeb.RunLive.Index do
   end
 
   def handle_event("batch_restart", _params, socket) do
-    {eligible_runs, skipped_runs} = socket |> selected_runs_on_dag() |> partition_batch_runs()
+    {eligible_runs, skipped_runs} =
+      socket |> selected_runs_on_dag() |> partition_restartable_runs()
+
     restarted_runs = Enum.map(eligible_runs, &Trigger.reset_run/1)
 
     {:noreply,
@@ -173,11 +176,21 @@ defmodule GustWeb.RunLive.Index do
   @impl true
   def handle_event("delete", %{"id" => id}, socket) do
     run = Flows.get_run!(id)
-    {:ok, _} = Flows.delete_run(run)
 
-    {:noreply,
-     socket
-     |> refresh_run_list(clear_selection?: false)}
+    case stop_run(run) do
+      :ok ->
+        {:ok, _run} = Flows.delete_run(run)
+
+        {:noreply, refresh_run_list(socket, clear_selection?: false)}
+
+      {:error, _reason} ->
+        {:noreply,
+         put_flash(
+           socket,
+           :error,
+           "Run could not be deleted because its process could not be stopped."
+         )}
+    end
   end
 
   @impl true
@@ -251,10 +264,21 @@ defmodule GustWeb.RunLive.Index do
     Flows.get_runs_on_dag(socket.assigns.dag_id, socket.assigns.selected_run_ids)
   end
 
-  defp partition_batch_runs(runs) do
+  defp partition_stoppable_runs(runs) do
+    Enum.split_with(runs, &(stop_run(&1) == :ok))
+  end
+
+  defp partition_restartable_runs(runs) do
     Enum.split_with(runs, &(&1.status in @completed_run_statuses))
   end
 
+  defp stop_run(run) do
+    case Terminator.stop_run(run) do
+      {:ok, _run} -> :ok
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
   defp batch_summary(action, processed_runs, []) do
     processed_runs |> length() |> processed_summary(action)
   end

apps/gust_web/lib/gust_web/live/run_live/index.html.heex

@@ -199,8 +199,9 @@
 
             <:action :let={{id, run}}>
               <.link
+                id={"delete-run-#{run.id}"}
                 phx-click={JS.push("delete", value: %{id: run.id}) |> hide("##{id}")}
-                data-confirm="Are you sure? Make sure it is completed, otherwise it may cause unexpected behavior."
+                data-confirm="Are you sure? An active run will be stopped before it is deleted."
                 class="btn btn-sm btn-soft btn-error"
               >
                 Delete