diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4bb64d5790..cc75029e88 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -21,6 +21,9 @@ and this project adheres to
 - Make root layout configurable
   [#2310](https://github.com/OpenFn/lightning/pull/2310)
 
+- Use snapshots when initiating Github Sync
+  [#1827](https://github.com/OpenFn/lightning/issues/1827)
+
 ### Fixed
 
 ## [v2.7.11] - 2024-07-26
diff --git a/lib/lightning/export_utils.ex b/lib/lightning/export_utils.ex
index 6682a3ef4d..3bc414cb84 100644
--- a/lib/lightning/export_utils.ex
+++ b/lib/lightning/export_utils.ex
@@ -5,8 +5,8 @@ defmodule Lightning.ExportUtils do
   """
 
   alias Lightning.Projects
-  alias Lightning.Repo
   alias Lightning.Workflows
+  alias Lightning.Workflows.Snapshot
 
   defp hyphenate(string) when is_binary(string) do
     string |> String.replace(" ", "-")
@@ -41,36 +41,38 @@ defmodule Lightning.ExportUtils do
       else: base
   end
 
-  defp edge_to_treenode(%{source_job_id: nil} = edge, triggers) do
-    edge = Repo.preload(edge, [:source_trigger, :target_job])
-    trigger_name = edge.source_trigger.type |> Atom.to_string()
-    target_job = edge.target_job.name |> hyphenate()
+  defp edge_to_treenode(%{source_job_id: nil} = edge, triggers, jobs) do
+    source_trigger =
+      Enum.find(triggers, fn t -> t.id == edge.source_trigger_id end)
+
+    target_job = Enum.find(jobs, fn j -> j.id == edge.target_job_id end)
+    trigger_name = to_string(source_trigger.type)
+    target_job_name = hyphenate(target_job.name)
 
     %{
-      name: "#{trigger_name}->#{target_job}",
-      source_trigger: find_trigger_name(edge, triggers)
+      name: "#{trigger_name}->#{target_job_name}",
+      source_trigger: trigger_name
     }
-    |> merge_edge_common_fields(edge)
+    |> merge_edge_common_fields(edge, target_job)
   end
 
-  defp edge_to_treenode(%{source_trigger_id: nil} = edge, _unused_triggers) do
-    edge = Repo.preload(edge, [:source_job, :target_job])
-    source_job = edge.source_job.name |> hyphenate()
-    target_job = edge.target_job.name |> hyphenate()
+  defp edge_to_treenode(%{source_trigger_id: nil} = edge, _triggers, jobs) do
+    target_job = Enum.find(jobs, fn j -> j.id == edge.target_job_id end)
+    source_job = Enum.find(jobs, fn j -> j.id == edge.source_job_id end)
+    source_job_name = hyphenate(source_job.name)
+    target_job_name = hyphenate(target_job.name)
 
     %{
-      name: "#{source_job}->#{target_job}",
-      source_job: source_job
+      name: "#{source_job_name}->#{target_job_name}",
+      source_job: source_job_name
     }
-    |> merge_edge_common_fields(edge)
+    |> merge_edge_common_fields(edge, target_job)
   end
 
-  defp merge_edge_common_fields(json, edge) do
-    target_job = edge.target_job.name |> hyphenate()
-
+  defp merge_edge_common_fields(json, edge, target_job) do
     json
     |> Map.merge(%{
-      target_job: target_job,
+      target_job: hyphenate(target_job.name),
       condition_type: edge.condition_type |> Atom.to_string(),
       enabled: edge.enabled,
       node_type: :edge
@@ -87,12 +89,6 @@ defmodule Lightning.ExportUtils do
     end)
   end
 
-  defp find_trigger_name(edge, triggers) do
-    [trigger] = Enum.filter(triggers, fn t -> t.id == edge.source_trigger_id end)
-
-    trigger.name
-  end
-
   defp pick_and_sort(map) do
     ordering_map = %{
       project: [:name, :description, :credentials, :globals, :workflows],
@@ -229,7 +225,9 @@ defmodule Lightning.ExportUtils do
     edges =
       workflow.edges
       |> Enum.sort_by(& &1.inserted_at, NaiveDateTime)
-      |> Enum.map(fn e -> edge_to_treenode(e, triggers) end)
+      |> Enum.map(fn e ->
+        edge_to_treenode(e, workflow.triggers, workflow.jobs)
+      end)
 
     flow_map = %{jobs: jobs, edges: edges, triggers: triggers}
 
@@ -237,9 +235,11 @@ defmodule Lightning.ExportUtils do
     |> to_workflow_yaml_tree(workflow)
   end
 
-  def generate_new_yaml(project_id) do
-    project = Projects.get_project!(project_id)
+  @spec generate_new_yaml(Projects.Project.t(), [Snapshot.t()] | nil) ::
+          {:ok, binary()}
+  def generate_new_yaml(project, snapshots \\ nil)
 
+  def generate_new_yaml(project, nil) do
     yaml =
       project
       |> Workflows.get_workflows_for()
@@ -248,4 +248,14 @@ defmodule Lightning.ExportUtils do
 
     {:ok, yaml}
   end
+
+  def generate_new_yaml(project, snapshots) when is_list(snapshots) do
+    yaml =
+      snapshots
+      |> Enum.sort_by(& &1.name)
+      |> build_yaml_tree(project)
+      |> to_new_yaml()
+
+    {:ok, yaml}
+  end
 end
diff --git a/lib/lightning/projects.ex b/lib/lightning/projects.ex
index 877bc4ed7e..ce3445b865 100644
--- a/lib/lightning/projects.ex
+++ b/lib/lightning/projects.ex
@@ -25,6 +25,7 @@ defmodule Lightning.Projects do
   alias Lightning.RunStep
   alias Lightning.Services.ProjectHook
   alias Lightning.Workflows.Job
+  alias Lightning.Workflows.Snapshot
   alias Lightning.Workflows.Trigger
   alias Lightning.Workflows.Workflow
   alias Lightning.WorkOrder
@@ -558,11 +559,15 @@ defmodule Lightning.Projects do
       {:ok, string}
 
   """
-  @spec export_project(:yaml, any) :: {:ok, binary}
-  def export_project(:yaml, project_id) do
-    {:ok, yaml} = ExportUtils.generate_new_yaml(project_id)
+  @spec export_project(atom(), Ecto.UUID.t(), [Ecto.UUID.t()] | nil) ::
+          {:ok, binary}
+  def export_project(:yaml, project_id, snapshot_ids \\ nil) do
+    project = get_project!(project_id)
 
-    {:ok, yaml}
+    snapshots =
+      if snapshot_ids, do: Snapshot.get_all_by_ids(snapshot_ids), else: nil
+
+    {:ok, _yaml} = ExportUtils.generate_new_yaml(project, snapshots)
   end
 
   @doc """
diff --git a/lib/lightning/projects/provisioner.ex b/lib/lightning/projects/provisioner.ex
index d6c92c94ff..80b4a7ae19 100644
--- a/lib/lightning/projects/provisioner.ex
+++ b/lib/lightning/projects/provisioner.ex
@@ -132,8 +132,11 @@ defmodule Lightning.Projects.Provisioner do
 
   Exclude deleted workflows.
   """
-  @spec preload_dependencies(Project.t()) :: Project.t()
-  def preload_dependencies(project) do
+  @spec preload_dependencies(Project.t(), nil | [Ecto.UUID.t(), ...]) ::
+          Project.t()
+  def preload_dependencies(project, snapshots \\ nil)
+
+  def preload_dependencies(project, nil) do
     w = from(w in Workflow, where: is_nil(w.deleted_at))
 
     Repo.preload(
@@ -146,6 +149,10 @@ defmodule Lightning.Projects.Provisioner do
     )
   end
 
+  def preload_dependencies(project, snapshots) when is_list(snapshots) do
+    %{project | workflows: Snapshot.get_all_by_ids(snapshots)}
+  end
+
   defp project_changeset(project, attrs) do
     project
     |> cast(attrs, [:id, :name, :description])
diff --git a/lib/lightning/version_control/version_control.ex b/lib/lightning/version_control/version_control.ex
index 88deefeea0..db966a13a2 100644
--- a/lib/lightning/version_control/version_control.ex
+++ b/lib/lightning/version_control/version_control.ex
@@ -16,6 +16,8 @@ defmodule Lightning.VersionControl do
   alias Lightning.VersionControl.GithubError
   alias Lightning.VersionControl.ProjectRepoConnection
   alias Lightning.VersionControl.VersionControlUsageLimiter
+  alias Lightning.Workflows.Snapshot
+  alias Lightning.Workflows.Workflow
 
   defdelegate subscribe(user), to: Events
 
@@ -130,6 +132,8 @@ defmodule Lightning.VersionControl do
            VersionControlUsageLimiter.limit_github_sync(
              repo_connection.project_id
            ),
+         snapshots <-
+           list_or_create_snapshots_for_project(repo_connection.project_id),
          {:ok, client} <-
            GithubClient.build_installation_client(
              repo_connection.github_installation_id
@@ -143,20 +147,50 @@ defmodule Lightning.VersionControl do
              pull_yml_target_path() |> Path.basename(),
              %{
                ref: default_branch,
-               inputs: %{
-                 projectId: repo_connection.project_id,
-                 apiSecretName: api_secret_name(repo_connection),
-                 pathToConfig: config_target_path(repo_connection),
-                 branch: repo_connection.branch,
-                 commitMessage:
-                   "user #{user_email} initiated a sync from Lightning"
-               }
+               inputs:
+                 %{
+                   projectId: repo_connection.project_id,
+                   apiSecretName: api_secret_name(repo_connection),
+                   pathToConfig: config_target_path(repo_connection),
+                   branch: repo_connection.branch,
+                   commitMessage:
+                     "user #{user_email} initiated a sync from Lightning"
+                 }
+                 |> maybe_add_snapshots(snapshots)
              }
            ) do
       :ok
     end
   end
 
+  defp list_or_create_snapshots_for_project(project_id) do
+    current_query =
+      from w in Workflow,
+        left_join: s in assoc(w, :snapshots),
+        on: s.lock_version == w.lock_version,
+        where: w.project_id == ^project_id and is_nil(w.deleted_at),
+        select: {w, s.id}
+
+    workflows = Repo.all(current_query)
+
+    Enum.reduce(workflows, [], fn {workflow, snapshot_id}, acc ->
+      if is_nil(snapshot_id) do
+        {:ok, snapshot} = Snapshot.get_or_create_latest_for(workflow)
+        [snapshot.id | acc]
+      else
+        [snapshot_id | acc]
+      end
+    end)
+  end
+
+  defp maybe_add_snapshots(inputs, snapshot_ids) do
+    if Enum.empty?(snapshot_ids) do
+      inputs
+    else
+      Map.put(inputs, :snapshots, Enum.join(snapshot_ids, " "))
+    end
+  end
+
   def fetch_user_installations(user) do
     with {:ok, access_token} <- fetch_user_access_token(user),
          {:ok, client} <- GithubClient.build_bearer_client(access_token) do
diff --git a/lib/lightning/workflows/snapshot.ex b/lib/lightning/workflows/snapshot.ex
index 5837ab31b5..fa7f7d0d9a 100644
--- a/lib/lightning/workflows/snapshot.ex
+++ b/lib/lightning/workflows/snapshot.ex
@@ -157,6 +157,13 @@ defmodule Lightning.Workflows.Snapshot do
     |> Repo.all()
   end
 
+  def get_all_by_ids(ids) do
+    from(s in __MODULE__,
+      where: s.id in ^ids
+    )
+    |> Repo.all()
+  end
+
   @doc """
   Get the latest snapshot for a workflow, based on the lock_version.
 
diff --git a/lib/lightning_web/controllers/api/provisioning_controller.ex b/lib/lightning_web/controllers/api/provisioning_controller.ex
index ef9328873e..96374b4585 100644
--- a/lib/lightning_web/controllers/api/provisioning_controller.ex
+++ b/lib/lightning_web/controllers/api/provisioning_controller.ex
@@ -67,7 +67,8 @@ defmodule LightningWeb.API.ProvisioningController do
              conn.assigns.current_resource,
              project
            ),
-         project <- Provisioner.preload_dependencies(project) do
+         project <-
+           Provisioner.preload_dependencies(project, params["snapshots"]) do
       conn
       |> put_status(:ok)
       |> render("create.json", project: project)
@@ -78,7 +79,7 @@ defmodule LightningWeb.API.ProvisioningController do
   Returns a description of the project as yaml. Same as the export project to
   yaml button (see Downloads Controller) but made for the API.
   """
-  def show_yaml(conn, %{"id" => id}) do
+  def show_yaml(conn, %{"id" => id} = params) do
     with %Projects.Project{} = project <-
            Projects.get_project(id) || {:error, :not_found},
          :ok <-
@@ -88,7 +89,7 @@ defmodule LightningWeb.API.ProvisioningController do
              conn.assigns.current_resource,
              project
            ) do
-      {:ok, yaml} = Projects.export_project(:yaml, id)
+      {:ok, yaml} = Projects.export_project(:yaml, id, params["snapshots"])
 
       conn
       |> put_resp_content_type("text/yaml")
diff --git a/lib/lightning_web/controllers/api/provisioning_json.ex b/lib/lightning_web/controllers/api/provisioning_json.ex
index 57ec8ff208..82617c0510 100644
--- a/lib/lightning_web/controllers/api/provisioning_json.ex
+++ b/lib/lightning_web/controllers/api/provisioning_json.ex
@@ -7,6 +7,7 @@ defmodule LightningWeb.API.ProvisioningJSON do
   alias Lightning.Projects.Project
   alias Lightning.Workflows.Edge
   alias Lightning.Workflows.Job
+  alias Lightning.Workflows.Snapshot
   alias Lightning.Workflows.Trigger
   alias Lightning.Workflows.Workflow
 
@@ -21,47 +22,57 @@ defmodule LightningWeb.API.ProvisioningJSON do
   def as_json(%Project{} = project) do
     Ecto.embedded_dump(project, :json)
     |> Map.put(
-      "workflows",
+      :workflows,
       project.workflows
       |> Enum.sort_by(& &1.inserted_at, NaiveDateTime)
       |> Enum.map(&as_json/1)
     )
   end
 
-  def as_json(%Workflow{} = workflow) do
-    Ecto.embedded_dump(workflow, :json)
+  def as_json(%module{} = workflow_or_snapshot)
+      when module in [Workflow, Snapshot] do
+    workflow_id =
+      if module == Workflow do
+        workflow_or_snapshot.id
+      else
+        workflow_or_snapshot.workflow_id
+      end
+
+    workflow_or_snapshot
+    |> Ecto.embedded_dump(:json)
+    |> Map.put(:id, workflow_id)
     |> Map.put(
-      "jobs",
-      workflow.jobs
+      :jobs,
+      workflow_or_snapshot.jobs
       |> Enum.sort_by(& &1.inserted_at, NaiveDateTime)
       |> Enum.map(&as_json/1)
     )
     |> Map.put(
-      "triggers",
-      workflow.triggers
+      :triggers,
+      workflow_or_snapshot.triggers
       |> Enum.sort_by(& &1.inserted_at, NaiveDateTime)
       |> Enum.map(&as_json/1)
     )
     |> Map.put(
-      "edges",
-      workflow.edges
+      :edges,
+      workflow_or_snapshot.edges
       |> Enum.sort_by(& &1.inserted_at, NaiveDateTime)
       |> Enum.map(&as_json/1)
     )
   end
 
-  def as_json(%Job{} = job) do
+  def as_json(%module{} = job) when module in [Job, Snapshot.Job] do
     Ecto.embedded_dump(job, :json)
     |> Map.take(~w(id adaptor body name)a)
   end
 
-  def as_json(%Trigger{} = trigger) do
+  def as_json(%module{} = trigger) when module in [Trigger, Snapshot.Trigger] do
     Ecto.embedded_dump(trigger, :json)
     |> Map.take(~w(id type cron_expression enabled)a)
     |> drop_keys_with_nil_value()
   end
 
-  def as_json(%Edge{} = edge) do
+  def as_json(%module{} = edge) when module in [Edge, Snapshot.Edge] do
     Ecto.embedded_dump(edge, :json)
     |> Map.take(~w(
       id enabled source_job_id source_trigger_id
diff --git a/priv/github/pull.yml b/priv/github/pull.yml
index 1f250025fc..decf89e3a0 100644
--- a/priv/github/pull.yml
+++ b/priv/github/pull.yml
@@ -16,6 +16,9 @@ on:
       commitMessage:
         description: 'Commit message for project state and spec'
         required: true
+      snapshots:
+        description: 'IDs of snapshots separated by spaces'
+        required: false
 
 jobs:
   pull-from-lightning:
@@ -25,10 +28,11 @@ jobs:
     name: A job to pull changes from Lightning
     steps:
       - name: openfn pull and commit
-        uses: openfn/cli-pull-action@v1.0.0
+        uses: openfn/cli-pull-action@v1.1.0
         with:
           secret_input: ${{ secrets[inputs.apiSecretName] }}
           project_id_input: ${{ inputs.projectId }}
           config_path_input: ${{ inputs.pathToConfig }}
           branch_input: ${{ inputs.branch }}
           commit_message_input: ${{ inputs.commitMessage }}
+          snapshots_input: ${{ inputs.snapshots }}
diff --git a/test/lightning/version_control_test.exs b/test/lightning/version_control_test.exs
index 8898202181..62f91fd495 100644
--- a/test/lightning/version_control_test.exs
+++ b/test/lightning/version_control_test.exs
@@ -3,10 +3,12 @@ defmodule Lightning.VersionControlTest do
   alias Lightning.VersionControl
   alias Lightning.VersionControl.ProjectRepoConnection
   alias Lightning.Repo
+  alias Lightning.Workflows.Snapshot
 
   import Lightning.Factories
 
   import Lightning.GithubHelpers
+  import Mox
 
   describe "create_github_connection/2" do
     test "user with valid oauth token creates connection successfully" do
@@ -387,6 +389,39 @@ defmodule Lightning.VersionControlTest do
     end
   end
 
+  describe "initiate_sync/2" do
+    setup do
+      verify_on_exit!()
+
+      project = insert(:project)
+      workflow = insert(:simple_workflow, project: project)
+      user = user_with_valid_github_oauth()
+      repo_connection = insert(:project_repo_connection, project: project)
+
+      [
+        project: project,
+        user: user,
+        repo_connection: repo_connection,
+        workflow: workflow
+      ]
+    end
+
+    test "creates snapshots for workflows without snapshots", %{
+      user: user,
+      repo_connection: repo_connection,
+      workflow: workflow
+    } do
+      refute Snapshot.get_current_for(workflow)
+
+      expect_create_installation_token(repo_connection.github_installation_id)
+      expect_get_repo(repo_connection.repo)
+      expect_create_workflow_dispatch(repo_connection.repo, "openfn-pull.yml")
+
+      assert :ok = VersionControl.initiate_sync(repo_connection, user.email)
+      assert Snapshot.get_current_for(workflow)
+    end
+  end
+
   defp user_with_valid_github_oauth do
     active_token = %{
       "access_token" => "access-token",
diff --git a/test/lightning_web/controllers/api/provisioning_controller_test.exs b/test/lightning_web/controllers/api/provisioning_controller_test.exs
index 19ca5ee3e4..c590ca264e 100644
--- a/test/lightning_web/controllers/api/provisioning_controller_test.exs
+++ b/test/lightning_web/controllers/api/provisioning_controller_test.exs
@@ -4,6 +4,7 @@ defmodule LightningWeb.API.ProvisioningControllerTest do
   import Ecto.Query
   import Lightning.Factories
 
+  alias Lightning.Workflows.Snapshot
   alias Lightning.Workflows.Workflow
   alias LightningWeb.API.ProvisioningJSON
 
@@ -193,6 +194,66 @@ defmodule LightningWeb.API.ProvisioningControllerTest do
       assert workflow_resp["id"] == existing_workflow.id
     end
 
+    test "returns a project only with the specified snapshots", %{
+      conn: conn,
+      user: user
+    } do
+      %{id: project_id, name: project_name} =
+        project =
+        insert(:project,
+          project_users: [%{user_id: user.id}]
+        )
+
+      workflow_1 =
+        insert(:simple_workflow,
+          project: project,
+          name: "workflow 1"
+        )
+
+      {:ok, snapshot_1} = Snapshot.get_or_create_latest_for(workflow_1)
+
+      {:ok, updated_workflow_1} =
+        workflow_1
+        |> Ecto.Changeset.change(%{name: "updated-workflow-name"})
+        |> Lightning.Repo.update()
+
+      workflow_2 =
+        insert(:simple_workflow,
+          project: project,
+          name: "workflow 2"
+        )
+
+      {:ok, snapshot_2} = Snapshot.get_or_create_latest_for(workflow_2)
+
+      conn =
+        get(conn, ~p"/api/provision/#{project_id}", snapshots: [snapshot_1.id])
+
+      response = json_response(conn, 200)
+
+      assert %{
+               "id" => ^project_id,
+               "name" => ^project_name,
+               "workflows" => [workflow_resp]
+             } = response["data"]
+
+      # Only the first workflow is returned because its snapshot was specified
+      assert workflow_resp["id"] == workflow_1.id
+      # The name of the workflow is the original name, not the updated name
+      assert workflow_resp["name"] == workflow_1.name
+      assert updated_workflow_1.name != workflow_1.name
+
+      # Now we specify both snapshots
+      conn =
+        get(conn, ~p"/api/provision/#{project_id}",
+          snapshots: [snapshot_1.id, snapshot_2.id]
+        )
+
+      response = json_response(conn, 200)
+
+      assert %{"workflows" => workflows} = response["data"]
+      assert Enum.count(workflows) == 2
+    end
+
     test "returns a project if user has owner access", %{
       conn: conn,
       user: user
@@ -296,6 +357,61 @@ defmodule LightningWeb.API.ProvisioningControllerTest do
       assert response.status == 200
     end
 
+    test "returns valid project yaml for snapshots provided" do
+      project = insert(:project)
+      repo_connection = insert(:project_repo_connection, project: project)
+
+      workflow_1 =
+        insert(:simple_workflow,
+          project: project,
+          name: "workflow 1"
+        )
+
+      {:ok, snapshot_1} = Snapshot.get_or_create_latest_for(workflow_1)
+
+      {:ok, updated_workflow_1} =
+        workflow_1
+        |> Ecto.Changeset.change(%{name: "updated-workflow-name"})
+        |> Lightning.Repo.update()
+
+      workflow_2 =
+        insert(:simple_workflow,
+          project: project,
+          name: "workflow 2"
+        )
+
+      {:ok, snapshot_2} = Snapshot.get_or_create_latest_for(workflow_2)
+
+      conn =
+        Plug.Conn.put_req_header(
+          build_conn(),
+          "authorization",
+          "Bearer #{repo_connection.access_token}"
+        )
+
+      response =
+        get(
+          conn,
+          ~p"/api/provision/yaml?#{%{id: project.id, snapshots: [snapshot_1.id]}}"
+        )
+        |> response(200)
+
+      assert response =~ workflow_1.name
+      refute response =~ updated_workflow_1.name
+      refute response =~ workflow_2.name
+
+      response =
+        get(
+          conn,
+          ~p"/api/provision/yaml?#{%{id: project.id, snapshots: [snapshot_1.id, snapshot_2.id]}}"
+        )
+        |> response(200)
+
+      assert response =~ workflow_1.name
+      refute response =~ updated_workflow_1.name
+      assert response =~ workflow_2.name
+    end
+
     test "returns a 403 if an invalid repo conenction token is provided" do
       project_1 = insert(:project)
       project_2 = insert(:project)