diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index da6c251d3..a02028404 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -299,7 +299,8 @@ jobs:
       # this backend declares no vector search capability, so its tests drive the
       # storage layer directly against this job's PostgreSQL. The PutItem create
       # race tests hold a create open in an outside transaction, the
-      # write-transaction conflict tests force a deadlock, and the backup restore
+      # write-transaction conflict tests force a deadlock, the stream clock test
+      # holds a row lock open while a write waits on it, and the backup restore
       # cases need catalog rows the current binary would not write; all need
       # direct access. The online GSI build tests reorder queue rows by hand and
       # pause writers with a test hook. They build their own throwaway databases;
@@ -316,7 +317,7 @@ jobs:
           cargo test --release -p extenddb-storage-postgres
           --test vector_control_plane --test put_create_race --test twi_conflict
           --test stream_shard_retention --test stream_shard_ownership --test stream_records_page_order --test backup_restore
-          --test gsi_online_build --test gsi_stale_writer --test table_snapshot
+          --test gsi_online_build --test gsi_stale_writer --test table_snapshot --test stream_record_clock
 
       # The daemonized server logs to syslog; dump it so server-side failures
       # are diagnosable from the job log.
diff --git a/crates/storage-cassandra/src/stream_util.rs b/crates/storage-cassandra/src/stream_util.rs
index 1ad6d8fbc..09746e9d2 100644
--- a/crates/storage-cassandra/src/stream_util.rs
+++ b/crates/storage-cassandra/src/stream_util.rs
@@ -58,8 +58,30 @@ impl HybridClock {
     /// Returns a 23-digit zero-padded string. Lexicographic ordering matches
     /// temporal ordering.
     pub fn generate(&mut self) -> String {
+        self.generate_with_created_at().0
+    }
+
+    /// Generate the next sequence number and the wall-clock reading taken by
+    /// this call, as `(sequence_number, created_at_ms)`.
+    ///
+    /// The sequence prefix is that reading or a later clamped value, so a
+    /// record's stored creation time never exceeds the timestamp prefix of
+    /// its own sequence number (`created_at_ms <= hlc_ms`). A reading at or
+    /// below the last prefix keeps the last prefix; when the counter
+    /// overflows, the prefix moves to the later of a fresh reading and the
+    /// last prefix plus one.
+    pub fn generate_with_created_at(&mut self) -> (String, i64) {
         let now_ms = chrono::Utc::now().timestamp_millis();
+        self.generate_from(now_ms, || {
+            std::thread::sleep(std::time::Duration::from_millis(1));
+            chrono::Utc::now().timestamp_millis()
+        })
+    }
 
+    /// One clock step for the reading `now_ms`. `reread` is called only when
+    /// the counter overflows and returns the reading taken after the wait;
+    /// tests pass both readings in.
+    fn generate_from(&mut self, now_ms: i64, reread: impl FnOnce() -> i64) -> (String, i64) {
         if now_ms > self.last_timestamp_ms {
             self.last_timestamp_ms = now_ms;
             self.logical_counter = 0;
@@ -80,22 +102,21 @@ impl HybridClock {
             // component stays close to real time; ordering is carried by the
             // floor below, not by the sleep.
             if self.logical_counter > 999_999 {
-                std::thread::sleep(std::time::Duration::from_millis(1));
-                // The re-read can still sit behind `last_timestamp_ms` when
-                // the clock was clamped forward by an earlier backwards step.
-                // The physical component must move past the millisecond that
-                // already issued counter 999_999, so it never goes below
-                // `last_timestamp_ms + 1`.
-                let reread_ms = chrono::Utc::now().timestamp_millis();
-                self.last_timestamp_ms = reread_ms.max(self.last_timestamp_ms + 1);
+                // The re-read can sit below the last prefix, after a backwards
+                // step during the wait or under a forward clamp, so the prefix
+                // moves at least one past the last one. It stays above every
+                // earlier reading, and the sequence number sorts after every
+                // earlier one.
+                self.last_timestamp_ms = reread().max(self.last_timestamp_ms + 1);
                 self.logical_counter = 0;
             }
         }
 
-        format!(
+        let seq = format!(
             "{:013}{:06}{:04}",
             self.last_timestamp_ms, self.logical_counter, self.node_id
-        )
+        );
+        (seq, now_ms)
     }
 }
 
@@ -161,13 +182,14 @@ pub fn stream_record_statement(
     hlc: &Arc<Mutex<HybridClock>>,
     retention_seconds: u32,
 ) -> Option<String> {
+    let (sequence_number, created_at_ms) = hlc
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .generate_with_created_at();
     let identity = StreamRecordIdentity {
         event_id: uuid::Uuid::new_v4().to_string(),
-        sequence_number: hlc
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .generate(),
-        created_at_ms: chrono::Utc::now().timestamp_millis(),
+        sequence_number,
+        created_at_ms,
     };
     stream_record_statement_with_identity(
         account_keyspace,
@@ -287,6 +309,146 @@ mod tests {
         assert!(b < c, "sequence numbers must be ordered: {b} < {c}");
     }
 
+    /// The stored creation time never exceeds the millisecond prefix of the
+    /// record's own sequence number, including when the clock is clamped to
+    /// an earlier read (same-millisecond bursts and backwards steps both take
+    /// the clamp branch). On the clamp branch the returned value is the
+    /// call's own wall-clock reading, not the clamped prefix.
+    #[test]
+    fn created_at_never_exceeds_sequence_prefix() {
+        let mut clock = HybridClock::new(7);
+        // Force the clamp branch: pretend the last read was far in the future.
+        clock.last_timestamp_ms = chrono::Utc::now().timestamp_millis() + 5_000;
+        for _ in 0..1_000 {
+            let before = chrono::Utc::now().timestamp_millis();
+            let (seq, created_at_ms) = clock.generate_with_created_at();
+            let after = chrono::Utc::now().timestamp_millis();
+            let hlc_ms: i64 = seq[..13].parse().unwrap();
+            assert!(
+                created_at_ms <= hlc_ms,
+                "created_at_ms {created_at_ms} exceeds sequence prefix {hlc_ms} in {seq}"
+            );
+            assert!(
+                (before..=after).contains(&created_at_ms),
+                "created_at_ms {created_at_ms} must be the reading between {before} and \
+                 {after}, below the clamped prefix {hlc_ms}"
+            );
+        }
+        // On the normal branch the prefix is the reading itself.
+        let mut fresh = HybridClock::new(7);
+        let before = chrono::Utc::now().timestamp_millis();
+        let (seq, created_at_ms) = fresh.generate_with_created_at();
+        let after = chrono::Utc::now().timestamp_millis();
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        assert_eq!(created_at_ms, hlc_ms);
+        assert!(
+            (before..=after).contains(&created_at_ms),
+            "created_at_ms {created_at_ms} is the wall clock read between {before} and {after}"
+        );
+    }
+
+    /// The counter-overflow branch re-reads the clock for the sequence
+    /// prefix. The readings are injected, so the branch runs on every run.
+    /// The returned `created_at_ms` is still the call's own reading, at or
+    /// below the new prefix.
+    #[test]
+    fn created_at_holds_across_counter_overflow() {
+        let now_ms: i64 = 1_800_000_000_000;
+        let mut clock = HybridClock::new(42);
+        let (first, _) =
+            clock.generate_from(now_ms, || unreachable!("no overflow on a fresh clock"));
+        clock.logical_counter = 999_999;
+        let mut reread = false;
+        let (seq, created_at_ms) = clock.generate_from(now_ms, || {
+            reread = true;
+            now_ms + 1
+        });
+        assert!(reread, "the call takes the overflow branch");
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        assert_eq!(
+            created_at_ms, now_ms,
+            "created_at_ms is the call's own reading"
+        );
+        assert_eq!(hlc_ms, now_ms + 1, "the prefix moves to the re-read");
+        assert!(
+            created_at_ms <= hlc_ms,
+            "overflow branch: created_at_ms {created_at_ms} exceeds prefix {hlc_ms} in {seq}"
+        );
+        assert_eq!(
+            clock.logical_counter, 0,
+            "the counter resets after overflow"
+        );
+        assert_eq!(
+            &seq[13..19],
+            "000000",
+            "the counter field resets in the sequence"
+        );
+        assert!(
+            seq > first,
+            "the overflow sequence follows the previous one: {first} < {seq}"
+        );
+    }
+
+    /// When the wall clock steps back during the overflow wait, the re-read
+    /// lands below the call's own reading. The prefix still stays at or
+    /// above that reading, and the sequence still follows the previous one.
+    #[test]
+    fn overflow_prefix_stays_above_the_reading_when_the_reread_steps_back() {
+        let now_ms: i64 = 1_800_000_000_000;
+        let mut clock = HybridClock::new(3);
+        clock.last_timestamp_ms = now_ms;
+        clock.logical_counter = 999_999;
+        let previous = format!("{now_ms:013}{:06}{:04}", 999_999, 3);
+        let (seq, created_at_ms) = clock.generate_from(now_ms, || now_ms - 10);
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        assert!(
+            created_at_ms <= hlc_ms,
+            "created_at_ms {created_at_ms} exceeds sequence prefix {hlc_ms} in {seq}"
+        );
+        assert!(
+            seq > previous,
+            "the overflow sequence follows the previous one: {previous} < {seq}"
+        );
+    }
+
+    /// Overflow under a forward clamp: the last prefix sits 5 s ahead of the
+    /// call's reading and the re-read lands a further 5 s back. The prefix
+    /// moves to the last prefix plus one, not to the reading plus one, so the
+    /// sequence still follows the previous one and `created_at_ms` stays at
+    /// or below the prefix.
+    #[test]
+    fn overflow_under_a_forward_clamp_moves_the_prefix_past_the_last_one() {
+        let now_ms: i64 = 1_800_000_000_000;
+        let mut clock = HybridClock::new(5);
+        clock.last_timestamp_ms = now_ms + 5_000;
+        clock.logical_counter = 999_999;
+        let previous = format!("{:013}{:06}{:04}", now_ms + 5_000, 999_999, 5);
+        let mut reread_called = false;
+        let (seq, created_at_ms) = clock.generate_from(now_ms, || {
+            reread_called = true;
+            now_ms - 5_000
+        });
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        assert!(reread_called, "the call takes the overflow branch");
+        assert_eq!(
+            created_at_ms, now_ms,
+            "created_at_ms is the call's own reading"
+        );
+        assert!(
+            created_at_ms <= hlc_ms,
+            "created_at_ms {created_at_ms} exceeds sequence prefix {hlc_ms} in {seq}"
+        );
+        assert!(
+            seq > previous,
+            "the overflow sequence follows the previous one: {previous} < {seq}"
+        );
+        assert_eq!(
+            hlc_ms,
+            now_ms + 5_001,
+            "the prefix moves to the last prefix plus one: {seq}"
+        );
+    }
+
     #[test]
     fn test_sequence_length() {
         let mut clock = HybridClock::new(1);
diff --git a/crates/storage-cassandra/src/ttl_worker.rs b/crates/storage-cassandra/src/ttl_worker.rs
index f6a170ddd..b85b9900b 100644
--- a/crates/storage-cassandra/src/ttl_worker.rs
+++ b/crates/storage-cassandra/src/ttl_worker.rs
@@ -871,19 +871,20 @@ async fn process_ttl_work_row(
             .as_ref()
             .and_then(|specification| {
                 if specification.stream_enabled {
-                    specification
-                        .stream_view_type
-                        .map(|view_type| TtlStreamPlan {
+                    specification.stream_view_type.map(|view_type| {
+                        let (sequence_number, created_at_ms) = storage
+                            .hlc
+                            .lock()
+                            .unwrap_or_else(std::sync::PoisonError::into_inner)
+                            .generate_with_created_at();
+                        TtlStreamPlan {
                             event_id: uuid::Uuid::new_v4().to_string(),
-                            sequence_number: storage
-                                .hlc
-                                .lock()
-                                .unwrap_or_else(std::sync::PoisonError::into_inner)
-                                .generate(),
-                            created_at_ms: chrono::Utc::now().timestamp_millis(),
+                            sequence_number,
+                            created_at_ms,
                             region: storage.region.clone(),
                             view_type,
-                        })
+                        }
+                    })
                 } else {
                     None
                 }
diff --git a/crates/storage-cassandra/tests/direct/streams.rs b/crates/storage-cassandra/tests/direct/streams.rs
index 16c86540e..bc0efa206 100644
--- a/crates/storage-cassandra/tests/direct/streams.rs
+++ b/crates/storage-cassandra/tests/direct/streams.rs
@@ -653,3 +653,92 @@ async fn test_list_streams_includes_stream_enabled_table() {
 
     assert!(streams.iter().any(|s| s.stream_arn == expected_arn));
 }
+
+/// `created_at` and `ApproximateCreationDateTime` derive from one clock
+/// read, and the sequence prefix is that reading or a later clamped value.
+/// For every stored record `created_at_ms <= hlc_ms`, and
+/// `created_at_ms / 1000` equals the record's timestamp. Covers plain
+/// writes and a service-identity deletion through `delete_item`.
+#[tokio::test]
+async fn test_stream_record_created_at_matches_the_record_and_sequence() {
+    if crate::helpers::skip_without_cassandra() {
+        return;
+    }
+    let engine = setup_engine().await;
+    let table = TestTable::new(&engine, "StreamClock", false).await;
+    let view = StreamViewType::NewAndOldImages;
+
+    for v in ["one", "two"] {
+        let mut item = BTreeMap::new();
+        item.insert("id".to_string(), AttributeValue::S("pk-clock".to_string()));
+        item.insert("val".to_string(), AttributeValue::S(v.to_string()));
+        engine
+            .put_item(
+                &table.key_info,
+                item,
+                false,
+                None,
+                &Default::default(),
+                Some(&capture(view)),
+            )
+            .await
+            .unwrap();
+    }
+    let mut key = BTreeMap::new();
+    key.insert("id".to_string(), AttributeValue::S("pk-clock".to_string()));
+    engine
+        .delete_item(
+            &table.key_info,
+            &key,
+            false,
+            None,
+            &Default::default(),
+            Some(&StreamCapture {
+                view_type: view,
+                user_identity: Some(extenddb_core::types::UserIdentity {
+                    identity_type: "Service".to_string(),
+                    principal_id: "dynamodb.amazonaws.com".to_string(),
+                }),
+                region: Arc::from("us-east-1"),
+            }),
+        )
+        .await
+        .unwrap();
+
+    let shard_id = shard_for("pk-clock", &table.key_info.table_id);
+    let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id);
+    let query = format!(
+        "SELECT sequence_number, created_at, record_data FROM {keyspace}.stream_records \
+         WHERE shard_id = ?"
+    );
+    let result = engine
+        .session_arc()
+        .query_with_values(&query, cdrs_tokio::query_values!(shard_id.as_str()))
+        .await
+        .unwrap();
+    let rows = result
+        .response_body()
+        .unwrap()
+        .into_rows()
+        .unwrap_or_default();
+    assert_eq!(rows.len(), 3, "two puts and one delete");
+    for row in rows {
+        let seq: String = row.get_r_by_name("sequence_number").unwrap();
+        let created_at_ms: i64 = row.get_r_by_name("created_at").unwrap();
+        let data: String = row.get_r_by_name("record_data").unwrap();
+        let record: serde_json::Value = serde_json::from_str(&data).unwrap();
+        let hlc_ms: i64 = seq[..13].parse().unwrap();
+        let approx = record["dynamodb"]["ApproximateCreationDateTime"]
+            .as_i64()
+            .unwrap();
+        assert!(
+            created_at_ms <= hlc_ms,
+            "created_at {created_at_ms} exceeds the sequence prefix {hlc_ms} in {seq}"
+        );
+        assert_eq!(
+            created_at_ms.div_euclid(1_000),
+            approx,
+            "created_at {created_at_ms} and ApproximateCreationDateTime {approx} disagree"
+        );
+    }
+}
diff --git a/crates/storage-cassandra/tests/ttl_integration.rs b/crates/storage-cassandra/tests/ttl_integration.rs
index 279d79492..4f712918a 100644
--- a/crates/storage-cassandra/tests/ttl_integration.rs
+++ b/crates/storage-cassandra/tests/ttl_integration.rs
@@ -593,7 +593,8 @@ async fn test_ttl_sweep_emits_service_remove_stream_record() {
         &key_info.table_id,
     );
     let query = format!(
-        "SELECT event_name, record_data FROM {account_keyspace}.stream_records WHERE shard_id = ?"
+        "SELECT event_name, sequence_number, created_at, record_data \
+         FROM {account_keyspace}.stream_records WHERE shard_id = ?"
     );
     let rows = engine
         .session_arc()
@@ -604,20 +605,38 @@ async fn test_ttl_sweep_emits_service_remove_stream_record() {
         .expect("parse TTL stream response")
         .into_rows()
         .unwrap_or_default();
-    let remove_records: Vec<String> = rows
+    let remove_records: Vec<(String, i64, String)> = rows
         .into_iter()
         .filter_map(|row| {
             let event: String = row.get_r_by_name("event_name").ok()?;
             if event == "Remove" {
-                row.get_r_by_name("record_data").ok()
+                Some((
+                    row.get_r_by_name("sequence_number")
+                        .expect("sequence_number"),
+                    row.get_r_by_name("created_at").expect("created_at"),
+                    row.get_r_by_name("record_data").expect("record_data"),
+                ))
             } else {
                 None
             }
         })
         .collect();
     assert_eq!(remove_records.len(), 1);
+    let (seq, created_at_ms, data) = &remove_records[0];
     let record: serde_json::Value =
-        serde_json::from_str(&remove_records[0]).expect("parse TTL stream record JSON");
+        serde_json::from_str(data).expect("parse TTL stream record JSON");
+    // The TTL plan takes its sequence number and created_at from one clock
+    // call, so the pair the worker persisted keeps the stream invariants.
+    let hlc_ms: i64 = seq[..13].parse().expect("sequence number prefix");
+    assert!(
+        *created_at_ms <= hlc_ms,
+        "TTL record created_at_ms {created_at_ms} exceeds sequence prefix {hlc_ms} in {seq}"
+    );
+    assert_eq!(
+        record["dynamodb"]["ApproximateCreationDateTime"].as_i64(),
+        Some(created_at_ms / 1_000),
+        "TTL record timestamp must be created_at_ms / 1000"
+    );
     assert_eq!(record["userIdentity"]["Type"], "Service");
     assert_eq!(
         record["userIdentity"]["PrincipalId"],
diff --git a/crates/storage-mongodb/src/backup_live_tests.rs b/crates/storage-mongodb/src/backup_live_tests.rs
index 75ed23761..fb42f09da 100644
--- a/crates/storage-mongodb/src/backup_live_tests.rs
+++ b/crates/storage-mongodb/src/backup_live_tests.rs
@@ -1773,3 +1773,104 @@ async fn live_update_table_on_a_creating_restore_target_is_refused() {
     assert!(desc.deletion_protection_enabled);
     fx.finish().await;
 }
+
+/// A stream record's `created_at` field and its own `ApproximateCreationDateTime`
+/// come from one clock read on the live write path, for a plain write and for
+/// a service-identity deletion (the TTL worker's shape).
+#[tokio::test]
+#[ignore = "needs EXTENDDB_TEST_MONGODB_URI"]
+async fn live_stream_record_created_at_matches_the_record_timestamp() {
+    use extenddb_core::expression::ExpressionMaps;
+    use extenddb_core::types::{StreamSpecification, StreamViewType, UserIdentity};
+    use extenddb_storage::{DataEngine, StreamCapture};
+
+    let fx = Fixture::new().await;
+    fx.engine
+        .create_table(
+            ACCOUNT,
+            CreateTableInput {
+                table_name: "clock".to_owned(),
+                key_schema: vec![key("pk", KeyType::Hash)],
+                attribute_definitions: vec![attr("pk", ScalarAttributeType::S)],
+                billing_mode: Some(BillingMode::PayPerRequest),
+                stream_specification: Some(StreamSpecification {
+                    stream_enabled: true,
+                    stream_view_type: Some(StreamViewType::NewAndOldImages),
+                }),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("create the table");
+    let key_info = fx
+        .engine
+        .table_key_info(ACCOUNT, "clock")
+        .await
+        .expect("key info");
+    let maps = ExpressionMaps::default();
+    let capture = StreamCapture {
+        view_type: StreamViewType::NewAndOldImages,
+        user_identity: None,
+        region: Arc::from("us-east-1"),
+    };
+    for v in ["one", "two"] {
+        let item: Item = [
+            ("pk".to_owned(), AttributeValue::S("k".to_owned())),
+            ("v".to_owned(), AttributeValue::S(v.to_owned())),
+        ]
+        .into_iter()
+        .collect();
+        fx.engine
+            .put_item(&key_info, item, false, None, &maps, Some(&capture))
+            .await
+            .expect("put");
+    }
+    let key: Item = [("pk".to_owned(), AttributeValue::S("k".to_owned()))]
+        .into_iter()
+        .collect();
+    fx.engine
+        .delete_item(
+            &key_info,
+            &key,
+            false,
+            None,
+            &maps,
+            Some(&StreamCapture {
+                view_type: StreamViewType::NewAndOldImages,
+                user_identity: Some(UserIdentity {
+                    identity_type: "Service".to_owned(),
+                    principal_id: "dynamodb.amazonaws.com".to_owned(),
+                }),
+                region: Arc::from("us-east-1"),
+            }),
+        )
+        .await
+        .expect("delete");
+
+    let mut cursor = fx
+        .engine
+        .data_db
+        .collection::<Document>("stream_records")
+        .find(doc! {})
+        .await
+        .expect("read back");
+    let mut seen = 0;
+    while cursor.advance().await.expect("cursor") {
+        let d = cursor.deserialize_current().expect("document");
+        let created_at = d.get_datetime("created_at").expect("created_at");
+        let approx = d
+            .get_document("record_data")
+            .and_then(|r| r.get_document("dynamodb"))
+            .expect("record_data.dynamodb")
+            .get_i64("ApproximateCreationDateTime")
+            .expect("ApproximateCreationDateTime");
+        assert_eq!(
+            created_at.timestamp_millis().div_euclid(1_000),
+            approx,
+            "created_at {created_at} and ApproximateCreationDateTime {approx} disagree"
+        );
+        seen += 1;
+    }
+    assert_eq!(seen, 3, "two puts and one delete");
+    fx.finish().await;
+}
diff --git a/crates/storage-mongodb/src/data_engine.rs b/crates/storage-mongodb/src/data_engine.rs
index dd8f4cc9f..1582c7c5b 100644
--- a/crates/storage-mongodb/src/data_engine.rs
+++ b/crates/storage-mongodb/src/data_engine.rs
@@ -2038,6 +2038,9 @@ impl MongoEngine {
             .next_sequence_number_in_session(&shard_id, &mut *session)
             .await?;
 
+        // One wall-clock read feeds both the record's ApproximateCreationDateTime
+        // and the created_at field, so the two never disagree.
+        let created_at = mongodb::bson::DateTime::now();
         let record = StreamRecord {
             event_id: uuid::Uuid::new_v4().to_string(),
             event_name: event,
@@ -2045,13 +2048,7 @@ impl MongoEngine {
             event_source: "aws:dynamodb".to_owned(),
             aws_region: capture.region.to_string(),
             dynamodb: StreamRecordData {
-                approximate_creation_date_time: i64::try_from(
-                    std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .unwrap_or_default()
-                        .as_secs(),
-                )
-                .unwrap_or(i64::MAX),
+                approximate_creation_date_time: created_at.timestamp_millis().div_euclid(1_000),
                 keys,
                 new_image,
                 old_image,
@@ -2081,7 +2078,7 @@ impl MongoEngine {
                 "table_id": table_id,
                 "event_name": crate::stream_engine::event_name_ddb_str(record.event_name),
                 "record_data": record_bson,
-                "created_at": mongodb::bson::DateTime::now(),
+                "created_at": created_at,
             })
             .session(&mut *session)
             .await
diff --git a/crates/storage-postgres/src/data/tx_helpers.rs b/crates/storage-postgres/src/data/tx_helpers.rs
index 6468f159b..d17d0c437 100755
--- a/crates/storage-postgres/src/data/tx_helpers.rs
+++ b/crates/storage-postgres/src/data/tx_helpers.rs
@@ -337,12 +337,21 @@ pub(super) async fn write_stream_record_in_tx(
     let idx = (hash as usize) % shards.len();
     let shard_id = &shards[idx].0;
 
-    // Generate monotonic sequence number within the transaction (CB-21).
-    let (seq_val,): (i64,) = sqlx::query_as("SELECT nextval('stream_seq')")
-        .fetch_one(&mut **tx)
-        .await
-        .map_err(db_error)?;
+    // Generate monotonic sequence number within the transaction (CB-21). The
+    // same statement takes the record's one clock read: clock_timestamp() is
+    // the database clock when this statement runs, after the write's locking
+    // reads, while the column default (NOW()) is the transaction start, which
+    // in a long transaction sits seconds before the record. created_at and
+    // ApproximateCreationDateTime both derive from this reading, and the
+    // retention sweep compares created_at with the database's NOW(), so the
+    // retention window is measured on the database clock alone.
+    let (seq_val, created_at): (i64, time::OffsetDateTime) =
+        sqlx::query_as("SELECT nextval('stream_seq'), clock_timestamp()")
+            .fetch_one(&mut **tx)
+            .await
+            .map_err(db_error)?;
     let seq = format!("{seq_val:021}");
+    let approximate_creation_date_time = created_at.unix_timestamp();
 
     let record = StreamRecord {
         event_id: uuid::Uuid::new_v4().to_string(),
@@ -351,13 +360,7 @@ pub(super) async fn write_stream_record_in_tx(
         event_source: "aws:dynamodb".to_owned(),
         aws_region: capture.region.to_string(),
         dynamodb: StreamRecordData {
-            approximate_creation_date_time: i64::try_from(
-                std::time::SystemTime::now()
-                    .duration_since(std::time::UNIX_EPOCH)
-                    .unwrap_or_default()
-                    .as_secs(),
-            )
-            .unwrap_or(i64::MAX),
+            approximate_creation_date_time,
             keys,
             new_image,
             old_image,
@@ -372,14 +375,15 @@ pub(super) async fn write_stream_record_in_tx(
         serde_json::to_value(&record).map_err(|e| StorageError::Internal(e.to_string()))?;
 
     sqlx::query(
-        "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data) \
-         VALUES ($1, $2, $3, $4, $5)",
+        "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data, created_at) \
+         VALUES ($1, $2, $3, $4, $5, $6)",
     )
     .bind(&record.dynamodb.sequence_number)
     .bind(shard_id)
     .bind(&key_info.table_id)
     .bind(format!("{:?}", record.event_name))
     .bind(&record_json)
+    .bind(created_at)
     .execute(&mut **tx)
     .await
     .map_err(db_error)?;
diff --git a/crates/storage-postgres/src/stream_engine.rs b/crates/storage-postgres/src/stream_engine.rs
index 970a81755..da7fd122d 100755
--- a/crates/storage-postgres/src/stream_engine.rs
+++ b/crates/storage-postgres/src/stream_engine.rs
@@ -177,15 +177,25 @@ impl StreamEngine for PostgresEngine {
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
+            // created_at follows the record's own timestamp rather than a
+            // second clock read (NOW() would be the transaction start time).
+            // A timestamp outside the range `time` can represent, which a
+            // SystemTime reading never is, is stamped with the current time,
+            // as the SQLite engine stamps it.
+            let created_at = time::OffsetDateTime::from_unix_timestamp(
+                record.dynamodb.approximate_creation_date_time,
+            )
+            .unwrap_or_else(|_| time::OffsetDateTime::now_utc());
             sqlx::query(
-                "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data) \
-                 VALUES ($1, $2, $3, $4, $5)",
+                "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data, created_at) \
+                 VALUES ($1, $2, $3, $4, $5, $6)",
             )
             .bind(&record.dynamodb.sequence_number)
             .bind(&shard_id)
             .bind(&table_id)
             .bind(format!("{:?}", record.event_name))
             .bind(&record_json)
+            .bind(created_at)
             .execute(&self.data_pool)
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;
diff --git a/crates/storage-postgres/tests/stream_record_clock.rs b/crates/storage-postgres/tests/stream_record_clock.rs
new file mode 100644
index 000000000..f3913f9ed
--- /dev/null
+++ b/crates/storage-postgres/tests/stream_record_clock.rs
@@ -0,0 +1,351 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! A stream record's `created_at` column and its `ApproximateCreationDateTime`
+//! come from one clock read.
+//!
+//! The column default (`NOW()`) is the transaction start time. A write that
+//! waits inside its transaction, on a row lock here, stores a `created_at`
+//! that is earlier than the record's own timestamp by the length of the wait.
+//! The retention sweep filters on `created_at`, so the two values must agree.
+//! The one reading is the database's `clock_timestamp()`, the clock the
+//! sweep's `NOW()` cutoff is taken on, so the test also bounds it with
+//! database clock reads taken before the lock is released and after the
+//! write returns.
+//!
+//! Needs `EXTENDDB_TEST_PG_CONNECTION_STRING` (host-only, e.g.
+//! `postgres://user:pass@127.0.0.1:5432`); skips otherwise. Each test builds
+//! its own scratch database and drops it afterwards.
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use extenddb_core::expression::ExpressionMaps;
+use extenddb_core::types::{
+    AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, Item, KeySchemaElement,
+    KeyType, ScalarAttributeType, StreamSpecification, StreamViewType, TableKeyInfo, UserIdentity,
+};
+use extenddb_storage::{DataEngine, StreamCapture, TableEngine};
+use extenddb_storage_postgres::{PostgresConfig, PostgresEngine};
+use sqlx::PgPool;
+use sqlx::postgres::PgPoolOptions;
+
+const ACCOUNT: &str = "123456789012";
+const REGION: &str = "us-east-1";
+const TABLE: &str = "t_stream_clock";
+
+struct Scratch {
+    engine: PostgresEngine,
+    db: PgPool,
+    admin: PgPool,
+    db_name: String,
+}
+
+impl Scratch {
+    async fn cleanup(self) {
+        let Scratch {
+            engine,
+            db,
+            admin,
+            db_name,
+        } = self;
+        drop(engine);
+        db.close().await;
+        sqlx::query(&format!(
+            "DROP DATABASE IF EXISTS \"{db_name}\" WITH (FORCE)"
+        ))
+        .execute(&admin)
+        .await
+        .expect("drop the scratch database");
+        admin.close().await;
+    }
+}
+
+fn base_conn() -> Option<String> {
+    let conn = std::env::var("EXTENDDB_TEST_PG_CONNECTION_STRING").ok()?;
+    (!conn.trim().is_empty()).then(|| conn.trim_end_matches('/').to_owned())
+}
+
+fn skip(test: &str) {
+    eprintln!(
+        "SKIP {test}: EXTENDDB_TEST_PG_CONNECTION_STRING is not set, so there is no PostgreSQL \
+         to build a scratch catalog in."
+    );
+}
+
+fn shipped_migrations() -> Vec<String> {
+    let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
+    let mut out = Vec::new();
+    for dir in ["migrations", "data_migrations"] {
+        let mut files: Vec<_> = std::fs::read_dir(root.join(dir))
+            .expect("read the migrations directory")
+            .map(|e| e.expect("directory entry").path())
+            .filter(|p| p.extension().is_some_and(|e| e == "sql"))
+            .collect();
+        files.sort();
+        for f in files {
+            out.push(std::fs::read_to_string(&f).expect("read a migration file"));
+        }
+    }
+    out
+}
+
+async fn scratch() -> Scratch {
+    let base = base_conn().expect("caller checks base_conn() first");
+    let db_name = format!("eddb_sclk_{}", uuid::Uuid::new_v4().simple())[..24].to_owned();
+    let admin = PgPoolOptions::new()
+        .max_connections(1)
+        .connect(&format!("{base}/postgres"))
+        .await
+        .expect("connect to the postgres maintenance database");
+    sqlx::query(&format!("CREATE DATABASE \"{db_name}\""))
+        .execute(&admin)
+        .await
+        .expect("create the scratch database");
+    let url = format!("{base}/{db_name}");
+    let db = PgPoolOptions::new()
+        .max_connections(3)
+        .connect(&url)
+        .await
+        .expect("connect to the scratch database");
+    for sql in shipped_migrations() {
+        sqlx::raw_sql(&sql)
+            .execute(&db)
+            .await
+            .expect("apply a shipped migration");
+    }
+    sqlx::query("ALTER TABLE stream_shards DROP CONSTRAINT IF EXISTS stream_shards_table_id_fkey")
+        .execute(&db)
+        .await
+        .expect("match the two-database layout");
+    sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'")
+        .execute(&db)
+        .await
+        .expect("pin the control-plane delay to zero");
+    sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES ($1, $2)")
+        .bind(ACCOUNT)
+        .bind(format!("acct-{db_name}"))
+        .execute(&db)
+        .await
+        .expect("seed the account row");
+    let engine = PostgresEngine::new(
+        &PostgresConfig {
+            connection_string: url,
+            pool_size: 10,
+            max_item_size_bytes: 400_000,
+        },
+        REGION,
+    )
+    .await
+    .expect("open a PostgresEngine on the scratch database");
+    Scratch {
+        engine,
+        db,
+        admin,
+        db_name,
+    }
+}
+
+fn item(pk: &str, v: &str) -> Item {
+    BTreeMap::from([
+        ("pk".to_owned(), AttributeValue::S(pk.to_owned())),
+        ("v".to_owned(), AttributeValue::S(v.to_owned())),
+    ])
+}
+
+async fn stream_table(s: &Scratch) -> TableKeyInfo {
+    s.engine
+        .create_table(
+            ACCOUNT,
+            CreateTableInput {
+                table_name: TABLE.to_owned(),
+                key_schema: vec![KeySchemaElement {
+                    attribute_name: "pk".to_owned(),
+                    key_type: KeyType::Hash,
+                }],
+                attribute_definitions: vec![AttributeDefinition {
+                    attribute_name: "pk".to_owned(),
+                    attribute_type: ScalarAttributeType::S,
+                }],
+                billing_mode: Some(BillingMode::PayPerRequest),
+                stream_specification: Some(StreamSpecification {
+                    stream_enabled: true,
+                    stream_view_type: Some(StreamViewType::NewImage),
+                }),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("create the table");
+    s.engine
+        .table_key_info(ACCOUNT, TABLE)
+        .await
+        .expect("read the key info")
+}
+
+async fn data_table(db: &PgPool) -> String {
+    let id: String =
+        sqlx::query_scalar("SELECT table_id FROM tables WHERE account_id = $1 AND table_name = $2")
+            .bind(ACCOUNT)
+            .bind(TABLE)
+            .fetch_one(db)
+            .await
+            .expect("look up the table id");
+    format!("\"_ddb_{id}\"")
+}
+
+async fn wait_for_lock_waiters(db: &PgPool, own_pid: i32, n: i64) {
+    for _ in 0..500 {
+        let waiting: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM pg_stat_activity \
+             WHERE datname = current_database() AND wait_event_type = 'Lock' AND pid <> $1",
+        )
+        .bind(own_pid)
+        .fetch_one(db)
+        .await
+        .expect("read pg_stat_activity");
+        if waiting >= n {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("{n} backends never waited on a lock");
+}
+
+/// Stored `(created_at, ApproximateCreationDateTime)` for every record in the
+/// scratch database, in sequence order.
+async fn stored_times(db: &PgPool) -> Vec<(time::OffsetDateTime, i64)> {
+    let rows: Vec<(time::OffsetDateTime, serde_json::Value)> = sqlx::query_as(
+        "SELECT created_at, record_data FROM stream_records ORDER BY sequence_number",
+    )
+    .fetch_all(db)
+    .await
+    .expect("read the stream records");
+    rows.into_iter()
+        .map(|(created, data)| {
+            let approx = data["dynamodb"]["ApproximateCreationDateTime"]
+                .as_i64()
+                .expect("ApproximateCreationDateTime is an integer");
+            (created, approx)
+        })
+        .collect()
+}
+
+#[tokio::test]
+async fn created_at_matches_the_record_timestamp_when_the_write_waited() {
+    let test = "created_at_matches_the_record_timestamp_when_the_write_waited";
+    if base_conn().is_none() {
+        return skip(test);
+    }
+    let s = scratch().await;
+    let maps = ExpressionMaps::default();
+    let key_info = stream_table(&s).await;
+    let capture = StreamCapture {
+        view_type: StreamViewType::NewImage,
+        user_identity: None,
+        region: Arc::from(REGION),
+    };
+    // Seed so the write below takes the locked update path.
+    s.engine
+        .put_item(
+            &key_info,
+            item("k", "seed"),
+            false,
+            None,
+            &maps,
+            Some(&capture),
+        )
+        .await
+        .expect("seed the item");
+    let table = data_table(&s.db).await;
+
+    // An outside transaction holds the row. The write begins its own
+    // transaction (fixing NOW()), waits on the lock for longer than a second,
+    // then builds its record once the lock is released.
+    let mut holder = s.db.begin().await.expect("begin the outside transaction");
+    let holder_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
+        .fetch_one(&mut *holder)
+        .await
+        .expect("read the backend pid");
+    sqlx::query(&format!("SELECT 1 FROM {table} WHERE pk = 'k' FOR UPDATE"))
+        .execute(&mut *holder)
+        .await
+        .expect("lock the row");
+
+    let write = s.engine.put_item(
+        &key_info,
+        item("k", "late"),
+        false,
+        None,
+        &maps,
+        Some(&capture),
+    );
+    let release = async {
+        wait_for_lock_waiters(&s.db, holder_pid, 1).await;
+        // Past the next second boundary and then some, so NOW() and the
+        // record's own clock read fall in different seconds.
+        tokio::time::sleep(Duration::from_millis(2_100)).await;
+        let released_at: time::OffsetDateTime = sqlx::query_scalar("SELECT clock_timestamp()")
+            .fetch_one(&mut *holder)
+            .await
+            .expect("read the database clock before the release");
+        holder.commit().await.expect("release the row");
+        released_at
+    };
+    let (result, released_at) = tokio::time::timeout(Duration::from_secs(30), async {
+        tokio::join!(write, release)
+    })
+    .await
+    .expect("the write finished");
+    result.expect("the write succeeded");
+    let written_by: time::OffsetDateTime = sqlx::query_scalar("SELECT clock_timestamp()")
+        .fetch_one(&s.db)
+        .await
+        .expect("read the database clock after the write");
+
+    // A service-identity deletion, the TTL worker's shape, through the same
+    // helper.
+    s.engine
+        .delete_item(
+            &key_info,
+            &BTreeMap::from([("pk".to_owned(), AttributeValue::S("k".to_owned()))]),
+            false,
+            None,
+            &maps,
+            Some(&StreamCapture {
+                view_type: StreamViewType::NewImage,
+                user_identity: Some(UserIdentity {
+                    identity_type: "Service".to_owned(),
+                    principal_id: "dynamodb.amazonaws.com".to_owned(),
+                }),
+                region: Arc::from(REGION),
+            }),
+        )
+        .await
+        .expect("the TTL-shaped delete succeeded");
+
+    let times = stored_times(&s.db).await;
+    assert_eq!(times.len(), 3, "seed, late, and remove records: {times:?}");
+    for (created_at, approx) in &times {
+        assert_eq!(
+            created_at.unix_timestamp(),
+            *approx,
+            "created_at second must equal ApproximateCreationDateTime: {times:?}"
+        );
+    }
+    // And the waited record is at least two seconds after the seed.
+    assert!(
+        times[1].1 - times[0].1 >= 2,
+        "the late record carries the post-wait time: {times:?}"
+    );
+    // Its created_at is a database clock read taken after the lock was
+    // released and before the write returned.
+    let late = times[1].0;
+    assert!(
+        released_at <= late && late <= written_by,
+        "created_at {late} must be a database clock read between {released_at} and {written_by}"
+    );
+
+    s.cleanup().await;
+}
diff --git a/crates/storage-sqlite/src/data/tx_helpers.rs b/crates/storage-sqlite/src/data/tx_helpers.rs
index 1fd1bfca8..0ff033523 100644
--- a/crates/storage-sqlite/src/data/tx_helpers.rs
+++ b/crates/storage-sqlite/src/data/tx_helpers.rs
@@ -230,13 +230,16 @@ pub(super) async fn write_stream_record_in_tx(
     let shard_id = shards[idx].0.clone();
 
     let seq = format!("{:021}", next_stream_seq(tx).await?);
-    let approximate_creation_date_time = i64::try_from(
+    // One wall-clock read feeds both the record's ApproximateCreationDateTime
+    // and the created_at column, so the two never disagree.
+    let created_at_ms = i64::try_from(
         std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap_or_default()
-            .as_secs(),
+            .as_millis(),
     )
     .unwrap_or(i64::MAX);
+    let approximate_creation_date_time = created_at_ms.div_euclid(1_000);
 
     let record = StreamRecord {
         event_id: uuid::Uuid::new_v4().to_string(),
@@ -259,14 +262,15 @@ pub(super) async fn write_stream_record_in_tx(
         serde_json::to_string(&record).map_err(|e| StorageError::Internal(e.to_string()))?;
 
     sqlx::query(
-        "INSERT INTO stream_records (shard_id, sequence_number, table_id, event_name, record_data) \
-         VALUES (?, ?, ?, ?, ?)",
+        "INSERT INTO stream_records (shard_id, sequence_number, table_id, event_name, record_data, created_at) \
+         VALUES (?, ?, ?, ?, ?, ?)",
     )
     .bind(&shard_id)
     .bind(&seq)
     .bind(&key_info.table_id)
     .bind(format!("{:?}", record.event_name))
     .bind(&record_json)
+    .bind(crate::sqlite_util::format_epoch_millis(created_at_ms))
     .execute(&mut **tx)
     .await
     .map_err(|e| StorageError::Internal(e.to_string()))?;
@@ -378,3 +382,116 @@ mod idempotency_tests {
         check(&pool, "acctB", "tok", "fpX").await.unwrap();
     }
 }
+
+#[cfg(test)]
+mod stream_clock_tests {
+    //! The stored `created_at` and the record's own `ApproximateCreationDateTime`
+    //! come from one clock read on the engine write path.
+
+    use std::collections::BTreeMap;
+    use std::sync::Arc;
+
+    use extenddb_core::expression::ExpressionMaps;
+    use extenddb_core::types::{AttributeValue, Item, StreamViewType};
+    use extenddb_storage::{DataEngine, StreamCapture, TableEngine};
+    use serde_json::json;
+
+    use crate::SqliteEngine;
+
+    const ACCOUNT: &str = "123456789012";
+
+    async fn engine() -> SqliteEngine {
+        let engine = SqliteEngine::new(":memory:", 1, "us-east-1", 409_600)
+            .await
+            .expect("engine");
+        crate::schema::apply(&engine.pool).await.expect("schema");
+        sqlx::query("UPDATE settings SET value = '0' WHERE key = 'control_plane_delay_seconds'")
+            .execute(&engine.pool)
+            .await
+            .expect("delay");
+        sqlx::query("INSERT INTO accounts (account_id, account_name) VALUES (?, 'default')")
+            .bind(ACCOUNT)
+            .execute(&engine.pool)
+            .await
+            .expect("account");
+        engine
+    }
+
+    #[tokio::test]
+    async fn stored_created_at_matches_the_record_timestamp() {
+        let engine = engine().await;
+        let input: extenddb_core::types::CreateTableInput = serde_json::from_value(json!({
+            "TableName": "t_clock",
+            "KeySchema": [{"AttributeName": "pk", "KeyType": "HASH"}],
+            "AttributeDefinitions": [{"AttributeName": "pk", "AttributeType": "S"}],
+            "BillingMode": "PAY_PER_REQUEST",
+            "StreamSpecification": {"StreamEnabled": true, "StreamViewType": "NEW_AND_OLD_IMAGES"}
+        }))
+        .expect("input");
+        engine.create_table(ACCOUNT, input).await.expect("create");
+        let key_info = engine
+            .table_key_info(ACCOUNT, "t_clock")
+            .await
+            .expect("key info");
+        let capture = StreamCapture {
+            view_type: StreamViewType::NewAndOldImages,
+            user_identity: None,
+            region: Arc::from("us-east-1"),
+        };
+        let maps = ExpressionMaps::default();
+        for v in ["one", "two", "three"] {
+            let item: Item = BTreeMap::from([
+                ("pk".to_owned(), AttributeValue::S("k".to_owned())),
+                ("v".to_owned(), AttributeValue::S(v.to_owned())),
+            ]);
+            engine
+                .put_item(&key_info, item, false, None, &maps, Some(&capture))
+                .await
+                .expect("put");
+        }
+        // Expiry of the item deletes it with a service identity, the TTL
+        // worker's shape, through the same helper.
+        engine
+            .delete_item(
+                &key_info,
+                &BTreeMap::from([("pk".to_owned(), AttributeValue::S("k".to_owned()))]),
+                false,
+                None,
+                &maps,
+                Some(&StreamCapture {
+                    view_type: StreamViewType::NewAndOldImages,
+                    user_identity: Some(extenddb_core::types::UserIdentity {
+                        identity_type: "Service".to_owned(),
+                        principal_id: "dynamodb.amazonaws.com".to_owned(),
+                    }),
+                    region: Arc::from("us-east-1"),
+                }),
+            )
+            .await
+            .expect("delete");
+
+        let rows: Vec<(String, String)> = sqlx::query_as(
+            "SELECT created_at, record_data FROM stream_records ORDER BY sequence_number",
+        )
+        .fetch_all(&engine.pool)
+        .await
+        .expect("read back");
+        assert_eq!(rows.len(), 4, "{rows:?}");
+        for (created_at, data) in &rows {
+            // Same shape as the column default: 24 characters, millisecond
+            // fraction, trailing Z.
+            assert_eq!(created_at.len(), 24, "{created_at}");
+            let record: serde_json::Value = serde_json::from_str(data).expect("record json");
+            let approx = record["dynamodb"]["ApproximateCreationDateTime"]
+                .as_i64()
+                .expect("ApproximateCreationDateTime");
+            let stored_secs = crate::sqlite_util::timestamp_epoch_seconds(created_at).floor();
+            #[allow(clippy::cast_possible_truncation)]
+            let stored_secs = stored_secs as i64;
+            assert_eq!(
+                stored_secs, approx,
+                "created_at {created_at} second must equal ApproximateCreationDateTime {approx}"
+            );
+        }
+    }
+}
diff --git a/crates/storage-sqlite/src/sqlite_util.rs b/crates/storage-sqlite/src/sqlite_util.rs
index c37ce480f..903f19019 100644
--- a/crates/storage-sqlite/src/sqlite_util.rs
+++ b/crates/storage-sqlite/src/sqlite_util.rs
@@ -53,6 +53,31 @@ pub(crate) fn format_epoch_seconds(secs: f64) -> String {
     )
 }
 
+/// Format epoch milliseconds in the shape the `stream_records.created_at`
+/// column default produces (`strftime('%Y-%m-%dT%H:%M:%fZ')`, millisecond
+/// precision, 24 characters): `2026-10-15T12:00:00.438Z`. Same width as the
+/// default, so rows written before and after this formatter was introduced
+/// sort together as text in time order.
+///
+/// A value outside the range `time` can represent (beyond the year 9999)
+/// cannot come from a `SystemTime` reading; should one arrive, the row is
+/// stamped with the current time, as the PostgreSQL engine does, so the
+/// retention sweep treats it as a fresh record rather than one from 1970.
+pub(crate) fn format_epoch_millis(ms: i64) -> String {
+    let dt = OffsetDateTime::from_unix_timestamp_nanos(i128::from(ms) * 1_000_000)
+        .unwrap_or_else(|_| OffsetDateTime::now_utc());
+    format!(
+        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z",
+        dt.year(),
+        u8::from(dt.month()),
+        dt.day(),
+        dt.hour(),
+        dt.minute(),
+        dt.second(),
+        dt.millisecond()
+    )
+}
+
 /// Epoch seconds, sub-second precision, of a catalog timestamp; 0 if it does
 /// not parse.
 #[allow(clippy::cast_precision_loss)] // epoch seconds fit f64's mantissa
@@ -93,6 +118,44 @@ pub(crate) fn is_fk_violation(e: &sqlx::Error) -> bool {
 mod tests {
     use super::*;
 
+    /// Same shape and width as the `stream_records.created_at` column default
+    /// (`strftime('%Y-%m-%dT%H:%M:%fZ')`), so rows from before and after the
+    /// explicit bind sort together as text.
+    #[test]
+    fn epoch_millis_matches_the_column_default_shape() {
+        let s = format_epoch_millis(1_792_765_147_123);
+        assert_eq!(s, "2026-10-23T14:19:07.123Z");
+        assert_eq!(s.len(), 24);
+        // Text order equals time order against a default-shaped neighbour.
+        assert!(s.as_str() < "2026-10-23T14:19:07.124Z");
+        assert!(s.as_str() > "2026-10-23T14:19:06.999Z");
+        // Against the retention cutoff (RFC 3339 from the `time` crate, no
+        // fraction on a whole second) the comparison is decided by the second
+        // digits in every case but the cutoff's own second, where `Z` sorts
+        // after `.`; that edge predates this formatter and is the default's
+        // behaviour too.
+        let cutoff = format_timestamp(OffsetDateTime::from_unix_timestamp(1_792_765_146).unwrap());
+        assert!(s.as_str() > cutoff.as_str(), "{s} vs {cutoff}");
+    }
+
+    /// A millisecond value `time` cannot represent is stamped with the current
+    /// time in the same shape, as the PostgreSQL engine stamps it, so the row
+    /// is not dated 1970 and swept at the next retention pass.
+    #[test]
+    fn epoch_millis_out_of_range_is_stamped_now() {
+        let before = OffsetDateTime::now_utc();
+        let s = format_epoch_millis(i64::MAX);
+        let after = OffsetDateTime::now_utc();
+        assert_eq!(s.len(), 24);
+        assert!(!s.starts_with("1970-"), "{s}");
+        let lo = format_epoch_millis((before.unix_timestamp_nanos() / 1_000_000) as i64);
+        let hi = format_epoch_millis((after.unix_timestamp_nanos() / 1_000_000) as i64);
+        assert!(
+            lo.as_str() <= s.as_str() && s.as_str() <= hi.as_str(),
+            "{lo} <= {s} <= {hi}"
+        );
+    }
+
     #[test]
     fn sqlite_url_variants() {
         assert_eq!(sqlite_url(":memory:"), "sqlite::memory:");
diff --git a/crates/storage-sqlite/src/stream.rs b/crates/storage-sqlite/src/stream.rs
index 7cd3d4729..5b00eabfb 100644
--- a/crates/storage-sqlite/src/stream.rs
+++ b/crates/storage-sqlite/src/stream.rs
@@ -93,15 +93,24 @@ impl StreamEngine for SqliteEngine {
             .map_err(|e| StorageError::Internal(e.to_string()))?;
 
             let _writer = self.write_lock.lock().await;
+            // created_at follows the record's own timestamp (whole seconds here,
+            // since that is all the record carries), never a second clock read.
+            let created_at = crate::sqlite_util::format_epoch_millis(
+                record
+                    .dynamodb
+                    .approximate_creation_date_time
+                    .saturating_mul(1_000),
+            );
             sqlx::query(
-                "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data) \
-                 VALUES (?, ?, ?, ?, ?)",
+                "INSERT INTO stream_records (sequence_number, shard_id, table_id, event_name, record_data, created_at) \
+                 VALUES (?, ?, ?, ?, ?, ?)",
             )
             .bind(&record.dynamodb.sequence_number)
             .bind(&shard_id)
             .bind(&table_id)
             .bind(format!("{:?}", record.event_name))
             .bind(&record_json)
+            .bind(created_at)
             .execute(&self.pool)
             .await
             .map_err(|e| StorageError::Internal(e.to_string()))?;