diff --git a/CHANGELOG.md b/CHANGELOG.md
index 90e82d0b6cf..fe70940feaf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,6 +17,15 @@ and this project adheres to
 
 ### Added
 
+- Support collections in sandboxes. Collection names are now scoped per project,
+  empty collections are cloned into a sandbox on provision, and collection names
+  (not data) are synchronised when a sandbox is merged back into its parent. The
+  collections API accepts an optional `?project_id=<uuid>` query param to scope
+  a request to a specific project. When the query param is omitted and the name
+  is ambiguous across projects, the API returns 409 with guidance to add
+  `?project_id=`. Existing unscoped calls keep working for unambiguous names.
+  [#3548](https://github.com/OpenFn/lightning/issues/3548)
+
 ### Changed
 
 ### Fixed
diff --git a/lib/lightning/collections.ex b/lib/lightning/collections.ex
index d89671a8de5..212cab205a8 100644
--- a/lib/lightning/collections.ex
+++ b/lib/lightning/collections.ex
@@ -52,10 +52,32 @@ defmodule Lightning.Collections do
     Repo.all(query)
   end
 
+  @doc """
+  Looks up a collection by name across all projects.
+
+  Returns `{:error, :conflict}` when the name exists in more than one
+  project; the caller should then retry with `get_collection/2`.
+  """
   @spec get_collection(String.t()) ::
-          {:ok, Collection.t()} | {:error, :not_found}
+          {:ok, Collection.t()} | {:error, :not_found} | {:error, :conflict}
   def get_collection(name) do
-    case Repo.get_by(Collection, name: name) do
+    case Repo.all(from c in Collection, where: c.name == ^name) do
+      [] -> {:error, :not_found}
+      [collection] -> {:ok, collection}
+      [_ | _] -> {:error, :conflict}
+    end
+  end
+
+  @doc """
+  Looks up a collection scoped to a specific project.
+
+  Unambiguous by construction: there is at most one collection with a given
+  name in a project.
+  """
+  @spec get_collection(Ecto.UUID.t(), String.t()) ::
+          {:ok, Collection.t()} | {:error, :not_found}
+  def get_collection(project_id, name) do
+    case Repo.get_by(Collection, project_id: project_id, name: name) do
       nil -> {:error, :not_found}
       collection -> {:ok, collection}
     end
diff --git a/lib/lightning/collections/collection.ex b/lib/lightning/collections/collection.ex
index ecb4c1a0791..fc6b947b2f2 100644
--- a/lib/lightning/collections/collection.ex
+++ b/lib/lightning/collections/collection.ex
@@ -40,7 +40,8 @@ defmodule Lightning.Collections.Collection do
     |> validate_format(:name, ~r/^[a-z0-9]+([\-_.][a-z0-9]+)*$/,
       message: "Collection name must be URL safe"
     )
-    |> unique_constraint([:name],
+    |> unique_constraint(:name,
+      name: :collections_project_id_name_index,
       message: "A collection with this name already exists"
     )
   end
@@ -50,7 +51,8 @@ defmodule Lightning.Collections.Collection do
     |> validate_format(:name, ~r/^[a-z0-9]+([\-_.][a-z0-9]+)*$/,
       message: "Collection name must be URL safe"
     )
-    |> unique_constraint([:name],
+    |> unique_constraint(:name,
+      name: :collections_project_id_name_index,
       message: "A collection with this name already exists"
     )
   end
diff --git a/lib/lightning/projects/merge_projects.ex b/lib/lightning/projects/merge_projects.ex
index e286116bc93..b0f9db334a7 100644
--- a/lib/lightning/projects/merge_projects.ex
+++ b/lib/lightning/projects/merge_projects.ex
@@ -20,6 +20,14 @@ defmodule Lightning.Projects.MergeProjects do
   Workflows that don't match are marked for deletion (target) or creation
   (source).
 
+  Pure transformation — returns a merge document without touching the
+  database. Scope is workflow structure only; credentials, collections, and
+  other project-scoped resources are not part of the document.
+
+  For sandbox merges use `Lightning.Projects.Sandboxes.merge/4`, which
+  composes this with `Provisioner.import_document/4` and sandbox-specific
+  steps (e.g. collection name sync) inside a single transaction.
+
   ## Parameters
     * `source_project` - The project with modifications to merge
     * `target_project` - The target project to merge changes onto
diff --git a/lib/lightning/projects/provisioner.ex b/lib/lightning/projects/provisioner.ex
index 37aed26c40d..9d10373154b 100644
--- a/lib/lightning/projects/provisioner.ex
+++ b/lib/lightning/projects/provisioner.ex
@@ -37,7 +37,17 @@ defmodule Lightning.Projects.Provisioner do
   alias Lightning.WorkflowVersions
 
   @doc """
-  Import a project.
+  Import a project document into the database.
+
+  Upserts the project and the associations carried in the document
+  (workflows, project credentials, collections) inside a single transaction,
+  then fires audit, version-bump, and snapshot side effects.
+
+  Generic pipeline shared by YAML provisioning, CLI deploys, GitHub syncs,
+  and sandbox merges. It only acts on what the document contains —
+  sandbox-specific behaviours (credential cloning, dataclip copying,
+  collection name sync) are composed around this call in
+  `Lightning.Projects.Sandboxes`.
 
   ## Options
     * `:allow_stale` - If true, allows stale operations during import (useful for
diff --git a/lib/lightning/projects/sandboxes.ex b/lib/lightning/projects/sandboxes.ex
index cebcd8edfcb..c7d1192e15e 100644
--- a/lib/lightning/projects/sandboxes.ex
+++ b/lib/lightning/projects/sandboxes.ex
@@ -18,6 +18,7 @@ defmodule Lightning.Projects.Sandboxes do
   ## Operations
 
   * `provision/3` - Create a new sandbox from a parent project
+  * `merge/4` - Merge a sandbox into its target (workflows + collections)
   * `update_sandbox/3` - Update sandbox name, color, or environment
   * `delete_sandbox/2` - Delete a sandbox and all its descendants
 
@@ -36,12 +37,17 @@ defmodule Lightning.Projects.Sandboxes do
   import Ecto.Query
 
   alias Lightning.Accounts.User
+  alias Lightning.Collections
+  alias Lightning.Collections.Collection
   alias Lightning.Credentials.KeychainCredential
   alias Lightning.Policies.Permissions
+  alias Lightning.Projects.MergeProjects
   alias Lightning.Projects.Project
   alias Lightning.Projects.ProjectCredential
+  alias Lightning.Projects.Provisioner
   alias Lightning.Projects.SandboxPromExPlugin
   alias Lightning.Repo
+  alias Lightning.Services.CollectionHook
   alias Lightning.Workflows
   alias Lightning.Workflows.Edge
   alias Lightning.Workflows.Job
@@ -121,6 +127,46 @@ defmodule Lightning.Projects.Sandboxes do
     end
   end
 
+  @doc """
+  Merges a sandbox into its target project.
+
+  Imports the sandbox's workflow configuration into the target via the
+  provisioner and synchronises collection names. Runs inside a single
+  transaction. Collection data is never copied.
+
+  Callers must authorise the merge before calling (e.g. `:merge_sandbox`).
+
+  ## Parameters
+  * `source` - The sandbox project being merged
+  * `target` - The project receiving the merge
+  * `actor` - The user performing the merge
+  * `opts` - Merge options (`:selected_workflow_ids`, `:deleted_target_workflow_ids`)
+
+  ## Returns
+  * `{:ok, updated_target}` - Merge succeeded
+  * `{:error, reason}` - Workflow merge or collection sync failed
+  """
+  @spec merge(Project.t(), Project.t(), User.t(), map()) ::
+          {:ok, Project.t()} | {:error, term()}
+  def merge(
+        %Project{} = source,
+        %Project{} = target,
+        %User{} = actor,
+        opts \\ %{}
+      ) do
+    merge_doc = MergeProjects.merge_project(source, target, opts)
+
+    Repo.transact(fn ->
+      with {:ok, updated_target} <-
+             Provisioner.import_document(target, actor, merge_doc,
+               allow_stale: true
+             ),
+           {:ok, _} <- sync_collections(source, target) do
+        {:ok, updated_target}
+      end
+    end)
+  end
+
   @doc """
   Updates a sandbox project's basic attributes.
 
@@ -566,6 +612,87 @@ defmodule Lightning.Projects.Sandboxes do
     |> copy_workflow_version_history(sandbox.workflow_id_mapping)
     |> create_initial_workflow_snapshots()
     |> copy_selected_dataclips(parent.id, Map.get(original_attrs, :dataclip_ids))
+    |> clone_collections_from_parent(parent)
+  end
+
+  defp clone_collections_from_parent(sandbox, parent) do
+    parent_names = parent |> Collections.list_project_collections() |> names()
+    insert_empty_collections(sandbox.id, parent_names)
+    sandbox
+  end
+
+  @doc """
+  Synchronises collection names from a sandbox to its merge target.
+
+  Names only in the source are created empty in the target; names only in
+  the target are deleted along with their items. Collection data is never
+  copied. The combined byte-size of deleted collections is reported via
+  `CollectionHook.handle_delete/2` for usage accounting.
+
+  Runs inside a single transaction.
+  """
+  @spec sync_collections(Project.t(), Project.t()) ::
+          {:ok, %{created: non_neg_integer(), deleted: non_neg_integer()}}
+          | {:error, term()}
+  def sync_collections(%Project{} = source, %Project{} = target) do
+    source_names = source |> Collections.list_project_collections() |> names()
+
+    target_collections = Collections.list_project_collections(target)
+    target_names = names(target_collections)
+
+    to_create = MapSet.difference(source_names, target_names)
+
+    names_to_delete = MapSet.difference(target_names, source_names)
+
+    collections_to_delete =
+      Enum.filter(target_collections, &(&1.name in names_to_delete))
+
+    to_delete_ids = Enum.map(collections_to_delete, & &1.id)
+
+    deleted_byte_size =
+      Enum.reduce(collections_to_delete, 0, &(&1.byte_size_sum + &2))
+
+    Repo.transaction(fn ->
+      {created, _} = insert_empty_collections(target.id, to_create)
+      {deleted, _} = delete_collections(to_delete_ids)
+
+      if deleted_byte_size > 0 do
+        :ok = CollectionHook.handle_delete(target.id, deleted_byte_size)
+      end
+
+      %{created: created, deleted: deleted}
+    end)
+  end
+
+  defp names(collections), do: MapSet.new(collections, & &1.name)
+
+  defp insert_empty_collections(project_id, names) do
+    if Enum.empty?(names) do
+      {0, nil}
+    else
+      now = DateTime.utc_now() |> DateTime.truncate(:second)
+
+      rows =
+        Enum.map(names, fn name ->
+          %{
+            id: Ecto.UUID.generate(),
+            name: name,
+            project_id: project_id,
+            byte_size_sum: 0,
+            inserted_at: now,
+            updated_at: now
+          }
+        end)
+
+      # Concurrent merges may race to create the same collection.
+      Repo.insert_all(Collection, rows, on_conflict: :nothing)
+    end
+  end
+
+  defp delete_collections([]), do: {0, nil}
+
+  defp delete_collections(ids) do
+    Repo.delete_all(from c in Collection, where: c.id in ^ids)
   end
 
   defp copy_workflow_version_history(sandbox, workflow_id_mapping) do
diff --git a/lib/lightning_web/controllers/collections_controller.ex b/lib/lightning_web/controllers/collections_controller.ex
index a75398030fb..b515e5e098b 100644
--- a/lib/lightning_web/controllers/collections_controller.ex
+++ b/lib/lightning_web/controllers/collections_controller.ex
@@ -25,19 +25,8 @@ defmodule LightningWeb.CollectionsController do
 
   @valid_params ["key", "cursor", "limit" | @timestamp_params]
 
-  defp authorize(conn, collection) do
-    subject = conn.assigns[:subject] || conn.assigns[:current_user]
-
-    Permissions.can(
-      Lightning.Policies.Collections,
-      :access_collection,
-      subject,
-      collection
-    )
-  end
-
-  def put(conn, %{"name" => col_name, "key" => key, "value" => value}) do
-    with {:ok, collection} <- Collections.get_collection(col_name),
+  def put(conn, %{"name" => name, "key" => key, "value" => value}) do
+    with {:ok, collection} <- resolve(conn, name),
          :ok <- authorize(conn, collection),
          :ok <- Collections.put(collection, key, value) do
       json(conn, %{upserted: 1, error: nil})
@@ -50,8 +39,10 @@ defmodule LightningWeb.CollectionsController do
     end
   end
 
-  def put_all(conn, %{"name" => col_name, "items" => items}) do
-    with {:ok, collection} <- Collections.get_collection(col_name),
+  def put(conn, %{"name" => _, "key" => _}), do: missing_body(conn, "value")
+
+  def put_all(conn, %{"name" => name, "items" => items}) do
+    with {:ok, collection} <- resolve(conn, name),
          :ok <- authorize(conn, collection),
          {:ok, count} <- Collections.put_all(collection, items) do
       json(conn, %{upserted: count, error: nil})
@@ -66,21 +57,20 @@ defmodule LightningWeb.CollectionsController do
     end
   end
 
-  def get(conn, %{"name" => col_name, "key" => key}) do
-    with {:ok, collection} <- Collections.get_collection(col_name),
+  def put_all(conn, %{"name" => _}), do: missing_body(conn, "items")
+
+  def get(conn, %{"name" => name, "key" => key}) do
+    with {:ok, collection} <- resolve(conn, name),
          :ok <- authorize(conn, collection) do
       case Collections.get(collection, key) do
-        nil ->
-          resp(conn, :no_content, "")
-
-        item ->
-          json(conn, item)
+        nil -> resp(conn, :no_content, "")
+        item -> json(conn, item)
       end
     end
   end
 
-  def delete(conn, %{"name" => col_name, "key" => key}) do
-    with {:ok, collection} <- Collections.get_collection(col_name),
+  def delete(conn, %{"name" => name, "key" => key}) do
+    with {:ok, collection} <- resolve(conn, name),
          :ok <- authorize(conn, collection) do
       case Collections.delete(collection, key) do
         :ok ->
@@ -92,19 +82,34 @@ defmodule LightningWeb.CollectionsController do
     end
   end
 
-  def delete_all(conn, %{"name" => col_name} = params) do
-    with {:ok, collection} <- Collections.get_collection(col_name),
+  def delete_all(conn, %{"name" => name} = params) do
+    with {:ok, collection} <- resolve(conn, name),
          :ok <- authorize(conn, collection) do
       key_param = params["key"]
-
       {:ok, n} = Collections.delete_all(collection, key_param)
-
       json(conn, %{key: key_param, deleted: n, error: nil})
     end
   end
 
-  def download(conn, %{"name" => col_name}) do
-    with {:ok, collection} <- Collections.get_collection(col_name),
+  def stream(conn, %{"name" => name}) do
+    with {:ok, collection} <- resolve(conn, name),
+         :ok <- authorize(conn, collection),
+         {:ok, filters} <-
+           parse_query_params(Map.drop(conn.query_params, ["project_id"])) do
+      key_pattern = conn.query_params["key"]
+      items_stream = stream_all_in_chunks(collection, filters, key_pattern)
+      response_limit = Map.fetch!(filters, :limit)
+
+      case stream_chunked(conn, items_stream, response_limit) do
+        {:error, conn} -> conn
+        {:ok, conn} -> conn
+      end
+    end
+  end
+
+  def download(conn, %{"project_id" => project_id, "name" => name}) do
+    with {:ok, uuid} <- cast_uuid(project_id),
+         {:ok, collection} <- Collections.get_collection(uuid, name),
          :ok <- authorize(conn, collection) do
       items_stream =
         stream_all_in_chunks(
@@ -117,26 +122,48 @@ defmodule LightningWeb.CollectionsController do
       |> put_resp_content_type("application/json")
       |> put_resp_header(
         "content-disposition",
-        ~s(attachment; filename="#{col_name}.json")
+        ~s(attachment; filename="#{name}.json")
       )
       |> stream_as_json_array(items_stream)
     end
   end
 
-  def stream(conn, %{"name" => col_name} = params) do
-    with {:ok, collection, filters} <- validate_query(conn, col_name) do
-      key_pattern = Map.get(params, "key")
+  defp resolve(conn, name) do
+    case Map.get(conn.query_params, "project_id") do
+      nil ->
+        Collections.get_collection(name)
 
-      items_stream = stream_all_in_chunks(collection, filters, key_pattern)
-      response_limit = Map.fetch!(filters, :limit)
+      project_id ->
+        with {:ok, uuid} <- cast_uuid(project_id) do
+          Collections.get_collection(uuid, name)
+        end
+    end
+  end
 
-      case stream_chunked(conn, items_stream, response_limit) do
-        {:error, conn} -> conn
-        {:ok, conn} -> conn
-      end
+  defp cast_uuid(project_id) do
+    case Ecto.UUID.cast(project_id) do
+      {:ok, uuid} -> {:ok, uuid}
+      :error -> {:error, :bad_request}
     end
   end
 
+  defp authorize(conn, collection) do
+    subject = conn.assigns[:subject] || conn.assigns[:current_user]
+
+    Permissions.can(
+      Lightning.Policies.Collections,
+      :access_collection,
+      subject,
+      collection
+    )
+  end
+
+  defp missing_body(conn, field) do
+    conn
+    |> put_status(:unprocessable_entity)
+    |> json(%{error: "Missing required field: #{field}"})
+  end
+
   defp stream_as_json_array(conn, items_stream) do
     conn = send_chunked(conn, 200)
     {:ok, conn} = Plug.Conn.chunk(conn, "[")
@@ -203,17 +230,13 @@ defmodule LightningWeb.CollectionsController do
     end
   end
 
-  defp validate_query(conn, col_name) do
-    with {:ok, collection} <- Collections.get_collection(col_name),
-         :ok <- authorize(conn, collection),
-         query_params <-
-           Enum.into(conn.query_params, %{
-             "cursor" => nil,
-             "limit" => "#{@default_stream_limit}"
-           }),
-         {:ok, filters} <- validate_query_params(query_params) do
-      {:ok, collection, filters}
-    end
+  defp parse_query_params(query_params) do
+    query_params
+    |> Enum.into(%{
+      "cursor" => nil,
+      "limit" => "#{@default_stream_limit}"
+    })
+    |> validate_query_params()
   end
 
   defp validate_query_params(
diff --git a/lib/lightning_web/controllers/fallback_controller.ex b/lib/lightning_web/controllers/fallback_controller.ex
index 6e3f3c590bf..5098c4aea94 100644
--- a/lib/lightning_web/controllers/fallback_controller.ex
+++ b/lib/lightning_web/controllers/fallback_controller.ex
@@ -28,6 +28,16 @@ defmodule LightningWeb.FallbackController do
     |> render(:"401")
   end
 
+  def call(conn, {:error, :conflict}) do
+    conn
+    |> put_status(:conflict)
+    |> json(%{
+      error:
+        "Multiple collections found with this name. " <>
+          "Add ?project_id=<uuid> to disambiguate."
+    })
+  end
+
   def call(conn, {:error, :forbidden}) do
     conn
     |> put_status(:forbidden)
diff --git a/lib/lightning_web/live/project_live/collections_component.ex b/lib/lightning_web/live/project_live/collections_component.ex
index 5ffd9d96ed9..8374aacb335 100644
--- a/lib/lightning_web/live/project_live/collections_component.ex
+++ b/lib/lightning_web/live/project_live/collections_component.ex
@@ -60,7 +60,8 @@ defmodule LightningWeb.ProjectLive.CollectionsComponent do
         socket
       ) do
     with :ok <- can_create_collection(socket),
-         {:ok, collection} <- Collections.get_collection(collection_name),
+         {:ok, collection} <-
+           fetch_project_collection(socket, collection_name),
          :ok <- can_access_collection(socket, collection) do
       changeset =
         Collection.form_changeset(collection, %{raw_name: collection.name})
@@ -80,7 +81,8 @@ defmodule LightningWeb.ProjectLive.CollectionsComponent do
         socket
       ) do
     with :ok <- can_create_collection(socket),
-         {:ok, collection} <- Collections.get_collection(collection_name),
+         {:ok, collection} <-
+           fetch_project_collection(socket, collection_name),
          :ok <- can_access_collection(socket, collection) do
       {:noreply, assign(socket, collection: collection, action: :delete)}
     end
@@ -91,7 +93,7 @@ defmodule LightningWeb.ProjectLive.CollectionsComponent do
         %{"collection" => collection_name},
         socket
       ) do
-    with {:ok, collection} <- Collections.get_collection(collection_name),
+    with {:ok, collection} <- fetch_project_collection(socket, collection_name),
          :ok <- can_access_collection(socket, collection) do
       preview_json =
         case Collections.get_all(collection, limit: 1, cursor: nil) do
@@ -128,7 +130,8 @@ defmodule LightningWeb.ProjectLive.CollectionsComponent do
         socket
       ) do
     with :ok <- can_create_collection(socket),
-         {:ok, collection} <- Collections.get_collection(collection_name),
+         {:ok, collection} <-
+           fetch_project_collection(socket, collection_name),
          :ok <- can_access_collection(socket, collection) do
       case Collections.delete_collection(collection.id) do
         {:ok, _collection} ->
@@ -213,6 +216,22 @@ defmodule LightningWeb.ProjectLive.CollectionsComponent do
     end
   end
 
+  defp fetch_project_collection(socket, collection_name) do
+    case Collections.get_collection(
+           socket.assigns.project.id,
+           collection_name
+         ) do
+      {:ok, collection} ->
+        {:ok, collection}
+
+      {:error, :not_found} ->
+        {:noreply,
+         socket
+         |> put_flash(:error, "Collection not found")
+         |> push_navigate(to: socket.assigns.return_to)}
+    end
+  end
+
   defp can_access_collection(socket, collection) do
     Permissions.can(
       Lightning.Policies.Collections,
@@ -296,7 +315,9 @@ defmodule LightningWeb.ProjectLive.CollectionsComponent do
             </.button>
             <a
               id={"download-collection-#{collection.id}-button"}
-              href={~p"/download/collections/#{collection.name}"}
+              href={
+                ~p"/download/collections/#{collection.project_id}/#{collection.name}"
+              }
               class="table-action"
             >
               Download
diff --git a/lib/lightning_web/live/sandbox_live/components.ex b/lib/lightning_web/live/sandbox_live/components.ex
index f6e20e2fe61..44d4713bd0e 100644
--- a/lib/lightning_web/live/sandbox_live/components.ex
+++ b/lib/lightning_web/live/sandbox_live/components.ex
@@ -382,6 +382,7 @@ defmodule LightningWeb.SandboxLive.Components do
           >
             <:message>
               Sandbox merging is in beta. For production projects, use the CLI to merge locally and preview changes first.
+              Collection names will be synced: new collections are added (empty) to the target, and collections missing from the sandbox are removed from the target. Collection data is never merged.
             </:message>
           </Common.alert>
 
diff --git a/lib/lightning_web/live/sandbox_live/index.ex b/lib/lightning_web/live/sandbox_live/index.ex
index 33110a574d3..17e67fb9866 100644
--- a/lib/lightning_web/live/sandbox_live/index.ex
+++ b/lib/lightning_web/live/sandbox_live/index.ex
@@ -6,10 +6,13 @@ defmodule LightningWeb.SandboxLive.Index do
   alias Lightning.Projects
   alias Lightning.Projects.MergeProjects
   alias Lightning.Projects.ProjectLimiter
+  alias Lightning.Projects.Sandboxes
   alias Lightning.Repo
   alias Lightning.VersionControl
   alias LightningWeb.SandboxLive.Components
 
+  require Logger
+
   defmodule MergeWorkflow do
     defstruct [:id, :name, :is_diverged, :is_new, :is_deleted]
   end
@@ -795,16 +798,7 @@ defmodule LightningWeb.SandboxLive.Index do
         %{}
       end
 
-    result =
-      source
-      |> MergeProjects.merge_project(target, opts)
-      |> then(
-        &Lightning.Projects.Provisioner.import_document(target, actor, &1,
-          allow_stale: true
-        )
-      )
-
-    case result do
+    case Sandboxes.merge(source, target, actor, opts) do
       {:ok, _updated_target} = success ->
         maybe_commit_to_github(target, "Merged sandbox #{source.name}")
         success
@@ -887,5 +881,6 @@ defmodule LightningWeb.SandboxLive.Index do
   end
 
   defp format_merge_error(%{text: text}), do: text
+  defp format_merge_error(reason) when is_binary(reason), do: reason
   defp format_merge_error(reason), do: "Failed to merge: #{inspect(reason)}"
 end
diff --git a/lib/lightning_web/router.ex b/lib/lightning_web/router.ex
index ac2ce518878..b08cac16ed7 100644
--- a/lib/lightning_web/router.ex
+++ b/lib/lightning_web/router.ex
@@ -145,7 +145,11 @@ defmodule LightningWeb.Router do
     post "/users/two-factor", UserTOTPController, :create
     get "/setup_vcs", VersionControlController, :index
     get "/download/yaml", DownloadsController, :download_project_yaml
-    get "/download/collections/:name", CollectionsController, :download
+
+    get "/download/collections/:project_id/:name",
+        CollectionsController,
+        :download
+
     get "/dataclip/body/:id", DataclipController, :show
 
     get "/projects/:project_id/jobs/:job_id/dataclips",
diff --git a/priv/repo/migrations/20260410000000_scope_collection_name_uniqueness_to_project.exs b/priv/repo/migrations/20260410000000_scope_collection_name_uniqueness_to_project.exs
new file mode 100644
index 00000000000..cc21e88141f
--- /dev/null
+++ b/priv/repo/migrations/20260410000000_scope_collection_name_uniqueness_to_project.exs
@@ -0,0 +1,23 @@
+defmodule Lightning.Repo.Migrations.ScopeCollectionNameUniquenessToProject do
+  use Ecto.Migration
+
+  def up do
+    drop_if_exists unique_index(:collections, [:name])
+    create unique_index(:collections, [:project_id, :name])
+  end
+
+  def down do
+    dupes =
+      repo().query!("SELECT name FROM collections GROUP BY name HAVING count(*) > 1")
+
+    if dupes.num_rows > 0 do
+      raise Ecto.MigrationError,
+        message:
+          "Cannot rollback: #{dupes.num_rows} collection name(s) exist in multiple projects. " <>
+            "Remove duplicates before rolling back."
+    end
+
+    drop_if_exists unique_index(:collections, [:project_id, :name])
+    create unique_index(:collections, [:name])
+  end
+end
diff --git a/test/lightning/collections_test.exs b/test/lightning/collections_test.exs
index 02b35505a0d..beb99d97ac4 100644
--- a/test/lightning/collections_test.exs
+++ b/test/lightning/collections_test.exs
@@ -17,6 +17,37 @@ defmodule Lightning.CollectionsTest do
       assert {:error, :not_found} =
                Collections.get_collection("nonexistent")
     end
+
+    test "returns a conflict error when the same name exists in multiple projects" do
+      name = "shared-name"
+      insert(:collection, name: name)
+      insert(:collection, name: name)
+
+      assert {:error, :conflict} = Collections.get_collection(name)
+    end
+  end
+
+  describe "get_collection/2" do
+    test "returns the collection for the given project" do
+      name = "shared-name"
+      %{id: project_id_1} = project_1 = insert(:project)
+      %{id: project_id_2} = project_2 = insert(:project)
+      %{id: id_1} = insert(:collection, name: name, project: project_1)
+      %{id: id_2} = insert(:collection, name: name, project: project_2)
+
+      assert {:ok, %Collection{id: ^id_1}} =
+               Collections.get_collection(project_id_1, name)
+
+      assert {:ok, %Collection{id: ^id_2}} =
+               Collections.get_collection(project_id_2, name)
+    end
+
+    test "returns not_found when the collection does not exist in the project" do
+      %{id: project_id} = insert(:project)
+
+      assert {:error, :not_found} =
+               Collections.get_collection(project_id, "nonexistent")
+    end
   end
 
   describe "create_collection/2" do
@@ -48,7 +79,7 @@ defmodule Lightning.CollectionsTest do
                     {"A collection with this name already exists",
                      [
                        constraint: :unique,
-                       constraint_name: "collections_name_index"
+                       constraint_name: "collections_project_id_name_index"
                      ]}
                 ]
               }} =
diff --git a/test/lightning/sandboxes_test.exs b/test/lightning/sandboxes_test.exs
index aaceac714af..351a81abe3a 100644
--- a/test/lightning/sandboxes_test.exs
+++ b/test/lightning/sandboxes_test.exs
@@ -1,21 +1,28 @@
 defmodule Lightning.Projects.SandboxesTest do
   use Lightning.DataCase, async: true
+  use Mimic
 
   import Ecto.Query
   import Lightning.Factories
-  alias Lightning.Repo
 
   alias Lightning.Accounts.User
   alias Lightning.Credentials.KeychainCredential
+  alias Lightning.Invocation.Dataclip
   alias Lightning.Projects.Project
   alias Lightning.Projects.ProjectCredential
   alias Lightning.Projects.Sandboxes
-  alias Lightning.Workflows.Workflow
-  alias Lightning.Workflows.WorkflowVersion
+  alias Lightning.Repo
+  alias Lightning.Workflows.Edge
   alias Lightning.Workflows.Job
   alias Lightning.Workflows.Trigger
-  alias Lightning.Workflows.Edge
-  alias Lightning.Invocation.Dataclip
+  alias Lightning.Workflows.Workflow
+  alias Lightning.Workflows.WorkflowVersion
+
+  setup_all do
+    Mimic.copy(Lightning.Collections)
+    Mimic.copy(Lightning.Projects.Provisioner)
+    :ok
+  end
 
   defp ensure_member!(%Project{} = project, %User{} = user, role) do
     insert(:project_user, %{project: project, user: user, role: role})
@@ -547,6 +554,286 @@ defmodule Lightning.Projects.SandboxesTest do
     end
   end
 
+  describe "collections provisioning" do
+    test "clones empty collection records from parent into sandbox" do
+      actor = insert(:user)
+      parent = insert(:project)
+      ensure_member!(parent, actor, :owner)
+
+      insert(:collection, project: parent, name: "col-a")
+
+      insert(:collection,
+        project: parent,
+        name: "col-b",
+        items: [%{key: "k", value: "v"}]
+      )
+
+      {:ok, sandbox} = Sandboxes.provision(parent, actor, %{name: "sandbox-x"})
+
+      sandbox_collections =
+        Lightning.Collections.list_project_collections(sandbox)
+
+      assert Enum.map(sandbox_collections, & &1.name) |> Enum.sort() == [
+               "col-a",
+               "col-b"
+             ]
+
+      Enum.each(sandbox_collections, fn col ->
+        assert Lightning.Collections.get_all(
+                 col,
+                 %{cursor: nil, limit: 100},
+                 nil
+               ) ==
+                 []
+      end)
+    end
+
+    test "provisioning a parent with no collections creates no sandbox collections" do
+      actor = insert(:user)
+      parent = insert(:project)
+      ensure_member!(parent, actor, :owner)
+
+      {:ok, sandbox} = Sandboxes.provision(parent, actor, %{name: "sandbox-x"})
+
+      assert Lightning.Collections.list_project_collections(sandbox) == []
+    end
+
+    test "each sandbox gets its own copy of parent collections" do
+      actor = insert(:user)
+      parent = insert(:project)
+      ensure_member!(parent, actor, :owner)
+
+      insert(:collection, project: parent, name: "col-a")
+
+      {:ok, sandbox_1} = Sandboxes.provision(parent, actor, %{name: "sandbox-1"})
+      {:ok, sandbox_2} = Sandboxes.provision(parent, actor, %{name: "sandbox-2"})
+
+      assert length(Lightning.Collections.list_project_collections(sandbox_1)) ==
+               1
+
+      assert length(Lightning.Collections.list_project_collections(sandbox_2)) ==
+               1
+    end
+  end
+
+  describe "sync_collections/2" do
+    test "creates collections in target that exist in source but not target" do
+      source = insert(:project)
+      target = insert(:project)
+
+      insert(:collection, project: source, name: "shared")
+      insert(:collection, project: source, name: "only-in-source")
+      insert(:collection, project: target, name: "shared")
+
+      assert {:ok, %{created: 1, deleted: 0}} =
+               Sandboxes.sync_collections(source, target)
+
+      target_names =
+        target
+        |> Lightning.Collections.list_project_collections()
+        |> Enum.map(& &1.name)
+        |> Enum.sort()
+
+      assert target_names == ["only-in-source", "shared"]
+    end
+
+    test "deletes collections in target that are missing from source, including items" do
+      source = insert(:project)
+      target = insert(:project)
+
+      insert(:collection, project: source, name: "shared")
+      insert(:collection, project: target, name: "shared")
+
+      dropped =
+        insert(:collection,
+          project: target,
+          name: "only-in-target",
+          items: [%{key: "k", value: "v"}]
+        )
+
+      assert {:ok, %{created: 0, deleted: 1}} =
+               Sandboxes.sync_collections(source, target)
+
+      refute Lightning.Repo.get(Lightning.Collections.Collection, dropped.id)
+
+      assert Lightning.Repo.all(
+               from i in Lightning.Collections.Item,
+                 where: i.collection_id == ^dropped.id
+             ) == []
+    end
+
+    test "is a no-op when both projects have the same collections" do
+      source = insert(:project)
+      target = insert(:project)
+
+      insert(:collection, project: source, name: "a")
+      insert(:collection, project: target, name: "a")
+
+      assert {:ok, %{created: 0, deleted: 0}} =
+               Sandboxes.sync_collections(source, target)
+    end
+
+    test "fires the collection-delete hook with the combined byte size" do
+      Mox.verify_on_exit!()
+
+      source = insert(:project)
+      %{id: target_id} = target = insert(:project)
+
+      insert(:collection, project: source, name: "keep")
+      insert(:collection, project: target, name: "keep")
+
+      insert(:collection,
+        project: target,
+        name: "drop-a",
+        byte_size_sum: 120
+      )
+
+      insert(:collection,
+        project: target,
+        name: "drop-b",
+        byte_size_sum: 45
+      )
+
+      Mox.expect(
+        Lightning.Extensions.MockCollectionHook,
+        :handle_delete,
+        fn ^target_id, 165 -> :ok end
+      )
+
+      assert {:ok, %{created: 0, deleted: 2}} =
+               Sandboxes.sync_collections(source, target)
+    end
+
+    test "does not copy collection data across" do
+      source = insert(:project)
+      target = insert(:project)
+
+      insert(:collection,
+        project: source,
+        name: "with-data",
+        items: [%{key: "k", value: "v"}]
+      )
+
+      assert {:ok, %{created: 1, deleted: 0}} =
+               Sandboxes.sync_collections(source, target)
+
+      [new_collection] = Lightning.Collections.list_project_collections(target)
+
+      assert Lightning.Collections.get_all(
+               new_collection,
+               %{cursor: nil, limit: 100},
+               nil
+             ) == []
+    end
+
+    test "runs in a single transaction -- either everything or nothing" do
+      source = insert(:project)
+      target = insert(:project)
+
+      insert(:collection, project: source, name: "to-create")
+      insert(:collection, project: target, name: "to-delete")
+
+      result =
+        Lightning.Repo.transaction(fn ->
+          {:ok, _summary} =
+            Sandboxes.sync_collections(source, target)
+
+          Lightning.Repo.rollback(:simulated_failure)
+        end)
+
+      assert result == {:error, :simulated_failure}
+
+      target_names =
+        target
+        |> Lightning.Collections.list_project_collections()
+        |> Enum.map(& &1.name)
+
+      assert target_names == ["to-delete"]
+    end
+  end
+
+  describe "merge/4" do
+    test "imports the merge document and syncs collections" do
+      actor = insert(:user)
+      parent = insert(:project)
+      ensure_member!(parent, actor, :owner)
+
+      insert(:simple_workflow, project: parent)
+
+      sandbox =
+        insert(:project,
+          parent: parent,
+          project_users: [%{user: actor, role: :owner}]
+        )
+
+      insert(:simple_workflow, project: sandbox)
+
+      insert(:collection, project: sandbox, name: "new-col")
+
+      assert {:ok, _updated} = Sandboxes.merge(sandbox, parent, actor)
+
+      parent_names =
+        parent
+        |> Lightning.Collections.list_project_collections()
+        |> Enum.map(& &1.name)
+
+      assert "new-col" in parent_names
+    end
+
+    test "defaults opts to empty map" do
+      actor = insert(:user)
+      parent = insert(:project)
+      ensure_member!(parent, actor, :owner)
+
+      insert(:simple_workflow, project: parent)
+
+      sandbox =
+        insert(:project,
+          parent: parent,
+          project_users: [%{user: actor, role: :owner}]
+        )
+
+      insert(:simple_workflow, project: sandbox)
+
+      assert {:ok, _updated} = Sandboxes.merge(sandbox, parent, actor)
+    end
+
+    test "rolls back DB changes from import_document when collection sync fails" do
+      actor = insert(:user)
+      parent = insert(:project)
+      ensure_member!(parent, actor, :owner)
+
+      sandbox =
+        insert(:project,
+          parent: parent,
+          project_users: [%{user: actor, role: :owner}]
+        )
+
+      marker_name = "atomicity-marker-#{System.unique_integer([:positive])}"
+
+      Mimic.stub(Lightning.Projects.Provisioner, :import_document, fn target,
+                                                                      _actor,
+                                                                      _doc,
+                                                                      _opts ->
+        %Project{name: marker_name}
+        |> Ecto.Changeset.change()
+        |> Repo.insert!()
+
+        {:ok, target}
+      end)
+
+      Mimic.stub(Lightning.Collections, :list_project_collections, fn _ ->
+        raise "simulated sync failure"
+      end)
+
+      assert_raise RuntimeError, ~r/simulated sync failure/, fn ->
+        Sandboxes.merge(sandbox, parent, actor)
+      end
+
+      refute Repo.exists?(from p in Project, where: p.name == ^marker_name)
+    end
+  end
+
   describe "keychains" do
     test "clones only used keychains and rewires jobs to cloned keychains" do
       %{
diff --git a/test/lightning_web/collections_controller_test.exs b/test/lightning_web/collections_controller_test.exs
index 4451f58d2fe..6ee0f1bc9fb 100644
--- a/test/lightning_web/collections_controller_test.exs
+++ b/test/lightning_web/collections_controller_test.exs
@@ -235,6 +235,25 @@ defmodule LightningWeb.API.CollectionsControllerTest do
                "error" => "some limit error message"
              }
     end
+
+    test "returns 'Format error' when the value fails validation", %{conn: conn} do
+      user = insert(:user)
+      project = insert(:project, project_users: [%{user: user}])
+      collection = insert(:collection, project: project)
+      token = Lightning.Accounts.generate_api_token(user)
+
+      oversized = String.duplicate("x", 1_000_001)
+
+      conn =
+        conn
+        |> assign_bearer(token)
+        |> put(~p"/collections/#{collection.name}/key", value: oversized)
+
+      assert json_response(conn, 200) == %{
+               "upserted" => 0,
+               "error" => "Format error"
+             }
+    end
   end
 
   describe "POST /collections/:name" do
@@ -400,6 +419,24 @@ defmodule LightningWeb.API.CollectionsControllerTest do
              }
     end
 
+    test "returns 'Item Not Found' when deleting a missing key", %{conn: conn} do
+      user = insert(:user)
+      project = insert(:project, project_users: [%{user: user}])
+      collection = insert(:collection, project: project)
+      token = Lightning.Accounts.generate_api_token(user)
+
+      conn =
+        conn
+        |> assign_bearer(token)
+        |> delete(~p"/collections/#{collection.name}/missing-key")
+
+      assert json_response(conn, 200) == %{
+               "key" => "missing-key",
+               "deleted" => 0,
+               "error" => "Item Not Found"
+             }
+    end
+
     test "deletes matching items", %{conn: conn} do
       user = insert(:user)
 
@@ -996,7 +1033,7 @@ defmodule LightningWeb.API.CollectionsControllerTest do
     end
   end
 
-  describe "GET /download/collections/:name" do
+  describe "GET /download/collections/:project_id/:name" do
     setup :register_and_log_in_user
     setup :create_project_for_current_user
 
@@ -1022,7 +1059,8 @@ defmodule LightningWeb.API.CollectionsControllerTest do
         value: ~s({"name": "Bob"})
       )
 
-      response = get(conn, ~p"/download/collections/#{collection.name}")
+      response =
+        get(conn, ~p"/download/collections/#{project.id}/#{collection.name}")
 
       assert response.status == 200
 
@@ -1049,7 +1087,8 @@ defmodule LightningWeb.API.CollectionsControllerTest do
     } do
       collection = insert(:collection, project: project)
 
-      response = get(conn, ~p"/download/collections/#{collection.name}")
+      response =
+        get(conn, ~p"/download/collections/#{project.id}/#{collection.name}")
 
       assert response.status == 200
       assert Jason.decode!(response.resp_body) == []
@@ -1059,16 +1098,422 @@ defmodule LightningWeb.API.CollectionsControllerTest do
       other_project = insert(:project)
       collection = insert(:collection, project: other_project)
 
-      response = get(conn, ~p"/download/collections/#{collection.name}")
+      response =
+        get(
+          conn,
+          ~p"/download/collections/#{other_project.id}/#{collection.name}"
+        )
 
       assert response.status == 401
     end
 
-    test "returns 404 for non-existent collection", %{conn: conn} do
-      response = get(conn, ~p"/download/collections/non-existent")
+    test "returns 404 for non-existent collection", %{
+      conn: conn,
+      project: project
+    } do
+      response = get(conn, ~p"/download/collections/#{project.id}/non-existent")
 
       assert response.status == 404
     end
+
+    test "returns 400 when project_id is not a valid UUID", %{conn: conn} do
+      response = get(conn, ~p"/download/collections/not-a-uuid/anything")
+
+      assert response.status == 400
+    end
+  end
+
+  describe "malformed request bodies" do
+    setup %{conn: conn} do
+      user = insert(:user)
+      project = insert(:project, project_users: [%{user: user}])
+      collection = insert(:collection, project: project)
+      token = Lightning.Accounts.generate_api_token(user)
+      conn = assign_bearer(conn, token)
+      {:ok, conn: conn, project: project, collection: collection}
+    end
+
+    test "PUT with no value returns 422", %{conn: conn, collection: collection} do
+      conn = put(conn, ~p"/collections/#{collection.name}/some-key", %{})
+
+      assert json_response(conn, 422)["error"] =~ "Missing required field: value"
+    end
+
+    test "POST with no items returns 422", %{conn: conn, collection: collection} do
+      conn = post(conn, ~p"/collections/#{collection.name}", %{})
+
+      assert json_response(conn, 422)["error"] =~ "Missing required field: items"
+    end
+  end
+
+  describe "ambiguous name without ?project_id" do
+    setup %{conn: conn} do
+      user = insert(:user)
+      project_1 = insert(:project, project_users: [%{user: user}])
+      project_2 = insert(:project, project_users: [%{user: user}])
+      name = "shared-collection"
+      insert(:collection, name: name, project: project_1)
+      insert(:collection, name: name, project: project_2)
+      token = Lightning.Accounts.generate_api_token(user)
+      conn = assign_bearer(conn, token)
+      {:ok, conn: conn, name: name}
+    end
+
+    test "GET /:name returns 409", %{conn: conn, name: name} do
+      conn = get(conn, ~p"/collections/#{name}")
+      assert json_response(conn, 409)["error"] =~ "Add ?project_id="
+    end
+
+    test "GET /:name/:key returns 409", %{conn: conn, name: name} do
+      conn = get(conn, ~p"/collections/#{name}/some-key")
+      assert json_response(conn, 409)["error"] =~ "Add ?project_id="
+    end
+
+    test "PUT /:name/:key returns 409", %{conn: conn, name: name} do
+      conn = put(conn, ~p"/collections/#{name}/some-key", value: "val")
+      assert json_response(conn, 409)["error"] =~ "Add ?project_id="
+    end
+
+    test "POST /:name returns 409", %{conn: conn, name: name} do
+      conn = post(conn, ~p"/collections/#{name}", items: [])
+      assert json_response(conn, 409)["error"] =~ "Add ?project_id="
+    end
+
+    test "DELETE /:name/:key returns 409", %{conn: conn, name: name} do
+      conn = delete(conn, ~p"/collections/#{name}/some-key")
+      assert json_response(conn, 409)["error"] =~ "Add ?project_id="
+    end
+
+    test "DELETE /:name returns 409", %{conn: conn, name: name} do
+      conn = delete(conn, ~p"/collections/#{name}")
+      assert json_response(conn, 409)["error"] =~ "Add ?project_id="
+    end
+  end
+
+  describe "?project_id=<uuid> query param" do
+    setup %{conn: conn} do
+      user = insert(:user)
+      project = insert(:project, project_users: [%{user: user}])
+
+      collection =
+        insert(:collection,
+          project: project,
+          items: [%{key: "foo", value: "bar"}]
+        )
+
+      token = Lightning.Accounts.generate_api_token(user)
+      conn = assign_bearer(conn, token)
+      {:ok, conn: conn, project: project, collection: collection}
+    end
+
+    test "GET /:name/:key returns the item", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      item = hd(collection.items)
+
+      conn =
+        get(
+          conn,
+          ~p"/collections/#{collection.name}/foo?project_id=#{project.id}"
+        )
+
+      assert json_response(conn, 200) == %{
+               "key" => item.key,
+               "value" => item.value,
+               "created" => DateTime.to_iso8601(item.inserted_at),
+               "updated" => DateTime.to_iso8601(item.updated_at)
+             }
+    end
+
+    test "GET /:name/:key returns 204 when item not found", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      conn =
+        get(
+          conn,
+          ~p"/collections/#{collection.name}/nonexistent?project_id=#{project.id}"
+        )
+
+      assert response(conn, 204) == ""
+    end
+
+    test "GET /:name/:key returns 404 when collection not found", %{
+      conn: conn,
+      project: project
+    } do
+      conn =
+        get(
+          conn,
+          ~p"/collections/nonexistent-collection/foo?project_id=#{project.id}"
+        )
+
+      assert json_response(conn, 404)
+    end
+
+    test "GET /:name streams items scoped by project", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      conn =
+        get(conn, ~p"/collections/#{collection.name}?project_id=#{project.id}")
+
+      body = json_response(conn, 200)
+      assert length(body["items"]) == 1
+      assert hd(body["items"])["key"] == "foo"
+    end
+
+    test "PUT /:name/:key upserts an item", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      conn =
+        put(
+          conn,
+          ~p"/collections/#{collection.name}/new-key?project_id=#{project.id}",
+          value: "new-val"
+        )
+
+      assert json_response(conn, 200) == %{"upserted" => 1, "error" => nil}
+    end
+
+    test "POST /:name upserts multiple items", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      conn =
+        post(
+          conn,
+          ~p"/collections/#{collection.name}?project_id=#{project.id}",
+          items: [%{key: "a", value: "1"}, %{key: "b", value: "2"}]
+        )
+
+      assert json_response(conn, 200) == %{"upserted" => 2, "error" => nil}
+    end
+
+    test "DELETE /:name/:key deletes an item", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      conn =
+        delete(
+          conn,
+          ~p"/collections/#{collection.name}/foo?project_id=#{project.id}"
+        )
+
+      assert json_response(conn, 200) == %{
+               "key" => "foo",
+               "deleted" => 1,
+               "error" => nil
+             }
+    end
+
+    test "DELETE /:name deletes all items", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      conn =
+        delete(
+          conn,
+          ~p"/collections/#{collection.name}?project_id=#{project.id}"
+        )
+
+      assert json_response(conn, 200)["deleted"] == 1
+    end
+
+    test "returns 401 when user does not have access to the project", %{
+      conn: conn
+    } do
+      other_project = insert(:project)
+      other_collection = insert(:collection, project: other_project)
+
+      conn =
+        get(
+          conn,
+          ~p"/collections/#{other_collection.name}/foo?project_id=#{other_project.id}"
+        )
+
+      assert json_response(conn, 401)
+    end
+
+    test "resolves correctly when same name exists in multiple projects", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      other_project = insert(:project, project_users: [])
+      insert(:collection, name: collection.name, project: other_project)
+
+      conn =
+        get(
+          conn,
+          ~p"/collections/#{collection.name}/foo?project_id=#{project.id}"
+        )
+
+      assert json_response(conn, 200)["key"] == "foo"
+    end
+
+    test "returns 400 when project_id is not a valid UUID", %{
+      conn: conn,
+      collection: collection
+    } do
+      conn =
+        get(conn, ~p"/collections/#{collection.name}?project_id=not-a-uuid")
+
+      assert response(conn, 400)
+    end
+
+    test "with a valid run token, can access own project's collection", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      workflow = insert(:simple_workflow, project: project)
+
+      workorder =
+        insert(:workorder, workflow: workflow, dataclip: insert(:dataclip))
+
+      run =
+        insert(:run,
+          work_order: workorder,
+          dataclip: workorder.dataclip,
+          starting_trigger: hd(workflow.triggers)
+        )
+
+      token = Lightning.Workers.generate_run_token(run)
+
+      conn =
+        conn
+        |> assign_bearer(token)
+        |> get(~p"/collections/#{collection.name}/foo?project_id=#{project.id}")
+
+      assert json_response(conn, 200)["key"] == "foo"
+    end
+
+    test "with a valid run token, can write to own project's collection", %{
+      conn: conn,
+      project: project,
+      collection: collection
+    } do
+      workflow = insert(:simple_workflow, project: project)
+
+      workorder =
+        insert(:workorder, workflow: workflow, dataclip: insert(:dataclip))
+
+      run =
+        insert(:run,
+          work_order: workorder,
+          dataclip: workorder.dataclip,
+          starting_trigger: hd(workflow.triggers)
+        )
+
+      token = Lightning.Workers.generate_run_token(run)
+
+      conn =
+        conn
+        |> assign_bearer(token)
+        |> put(
+          ~p"/collections/#{collection.name}/new-key?project_id=#{project.id}",
+          value: "new-val"
+        )
+
+      assert json_response(conn, 200) == %{"upserted" => 1, "error" => nil}
+    end
+
+    test "with a run token, cannot access a different project's collection", %{
+      conn: conn
+    } do
+      other_project = insert(:project)
+      other_collection = insert(:collection, project: other_project)
+
+      workflow = insert(:simple_workflow)
+
+      workorder =
+        insert(:workorder, workflow: workflow, dataclip: insert(:dataclip))
+
+      run =
+        insert(:run,
+          work_order: workorder,
+          dataclip: workorder.dataclip,
+          starting_trigger: hd(workflow.triggers)
+        )
+
+      token = Lightning.Workers.generate_run_token(run)
+
+      conn =
+        conn
+        |> assign_bearer(token)
+        |> get(
+          ~p"/collections/#{other_collection.name}/foo?project_id=#{other_project.id}"
+        )
+
+      assert json_response(conn, 401)
+    end
+
+    test "with a run token, cannot write to a different project's collection",
+         %{conn: conn} do
+      other_project = insert(:project)
+      other_collection = insert(:collection, project: other_project)
+
+      workflow = insert(:simple_workflow)
+
+      workorder =
+        insert(:workorder, workflow: workflow, dataclip: insert(:dataclip))
+
+      run =
+        insert(:run,
+          work_order: workorder,
+          dataclip: workorder.dataclip,
+          starting_trigger: hd(workflow.triggers)
+        )
+
+      token = Lightning.Workers.generate_run_token(run)
+
+      conn =
+        conn
+        |> assign_bearer(token)
+        |> put(
+          ~p"/collections/#{other_collection.name}/k?project_id=#{other_project.id}",
+          value: "v"
+        )
+
+      assert json_response(conn, 401)
+    end
+
+    test "with a run token, bare ambiguous name returns 409 before authorization",
+         %{conn: conn, project: project, collection: collection} do
+      other_project = insert(:project)
+      insert(:collection, name: collection.name, project: other_project)
+
+      workflow = insert(:simple_workflow, project: project)
+
+      workorder =
+        insert(:workorder, workflow: workflow, dataclip: insert(:dataclip))
+
+      run =
+        insert(:run,
+          work_order: workorder,
+          dataclip: workorder.dataclip,
+          starting_trigger: hd(workflow.triggers)
+        )
+
+      token = Lightning.Workers.generate_run_token(run)
+
+      conn =
+        conn
+        |> assign_bearer(token)
+        |> get(~p"/collections/#{collection.name}/foo")
+
+      assert json_response(conn, 409)["error"] =~ "Add ?project_id="
+    end
   end
 
   defp encode_decode(item) do
diff --git a/test/lightning_web/live/project_live_test.exs b/test/lightning_web/live/project_live_test.exs
index 9d21cfe93ed..a01681f2968 100644
--- a/test/lightning_web/live/project_live_test.exs
+++ b/test/lightning_web/live/project_live_test.exs
@@ -6841,7 +6841,7 @@ defmodule LightningWeb.ProjectLiveTest do
           ~p"/projects/#{their_project.id}/settings#collections"
         )
 
-      assert flash["error"] == "You are not authorized to perform this action"
+      assert flash["error"] == "Collection not found"
     end
 
     test "cannot edit a collection belonging to another project", %{
@@ -6870,7 +6870,7 @@ defmodule LightningWeb.ProjectLiveTest do
           ~p"/projects/#{their_project.id}/settings#collections"
         )
 
-      assert flash["error"] == "You are not authorized to perform this action"
+      assert flash["error"] == "Collection not found"
     end
 
     test "cannot delete a collection belonging to another project", %{
@@ -6899,7 +6899,7 @@ defmodule LightningWeb.ProjectLiveTest do
           ~p"/projects/#{their_project.id}/settings#collections"
         )
 
-      assert flash["error"] == "You are not authorized to perform this action"
+      assert flash["error"] == "Collection not found"
     end
   end
 
diff --git a/test/lightning_web/live/sandbox_live/index_test.exs b/test/lightning_web/live/sandbox_live/index_test.exs
index 75a1e038430..2163e501ae3 100644
--- a/test/lightning_web/live/sandbox_live/index_test.exs
+++ b/test/lightning_web/live/sandbox_live/index_test.exs
@@ -1482,6 +1482,154 @@ defmodule LightningWeb.SandboxLive.IndexTest do
     end
   end
 
+  describe "collection sync on merge" do
+    setup :register_and_log_in_user
+
+    setup %{user: user} do
+      root =
+        insert(:project,
+          name: "root",
+          project_users: [%{user: user, role: :owner}]
+        )
+
+      sandbox =
+        insert(:project,
+          name: "sandbox",
+          parent: root,
+          project_users: [%{user: user, role: :owner}]
+        )
+
+      {:ok, root: root, sandbox: sandbox}
+    end
+
+    defp mock_provisioner_ok(target) do
+      Mimic.expect(Lightning.Projects.MergeProjects, :merge_project, fn _src,
+                                                                        _tgt,
+                                                                        _opts ->
+        "merged_yaml"
+      end)
+
+      Mimic.expect(Lightning.Projects.Provisioner, :import_document, fn _tgt,
+                                                                        _actor,
+                                                                        _yaml,
+                                                                        _opts ->
+        {:ok, target}
+      end)
+
+      Mimic.expect(Lightning.Projects, :delete_sandbox, fn source, _actor ->
+        {:ok, source}
+      end)
+    end
+
+    test "new collections in sandbox are created in parent on merge", %{
+      conn: conn,
+      root: root,
+      sandbox: sandbox
+    } do
+      insert(:collection, project: sandbox, name: "new-col")
+
+      {:ok, view, _} = live(conn, ~p"/projects/#{root.id}/sandboxes")
+      mock_provisioner_ok(root)
+
+      Mimic.allow(Lightning.Projects.MergeProjects, self(), view.pid)
+      Mimic.allow(Lightning.Projects.Provisioner, self(), view.pid)
+      Mimic.allow(Lightning.Projects, self(), view.pid)
+
+      view
+      |> element("#branch-rewire-sandbox-#{sandbox.id} button")
+      |> render_click()
+
+      view |> form("#merge-sandbox-modal form") |> render_submit()
+
+      parent_names =
+        Lightning.Collections.list_project_collections(root)
+        |> Enum.map(& &1.name)
+
+      assert "new-col" in parent_names
+    end
+
+    test "collections deleted from sandbox are removed from parent on merge", %{
+      conn: conn,
+      root: root,
+      sandbox: sandbox
+    } do
+      # Parent has a collection, sandbox does not
+      insert(:collection, project: root, name: "to-delete")
+
+      {:ok, view, _} = live(conn, ~p"/projects/#{root.id}/sandboxes")
+      mock_provisioner_ok(root)
+
+      Mimic.allow(Lightning.Projects.MergeProjects, self(), view.pid)
+      Mimic.allow(Lightning.Projects.Provisioner, self(), view.pid)
+      Mimic.allow(Lightning.Projects, self(), view.pid)
+
+      view
+      |> element("#branch-rewire-sandbox-#{sandbox.id} button")
+      |> render_click()
+
+      view |> form("#merge-sandbox-modal form") |> render_submit()
+
+      parent_names =
+        Lightning.Collections.list_project_collections(root)
+        |> Enum.map(& &1.name)
+
+      refute "to-delete" in parent_names
+    end
+
+    test "collections present in both are unchanged after merge", %{
+      conn: conn,
+      root: root,
+      sandbox: sandbox
+    } do
+      insert(:collection, project: root, name: "shared")
+      insert(:collection, project: sandbox, name: "shared")
+
+      {:ok, view, _} = live(conn, ~p"/projects/#{root.id}/sandboxes")
+      mock_provisioner_ok(root)
+
+      Mimic.allow(Lightning.Projects.MergeProjects, self(), view.pid)
+      Mimic.allow(Lightning.Projects.Provisioner, self(), view.pid)
+      Mimic.allow(Lightning.Projects, self(), view.pid)
+
+      view
+      |> element("#branch-rewire-sandbox-#{sandbox.id} button")
+      |> render_click()
+
+      view |> form("#merge-sandbox-modal form") |> render_submit()
+
+      parent_collections = Lightning.Collections.list_project_collections(root)
+      assert length(parent_collections) == 1
+      assert hd(parent_collections).name == "shared"
+    end
+
+    test "merge fails with flash error when collection sync fails", %{
+      conn: conn,
+      root: root,
+      sandbox: sandbox
+    } do
+      {:ok, view, _} = live(conn, ~p"/projects/#{root.id}/sandboxes")
+
+      Mimic.expect(Lightning.Projects.Sandboxes, :merge, fn _src,
+                                                            _tgt,
+                                                            _actor,
+                                                            _opts ->
+        {:error, "Failed to sync collections: :boom"}
+      end)
+
+      Mimic.allow(Lightning.Projects.Sandboxes, self(), view.pid)
+
+      view
+      |> element("#branch-rewire-sandbox-#{sandbox.id} button")
+      |> render_click()
+
+      view |> form("#merge-sandbox-modal form") |> render_submit()
+
+      html = render(view)
+      assert html =~ "Failed to sync collections"
+      refute has_element?(view, "#merge-sandbox-modal")
+    end
+  end
+
   describe "Merge modal authorization" do
     setup :register_and_log_in_user