diff --git a/lib/microwaveprop/weather/rebatch_asos.ex b/lib/microwaveprop/weather/rebatch_asos.ex
new file mode 100644
index 00000000..2793587f
--- /dev/null
+++ b/lib/microwaveprop/weather/rebatch_asos.ex
@@ -0,0 +1,111 @@
+defmodule Microwaveprop.Weather.RebatchAsos do
+  @moduledoc """
+  Converts pending single-station `WeatherFetchWorker` ASOS jobs into
+  the batched `asos_batch` shape — one job per unique
+  `(start_dt, end_dt)` window covering every station that had a
+  pending job for that window.
+
+  Callable from both a Mix task (`mix weather.rebatch_asos`) and a
+  release `eval` shell (`bin/microwaveprop eval
+  'Microwaveprop.Weather.RebatchAsos.run()'`). In prod the Mix API
+  isn't available, so this module is Mix-free.
+
+  Idempotent: a second run with no pending "asos" jobs is a no-op.
+  """
+
+  import Ecto.Query
+
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Workers.WeatherFetchWorker
+
+  @doc """
+  Run the rebatch. Options:
+    * `:dry_run` (boolean) — log the plan but don't mutate the queue.
+
+  Returns a summary map: `%{found: n, windows: n, inserted: n, cancelled: n}`.
+  """
+  @spec run(keyword()) :: %{
+          found: non_neg_integer(),
+          windows: non_neg_integer(),
+          inserted: non_neg_integer(),
+          cancelled: non_neg_integer()
+        }
+  def run(opts \\ []) do
+    dry_run? = Keyword.get(opts, :dry_run, false)
+    jobs = load_pending_asos_jobs()
+
+    if jobs == [] do
+      IO.puts("No pending `asos` jobs. Nothing to do.")
+      %{found: 0, windows: 0, inserted: 0, cancelled: 0}
+    else
+      do_rebatch(jobs, dry_run?)
+    end
+  end
+
+  defp load_pending_asos_jobs do
+    query =
+      from(j in Oban.Job,
+        where:
+          j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and
+            j.state in ["available", "scheduled", "retryable"] and
+            fragment("?->>'fetch_type' = ?", j.args, "asos"),
+        select: %{id: j.id, args: j.args}
+      )
+
+    Repo.all(query)
+  end
+
+  defp do_rebatch(jobs, dry_run?) do
+    groups =
+      jobs
+      |> Enum.group_by(fn %{args: a} ->
+        {Map.get(a, "start_dt"), Map.get(a, "end_dt")}
+      end)
+      |> Enum.reject(fn {{s, e}, _} -> is_nil(s) or is_nil(e) end)
+
+    IO.puts("Found #{length(jobs)} pending `asos` jobs across #{length(groups)} time windows.")
+
+    batch_args =
+      Enum.map(groups, fn {{start_dt, end_dt}, members} ->
+        sorted = Enum.sort_by(members, fn %{args: a} -> a["station_code"] end)
+
+        %{
+          "fetch_type" => "asos_batch",
+          "station_ids" => Enum.map(sorted, fn %{args: a} -> a["station_id"] end),
+          "station_codes" => Enum.map(sorted, fn %{args: a} -> a["station_code"] end),
+          "start_dt" => start_dt,
+          "end_dt" => end_dt,
+          "_source_ids" => Enum.map(sorted, & &1.id)
+        }
+      end)
+
+    if dry_run? do
+      Enum.each(batch_args, fn a ->
+        IO.puts("[dry-run] #{length(a["station_codes"])} stations for #{a["start_dt"]}..#{a["end_dt"]}")
+      end)
+
+      %{found: length(jobs), windows: length(groups), inserted: 0, cancelled: 0}
+    else
+      apply_rebatch(batch_args, length(jobs))
+    end
+  end
+
+  defp apply_rebatch(batch_args, found) do
+    {inserted, cancelled} =
+      Enum.reduce(batch_args, {0, 0}, fn a, {ins, can} ->
+        source_ids = a["_source_ids"]
+        job_args = Map.delete(a, "_source_ids")
+
+        Repo.transaction(fn ->
+          _ = job_args |> WeatherFetchWorker.new() |> Oban.insert!()
+          {:ok, _n} = Oban.cancel_all_jobs(from(j in Oban.Job, where: j.id in ^source_ids))
+        end)
+
+        {ins + 1, can + length(source_ids)}
+      end)
+
+    IO.puts("Rebatched: inserted #{inserted} batch jobs, cancelled #{cancelled} single-station jobs.")
+
+    %{found: found, windows: length(batch_args), inserted: inserted, cancelled: cancelled}
+  end
+end
diff --git a/lib/mix/tasks/weather_rebatch_asos.ex b/lib/mix/tasks/weather_rebatch_asos.ex
index a42b8c1f..e11fe5f9 100644
--- a/lib/mix/tasks/weather_rebatch_asos.ex
+++ b/lib/mix/tasks/weather_rebatch_asos.ex
@@ -2,111 +2,25 @@ defmodule Mix.Tasks.Weather.RebatchAsos do
   @shortdoc "Collapse pending single-station ASOS jobs into batched jobs"
 
   @moduledoc """
-  Converts every pending single-station `WeatherFetchWorker` ASOS job
-  into the batched (`asos_batch`) shape — one job per unique
-  `(start_dt, end_dt)` window covering every station that had a
-  pending job for that window.
+  Thin Mix-task wrapper over `Microwaveprop.Weather.RebatchAsos`.
+  The real logic lives in the plain module so it's callable from a
+  release shell too.
 
-  Why: the worker still accepts the old `"asos"` fetch_type for
-  backward compat, but every such job pays the full `IemRateLimiter`
-  gap + IEM 429-retry tail for a single station's rows. Collapsing
-  them into batched jobs gets the queue through the backlog at the
-  new per-request efficiency.
-
-  Usage (idempotent — running it twice is a no-op):
-
-      mix weather.rebatch_asos               # rewrites everything
-      mix weather.rebatch_asos --dry-run     # report only
+  Usage:
 
+      mix weather.rebatch_asos                # rewrite
+      mix weather.rebatch_asos --dry-run      # report only
   """
 
   use Mix.Task
 
-  import Ecto.Query
-
-  alias Microwaveprop.Repo
-  alias Microwaveprop.Workers.WeatherFetchWorker
-
-  require Logger
+  alias Microwaveprop.Weather.RebatchAsos
 
   @impl Mix.Task
   def run(args) do
     {opts, _, _} = OptionParser.parse(args, strict: [dry_run: :boolean])
     Mix.Task.run("app.start")
-    dry_run? = Keyword.get(opts, :dry_run, false)
-
-    jobs = load_pending_asos_jobs()
-
-    if jobs == [] do
-      Mix.shell().info("No pending `asos` jobs. Nothing to do.")
-    else
-      do_rebatch(jobs, dry_run?)
-    end
-  end
-
-  @doc false
-  def load_pending_asos_jobs do
-    query =
-      from(j in Oban.Job,
-        where:
-          j.worker == "Microwaveprop.Workers.WeatherFetchWorker" and
-            j.state in ["available", "scheduled", "retryable"] and
-            fragment("?->>'fetch_type' = ?", j.args, "asos"),
-        select: %{id: j.id, args: j.args}
-      )
-
-    Repo.all(query)
-  end
-
-  defp do_rebatch(jobs, dry_run?) do
-    groups =
-      jobs
-      |> Enum.group_by(fn %{args: a} ->
-        {Map.get(a, "start_dt"), Map.get(a, "end_dt")}
-      end)
-      |> Enum.reject(fn {{s, e}, _} -> is_nil(s) or is_nil(e) end)
-
-    Mix.shell().info("Found #{length(jobs)} pending `asos` jobs across #{length(groups)} time windows.")
-
-    batch_args =
-      Enum.map(groups, fn {{start_dt, end_dt}, members} ->
-        # Deterministic sort (matches the enqueuer's ordering) so a
-        # re-run with identical membership hits Oban's unique dedup.
-        sorted = Enum.sort_by(members, fn %{args: a} -> a["station_code"] end)
-
-        %{
-          "fetch_type" => "asos_batch",
-          "station_ids" => Enum.map(sorted, fn %{args: a} -> a["station_id"] end),
-          "station_codes" => Enum.map(sorted, fn %{args: a} -> a["station_code"] end),
-          "start_dt" => start_dt,
-          "end_dt" => end_dt,
-          "_source_ids" => Enum.map(sorted, & &1.id)
-        }
-      end)
-
-    if dry_run? do
-      Enum.each(batch_args, fn a ->
-        Mix.shell().info("[dry-run] #{length(a["station_codes"])} stations for #{a["start_dt"]}..#{a["end_dt"]}")
-      end)
-    else
-      apply_rebatch(batch_args)
-    end
-  end
-
-  defp apply_rebatch(batch_args) do
-    {inserted, cancelled} =
-      Enum.reduce(batch_args, {0, 0}, fn a, {ins, can} ->
-        source_ids = a["_source_ids"]
-        job_args = Map.delete(a, "_source_ids")
-
-        Repo.transaction(fn ->
-          _ = job_args |> WeatherFetchWorker.new() |> Oban.insert!()
-          {:ok, _n} = Oban.cancel_all_jobs(from(j in Oban.Job, where: j.id in ^source_ids))
-        end)
-
-        {ins + 1, can + length(source_ids)}
-      end)
-
-    Mix.shell().info("Rebatched: inserted #{inserted} batch jobs, cancelled #{cancelled} single-station jobs.")
+    _ = RebatchAsos.run(opts)
+    :ok
   end
 end
diff --git a/test/mix/tasks/weather_rebatch_asos_test.exs b/test/mix/tasks/weather_rebatch_asos_test.exs
index c7eb931e..0c4625de 100644
--- a/test/mix/tasks/weather_rebatch_asos_test.exs
+++ b/test/mix/tasks/weather_rebatch_asos_test.exs
@@ -2,8 +2,8 @@ defmodule Mix.Tasks.Weather.RebatchAsosTest do
   use Microwaveprop.DataCase, async: false
 
   alias Microwaveprop.Weather.IemClient
+  alias Microwaveprop.Weather.RebatchAsos
   alias Microwaveprop.Workers.WeatherFetchWorker
-  alias Mix.Tasks.Weather.RebatchAsos
 
   setup do
     # Oban runs inline in tests, so the batch job the task inserts
@@ -75,7 +75,7 @@ defmodule Mix.Tasks.Weather.RebatchAsosTest do
   test "dry-run does not mutate the queue" do
     insert_asos_job(Ecto.UUID.generate(), "KDFW", "2026-03-28T16:00:00Z", "2026-03-28T20:00:00Z")
 
-    ExUnit.CaptureIO.capture_io(fn -> RebatchAsos.run(["--dry-run"]) end)
+    ExUnit.CaptureIO.capture_io(fn -> RebatchAsos.run(dry_run: true) end)
 
     counts = pending_count_by_type()
     assert counts["asos"] == 1