close #4584 (#4585)

Author: taylordowns2000

CHANGELOG.md

@@ -39,6 +39,9 @@ and this project adheres to
 
 ### Fixed
 
+- Consider manual runs for "next cron run input" via the
+  `last_run_final_dataclip` function
+  [#4584](https://github.com/OpenFn/lightning/issues/4584)
 - Proper warn & error for exceeding max dataclip size
   [#4524](https://github.com/OpenFn/lightning/issues/4524)
 - Copying api tokens doesn't work on unsecure non-localhost contexts

lib/lightning/invocation.ex

@@ -229,14 +229,16 @@ defmodule Lightning.Invocation do
   end
 
   @doc """
-  Returns the final dataclip from the last successful run for a trigger.
-  Used when cron_cursor_job_id is nil (use final run state).
+  Returns the final dataclip from the last successful run for a trigger's
+  workflow. Used when cron_cursor_job_id is nil (use final run state).
+
+  Scopes by workflow (not trigger) so that manual runs are also considered.
   """
-  def last_run_final_dataclip(%Trigger{id: trigger_id}) do
+  def last_run_final_dataclip(%Trigger{workflow_id: workflow_id}) do
     from(r in Run,
       join: wo in assoc(r, :work_order),
       join: d in assoc(r, :final_dataclip),
-      where: wo.trigger_id == ^trigger_id,
+      where: wo.workflow_id == ^workflow_id,
       where: r.state == :success,
       where: is_nil(d.wiped_at),
       order_by: [desc: r.finished_at],

test/lightning/invocation_test.exs

@@ -714,6 +714,115 @@ defmodule Lightning.InvocationTest do
       assert result.id == final_dataclip.id
     end
 
+    test "returns the final dataclip from a manual run (no trigger on work order)" do
+      project = insert(:project)
+
+      %{workflow: workflow, trigger: trigger, snapshot: snapshot} =
+        build_workflow(project: project)
+
+      dataclip = insert(:dataclip, project: project)
+
+      final_dataclip =
+        insert(:dataclip,
+          project: project,
+          body: %{"manual_final" => "state"},
+          type: :step_result
+        )
+
+      [job] = workflow.jobs
+
+      # Manual run: work order has no trigger, run has starting_job instead
+      wo =
+        insert(:workorder,
+          workflow: workflow,
+          trigger: nil,
+          dataclip: dataclip,
+          snapshot: snapshot
+        )
+
+      insert(:run,
+        work_order: wo,
+        dataclip: dataclip,
+        starting_job: job,
+        snapshot: snapshot,
+        state: :success,
+        finished_at: DateTime.utc_now(),
+        final_dataclip: final_dataclip
+      )
+
+      result = Invocation.get_next_cron_run_dataclip(trigger)
+      assert result.id == final_dataclip.id
+    end
+
+    test "prefers the most recent successful run regardless of trigger" do
+      project = insert(:project)
+
+      %{workflow: workflow, trigger: trigger, snapshot: snapshot} =
+        build_workflow(project: project)
+
+      [job] = workflow.jobs
+
+      old_dataclip = insert(:dataclip, project: project)
+
+      old_final =
+        insert(:dataclip,
+          project: project,
+          body: %{"old" => "state"},
+          type: :step_result
+        )
+
+      # Older cron-triggered run
+      old_wo =
+        insert(:workorder,
+          workflow: workflow,
+          trigger: trigger,
+          dataclip: old_dataclip,
+          snapshot: snapshot
+        )
+
+      insert(:run,
+        work_order: old_wo,
+        dataclip: old_dataclip,
+        starting_trigger: trigger,
+        snapshot: snapshot,
+        state: :success,
+        finished_at: ~U[2025-01-01 00:00:00Z],
+        final_dataclip: old_final
+      )
+
+      # Newer manual run
+      new_dataclip = insert(:dataclip, project: project)
+
+      new_final =
+        insert(:dataclip,
+          project: project,
+          body: %{"new" => "state"},
+          type: :step_result
+        )
+
+      new_wo =
+        insert(:workorder,
+          workflow: workflow,
+          trigger: nil,
+          dataclip: new_dataclip,
+          snapshot: snapshot
+        )
+
+      insert(:run,
+        work_order: new_wo,
+        dataclip: new_dataclip,
+        starting_job: job,
+        snapshot: snapshot,
+        state: :success,
+        finished_at: ~U[2025-06-01 00:00:00Z],
+        final_dataclip: new_final
+      )
+
+      # Should use the newer manual run's final dataclip
+      result = Invocation.get_next_cron_run_dataclip(trigger)
+      assert result.id == new_final.id
+    end
+
     test "skips wiped dataclips and returns nil" do
       project = insert(:project)