diff --git a/crates/persistence/src/backends/s3/bulk_submit.rs b/crates/persistence/src/backends/s3/bulk_submit.rs
index f76f1a39b..787c5129e 100644
--- a/crates/persistence/src/backends/s3/bulk_submit.rs
+++ b/crates/persistence/src/backends/s3/bulk_submit.rs
@@ -21,7 +21,7 @@ use crate::core::bulk_submit::{
     SubmissionManifest, SubmissionStatus, SubmissionSummary, UnindexedEntry,
     invalid_entry_result_page,
 };
-use crate::error::{BulkSubmitError, ResourceError, StorageError, StorageResult};
+use crate::error::{BackendError, BulkSubmitError, ResourceError, StorageError, StorageResult};
 use crate::tenant::TenantContext;
 
 use super::backend::{S3Backend, TenantLocation};
@@ -336,6 +336,10 @@ impl BulkSubmitProvider for S3Backend {
         .await?;
 
         let mut results = Vec::new();
+        // Every rollback change the batch produces, gathered here and written as
+        // one coalesced object after the walk instead of one PUT per resource
+        // (#1429). Order does not matter — `list_changes` sorts by timestamp.
+        let mut changes: Vec<SubmissionChange> = Vec::new();
         let mut error_count = 0u32;
         let file_url = options.file_url.as_deref();
 
@@ -347,6 +351,11 @@ impl BulkSubmitProvider for S3Backend {
         self.persist_raw_batch(&location, submission_id, manifest_id, file_url, &entries)
             .await?;
 
+        // The batch's first line keys its coalesced raw archive and its
+        // coalesced change log alike; captured here because the walk below
+        // consumes `entries`.
+        let batch_first_line = entries.first().map(|entry| entry.line_number);
+
         // S3 writes each entry on its own, so an entry is durable as soon as
         // its write returns. The loop runs in a block so that however it ends —
         // exhausted, max errors reached, or a storage error part-way — the
@@ -382,28 +391,34 @@ impl BulkSubmitProvider for S3Backend {
             && !has_id_collision
         {
             use futures::stream::{self, StreamExt};
-            let outcomes: Vec<StorageResult<BulkEntryResult>> = stream::iter(entries)
-                .map(|entry| {
-                    self.process_one_entry(
-                        &location,
-                        tenant,
-                        submission_id,
-                        manifest_id,
-                        file_url,
-                        entry,
-                        options,
-                    )
-                })
-                .buffered(concurrency)
-                .collect()
-                .await;
+            let outcomes: Vec<StorageResult<(BulkEntryResult, Option<SubmissionChange>)>> =
+                stream::iter(entries)
+                    .map(|entry| {
+                        self.process_one_entry(
+                            &location,
+                            tenant,
+                            submission_id,
+                            manifest_id,
+                            file_url,
+                            entry,
+                            options,
+                        )
+                    })
+                    .buffered(concurrency)
+                    .collect()
+                    .await;
             // `error_count` is only the serial path's early-stop counter; with
             // no cap the final tallies come from `results` below, so it is not
             // touched here.
             let mut first_err = None;
             for outcome in outcomes {
                 match outcome {
-                    Ok(result) => results.push(result),
+                    Ok((result, change)) => {
+                        results.push(result);
+                        if let Some(change) = change {
+                            changes.push(change);
+                        }
+                    }
                     Err(err) => {
                         if first_err.is_none() {
                             first_err = Some(err);
@@ -447,15 +462,18 @@ impl BulkSubmitProvider for S3Backend {
 
                     // The raw line was archived for the whole batch upfront
                     // (persist_raw_batch), so the loop no longer PUTs it per entry.
-                    let result = match self
+                    let (result, change) = match self
                         .process_single_entry(tenant, submission_id, manifest_id, &entry, options)
                         .await
                     {
-                        Ok(result) => result,
-                        Err(err) => BulkEntryResult::processing_error(
-                            entry.line_number,
-                            &entry.resource_type,
-                            Self::bulk_submit_operation_outcome(&err),
+                        Ok((result, change)) => (result, change),
+                        Err(err) => (
+                            BulkEntryResult::processing_error(
+                                entry.line_number,
+                                &entry.resource_type,
+                                Self::bulk_submit_operation_outcome(&err),
+                            ),
+                            None,
                         ),
                     };
 
@@ -471,18 +489,44 @@ impl BulkSubmitProvider for S3Backend {
                         &result,
                     )
                     .await?;
+                    if let Some(change) = change {
+                        changes.push(change);
+                    }
                     results.push(result);
                 }
                 Ok(())
             }
             .await
         };
+        // Write every rollback change the batch produced in one coalesced
+        // object (#1429). This is deferred to here — after the resources and
+        // their receipts — so it costs one PUT per batch instead of one per
+        // resource; whatever changes were gathered before a hard failure are
+        // still written, so a partially-processed batch stays rollback-covered.
+        // A failure here is reported after the receipts (below), like the walk's
+        // own #1078 contract, and yields to the walk error when both occur.
+        let change_write = match batch_first_line {
+            Some(first_line) => {
+                self.persist_change_batch(
+                    &location,
+                    submission_id,
+                    manifest_id,
+                    file_url,
+                    first_line,
+                    &changes,
+                )
+                .await
+            }
+            None => Ok(()),
+        };
+
         // Entries are written one by one and not kept, so observers that need
         // the resources re-read the ids from the primary (#1127).
         options
             .notify_batch_committed(tenant, submission_id, manifest_id, &results, &[])
             .await;
         walked?;
+        change_write?;
 
         let success_count = results.iter().filter(|r| r.is_success()).count() as u64;
         let failed_count = results.iter().filter(|r| r.is_error()).count() as u64;
@@ -843,11 +887,13 @@ impl BulkSubmitRollbackProvider for S3Backend {
 }
 
 impl S3Backend {
-    /// Processes one ingest entry end to end — writes its raw line, runs it,
-    /// and writes its receipt — returning the receipt. This is the unit the
-    /// batch runs, serially or concurrently. A per-entry *processing* failure
-    /// becomes a `processing-error` receipt (not an error); only a hard store
-    /// failure writing the raw line or the receipt propagates.
+    /// Processes one ingest entry end to end — runs it and writes its receipt —
+    /// returning the receipt together with the rollback change it produced (if
+    /// any) for the caller to coalesce. This is the unit the batch runs,
+    /// serially or concurrently. A per-entry *processing* failure becomes a
+    /// `processing-error` receipt (not an error) and no change; only a hard
+    /// store failure writing the receipt propagates. The raw line was archived
+    /// for the whole batch upfront (persist_raw_batch).
     #[allow(clippy::too_many_arguments)]
     async fn process_one_entry(
         &self,
@@ -858,22 +904,24 @@ impl S3Backend {
         file_url: Option<&str>,
         entry: NdjsonEntry,
         options: &BulkProcessingOptions,
-    ) -> StorageResult<BulkEntryResult> {
-        // The raw line was archived for the whole batch upfront (persist_raw_batch).
-        let result = match self
+    ) -> StorageResult<(BulkEntryResult, Option<SubmissionChange>)> {
+        let (result, change) = match self
             .process_single_entry(tenant, submission_id, manifest_id, &entry, options)
             .await
         {
-            Ok(result) => result,
-            Err(err) => BulkEntryResult::processing_error(
-                entry.line_number,
-                &entry.resource_type,
-                Self::bulk_submit_operation_outcome(&err),
+            Ok((result, change)) => (result, change),
+            Err(err) => (
+                BulkEntryResult::processing_error(
+                    entry.line_number,
+                    &entry.resource_type,
+                    Self::bulk_submit_operation_outcome(&err),
+                ),
+                None,
             ),
         };
         self.persist_entry_result(location, submission_id, manifest_id, file_url, &result)
             .await?;
-        Ok(result)
+        Ok((result, change))
     }
 
     /// How many ingest entries run at once on S3 when no per-entry error cap is
@@ -889,36 +937,43 @@ impl S3Backend {
             .unwrap_or(8)
     }
 
-    /// Processes a single NDJSON entry: validates it, upserts the resource,
-    /// and records a change log entry for rollback.
+    /// Processes a single NDJSON entry: validates it and upserts the resource.
     ///
-    /// Returns a `BulkEntryResult` describing the outcome. Storage errors are
-    /// promoted to entry-level processing errors rather than aborting the whole
-    /// batch.
+    /// Returns the `BulkEntryResult` describing the outcome together with the
+    /// rollback change it produced, if any. The change is *returned*, not
+    /// written, so the caller can coalesce a whole batch's changes into one
+    /// object instead of one PUT per resource (#1429); the receipt path already
+    /// batches the raw archive the same way. A validation error, a mismatch, or
+    /// a skipped update produces no change. Storage errors are promoted to
+    /// entry-level processing errors by the caller rather than aborting the
+    /// whole batch.
     async fn process_single_entry(
         &self,
         tenant: &TenantContext,
-        submission_id: &SubmissionId,
+        _submission_id: &SubmissionId,
         manifest_id: &str,
         entry: &NdjsonEntry,
         options: &BulkProcessingOptions,
-    ) -> StorageResult<BulkEntryResult> {
+    ) -> StorageResult<(BulkEntryResult, Option<SubmissionChange>)> {
         if let Some(resource_type) = entry.resource.get("resourceType").and_then(|v| v.as_str()) {
             if resource_type != entry.resource_type {
-                return Ok(BulkEntryResult::validation_error(
-                    entry.line_number,
-                    &entry.resource_type,
-                    serde_json::json!({
-                        "resourceType": "OperationOutcome",
-                        "issue": [{
-                            "severity": "error",
-                            "code": "invalid",
-                            "diagnostics": format!(
-                                "resourceType mismatch: entry={}, payload={}",
-                                entry.resource_type, resource_type
-                            )
-                        }]
-                    }),
+                return Ok((
+                    BulkEntryResult::validation_error(
+                        entry.line_number,
+                        &entry.resource_type,
+                        serde_json::json!({
+                            "resourceType": "OperationOutcome",
+                            "issue": [{
+                                "severity": "error",
+                                "code": "invalid",
+                                "diagnostics": format!(
+                                    "resourceType mismatch: entry={}, payload={}",
+                                    entry.resource_type, resource_type
+                                )
+                            }]
+                        }),
+                    ),
+                    None,
                 ));
             }
         }
@@ -927,10 +982,13 @@ impl S3Backend {
             match self.read(tenant, &entry.resource_type, id).await {
                 Ok(Some(current)) => {
                     if !options.allow_updates {
-                        return Ok(BulkEntryResult::skipped(
-                            entry.line_number,
-                            &entry.resource_type,
-                            "updates not allowed",
+                        return Ok((
+                            BulkEntryResult::skipped(
+                                entry.line_number,
+                                &entry.resource_type,
+                                "updates not allowed",
+                            ),
+                            None,
                         ));
                     }
 
@@ -946,13 +1004,15 @@ impl S3Backend {
                         updated.version_id(),
                         current.content().clone(),
                     );
-                    self.record_change(tenant, submission_id, &change).await?;
 
-                    Ok(BulkEntryResult::success(
-                        entry.line_number,
-                        &entry.resource_type,
-                        updated.id(),
-                        false,
+                    Ok((
+                        BulkEntryResult::success(
+                            entry.line_number,
+                            &entry.resource_type,
+                            updated.id(),
+                            false,
+                        ),
+                        Some(change),
                     ))
                 }
                 Ok(None) | Err(StorageError::Resource(ResourceError::Gone { .. })) => {
@@ -971,13 +1031,15 @@ impl S3Backend {
                         created.id(),
                         created.version_id(),
                     );
-                    self.record_change(tenant, submission_id, &change).await?;
 
-                    Ok(BulkEntryResult::success(
-                        entry.line_number,
-                        &entry.resource_type,
-                        created.id(),
-                        true,
+                    Ok((
+                        BulkEntryResult::success(
+                            entry.line_number,
+                            &entry.resource_type,
+                            created.id(),
+                            true,
+                        ),
+                        Some(change),
                     ))
                 }
                 Err(err) => Err(err),
@@ -998,13 +1060,15 @@ impl S3Backend {
                 created.id(),
                 created.version_id(),
             );
-            self.record_change(tenant, submission_id, &change).await?;
 
-            Ok(BulkEntryResult::success(
-                entry.line_number,
-                &entry.resource_type,
-                created.id(),
-                true,
+            Ok((
+                BulkEntryResult::success(
+                    entry.line_number,
+                    &entry.resource_type,
+                    created.id(),
+                    true,
+                ),
+                Some(change),
             ))
         }
     }
@@ -1061,6 +1125,44 @@ impl S3Backend {
         Ok(())
     }
 
+    /// Writes an ingest batch's rollback changes to S3 as a single object — the
+    /// whole batch's changes in one array rather than one PUT per resource
+    /// (#1429).
+    ///
+    /// Stored under `changes/<manifest>/<file>/batch-<first_line>.json` (see
+    /// [`crate::backends::s3::keyspace::S3Keyspace::submit_change_batch_key`]),
+    /// which sits below the same `changes/` prefix that `load_changes` reads and
+    /// the per-change [`Self::record_change`] writes — so `load_changes` picks
+    /// up both this array form and any legacy single-change object. An empty
+    /// change set writes nothing. The batch's first line keys the object and
+    /// `file_url` discriminates it, exactly as the raw archive and the receipts
+    /// are keyed, so batches never overwrite one another.
+    async fn persist_change_batch(
+        &self,
+        location: &TenantLocation,
+        submission_id: &SubmissionId,
+        manifest_id: &str,
+        file_url: Option<&str>,
+        first_line: u64,
+        changes: &[SubmissionChange],
+    ) -> StorageResult<()> {
+        if changes.is_empty() {
+            return Ok(());
+        }
+        let key = location.keyspace.submit_change_batch_key(
+            &submission_id.submitter,
+            &submission_id.submission_id,
+            manifest_id,
+            file_url,
+            first_line,
+        );
+
+        let payload = self.serialize_json(&changes)?;
+        self.put_json_object(&location.bucket, &key, &payload, None, None)
+            .await?;
+        Ok(())
+    }
+
     /// Persists the processing result for a single entry to S3.
     ///
     /// `file_url` is the manifest output file the line came from. It is part of
@@ -1196,11 +1298,34 @@ impl S3Backend {
                 continue;
             }
 
-            if let Some((change, _)) = self
-                .get_json_object::<SubmissionChange>(&location.bucket, &object.key)
+            // The `changes/` prefix now holds two shapes: a coalesced batch is a
+            // JSON array of changes (#1429), and a legacy per-change object — or
+            // one written by the trait's `record_change` — is a single change.
+            // Discriminate on the parsed JSON so both are read back.
+            let Some((value, _)) = self
+                .get_json_object::<serde_json::Value>(&location.bucket, &object.key)
                 .await?
-            {
-                changes.push(change);
+            else {
+                continue;
+            };
+            match value {
+                serde_json::Value::Array(_) => {
+                    let batch: Vec<SubmissionChange> =
+                        serde_json::from_value(value).map_err(|e| {
+                            StorageError::Backend(BackendError::SerializationError {
+                                message: format!("failed to deserialize change batch: {e}"),
+                            })
+                        })?;
+                    changes.extend(batch);
+                }
+                _ => {
+                    let change: SubmissionChange = serde_json::from_value(value).map_err(|e| {
+                        StorageError::Backend(BackendError::SerializationError {
+                            message: format!("failed to deserialize change: {e}"),
+                        })
+                    })?;
+                    changes.push(change);
+                }
             }
         }
 
diff --git a/crates/persistence/src/backends/s3/keyspace.rs b/crates/persistence/src/backends/s3/keyspace.rs
index c0dfe3a26..98a902a61 100644
--- a/crates/persistence/src/backends/s3/keyspace.rs
+++ b/crates/persistence/src/backends/s3/keyspace.rs
@@ -343,6 +343,34 @@ impl S3Keyspace {
         ])
     }
 
+    /// Key for one ingest batch's coalesced change log — every change the batch
+    /// recorded, in a single object rather than one per resource (#1429).
+    ///
+    /// Sits under the same `changes/` prefix that [`Self::submit_change_key`]
+    /// writes and that `load_changes` lists, but nested by manifest and file so
+    /// batches never collide: like the raw archive and the entry receipts, the
+    /// key is discriminated by `file_url` (line numbers restart per file, see
+    /// [`submit_file_segment`]) and keyed by the batch's first line.
+    pub fn submit_change_batch_key(
+        &self,
+        submitter: &str,
+        submission_id: &str,
+        manifest_id: &str,
+        file_url: Option<&str>,
+        first_line: u64,
+    ) -> String {
+        self.join(&[
+            "bulk",
+            "submit",
+            submitter,
+            submission_id,
+            "changes",
+            manifest_id,
+            &submit_file_segment(file_url),
+            &format!("batch-{}.json", first_line),
+        ])
+    }
+
     /// Key for one finalized status-manifest artifact row of a submission.
     ///
     /// The identity is `(submitter, submission_id, manifest_id, file_type,
diff --git a/crates/persistence/src/backends/s3/tests.rs b/crates/persistence/src/backends/s3/tests.rs
index 023ea111b..6f2960bfc 100644
--- a/crates/persistence/src/backends/s3/tests.rs
+++ b/crates/persistence/src/backends/s3/tests.rs
@@ -24,7 +24,8 @@ use crate::backends::s3::user_settings::settings_object_id;
 use crate::core::bulk_export::{ExportDataProvider, ExportRequest};
 use crate::core::bulk_submit::{
     BulkProcessingOptions, BulkSubmitProvider, BulkSubmitRollbackProvider, CANCELLED_ABORT_REASON,
-    CancelToken, NdjsonEntry, StreamingBulkSubmitProvider, SubmissionId, SubmissionStatus,
+    CancelToken, NdjsonEntry, StreamingBulkSubmitProvider, SubmissionChange, SubmissionId,
+    SubmissionStatus,
 };
 use crate::core::history::{
     HistoryParams, InstanceHistoryProvider, SystemHistoryProvider, TypeHistoryProvider,
@@ -1074,6 +1075,131 @@ async fn bulk_submit_raw_archive_is_one_object_per_batch() {
     }
 }
 
+/// A batch records its rollback changes in one coalesced object, not one PUT
+/// per resource, and `list_changes` still reads every change back (#1429).
+#[tokio::test]
+async fn bulk_submit_change_log_is_one_object_per_batch() {
+    let mock = Arc::new(MockS3Client::with_buckets(&["test-bucket"]));
+    let backend = make_prefix_backend(mock.clone());
+    let tenant = tenant("tenant-a");
+
+    let submission_id = SubmissionId::new("client-a", "sub-changes");
+    backend
+        .create_submission(&tenant, &submission_id, None)
+        .await
+        .unwrap();
+    let manifest = backend
+        .add_manifest(&tenant, &submission_id, None, None)
+        .await
+        .unwrap();
+
+    let entries: Vec<NdjsonEntry> = (1..=3)
+        .map(|i| {
+            NdjsonEntry::new(
+                i,
+                "Patient",
+                json!({"resourceType": "Patient", "id": format!("c{i}")}),
+            )
+        })
+        .collect();
+    backend
+        .process_entries(
+            &tenant,
+            &submission_id,
+            &manifest.manifest_id,
+            entries,
+            &BulkProcessingOptions::new(),
+        )
+        .await
+        .unwrap();
+
+    let change_puts = mock
+        .recorded_puts()
+        .into_iter()
+        .filter(|put| put.key.contains("/changes/"))
+        .count();
+    assert_eq!(
+        change_puts, 1,
+        "the three-entry batch must record its changes in one object, not three"
+    );
+
+    let changes = backend
+        .list_changes(&tenant, &submission_id, 10, 0)
+        .await
+        .unwrap();
+    assert_eq!(
+        changes.len(),
+        3,
+        "all three changes must be readable back from the coalesced object"
+    );
+}
+
+/// `load_changes` reads both shapes under the `changes/` prefix: the coalesced
+/// batch array (#1429) and a legacy single-change object — such as one the
+/// rollback trait's `record_change` still writes for composite backends.
+#[tokio::test]
+async fn bulk_submit_change_log_reads_batch_and_legacy_objects() {
+    let mock = Arc::new(MockS3Client::with_buckets(&["test-bucket"]));
+    let backend = make_prefix_backend(mock);
+    let tenant = tenant("tenant-a");
+
+    let submission_id = SubmissionId::new("client-a", "sub-mixed");
+    backend
+        .create_submission(&tenant, &submission_id, None)
+        .await
+        .unwrap();
+    let manifest = backend
+        .add_manifest(&tenant, &submission_id, None, None)
+        .await
+        .unwrap();
+
+    // A batch writes the coalesced array form.
+    let entries: Vec<NdjsonEntry> = (1..=2)
+        .map(|i| {
+            NdjsonEntry::new(
+                i,
+                "Patient",
+                json!({"resourceType": "Patient", "id": format!("m{i}")}),
+            )
+        })
+        .collect();
+    backend
+        .process_entries(
+            &tenant,
+            &submission_id,
+            &manifest.manifest_id,
+            entries,
+            &BulkProcessingOptions::new(),
+        )
+        .await
+        .unwrap();
+
+    // A single change written through the trait method takes the legacy shape.
+    let legacy = SubmissionChange::create(&manifest.manifest_id, "Observation", "legacy-1", "1");
+    backend
+        .record_change(&tenant, &submission_id, &legacy)
+        .await
+        .unwrap();
+
+    let changes = backend
+        .list_changes(&tenant, &submission_id, 10, 0)
+        .await
+        .unwrap();
+    assert_eq!(
+        changes.len(),
+        3,
+        "both the two-change batch object and the one legacy object must be read"
+    );
+    assert!(
+        changes.iter().any(|c| c.resource_id == "legacy-1"),
+        "the legacy single-change object must be read back"
+    );
+    assert!(
+        changes.iter().any(|c| c.resource_id == "m1"),
+        "the coalesced batch changes must be read back"
+    );
+}
+
 /// A batch with two entries for the same resource id is order-dependent
 /// (last write wins), so it ingests serially even with concurrency enabled —
 /// the concurrent path would race them to a non-deterministic result (#945).