diff --git a/crates/core/src/types/stream.rs b/crates/core/src/types/stream.rs
index 870169b9a..41426e7cc 100755
--- a/crates/core/src/types/stream.rs
+++ b/crates/core/src/types/stream.rs
@@ -27,6 +27,9 @@ pub enum ShardIteratorType {
     Latest,
     AtSequenceNumber,
     AfterSequenceNumber,
+    /// Start at the first record whose creation time is at or after
+    /// `GetShardIteratorInput::timestamp`.
+    AtTimestamp,
 }
 
 /// Stream status.
@@ -201,6 +204,10 @@ pub struct GetShardIteratorInput {
     pub shard_iterator_type: ShardIteratorType,
     #[serde(rename = "SequenceNumber", default)]
     pub sequence_number: Option<String>,
+    /// Epoch seconds, fractional milliseconds allowed. Used with
+    /// `AT_TIMESTAMP`; the service refuses it with every other type.
+    #[serde(rename = "Timestamp", default)]
+    pub timestamp: Option<f64>,
 }
 
 /// `GetShardIterator` response.
diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs
index 74083fbfa..46278b5d7 100755
--- a/crates/engine/src/lib.rs
+++ b/crates/engine/src/lib.rs
@@ -33,6 +33,7 @@ mod scan;
 mod search_vectors;
 pub mod stream_capture;
 mod streams;
+mod streams_iterator_params;
 mod tagging;
 mod transact_get_items;
 mod transact_write_helpers;
diff --git a/crates/engine/src/streams.rs b/crates/engine/src/streams.rs
index c8ba419bc..5ba804d35 100755
--- a/crates/engine/src/streams.rs
+++ b/crates/engine/src/streams.rs
@@ -14,6 +14,7 @@ use serde_json::Value;
 
 use crate::OperationContext;
 use crate::serialize_output;
+use crate::streams_iterator_params as iterator_params;
 
 /// Handle `DescribeStream`.
 ///
@@ -75,6 +76,11 @@ pub async fn handle_list_streams(
 ///
 /// Encodes the shard ID and starting position into a base64 iterator token.
 ///
+/// Validation runs in the service's order, measured 2026-10-23 (us-east-1):
+/// the `ShardIteratorType` enum, then `Timestamp` presence or absence for the
+/// type, then `SequenceNumber` presence or absence for the type, then the
+/// future check, then stream and shard existence.
+///
 /// # Errors
 ///
 /// Returns [`DynamoDbError::ResourceNotFoundException`] if the stream/shard does not exist.
@@ -83,8 +89,36 @@ pub async fn handle_get_shard_iterator(
     body: Value,
     ctx: &OperationContext,
 ) -> Result<Value, DynamoDbError> {
-    let input: GetShardIteratorInput = serde_json::from_value(body)
-        .map_err(|e| DynamoDbError::SerializationException(e.to_string()))?;
+    // The service renders both ShardIteratorType violations without the
+    // offending value (measured 2026-10-23): a missing member is "Member must
+    // not be null", a bad member lists the set in this order.
+    let type_value = body.get("ShardIteratorType");
+    if type_value.is_none_or(serde_json::Value::is_null) {
+        return Err(DynamoDbError::ValidationException(
+            "1 validation error detected: Value at 'shardIteratorType' failed to satisfy \
+             constraint: Member must not be null"
+                .to_owned(),
+        ));
+    }
+    crate::validate_enum_fields(
+        &body,
+        &[crate::EnumField {
+            json_name: "ShardIteratorType",
+            valid: &[
+                "AFTER_SEQUENCE_NUMBER",
+                "LATEST",
+                "AT_TIMESTAMP",
+                "AT_SEQUENCE_NUMBER",
+                "TRIM_HORIZON",
+            ],
+            clause: crate::EnumClause::Bare(iterator_params::SHARD_ITERATOR_TYPE_ENUM_CLAUSE),
+        }],
+    )?;
+    iterator_params::check_timestamp_shape(&body)?;
+    let input: GetShardIteratorInput =
+        serde_json::from_value(body).map_err(crate::deserialize_error)?;
+
+    let at_timestamp_ms = iterator_params::validate_iterator_parameters(&input)?;
 
     // Validate the stream and shard exist before issuing an iterator.
     ctx.storage
@@ -92,6 +126,7 @@ pub async fn handle_get_shard_iterator(
         .await
         .map_err(storage_to_dynamo)?;
 
+    let width = ctx.storage.sequence_number_width();
     let seq = match input.shard_iterator_type {
         ShardIteratorType::TrimHorizon => String::new(),
         ShardIteratorType::Latest => {
@@ -104,33 +139,18 @@ pub async fn handle_get_shard_iterator(
                 .unwrap_or_default()
         }
         ShardIteratorType::AtSequenceNumber => {
-            // Convert to AFTER_SEQUENCE_NUMBER by subtracting 1, so the
-            // exclusive "after" semantics produce inclusive "at" behavior.
-            let raw = input.sequence_number.clone().ok_or_else(|| {
-                DynamoDbError::ValidationException(
-                    "SequenceNumber is required for AT_SEQUENCE_NUMBER iterator type".to_owned(),
-                )
-            })?;
-            let n = raw.parse::<u128>().map_err(|_| {
-                DynamoDbError::ValidationException("Invalid SequenceNumber".to_owned())
-            })?;
-            // n == 0: sequence 0 is the first possible record, so "at 0"
-            // means "read from the beginning" — same as TRIM_HORIZON.
-            if n > 0 {
-                // Pad to the backend's stored width so lexicographic order
-                // matches numeric order against stored sequence numbers.
-                let width = ctx.storage.sequence_number_width();
-                format!("{:0>width$}", n - 1)
-            } else {
-                String::new()
-            }
+            let raw = input.sequence_number.as_deref().unwrap_or_default();
+            iterator_params::at_sequence_to_after(raw, width)?
         }
-        ShardIteratorType::AfterSequenceNumber => {
-            input.sequence_number.clone().ok_or_else(|| {
-                DynamoDbError::ValidationException(
-                    "SequenceNumber is required for AFTER_SEQUENCE_NUMBER iterator type".to_owned(),
-                )
-            })?
+        ShardIteratorType::AfterSequenceNumber => input.sequence_number.clone().unwrap_or_default(),
+        ShardIteratorType::AtTimestamp => {
+            resolve_at_timestamp(
+                ctx.storage.as_ref(),
+                &input.shard_id,
+                at_timestamp_ms.unwrap_or_default(),
+                width,
+            )
+            .await?
         }
     };
 
@@ -140,8 +160,8 @@ pub async fn handle_get_shard_iterator(
     let type_str = "AFTER_SEQUENCE_NUMBER";
 
     // Encode creation timestamp (seconds since epoch) for 15-minute expiration.
-    // unwrap_or_default: returns epoch 0 if system clock is before 1970 — safe
-    // because the iterator would just expire immediately on the next GetRecords.
+    // unwrap_or_default: returns epoch 0 if the system clock is before 1970,
+    // which only makes the iterator expire on the next GetRecords.
     let created_at = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .unwrap_or_default()
@@ -161,6 +181,33 @@ pub async fn handle_get_shard_iterator(
     serialize_output(&output)
 }
 
+/// The AFTER position for an `AT_TIMESTAMP` request: at the first record
+/// created at or after `at_ms`, or after the newest record when none
+/// qualifies (the service's tip-plus-one resolution).
+///
+/// The tip is read before the seek. If the seek then finds nothing, the
+/// iterator starts after that tip, and a record landing between the two
+/// reads is still ahead of it. The other order would skip such a record.
+pub(crate) async fn resolve_at_timestamp(
+    storage: &dyn extenddb_storage::StreamEngine,
+    shard_id: &str,
+    at_ms: i64,
+    width: usize,
+) -> Result<String, DynamoDbError> {
+    let tip = storage
+        .latest_sequence_number(shard_id)
+        .await
+        .map_err(storage_to_dynamo)?;
+    let first = storage
+        .first_sequence_number_at_or_after(shard_id, at_ms)
+        .await
+        .map_err(storage_to_dynamo)?;
+    match first {
+        Some(first) => iterator_params::at_sequence_to_after(&first, width),
+        None => Ok(tip.unwrap_or_default()),
+    }
+}
+
 /// Shard iterator expiration: 15 minutes (900 seconds), matching real `DynamoDB`.
 const SHARD_ITERATOR_EXPIRY_SECS: u64 = 900;
 
@@ -419,32 +466,6 @@ fn storage_to_dynamo(e: StorageError) -> DynamoDbError {
 
 #[cfg(test)]
 mod tests {
-    /// AT_SEQUENCE_NUMBER converts to AFTER by subtracting 1 and padding to
-    /// the backend's stored width. Verify that an unpadded client input is
-    /// normalised correctly for both the 21-digit (postgres/sqlite/mongodb)
-    /// and 23-digit (cassandra) cases, and that the edge case n=0 maps to
-    /// TRIM_HORIZON (empty string).
-    #[test]
-    fn at_sequence_number_padding() {
-        let cases: &[(&str, usize, &str)] = &[
-            ("5", 21, "000000000000000000004"),
-            ("000000000000000000005", 21, "000000000000000000004"),
-            ("5", 23, "00000000000000000000004"),
-            ("00000000000000000000005", 23, "00000000000000000000004"),
-            ("0", 21, ""),
-            ("1", 21, "000000000000000000000"),
-        ];
-        for (input, width, expected) in cases {
-            let n: u128 = input.parse().unwrap();
-            let result = if n > 0 {
-                format!("{:0>width$}", n - 1)
-            } else {
-                String::new()
-            };
-            assert_eq!(&result, expected, "input={input} width={width}");
-        }
-    }
-
     #[test]
     fn shard_iterator_round_trip_backward_compat_and_fail_closed() {
         let stream = "arn:aws:dynamodb:us-east-1:1:table/T/stream/2026-10-20T16:17:40.188";
diff --git a/crates/engine/src/streams_iterator_params.rs b/crates/engine/src/streams_iterator_params.rs
new file mode 100644
index 000000000..623199740
--- /dev/null
+++ b/crates/engine/src/streams_iterator_params.rs
@@ -0,0 +1,463 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! `GetShardIterator` parameter rules and position conversions.
+//!
+//! Messages and the order of the checks were measured on the service
+//! 2026-10-23 (us-east-1); see `tests/test_streams_at_timestamp.py` for the
+//! wire-level expectations taken from the same run.
+
+use extenddb_core::error::DynamoDbError;
+use extenddb_core::types::{GetShardIteratorInput, ShardIteratorType};
+
+/// How the service renders a `ShardIteratorType` that is not a member: no
+/// value, and the set in this order. Measured 2026-10-23 (us-east-1) for an
+/// unknown name and for a lowercase one alike.
+pub(crate) const SHARD_ITERATOR_TYPE_ENUM_CLAUSE: &str = "Value at 'shardIteratorType' failed to \
+     satisfy constraint: Member must satisfy enum value set: [AFTER_SEQUENCE_NUMBER, LATEST, \
+     AT_TIMESTAMP, AT_SEQUENCE_NUMBER, TRIM_HORIZON]";
+
+/// Refuse a `Timestamp` whose JSON type is not a number with the text the
+/// service's deserializer renders for it (measured 2026-10-23): a string is
+/// `STRING_VALUE`, a boolean `TRUE_VALUE` or `FALSE_VALUE`, an array or object
+/// the structure message. Numbers and null pass through to the parameter
+/// rules, which treat null as absent.
+pub(crate) fn check_timestamp_shape(body: &serde_json::Value) -> Result<(), DynamoDbError> {
+    use serde_json::Value;
+    let Some(v) = body.get("Timestamp") else {
+        return Ok(());
+    };
+    let msg = match v {
+        Value::Null | Value::Number(_) => return Ok(()),
+        Value::String(_) => "STRING_VALUE cannot be converted to Date",
+        Value::Bool(true) => "TRUE_VALUE cannot be converted to Date",
+        Value::Bool(false) => "FALSE_VALUE cannot be converted to Date",
+        Value::Array(_) | Value::Object(_) => "Start of structure or map found where not expected",
+    };
+    Err(DynamoDbError::SerializationException(msg.to_owned()))
+}
+
+/// The service's prefix on every parameter-value refusal below.
+const INVALID_PARAMETER_PREFIX: &str = "One or more parameter values were invalid: ";
+
+/// A future `Timestamp` is accepted up to this far ahead of the server clock.
+/// Measured 2026-10-23 (us-east-1): now + 1 s accepted, now + 2 s refused.
+const AT_TIMESTAMP_FUTURE_BUFFER_MS: i64 = 1_000;
+
+/// Check the per-type parameter rules in the service's order and return the
+/// `AT_TIMESTAMP` position in milliseconds when that is the type.
+///
+/// Messages and order measured on the service 2026-10-23 (us-east-1).
+pub(crate) fn validate_iterator_parameters(
+    input: &GetShardIteratorInput,
+) -> Result<Option<i64>, DynamoDbError> {
+    let refuse = |msg: &str| {
+        Err(DynamoDbError::ValidationException(format!(
+            "{INVALID_PARAMETER_PREFIX}{msg}"
+        )))
+    };
+    let is_at_timestamp = input.shard_iterator_type == ShardIteratorType::AtTimestamp;
+    let takes_sequence = matches!(
+        input.shard_iterator_type,
+        ShardIteratorType::AtSequenceNumber | ShardIteratorType::AfterSequenceNumber
+    );
+
+    // Timestamp presence or absence for the type comes first.
+    let at_ms = if is_at_timestamp {
+        match input.timestamp {
+            Some(t) if t > 0.0 && t.is_finite() => Some(timestamp_to_millis(t)),
+            _ => {
+                return refuse(
+                    "Valid timestamp must be provided when the iterator type is AT_TIMESTAMP",
+                );
+            }
+        }
+    } else {
+        if input.timestamp.is_some() {
+            return refuse("Timestamp should not be passed for this iterator type");
+        }
+        None
+    };
+
+    // Then SequenceNumber presence or absence for the type.
+    if takes_sequence {
+        if input.sequence_number.is_none() {
+            return refuse("SequenceNumber cannot be null");
+        }
+    } else if input.sequence_number.is_some() {
+        return refuse("SequenceNumber should not be passed for this iterator type");
+    }
+
+    // Then the future check.
+    if let Some(at_ms) = at_ms {
+        let now_ms = i64::try_from(
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_millis(),
+        )
+        .unwrap_or(i64::MAX);
+        if at_ms > now_ms.saturating_add(AT_TIMESTAMP_FUTURE_BUFFER_MS) {
+            return refuse("Timestamp parameter cannot be greater than the current timestamp");
+        }
+    }
+    Ok(at_ms)
+}
+
+/// Epoch seconds with an optional fraction to whole milliseconds, floored.
+/// Saturates at the `i64` edge; the caller has already refused values that
+/// are not positive and finite.
+pub(crate) fn timestamp_to_millis(secs: f64) -> i64 {
+    let ms = (secs * 1000.0).floor();
+    if ms >= 9.2e18 {
+        i64::MAX
+    } else {
+        // Non-negative and in range after the guard above.
+        #[allow(clippy::cast_possible_truncation)]
+        let v = ms as i64;
+        v
+    }
+}
+
+/// Convert an AT_SEQUENCE_NUMBER position to the AFTER position the token
+/// carries: `n - 1`, zero-padded to the backend's stored width so lexicographic
+/// order matches numeric order. `n == 0` is the first possible record, so
+/// "at 0" reads from the beginning, the same as TRIM_HORIZON.
+pub(crate) fn at_sequence_to_after(raw: &str, width: usize) -> Result<String, DynamoDbError> {
+    let n = raw
+        .parse::<u128>()
+        .map_err(|_| DynamoDbError::ValidationException("Invalid SequenceNumber".to_owned()))?;
+    Ok(if n > 0 {
+        format!("{:0>width$}", n - 1)
+    } else {
+        String::new()
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use extenddb_core::error::DynamoDbError;
+    use extenddb_core::types::{GetShardIteratorInput, ShardIteratorType};
+
+    /// AT_SEQUENCE_NUMBER converts to AFTER by subtracting 1 and padding to
+    /// the backend's stored width. Verify that an unpadded client input is
+    /// normalised correctly for both the 21-digit (postgres/sqlite/mongodb)
+    /// and 23-digit (cassandra) cases, and that the edge case n=0 maps to
+    /// TRIM_HORIZON (empty string).
+    #[test]
+    fn at_sequence_number_padding() {
+        let cases: &[(&str, usize, &str)] = &[
+            ("5", 21, "000000000000000000004"),
+            ("000000000000000000005", 21, "000000000000000000004"),
+            ("5", 23, "00000000000000000000004"),
+            ("00000000000000000000005", 23, "00000000000000000000004"),
+            ("0", 21, ""),
+            ("1", 21, "000000000000000000000"),
+        ];
+        for (input, width, expected) in cases {
+            let result = super::at_sequence_to_after(input, *width).unwrap();
+            assert_eq!(&result, expected, "input={input} width={width}");
+        }
+        assert!(super::at_sequence_to_after("abc", 21).is_err());
+    }
+
+    /// Epoch seconds with a fraction floor to whole milliseconds, as the
+    /// service seeks at millisecond precision (measured: a `Timestamp` of
+    /// `.5` skipped records written at `.050`).
+    #[test]
+    fn timestamp_to_millis_floors_to_the_millisecond() {
+        assert_eq!(
+            super::timestamp_to_millis(1_792_764_248.0),
+            1_792_764_248_000
+        );
+        assert_eq!(
+            super::timestamp_to_millis(1_792_764_248.05),
+            1_792_764_248_050
+        );
+        assert_eq!(
+            super::timestamp_to_millis(1_792_764_248.999),
+            1_792_764_248_999
+        );
+        assert_eq!(
+            super::timestamp_to_millis(1_792_764_248.123_456),
+            1_792_764_248_123
+        );
+        // Floors, never rounds: .9996 ms stays at 123.
+        assert_eq!(
+            super::timestamp_to_millis(1_792_764_248.123_999_6),
+            1_792_764_248_123
+        );
+        assert_eq!(super::timestamp_to_millis(1e18), i64::MAX);
+    }
+
+    fn input(t: ShardIteratorType, seq: Option<&str>, ts: Option<f64>) -> GetShardIteratorInput {
+        GetShardIteratorInput {
+            stream_arn:
+                "arn:aws:dynamodb:us-east-1:123456789012:table/T/stream/2026-10-23T14:03:56.871"
+                    .to_owned(),
+            shard_id: "shardId-00000001792764241200-b719fef1".to_owned(),
+            shard_iterator_type: t,
+            sequence_number: seq.map(str::to_owned),
+            timestamp: ts,
+        }
+    }
+
+    fn message(r: Result<Option<i64>, DynamoDbError>) -> String {
+        match r {
+            Err(DynamoDbError::ValidationException(m)) => m,
+            other => panic!("expected ValidationException, got {other:?}"),
+        }
+    }
+
+    /// Messages, their prefix, and the order of the checks, as measured on
+    /// the service 2026-10-23 (us-east-1).
+    #[test]
+    fn parameter_validation_matches_the_service_order_and_text() {
+        use ShardIteratorType::{
+            AfterSequenceNumber, AtSequenceNumber, AtTimestamp, Latest, TrimHorizon,
+        };
+        const P: &str = "One or more parameter values were invalid: ";
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_secs_f64();
+        let past = 1_792_764_148.0;
+        let v = super::validate_iterator_parameters;
+
+        // AT_TIMESTAMP needs a positive Timestamp; null, zero and negative all refuse.
+        for ts in [None, Some(0.0), Some(-1.0), Some(f64::NAN)] {
+            assert_eq!(
+                message(v(&input(AtTimestamp, None, ts))),
+                format!(
+                    "{P}Valid timestamp must be provided when the iterator type is AT_TIMESTAMP"
+                ),
+                "ts={ts:?}"
+            );
+        }
+        // The Timestamp check runs before the SequenceNumber check.
+        assert_eq!(
+            message(v(&input(AtTimestamp, Some("1"), None))),
+            format!("{P}Valid timestamp must be provided when the iterator type is AT_TIMESTAMP")
+        );
+        // SequenceNumber with AT_TIMESTAMP refuses, and does so before the future check.
+        for ts in [past, now + 3_600.0] {
+            assert_eq!(
+                message(v(&input(AtTimestamp, Some("1"), Some(ts)))),
+                format!("{P}SequenceNumber should not be passed for this iterator type")
+            );
+        }
+        // Timestamp with any other type refuses, before the SequenceNumber check.
+        for t in [TrimHorizon, Latest, AtSequenceNumber, AfterSequenceNumber] {
+            for seq in [None, Some("1")] {
+                assert_eq!(
+                    message(v(&input(t, seq, Some(past)))),
+                    format!("{P}Timestamp should not be passed for this iterator type"),
+                    "type={t:?} seq={seq:?}"
+                );
+            }
+        }
+        // SequenceNumber rules for the four other types.
+        for t in [TrimHorizon, Latest] {
+            assert_eq!(
+                message(v(&input(t, Some("1"), None))),
+                format!("{P}SequenceNumber should not be passed for this iterator type")
+            );
+            assert_eq!(v(&input(t, None, None)).unwrap(), None);
+        }
+        for t in [AtSequenceNumber, AfterSequenceNumber] {
+            assert_eq!(
+                message(v(&input(t, None, None))),
+                format!("{P}SequenceNumber cannot be null")
+            );
+            assert_eq!(v(&input(t, Some("1"), None)).unwrap(), None);
+        }
+        // A far-future Timestamp refuses; a past one resolves to milliseconds.
+        assert_eq!(
+            message(v(&input(AtTimestamp, None, Some(now + 3_600.0)))),
+            format!("{P}Timestamp parameter cannot be greater than the current timestamp")
+        );
+        assert_eq!(
+            v(&input(AtTimestamp, None, Some(past))).unwrap(),
+            Some(1_792_764_148_000)
+        );
+    }
+
+    /// Malformed `Timestamp` JSON types render the service's deserializer
+    /// texts (measured); numbers and null pass to the parameter rules.
+    #[test]
+    fn timestamp_shape_texts_match_the_service() {
+        use serde_json::json;
+        let msg =
+            |v: serde_json::Value| match super::check_timestamp_shape(&json!({"Timestamp": v})) {
+                Err(DynamoDbError::SerializationException(m)) => m,
+                other => panic!("expected SerializationException, got {other:?}"),
+            };
+        assert_eq!(
+            msg(json!("1792764248.5")),
+            "STRING_VALUE cannot be converted to Date"
+        );
+        assert_eq!(msg(json!("")), "STRING_VALUE cannot be converted to Date");
+        assert_eq!(msg(json!(true)), "TRUE_VALUE cannot be converted to Date");
+        assert_eq!(msg(json!(false)), "FALSE_VALUE cannot be converted to Date");
+        assert_eq!(
+            msg(json!({"S": "x"})),
+            "Start of structure or map found where not expected"
+        );
+        assert_eq!(
+            msg(json!([1])),
+            "Start of structure or map found where not expected"
+        );
+        assert!(super::check_timestamp_shape(&json!({"Timestamp": 1.5})).is_ok());
+        assert!(super::check_timestamp_shape(&json!({"Timestamp": null})).is_ok());
+        assert!(super::check_timestamp_shape(&json!({})).is_ok());
+    }
+
+    /// The future buffer is about one second: now + 0.5 s is accepted, now +
+    /// 2 s is refused (measured; the 1 s to 2 s band depends on clock skew).
+    #[test]
+    fn future_buffer_admits_half_a_second_and_refuses_two() {
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_secs_f64();
+        let v = super::validate_iterator_parameters;
+        assert!(
+            v(&input(
+                ShardIteratorType::AtTimestamp,
+                None,
+                Some(now + 0.5)
+            ))
+            .is_ok()
+        );
+        assert!(
+            v(&input(
+                ShardIteratorType::AtTimestamp,
+                None,
+                Some(now + 2.0)
+            ))
+            .is_err()
+        );
+    }
+
+    /// A write that lands while the tip is being read is still ahead of the
+    /// resolved position. The double performs that write inside
+    /// `latest_sequence_number` and reports the tip including it. Tip first:
+    /// the seek then finds the new record and the iterator positions AT it.
+    /// Seek first: the seek finds nothing, the tip read then includes the new
+    /// record, and the iterator starts after it, skipping it.
+    #[tokio::test]
+    async fn at_timestamp_positions_before_a_write_that_lands_between_the_two_reads() {
+        use std::sync::Mutex;
+
+        use extenddb_storage::BoxedFuture as BoxFuture;
+        use extenddb_storage::error::StorageError;
+        use extenddb_storage::{StreamEngine, StreamListResult, StreamRecordsResult};
+
+        struct RacingShard {
+            /// Sequence numbers in the shard, with their creation ms.
+            records: Mutex<Vec<(String, i64)>>,
+            /// The record the racing writer adds during the tip read.
+            late: (String, i64),
+        }
+        impl StreamEngine for RacingShard {
+            fn write_stream_record(
+                &self,
+                _: &str,
+                _: &extenddb_core::types::StreamRecord,
+                _: &str,
+                _: &str,
+            ) -> BoxFuture<'_, Result<(), StorageError>> {
+                unreachable!()
+            }
+            fn get_stream_records(
+                &self,
+                _: &str,
+                _: &str,
+                _: &str,
+                _: Option<&str>,
+                _: i64,
+            ) -> BoxFuture<'_, StreamRecordsResult> {
+                unreachable!()
+            }
+            fn describe_stream(
+                &self,
+                _: &str,
+                _: &extenddb_core::types::DescribeStreamInput,
+            ) -> BoxFuture<'_, Result<extenddb_core::types::StreamDescription, StorageError>>
+            {
+                unreachable!()
+            }
+            fn list_streams(
+                &self,
+                _: &str,
+                _: Option<&str>,
+                _: i64,
+                _: Option<&str>,
+            ) -> BoxFuture<'_, StreamListResult> {
+                unreachable!()
+            }
+            fn cleanup_expired_stream_records(
+                &self,
+                _: i64,
+            ) -> BoxFuture<'_, Result<u64, StorageError>> {
+                unreachable!()
+            }
+            fn assign_shard(
+                &self,
+                _: &str,
+                _: &str,
+                _: &str,
+            ) -> BoxFuture<'_, Result<String, StorageError>> {
+                unreachable!()
+            }
+            fn next_sequence_number(&self, _: &str) -> BoxFuture<'_, Result<String, StorageError>> {
+                unreachable!()
+            }
+            fn validate_shard(
+                &self,
+                _: &str,
+                _: &str,
+                _: &str,
+            ) -> BoxFuture<'_, Result<(), StorageError>> {
+                Box::pin(async { Ok(()) })
+            }
+            fn latest_sequence_number(
+                &self,
+                _: &str,
+            ) -> BoxFuture<'_, Result<Option<String>, StorageError>> {
+                Box::pin(async move {
+                    let mut recs = self.records.lock().unwrap();
+                    // The racing write lands while this read runs; the store
+                    // reports the tip as it stands, which includes it.
+                    recs.push(self.late.clone());
+                    Ok(recs.last().map(|(s, _)| s.clone()))
+                })
+            }
+            fn first_sequence_number_at_or_after(
+                &self,
+                _: &str,
+                at_ms: i64,
+            ) -> BoxFuture<'_, Result<Option<String>, StorageError>> {
+                Box::pin(async move {
+                    let recs = self.records.lock().unwrap();
+                    Ok(recs
+                        .iter()
+                        .filter(|(_, ms)| *ms >= at_ms)
+                        .map(|(s, _)| s.clone())
+                        .min())
+                })
+            }
+        }
+
+        let shard = RacingShard {
+            records: Mutex::new(vec![("000000000000000000005".to_owned(), 1_000)]),
+            late: ("000000000000000000006".to_owned(), 2_500),
+        };
+        // T is after the only stored record and before the racing write.
+        let pos = crate::streams::resolve_at_timestamp(&shard, "shardId-x", 2_000, 21)
+            .await
+            .unwrap();
+        // AT record 6 is encoded as AFTER record 5: GetRecords returns 6 first.
+        assert_eq!(pos, "000000000000000000005");
+    }
+}
diff --git a/crates/storage-cassandra/src/lib.rs b/crates/storage-cassandra/src/lib.rs
index 7dbfe98c4..c38891fa5 100644
--- a/crates/storage-cassandra/src/lib.rs
+++ b/crates/storage-cassandra/src/lib.rs
@@ -28,6 +28,7 @@ pub mod migrations;
 pub mod operations;
 pub(crate) mod propagation_hold;
 mod snapshot_engine;
+pub mod stream_at_timestamp;
 mod stream_engine;
 pub mod stream_util;
 pub mod table_engine;
diff --git a/crates/storage-cassandra/src/stream_at_timestamp.rs b/crates/storage-cassandra/src/stream_at_timestamp.rs
new file mode 100644
index 000000000..5654744e3
--- /dev/null
+++ b/crates/storage-cassandra/src/stream_at_timestamp.rs
@@ -0,0 +1,85 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! `AT_TIMESTAMP` seek: the first stream record in a shard created at or
+//! after a point in time.
+//!
+//! `created_at` is not a clustering column, so the seek uses the sequence
+//! number's millisecond prefix as a clustering-key lower bound and then
+//! filters `created_at` within that single partition. For every record
+//! `created_at_ms <= hlc_ms`: the `HybridClock` clamps its physical
+//! component upward and never below the wall-clock read that `created_at`
+//! stores. So `created_at_ms >= at_ms` implies `hlc_ms >= at_ms`, and every
+//! qualifying record sits at or after the bound; the seek never skips one.
+//! The filter removes the other case: a record written under a clamped clock
+//! (after a backwards step) whose prefix clears the bound while its
+//! `created_at` is before `at_ms`. The filter walks at most that clamp
+//! window, which is zero under a monotonic clock.
+
+use cdrs_tokio::types::IntoRustByName;
+use extenddb_storage::error::StorageError;
+
+use crate::engine::CassandraEngine;
+
+/// The smallest sequence number whose millisecond prefix is `at_ms`.
+#[must_use]
+pub fn sequence_lower_bound(at_ms: i64) -> String {
+    format!("{:013}{:06}{:04}", at_ms.max(0), 0, 0)
+}
+
+impl CassandraEngine {
+    pub(crate) async fn first_sequence_number_at_or_after_impl(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> Result<Option<String>, StorageError> {
+        let keyspace = self.account_keyspace_for_shard(shard_id).await?;
+        let bound = sequence_lower_bound(at_ms);
+        // A single-partition clustering range with one non-key filter; the
+        // rows come back in clustering (sequence) order, so the first is the
+        // minimum qualifying sequence.
+        let query = format!(
+            "SELECT sequence_number FROM {keyspace}.stream_records \
+             WHERE shard_id = ? AND sequence_number >= ? AND created_at >= ? \
+             LIMIT 1 ALLOW FILTERING"
+        );
+        let result = self
+            .session
+            .query_with_values(
+                &query,
+                cdrs_tokio::query_values!(shard_id, bound.as_str(), at_ms),
+            )
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
+        let rows = result
+            .response_body()
+            .map_err(|e| StorageError::Internal(e.to_string()))?
+            .into_rows()
+            .unwrap_or_default();
+        Ok(rows
+            .into_iter()
+            .next()
+            .and_then(|row| row.get_r_by_name("sequence_number").ok()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::sequence_lower_bound;
+
+    #[test]
+    fn lower_bound_is_the_first_sequence_in_that_millisecond() {
+        let b = sequence_lower_bound(1_792_764_248_050);
+        assert_eq!(b, "17927642480500000000000");
+        assert_eq!(b.len(), 23);
+        // Any real sequence in that millisecond sorts at or after the bound,
+        // and every sequence from the previous millisecond sorts before it.
+        assert!("17927642480500000010042" > b.as_str());
+        assert!("17927642480499999990042" < b.as_str());
+    }
+
+    #[test]
+    fn negative_timestamps_clamp_to_the_epoch() {
+        assert_eq!(sequence_lower_bound(-5), "00000000000000000000000");
+    }
+}
diff --git a/crates/storage-cassandra/src/stream_engine.rs b/crates/storage-cassandra/src/stream_engine.rs
index 8c83b0927..62bbea71b 100755
--- a/crates/storage-cassandra/src/stream_engine.rs
+++ b/crates/storage-cassandra/src/stream_engine.rs
@@ -24,7 +24,10 @@ impl CassandraEngine {
     /// (embedded in the shard ID) via the secondary index on `tables.table_id`.
     ///
     /// Shard ID format: `shardId-{table_id}-{index:012}`
-    async fn account_keyspace_for_shard(&self, shard_id: &str) -> Result<String, StorageError> {
+    pub(crate) async fn account_keyspace_for_shard(
+        &self,
+        shard_id: &str,
+    ) -> Result<String, StorageError> {
         // Strip "shardId-" prefix and trailing "-{12 digits}" to get table_id.
         let without_prefix = shard_id.strip_prefix("shardId-").ok_or_else(|| {
             StorageError::Internal(format!("Invalid shard_id format: {shard_id}"))
@@ -595,4 +598,16 @@ impl StreamEngine for CassandraEngine {
                 .map(|row| row.get_r_by_name("sequence_number").unwrap_or_default()))
         })
     }
+
+    fn first_sequence_number_at_or_after(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> BoxFuture<'_, Result<Option<String>, StorageError>> {
+        let shard_id = shard_id.to_owned();
+        Box::pin(async move {
+            self.first_sequence_number_at_or_after_impl(&shard_id, at_ms)
+                .await
+        })
+    }
 }
diff --git a/crates/storage-cassandra/tests/direct/streams.rs b/crates/storage-cassandra/tests/direct/streams.rs
index bc0efa206..075d8cf2e 100644
--- a/crates/storage-cassandra/tests/direct/streams.rs
+++ b/crates/storage-cassandra/tests/direct/streams.rs
@@ -742,3 +742,111 @@ async fn test_stream_record_created_at_matches_the_record_and_sequence() {
         );
     }
 }
+
+/// A record written under a clamped clock (after a backwards step) carries a
+/// sequence prefix later than its `created_at`. The `AT_TIMESTAMP` seek must
+/// not return it for a `T` after its `created_at`, and must still return a
+/// record created at or after `T`.
+#[tokio::test]
+async fn test_at_timestamp_seek_skips_a_clamped_record_older_than_t() {
+    use extenddb_storage::StreamEngine;
+
+    if crate::helpers::skip_without_cassandra() {
+        return;
+    }
+    let engine = setup_engine().await;
+    let table = TestTable::new(&engine, "StreamClampSeek", false).await;
+    let view = StreamViewType::NewAndOldImages;
+
+    let mut item = BTreeMap::new();
+    item.insert("id".to_string(), AttributeValue::S("pk-clamp".to_string()));
+    item.insert("val".to_string(), AttributeValue::S("real".to_string()));
+    engine
+        .put_item(
+            &table.key_info,
+            item,
+            false,
+            None,
+            &Default::default(),
+            Some(&capture(view)),
+        )
+        .await
+        .unwrap();
+
+    let shard_id = shard_for("pk-clamp", &table.key_info.table_id);
+    let keyspace = format!("extenddb_ttl_test_account_{}", table.key_info.account_id);
+    let rows = engine
+        .session_arc()
+        .query_with_values(
+            &format!(
+                "SELECT sequence_number, created_at, record_data FROM {keyspace}.stream_records \
+                 WHERE shard_id = ?"
+            ),
+            cdrs_tokio::query_values!(shard_id.as_str()),
+        )
+        .await
+        .unwrap()
+        .response_body()
+        .unwrap()
+        .into_rows()
+        .unwrap_or_default();
+    assert_eq!(rows.len(), 1);
+    let real_seq: String = rows[0].get_r_by_name("sequence_number").unwrap();
+    let real_created: i64 = rows[0].get_r_by_name("created_at").unwrap();
+    let data: String = rows[0].get_r_by_name("record_data").unwrap();
+
+    // Model the clamped record: sequence prefix 8 s after the real record,
+    // created_at 8 s before it. T sits between the two: after the clamped
+    // created_at, before the real record's.
+    let clamped_hlc = real_created + 8_000;
+    let clamped_created = real_created - 8_000;
+    let clamped_seq = format!("{clamped_hlc:013}{:06}{:04}", 0, 9999);
+    engine
+        .session_arc()
+        .query_with_values(
+            &format!(
+                "INSERT INTO {keyspace}.stream_records \
+                 (shard_id, sequence_number, table_id, event_name, record_data, created_at) \
+                 VALUES (?, ?, ?, 'Insert', ?, ?)"
+            ),
+            cdrs_tokio::query_values!(
+                shard_id.as_str(),
+                clamped_seq.as_str(),
+                table.key_info.table_id.as_str(),
+                data.as_str(),
+                clamped_created
+            ),
+        )
+        .await
+        .unwrap();
+
+    // T after the clamped created_at and before the real record: the real
+    // record qualifies, the clamped one does not.
+    let t = real_created - 1_000;
+    let got = engine
+        .first_sequence_number_at_or_after(&shard_id, t)
+        .await
+        .unwrap();
+    assert_eq!(
+        got.as_deref(),
+        Some(real_seq.as_str()),
+        "the real record is first at or after T"
+    );
+
+    // T after every created_at: nothing qualifies, even though the clamped
+    // sequence prefix clears the bound.
+    let t = real_created + 1_000;
+    let got = engine
+        .first_sequence_number_at_or_after(&shard_id, t)
+        .await
+        .unwrap();
+    assert_eq!(got, None, "a clamped record older than T is not returned");
+
+    // T before everything: the clamped record is the first, as its
+    // created_at is the oldest.
+    let got = engine
+        .first_sequence_number_at_or_after(&shard_id, clamped_created - 1)
+        .await
+        .unwrap();
+    assert!(got.is_some());
+}
diff --git a/crates/storage-mongodb/src/bootstrapper.rs b/crates/storage-mongodb/src/bootstrapper.rs
index c37c3f36f..4487651e6 100644
--- a/crates/storage-mongodb/src/bootstrapper.rs
+++ b/crates/storage-mongodb/src/bootstrapper.rs
@@ -183,6 +183,20 @@ impl Bootstrapper for MongoBootstrapper {
             .await
             .map_err(|e| OpError::Internal(format!("stream_records shard index: {e}")))?;
 
+        // (shard_id, created_at) serves the AT_TIMESTAMP seek, which filters
+        // one shard by creation time and takes the lowest sequence. Without
+        // it the (shard_id, sequence_number) index walks every document up to
+        // the first match: 84 ms over 50,000 documents measured on one shard
+        // of 100,000.
+        stream_records
+            .create_index(
+                IndexModel::builder()
+                    .keys(doc! { "shard_id": 1, "created_at": 1 })
+                    .build(),
+            )
+            .await
+            .map_err(|e| OpError::Internal(format!("stream_records created_at index: {e}")))?;
+
         Ok(())
     }
 
diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs
index b27cc0baf..fef02e4c3 100644
--- a/crates/storage-mongodb/src/lib.rs
+++ b/crates/storage-mongodb/src/lib.rs
@@ -22,6 +22,7 @@ mod metadata_engine;
 mod operations;
 pub mod pushdown;
 mod snapshot_engine;
+mod stream_at_timestamp;
 mod stream_engine;
 mod table_engine;
 mod ttl_worker;
diff --git a/crates/storage-mongodb/src/stream_at_timestamp.rs b/crates/storage-mongodb/src/stream_at_timestamp.rs
new file mode 100644
index 000000000..964fe0ab1
--- /dev/null
+++ b/crates/storage-mongodb/src/stream_at_timestamp.rs
@@ -0,0 +1,44 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! `AT_TIMESTAMP` seek: the first stream record in a shard created at or
+//! after a point in time.
+
+use extenddb_storage::error::StorageError;
+use futures::TryStreamExt;
+use mongodb::bson::{DateTime as BsonDateTime, Document, doc};
+use mongodb::options::FindOptions;
+
+use crate::MongoEngine;
+
+impl MongoEngine {
+    pub(crate) async fn first_sequence_number_at_or_after_impl(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> Result<Option<String>, StorageError> {
+        let records_coll = self.data_db.collection::<Document>("stream_records");
+        let opts = FindOptions::builder()
+            .sort(doc! { "sequence_number": 1 })
+            .limit(1)
+            .projection(doc! { "sequence_number": 1 })
+            .build();
+        let cursor = records_coll
+            .find(doc! {
+                "shard_id": shard_id,
+                "created_at": { "$gte": BsonDateTime::from_millis(at_ms) },
+            })
+            .with_options(opts)
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
+        let docs: Vec<Document> = cursor
+            .try_collect()
+            .await
+            .map_err(|e| StorageError::Internal(e.to_string()))?;
+        Ok(docs.first().and_then(|d| {
+            d.get_str("sequence_number")
+                .ok()
+                .map(std::borrow::ToOwned::to_owned)
+        }))
+    }
+}
diff --git a/crates/storage-mongodb/src/stream_engine.rs b/crates/storage-mongodb/src/stream_engine.rs
index 097b3bb2a..85e5392e1 100644
--- a/crates/storage-mongodb/src/stream_engine.rs
+++ b/crates/storage-mongodb/src/stream_engine.rs
@@ -751,6 +751,18 @@ impl StreamEngine for MongoEngine {
             }))
         })
     }
+
+    fn first_sequence_number_at_or_after(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> BoxFuture<'_, Result<Option<String>, StorageError>> {
+        let shard_id = shard_id.to_owned();
+        Box::pin(async move {
+            self.first_sequence_number_at_or_after_impl(&shard_id, at_ms)
+                .await
+        })
+    }
 }
 
 #[cfg(test)]
diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs
index 9603d68ff..a90a5a9a3 100755
--- a/crates/storage-postgres/src/lib.rs
+++ b/crates/storage-postgres/src/lib.rs
@@ -27,6 +27,7 @@ mod pg_util;
 mod snapshot_engine;
 #[doc(hidden)]
 pub use snapshot_engine::map_fetch_error;
+mod stream_at_timestamp;
 mod stream_engine;
 #[doc(hidden)]
 pub mod stream_records_sql;
diff --git a/crates/storage-postgres/src/stream_at_timestamp.rs b/crates/storage-postgres/src/stream_at_timestamp.rs
new file mode 100644
index 000000000..1f43a12b2
--- /dev/null
+++ b/crates/storage-postgres/src/stream_at_timestamp.rs
@@ -0,0 +1,31 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! `AT_TIMESTAMP` seek: the first stream record in a shard created at or
+//! after a point in time.
+
+use extenddb_storage::error::StorageError;
+
+use crate::PostgresEngine;
+
+impl PostgresEngine {
+    pub(crate) async fn first_sequence_number_at_or_after_impl(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> Result<Option<String>, StorageError> {
+        let bound = time::OffsetDateTime::from_unix_timestamp_nanos(i128::from(at_ms) * 1_000_000)
+            .map_err(|e| StorageError::Validation(format!("Invalid Timestamp: {e}")))?;
+        // MIN over no rows is one NULL row, so the column is read as Option.
+        let row: Option<(Option<String>,)> = sqlx::query_as(
+            "SELECT MIN(sequence_number) FROM stream_records \
+             WHERE shard_id = $1 AND created_at >= $2",
+        )
+        .bind(shard_id)
+        .bind(bound)
+        .fetch_optional(&self.data_pool)
+        .await
+        .map_err(|e| StorageError::Internal(e.to_string()))?;
+        Ok(row.and_then(|(s,)| s))
+    }
+}
diff --git a/crates/storage-postgres/src/stream_engine.rs b/crates/storage-postgres/src/stream_engine.rs
index da7fd122d..0a546a717 100755
--- a/crates/storage-postgres/src/stream_engine.rs
+++ b/crates/storage-postgres/src/stream_engine.rs
@@ -616,4 +616,16 @@ impl StreamEngine for PostgresEngine {
             Ok(row.map(|(s,)| s))
         })
     }
+
+    fn first_sequence_number_at_or_after(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> BoxFuture<'_, Result<Option<String>, StorageError>> {
+        let shard_id = shard_id.to_owned();
+        Box::pin(async move {
+            self.first_sequence_number_at_or_after_impl(&shard_id, at_ms)
+                .await
+        })
+    }
 }
diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs
index 7e8085ff9..2ce3b45fe 100644
--- a/crates/storage-sqlite/src/lib.rs
+++ b/crates/storage-sqlite/src/lib.rs
@@ -41,6 +41,7 @@ mod snapshot_engine;
 mod sqlite_util;
 mod store;
 mod stream;
+mod stream_at_timestamp;
 #[cfg(test)]
 mod stream_tests;
 mod table_engine;
diff --git a/crates/storage-sqlite/src/schema.rs b/crates/storage-sqlite/src/schema.rs
index d04ee5b56..a6c5a974b 100644
--- a/crates/storage-sqlite/src/schema.rs
+++ b/crates/storage-sqlite/src/schema.rs
@@ -181,6 +181,8 @@ CREATE TABLE IF NOT EXISTS stream_records (
 );
 
 CREATE INDEX IF NOT EXISTS idx_stream_records_created ON stream_records (created_at);
+-- Serves the AT_TIMESTAMP seek: one shard filtered by creation time.
+CREATE INDEX IF NOT EXISTS idx_stream_records_shard_created ON stream_records (shard_id, created_at);
 
 -- Admin users.
 CREATE TABLE IF NOT EXISTS admin_users (
diff --git a/crates/storage-sqlite/src/stream.rs b/crates/storage-sqlite/src/stream.rs
index 5b00eabfb..719b78914 100644
--- a/crates/storage-sqlite/src/stream.rs
+++ b/crates/storage-sqlite/src/stream.rs
@@ -517,4 +517,16 @@ impl StreamEngine for SqliteEngine {
             Ok(row.map(|(s,)| s))
         })
     }
+
+    fn first_sequence_number_at_or_after(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> BoxFuture<'_, Result<Option<String>, StorageError>> {
+        let shard_id = shard_id.to_owned();
+        Box::pin(async move {
+            self.first_sequence_number_at_or_after_impl(&shard_id, at_ms)
+                .await
+        })
+    }
 }
diff --git a/crates/storage-sqlite/src/stream_at_timestamp.rs b/crates/storage-sqlite/src/stream_at_timestamp.rs
new file mode 100644
index 000000000..04e903249
--- /dev/null
+++ b/crates/storage-sqlite/src/stream_at_timestamp.rs
@@ -0,0 +1,33 @@
+// Copyright 2026 ExtendDB contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! `AT_TIMESTAMP` seek: the first stream record in a shard created at or
+//! after a point in time.
+
+use extenddb_storage::error::StorageError;
+
+use crate::SqliteEngine;
+use crate::sqlite_util::format_epoch_millis;
+
+impl SqliteEngine {
+    /// `created_at` is stored as fixed-width millisecond text in the column
+    /// default's shape, so a bound in the same shape compares chronologically.
+    pub(crate) async fn first_sequence_number_at_or_after_impl(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> Result<Option<String>, StorageError> {
+        let bound = format_epoch_millis(at_ms);
+        // MIN over no rows is one NULL row, so the column is read as Option.
+        let row: Option<(Option<String>,)> = sqlx::query_as(
+            "SELECT MIN(sequence_number) FROM stream_records \
+             WHERE shard_id = ? AND created_at >= ?",
+        )
+        .bind(shard_id)
+        .bind(&bound)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(|e| StorageError::Internal(e.to_string()))?;
+        Ok(row.and_then(|(s,)| s))
+    }
+}
diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs
index b7aec12ff..529a6da21 100755
--- a/crates/storage/src/lib.rs
+++ b/crates/storage/src/lib.rs
@@ -689,6 +689,19 @@ pub trait StreamEngine: Send + Sync {
         &self,
         shard_id: &str,
     ) -> BoxFuture<'_, Result<Option<String>, StorageError>>;
+
+    /// Sequence number of the first record in the shard whose creation time
+    /// is at or after `at_ms` (milliseconds since the Unix epoch), or `None`
+    /// when no stored record qualifies.
+    ///
+    /// Used by `GetShardIterator` with `AT_TIMESTAMP`. The comparison is
+    /// against the stored `created_at`, which every backend takes from the
+    /// same clock read as the record's `ApproximateCreationDateTime`.
+    fn first_sequence_number_at_or_after(
+        &self,
+        shard_id: &str,
+        at_ms: i64,
+    ) -> BoxFuture<'_, Result<Option<String>, StorageError>>;
 }
 
 /// Background worker operations that require storage access.
diff --git a/docs/design/07-component-streams.md b/docs/design/07-component-streams.md
index dbff306b5..ac2468d4d 100755
--- a/docs/design/07-component-streams.md
+++ b/docs/design/07-component-streams.md
@@ -18,7 +18,7 @@ Four operations, served on the same HTTP endpoint as DynamoDB operations:
 |-----------|-------------|
 | `DescribeStream` | Return stream ARN, status, shard list with parent/child relationships |
 | `ListStreams` | List streams, optionally filtered by table name |
-| `GetShardIterator` | Get an iterator for a shard (TRIM_HORIZON, LATEST, AT_SEQUENCE_NUMBER, AFTER_SEQUENCE_NUMBER) |
+| `GetShardIterator` | Get an iterator for a shard (TRIM_HORIZON, LATEST, AT_SEQUENCE_NUMBER, AFTER_SEQUENCE_NUMBER, AT_TIMESTAMP) |
 | `GetRecords` | Read up to 1000 records (or 1 MB) from a shard using an iterator |
 
 **Note:** In real DynamoDB, Streams is a separate service with its own endpoint (`streams.dynamodb.<region>.amazonaws.com`) and its own SDK client (`boto3.client('dynamodbstreams')` in Python, `DynamoDbStreamsClient` in Java). extenddb serves both services on a single port. This works because SDK clients accept `endpoint_url` overrides — the user sets both `AWS_ENDPOINT_URL_DYNAMODB` and `AWS_ENDPOINT_URL_DYNAMODB_STREAMS` to the same extenddb address. The `X-Amz-Target` prefix distinguishes the two services: DynamoDB operations use `DynamoDB_20120810.<Op>`, Streams operations use `DynamoDBStreams_20120810.<Op>`. Both use `dynamodb` as the SigV4 signing name, so authentication is identical. See `01-requirements.md` §11 and `08-component-config.md` §10 for SDK configuration details.
diff --git a/docs/differences-from-dynamodb.md b/docs/differences-from-dynamodb.md
index 80c1f671b..68fd37ff7 100755
--- a/docs/differences-from-dynamodb.md
+++ b/docs/differences-from-dynamodb.md
@@ -49,6 +49,13 @@ adaptation when switching between ExtendDB and the real service.
 | Table creation delay | Returns `CREATING` immediately; transitions to `ACTIVE` typically within seconds. Same behavior for on-demand and provisioned | Configurable via `control_plane_delay_seconds` runtime setting (default: 5s) |
 | DeletionProtectionEnabled | Enforced | Enforced (accepted and stored, DeleteTable rejects when enabled) |
 
+## Streams
+
+| Area | DynamoDB | ExtendDB |
+|------|----------|------|
+| Shards | Shards open and close over time, with parent and child shards | A fixed number of shards per table (default 4), assigned by partition key hash; shards never close |
+| `AT_TIMESTAMP` iterator | The first record created at or after `Timestamp`, compared at millisecond precision; older than retention resolves to the oldest record; past the newest record resolves to after it; more than about a second in the future is a `ValidationException` | The same, on all four backends. The seek compares the record's stored creation time, taken from the same clock read as its `ApproximateCreationDateTime`. The future buffer is one second against the server clock |
+
 ## Time to Live (TTL)
 
 | Area | DynamoDB | ExtendDB |
diff --git a/docs/manuals/03-usage-guide.md b/docs/manuals/03-usage-guide.md
index 861f6bba9..9d8c5e84e 100755
--- a/docs/manuals/03-usage-guide.md
+++ b/docs/manuals/03-usage-guide.md
@@ -391,6 +391,15 @@ aws dynamodbstreams get-shard-iterator \
     --shard-id "shard-0" \
     --shard-iterator-type TRIM_HORIZON
 
+# Get a shard iterator positioned at the first record created at or after a
+# point in time (epoch seconds, fractional milliseconds allowed)
+aws dynamodbstreams get-shard-iterator \
+    --endpoint-url https://127.0.0.1:18443 \
+    --stream-arn "<stream-arn>" \
+    --shard-id "shard-0" \
+    --shard-iterator-type AT_TIMESTAMP \
+    --timestamp 1792764248.050
+
 # Read records
 aws dynamodbstreams get-records \
     --endpoint-url https://127.0.0.1:18443 \
diff --git a/samples/stream_consumer.py b/samples/stream_consumer.py
index b360f82b6..2b254c6ae 100755
--- a/samples/stream_consumer.py
+++ b/samples/stream_consumer.py
@@ -125,11 +125,23 @@ def refresh_shards():
         for shard in resp["StreamDescription"]["Shards"]:
             sid = shard["ShardId"]
             if sid not in shard_iterators:
-                it = streams_client.get_shard_iterator(
-                    StreamArn=stream_arn,
-                    ShardId=sid,
-                    ShardIteratorType="TRIM_HORIZON",
-                )
+                # EXTENDDB_STREAM_SINCE (epoch seconds) starts each shard at
+                # the first record created at or after that time instead of
+                # the oldest retained record.
+                since = os.environ.get("EXTENDDB_STREAM_SINCE")
+                if since:
+                    it = streams_client.get_shard_iterator(
+                        StreamArn=stream_arn,
+                        ShardId=sid,
+                        ShardIteratorType="AT_TIMESTAMP",
+                        Timestamp=float(since),
+                    )
+                else:
+                    it = streams_client.get_shard_iterator(
+                        StreamArn=stream_arn,
+                        ShardId=sid,
+                        ShardIteratorType="TRIM_HORIZON",
+                    )
                 shard_iterators[sid] = it["ShardIterator"]
                 log("POLLER", f"Tracking shard {sid}")
 
diff --git a/tests/test_streams_at_timestamp.py b/tests/test_streams_at_timestamp.py
new file mode 100644
index 000000000..10811b02f
--- /dev/null
+++ b/tests/test_streams_at_timestamp.py
@@ -0,0 +1,399 @@
+# Copyright 2026 ExtendDB contributors
+# SPDX-License-Identifier: Apache-2.0
+
+"""DynamoDB Streams AT_TIMESTAMP shard iterator.
+
+Every expected value here was measured on the service (us-east-1,
+2026-10-23): positioning at millisecond precision, the four validation
+messages and their order, the future buffer, and the error classes for
+malformed Timestamp values.
+
+These tests require a running extenddb server (EXTENDDB_TEST_ENDPOINT must be
+set) because the service routes Streams to a separate endpoint; extenddb
+serves both on the same endpoint.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+import uuid
+from datetime import datetime, timezone
+
+import boto3
+import pytest
+from botocore.config import Config
+from botocore.exceptions import ClientError
+from conftest import wait_for_active, wait_for_deleted
+
+PREFIX = "One or more parameter values were invalid: "
+
+
+@pytest.fixture(scope="module")
+def endpoint_url() -> str:
+    return os.environ["EXTENDDB_TEST_ENDPOINT"].strip()
+
+
+def _client(service: str, endpoint_url: str, validate: bool = True):
+    kwargs: dict = dict(
+        service_name=service,
+        region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
+        endpoint_url=endpoint_url,
+        config=Config(parameter_validation=validate, retries={"max_attempts": 1}),
+    )
+    if endpoint_url.startswith("https://"):
+        kwargs["verify"] = False
+    return boto3.client(**kwargs)
+
+
+@pytest.fixture(scope="module")
+def dynamodb_client(endpoint_url: str):
+    return _client("dynamodb", endpoint_url)
+
+
+@pytest.fixture(scope="module")
+def streams_client(endpoint_url: str):
+    return _client("dynamodbstreams", endpoint_url)
+
+
+@pytest.fixture(scope="module")
+def raw_streams_client(endpoint_url: str):
+    """Parameter validation off, so malformed Timestamp shapes reach the server."""
+    return _client("dynamodbstreams", endpoint_url, validate=False)
+
+
+@pytest.fixture()
+def stream_table(dynamodb_client):
+    table_name = f"extenddb-ts-{uuid.uuid4().hex[:8]}"
+    dynamodb_client.create_table(
+        TableName=table_name,
+        AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}],
+        KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}],
+        BillingMode="PAY_PER_REQUEST",
+        StreamSpecification={"StreamEnabled": True, "StreamViewType": "NEW_AND_OLD_IMAGES"},
+    )
+    wait_for_active(dynamodb_client, table_name)
+    stream_arn = dynamodb_client.describe_table(TableName=table_name)["Table"]["LatestStreamArn"]
+    yield table_name, stream_arn
+    try:
+        dynamodb_client.delete_table(TableName=table_name)
+        wait_for_deleted(dynamodb_client, table_name)
+    except Exception:
+        pass
+
+
+def _shard_with_records(streams_client, stream_arn: str) -> str:
+    """The one shard holding a single partition key's records."""
+    for shard in streams_client.describe_stream(StreamArn=stream_arn)["StreamDescription"]["Shards"]:
+        it = streams_client.get_shard_iterator(
+            StreamArn=stream_arn, ShardId=shard["ShardId"], ShardIteratorType="TRIM_HORIZON"
+        )["ShardIterator"]
+        if streams_client.get_records(ShardIterator=it)["Records"]:
+            return shard["ShardId"]
+    raise AssertionError("no shard holds records")
+
+
+def _drain(streams_client, iterator: str, rounds: int = 6) -> list[dict]:
+    out: list[dict] = []
+    it = iterator
+    for _ in range(rounds):
+        resp = streams_client.get_records(ShardIterator=it)
+        out.extend(resp.get("Records", []))
+        it = resp.get("NextShardIterator")
+        if not resp.get("Records") or not it:
+            break
+    return out
+
+
+def _values(records: list[dict]) -> list[str]:
+    return [r["dynamodb"]["NewImage"]["v"]["S"] for r in records]
+
+
+def _at(streams_client, stream_arn: str, shard_id: str, ts) -> list[dict]:
+    it = streams_client.get_shard_iterator(
+        StreamArn=stream_arn, ShardId=shard_id, ShardIteratorType="AT_TIMESTAMP", Timestamp=ts
+    )["ShardIterator"]
+    return _drain(streams_client, it)
+
+
+def _write_group(dynamodb_client, table: str, tag: str, n: int) -> tuple[float, float]:
+    """Write n items just after a second boundary; return (start, end) wall times."""
+    now = time.time()
+    time.sleep((1.0 - (now % 1.0)) + 0.05)
+    start = time.time()
+    for i in range(n):
+        dynamodb_client.put_item(TableName=table, Item={"pk": {"S": "k"}, "v": {"S": f"{tag}-{i}"}})
+    return start, time.time()
+
+
+def _error(fn, **kw) -> tuple[str, str]:
+    try:
+        fn(**kw)
+    except ClientError as e:
+        err = e.response["Error"]
+        return err["Code"], err["Message"]
+    raise AssertionError("expected a ClientError")
+
+
+def _raw_error(endpoint_url: str, body: dict) -> tuple[str, str]:
+    """POST a GetShardIterator body exactly as given, SigV4-signed with the
+    test user's key from the environment, for shapes the SDK cannot
+    serialise (JSON null, an object). Returns (code, message)."""
+    import ssl
+    import urllib.request
+
+    from botocore.auth import SigV4Auth
+    from botocore.awsrequest import AWSRequest
+    from botocore.credentials import Credentials
+
+    data = json.dumps(body).encode()
+    req = AWSRequest(method="POST", url=endpoint_url, data=data, headers={
+        "Content-Type": "application/x-amz-json-1.0",
+        "X-Amz-Target": "DynamoDBStreams_20120810.GetShardIterator",
+    })
+    signer_creds = Credentials(os.environ["AWS_ACCESS_KEY_ID"], os.environ["AWS_SECRET_ACCESS_KEY"])
+    SigV4Auth(signer_creds, "dynamodb", os.environ.get("AWS_DEFAULT_REGION", "us-east-1")).add_auth(req)
+    prepped = req.prepare()
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    hr = urllib.request.Request(prepped.url, data=data, headers=dict(prepped.headers), method="POST")
+    try:
+        with urllib.request.urlopen(hr, timeout=30, context=ctx) as r:
+            raise AssertionError(f"expected an error, got {r.status}")
+    except urllib.error.HTTPError as e:
+        parsed = json.loads(e.read().decode())
+        return parsed.get("__type", "").split("#")[-1], parsed.get("message") or parsed.get("Message") or ""
+
+
+class TestPositioning:
+    """Measured on the service with three groups written in distinct seconds."""
+
+    def test_seek_positions_at_millisecond_precision(self, dynamodb_client, streams_client, stream_table):
+        table, arn = stream_table
+        g1 = _write_group(dynamodb_client, table, "g1", 3)
+        time.sleep(2.2)
+        g2 = _write_group(dynamodb_client, table, "g2", 3)
+        time.sleep(2.2)
+        g3 = _write_group(dynamodb_client, table, "g3", 2)
+        shard = _shard_with_records(streams_client, arn)
+        all_vals = ["g1-0", "g1-1", "g1-2", "g2-0", "g2-1", "g2-2", "g3-0", "g3-1"]
+
+        def ts(secs: float):
+            return datetime.fromtimestamp(secs, tz=timezone.utc)
+
+        # Before the first record and 25 hours ago both return everything, no error.
+        assert _values(_at(streams_client, arn, shard, ts(g1[0] - 10))) == all_vals
+        assert _values(_at(streams_client, arn, shard, ts(time.time() - 25 * 3600))) == all_vals
+        # Between groups.
+        assert _values(_at(streams_client, arn, shard, ts(g1[1] + 1.0))) == all_vals[3:]
+        assert _values(_at(streams_client, arn, shard, ts(g2[1] + 1.0))) == all_vals[6:]
+        # The group's own second, as an integer, includes the group.
+        assert _values(_at(streams_client, arn, shard, ts(int(g2[0])))) == all_vals[3:]
+        assert _values(_at(streams_client, arn, shard, ts(int(g3[0])))) == all_vals[6:]
+        # Fractional seconds inside the group's second skip it: the writes
+        # landed at +0.050 to about +0.100, so .5 and .999 start at group 3.
+        # Guard on the measured span so a slow write on a loaded host skips
+        # the probe instead of failing it.
+        if g2[1] < int(g2[0]) + 0.5:
+            assert _values(_at(streams_client, arn, shard, ts(int(g2[0]) + 0.5))) == all_vals[6:]
+            assert _values(_at(streams_client, arn, shard, ts(int(g2[0]) + 0.999))) == all_vals[6:]
+        else:
+            pytest.skip(f"group 2 spanned past +0.5 s ({g2[1] - int(g2[0]):.3f}); boundary probe not meaningful")
+        # One second after the last record: nothing.
+        assert _at(streams_client, arn, shard, ts(int(g3[1]) + 1)) == []
+        # ApproximateCreationDateTime is whole seconds.
+        recs = _at(streams_client, arn, shard, ts(g1[0] - 10))
+        for r in recs:
+            assert r["dynamodb"]["ApproximateCreationDateTime"].microsecond == 0
+
+    def test_boundary_is_at_or_after_the_record_millisecond(self, dynamodb_client, streams_client, stream_table):
+        """T equal to a record's own creation millisecond returns that record
+        (the service's at-or-after); one millisecond later does not."""
+        table, arn = stream_table
+        dynamodb_client.put_item(TableName=table, Item={"pk": {"S": "k"}, "v": {"S": "only"}})
+        shard = _shard_with_records(streams_client, arn)
+        it = streams_client.get_shard_iterator(StreamArn=arn, ShardId=shard, ShardIteratorType="TRIM_HORIZON")["ShardIterator"]
+        rec = _drain(streams_client, it)[0]
+        second = int(rec["dynamodb"]["ApproximateCreationDateTime"].timestamp())
+
+        def returns_it(ms: int) -> bool:
+            t = datetime.fromtimestamp(second + ms / 1000.0, tz=timezone.utc)
+            return _values(_at(streams_client, arn, shard, t)) == ["only"]
+
+        # Bisect within the record's second for the last millisecond that
+        # still returns it: that is the record's own creation millisecond.
+        assert returns_it(0)
+        lo, hi = 0, 1000
+        while hi - lo > 1:
+            mid = (lo + hi) // 2
+            if returns_it(mid):
+                lo = mid
+            else:
+                hi = mid
+        assert returns_it(lo), "T at the record's millisecond returns it"
+        assert not returns_it(lo + 1), "T one millisecond later does not"
+
+    def test_after_tip_returns_only_later_writes(self, dynamodb_client, streams_client, stream_table):
+        table, arn = stream_table
+        _write_group(dynamodb_client, table, "old", 2)
+        shard = _shard_with_records(streams_client, arn)
+        time.sleep(0.3)
+        it = streams_client.get_shard_iterator(
+            StreamArn=arn,
+            ShardId=shard,
+            ShardIteratorType="AT_TIMESTAMP",
+            Timestamp=datetime.fromtimestamp(time.time() - 0.2, tz=timezone.utc),
+        )["ShardIterator"]
+        dynamodb_client.put_item(TableName=table, Item={"pk": {"S": "k"}, "v": {"S": "post-tip"}})
+        assert _values(_drain(streams_client, it)) == ["post-tip"]
+
+
+class TestTimestampShapes:
+    def test_integer_and_fractional_numbers_are_accepted(self, dynamodb_client, streams_client, raw_streams_client, stream_table):
+        table, arn = stream_table
+        g = _write_group(dynamodb_client, table, "g", 2)
+        shard = _shard_with_records(streams_client, arn)
+        for ts in (int(g[0]), round(g[0], 3), g[0] + 0.123456):
+            resp = raw_streams_client.get_shard_iterator(
+                StreamArn=arn, ShardId=shard, ShardIteratorType="AT_TIMESTAMP", Timestamp=ts
+            )
+            assert "ShardIterator" in resp, ts
+
+    def test_zero_negative_and_null_refuse(self, dynamodb_client, streams_client, raw_streams_client, stream_table, endpoint_url):
+        table, arn = stream_table
+        _write_group(dynamodb_client, table, "g", 1)
+        shard = _shard_with_records(streams_client, arn)
+        expected = PREFIX + "Valid timestamp must be provided when the iterator type is AT_TIMESTAMP"
+        for ts in (0, -1):
+            code, msg = _error(
+                raw_streams_client.get_shard_iterator,
+                StreamArn=arn, ShardId=shard, ShardIteratorType="AT_TIMESTAMP", Timestamp=ts,
+            )
+            assert (code, msg) == ("ValidationException", expected), ts
+        # JSON null cannot go through the SDK; send it as the service saw it.
+        code, msg = _raw_error(endpoint_url, {
+            "StreamArn": arn, "ShardId": shard, "ShardIteratorType": "AT_TIMESTAMP", "Timestamp": None,
+        })
+        assert (code, msg) == ("ValidationException", expected), "null"
+
+    def test_milliseconds_as_seconds_is_in_the_future(self, dynamodb_client, streams_client, stream_table, endpoint_url):
+        table, arn = stream_table
+        g = _write_group(dynamodb_client, table, "g", 1)
+        shard = _shard_with_records(streams_client, arn)
+        for ts in (int(g[0]) * 1000, 10**18):
+            code, msg = _raw_error(endpoint_url, {
+                "StreamArn": arn, "ShardId": shard, "ShardIteratorType": "AT_TIMESTAMP", "Timestamp": ts,
+            })
+            assert code == "ValidationException", ts
+            assert msg == PREFIX + "Timestamp parameter cannot be greater than the current timestamp", ts
+
+    def test_non_numeric_timestamp_is_a_serialization_error(self, dynamodb_client, streams_client, stream_table, endpoint_url):
+        table, arn = stream_table
+        g = _write_group(dynamodb_client, table, "g", 1)
+        shard = _shard_with_records(streams_client, arn)
+        cases = [
+            (f"{int(g[0])}.5", "STRING_VALUE cannot be converted to Date"),
+            ("", "STRING_VALUE cannot be converted to Date"),
+            (True, "TRUE_VALUE cannot be converted to Date"),
+            ({"S": "x"}, "Start of structure or map found where not expected"),
+        ]
+        for ts, expected in cases:
+            code, msg = _raw_error(endpoint_url, {
+                "StreamArn": arn, "ShardId": shard, "ShardIteratorType": "AT_TIMESTAMP", "Timestamp": ts,
+            })
+            assert (code, msg) == ("SerializationException", expected), ts
+
+
+class TestValidationMatrix:
+    """Messages and check order measured on the service."""
+
+    def test_matrix(self, dynamodb_client, streams_client, raw_streams_client, stream_table):
+        table, arn = stream_table
+        g = _write_group(dynamodb_client, table, "g", 1)
+        shard = _shard_with_records(streams_client, arn)
+        it = streams_client.get_shard_iterator(StreamArn=arn, ShardId=shard, ShardIteratorType="TRIM_HORIZON")["ShardIterator"]
+        seq = _drain(streams_client, it)[0]["dynamodb"]["SequenceNumber"]
+        ts = int(g[0])
+        future = int(time.time()) + 3600
+        ts_required = "Valid timestamp must be provided when the iterator type is AT_TIMESTAMP"
+        seq_refused = "SequenceNumber should not be passed for this iterator type"
+        ts_refused = "Timestamp should not be passed for this iterator type"
+        future_refused = "Timestamp parameter cannot be greater than the current timestamp"
+        cases = [
+            (dict(ShardIteratorType="AT_TIMESTAMP"), ts_required),
+            (dict(ShardIteratorType="AT_TIMESTAMP", SequenceNumber=seq), ts_required),
+            (dict(ShardIteratorType="AT_TIMESTAMP", SequenceNumber=seq, Timestamp=ts), seq_refused),
+            (dict(ShardIteratorType="AT_TIMESTAMP", SequenceNumber=seq, Timestamp=future), seq_refused),
+            (dict(ShardIteratorType="AT_TIMESTAMP", SequenceNumber=seq, Timestamp=0), ts_required),
+            (dict(ShardIteratorType="AT_TIMESTAMP", Timestamp=future), future_refused),
+            (dict(ShardIteratorType="TRIM_HORIZON", Timestamp=ts), ts_refused),
+            (dict(ShardIteratorType="LATEST", Timestamp=ts), ts_refused),
+            (dict(ShardIteratorType="AT_SEQUENCE_NUMBER", SequenceNumber=seq, Timestamp=ts), ts_refused),
+            (dict(ShardIteratorType="AFTER_SEQUENCE_NUMBER", SequenceNumber=seq, Timestamp=ts), ts_refused),
+            (dict(ShardIteratorType="TRIM_HORIZON", SequenceNumber=seq, Timestamp=ts), ts_refused),
+            (dict(ShardIteratorType="TRIM_HORIZON", SequenceNumber=seq), seq_refused),
+            (dict(ShardIteratorType="LATEST", SequenceNumber=seq), seq_refused),
+            (dict(ShardIteratorType="AT_SEQUENCE_NUMBER"), "SequenceNumber cannot be null"),
+        ]
+        for extra, expected in cases:
+            code, msg = _error(raw_streams_client.get_shard_iterator, StreamArn=arn, ShardId=shard, **extra)
+            assert code == "ValidationException", extra
+            assert msg == PREFIX + expected, extra
+
+    def test_unknown_or_missing_type_renders_without_the_value(self, dynamodb_client, streams_client, raw_streams_client, stream_table, endpoint_url):
+        table, arn = stream_table
+        _write_group(dynamodb_client, table, "g", 1)
+        shard = _shard_with_records(streams_client, arn)
+        members = "[AFTER_SEQUENCE_NUMBER, LATEST, AT_TIMESTAMP, AT_SEQUENCE_NUMBER, TRIM_HORIZON]"
+        # The service omits the offending value from the enum violation.
+        for bad in ("BOGUS", "at_timestamp"):
+            code, msg = _error(raw_streams_client.get_shard_iterator, StreamArn=arn, ShardId=shard, ShardIteratorType=bad)
+            assert code == "ValidationException"
+            assert msg == (
+                "1 validation error detected: Value at 'shardIteratorType' failed to satisfy "
+                f"constraint: Member must satisfy enum value set: {members}"
+            )
+        # A missing or null type is a ValidationException, not a serialization error.
+        for body in ({"Timestamp": 1792764248.0}, {"ShardIteratorType": None}):
+            code, msg = _raw_error(endpoint_url, {"StreamArn": arn, "ShardId": shard, **body})
+            assert (code, msg) == (
+                "ValidationException",
+                "1 validation error detected: Value at 'shardIteratorType' failed to satisfy "
+                "constraint: Member must not be null",
+            ), body
+
+    def test_future_check_runs_before_shard_existence(self, dynamodb_client, streams_client, raw_streams_client, stream_table):
+        table, arn = stream_table
+        _write_group(dynamodb_client, table, "g", 1)
+        shard = _shard_with_records(streams_client, arn)
+        code, msg = _error(
+            raw_streams_client.get_shard_iterator,
+            StreamArn=arn, ShardId="shardId-00000000000000000000-00000000",
+            ShardIteratorType="AT_TIMESTAMP", Timestamp=int(time.time()) + 3600,
+        )
+        assert (code, msg) == ("ValidationException", PREFIX + "Timestamp parameter cannot be greater than the current timestamp")
+        code, _ = _error(
+            raw_streams_client.get_shard_iterator,
+            StreamArn=arn, ShardId="shardId-00000000000000000000-00000000",
+            ShardIteratorType="AT_TIMESTAMP", Timestamp=int(time.time()) - 60,
+        )
+        assert code == "ResourceNotFoundException"
+        assert shard  # the real shard still exists
+
+    def test_near_future_is_accepted(self, dynamodb_client, streams_client, stream_table):
+        table, arn = stream_table
+        _write_group(dynamodb_client, table, "g", 1)
+        shard = _shard_with_records(streams_client, arn)
+        resp = streams_client.get_shard_iterator(
+            StreamArn=arn, ShardId=shard, ShardIteratorType="AT_TIMESTAMP",
+            Timestamp=datetime.fromtimestamp(time.time() + 0.5, tz=timezone.utc),
+        )
+        assert "ShardIterator" in resp
+        code, _ = _error(
+            streams_client.get_shard_iterator,
+            StreamArn=arn, ShardId=shard, ShardIteratorType="AT_TIMESTAMP",
+            Timestamp=datetime.fromtimestamp(time.time() + 5, tz=timezone.utc),
+        )
+        assert code == "ValidationException"