Phase 2 saga engine: compensation-based runner + step contract

The architectural spine of every write workflow phase 2+ — provisioning,
suspension, offboarding, updates, rollback — all ride this engine. Each
becomes a step list, not new orchestration code.

Schemas:
- saga_runs           — kind, status (pending/running/completed/failed/
                        compensating/rolled_back), step_modules, current
                        step idx, accumulating context (jsonb), cancel
                        flag, error.
- saga_step_results   — per-step audit ledger: status (running/completed/
                        failed/compensated), output, attempts, timings,
                        unique (saga_id, step_idx).
- cloud_provisioned   — desired-state for resources WE provisioned. spec
                        + spec_version + saga_id; FK to cloud_resources.
                        Phase 2's drift-detection diff lands here.

Step contract (ArcadiaCloud.Provisioning.Step):
- execute(state) -> {:ok, state} | {:error, reason}
- compensate(state) -> :ok | {:error, _}   (optional)
- name() -> String.t()

SagaState carries the live execution state — accumulating context,
immutable inputs, current step_idx. Helpers: get_output/put_output
(context r/w), get_input (inputs read-only).

Runner (Oban worker, queue: provisioning, max_attempts: 1):
- kick_off: pending -> running, run_step(0)
- run_step: idempotent re-entry on saga.current_step_idx; persists step
  result + saga context after each step; recursive forward walk through
  the whole step list within one perform/1 call.
- safe_execute: try/rescue/catch around module.execute so a raised
  exception triggers compensation rather than blowing up the worker.
- start_compensation: status=compensating, walk from idx-1 down to 0,
  calling compensate/1 where it's exported; logs but doesn't halt on
  compensate failure (best-effort + audit log).
- cancellation: checked between steps; cancel_requested=true -> trigger
  compensation from current idx.
- crash recovery: max_attempts: 1 + run_step keyed on
  saga.current_step_idx means Oban requeue picks up at the right place,
  but full crash-resume infra is deferred to phase 2.5 (manual re-enqueue
  works for now).

Two proof-of-concept steps (Steps.Echo, Steps.Fail) demonstrate the
engine without any DO API exposure. First real DO write step lands in
the next chunk.

Provisioning context provides start_saga/1, list_sagas/1,
list_step_results/1, cancel_saga/1, upsert_step_result/3.

Live smoke verified end-to-end:
- [Echo, Echo, Echo] happy path: all 3 completed, context accumulated
  echoed_at_step_0/1/2 = "hello".
- [Echo, Echo, Fail] failure path: step 2 failed, compensation walked
  back through step 1 then step 0; final status rolled_back with error
  {compensate_from_idx: 1, reason: "step_failed:fail"}; ledger shows
  echo/echo/fail with statuses compensated/compensated/failed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

lib/arcadia_cloud/provisioning/saga_state.ex

@@ -0,0 +1,65 @@
+defmodule ArcadiaCloud.Provisioning.SagaState do
+  @moduledoc """
+  Carries the live saga context across step calls. Steps receive this
+  struct, mutate it via the helpers, return a new value.
+
+  `context` is the accumulating bag of created-resource IDs and other
+  per-saga state. Outputs from a step (e.g. `droplet_id` after
+  CreateDroplet) live here so later steps and compensation can find
+  what to act on.
+
+  `inputs` are the immutable arguments the saga was started with
+  (e.g. `template_id`, `deployment_slug`). Steps may read these but
+  never write to them.
+  """
+
+  alias ArcadiaCloud.Provisioning.SagaRun
+
+  defstruct [:saga_id, :saga, :context, :inputs, :step_idx]
+
+  @type t :: %__MODULE__{
+          saga_id: binary(),
+          saga: SagaRun.t() | nil,
+          context: map(),
+          inputs: map(),
+          step_idx: non_neg_integer() | nil
+        }
+
+  def from(%SagaRun{} = saga) do
+    %__MODULE__{
+      saga_id: saga.id,
+      saga: saga,
+      context: saga.context || %{},
+      inputs: (saga.context || %{})["__inputs__"] || %{},
+      step_idx: saga.current_step_idx
+    }
+  end
+
+  @doc """
+  Idempotency hook: was a value written by a prior attempt of this same step?
+  Returns the value or nil.
+  """
+  def get_output(%__MODULE__{context: ctx}, key) when is_atom(key) do
+    get_output(%__MODULE__{context: ctx}, Atom.to_string(key))
+  end
+
+  def get_output(%__MODULE__{context: ctx}, key) when is_binary(key) do
+    Map.get(ctx, key)
+  end
+
+  def put_output(%__MODULE__{} = state, key, value) when is_atom(key) do
+    put_output(state, Atom.to_string(key), value)
+  end
+
+  def put_output(%__MODULE__{context: ctx} = state, key, value) when is_binary(key) do
+    %{state | context: Map.put(ctx, key, value)}
+  end
+
+  def get_input(%__MODULE__{inputs: inputs}, key) when is_atom(key) do
+    Map.get(inputs, Atom.to_string(key))
+  end
+
+  def get_input(%__MODULE__{inputs: inputs}, key) when is_binary(key) do
+    Map.get(inputs, key)
+  end
+end