diff --git a/lib/arcadia_cloud/digital_ocean/client.ex b/lib/arcadia_cloud/digital_ocean/client.ex
index a1a90af..2a7bfc0 100644
--- a/lib/arcadia_cloud/digital_ocean/client.ex
+++ b/lib/arcadia_cloud/digital_ocean/client.ex
@@ -62,6 +62,43 @@ defmodule ArcadiaCloud.DigitalOcean.Client do
     request(:delete, "/snapshots/#{snapshot_id}", purpose: opts[:purpose] || "provisioning")
   end
 
+  # ---- DNS records ----------------------------------------------------------
+
+  def list_domain_records(domain, opts \\ []) do
+    list_paginated("/domains/#{domain}/records", "domain_records",
+      Keyword.put(opts, :purpose, opts[:purpose] || "sync_full"))
+  end
+
+  @doc """
+  Create a DNS record. `attrs` must include type + name + data; optional
+  ttl, priority, port, weight, flags, tag. Returns {:ok, record}.
+  """
+  def create_domain_record(domain, attrs, opts \\ []) do
+    case request(:post, "/domains/#{domain}/records",
+           body: attrs,
+           purpose: opts[:purpose] || "provisioning"
+         ) do
+      {:ok, %{"domain_record" => record}} -> {:ok, record}
+      other -> other
+    end
+  end
+
+  def update_domain_record(domain, record_id, attrs, opts \\ []) do
+    case request(:put, "/domains/#{domain}/records/#{record_id}",
+           body: attrs,
+           purpose: opts[:purpose] || "provisioning"
+         ) do
+      {:ok, %{"domain_record" => record}} -> {:ok, record}
+      other -> other
+    end
+  end
+
+  def delete_domain_record(domain, record_id, opts \\ []) do
+    request(:delete, "/domains/#{domain}/records/#{record_id}",
+      purpose: opts[:purpose] || "provisioning"
+    )
+  end
+
   # ---- billing --------------------------------------------------------------
 
   def get_balance(opts \\ []) do
diff --git a/lib/arcadia_cloud/provisioning/steps/delete_dns_record.ex b/lib/arcadia_cloud/provisioning/steps/delete_dns_record.ex
new file mode 100644
index 0000000..9422cbb
--- /dev/null
+++ b/lib/arcadia_cloud/provisioning/steps/delete_dns_record.ex
@@ -0,0 +1,101 @@
+defmodule ArcadiaCloud.Provisioning.Steps.DeleteDnsRecord do
+  @moduledoc """
+  Deletes a DNS record, identified by (type, name) within a zone.
+
+  Saga inputs:
+    dns_domain       — required; the zone
+    dns_record_type  — required
+    dns_record_name  — required
+
+  Idempotent: if the record is already gone, succeeds.
+
+  Compensation: recreates the record from the stashed prior state. If
+  the record didn't exist at execute time there's nothing to restore.
+  """
+
+  @behaviour ArcadiaCloud.Provisioning.Step
+
+  alias ArcadiaCloud.DigitalOcean.Client
+  alias ArcadiaCloud.Provisioning.SagaState
+
+  @impl true
+  def name, do: "delete_dns_record"
+
+  @impl true
+  def execute(state) do
+    domain = SagaState.get_input(state, :dns_domain)
+    type = SagaState.get_input(state, :dns_record_type)
+    rname = SagaState.get_input(state, :dns_record_name)
+
+    cond do
+      is_nil(domain) or is_nil(type) or is_nil(rname) ->
+        {:error, :missing_dns_inputs}
+
+      true ->
+        case find_record(domain, type, rname) do
+          {:ok, nil} ->
+            {:ok, SagaState.put_output(state, :dns_deleted, "absent")}
+
+          {:ok, record} ->
+            do_delete(state, domain, record)
+
+          {:error, reason} ->
+            {:error, reason}
+        end
+    end
+  end
+
+  @impl true
+  def compensate(state) do
+    domain = SagaState.get_input(state, :dns_domain)
+
+    case SagaState.get_output(state, :dns_deleted_record) do
+      nil ->
+        :ok
+
+      prior ->
+        attrs = %{
+          type: prior["type"],
+          name: prior["name"],
+          data: prior["data"],
+          ttl: prior["ttl"]
+        }
+
+        case Client.create_domain_record(domain, attrs) do
+          {:ok, _} -> :ok
+          {:error, reason} -> {:error, reason}
+        end
+    end
+  end
+
+  defp do_delete(state, domain, record) do
+    case Client.delete_domain_record(domain, record["id"]) do
+      {:ok, _} ->
+        {:ok,
+         state
+         |> SagaState.put_output(:dns_deleted, "deleted")
+         |> SagaState.put_output(:dns_deleted_record, %{
+           "type" => record["type"],
+           "name" => record["name"],
+           "data" => record["data"],
+           "ttl" => record["ttl"]
+         })}
+
+      {:error, {:http, 404, _}} ->
+        {:ok, SagaState.put_output(state, :dns_deleted, "absent")}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp find_record(domain, type, rname) do
+    case Client.list_domain_records(domain) do
+      {:ok, records} ->
+        {:ok, Enum.find(records, &(&1["type"] == type and &1["name"] == rname))}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+end
diff --git a/lib/arcadia_cloud/provisioning/steps/upsert_dns_record.ex b/lib/arcadia_cloud/provisioning/steps/upsert_dns_record.ex
new file mode 100644
index 0000000..f3adb14
--- /dev/null
+++ b/lib/arcadia_cloud/provisioning/steps/upsert_dns_record.ex
@@ -0,0 +1,171 @@
+defmodule ArcadiaCloud.Provisioning.Steps.UpsertDnsRecord do
+  @moduledoc """
+  Idempotently ensures a DNS record exists with the desired data.
+
+  Saga inputs:
+    dns_domain       — required; the zone (e.g. "sky-ai.com")
+    dns_record_type  — required; "A" | "CNAME" | "TXT" | ...
+    dns_record_name  — required; subdomain ("@" for apex)
+    dns_record_data  — required; the record value
+    dns_record_ttl   — optional; defaults to 1800
+
+  Behaviour:
+    - record (type, name) absent  → create it
+    - present with matching data  → no-op
+    - present with different data → update it, stashing the prior state
+
+  Compensation:
+    - if we created the record → delete it
+    - if we updated it         → restore the prior data/ttl
+    - if it was a no-op        → nothing
+  """
+
+  @behaviour ArcadiaCloud.Provisioning.Step
+
+  alias ArcadiaCloud.DigitalOcean.Client
+  alias ArcadiaCloud.Provisioning.SagaState
+
+  @default_ttl 1800
+
+  @impl true
+  def name, do: "upsert_dns_record"
+
+  @impl true
+  def execute(state) do
+    with {:ok, domain, type, rname, data, ttl} <- read_inputs(state) do
+      case find_record(domain, type, rname) do
+        {:ok, nil} ->
+          create(state, domain, type, rname, data, ttl)
+
+        {:ok, existing} ->
+          if record_matches?(existing, data, ttl) do
+            {:ok, record_outcome(state, existing["id"], "noop", nil)}
+          else
+            update(state, domain, existing, type, rname, data, ttl)
+          end
+
+        {:error, reason} ->
+          {:error, reason}
+      end
+    end
+  end
+
+  @impl true
+  def compensate(state) do
+    domain = SagaState.get_input(state, :dns_domain)
+    outcome = SagaState.get_output(state, :dns_outcome)
+    record_id = SagaState.get_output(state, :dns_record_id)
+
+    case outcome do
+      "created" ->
+        delete_record(domain, record_id)
+
+      "updated" ->
+        restore_record(state, domain, record_id)
+
+      _ ->
+        :ok
+    end
+  end
+
+  # ---- execute helpers ------------------------------------------------------
+
+  defp create(state, domain, type, rname, data, ttl) do
+    attrs = %{type: type, name: rname, data: data, ttl: ttl}
+
+    case Client.create_domain_record(domain, attrs) do
+      {:ok, %{"id" => id}} -> {:ok, record_outcome(state, id, "created", nil)}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  defp update(state, domain, existing, type, rname, data, ttl) do
+    prior = %{
+      "type" => existing["type"],
+      "name" => existing["name"],
+      "data" => existing["data"],
+      "ttl" => existing["ttl"]
+    }
+
+    attrs = %{type: type, name: rname, data: data, ttl: ttl}
+
+    case Client.update_domain_record(domain, existing["id"], attrs) do
+      {:ok, %{"id" => id}} -> {:ok, record_outcome(state, id, "updated", prior)}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  defp record_outcome(state, record_id, outcome, prior) do
+    state
+    |> SagaState.put_output(:dns_record_id, record_id)
+    |> SagaState.put_output(:dns_outcome, outcome)
+    |> SagaState.put_output(:dns_prior, prior)
+  end
+
+  # ---- compensate helpers ---------------------------------------------------
+
+  defp delete_record(domain, record_id) do
+    case Client.delete_domain_record(domain, record_id) do
+      {:ok, _} -> :ok
+      {:error, {:http, 404, _}} -> :ok
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
+  defp restore_record(state, domain, record_id) do
+    case SagaState.get_output(state, :dns_prior) do
+      nil ->
+        :ok
+
+      prior ->
+        attrs = %{
+          type: prior["type"],
+          name: prior["name"],
+          data: prior["data"],
+          ttl: prior["ttl"]
+        }
+
+        case Client.update_domain_record(domain, record_id, attrs) do
+          {:ok, _} -> :ok
+          {:error, {:http, 404, _}} -> :ok
+          {:error, reason} -> {:error, reason}
+        end
+    end
+  end
+
+  # ---- shared ---------------------------------------------------------------
+
+  defp read_inputs(state) do
+    domain = SagaState.get_input(state, :dns_domain)
+    type = SagaState.get_input(state, :dns_record_type)
+    rname = SagaState.get_input(state, :dns_record_name)
+    data = SagaState.get_input(state, :dns_record_data)
+    ttl = SagaState.get_input(state, :dns_record_ttl) || @default_ttl
+
+    if domain && type && rname && data do
+      {:ok, domain, type, rname, data, ttl}
+    else
+      {:error, :missing_dns_inputs}
+    end
+  end
+
+  defp find_record(domain, type, rname) do
+    case Client.list_domain_records(domain) do
+      {:ok, records} ->
+        {:ok, Enum.find(records, &(&1["type"] == type and &1["name"] == rname))}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  # DO normalizes record data (e.g. appends trailing dot to CNAMEs).
+  # Compare loosely on the trimmed/dotless form.
+  defp record_matches?(existing, data, ttl) do
+    normalize(existing["data"]) == normalize(data) and existing["ttl"] == ttl
+  end
+
+  defp normalize(nil), do: ""
+  defp normalize(v) when is_binary(v), do: v |> String.trim() |> String.trim_trailing(".")
+  defp normalize(v), do: to_string(v)
+end