diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md
index faaaa38fc2..c2fcd53af6 100644
--- a/python/CHANGELOG.md
+++ b/python/CHANGELOG.md
@@ -37,6 +37,9 @@ CsvDataColumn(
 
 Setting `enum_types` on a non-enum data type raises a validation error. Multi-channel Parquet single-channel-per-row imports don't support enums, since channel configs there are derived per row.
 
+### Bugfixes
+- Fix `channels.get_data` printing a `UserWarning: Discarding nonzero nanoseconds in conversion` once per channel on each run-scoped fetch. The channel data cache stored segment bounds as `datetime`, which truncated the nanosecond timestamps the data already carries; they are now `pd.Timestamp`. `start_time` and `end_time` on `get_data` and `get_data_as_arrow` also accept a `pd.Timestamp` now, and its nanoseconds survive to the wire, where the service filters on them.
+
 ## [v0.20.0] - August 25, 2026
 
 ### What's New
diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data.py b/python/lib/sift_client/_internal/low_level_wrappers/data.py
index 6661a710cb..9ac5bc071a 100644
--- a/python/lib/sift_client/_internal/low_level_wrappers/data.py
+++ b/python/lib/sift_client/_internal/low_level_wrappers/data.py
@@ -38,7 +38,7 @@
 REQUEST_BATCH_SIZE = 1
 
 
-TimeRange = Tuple[datetime, datetime]
+TimeRange = Tuple[pd.Timestamp, pd.Timestamp]
 
 
 class SegmentRef(BaseModel):
@@ -48,12 +48,17 @@ class SegmentRef(BaseModel):
     queried this range and the wire returned no data". Empty refs
     contribute to coverage (so a repeat of a known-empty range
     doesn't hit the wire).
+
+    Bounds are ``pd.Timestamp``: a ``datetime`` truncates to
+    microseconds, which would under-claim the first sample by up to
+    999 ns. The adapter's methods normalize what they're handed, so
+    callers can still pass a plain ``datetime``.
     """
 
-    model_config = ConfigDict(frozen=True)
+    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
     seg_id: int | None
-    start_time: datetime
-    end_time: datetime
+    start_time: pd.Timestamp
+    end_time: pd.Timestamp
 
 
 class SegmentIndex(BaseModel):
@@ -159,8 +164,8 @@ def get_range(
         self,
         channel_id: str,
         run_id: str | None,
-        start_time: datetime,
-        end_time: datetime,
+        start_time: pd.Timestamp | datetime,
+        end_time: pd.Timestamp | datetime,
     ) -> tuple[pd.DataFrame | None, list[TimeRange]]:
         """Return cached data covering ``[start_time, end_time]`` plus gaps.
 
@@ -176,6 +181,8 @@ def get_range(
             query window not covered by any present segment.
             ``gaps == []`` means the cache fully covers the request.
         """
+        start_time = to_timestamp_nanos(start_time)
+        end_time = to_timestamp_nanos(end_time)
         idx = self._load_index(channel_id, run_id)
         if idx is None or not idx.segments:
             return None, [(start_time, end_time)]
@@ -226,8 +233,8 @@ def put_segment(
         channel_id: str,
         run_id: str | None,
         data: pd.DataFrame,
-        start_time: datetime,
-        end_time: datetime,
+        start_time: pd.Timestamp | datetime,
+        end_time: pd.Timestamp | datetime,
     ) -> None:
         """Write a new segment and update the index.
 
@@ -254,6 +261,8 @@ def put_segment(
         puts skip the segment write entirely and update the index
         directly.
         """
+        start_time = to_timestamp_nanos(start_time)
+        end_time = to_timestamp_nanos(end_time)
         idx = self._load_index(channel_id, run_id) or SegmentIndex()
 
         if len(data) == 0:
@@ -486,8 +495,8 @@ def _replace_index_and_drop_segments(
 
     @staticmethod
     def _compute_gaps(
-        query_start: datetime,
-        query_end: datetime,
+        query_start: pd.Timestamp,
+        query_end: pd.Timestamp,
         covered: list[TimeRange],
     ) -> list[TimeRange]:
         """Sub-ranges of ``[query_start, query_end]`` not in ``covered``.
@@ -566,8 +575,8 @@ async def _get_data_impl(
         *,
         channel_ids: list[str],
         run_id: str | None = None,
-        start_time: datetime | None = None,
-        end_time: datetime,
+        start_time: pd.Timestamp | datetime | None = None,
+        end_time: pd.Timestamp | datetime,
         page_size: int | None = None,
         page_token: str | None = None,
         order_by: str | None = None,
@@ -642,8 +651,8 @@ def _update_cache(
         *,
         channel_data: dict[str, pd.DataFrame],
         fetched_ranges_per_channel: dict[str, list[TimeRange]],
-        start_time: datetime,
-        end_time: datetime,
+        start_time: pd.Timestamp,
+        end_time: pd.Timestamp,
         run_id: str | None = None,
     ):
         """Write each channel's fresh data or empty-ref as a new segment.
@@ -720,12 +729,13 @@ def _update_cache(
             # started yet). For unscoped queries, claim the full
             # requested range so a follow-up of the same range hits.
             seg_end = end_time
+            seg_start: pd.Timestamp
             if run_id:
                 # ``combined.index`` is a ``DatetimeIndex`` (built from
                 # the wire's nanosecond timestamps), so ``index[0]`` is
                 # always ``pd.Timestamp`` at runtime; pandas-stubs types
                 # it as the wider ``Scalar`` union.
-                seg_start = cast("pd.Timestamp", combined.index[0]).to_pydatetime()
+                seg_start = cast("pd.Timestamp", combined.index[0])
             else:
                 seg_start = start_time
 
@@ -767,8 +777,8 @@ async def get_channel_data(
         *,
         channels: list[Channel],
         run_id: str | None = None,
-        start_time: datetime | None = None,
-        end_time: datetime | None = None,
+        start_time: pd.Timestamp | datetime | None = None,
+        end_time: pd.Timestamp | datetime | None = None,
         max_results: int | None = None,
         page_size: int | None = None,
         ignore_cache: bool = False,
@@ -777,8 +787,8 @@ async def get_channel_data(
         """Get the data for a channel during a run."""
         ret_data: dict[str, pd.DataFrame] = {}
         # No data will be returned if end_time is not provided.
-        start_time = start_time or datetime.fromtimestamp(0, tz=timezone.utc)
-        end_time = end_time or datetime.now(timezone.utc)
+        query_start = to_timestamp_nanos(start_time or datetime.fromtimestamp(0, tz=timezone.utc))
+        query_end = to_timestamp_nanos(end_time or datetime.now(timezone.utc))
 
         self._update_name_id_map(channels)
 
@@ -803,9 +813,11 @@ async def get_channel_data(
             assert cid is not None
             if ignore_cache:
                 cached_data: pd.DataFrame | None = None
-                gaps: list[TimeRange] = [(start_time, end_time)]
+                gaps: list[TimeRange] = [(query_start, query_end)]
             else:
-                cached_data, gaps = self.channel_cache.get_range(cid, run_id, start_time, end_time)
+                cached_data, gaps = self.channel_cache.get_range(
+                    cid, run_id, query_start, query_end
+                )
 
             if cached_data is not None:
                 cached_points += int(cached_data.size)
@@ -820,7 +832,7 @@ async def get_channel_data(
             if not gaps:
                 continue
             fetched_ranges_per_channel.setdefault(cid, []).extend(gaps)
-            if len(gaps) == 1 and gaps[0] == (start_time, end_time):
+            if len(gaps) == 1 and gaps[0] == (query_start, query_end):
                 fully_uncached.append(cid)
             else:
                 partial_gaps.append((cid, gaps))
@@ -841,8 +853,8 @@ async def get_channel_data(
                     {
                         "channel_ids": batch,
                         "run_id": run_id,
-                        "start_time": start_time,
-                        "end_time": end_time,
+                        "start_time": query_start,
+                        "end_time": query_end,
                     },
                     ", ".join(id_to_name.get(cid, cid) for cid in batch),
                 )
@@ -963,8 +975,8 @@ async def _tick(kwargs: dict[str, Any], label: str) -> Any:
             self._update_cache(
                 channel_data=ret_data,
                 fetched_ranges_per_channel=fetched_ranges_per_channel,
-                start_time=start_time,
-                end_time=end_time,
+                start_time=query_start,
+                end_time=query_end,
                 run_id=run_id,
             )
 
diff --git a/python/lib/sift_client/_internal/low_level_wrappers/exports.py b/python/lib/sift_client/_internal/low_level_wrappers/exports.py
index fd438fac0b..13e35d17d5 100644
--- a/python/lib/sift_client/_internal/low_level_wrappers/exports.py
+++ b/python/lib/sift_client/_internal/low_level_wrappers/exports.py
@@ -19,7 +19,7 @@
 from sift.exports.v1.exports_pb2_grpc import ExportServiceStub
 
 from sift_client._internal.low_level_wrappers.base import LowLevelClientBase
-from sift_client._internal.util.timestamp import to_pb_timestamp
+from sift_client._internal.time import to_timestamp_pb
 from sift_client.sift_types.calculated_channel import CalculatedChannel, CalculatedChannelCreate
 from sift_client.transport import WithGrpcClient
 
@@ -123,25 +123,25 @@ async def export_data(
         if run_ids is not None:
             runs_and_time_range = RunsAndTimeRange(run_ids=run_ids)
             if start_time:
-                runs_and_time_range.start_time.CopyFrom(to_pb_timestamp(start_time))
+                runs_and_time_range.start_time.CopyFrom(to_timestamp_pb(start_time))
             if stop_time:
-                runs_and_time_range.stop_time.CopyFrom(to_pb_timestamp(stop_time))
+                runs_and_time_range.stop_time.CopyFrom(to_timestamp_pb(stop_time))
             request.runs_and_time_range.CopyFrom(runs_and_time_range)
 
         if asset_ids is not None:
             assets_and_time_range = AssetsAndTimeRange(asset_ids=asset_ids)
             if start_time:
-                assets_and_time_range.start_time.CopyFrom(to_pb_timestamp(start_time))
+                assets_and_time_range.start_time.CopyFrom(to_timestamp_pb(start_time))
             if stop_time:
-                assets_and_time_range.stop_time.CopyFrom(to_pb_timestamp(stop_time))
+                assets_and_time_range.stop_time.CopyFrom(to_timestamp_pb(stop_time))
             request.assets_and_time_range.CopyFrom(assets_and_time_range)
 
         if run_ids is None and asset_ids is None:
             time_range = TimeRange()
             if start_time:
-                time_range.start_time.CopyFrom(to_pb_timestamp(start_time))
+                time_range.start_time.CopyFrom(to_timestamp_pb(start_time))
             if stop_time:
-                time_range.stop_time.CopyFrom(to_pb_timestamp(stop_time))
+                time_range.stop_time.CopyFrom(to_timestamp_pb(stop_time))
             request.time_range.CopyFrom(time_range)
 
         response = await self._grpc_client.get_stub(ExportServiceStub).ExportData(request)
diff --git a/python/lib/sift_client/_internal/low_level_wrappers/rules.py b/python/lib/sift_client/_internal/low_level_wrappers/rules.py
index 364ee37e14..5a49acf113 100644
--- a/python/lib/sift_client/_internal/low_level_wrappers/rules.py
+++ b/python/lib/sift_client/_internal/low_level_wrappers/rules.py
@@ -54,7 +54,7 @@
 
 from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase
 from sift_client._internal.low_level_wrappers.jobs import JobsLowLevelClient
-from sift_client._internal.util.timestamp import to_pb_timestamp
+from sift_client._internal.time import to_timestamp_pb
 from sift_client._internal.util.util import count_non_none
 from sift_client.sift_types.rule import (
     Rule,
@@ -635,13 +635,13 @@ async def evaluate_rules(
             if run_id:
                 kwargs["run_time_range"] = RunTimeRange(
                     run=run_id,  # type: ignore
-                    start_time=to_pb_timestamp(start_time),
-                    end_time=to_pb_timestamp(end_time),  # type: ignore
+                    start_time=to_timestamp_pb(start_time),
+                    end_time=to_timestamp_pb(end_time),  # type: ignore
                 )
                 kwargs["assets"] = AssetsTimeRange(
                     assets={"ids": {"ids": asset_ids}},  # type: ignore
-                    start_time=to_pb_timestamp(start_time),
-                    end_time=to_pb_timestamp(end_time),
+                    start_time=to_timestamp_pb(start_time),
+                    end_time=to_timestamp_pb(end_time),
                 )
         elif run_id:
             kwargs["run"] = ResourceIdentifier(id=run_id)
diff --git a/python/lib/sift_client/_internal/low_level_wrappers/runs.py b/python/lib/sift_client/_internal/low_level_wrappers/runs.py
index 12d5b8a75d..7b329a5e32 100644
--- a/python/lib/sift_client/_internal/low_level_wrappers/runs.py
+++ b/python/lib/sift_client/_internal/low_level_wrappers/runs.py
@@ -19,7 +19,7 @@
 from sift.runs.v2.runs_pb2_grpc import RunServiceStub
 
 from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase
-from sift_client._internal.util.timestamp import to_pb_timestamp
+from sift_client._internal.time import to_timestamp_pb
 from sift_client.sift_types.run import Run, RunCreate, RunUpdate
 from sift_client.transport import WithGrpcClient
 
@@ -212,8 +212,8 @@ async def create_adhoc_run(
         request = CreateAdhocRunRequest(
             name=name,
             description=description or "",
-            start_time=to_pb_timestamp(start_time) if start_time else None,
-            stop_time=to_pb_timestamp(stop_time) if stop_time else None,
+            start_time=to_timestamp_pb(start_time) if start_time else None,
+            stop_time=to_timestamp_pb(stop_time) if stop_time else None,
             asset_ids=asset_ids,
             tags=tag_names,
             metadata=metadata_dict_to_proto(metadata) if metadata else None,
diff --git a/python/lib/sift_client/_internal/time.py b/python/lib/sift_client/_internal/time.py
index 9787996a6d..bcde622f35 100644
--- a/python/lib/sift_client/_internal/time.py
+++ b/python/lib/sift_client/_internal/time.py
@@ -31,15 +31,26 @@ def to_timestamp_nanos(arg: TimestampPb | pd.Timestamp | datetime | str | int) -
         return cast("pd.Timestamp", pd.Timestamp(arg))
 
 
-def to_timestamp_pb(arg: datetime | str | int | float) -> TimestampPb:
+def to_timestamp_pb(arg: pd.Timestamp | datetime | str | int | float) -> TimestampPb:
     """
-    Mainly used for testing at the moment. If using this for non-testing purposes
-    should probably make this more robust and support nano-second precision.
+    Converts a variety of time-types to a protobuf timestamp.
+
+    A ``pd.Timestamp`` keeps its nanoseconds. Every other input has
+    microsecond resolution at best, since ``TimestampPb.FromDatetime``
+    reads ``microsecond`` and nothing finer. A naive input is read as
+    UTC on either path.
     """
 
     ts = TimestampPb()
 
-    if isinstance(arg, datetime):
+    if isinstance(arg, pd.Timestamp) and arg.nanosecond:
+        # A nonzero sub-microsecond remainder is exactly what
+        # ``FromDatetime`` drops, and it implies nanosecond units, so
+        # ``value`` can't overflow here. Coarser units reach dates
+        # outside the ns range where it would.
+        ts.FromNanoseconds(arg.value)
+        return ts
+    elif isinstance(arg, datetime):
         ts.FromDatetime(arg)
         return ts
     elif isinstance(arg, (int, float)):
diff --git a/python/lib/sift_client/_internal/util/timestamp.py b/python/lib/sift_client/_internal/util/timestamp.py
deleted file mode 100644
index 5cc659fce7..0000000000
--- a/python/lib/sift_client/_internal/util/timestamp.py
+++ /dev/null
@@ -1,17 +0,0 @@
-from datetime import datetime
-
-from google.protobuf.timestamp_pb2 import Timestamp
-
-
-def to_pb_timestamp(timestamp: datetime) -> Timestamp:
-    """Convert a Python datetime to a Protocol Buffer Timestamp.
-
-    Args:
-        timestamp: The datetime to convert
-
-    Returns:
-        A Protocol Buffer Timestamp representation
-    """
-    timestamp_pb = Timestamp()
-    timestamp_pb.FromDatetime(timestamp)
-    return timestamp_pb
diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py
index 31fb8cd7b6..308824aaf0 100644
--- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py
+++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py
@@ -35,6 +35,7 @@
 
 from __future__ import annotations
 
+import warnings
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Any, Iterator, Mapping, Sequence, cast
@@ -47,6 +48,7 @@
 from sift_client._internal.low_level_wrappers.data import (
     ChannelDataCache,
     DataLowLevelClient,
+    SegmentRef,
 )
 from sift_client.sift_types.channel import (
     Channel,
@@ -56,6 +58,14 @@
 
 _NOW = datetime(2025, 1, 1, tzinfo=timezone.utc)
 _WINDOW_END = _NOW + timedelta(days=1)
+# ``_update_cache`` is called only by ``get_channel_data``, which
+# normalizes its window first, so its bounds arrive as ``pd.Timestamp``.
+_NOW_TS = pd.Timestamp(_NOW)
+_WINDOW_END_TS = pd.Timestamp(_WINDOW_END)
+# A start that falls off a whole microsecond, the way real sample
+# timestamps do. Anything derived from it keeps its 789 ns tail only if
+# every hop is nanosecond-capable.
+_NS_START = _NOW_TS + pd.Timedelta(123456789, unit="ns")
 
 
 # ---------- shared helpers -----------
@@ -78,6 +88,14 @@ def _frame(
     ).astype({cid: value_dtype})
 
 
+def _sole_ref(adapter: ChannelDataCache, channel_id: str, run_id: str | None = None) -> SegmentRef:
+    """The one ``SegmentRef`` in a bucket. Fails loudly if there isn't exactly one."""
+    idx = adapter._load_index(channel_id, run_id)
+    assert idx is not None
+    assert len(idx.segments) == 1
+    return idx.segments[0]
+
+
 def _channel(cid: str) -> Channel:
     """Minimal ``Channel`` with required fields populated."""
     return Channel(
@@ -217,11 +235,11 @@ def _put(
     *,
     data: pd.DataFrame | None = None,
     rows: int = 5,
-    start: datetime = _NOW,
+    start: pd.Timestamp | datetime = _NOW,
     offset: int = 0,
     freq: str = "ms",
-    seg_start: datetime | None = None,
-    seg_end: datetime | None = None,
+    seg_start: pd.Timestamp | datetime | None = None,
+    seg_end: pd.Timestamp | datetime | None = None,
     run_id: str | None = None,
 ) -> pd.DataFrame:
     """Convenience: build a frame, write it as one segment, return it.
@@ -233,9 +251,9 @@ def _put(
     if data is None:
         data = _frame(channel_id, rows=rows, start=start, offset=offset, freq=freq)
     if seg_start is None:
-        seg_start = cast("pd.Timestamp", data.index[0]).to_pydatetime()
+        seg_start = cast("pd.Timestamp", data.index[0])
     if seg_end is None:
-        seg_end = cast("pd.Timestamp", data.index[-1]).to_pydatetime()
+        seg_end = cast("pd.Timestamp", data.index[-1])
     adapter.put_segment(
         channel_id=channel_id,
         run_id=run_id,
@@ -415,9 +433,7 @@ def test_size_bytes_propagates_to_store(self, tmp_path):
         store = DiskCache(disk_path=tmp_path / "size", disk_max_bytes=size_bytes // 2)
         adapter = ChannelDataCache(store)
         try:
-            adapter.put_segment(
-                "c1", None, big, big.index[0].to_pydatetime(), big.index[-1].to_pydatetime()
-            )
+            adapter.put_segment("c1", None, big, big.index[0], big.index[-1])
             assert not adapter.has_any("c1")
         finally:
             store.close()
@@ -446,7 +462,7 @@ def test_get_range_stitches_multiple_segments(self, tmp_path):
         adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "stitch"))
         try:
             df1 = _frame("c1", rows=5, start=_NOW, freq="ms", offset=0)
-            df1_end = df1.index[-1].to_pydatetime()
+            df1_end = df1.index[-1]
             df2 = _frame(
                 "c1",
                 rows=5,
@@ -454,8 +470,8 @@ def test_get_range_stitches_multiple_segments(self, tmp_path):
                 freq="ms",
                 offset=100,
             )
-            query_start = df1.index[0].to_pydatetime()
-            query_end = df2.index[-1].to_pydatetime()
+            query_start = df1.index[0]
+            query_end = df2.index[-1]
             # Claim seg2 starts right where seg1 ends so the abutting
             # ranges leave no gap. Real callers do this via
             # ``_update_cache`` claiming the requested window.
@@ -534,8 +550,8 @@ def test_put_segment_normalizes_non_monotonic_index_before_write(self, tmp_path)
             shuffled = sorted_frame.iloc[[2, 0, 4, 1, 3]]
             assert not shuffled.index.is_monotonic_increasing
 
-            seg_start = sorted_frame.index[0].to_pydatetime()
-            seg_end = sorted_frame.index[-1].to_pydatetime()
+            seg_start = sorted_frame.index[0]
+            seg_end = sorted_frame.index[-1]
             adapter.put_segment("c1", None, shuffled, seg_start, seg_end)
 
             data, gaps = adapter.get_range("c1", None, seg_start, seg_end)
@@ -654,13 +670,13 @@ def test_empty_ref_alongside_data_segment(self, tmp_path):
         adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "mix"))
         try:
             data = _frame("c1", rows=5, freq="s")
-            mid = data.index[-1].to_pydatetime()
+            mid = data.index[-1]
             tail = mid + timedelta(seconds=5)
 
-            adapter.put_segment("c1", None, data, data.index[0].to_pydatetime(), mid)
+            adapter.put_segment("c1", None, data, data.index[0], mid)
             adapter.put_segment("c1", None, pd.DataFrame(), mid, tail)
 
-            got, gaps = adapter.get_range("c1", None, data.index[0].to_pydatetime(), tail)
+            got, gaps = adapter.get_range("c1", None, data.index[0], tail)
             assert got is not None
             pd.testing.assert_frame_equal(got, data)
             assert gaps == []
@@ -718,6 +734,154 @@ def test_compute_gaps(self, covered, expected):
         assert ChannelDataCache._compute_gaps(query_start, query_end, covered) == expected
 
 
+class TestNanosecondPrecision:
+    """Cache bounds hold the nanoseconds the wire delivers.
+
+    A ``datetime`` stops at microseconds, so a bound recorded as one
+    sits up to 999 ns below the sample it describes. That shows up two
+    ways. Pandas warns ("Discarding nonzero nanoseconds in conversion")
+    once per channel per fetch. The index also claims coverage starting
+    before the first row it holds, and ``get_range`` believes it,
+    serving a hit for a sub-microsecond sliver that was never fetched.
+
+    Rigs whose samples fall on exact microseconds see neither symptom,
+    which is why the bug reads as dataset-specific.
+    """
+
+    def test_put_segment_round_trips_nanoseconds(self, tmp_path):
+        """The index stores the exact ns bound it was handed."""
+        adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "ns"))
+        try:
+            df = _put(adapter, "c1", rows=4, start=_NS_START)
+            ref = _sole_ref(adapter, "c1")
+            assert ref.start_time == df.index[0]
+            assert ref.start_time.nanosecond == 789
+        finally:
+            adapter.store.close()
+
+    def test_index_survives_a_reopen_with_nanoseconds_intact(self, tmp_path):
+        """Bounds keep their nanoseconds across a store close/reopen.
+
+        The index is pickled, so this pins that the precision isn't
+        lost on the serialization hop rather than in memory.
+        """
+        path = tmp_path / "ns-reopen"
+        adapter = ChannelDataCache(DiskCache(disk_path=path))
+        try:
+            df = _put(adapter, "c1", rows=4, start=_NS_START)
+        finally:
+            adapter.store.close()
+
+        reopened = ChannelDataCache(DiskCache(disk_path=path))
+        try:
+            assert _sole_ref(reopened, "c1").start_time == df.index[0]
+        finally:
+            reopened.store.close()
+
+    def test_gaps_are_exact_to_the_nanosecond(self, tmp_path):
+        """Gaps land on the segment's real bounds, from either input type.
+
+        A query starting a nanosecond early has to report that sliver
+        as a gap. With a rounded-down bound it reads as covered and
+        never reaches the wire. Gaps go straight back out, so they stay
+        ``pd.Timestamp`` even when the caller asked in ``datetime``.
+        """
+        adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "claim"))
+        try:
+            df = _put(adapter, "c1", rows=4, start=_NS_START)
+
+            early = df.index[0] - pd.Timedelta(1, unit="ns")
+            _, gaps = adapter.get_range("c1", None, early, df.index[-1])
+            assert gaps == [(early, df.index[0])]
+
+            _, gaps = adapter.get_range("c1", None, _NOW, _WINDOW_END)
+            assert gaps == [(_NOW, df.index[0]), (df.index[-1], _WINDOW_END)]
+            assert all(isinstance(bound, pd.Timestamp) for gap in gaps for bound in gap)
+        finally:
+            adapter.store.close()
+
+    def test_run_scoped_update_records_first_sample_exactly(self, tmp_path):
+        """``_update_cache`` claims the first row's nanoseconds and warns about nothing.
+
+        The run-scoped branch takes the segment's lower bound from the
+        data itself. That is the one place a wire timestamp used to be
+        narrowed to a ``datetime``.
+        """
+        client = _client_with_cache(tmp_path)
+        client._update_name_id_map([_channel("c1")])
+        df = _frame("c1", rows=4, start=_NS_START, freq="ms")
+        try:
+            with warnings.catch_warnings(record=True) as caught:
+                warnings.simplefilter("always")
+                client._update_cache(
+                    channel_data={"c1": df},
+                    fetched_ranges_per_channel={"c1": [(_NOW_TS, _WINDOW_END_TS)]},
+                    start_time=_NOW_TS,
+                    end_time=_WINDOW_END_TS,
+                    run_id="r1",
+                )
+            assert [str(w.message) for w in caught] == []
+
+            ref = _sole_ref(client.channel_cache, "c1", run_id="r1")
+            assert ref.start_time == df.index[0]
+        finally:
+            client.channel_cache.store.close()
+
+    def test_index_written_before_the_fix_still_reads(self, tmp_path):
+        """A ref carrying a plain ``datetime`` bound keeps working.
+
+        The store lives under the temp dir and is shared across
+        sessions, so an index written by an older client is a normal
+        thing to open. Unpickling skips validation, so such a ref
+        arrives with a ``datetime`` and has to compare cleanly against
+        nanosecond query bounds.
+        """
+        adapter = ChannelDataCache(DiskCache(disk_path=tmp_path / "stale"))
+        try:
+            df = _put(adapter, "c1", rows=4, start=_NS_START)
+            idx = adapter._load_index("c1", None)
+            assert idx is not None
+            ref = idx.segments[0]
+            # What the old code stored: the same bound, floored to a datetime.
+            idx.segments[0] = SegmentRef.model_construct(
+                seg_id=ref.seg_id,
+                start_time=ref.start_time.to_pydatetime(warn=False),
+                end_time=ref.end_time.to_pydatetime(warn=False),
+            )
+            adapter._write_index("c1", None, idx)
+
+            got, gaps = adapter.get_range("c1", None, df.index[0], df.index[-1])
+            assert got is not None
+            assert len(got) == 4
+            # The floored upper bound leaves the last 789 ns to refetch,
+            # erring toward the wire rather than toward a silent hole.
+            stale_end = pd.Timestamp(df.index[-1].to_pydatetime(warn=False))
+            assert gaps == [(stale_end, df.index[-1])]
+        finally:
+            adapter.store.close()
+
+    @pytest.mark.asyncio
+    async def test_query_bounds_reach_the_wire_with_nanoseconds(self) -> None:
+        """A ``pd.Timestamp`` bound arrives at ``_get_data_impl`` unrounded.
+
+        The service filters on a millisecond plus sub-millisecond pair,
+        so a caller asking for a nanosecond-precise window gets one
+        rather than the surrounding microsecond.
+        """
+        client = DataLowLevelClient(MagicMock())
+        end = _NS_START + pd.Timedelta(seconds=10)
+        with _fake_grpc(client, {"c1": [_frame("c1", start=_NS_START)]}) as call_log:
+            await client.get_channel_data(
+                channels=[_channel("c1")],
+                start_time=_NS_START,
+                end_time=end,
+                ignore_cache=True,
+            )
+        assert call_log[0]["start_time"] == _NS_START
+        assert call_log[0]["start_time"].nanosecond == 789
+        assert call_log[0]["end_time"] == end
+
+
 class TestCompaction:
     """``ChannelDataCache.MAX_SEGMENTS_PER_BUCKET`` triggers a merge.
 
@@ -822,8 +986,8 @@ def test_compaction_preserves_data(self, tmp_path, monkeypatch):
             got, gaps = adapter.get_range(
                 "c1",
                 None,
-                frames[0].index[0].to_pydatetime(),
-                frames[-1].index[-1].to_pydatetime(),
+                frames[0].index[0],
+                frames[-1].index[-1],
             )
             assert gaps == []
             assert got is not None
@@ -877,7 +1041,7 @@ def test_empty_refs_fold_into_merged_claim(self, tmp_path, monkeypatch):
         adapter = ChannelDataCache(store)
         try:
             data = _put(adapter, "c1", rows=4, start=_NOW, freq="s")
-            data_end = data.index[-1].to_pydatetime()
+            data_end = data.index[-1]
             empty_end = data_end + timedelta(seconds=10)
             # Second put (empty) crosses the cap of 2 → compacts.
             adapter.put_segment("c1", None, pd.DataFrame(), data_end, empty_end)
@@ -890,12 +1054,12 @@ def test_empty_refs_fold_into_merged_claim(self, tmp_path, monkeypatch):
             assert len(idx.segments) == 1
             sole = idx.segments[0]
             assert sole.seg_id is not None  # data ref survives
-            assert sole.start_time == data.index[0].to_pydatetime()
+            assert sole.start_time == data.index[0]
             assert sole.end_time == empty_end + timedelta(seconds=10)
 
             # Query the wider claim — data slice returns the actual
             # rows, and the empty tail counts as covered (no gap).
-            got, gaps = adapter.get_range("c1", None, data.index[0].to_pydatetime(), sole.end_time)
+            got, gaps = adapter.get_range("c1", None, data.index[0], sole.end_time)
             assert got is not None
             pd.testing.assert_frame_equal(got, data)
             assert gaps == []
@@ -1565,9 +1729,9 @@ def test_update_cache_merges_elements_into_one_segment(self, tmp_path) -> None:
         try:
             client._update_cache(
                 channel_data={"ch1.lo": df_lo, "ch1.hi": df_hi},
-                fetched_ranges_per_channel={"abc": [(_NOW, _WINDOW_END)]},
-                start_time=_NOW,
-                end_time=_WINDOW_END,
+                fetched_ranges_per_channel={"abc": [(_NOW_TS, _WINDOW_END_TS)]},
+                start_time=_NOW_TS,
+                end_time=_WINDOW_END_TS,
             )
             assert client.channel_cache.has_any("abc")
             assert not client.channel_cache.has_any("ch1.lo")
@@ -1604,9 +1768,9 @@ def test_get_range_treats_bitfield_as_one_bucket(self, tmp_path) -> None:
                     "ch1.lo": _frame("ch1.lo", rows=5),
                     "ch1.hi": _frame("ch1.hi", rows=5),
                 },
-                fetched_ranges_per_channel={"abc": [(_NOW, _WINDOW_END)]},
-                start_time=_NOW,
-                end_time=_WINDOW_END,
+                fetched_ranges_per_channel={"abc": [(_NOW_TS, _WINDOW_END_TS)]},
+                start_time=_NOW_TS,
+                end_time=_WINDOW_END_TS,
             )
             data, gaps = client.channel_cache.get_range("abc", None, _NOW, _WINDOW_END)
             assert data is not None
diff --git a/python/lib/sift_client/_tests/_internal/test_time.py b/python/lib/sift_client/_tests/_internal/test_time.py
new file mode 100644
index 0000000000..bc509e5d9e
--- /dev/null
+++ b/python/lib/sift_client/_tests/_internal/test_time.py
@@ -0,0 +1,78 @@
+"""Tests for :mod:`sift_client._internal.time`.
+
+The conversions here sit between the wire (nanosecond protobuf
+timestamps) and pandas (nanosecond ``Timestamp``), with ``datetime`` —
+microseconds at best — as the lossy type in the middle. These tests pin
+where precision is kept and where it legitimately isn't.
+
+The service filters query bounds on a millisecond plus sub-millisecond
+pair, so a nanosecond bound is meaningful all the way down. Truncating
+one on the way out is a client-side loss, not a match to the server's
+resolution.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pandas as pd
+import pytest
+from google.protobuf.timestamp_pb2 import Timestamp as TimestampPb
+
+from sift_client._internal.time import to_timestamp_nanos, to_timestamp_pb
+
+_NANOS = 123456789
+_SECONDS = 1735689600  # 2025-01-01T00:00:00Z
+
+
+class TestToTimestampPb:
+    def test_pandas_timestamp_keeps_nanoseconds(self) -> None:
+        """A ``pd.Timestamp`` survives the hop to protobuf intact.
+
+        ``TimestampPb.FromDatetime`` reads ``microsecond`` and nothing
+        finer, so a nanosecond-precise bound would otherwise reach the
+        wire rounded down.
+        """
+        ts = pd.Timestamp(_SECONDS, unit="s", tz=timezone.utc) + pd.Timedelta(_NANOS, unit="ns")
+        pb = to_timestamp_pb(ts)
+        assert (pb.seconds, pb.nanos) == (_SECONDS, _NANOS)
+
+    def test_coarser_units_take_the_datetime_path(self) -> None:
+        """A non-nanosecond ``Timestamp`` converts without losing anything.
+
+        Pandas 2.0 timestamps carry a storage unit. At ``us`` or
+        coarser there is no sub-microsecond remainder, so
+        ``FromDatetime`` is lossless.
+        """
+        ts = pd.Timestamp(_SECONDS, unit="s", tz=timezone.utc) + pd.Timedelta(_NANOS, unit="ns")
+        assert to_timestamp_pb(ts.as_unit("us")).nanos == 123456000
+        assert to_timestamp_pb(ts.as_unit("ms")).nanos == 123000000
+
+    @pytest.mark.parametrize("stamp", ["2300-01-01", "1500-06-01"], ids=["future", "past"])
+    def test_dates_outside_the_nanosecond_range(self, stamp: str) -> None:
+        """Dates a ``Timestamp`` can hold but nanoseconds can't still convert.
+
+        Only ``ns``-unit timestamps are bounded to roughly 1677-2262. A
+        coarser unit reaches further, and reading ``value`` on one of
+        those raises ``OverflowError``, so those take the
+        ``FromDatetime`` path instead.
+        """
+        ts = pd.Timestamp(stamp, tz=timezone.utc).as_unit("s")
+        assert to_timestamp_pb(ts).ToDatetime(tzinfo=timezone.utc) == ts.to_pydatetime()
+
+    def test_datetime_carries_its_microseconds(self) -> None:
+        """A plain ``datetime`` converts at the resolution it has."""
+        dt = datetime(2025, 1, 1, microsecond=123456, tzinfo=timezone.utc)
+        pb = to_timestamp_pb(dt)
+        assert (pb.seconds, pb.nanos) == (_SECONDS, 123456000)
+
+    def test_epoch_seconds_and_iso_string(self) -> None:
+        """The numeric and string forms stay on the ``datetime`` path."""
+        assert to_timestamp_pb(_SECONDS).seconds == _SECONDS
+        assert to_timestamp_pb("2025-01-01T00:00:00+00:00").seconds == _SECONDS
+
+    def test_round_trips_through_to_timestamp_nanos(self) -> None:
+        """Proto to pandas and back is lossless at nanosecond resolution."""
+        pb = TimestampPb(seconds=_SECONDS, nanos=_NANOS)
+        again = to_timestamp_pb(to_timestamp_nanos(pb))
+        assert (again.seconds, again.nanos) == (_SECONDS, _NANOS)
diff --git a/python/lib/sift_client/resources/channels.py b/python/lib/sift_client/resources/channels.py
index 1d8c8d81f0..ee4b4de811 100644
--- a/python/lib/sift_client/resources/channels.py
+++ b/python/lib/sift_client/resources/channels.py
@@ -262,8 +262,8 @@ async def get_data(
         *,
         channels: list[Channel],
         run: Run | str | None = None,
-        start_time: datetime | None = None,
-        end_time: datetime | None = None,
+        start_time: pd.Timestamp | datetime | None = None,
+        end_time: pd.Timestamp | datetime | None = None,
         limit: int | None = None,
         page_size: int | None = None,
         ignore_cache: bool = False,
@@ -274,8 +274,11 @@ async def get_data(
         Args:
             channels: The channels to get data for.
             run: The Run or run_id to get data for.
-            start_time: The start time to get data for.
-            end_time: The end time to get data for.
+            start_time: The start time to get data for, inclusive. Pass a
+                ``pd.Timestamp`` to bound the query at nanosecond
+                resolution; a ``datetime`` bounds it at microseconds.
+            end_time: The end time to get data for, exclusive. Same
+                resolution rule as ``start_time``.
             limit: The maximum number of data points to return. Will be in increments of page_size or default page size defined by the call if no page_size is provided.
             page_size: Number of data points to fetch per request. Defaults to 10,000.
                 The server caps this at 1,000,000; values above that are coerced down.
@@ -310,8 +313,8 @@ async def get_data_as_arrow(
         *,
         channels: list[Channel],
         run: Run | str | None = None,
-        start_time: datetime | None = None,
-        end_time: datetime | None = None,
+        start_time: pd.Timestamp | datetime | None = None,
+        end_time: pd.Timestamp | datetime | None = None,
         limit: int | None = None,
         page_size: int | None = None,
         ignore_cache: bool = False,
diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi
index a030f42069..078d870b1d 100644
--- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi
+++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi
@@ -493,8 +493,8 @@ class ChannelsAPI:
         *,
         channels: list[Channel],
         run: Run | str | None = None,
-        start_time: datetime | None = None,
-        end_time: datetime | None = None,
+        start_time: pd.Timestamp | datetime | None = None,
+        end_time: pd.Timestamp | datetime | None = None,
         limit: int | None = None,
         page_size: int | None = None,
         ignore_cache: bool = False,
@@ -505,8 +505,11 @@ class ChannelsAPI:
         Args:
             channels: The channels to get data for.
             run: The Run or run_id to get data for.
-            start_time: The start time to get data for.
-            end_time: The end time to get data for.
+            start_time: The start time to get data for, inclusive. Pass a
+                ``pd.Timestamp`` to bound the query at nanosecond
+                resolution; a ``datetime`` bounds it at microseconds.
+            end_time: The end time to get data for, exclusive. Same
+                resolution rule as ``start_time``.
             limit: The maximum number of data points to return. Will be in increments of page_size or default page size defined by the call if no page_size is provided.
             page_size: Number of data points to fetch per request. Defaults to 10,000.
                 The server caps this at 1,000,000; values above that are coerced down.
@@ -526,8 +529,8 @@ class ChannelsAPI:
         *,
         channels: list[Channel],
         run: Run | str | None = None,
-        start_time: datetime | None = None,
-        end_time: datetime | None = None,
+        start_time: pd.Timestamp | datetime | None = None,
+        end_time: pd.Timestamp | datetime | None = None,
         limit: int | None = None,
         page_size: int | None = None,
         ignore_cache: bool = False,
diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py
index 1966109dfa..55147f808b 100644
--- a/python/lib/sift_client/sift_types/data_import.py
+++ b/python/lib/sift_client/sift_types/data_import.py
@@ -55,7 +55,7 @@
 from sift.data_imports.v2.data_imports_pb2 import UlogConfig as UlogConfigProto
 from sift.data_imports.v2.data_imports_pb2 import UlogDataConfig as UlogDataConfigProto
 
-from sift_client._internal.util.timestamp import to_pb_timestamp
+from sift_client._internal.time import to_timestamp_pb
 from sift_client.sift_types._base import BaseType
 from sift_client.sift_types.channel import ChannelDataType
 
@@ -217,7 +217,7 @@ def _to_proto(self) -> CsvTimeColumnProto:
             format=self.format.value,
         )
         if self.relative_start_time is not None:
-            proto.relative_start_time.CopyFrom(to_pb_timestamp(self.relative_start_time))
+            proto.relative_start_time.CopyFrom(to_timestamp_pb(self.relative_start_time))
         return proto
 
 
@@ -342,7 +342,7 @@ def _to_proto(self) -> ParquetTimeColumnProto:
             format=self.format.value,
         )
         if self.relative_start_time is not None:
-            proto.relative_start_time.CopyFrom(to_pb_timestamp(self.relative_start_time))
+            proto.relative_start_time.CopyFrom(to_timestamp_pb(self.relative_start_time))
         return proto
 
     @classmethod
@@ -687,12 +687,12 @@ def _to_proto(self) -> TDMSConfigProto:
             import_file_properties=self.import_file_properties,
         )
         if self.start_time_override is not None:
-            proto.start_time_override.CopyFrom(to_pb_timestamp(self.start_time_override))
+            proto.start_time_override.CopyFrom(to_timestamp_pb(self.start_time_override))
         proto.fallback_method = self.fallback_method.value
         if self.time_format is not None:
             proto.time_format = self.time_format.value
         if self.relative_start_time is not None:
-            proto.relative_start_time.CopyFrom(to_pb_timestamp(self.relative_start_time))
+            proto.relative_start_time.CopyFrom(to_timestamp_pb(self.relative_start_time))
         for d in self.data:
             entry = TdmsDataConfigProto(
                 group_name=d.group_name,
@@ -855,7 +855,7 @@ def _to_proto(self) -> Hdf5ConfigProto:
             ],
         )
         if self.relative_start_time is not None:
-            proto.relative_start_time.CopyFrom(to_pb_timestamp(self.relative_start_time))
+            proto.relative_start_time.CopyFrom(to_timestamp_pb(self.relative_start_time))
         return proto
 
 
@@ -964,7 +964,7 @@ def _to_proto(self) -> UlogConfigProto:
             parse_error_policy=self.parse_error_policy.value,
         )
         if self.relative_start_time is not None:
-            proto.relative_start_time.CopyFrom(to_pb_timestamp(self.relative_start_time))
+            proto.relative_start_time.CopyFrom(to_timestamp_pb(self.relative_start_time))
         for dc in self.data:
             proto.data.append(
                 UlogDataConfigProto(
diff --git a/python/lib/sift_client/sift_types/tag.py b/python/lib/sift_client/sift_types/tag.py
index 6a565c657e..8e33381878 100644
--- a/python/lib/sift_client/sift_types/tag.py
+++ b/python/lib/sift_client/sift_types/tag.py
@@ -6,7 +6,7 @@
 from sift.tags.v2.tags_pb2 import CreateTagRequest as CreateTagRequestProto
 from sift.tags.v2.tags_pb2 import Tag as TagProto
 
-from sift_client._internal.util.timestamp import to_pb_timestamp
+from sift_client._internal.time import to_timestamp_pb
 from sift_client.sift_types._base import (
     BaseType,
     ModelCreate,
@@ -67,7 +67,7 @@ def _to_proto(self) -> TagProto:
             tag_id=self.id_ or "",
             name=self.name,
             created_by_user_id=self.created_by_user_id,
-            created_date=to_pb_timestamp(self.created_date),
+            created_date=to_timestamp_pb(self.created_date),
         )
         return proto