diff --git a/apps/gust/lib/gust/dag/runner/dag_worker.ex b/apps/gust/lib/gust/dag/runner/dag_worker.ex
index deef5a9..53b15b9 100644
--- a/apps/gust/lib/gust/dag/runner/dag_worker.ex
+++ b/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)
diff --git a/apps/gust/lib/gust/dag/runner/run_gateway/default.ex b/apps/gust/lib/gust/dag/runner/run_gateway/default.ex
index 9449185..8603403 100644
--- a/apps/gust/lib/gust/dag/runner/run_gateway/default.ex
+++ b/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}
diff --git a/apps/gust/lib/gust/dag/terminator.ex b/apps/gust/lib/gust/dag/terminator.ex
index c209512..6e563f4 100644
--- a/apps/gust/lib/gust/dag/terminator.ex
+++ b/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
diff --git a/apps/gust/lib/gust/dag/terminator/gateway.ex b/apps/gust/lib/gust/dag/terminator/gateway.ex
index 02cd1ae..8f5a7c3 100644
--- a/apps/gust/lib/gust/dag/terminator/gateway.ex
+++ b/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
diff --git a/apps/gust/test/dag/runner/dag_worker_test.exs b/apps/gust/test/dag/runner/dag_worker_test.exs
index 9f8b592..24036bc 100644
--- a/apps/gust/test/dag/runner/dag_worker_test.exs
+++ b/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(
diff --git a/apps/gust/test/dag/runner/run_gateway/default_test.exs b/apps/gust/test/dag/runner/run_gateway/default_test.exs
index 26103a2..5ab2c28 100644
--- a/apps/gust/test/dag/runner/run_gateway/default_test.exs
+++ b/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()
 
diff --git a/apps/gust/test/dag/terminator/gateway_test.exs b/apps/gust/test/dag/terminator/gateway_test.exs
index 212f60d..03ddfd3 100644
--- a/apps/gust/test/dag/terminator/gateway_test.exs
+++ b/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
diff --git a/apps/gust/test/dag/terminator_test.exs b/apps/gust/test/dag/terminator_test.exs
index 6ab23f6..7e620ad 100644
--- a/apps/gust/test/dag/terminator_test.exs
+++ b/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
diff --git a/apps/gust_web/lib/gust_web/live/run_live/index.ex b/apps/gust_web/lib/gust_web/live/run_live/index.ex
index 1dd24ee..47af817 100644
--- a/apps/gust_web/lib/gust_web/live/run_live/index.ex
+++ b/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
diff --git a/apps/gust_web/lib/gust_web/live/run_live/index.html.heex b/apps/gust_web/lib/gust_web/live/run_live/index.html.heex
index bdffa24..099a0bc 100644
--- a/apps/gust_web/lib/gust_web/live/run_live/index.html.heex
+++ b/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
diff --git a/apps/gust_web/test/gust_web/live/run_live_test.exs b/apps/gust_web/test/gust_web/live/run_live_test.exs
index be97a4f..430984d 100644
--- a/apps/gust_web/test/gust_web/live/run_live_test.exs
+++ b/apps/gust_web/test/gust_web/live/run_live_test.exs
@@ -1,4 +1,5 @@
 defmodule GustWeb.RunLiveTest do
+  alias Gust.DAG.Terminator.Gateway
   alias Gust.{Flows, Repo}
   use GustWeb.ConnCase
 
@@ -13,6 +14,10 @@ defmodule GustWeb.RunLiveTest do
       dag = dag_fixture(%{name: "dag_with_runs"})
       run = run_fixture(%{dag_id: dag.id})
 
+      stub(GustWeb.DAGTerminatorMock, :stop_run, fn run ->
+        Gateway.stop_run(run)
+      end)
+
       %{conn: conn, run: run, dag: dag}
     end
 
@@ -364,10 +369,45 @@ defmodule GustWeb.RunLiveTest do
     test "deletes run in listing", %{conn: conn, dag: dag, run: run} do
       {:ok, index_live, _html} = live(conn, ~g"/dags/#{dag.name}/runs?page_size=30&page=1")
 
-      assert index_live |> element("#runs-#{run.id} a", "Delete") |> render_click()
+      assert index_live |> element("#delete-run-#{run.id}") |> render_click()
       refute has_element?(index_live, "#runs-#{run.id}")
     end
 
+    test "stops an active run process before deleting it", %{conn: conn, dag: dag, run: run} do
+      {:ok, run} = Flows.update_run_status(run, :running)
+      runner = start_registered_runner(run.id)
+      runner_ref = Process.monitor(runner)
+
+      {:ok, index_live, _html} = live(conn, ~g"/dags/#{dag.name}/runs?page_size=30&page=1")
+
+      assert index_live |> element("#delete-run-#{run.id}") |> render_click()
+      assert_receive {:run_stopped, run_id}
+      assert run_id == run.id
+      assert_receive {:DOWN, ^runner_ref, :process, ^runner, :normal}
+      assert_raise Ecto.NoResultsError, fn -> Flows.get_run!(run.id) end
+    end
+
+    test "keeps a run when its active process cannot be stopped", %{
+      conn: conn,
+      dag: dag,
+      run: run
+    } do
+      {:ok, run} = Flows.update_run_status(run, :running)
+      runner = start_rejecting_runner(run.id)
+
+      on_exit(fn ->
+        if Process.alive?(runner), do: send(runner, :stop)
+      end)
+
+      {:ok, index_live, _html} = live(conn, ~g"/dags/#{dag.name}/runs?page_size=30&page=1")
+
+      html = index_live |> element("#delete-run-#{run.id}") |> render_click()
+
+      assert html =~ "Run could not be deleted because its process could not be stopped."
+      assert Process.alive?(runner)
+      assert Flows.get_run!(run.id)
+    end
+
     test "batch deletes selected runs in listing", %{conn: conn, dag: dag, run: first_run} do
       {:ok, first_run} = Flows.update_run_status(first_run, :succeeded)
       second_run = run_fixture(%{dag_id: dag.id, status: :failed})
@@ -396,13 +436,15 @@ defmodule GustWeb.RunLiveTest do
       assert_raise Ecto.NoResultsError, fn -> Flows.get_run!(second_run.id) end
     end
 
-    test "batch delete skips active runs and reports the result", %{
+    test "batch delete stops active runs and deletes runs in any status", %{
       conn: conn,
       dag: dag,
       run: created_run
     } do
       succeeded_run = run_fixture(%{dag_id: dag.id, status: :succeeded})
       running_run = run_fixture(%{dag_id: dag.id, status: :running})
+      runner = start_registered_runner(running_run.id)
+      runner_ref = Process.monitor(runner)
 
       {:ok, index_live, _html} = live(conn, ~g"/dags/#{dag.name}/runs?page_size=30&page=1")
 
@@ -414,12 +456,14 @@ defmodule GustWeb.RunLiveTest do
 
       html = index_live |> element("#batch-delete-runs") |> render_click()
 
-      assert html =~
-               "1 run deleted; 2 runs skipped: Created runs cannot be deleted. Running runs cannot be deleted."
+      assert html =~ "3 runs deleted"
+      assert_receive {:run_stopped, run_id}
+      assert run_id == running_run.id
+      assert_receive {:DOWN, ^runner_ref, :process, ^runner, :normal}
 
       assert_raise Ecto.NoResultsError, fn -> Flows.get_run!(succeeded_run.id) end
-      assert Flows.get_run!(running_run.id)
-      assert Flows.get_run!(created_run.id)
+      assert_raise Ecto.NoResultsError, fn -> Flows.get_run!(running_run.id) end
+      assert_raise Ecto.NoResultsError, fn -> Flows.get_run!(created_run.id) end
     end
 
     test "batch delete with no selected runs does nothing", %{conn: conn, dag: dag, run: run} do
@@ -625,4 +669,44 @@ defmodule GustWeb.RunLiveTest do
       refute index_live |> has_element?("#runs-#{failed_run.id}")
     end
   end
+
+  defp start_registered_runner(run_id) do
+    owner = self()
+
+    runner =
+      spawn(fn ->
+        {:ok, _owner} = Registry.register(Gust.Registry, "dag_run_#{run_id}", nil)
+        send(owner, {:run_registered, self()})
+
+        receive do
+          {:"$gen_call", from, :stop} ->
+            send(owner, {:run_stopped, run_id})
+            GenServer.reply(from, {:ok, :stopped})
+        end
+      end)
+
+    assert_receive {:run_registered, ^runner}
+    runner
+  end
+
+  defp start_rejecting_runner(run_id) do
+    owner = self()
+
+    runner =
+      spawn(fn ->
+        {:ok, _owner} = Registry.register(Gust.Registry, "dag_run_#{run_id}", nil)
+        send(owner, {:run_registered, self()})
+
+        receive do
+          {:"$gen_call", from, :stop} -> GenServer.reply(from, {:error, :cannot_stop})
+        end
+
+        receive do
+          :stop -> :ok
+        end
+      end)
+
+    assert_receive {:run_registered, ^runner}
+    runner
+  end
 end