diff --git a/cmk-data-transfer/.env.example b/cmk-data-transfer/.env.example
index faf3894..3e3ed4b 100644
--- a/cmk-data-transfer/.env.example
+++ b/cmk-data-transfer/.env.example
@@ -1,116 +1,71 @@
-# ============================================================================
-# cmk-data-transfer — configuration
-# Copy to .env and fill in.  .env is git-ignored; never commit real creds.
-#   cp .env.example .env
-# ============================================================================
+# --- Source S3 ---
+SRC_S3_ACCESS_KEY_ID=
+SRC_S3_SECRET_ACCESS_KEY=
+SRC_S3_ENDPOINT=           # e.g. https://s3.us-east-1.amazonaws.com
+SRC_S3_REGION=             # e.g. us-east-1 (optional, depends on provider)
+SRC_S3_BUCKET=
+SRC_S3_PREFIX=             # optional: only copy objects under this prefix
 
-# --- OCI Object Storage credentials (S3-compatible "Customer Secret Key") ---
-# Create under: OCI Console > Profile > Customer Secret Keys.
-# These are the S3-compat access/secret pair, NOT your OCI API signing key.
-OCI_ACCESS_KEY_ID=
-OCI_SECRET_ACCESS_KEY=
+# --- Destination type ---
+# disk : write to a VAST-backed RWX shared disk (default)
+# s3   : write to an S3-compatible destination (e.g. Crusoe S3)
+DEST_TYPE=disk
 
-# --- OCI Object Storage location ---
-# NAMESPACE is the Object Storage namespace (Console > Tenancy details, or
-#   `oci os ns get`). REQUIRED to build the S3-compat endpoint.
-OCI_NAMESPACE=
+# --- Destination S3 (only when DEST_TYPE=s3) ---
+DEST_S3_ACCESS_KEY_ID=
+DEST_S3_SECRET_ACCESS_KEY=
+DEST_S3_ENDPOINT=          # e.g. https://s3.crusoe.ai
+DEST_S3_REGION=
+DEST_S3_BUCKET=
+DEST_S3_PREFIX=
 
-# REGION is the OCI region slug where the BUCKET lives, e.g. us-phoenix-1 /
-# us-ashburn-1. Must be a valid OCI region (not your destination/cloud region);
-# preflight warns if the slug doesn't look like OCI.
-OCI_REGION=us-phoenix-1
-
-# Source bucket and optional prefix (no leading slash on prefix).
-OCI_BUCKET=
-OCI_PREFIX=
-
-# Derived endpoint (leave blank to auto-build from NAMESPACE+REGION):
-#   https://<NAMESPACE>.compat.objectstorage.<REGION>.oraclecloud.com
-OCI_ENDPOINT=
-
-# --- Destination (VAST RWX shared disk) ---
+# --- Destination disk (only when DEST_TYPE=disk) ---
+DEST_PATH=/data/dataset
+DEST_MODE=dynamic           # dynamic | import | nfs
 PVC_NAME=cmk-data-transfer-fs
 PVC_SIZE=1000Ti
 STORAGE_CLASS=crusoe-csi-driver-fs-sc
-# DEST_PATH = WHERE objects are saved on the disk. The disk mounts at /data in
-# every pod, so this is the folder under it (e.g. /data/my-dataset). For an
-# existing disk (DEST_MODE=import/nfs), point it at where on that disk you want
-# the files; rclone copy is idempotent, so an existing partial folder is fine.
-DEST_PATH=/data/dataset
-
-# --- Destination mode: dynamic | import | nfs ---
-#   dynamic : provision a new shared disk via the fs CSI driver (default).
-#   import  : bind an EXISTING disk via a CSI static PV (id+name+serial below).
-#   nfs     : bind an EXISTING disk via an in-tree NFS PV straight to the VAST
-#             DNS endpoint, bypassing the CSI driver. Use this where CSI mounts
-#             time out on an unroutable fallback IP (disk returns no data-path
-#             connectivity fields). Needs only EXISTING_DISK_ID.
-DEST_MODE=dynamic
 
-# Existing-disk identity (find with `crusoe storage disks list -f json`):
-#   import needs all three; nfs needs only EXISTING_DISK_ID.
-# A static PV is created with reclaimPolicy=Retain (your disk is never deleted);
-# PVC_SIZE becomes the PV capacity; STORAGE_CLASS is ignored.
-EXISTING_DISK_ID=                # crusoe disk `id`  (nfs path = /volumes/<id>; CSI volumeHandle)
-EXISTING_DISK_NAME=              # crusoe disk `name`   -> csi.crusoe.ai/disk-name   (import only)
-EXISTING_DISK_SERIAL=            # crusoe disk `serial_number`                       (import only)
+# --- Import existing disk (DEST_MODE=import) ---
+EXISTING_DISK_ID=
+EXISTING_DISK_NAME=
+EXISTING_DISK_SERIAL=
 EXISTING_DISK_FSTYPE=ext4
 
-# NFS-mode endpoint (the VAST data path on CMK):
+# --- NFS mode (DEST_MODE=nfs) ---
 NFS_SERVER=nfs.crusoecloudcompute.com
-NFS_EXPORT_PATH=                 # blank => /volumes/<EXISTING_DISK_ID>
+NFS_EXPORT_PATH=
 NFS_MOUNT_OPTIONS=vers=3,nconnect=16,spread_reads,spread_writes,remoteports=dns
 
-# --- Fleet sizing ---
-# TARGET_GBPS is the single sizing knob — set to your target aggregate GB/s.
-TARGET_GBPS=30
-NUM_NODES=4                       # nodes to spread across
-PODS_PER_NODE=8                   # workers per node (spread evenly). More pods =
-                                  # smaller shards = shorter straggler tail. Raise
-                                  # for shorter tails; if each pod uses many
-                                  # streams, watch node memory.
-NUM_PODS=                         # absolute total override; default = NODES*PER_NODE
+# --- Sizing ---
+TARGET_GBPS=30.0
+NUM_NODES=4
+PODS_PER_NODE=8
+# NUM_PODS=               # absolute override
 INSTANCE_CLASS=s2a
-
-# Node hardware (defaults = s2a.80x). Set for other SKUs/clouds so the sizing
-# model uses the right CPU split, NIC line rate, and BDP.
 NODE_VCPU=80
 NODE_RAM_GIB=676
 NODE_NIC_GBPS=200
-
-# Path RTT (ms) source<->destination region. Drives the BDP / concurrency math.
-RTT_MS=150
-# Conservative per-stream throughput assumption (Mbps) for an untuned
-# intercontinental TCP stream. Lower => more streams provisioned. Tune via sweep.
-PER_STREAM_MBPS=250
-# Safety multiplier on the BDP-derived stream count.
+RTT_MS=150.0
+PER_STREAM_MBPS=250.0
 STREAM_SAFETY=1.5
 
-# --- rclone concurrency (blank => auto-derive per-pod from the BDP sizing) ---
-# Leave TRANSFERS/STREAMS blank to size from TARGET_GBPS + RTT. To pin them, the
-# Crusoe AWS-S3 reference used transfers=40, streams=40, no --checkers (strong
-# throughput + low latency on large files):
-#   RCLONE_TRANSFERS=40 ; RCLONE_MULTI_THREAD_STREAMS=40
-RCLONE_TRANSFERS=                 # parallel files per pod (blank = auto)
-RCLONE_MULTI_THREAD_STREAMS=      # ranged-GET streams per file >cutoff (blank = auto)
+# --- rclone (blank = auto-derive from sizing) ---
+# RCLONE_TRANSFERS=
+# RCLONE_MULTI_THREAD_STREAMS=
 RCLONE_MULTI_THREAD_CUTOFF=256M
-RCLONE_MULTI_THREAD_CHUNK_SIZE=32M  # smaller chunk = shorter per-file straggler tail
-RCLONE_CHECKERS=                  # blank => omit --checkers (rclone default; with
-                                  # --no-traverse there's little to check + it
-                                  # avoids extra HEAD latency)
+RCLONE_MULTI_THREAD_CHUNK_SIZE=32M
+# RCLONE_CHECKERS=
 RCLONE_BUFFER_SIZE=32M
 RCLONE_S3_CHUNK_SIZE=64M
-RCLONE_EXTRA_FLAGS=               # e.g. "--low-level-retries 20 --timeout 120s"
+# RCLONE_EXTRA_FLAGS=
 
-# --- Worker pod resource requests (blank => auto-derived; the auto memory
-# estimate is conservative for high stream counts, which can over-restrict how
-# many pods schedule — set an explicit value when running 40/40). Requests
-# affect scheduling only; with no limits set, actual use can exceed the request.
-WORKER_CPU_REQUEST=
-WORKER_MEM_REQUEST=16Gi
+# --- Worker resources (blank = auto-derive) ---
+# WORKER_CPU_REQUEST=
+# WORKER_MEM_REQUEST=
 
-# --- K8s plumbing ---
+# --- K8s ---
 NAMESPACE=default
-SECRET_NAME=cmk-data-transfer-oci
+SECRET_NAME=cmk-data-transfer-s3
 RCLONE_IMAGE=rclone/rclone:latest
-KUBECONFIG=
+# KUBECONFIG=
diff --git a/cmk-data-transfer/.gitignore b/cmk-data-transfer/.gitignore
index dc43630..5cce0b7 100644
--- a/cmk-data-transfer/.gitignore
+++ b/cmk-data-transfer/.gitignore
@@ -37,3 +37,4 @@ venv/
 .vscode/
 *.swp
 .claude/
+.superpowers/
diff --git a/cmk-data-transfer/Makefile b/cmk-data-transfer/Makefile
index 3b877d3..8e82e65 100644
--- a/cmk-data-transfer/Makefile
+++ b/cmk-data-transfer/Makefile
@@ -17,7 +17,7 @@ sizing: ## Print the BDP/target sizing plan (no cluster access)
 dry-run: ## Render manifests + run preflight checks, launch nothing
 	$(PY) -m orchestrator.run --dry-run
 
-preflight: ## VAST write-ceiling test (fio, one pod per node) — safe, no egress
+preflight: ## Disk write-ceiling test (fio, one pod per node) — safe, no egress
 	$(PY) preflight/run_fio.py
 
 fio-bench: ## Disk read+write benchmark: fio on every s2a node -> bench/results/
@@ -45,7 +45,7 @@ clean: ## Delete worker + master + fio pods (keep PVC/Secret/data)
 	-kubectl -n $(NAMESPACE) delete pod cmk-data-transfer-master --ignore-not-found
 
 clean-all: clean ## Also delete the Secret (keeps the PVC and downloaded data)
-	-kubectl -n $(NAMESPACE) delete secret cmk-data-transfer-oci --ignore-not-found
+	-kubectl -n $(NAMESPACE) delete secret cmk-data-transfer-s3 --ignore-not-found
 	@echo "PVC and downloaded data left intact. Delete the PVC manually if desired."
 
 # --- defaults for sweep knobs (override on CLI) ---
diff --git a/cmk-data-transfer/README.md b/cmk-data-transfer/README.md
index 50204d2..773d612 100644
--- a/cmk-data-transfer/README.md
+++ b/cmk-data-transfer/README.md
@@ -1,11 +1,13 @@
 # CMK Cross-Region Object Storage Data Transfer
 
-Parallel-pull a dataset from **any S3-compatible object store** to a **VAST-backed
-RWX shared disk** on **Crusoe Managed Kubernetes (CMK)**, tuned to saturate worker
-hosts across a high-latency path. **OCI Object Storage is the worked example**, but
-the source backend is rclone's generic `provider = Other`, so AWS S3, MinIO/Ceph,
-GCS (S3 interop), Cloudflare R2, Backblaze B2, etc. work too — see
-[Other S3-compatible sources](#other-s3-compatible-sources).
+Parallel-pull a dataset from **any S3-compatible object store** to a **Crusoe shared filesystem
+RWX shared disk** or **S3-compatible destination** on **Crusoe Managed Kubernetes
+(CMK)**, tuned to saturate worker hosts across a high-latency path. **AWS S3 is the
+worked example**, but the source backend is rclone's generic `provider = Other`, so
+OCI Object Storage, MinIO/Ceph, GCS (S3 interop), Cloudflare R2, Backblaze B2,
+etc. work too — see [Other S3-compatible sources](#other-s3-compatible-sources).
+Set `DEST_TYPE=s3` to write directly to Crusoe S3 (or any S3-compatible
+destination) instead of a shared disk.
 
 A **master pod** lists the source and writes balanced shard manifests to the shared
 disk; **N worker pods** each `rclone copy` one shard in parallel. It follows Crusoe's
@@ -18,17 +20,36 @@ concurrency**.
 ## Quick Start
 
 ```bash
-cp .env.example .env    # fill in creds, namespace, region, bucket, destination disk
+cp .env.example .env    # fill in creds, namespace, bucket, and destination
 make sizing             # (optional) print the concurrency plan — no cluster access
 make dry-run            # render manifests + preflight checks (launches nothing)
 make preflight          # (optional) fio write-ceiling test (safe, no egress)
 make run                # full pipeline; PROMPTS before the large transfer
 ```
 
-> **Where does the data land?** The shared disk mounts at **`/data`** in every pod,
-> and objects copy to **`DEST_PATH`** (default `/data/dataset`) — set it in `.env`.
-> **Using an existing disk?** Set `DEST_MODE=import` (or `nfs`) + the disk id, then
-> point `DEST_PATH` at where on that disk you want the files. See
+**Disk destination** (default — writes to a Crusoe shared filesystem RWX shared disk):
+
+```bash
+# In .env:
+DEST_TYPE=disk          # or omit (disk is the default)
+DEST_PATH=/data/dataset
+```
+
+**S3 destination** (copies directly to Crusoe S3 or any S3-compatible store):
+
+```bash
+# In .env:
+DEST_TYPE=s3
+DEST_S3_ENDPOINT=https://s3.crusoe.ai
+DEST_S3_BUCKET=my-dest-bucket
+DEST_S3_ACCESS_KEY_ID=...
+DEST_S3_SECRET_ACCESS_KEY=...
+```
+
+> **Disk mode — where does the data land?** The shared disk mounts at **`/data`** in
+> every pod, and objects copy to **`DEST_PATH`** (default `/data/dataset`) — set it in
+> `.env`. **Using an existing disk?** Set `DEST_MODE=import` (or `nfs`) + the disk id,
+> then point `DEST_PATH` at where on that disk you want the files. See
 > [Destination modes](#destination-modes-dest_mode).
 
 ---
@@ -41,7 +62,7 @@ make run                # full pipeline; PROMPTS before the large transfer
         │  list (via master) → bin-pack shards locally → push to shared disk
         ▼
 ┌──────────────────────────── Crusoe Managed Kubernetes ───────────────────────┐
-│   Secret(rclone.conf, RO)        PVC: shared disk (VAST, RWX)                 │
+│   Secret(rclone.conf, RO)        PVC: shared disk (Crusoe shared filesystem, RWX)                 │
 │        ▼                                   ▼                                  │
 │   ┌─────────────┐   rclone lsf      ┌─────────────────────────────────────┐  │
 │   │ master pod  │ ───────────────►  │  shared disk  /data                 │  │
@@ -56,8 +77,14 @@ make run                # full pipeline; PROMPTS before the large transfer
 └─────────┼──────────────┼────────────────────┼──────────────┼──────────────────┘
           ▼              ▼                    ▼              ▼  hundreds of parallel
         ┌──────────────────────────────────────────────────┐   ranged-GET streams
-        │   S3-compatible object store (e.g. OCI)            │   high RTT (intercontinental)
+        │   S3-compatible source (AWS S3, OCI, R2, …)       │   high RTT (intercontinental)
         └──────────────────────────────────────────────────┘
+                                                              DEST_TYPE=s3: workers
+                                                              also write to an
+                                                              S3-compatible destination
+                                                              (e.g. Crusoe S3) using a
+                                                              small 1 TiB coordination
+                                                              PVC (shards + logs only)
 ```
 
 The fleet runs **`PODS_PER_NODE` independent rclone processes per node** — multiple
@@ -128,34 +155,34 @@ All inputs come from `.env` (copied from `.env.example`), process env, or CLI fl
 
 | Variable | Meaning |
 |---|---|
-| `OCI_ACCESS_KEY_ID` / `OCI_SECRET_ACCESS_KEY` | S3-compat access/secret key pair (OCI: "Customer Secret Key") |
-| `OCI_NAMESPACE` / `OCI_REGION` | OCI Object Storage namespace + region (auto-build the endpoint) |
-| `OCI_BUCKET` / `OCI_PREFIX` | source bucket and optional prefix |
-| `OCI_ENDPOINT` | explicit S3 endpoint (for non-OCI stores; blank = auto from namespace+region) |
+| `SRC_S3_ACCESS_KEY_ID` / `SRC_S3_SECRET_ACCESS_KEY` | S3-compat access/secret key pair for the source |
+| `SRC_S3_ENDPOINT` | source S3 endpoint URL (e.g. `https://s3.us-east-1.amazonaws.com`) |
+| `SRC_S3_REGION` | source region (optional, depends on provider) |
+| `SRC_S3_BUCKET` / `SRC_S3_PREFIX` | source bucket and optional prefix |
+| `DEST_TYPE` | `disk` (default) or `s3` — destination type |
+| `DEST_S3_ACCESS_KEY_ID` / `DEST_S3_SECRET_ACCESS_KEY` | S3-compat credentials for the destination (when `DEST_TYPE=s3`) |
+| `DEST_S3_ENDPOINT` / `DEST_S3_REGION` | destination S3 endpoint + region (when `DEST_TYPE=s3`) |
+| `DEST_S3_BUCKET` / `DEST_S3_PREFIX` | destination bucket and optional prefix (when `DEST_TYPE=s3`) |
 | `TARGET_GBPS` | **the single sizing knob** (default 30) |
 | `NUM_NODES` / `PODS_PER_NODE` | fleet size; total = nodes × pods/node (`NUM_PODS` forces an absolute total) |
 | `INSTANCE_CLASS` | node class for the nodeSelector (default `s2a`) |
 | `NODE_VCPU` / `NODE_RAM_GIB` / `NODE_NIC_GBPS` | per-node hardware for the sizing model (defaults = s2a.80x) |
 | `RTT_MS` / `PER_STREAM_MBPS` / `STREAM_SAFETY` | BDP model inputs |
-| `DEST_PATH` | **where objects are saved** — folder under the `/data` mount (default `/data/dataset`) |
-| `DEST_MODE` | `dynamic` / `import` / `nfs` destination (see below) |
-| `STORAGE_CLASS` / `PVC_NAME` / `PVC_SIZE` | destination shared disk |
+| `DEST_PATH` | **where objects are saved** — folder under the `/data` mount (default `/data/dataset`, disk mode only) |
+| `DEST_MODE` | `dynamic` / `import` / `nfs` destination (disk mode only; see below) |
+| `STORAGE_CLASS` / `PVC_NAME` / `PVC_SIZE` | destination shared disk (disk mode only) |
 | `EXISTING_DISK_ID` / `_NAME` / `_SERIAL` / `NFS_SERVER` | bind an existing disk (import/nfs modes) |
 | `RCLONE_*` / `WORKER_MEM_REQUEST` | per-flag / resource overrides (blank = auto-derive) |
 
-> **Region note:** use the **OCI** region slug where the *bucket* lives (not your
-> destination region). Preflight warns if it doesn't look like an OCI slug.
-
 ### Other S3-compatible sources
 
 The source backend is rclone `provider = Other`, so any S3-compatible store works.
-The `OCI_*` names are just labels — for a non-OCI store, set `OCI_ENDPOINT` to its S3
-endpoint and use its access/secret key + bucket:
+Set `SRC_S3_ENDPOINT` to its S3 endpoint and use its access/secret key + bucket:
 
-| Source | `OCI_ENDPOINT` |
+| Source | `SRC_S3_ENDPOINT` |
 |---|---|
-| OCI Object Storage | *(auto from namespace + region)* |
 | AWS S3 | `https://s3.<region>.amazonaws.com` (or blank with a real AWS key) |
+| OCI Object Storage | `https://<namespace>.compat.objectstorage.<region>.oraclecloud.com` |
 | MinIO / Ceph RGW | `https://<host>:<port>` |
 | Google Cloud Storage | `https://storage.googleapis.com` (S3 interop + HMAC key) |
 | Cloudflare R2 | `https://<account>.r2.cloudflarestorage.com` |
@@ -176,14 +203,16 @@ lands in the repo, in pod args, or in shell history. `.gitignore` blocks `.env`,
 
 1. **Sizing + preflight** — prints the concurrency plan; verifies `kubectl`,
    enough schedulable nodes (`INSTANCE_CLASS`), and the StorageClass (creates it if
-   absent).
-2. **Secret** — builds `rclone.conf`, applies it in-cluster.
-3. **PVC + master pod** — applies the RWX claim and the master.
+   absent, disk mode only).
+2. **Secret** — builds `rclone.conf` (source + optional dest remote), applies it in-cluster.
+3. **PVC + master pod** — provisions the coordination PVC (large RWX disk for
+   `DEST_TYPE=disk`; small 1 TiB for `DEST_TYPE=s3`) and launches the master pod.
 4. **List + shard** — `rclone lsf` in the master → pull the listing → bin-pack (LPT)
    into N balanced shard files → push to `/data/shards/`.
 5. **Confirm** — prompts before the large transfer (skip with `--yes`).
 6. **Launch** — `PODS_PER_NODE` workers per node (pinned), each `rclone copy
-   --files-from shard-i.txt --no-traverse …`.
+   --files-from shard-i.txt --no-traverse …` (to disk or S3 dest depending on
+   `DEST_TYPE`).
 7. **Monitor → teardown** — polls to completion, then deletes worker + master pods
    (keeps PVC, Secret, data) unless `--keep`.
 
@@ -198,7 +227,7 @@ only fetches what's missing.
 |---|---|---|---|
 | `dynamic` (default) | provisions a new disk via the fs CSI driver | — | greenfield |
 | `import` | binds an **existing** disk via a CSI **static PV** | disk `id` + `name` + `serial` | migrate into an existing disk, CSI healthy |
-| `nfs` | binds an **existing** disk via an **in-tree NFS PV** to the VAST DNS endpoint (bypasses CSI) | disk `id` (+ `NFS_SERVER`) | when a CSI mount times out on an unroutable fallback IP |
+| `nfs` | binds an **existing** disk via an **in-tree NFS PV** to the Crusoe shared filesystem DNS endpoint (bypasses CSI) | disk `id` (+ `NFS_SERVER`) | when a CSI mount times out on an unroutable fallback IP |
 
 Find the disk with `crusoe storage disks list -f json` (pick the `shared-volume` in
 your region): `.id`, `.name`, `.serial_number`. All modes use `reclaimPolicy: Retain`,
@@ -217,13 +246,45 @@ Because `rclone copy` is idempotent, `DEST_PATH` can point at a folder already h
 part of the dataset — only missing/changed objects are pulled.
 
 > **`nfs` mode** exists because the fs CSI driver can fall back to a fixed IP that may
-> be unroutable from your nodepool (mounts hang). The VAST DNS endpoint
+> be unroutable from your nodepool (mounts hang). The Crusoe shared filesystem DNS endpoint
 > (`nfs.crusoecloudcompute.com`, `remoteports=dns`) mounts cleanly; `nfs` mode creates
 > an in-tree NFS PV straight to it (see `k8s/nfs-pv.yaml`). If a CSI mount hangs,
 > prefer `nfs`. Export path defaults to `/volumes/<EXISTING_DISK_ID>`.
 
 ---
 
+## S3-to-S3 mode (`DEST_TYPE=s3`)
+
+Set `DEST_TYPE=s3` to copy directly from a source S3-compatible store to a destination
+S3-compatible store (e.g. Crusoe S3), bypassing the shared disk entirely. Workers run
+`rclone copy src:bucket dest:bucket` instead of writing to the mounted PVC.
+
+A **small 1 TiB coordination PVC** is still provisioned to hold shard manifests and
+worker logs — no large data disk is needed.
+
+```bash
+SRC_S3_ACCESS_KEY_ID=<src-key>         \
+SRC_S3_SECRET_ACCESS_KEY=<src-secret>  \
+SRC_S3_ENDPOINT=https://s3.us-east-1.amazonaws.com \
+SRC_S3_BUCKET=my-source-bucket         \
+DEST_TYPE=s3                            \
+DEST_S3_ACCESS_KEY_ID=<dest-key>       \
+DEST_S3_SECRET_ACCESS_KEY=<dest-secret> \
+DEST_S3_ENDPOINT=https://s3.crusoe.ai  \
+DEST_S3_BUCKET=my-dest-bucket          \
+make run
+```
+
+Or place all variables in `.env` and just run `make run`. The `DEST_S3_*` credentials
+are kept in the same Kubernetes Secret as the source credentials, assembled into a
+single `rclone.conf` with `[src]` and `[dest]` remotes.
+
+> **Note:** `DEST_MODE`, `STORAGE_CLASS`, `PVC_SIZE`, and disk-specific settings are
+> ignored when `DEST_TYPE=s3`. Use `RCLONE_EXTRA_FLAGS` as an escape hatch for
+> destination-specific rclone options.
+
+---
+
 ## Benchmarking
 
 - **`make preflight`** — fio write-ceiling, one pod per node, no egress. If the
@@ -289,26 +350,6 @@ kubectl get po -l app=cmk-data-transfer-worker --field-selector=status.phase=Run
 
 ---
 
-## Alternative: native OCI backend
-
-Instead of the S3-compat path you can use rclone's native `oracleobjectstorage`
-backend (OCI IAM auth, no access/secret key):
-
-```ini
-[oci-native]
-type = oracleobjectstorage
-namespace = <namespace>
-region = <region>
-provider = user_principal_auth      # or instance_principal_auth / resource_principal_auth
-config_file = /root/.oci/config
-config_profile = DEFAULT
-```
-
-Mount your `~/.oci` config + key into the pods and point the remote at `oci-native:`.
-The S3-compat path is the default because it matches the credential most users hold.
-
----
-
 ## Repo layout
 
 ```
diff --git a/cmk-data-transfer/bench/fio/fio-bench-job.yaml b/cmk-data-transfer/bench/fio/fio-bench-job.yaml
index e3d37db..aedafde 100644
--- a/cmk-data-transfer/bench/fio/fio-bench-job.yaml
+++ b/cmk-data-transfer/bench/fio/fio-bench-job.yaml
@@ -1,7 +1,7 @@
-# Disk benchmark: one fio pod per s2a node, against the shared VAST PVC.
+# Disk benchmark: one fio pod per s2a node, against the shared filesystem PVC.
 #
 # Runs four stonewalled (sequential) fio profiles against a single per-node file
-# so reads measure data actually on VAST (direct=1 bypasses the page cache):
+# so reads measure data actually on disk (direct=1 bypasses the page cache):
 #   seqwrite  (1M)  -> sequential write bandwidth
 #   seqread   (1M)  -> sequential read bandwidth (re-reads the written file)
 #   randwrite (4k)  -> random write IOPS
diff --git a/cmk-data-transfer/bench/fio/run_fio_bench.py b/cmk-data-transfer/bench/fio/run_fio_bench.py
index 079180a..3a1ba2d 100644
--- a/cmk-data-transfer/bench/fio/run_fio_bench.py
+++ b/cmk-data-transfer/bench/fio/run_fio_bench.py
@@ -10,8 +10,8 @@
         [--size 10G] [--jobs 4] [--direct 1] [--instance-class s2a]
         [--namespace default] [--image alpine:3.20] [--timeout 1200]
 
-Safe: writes only under /data/_fiobench/<node>/ on the PVC and cleans up. No OCI
-egress (VAST I/O is in-region).
+Safe: writes only under /data/_fiobench/<node>/ on the PVC and cleans up. No
+egress (disk I/O is in-region).
 """
 from __future__ import annotations
 
diff --git a/cmk-data-transfer/bench/oci_experiment.py b/cmk-data-transfer/bench/oci_experiment.py
index f262520..af9cbfd 100644
--- a/cmk-data-transfer/bench/oci_experiment.py
+++ b/cmk-data-transfer/bench/oci_experiment.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""OCI -> VAST transfer experiment: measure sustained pull rate at a given
+"""S3 transfer experiment: measure sustained pull rate at a given
 node/pod concurrency, writing to a DISTINCT destination subfolder per run.
 
 Drives the orchestrator (secret -> PVC/master -> list -> shard -> workers) for a
@@ -8,7 +8,7 @@
   - bytes transferred + wall-clock elapsed
   - average GB/s   (= bytes / elapsed)  <- the headline "how fast did we pull"
   - peak windowed GB/s (steady-state)
-  - per-node GB/s, and any OCI throttling (429/503 SlowDown) seen in logs
+  - per-node GB/s, and any S3 throttling (429/503 SlowDown) seen in logs
 Time-series + summary saved to bench/results/oci-exp-<label>-<ts>/ (git-ignored).
 
 Each run writes to DEST_PATH=/data/dataset/<label> so it re-pulls the full data
@@ -34,7 +34,7 @@
 from orchestrator import manifests, run as orun          # noqa: E402
 from orchestrator.config import load_config               # noqa: E402
 from orchestrator.k8s import Kubectl                       # noqa: E402
-from orchestrator.rclone_conf import build_s3_compat_conf  # noqa: E402
+from orchestrator.rclone_conf import build_rclone_conf  # noqa: E402
 from orchestrator.sizing import compute_sizing             # noqa: E402
 from bench import collect                                  # noqa: E402
 
@@ -78,7 +78,7 @@ def main(argv=None) -> int:
     cfg.rclone_multi_thread_streams = args.mts
     cfg.dest_path = f"{cfg.mount_root}/dataset/{args.label}"   # DISTINCT per run
     if args.prefix_override is not None:
-        cfg.prefix = args.prefix_override
+        cfg.src_s3_prefix = args.prefix_override
     errs = cfg.validate()
     if errs:
         for e in errs:
@@ -88,10 +88,10 @@ def main(argv=None) -> int:
     sizing = compute_sizing(cfg)
     ns = cfg.k8s_namespace
     kc = Kubectl(namespace=ns)
-    run_dir = tempfile.mkdtemp(prefix=f"oci-exp-{args.label}-")
+    run_dir = tempfile.mkdtemp(prefix=f"s3-exp-{args.label}-")
 
-    print(f"=== OCI experiment: {args.label} ===")
-    print(f"  source : {cfg.remote_root()}")
+    print(f"=== S3 transfer experiment: {args.label} ===")
+    print(f"  source : {cfg.source_remote_root()}")
     print(f"  dest   : {cfg.dest_path}  (PVC {cfg.pvc_name}, nfs)")
     print(f"  fleet  : {cfg.num_nodes} nodes x {sizing.pods_per_node} pods "
           f"= {sizing.total_pods} workers; per-pod T={sizing.transfers} "
@@ -101,7 +101,7 @@ def main(argv=None) -> int:
     # --- setup (reuse orchestrator steps) ---
     orun.preflight(cfg, kc)
     kc.create_secret_from_files(cfg.secret_name,
-                                {"rclone.conf": build_s3_compat_conf(cfg)})
+                                {"rclone.conf": build_rclone_conf(cfg)})
     if cfg.is_nfs():
         kc.apply(manifests.nfs_pv(cfg))
     elif cfg.is_import():
@@ -305,7 +305,7 @@ def fleet_nic_rx(pods):
     print(f"  NIC     avg/steady/peak: {nic_avg/GB:.2f} / {nic_steady_med/GB:.2f} / "
           f"{peak_nic/GB:.2f} GB/s  <- GROUND TRUTH "
           f"({nic_avg/GB/cfg.num_nodes:.2f} GB/s/node)")
-    print(f"  OCI throttle signals: {throttle}")
+    print(f"  S3 throttle signals: {throttle}")
     print(f"  -> {out}/summary.json , timeseries.csv")
 
     if not args.keep:
diff --git a/cmk-data-transfer/bench/sweep.py b/cmk-data-transfer/bench/sweep.py
index 71311b6..cf79402 100644
--- a/cmk-data-transfer/bench/sweep.py
+++ b/cmk-data-transfer/bench/sweep.py
@@ -5,9 +5,9 @@
 against a BOUNDED sample of the source, records the throughput curve via
 collect.py, tears the fleet down, and appends peak/median aggregate GB/s to a
 results CSV. The goal is to find the saturation knee and identify the binding
-constraint (network vs VAST write ceiling vs OCI request throttling).
+constraint (network vs disk write ceiling vs S3 request throttling).
 
-COST WARNING: each grid point re-downloads the sample (OCI egress is billed).
+COST WARNING: each grid point re-downloads the sample (source egress is billed).
 Keep --sample-prefix small and --max-seconds short. Requires --yes.
 
 Prereqs: run the main orchestrator once (or this with --do-list) so the Secret,
@@ -37,7 +37,7 @@
 from orchestrator import manifests                      # noqa: E402
 from orchestrator.config import load_config             # noqa: E402
 from orchestrator.k8s import Kubectl                     # noqa: E402
-from orchestrator.rclone_conf import build_s3_compat_conf  # noqa: E402
+from orchestrator.rclone_conf import build_rclone_conf  # noqa: E402
 from orchestrator.sizing import compute_sizing           # noqa: E402
 
 from bench import collect  # noqa: E402
@@ -50,7 +50,7 @@ def _grid(s: str) -> list[int]:
 def ensure_base(cfg, kc: Kubectl, do_list: bool) -> None:
     """Make sure Secret / PVC / master pod exist; optionally (re)list+shard."""
     kc.create_secret_from_files(
-        cfg.secret_name, {"rclone.conf": build_s3_compat_conf(cfg)})
+        cfg.secret_name, {"rclone.conf": build_rclone_conf(cfg)})
     kc.apply(manifests.pvc(cfg))
     if not kc.exists("pod", cfg.master_pod_name):
         kc.apply(manifests.master_pod(cfg))
@@ -59,7 +59,7 @@ def ensure_base(cfg, kc: Kubectl, do_list: bool) -> None:
         listing = f"{cfg.mount_root}/listing.tsv"
         lsf = ("SEP=$(printf '\\t'); rclone lsf --recursive --files-only "
                "--format sp --separator \"$SEP\" --config /config/rclone.conf "
-               f"{cfg.remote_root()} > {listing}")
+               f"{cfg.source_remote_root()} > {listing}")
         kc.exec(cfg.master_pod_name, ["/bin/sh", "-c", lsf])
 
 
@@ -128,7 +128,7 @@ def main(argv=None) -> int:
     p.add_argument("--pods-per-node-grid", default="2,4,8",
                    help="sweep workers-per-node (the new throughput lever)")
     p.add_argument("--sample-prefix", default=None,
-                   help="override OCI_PREFIX with a small sample subdir")
+                   help="override SRC_S3_PREFIX with a small sample subdir")
     p.add_argument("--max-seconds", type=int, default=120)
     p.add_argument("--out", default="bench/results/sweep.csv")
     p.add_argument("--do-list", action="store_true",
@@ -138,10 +138,10 @@ def main(argv=None) -> int:
     args, _unknown = p.parse_known_args(argv)
 
     if not args.yes:
-        print("Refusing to run without --yes (each point bills OCI egress).")
+        print("Refusing to run without --yes (each point bills source egress).")
         return 2
     if args.sample_prefix is not None:
-        cfg.prefix = args.sample_prefix
+        cfg.src_s3_prefix = args.sample_prefix
 
     import tempfile
     run_dir = tempfile.mkdtemp(prefix="cmk-sweep-")
diff --git a/cmk-data-transfer/k8s/master-pod.yaml b/cmk-data-transfer/k8s/master-pod.yaml
index 7679f4f..ed740d3 100644
--- a/cmk-data-transfer/k8s/master-pod.yaml
+++ b/cmk-data-transfer/k8s/master-pod.yaml
@@ -1,7 +1,7 @@
 # Master / distribution pod: mounts the shared disk and the rclone Secret.
 # The orchestrator execs `rclone lsf` here to list the source, then pushes the
 # computed shard files into ${MOUNT_ROOT}/shards. Pinned to the same instance
-# class as the workers so its egress path to OCI matches theirs.
+# class as the workers so its egress path to the source matches theirs.
 apiVersion: v1
 kind: Pod
 metadata:
@@ -34,7 +34,7 @@ spec:
         claimName: ${PVC_NAME}
     - name: rclone-conf
       secret:
-        secretName: ${SECRET_NAME}                  # cmk-data-transfer-oci
+        secretName: ${SECRET_NAME}                  # cmk-data-transfer-s3
         defaultMode: 0400
         items:
           - key: rclone.conf
diff --git a/cmk-data-transfer/k8s/nfs-pv.yaml b/cmk-data-transfer/k8s/nfs-pv.yaml
index 62026c4..e4cde24 100644
--- a/cmk-data-transfer/k8s/nfs-pv.yaml
+++ b/cmk-data-transfer/k8s/nfs-pv.yaml
@@ -1,7 +1,7 @@
-# Bind an existing VAST shared disk via an in-tree NFS PV (bypasses the CSI
+# Bind an existing Crusoe shared disk via an in-tree NFS PV (bypasses the CSI
 # driver). Use this where the fs CSI driver's mount times out on an unroutable
-# fallback IP (the disk returns no "data path connectivity fields"): the VAST
-# DNS endpoint resolves to the in-VPC data IPs and mounts cleanly with
+# fallback IP (the disk returns no "data path connectivity fields"): the Crusoe
+# shared filesystem DNS endpoint resolves to the in-VPC data IPs and mounts cleanly with
 # remoteports=dns.
 #
 # The orchestrator generates the authoritative versions when DEST_MODE=nfs; this
@@ -23,7 +23,7 @@ spec:
   persistentVolumeReclaimPolicy: Retain
   storageClassName: ""
   volumeMode: Filesystem
-  mountOptions:                          # VAST multipath NFS over DNS
+  mountOptions:                          # multipath NFS over DNS
     - vers=3
     - nconnect=16
     - spread_reads
diff --git a/cmk-data-transfer/k8s/pvc.yaml b/cmk-data-transfer/k8s/pvc.yaml
index e4b7991..6713b04 100644
--- a/cmk-data-transfer/k8s/pvc.yaml
+++ b/cmk-data-transfer/k8s/pvc.yaml
@@ -8,7 +8,7 @@ metadata:
 spec:
   accessModes:
     - ReadWriteMany
-  storageClassName: ${STORAGE_CLASS}   # crusoe-csi-driver-fs-sc (VAST)
+  storageClassName: ${STORAGE_CLASS}   # crusoe-csi-driver-fs-sc
   resources:
     requests:
       storage: ${PVC_SIZE}    # e.g. 1000Ti
diff --git a/cmk-data-transfer/k8s/secret.example.yaml b/cmk-data-transfer/k8s/secret.example.yaml
index 353d110..f2810b7 100644
--- a/cmk-data-transfer/k8s/secret.example.yaml
+++ b/cmk-data-transfer/k8s/secret.example.yaml
@@ -5,33 +5,44 @@
 # The single key `rclone.conf` is mounted read-only at /config/rclone.conf.
 #
 # Build it safely from a local rclone.conf without echoing creds into argv:
-#   kubectl create secret generic cmk-data-transfer-oci \
+#   kubectl create secret generic cmk-data-transfer-s3 \
 #       --from-file=rclone.conf=./rclone.conf \
 #       --dry-run=client -o yaml | kubectl apply -f -
 #
 # ...where ./rclone.conf (git-ignored) contains:
-#   [oci]
+#   [src]
 #   type = s3
 #   provider = Other
-#   access_key_id = <OCI_ACCESS_KEY_ID>
-#   secret_access_key = <OCI_SECRET_ACCESS_KEY>
-#   endpoint = https://<NAMESPACE>.compat.objectstorage.<REGION>.oraclecloud.com
+#   access_key_id = <SRC_S3_ACCESS_KEY_ID>
+#   secret_access_key = <SRC_S3_SECRET_ACCESS_KEY>
+#   endpoint = https://s3.<region>.amazonaws.com
+#   region = <REGION>
+#   force_path_style = true
+#   no_check_bucket = true
+#
+#   # (only when DEST_TYPE=s3)
+#   [dest]
+#   type = s3
+#   provider = Other
+#   access_key_id = <DEST_S3_ACCESS_KEY_ID>
+#   secret_access_key = <DEST_S3_SECRET_ACCESS_KEY>
+#   endpoint = https://s3.crusoe.ai
 #   region = <REGION>
 #   force_path_style = true
 #   no_check_bucket = true
 apiVersion: v1
 kind: Secret
 metadata:
-  name: cmk-data-transfer-oci
+  name: cmk-data-transfer-s3
 type: Opaque
 stringData:
   rclone.conf: |
-    [oci]
+    [src]
     type = s3
     provider = Other
     access_key_id = REDACTED
     secret_access_key = REDACTED
-    endpoint = https://NAMESPACE.compat.objectstorage.REGION.oraclecloud.com
-    region = REGION
+    endpoint = https://s3.us-east-1.amazonaws.com
+    region = us-east-1
     force_path_style = true
     no_check_bucket = true
diff --git a/cmk-data-transfer/k8s/storageclass.yaml b/cmk-data-transfer/k8s/storageclass.yaml
index d4aff4f..d4e7588 100644
--- a/cmk-data-transfer/k8s/storageclass.yaml
+++ b/cmk-data-transfer/k8s/storageclass.yaml
@@ -1,4 +1,4 @@
-# Crusoe VAST shared-filesystem (ReadWriteMany) StorageClass.
+# Crusoe shared-filesystem (ReadWriteMany) StorageClass.
 #
 # This cluster ships the fs CSI driver (fs.csi.crusoe.ai) but does not always
 # create the StorageClass object. The orchestrator creates this automatically
diff --git a/cmk-data-transfer/k8s/worker-pod.yaml b/cmk-data-transfer/k8s/worker-pod.yaml
index 4638f6d..c5c8dd3 100644
--- a/cmk-data-transfer/k8s/worker-pod.yaml
+++ b/cmk-data-transfer/k8s/worker-pod.yaml
@@ -45,7 +45,7 @@ spec:
           mkdir -p /data/dataset /data/logs;
           rclone copy
           --files-from /data/shards/shard-${SHARD_INDEX}.txt
-          oci:${OCI_BUCKET}/${OCI_PREFIX} /data/dataset
+          src:${SRC_S3_BUCKET}/${SRC_S3_PREFIX} /data/dataset
           --config /config/rclone.conf
           --transfers ${RCLONE_TRANSFERS}
           --multi-thread-streams ${RCLONE_MULTI_THREAD_STREAMS}
diff --git a/cmk-data-transfer/orchestrator/__init__.py b/cmk-data-transfer/orchestrator/__init__.py
index b4b3392..b70af96 100644
--- a/cmk-data-transfer/orchestrator/__init__.py
+++ b/cmk-data-transfer/orchestrator/__init__.py
@@ -1,8 +1,8 @@
 """cmk-data-transfer orchestrator.
 
-Parallel-pulls a dataset from OCI Object Storage (S3-compatible endpoint) to a
-VAST-backed ReadWriteMany shared disk on Crusoe Managed Kubernetes, engineered
-to saturate s2a worker hosts over a high-RTT intercontinental path.
+Parallel data transfer from any S3-compatible source to a Crusoe shared
+filesystem or S3-compatible destination on Crusoe Managed Kubernetes, engineered to
+saturate worker hosts over a high-RTT intercontinental path.
 """
 
 __version__ = "0.1.0"
diff --git a/cmk-data-transfer/orchestrator/config.py b/cmk-data-transfer/orchestrator/config.py
index 7b63731..900ed64 100644
--- a/cmk-data-transfer/orchestrator/config.py
+++ b/cmk-data-transfer/orchestrator/config.py
@@ -8,7 +8,6 @@
 
 import argparse
 import os
-import re
 from dataclasses import dataclass, field, fields
 from pathlib import Path
 from typing import Optional
@@ -21,9 +20,6 @@
 S2A_NIC_GBPS = 200
 
 
-KNOWN_OCI_REGION_RE = re.compile(r"^[a-z]{2,3}-[a-z]+-\d+$")  # e.g. us-phoenix-1
-
-
 def _load_dotenv(path: Path) -> dict[str, str]:
     """Minimal .env parser (no dependency on python-dotenv)."""
     out: dict[str, str] = {}
@@ -54,14 +50,26 @@ def _load_dotenv(path: Path) -> dict[str, str]:
 
 @dataclass
 class Config:
-    # --- OCI source ---
-    access_key_id: str = ""
-    secret_access_key: str = ""
-    namespace: str = ""
-    region: str = "us-phoenix-1"
-    bucket: str = ""
-    prefix: str = ""
-    endpoint: str = ""
+    # --- Source S3 ---
+    src_s3_access_key_id: str = ""
+    src_s3_secret_access_key: str = ""
+    src_s3_endpoint: str = ""
+    src_s3_region: str = ""
+    src_s3_bucket: str = ""
+    src_s3_prefix: str = ""
+
+    # --- Destination type ---
+    #   disk : write to a Crusoe shared filesystem (RWX, default)
+    #   s3   : write to an S3-compatible destination (e.g. Crusoe S3)
+    dest_type: str = "disk"
+
+    # --- Destination S3 (only when dest_type=s3) ---
+    dest_s3_access_key_id: str = ""
+    dest_s3_secret_access_key: str = ""
+    dest_s3_endpoint: str = ""
+    dest_s3_region: str = ""
+    dest_s3_bucket: str = ""
+    dest_s3_prefix: str = ""
 
     # --- Destination ---
     pvc_name: str = "cmk-data-transfer-fs"
@@ -83,8 +91,8 @@ class Config:
     # --- Destination mode ---
     #   dynamic : provision a new disk via the fs CSI driver (default)
     #   import  : bind an existing disk via a CSI static PV (disk id/name/serial)
-    #   nfs     : bind an existing disk via an in-tree NFS PV straight to the VAST
-    #             DNS endpoint, bypassing the CSI driver. Use this where CSI mounts
+    #   nfs     : bind an existing disk via an in-tree NFS PV straight to the
+    #             Crusoe shared filesystem DNS endpoint, bypassing the CSI driver. Use this where CSI mounts
     #             time out on an unroutable fallback IP (disk returns no data-path
     #             connectivity fields).
     dest_mode: str = "dynamic"
@@ -123,7 +131,7 @@ class Config:
 
     # --- K8s plumbing ---
     k8s_namespace: str = "default"
-    secret_name: str = "cmk-data-transfer-oci"
+    secret_name: str = "cmk-data-transfer-s3"
     rclone_image: str = "rclone/rclone:latest"
     master_pod_name: str = "cmk-data-transfer-master"
     mount_root: str = "/data"  # shared-disk (PVC) mount path in all pods
@@ -141,19 +149,14 @@ def effective_pods_per_node(self) -> int:
             return max(1, -(-self.num_pods // self.num_nodes))  # ceil div
         return max(1, self.pods_per_node)
 
-    def effective_endpoint(self) -> str:
-        if self.endpoint:
-            return self.endpoint
-        return (
-            f"https://{self.namespace}.compat.objectstorage."
-            f"{self.region}.oraclecloud.com"
-        )
-
-    def remote_root(self) -> str:
-        """`oci:bucket[/prefix]` for rclone."""
-        root = f"oci:{self.bucket}"
-        if self.prefix:
-            root += "/" + self.prefix.strip("/")
+    def effective_source_endpoint(self) -> str:
+        return self.src_s3_endpoint
+
+    def source_remote_root(self) -> str:
+        """`src:bucket[/prefix]` for rclone."""
+        root = f"src:{self.src_s3_bucket}"
+        if self.src_s3_prefix:
+            root += "/" + self.src_s3_prefix.strip("/")
         return root
 
     def is_import(self) -> bool:
@@ -171,53 +174,69 @@ def static_pv_name(self) -> str:
     def nfs_path(self) -> str:
         return self.nfs_export_path or f"/volumes/{self.existing_disk_id}"
 
+    def is_s3_dest(self) -> bool:
+        """Write directly to an S3-compatible destination."""
+        return self.dest_type == "s3"
+
+    def dest_remote_root(self) -> str:
+        """`dest:bucket[/prefix]` for rclone."""
+        root = f"dest:{self.dest_s3_bucket}"
+        if self.dest_s3_prefix:
+            root += "/" + self.dest_s3_prefix.strip("/")
+        return root
+
     # ----------------------------------------------------------------- validate
     def validate(self, require_secrets: bool = True) -> list[str]:
         errs: list[str] = []
         if require_secrets:
-            if not self.access_key_id:
-                errs.append("OCI_ACCESS_KEY_ID is required")
-            if not self.secret_access_key:
-                errs.append("OCI_SECRET_ACCESS_KEY is required")
-        if not self.namespace and not self.endpoint:
-            errs.append("OCI_NAMESPACE (or explicit OCI_ENDPOINT) is required")
-        if not self.bucket:
-            errs.append("OCI_BUCKET is required")
+            if not self.src_s3_access_key_id:
+                errs.append("SRC_S3_ACCESS_KEY_ID is required")
+            if not self.src_s3_secret_access_key:
+                errs.append("SRC_S3_SECRET_ACCESS_KEY is required")
+        if not self.src_s3_endpoint:
+            errs.append("SRC_S3_ENDPOINT is required")
+        if not self.src_s3_bucket:
+            errs.append("SRC_S3_BUCKET is required")
         if self.num_nodes < 1:
             errs.append("NUM_NODES must be >= 1")
         if self.target_gbps <= 0:
             errs.append("TARGET_GBPS must be > 0")
         if self.pods_per_node < 1:
             errs.append("PODS_PER_NODE must be >= 1")
-        if self.dest_mode not in ("dynamic", "import", "nfs"):
-            errs.append(f"DEST_MODE must be dynamic|import|nfs (got "
-                        f"'{self.dest_mode}')")
-        if self.is_import() and not (self.existing_disk_id
-                                     and self.existing_disk_name
-                                     and self.existing_disk_serial):
-            errs.append(
-                "import mode needs EXISTING_DISK_ID, EXISTING_DISK_NAME, and "
-                "EXISTING_DISK_SERIAL together (from `crusoe storage disks "
-                "list -f json`: the disk's id, name, and serial_number)."
-            )
-        if self.is_nfs():
-            if not self.existing_disk_id and not self.nfs_export_path:
-                errs.append("nfs mode needs EXISTING_DISK_ID (export path "
-                            "/volumes/<id>) or an explicit NFS_EXPORT_PATH")
-            if not self.nfs_server:
-                errs.append("nfs mode needs NFS_SERVER")
+        # --- destination validation ---
+        if self.dest_type not in ("disk", "s3"):
+            errs.append(f"DEST_TYPE must be disk|s3 (got '{self.dest_type}')")
+        if self.is_s3_dest():
+            if require_secrets:
+                if not self.dest_s3_access_key_id:
+                    errs.append("DEST_S3_ACCESS_KEY_ID is required when DEST_TYPE=s3")
+                if not self.dest_s3_secret_access_key:
+                    errs.append("DEST_S3_SECRET_ACCESS_KEY is required when DEST_TYPE=s3")
+            if not self.dest_s3_endpoint:
+                errs.append("DEST_S3_ENDPOINT is required when DEST_TYPE=s3")
+            if not self.dest_s3_bucket:
+                errs.append("DEST_S3_BUCKET is required when DEST_TYPE=s3")
+        else:
+            # disk destination validation (existing logic)
+            if self.dest_mode not in ("dynamic", "import", "nfs"):
+                errs.append(f"DEST_MODE must be dynamic|import|nfs (got "
+                            f"'{self.dest_mode}')")
+            if self.is_import() and not (self.existing_disk_id
+                                         and self.existing_disk_name
+                                         and self.existing_disk_serial):
+                errs.append(
+                    "import mode needs EXISTING_DISK_ID, EXISTING_DISK_NAME, and "
+                    "EXISTING_DISK_SERIAL together (from `crusoe storage disks "
+                    "list -f json`: the disk's id, name, and serial_number)."
+                )
+            if self.is_nfs():
+                if not self.existing_disk_id and not self.nfs_export_path:
+                    errs.append("nfs mode needs EXISTING_DISK_ID (export path "
+                                "/volumes/<id>) or an explicit NFS_EXPORT_PATH")
+                if not self.nfs_server:
+                    errs.append("nfs mode needs NFS_SERVER")
         return errs
 
-    def warnings(self) -> list[str]:
-        warns: list[str] = []
-        if not self.endpoint and not KNOWN_OCI_REGION_RE.match(self.region):
-            warns.append(
-                f"OCI_REGION='{self.region}' does not look like an OCI region "
-                "slug (expected e.g. us-phoenix-1 / us-ashburn-1). Use the OCI "
-                "region where the bucket lives, not the destination/cloud region."
-            )
-        return warns
-
 
 def _coerce(name: str, raw: str):
     """Coerce a string env value to the dataclass field type."""
@@ -234,13 +253,19 @@ def _coerce(name: str, raw: str):
 
 # env var name -> Config field name
 _ENV_MAP = {
-    "OCI_ACCESS_KEY_ID": "access_key_id",
-    "OCI_SECRET_ACCESS_KEY": "secret_access_key",
-    "OCI_NAMESPACE": "namespace",
-    "OCI_REGION": "region",
-    "OCI_BUCKET": "bucket",
-    "OCI_PREFIX": "prefix",
-    "OCI_ENDPOINT": "endpoint",
+    "SRC_S3_ACCESS_KEY_ID": "src_s3_access_key_id",
+    "SRC_S3_SECRET_ACCESS_KEY": "src_s3_secret_access_key",
+    "SRC_S3_ENDPOINT": "src_s3_endpoint",
+    "SRC_S3_REGION": "src_s3_region",
+    "SRC_S3_BUCKET": "src_s3_bucket",
+    "SRC_S3_PREFIX": "src_s3_prefix",
+    "DEST_TYPE": "dest_type",
+    "DEST_S3_ACCESS_KEY_ID": "dest_s3_access_key_id",
+    "DEST_S3_SECRET_ACCESS_KEY": "dest_s3_secret_access_key",
+    "DEST_S3_ENDPOINT": "dest_s3_endpoint",
+    "DEST_S3_REGION": "dest_s3_region",
+    "DEST_S3_BUCKET": "dest_s3_bucket",
+    "DEST_S3_PREFIX": "dest_s3_prefix",
     "PVC_NAME": "pvc_name",
     "PVC_SIZE": "pvc_size",
     "STORAGE_CLASS": "storage_class",
@@ -311,10 +336,10 @@ def load_config(argv: Optional[list[str]] = None):
         "num_nodes": args.num_nodes,
         "pods_per_node": args.pods_per_node,
         "num_pods": args.num_pods,
-        "bucket": args.bucket,
-        "prefix": args.prefix,
-        "namespace": args.namespace,
-        "region": args.region,
+        "src_s3_bucket": args.src_bucket,
+        "src_s3_prefix": args.src_prefix,
+        "src_s3_endpoint": args.src_endpoint,
+        "src_s3_region": args.src_region,
         "rtt_ms": args.rtt_ms,
         "per_stream_mbps": args.per_stream_mbps,
         "rclone_transfers": args.transfers,
@@ -326,6 +351,10 @@ def load_config(argv: Optional[list[str]] = None):
         "existing_disk_name": args.import_disk_name,
         "existing_disk_serial": args.import_disk_serial,
         "dest_mode": args.dest_mode,
+        "dest_type": args.dest_type,
+        "dest_s3_bucket": args.dest_s3_bucket,
+        "dest_s3_prefix": args.dest_s3_prefix,
+        "dest_s3_endpoint": args.dest_s3_endpoint,
         "nfs_server": args.nfs_server,
     }
     for field_name, val in cli_map.items():
@@ -340,7 +369,7 @@ def load_config(argv: Optional[list[str]] = None):
 
 def build_arg_parser() -> argparse.ArgumentParser:
     p = argparse.ArgumentParser(
-        description="Parallel OCI->VAST pull on Crusoe CMK, tuned for s2a.",
+        description="Parallel S3->S3/disk pull on Crusoe CMK, tuned for high-RTT.",
     )
     p.add_argument("--env-file", help="Path to .env (default: ./.env)")
     p.add_argument("--target-gbps", type=float, help="Single sizing knob")
@@ -349,10 +378,10 @@ def build_arg_parser() -> argparse.ArgumentParser:
                    help="worker pods per node (spread evenly)")
     p.add_argument("--num-pods", type=int,
                    help="absolute total pods override (spread across nodes)")
-    p.add_argument("--bucket")
-    p.add_argument("--prefix")
-    p.add_argument("--namespace", help="OCI Object Storage namespace")
-    p.add_argument("--region", help="OCI region slug")
+    p.add_argument("--src-bucket", help="Source S3 bucket")
+    p.add_argument("--src-prefix", help="Source S3 prefix")
+    p.add_argument("--src-endpoint", help="Source S3 endpoint URL")
+    p.add_argument("--src-region", help="Source S3 region")
     p.add_argument("--rtt-ms", type=float)
     p.add_argument("--per-stream-mbps", type=float)
     p.add_argument("--transfers", type=int, help="Override rclone --transfers")
@@ -370,7 +399,12 @@ def build_arg_parser() -> argparse.ArgumentParser:
                    help="Existing disk `serial_number`")
     p.add_argument("--dest-mode", choices=("dynamic", "import", "nfs"),
                    help="destination: provision / CSI import / in-tree NFS PV")
-    p.add_argument("--nfs-server", help="VAST NFS DNS endpoint (nfs mode)")
+    p.add_argument("--dest-type", choices=("disk", "s3"),
+                   help="destination type: disk (Crusoe shared filesystem) or s3")
+    p.add_argument("--dest-s3-bucket", help="Destination S3 bucket (DEST_TYPE=s3)")
+    p.add_argument("--dest-s3-prefix", help="Destination S3 prefix (DEST_TYPE=s3)")
+    p.add_argument("--dest-s3-endpoint", help="Destination S3 endpoint (DEST_TYPE=s3)")
+    p.add_argument("--nfs-server", help="Crusoe shared filesystem NFS DNS endpoint (nfs mode)")
     p.add_argument("--dry-run", action="store_true",
                    help="Render plan + manifests, do not apply to the cluster")
     p.add_argument("--keep", action="store_true",
diff --git a/cmk-data-transfer/orchestrator/manifests.py b/cmk-data-transfer/orchestrator/manifests.py
index 53d0580..f0ae273 100644
--- a/cmk-data-transfer/orchestrator/manifests.py
+++ b/cmk-data-transfer/orchestrator/manifests.py
@@ -28,7 +28,7 @@ def log_dir(cfg: Config) -> str:
 
 
 def storage_class(name: str) -> dict:
-    """Crusoe VAST RWX storage class (created only if absent)."""
+    """Crusoe shared filesystem RWX storage class (created only if absent)."""
     return {
         "apiVersion": "storage.k8s.io/v1",
         "kind": "StorageClass",
@@ -73,12 +73,12 @@ def import_pv(cfg: Config) -> dict:
 
 
 def nfs_pv(cfg: Config) -> dict:
-    """Static IN-TREE NFS PV bound to an existing VAST volume via the DNS
+    """Static IN-TREE NFS PV bound to an existing Crusoe shared volume via the DNS
     endpoint — bypasses the CSI driver.
 
     Why this exists: where the fs CSI driver falls back to an unroutable IP
     (the disk returns no data-path connectivity fields) and mounts time out, the
-    VAST DNS endpoint resolves to the in-VPC data IPs and mounts cleanly with
+    Crusoe shared filesystem DNS endpoint resolves to the in-VPC data IPs and mounts cleanly with
     remoteports=dns. reclaimPolicy Retain so the disk/data is never deleted.
     """
     return {
@@ -99,16 +99,22 @@ def nfs_pv(cfg: Config) -> dict:
 
 
 def pvc(cfg: Config) -> dict:
+    # S3 dest: small coordination-only PVC for shards + logs
+    # Crusoe fs CSI minimum disk size is 1 Ti; use that for S3 coordination
+    pvc_size = "1Ti" if cfg.is_s3_dest() else cfg.pvc_size
+
     spec = {
         "accessModes": ["ReadWriteMany"],
-        "resources": {"requests": {"storage": cfg.pvc_size}},
+        "resources": {"requests": {"storage": pvc_size}},
         "volumeMode": "Filesystem",
     }
-    if cfg.is_import() or cfg.is_nfs():
-        # bind to the static PV; empty class disables dynamic provisioning
+    if not cfg.is_s3_dest() and (cfg.is_import() or cfg.is_nfs()):
         spec["storageClassName"] = ""
         spec["volumeName"] = cfg.static_pv_name()
+    elif not cfg.is_s3_dest():
+        spec["storageClassName"] = cfg.storage_class
     else:
+        # S3 dest: always dynamic provisioning for the small coordination disk
         spec["storageClassName"] = cfg.storage_class
     return {
         "apiVersion": "v1",
@@ -132,13 +138,13 @@ def _rclone_conf_volume(cfg: Config) -> dict:
 
 
 def master_pod(cfg: Config) -> dict:
-    """Lightweight pod that mounts the shared disk; lists + shards run here.
-
-    Pinned to the worker instance class so its in-cluster egress path to OCI
-    matches the workers' (relevant for the listing pass).
-    """
-    init = (f"mkdir -p {shard_dir(cfg)} {log_dir(cfg)} {cfg.dest_path}; "
-            "while true; do sleep 3600; done")
+    """Lightweight pod that mounts the shared disk; lists + shards run here."""
+    if cfg.is_s3_dest():
+        init = (f"mkdir -p {shard_dir(cfg)} {log_dir(cfg)}; "
+                "while true; do sleep 3600; done")
+    else:
+        init = (f"mkdir -p {shard_dir(cfg)} {log_dir(cfg)} {cfg.dest_path}; "
+                "while true; do sleep 3600; done")
     return {
         "apiVersion": "v1",
         "kind": "Pod",
@@ -176,10 +182,16 @@ def _worker_command(cfg: Config, sizing: Sizing) -> list[str]:
     shard_file = f"{shard_dir(cfg)}/shard-${{SHARD_INDEX}}.txt"
     log_file = f"{log_dir(cfg)}/worker-${{SHARD_INDEX}}.log"
 
+    # destination: S3 remote or local disk path
+    if cfg.is_s3_dest():
+        dest_target = cfg.dest_remote_root()
+    else:
+        dest_target = cfg.dest_path
+
     rclone = [
         "rclone", "copy",
         "--files-from", shard_file,
-        cfg.remote_root(), cfg.dest_path,
+        cfg.source_remote_root(), dest_target,
         "--config", "/config/rclone.conf",
         "--transfers", str(sizing.transfers),
         "--multi-thread-streams", str(sizing.multi_thread_streams),
@@ -206,9 +218,12 @@ def _worker_command(cfg: Config, sizing: Sizing) -> list[str]:
         rclone += cfg.rclone_extra_flags.split()
 
     rclone_str = " ".join(rclone)
+    if cfg.is_s3_dest():
+        setup = f"mkdir -p {log_dir(cfg)}"
+    else:
+        setup = f"mkdir -p {cfg.dest_path} {log_dir(cfg)}"
     script = (
-        "set -eu; "
-        f"mkdir -p {cfg.dest_path} {log_dir(cfg)}; "
+        f"set -eu; {setup}; "
         'echo "[worker ${SHARD_INDEX}] $(date -u) starting"; '
         f"{rclone_str} 2>&1 | tee {log_file}"
     )
diff --git a/cmk-data-transfer/orchestrator/rclone_conf.py b/cmk-data-transfer/orchestrator/rclone_conf.py
index a518eb3..9a3c5ca 100644
--- a/cmk-data-transfer/orchestrator/rclone_conf.py
+++ b/cmk-data-transfer/orchestrator/rclone_conf.py
@@ -1,7 +1,4 @@
-"""Build the rclone.conf body for the OCI remote.
-
-Primary path: S3-compatible backend pointed at OCI's S3-compat endpoint, using
-the customer's Access Key / Secret Key (OCI "Customer Secret Key"). 
+"""Build the rclone.conf body for S3-compatible remotes.
 
 The generated config is placed ONLY into a Kubernetes Secret (see k8s.py) and
 mounted read-only into pods. It is never written to the repo or committed.
@@ -11,48 +8,44 @@
 from .config import Config
 
 
-def build_s3_compat_conf(cfg: Config) -> str:
-    """rclone remote named `oci` using the s3 backend against OCI S3-compat.
+def build_rclone_conf(cfg: Config) -> str:
+    """Build rclone.conf with a source [src] remote, and optionally a [dest]
+    remote when DEST_TYPE=s3.
 
-    provider=Other + force_path_style is the generic S3-compatible profile;
-    OCI's S3-compat endpoint is path-style and validates region in the v4
-    signature, so we set region explicitly.
+    provider=Other + force_path_style is the generic S3-compatible profile.
     """
-    endpoint = cfg.effective_endpoint()
-    return "\n".join(
-        [
-            "[oci]",
+    endpoint = cfg.effective_source_endpoint()
+    lines = [
+        "[src]",
+        "type = s3",
+        "provider = Other",
+        f"access_key_id = {cfg.src_s3_access_key_id}",
+        f"secret_access_key = {cfg.src_s3_secret_access_key}",
+        f"endpoint = {endpoint}",
+    ]
+    if cfg.src_s3_region:
+        lines.append(f"region = {cfg.src_s3_region}")
+    lines += [
+        "force_path_style = true",
+        "no_check_bucket = true",
+        "",
+    ]
+
+    if cfg.is_s3_dest():
+        lines += [
+            "[dest]",
             "type = s3",
             "provider = Other",
-            f"access_key_id = {cfg.access_key_id}",
-            f"secret_access_key = {cfg.secret_access_key}",
-            f"endpoint = {endpoint}",
-            f"region = {cfg.region}",
+            f"access_key_id = {cfg.dest_s3_access_key_id}",
+            f"secret_access_key = {cfg.dest_s3_secret_access_key}",
+            f"endpoint = {cfg.dest_s3_endpoint}",
+        ]
+        if cfg.dest_s3_region:
+            lines.append(f"region = {cfg.dest_s3_region}")
+        lines += [
             "force_path_style = true",
-            # OCI S3-compat does not implement bucket-create/HEAD-bucket the way
-            # AWS does; skip it so copy doesn't error on a read-only key.
             "no_check_bucket = true",
             "",
         ]
-    )
-
-
-def native_oracle_conf_example(cfg: Config) -> str:
-    """Documented ALTERNATIVE: native oracleobjectstorage backend.
 
-    Uses OCI IAM (user/instance/resource principal), NOT the S3 access/secret
-    key. Shown for completeness; not the default path.
-    """
-    return "\n".join(
-        [
-            "[oci-native]",
-            "type = oracleobjectstorage",
-            f"namespace = {cfg.namespace or '<namespace>'}",
-            f"region = {cfg.region}",
-            "provider = user_principal_auth",
-            "config_file = /root/.oci/config",
-            "config_profile = DEFAULT",
-            "# compartment = ocid1.compartment.oc1..<...>   # for bucket listing",
-            "",
-        ]
-    )
+    return "\n".join(lines)
diff --git a/cmk-data-transfer/orchestrator/run.py b/cmk-data-transfer/orchestrator/run.py
index 622792f..7aae7b0 100644
--- a/cmk-data-transfer/orchestrator/run.py
+++ b/cmk-data-transfer/orchestrator/run.py
@@ -4,7 +4,7 @@
   1. load config + compute sizing (BDP-grounded)               [always]
   2. preflight: kubectl reachable, enough s2a nodes, StorageClass present
   3. create K8s Secret (rclone.conf) — creds only live in-cluster
-  4. apply PVC (RWX VAST) + master pod
+  4. apply PVC (RWX shared filesystem) + master pod
   5. list source via `rclone lsf` (in master pod) -> /data/listing.tsv
   6. pull listing, shard locally (LPT bin-pack), push shard files back
   7. CONFIRM, then launch N pinned worker pods (the large transfer)
@@ -28,7 +28,7 @@
 from . import manifests  # noqa: E402
 from .config import load_config  # noqa: E402
 from .k8s import Kubectl, die  # noqa: E402
-from .rclone_conf import build_s3_compat_conf  # noqa: E402
+from .rclone_conf import build_rclone_conf  # noqa: E402
 from .sizing import compute_sizing  # noqa: E402
 
 
@@ -64,6 +64,18 @@ def preflight(cfg, kc: Kubectl) -> None:
         die(f"only {len(nodes)} schedulable {cfg.instance_class} nodes; "
             f"NUM_NODES={cfg.num_nodes}. Lower NUM_NODES or add nodes.")
 
+    if cfg.is_s3_dest():
+        print(f"  destination: S3 -> {cfg.dest_s3_endpoint} "
+              f"(coordination PVC only)")
+        # S3 dest still needs a StorageClass for the coordination PVC
+        if kc.cluster_exists("storageclass", cfg.storage_class):
+            print(f"  StorageClass {cfg.storage_class}: present")
+        else:
+            print(f"  StorageClass {cfg.storage_class}: ABSENT — creating "
+                  "(fs.csi.crusoe.ai)")
+            kc.apply_cluster(manifests.storage_class(cfg.storage_class))
+        return
+
     if cfg.is_nfs():
         print(f"  destination: NFS in-tree PV -> {cfg.nfs_server}:"
               f"{cfg.nfs_path()} (bypasses CSI; reclaimPolicy=Retain)")
@@ -92,10 +104,10 @@ def list_source(cfg, kc: Kubectl) -> None:
     lsf = (
         "SEP=$(printf '\\t'); "
         "rclone lsf --recursive --files-only --format sp --separator \"$SEP\" "
-        f"--config /config/rclone.conf {cfg.remote_root()} > {listing} && "
+        f"--config /config/rclone.conf {cfg.source_remote_root()} > {listing} && "
         f"wc -l {listing}"
     )
-    print(f"  {cfg.remote_root()}  ->  {listing}")
+    print(f"  {cfg.source_remote_root()}  ->  {listing}")
     res = kc.exec(cfg.master_pod_name, ["/bin/sh", "-c", lsf])
     if res.stdout:
         print("  " + res.stdout.decode(errors="replace").strip())
@@ -199,8 +211,6 @@ def main(argv: list[str] | None = None) -> int:
     cfg, args = load_config(argv)
 
     errs = cfg.validate(require_secrets=not args.dry_run)
-    for w in cfg.warnings():
-        print(f"WARNING: {w}", file=sys.stderr)
     if errs:
         for e in errs:
             print(f"ERROR: {e}", file=sys.stderr)
@@ -208,9 +218,12 @@ def main(argv: list[str] | None = None) -> int:
 
     sizing = compute_sizing(cfg)
     banner("Plan")
-    print(f"  source : {cfg.remote_root()}")
-    print(f"  via    : {cfg.effective_endpoint()}")
-    if cfg.is_nfs():
+    print(f"  source : {cfg.source_remote_root()}")
+    print(f"  via    : {cfg.effective_source_endpoint()}")
+    if cfg.is_s3_dest():
+        print(f"  dest   : {cfg.dest_remote_root()} (S3)")
+        print(f"  via    : {cfg.dest_s3_endpoint}")
+    elif cfg.is_nfs():
         print(f"  dest   : PVC {cfg.pvc_name} -> NFS {cfg.nfs_server}:"
               f"{cfg.nfs_path()} (in-tree PV, Retain) -> {cfg.dest_path}")
     elif cfg.is_import():
@@ -238,17 +251,19 @@ def main(argv: list[str] | None = None) -> int:
 
     banner("Creating Secret (rclone.conf) — creds stay in-cluster")
     kc.create_secret_from_files(
-        cfg.secret_name, {"rclone.conf": build_s3_compat_conf(cfg)})
+        cfg.secret_name, {"rclone.conf": build_rclone_conf(cfg)})
     print(f"  secret/{cfg.secret_name} applied")
 
     banner("Applying PVC + master pod")
-    if cfg.is_nfs():
-        kc.apply(manifests.nfs_pv(cfg))
-        print(f"  static NFS PV {cfg.static_pv_name()} -> {cfg.nfs_server}:"
-              f"{cfg.nfs_path()}")
-    elif cfg.is_import():
-        kc.apply(manifests.import_pv(cfg))
-        print(f"  static PV {cfg.static_pv_name()} -> disk {cfg.existing_disk_id}")
+    if not cfg.is_s3_dest():
+        if cfg.is_nfs():
+            kc.apply(manifests.nfs_pv(cfg))
+            print(f"  static NFS PV {cfg.static_pv_name()} -> {cfg.nfs_server}:"
+                  f"{cfg.nfs_path()}")
+        elif cfg.is_import():
+            kc.apply(manifests.import_pv(cfg))
+            print(f"  static PV {cfg.static_pv_name()} -> disk "
+                  f"{cfg.existing_disk_id}")
     kc.apply(manifests.pvc(cfg))
     kc.apply(manifests.master_pod(cfg))
     kc.wait_ready(cfg.master_pod_name)
@@ -278,7 +293,10 @@ def main(argv: list[str] | None = None) -> int:
         teardown(cfg, kc)
 
     banner("Done")
-    print(f"  dataset on PVC {cfg.pvc_name} at {cfg.dest_path}")
+    if cfg.is_s3_dest():
+        print(f"  data written to {cfg.dest_remote_root()}")
+    else:
+        print(f"  dataset on PVC {cfg.pvc_name} at {cfg.dest_path}")
     print(f"  per-worker logs: {manifests.log_dir(cfg)}/worker-<i>.log")
     return 0
 
@@ -292,12 +310,13 @@ def _render_generated(cfg, sizing) -> None:
         "master-pod.json": manifests.master_pod(cfg),
         "worker-pod-0.json": manifests.worker_pod(cfg, sizing, 0),
     }
-    if cfg.is_nfs():
-        docs["nfs-pv.json"] = manifests.nfs_pv(cfg)
-    elif cfg.is_import():
-        docs["import-pv.json"] = manifests.import_pv(cfg)
-    else:
-        docs["storageclass.json"] = manifests.storage_class(cfg.storage_class)
+    if not cfg.is_s3_dest():
+        if cfg.is_nfs():
+            docs["nfs-pv.json"] = manifests.nfs_pv(cfg)
+        elif cfg.is_import():
+            docs["import-pv.json"] = manifests.import_pv(cfg)
+        else:
+            docs["storageclass.json"] = manifests.storage_class(cfg.storage_class)
     for fname, doc in docs.items():
         with open(os.path.join(out, fname), "w") as fh:
             json.dump(doc, fh, indent=2)
diff --git a/cmk-data-transfer/preflight/fio-vast.yaml b/cmk-data-transfer/preflight/fio-vast.yaml
index 253750c..f1263fe 100644
--- a/cmk-data-transfer/preflight/fio-vast.yaml
+++ b/cmk-data-transfer/preflight/fio-vast.yaml
@@ -1,11 +1,11 @@
-# Reference fio pod for the VAST write-ceiling preflight (one per s2a node).
+# Reference fio pod for the shared-disk write-ceiling preflight (one per node).
 # preflight/run_fio.py templates N of these (anti-affinity => one per node),
 # runs a sequential write to the shared PVC, and sums the write bandwidth so a
-# later throughput plateau can be attributed to VAST vs network vs OCI.
+# later throughput plateau can be attributed to disk vs network vs source.
 #
 # Why this matters: if the download plateaus at, say, 18 GB/s and fio shows the
-# VAST mount also tops out near there, the binding constraint is the write path,
-# not OCI or the NIC. Run it BEFORE the big pull.
+# shared disk also tops out near there, the binding constraint is the write path,
+# not the source or the NIC. Run it BEFORE the big pull.
 apiVersion: v1
 kind: Pod
 metadata:
@@ -32,7 +32,7 @@ spec:
         - >
           apk add --no-cache fio >/dev/null &&
           mkdir -p /data/fio/$(hostname) &&
-          fio --name=vastwrite --directory=/data/fio/$(hostname)
+          fio --name=diskwrite --directory=/data/fio/$(hostname)
           --rw=write --bs=${FIO_BS} --size=${FIO_SIZE} --numjobs=${FIO_JOBS}
           --ioengine=libaio --iodepth=32 --direct=0 --group_reporting
           --output-format=json
diff --git a/cmk-data-transfer/preflight/run_fio.py b/cmk-data-transfer/preflight/run_fio.py
index 2d27cf6..62aa8a7 100644
--- a/cmk-data-transfer/preflight/run_fio.py
+++ b/cmk-data-transfer/preflight/run_fio.py
@@ -1,10 +1,10 @@
 #!/usr/bin/env python3
-"""VAST write-ceiling preflight.
+"""Shared-disk write-ceiling preflight.
 
 Launches one fio pod per s2a node (anti-affinity on hostname), each doing a
 sequential buffered write to the shared RWX PVC, then sums per-pod write
 bandwidth into an aggregate. This establishes the destination write ceiling so a
-later download plateau can be attributed correctly (VAST vs NIC vs OCI).
+later download plateau can be attributed correctly (disk vs NIC vs source).
 
 Safe to run before the big pull — it only writes to /data/fio and is cleaned up.
 
@@ -32,14 +32,14 @@ def kc(ns, *args, check=True, capture=True):
 
 
 def fio_pod(cfg, index, size, jobs, bs, direct):
-    # direct=1 (O_DIRECT) bypasses the page cache so we measure the real VAST
-    # write path, not RAM. end_fsync=1 forces a flush so the reported bandwidth
-    # reflects bytes durably on VAST. If the mount rejects O_DIRECT, rerun with
+    # direct=1 (O_DIRECT) bypasses the page cache so we measure the real shared
+    # disk write path, not RAM. end_fsync=1 forces a flush so the reported
+    # bandwidth reflects bytes durably on disk. If the mount rejects O_DIRECT, rerun with
     # --direct 0 (buffered) and a size well above node RAM.
     cmd = (
         "apk add --no-cache fio >/dev/null && "
         "mkdir -p /data/fio/$(hostname) && "
-        f"fio --name=vastwrite --directory=/data/fio/$(hostname) "
+        f"fio --name=diskwrite --directory=/data/fio/$(hostname) "
         f"--rw=write --bs={bs} --size={size} --numjobs={jobs} "
         f"--ioengine=libaio --iodepth=32 --direct={direct} --end_fsync=1 "
         "--group_reporting --output-format=json"
@@ -128,7 +128,7 @@ def main(argv=None) -> int:
         bw = parse_write_bw_bytes(log)
         total += bw
         print(f"  {name}: {bw/1e9:.2f} GB/s")
-    print(f"\nVAST aggregate write ceiling (this fleet): {total/1e9:.2f} GB/s")
+    print(f"\nDisk aggregate write ceiling (this fleet): {total/1e9:.2f} GB/s")
     print(f"  => download throughput cannot durably exceed this. Compare to "
           f"TARGET_GBPS={cfg.target_gbps}.")
 
diff --git a/cmk-data-transfer/tests/__init__.py b/cmk-data-transfer/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/cmk-data-transfer/tests/test_config.py b/cmk-data-transfer/tests/test_config.py
new file mode 100644
index 0000000..c7fdb34
--- /dev/null
+++ b/cmk-data-transfer/tests/test_config.py
@@ -0,0 +1,180 @@
+# tests/test_config.py
+"""Tests for orchestrator.config after OCI->SRC_S3 rename."""
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from orchestrator.config import Config
+
+
+def test_source_remote_root_bucket_only():
+    cfg = Config(src_s3_bucket="my-bucket")
+    assert cfg.source_remote_root() == "src:my-bucket"
+
+
+def test_source_remote_root_with_prefix():
+    cfg = Config(src_s3_bucket="my-bucket", src_s3_prefix="data/sub/")
+    assert cfg.source_remote_root() == "src:my-bucket/data/sub"
+
+
+def test_effective_source_endpoint_explicit():
+    cfg = Config(src_s3_endpoint="https://s3.us-east-1.amazonaws.com")
+    assert cfg.effective_source_endpoint() == "https://s3.us-east-1.amazonaws.com"
+
+
+def test_effective_source_endpoint_empty():
+    cfg = Config(src_s3_endpoint="")
+    # With no endpoint set, should return empty string (no OCI auto-build anymore)
+    assert cfg.effective_source_endpoint() == ""
+
+
+def test_validate_requires_src_s3_access_key():
+    cfg = Config(src_s3_access_key_id="", src_s3_secret_access_key="secret",
+                 src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="b")
+    errs = cfg.validate()
+    assert any("SRC_S3_ACCESS_KEY_ID" in e for e in errs)
+
+
+def test_validate_requires_src_s3_secret_key():
+    cfg = Config(src_s3_access_key_id="key", src_s3_secret_access_key="",
+                 src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="b")
+    errs = cfg.validate()
+    assert any("SRC_S3_SECRET_ACCESS_KEY" in e for e in errs)
+
+
+def test_validate_requires_src_s3_endpoint():
+    cfg = Config(src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+                 src_s3_endpoint="", src_s3_bucket="b")
+    errs = cfg.validate()
+    assert any("SRC_S3_ENDPOINT" in e for e in errs)
+
+
+def test_validate_requires_src_s3_bucket():
+    cfg = Config(src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+                 src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="")
+    errs = cfg.validate()
+    assert any("SRC_S3_BUCKET" in e for e in errs)
+
+
+def test_validate_passes_with_all_required():
+    cfg = Config(src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+                 src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="b")
+    errs = cfg.validate()
+    assert errs == []
+
+
+def test_env_map_has_src_s3_keys():
+    from orchestrator.config import _ENV_MAP
+    assert "SRC_S3_ACCESS_KEY_ID" in _ENV_MAP
+    assert "SRC_S3_SECRET_ACCESS_KEY" in _ENV_MAP
+    assert "SRC_S3_ENDPOINT" in _ENV_MAP
+    assert "SRC_S3_BUCKET" in _ENV_MAP
+    assert "SRC_S3_PREFIX" in _ENV_MAP
+    assert "SRC_S3_REGION" in _ENV_MAP
+    # OCI keys must be gone
+    assert "OCI_ACCESS_KEY_ID" not in _ENV_MAP
+    assert "OCI_NAMESPACE" not in _ENV_MAP
+
+
+def test_dest_type_defaults_to_disk():
+    cfg = Config()
+    assert cfg.dest_type == "disk"
+
+
+def test_is_s3_dest():
+    cfg = Config(dest_type="s3")
+    assert cfg.is_s3_dest()
+
+
+def test_is_not_s3_dest():
+    cfg = Config(dest_type="disk")
+    assert not cfg.is_s3_dest()
+
+
+def test_dest_remote_root_bucket_only():
+    cfg = Config(dest_type="s3", dest_s3_bucket="dest-bucket")
+    assert cfg.dest_remote_root() == "dest:dest-bucket"
+
+
+def test_dest_remote_root_with_prefix():
+    cfg = Config(dest_type="s3", dest_s3_bucket="dest-bucket",
+                 dest_s3_prefix="output/path/")
+    assert cfg.dest_remote_root() == "dest:dest-bucket/output/path"
+
+
+def test_validate_s3_dest_requires_creds():
+    cfg = Config(
+        src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+        src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="src",
+        dest_type="s3", dest_s3_bucket="dest",
+        dest_s3_endpoint="https://s3.crusoe.ai",
+    )
+    errs = cfg.validate()
+    assert any("DEST_S3_ACCESS_KEY_ID" in e for e in errs)
+
+
+def test_validate_s3_dest_requires_bucket():
+    cfg = Config(
+        src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+        src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="src",
+        dest_type="s3", dest_s3_access_key_id="dk", dest_s3_secret_access_key="ds",
+        dest_s3_endpoint="https://s3.crusoe.ai", dest_s3_bucket="",
+    )
+    errs = cfg.validate()
+    assert any("DEST_S3_BUCKET" in e for e in errs)
+
+
+def test_validate_s3_dest_requires_endpoint():
+    cfg = Config(
+        src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+        src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="src",
+        dest_type="s3", dest_s3_access_key_id="dk", dest_s3_secret_access_key="ds",
+        dest_s3_endpoint="", dest_s3_bucket="dest",
+    )
+    errs = cfg.validate()
+    assert any("DEST_S3_ENDPOINT" in e for e in errs)
+
+
+def test_validate_s3_dest_passes():
+    cfg = Config(
+        src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+        src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="src",
+        dest_type="s3", dest_s3_access_key_id="dk", dest_s3_secret_access_key="ds",
+        dest_s3_endpoint="https://s3.crusoe.ai", dest_s3_bucket="dest",
+    )
+    errs = cfg.validate()
+    assert errs == []
+
+
+def test_validate_s3_dest_skips_disk_checks():
+    """When dest_type=s3, DEST_MODE validation for import/nfs is irrelevant."""
+    cfg = Config(
+        src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+        src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="src",
+        dest_type="s3", dest_s3_access_key_id="dk", dest_s3_secret_access_key="ds",
+        dest_s3_endpoint="https://s3.crusoe.ai", dest_s3_bucket="dest",
+        dest_mode="import",  # should NOT trigger import validation
+    )
+    errs = cfg.validate()
+    assert not any("EXISTING_DISK" in e for e in errs)
+
+
+def test_validate_disk_dest_unchanged():
+    """When dest_type=disk, existing DEST_MODE validation still works."""
+    cfg = Config(
+        src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+        src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="src",
+        dest_type="disk", dest_mode="import",
+    )
+    errs = cfg.validate()
+    assert any("EXISTING_DISK" in e for e in errs)
+
+
+def test_env_map_has_dest_s3_keys():
+    from orchestrator.config import _ENV_MAP
+    assert "DEST_TYPE" in _ENV_MAP
+    assert "DEST_S3_ACCESS_KEY_ID" in _ENV_MAP
+    assert "DEST_S3_SECRET_ACCESS_KEY" in _ENV_MAP
+    assert "DEST_S3_ENDPOINT" in _ENV_MAP
+    assert "DEST_S3_BUCKET" in _ENV_MAP
+    assert "DEST_S3_PREFIX" in _ENV_MAP
+    assert "DEST_S3_REGION" in _ENV_MAP
diff --git a/cmk-data-transfer/tests/test_manifests.py b/cmk-data-transfer/tests/test_manifests.py
new file mode 100644
index 0000000..2ccdb71
--- /dev/null
+++ b/cmk-data-transfer/tests/test_manifests.py
@@ -0,0 +1,72 @@
+# tests/test_manifests.py
+"""Tests for manifest generation with S3 destination support."""
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from orchestrator.config import Config
+from orchestrator.sizing import compute_sizing
+from orchestrator import manifests
+
+
+def _s3_dest_config() -> Config:
+    return Config(
+        src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+        src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="src-bucket",
+        src_s3_prefix="data/",
+        dest_type="s3", dest_s3_access_key_id="dk", dest_s3_secret_access_key="ds",
+        dest_s3_endpoint="https://s3.crusoe.ai", dest_s3_bucket="dest-bucket",
+        dest_s3_prefix="output/",
+        num_nodes=1, pods_per_node=2,
+    )
+
+
+def _disk_dest_config() -> Config:
+    return Config(
+        src_s3_access_key_id="key", src_s3_secret_access_key="secret",
+        src_s3_endpoint="https://s3.amazonaws.com", src_s3_bucket="src-bucket",
+        dest_type="disk",
+        num_nodes=1, pods_per_node=2,
+    )
+
+
+def test_worker_command_s3_dest():
+    cfg = _s3_dest_config()
+    sizing = compute_sizing(cfg)
+    pod = manifests.worker_pod(cfg, sizing, 0)
+    cmd = pod["spec"]["containers"][0]["command"][-1]  # the shell script
+    assert "src:src-bucket/data" in cmd
+    assert "dest:dest-bucket/output" in cmd
+    assert "/data/dataset" not in cmd  # should NOT write to local disk
+
+
+def test_worker_command_disk_dest():
+    cfg = _disk_dest_config()
+    sizing = compute_sizing(cfg)
+    pod = manifests.worker_pod(cfg, sizing, 0)
+    cmd = pod["spec"]["containers"][0]["command"][-1]
+    assert "src:src-bucket" in cmd
+    assert "/data/dataset" in cmd
+    assert "dest:" not in cmd
+
+
+def test_pvc_size_s3_dest():
+    cfg = _s3_dest_config()
+    p = manifests.pvc(cfg)
+    assert p["spec"]["resources"]["requests"]["storage"] == "1Ti"
+
+
+def test_pvc_size_disk_dest():
+    cfg = _disk_dest_config()
+    p = manifests.pvc(cfg)
+    assert p["spec"]["resources"]["requests"]["storage"] == "1000Ti"
+
+
+def test_master_pod_no_dest_mkdir_s3():
+    """When dest is S3, master pod should not mkdir DEST_PATH (no data on disk)."""
+    cfg = _s3_dest_config()
+    pod = manifests.master_pod(cfg)
+    init_cmd = pod["spec"]["containers"][0]["command"][-1]
+    assert "/data/dataset" not in init_cmd
+    # But shard and log dirs should still be created
+    assert "/data/shards" in init_cmd
+    assert "/data/logs" in init_cmd
diff --git a/cmk-data-transfer/tests/test_rclone_conf.py b/cmk-data-transfer/tests/test_rclone_conf.py
new file mode 100644
index 0000000..22872bf
--- /dev/null
+++ b/cmk-data-transfer/tests/test_rclone_conf.py
@@ -0,0 +1,94 @@
+# tests/test_rclone_conf.py
+"""Tests for rclone.conf generation after OCI->SRC_S3 rename."""
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from orchestrator.config import Config
+from orchestrator.rclone_conf import build_rclone_conf
+
+
+def test_source_remote_named_src():
+    cfg = Config(
+        src_s3_access_key_id="AKID",
+        src_s3_secret_access_key="SECRET",
+        src_s3_endpoint="https://s3.us-east-1.amazonaws.com",
+        src_s3_region="us-east-1",
+    )
+    conf = build_rclone_conf(cfg)
+    assert "[src]" in conf
+    assert "[oci]" not in conf
+
+
+def test_source_remote_contains_creds():
+    cfg = Config(
+        src_s3_access_key_id="AKID",
+        src_s3_secret_access_key="SECRET",
+        src_s3_endpoint="https://s3.us-east-1.amazonaws.com",
+        src_s3_region="us-east-1",
+    )
+    conf = build_rclone_conf(cfg)
+    assert "access_key_id = AKID" in conf
+    assert "secret_access_key = SECRET" in conf
+    assert "endpoint = https://s3.us-east-1.amazonaws.com" in conf
+    assert "region = us-east-1" in conf
+
+
+def test_source_remote_generic_s3():
+    cfg = Config(
+        src_s3_access_key_id="AKID",
+        src_s3_secret_access_key="SECRET",
+        src_s3_endpoint="https://s3.us-east-1.amazonaws.com",
+    )
+    conf = build_rclone_conf(cfg)
+    assert "type = s3" in conf
+    assert "provider = Other" in conf
+    assert "force_path_style = true" in conf
+    assert "no_check_bucket = true" in conf
+
+
+def test_no_dest_remote_when_disk():
+    cfg = Config(
+        src_s3_access_key_id="AKID",
+        src_s3_secret_access_key="SECRET",
+        src_s3_endpoint="https://s3.us-east-1.amazonaws.com",
+        dest_type="disk",
+    )
+    conf = build_rclone_conf(cfg)
+    assert "[dest]" not in conf
+
+
+def test_dest_remote_when_s3():
+    cfg = Config(
+        src_s3_access_key_id="AKID",
+        src_s3_secret_access_key="SECRET",
+        src_s3_endpoint="https://s3.us-east-1.amazonaws.com",
+        dest_type="s3",
+        dest_s3_access_key_id="DAKID",
+        dest_s3_secret_access_key="DSECRET",
+        dest_s3_endpoint="https://s3.crusoe.ai",
+        dest_s3_region="us-east",
+    )
+    conf = build_rclone_conf(cfg)
+    assert "[src]" in conf
+    assert "[dest]" in conf
+    # dest creds
+    assert "DAKID" in conf
+    assert "DSECRET" in conf
+    assert "https://s3.crusoe.ai" in conf
+
+
+def test_dest_remote_region_omitted_when_empty():
+    cfg = Config(
+        src_s3_access_key_id="AKID",
+        src_s3_secret_access_key="SECRET",
+        src_s3_endpoint="https://s3.us-east-1.amazonaws.com",
+        dest_type="s3",
+        dest_s3_access_key_id="DAKID",
+        dest_s3_secret_access_key="DSECRET",
+        dest_s3_endpoint="https://s3.crusoe.ai",
+        dest_s3_region="",
+    )
+    conf = build_rclone_conf(cfg)
+    # The [dest] section should NOT have a region line
+    dest_section = conf[conf.index("[dest]"):]
+    assert "region" not in dest_section