From 4f3d41e2ce460cf5ee54a8bd2863432f7f9cba65 Mon Sep 17 00:00:00 2001
From: GNERSIS <gornersisyan4@gmail.com>
Date: Wed, 2 Sep 2026 22:15:53 +0100
Subject: [PATCH 1/4] feat(sdk): shared timestamp-axis policy header (0.27.0)

Five official plugins each carry their own rule for choosing a topic's time
axis, with four different name lists and plausibility rules and no test that
any two agree: the same Parquet file gets a different axis through the parquet
loader than through arrow-ipc, an int8 column named `ts` could silently become
an axis, and two plugins carried verbatim copies of each other's seconds-to-
nanoseconds conversion that have since diverged (#186).

pj_plugins/sdk/timestamp_policy.hpp is the one contract, in the shape of
parser_array_policy.hpp: a native-timestamp type pass, then a name pass over
the union of every list in use, restricted to storage that can actually hold
an epoch nanosecond (TIMESTAMP, int64, uint64, double) on scalar leaves; an
explicit narrow-integer, uint32 or float32 axis is accepted with a warning the
plugin surfaces; secondsToNanoseconds is integer-split so every plugin and
every platform computes the same nanosecond; and timestamp_column /
timestamp_unit are canonical config keys so the unit stops being inferred from
storage type.

Header-only, additive, no ABI change. Adoption in parser_arrow, toolbox_mosaico,
data_load_parquet and data_load_lerobot is separate work in pj-official-plugins.
---
 CHANGELOG.md                                  |  15 +
 CLAUDE.md                                     |   4 +-
 VERSION                                       |   2 +-
 .../pj_plugins/sdk/timestamp_policy.hpp       | 297 ++++++++++++++++++
 pj_plugins/tests/plugin_sdk_helpers_test.cpp  | 168 ++++++++++
 5 files changed, 484 insertions(+), 2 deletions(-)
 create mode 100644 pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 54127ce7..b72bcf44 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,21 @@
 All notable changes to `plotjuggler_sdk` are recorded here. Versioning policy is in
 [`CLAUDE.md`](./CLAUDE.md) → "Release Versioning".
 
+## [0.27.0]
+
+### Feature: shared timestamp-axis policy for plugins (MINOR)
+
+`pj_plugins/sdk/timestamp_policy.hpp` is meant to replace the five divergent per-plugin
+timestamp-axis detectors inventoried in
+[#186](https://github.com/PlotJuggler/plotjuggler_sdk/issues/186) with one
+header-only contract: native timestamp storage first, then canonical names restricted to
+plausible scalar storage (`TIMESTAMP`, `int64`, `uint64`, or `double`), never expanded list
+elements. Explicit narrow-integer, `uint32`, and `float32` axes carry a shared warning.
+Canonical `timestamp_column` / `timestamp_unit` (`ns` | `us` | `ms` | `s`) JSON keys stop
+unit inference from being private plugin policy, and the integer-split seconds-to-nanoseconds
+helper makes rounding and overflow handling platform-independent. No ABI change;
+`abi/baseline.abi` untouched.
+
 ## [0.26.0]
 
 ### Feature: `GridMap` canonical builtin object (MINOR)
diff --git a/CLAUDE.md b/CLAUDE.md
index 40b7347f..5d8508ea 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -32,7 +32,9 @@ not in the PJ4 superproject. This file is the root navigation node for the whole
 - **pj_plugins** — host-side loaders + RAII handles + plugin **discovery** (directory scan +
   embedded-manifest inspection) for four plugin families (DataSource, MessageParser, Dialog, Toolbox),
   parser claim admission/resolution and native functional parser-module execution,
-  config-envelope helpers, and the **dialog C ABI** (`pj_plugins/dialog_protocol/`). The
+  config-envelope helpers, shared plugin-authoring policies
+  (`pj_plugins/sdk/parser_array_policy.hpp`, `pj_plugins/sdk/timestamp_policy.hpp`), and the
+  **dialog C ABI** (`pj_plugins/dialog_protocol/`). The
   duplicate-resolution *catalog* (which copy wins by priority/version/compatibility) is host policy
   and lives in the app (`pj_runtime`), built on these discovery primitives. Note the split: the DataSource/MessageParser/Toolbox C-ABI
   protocol headers live in `pj_base`; the **Dialog** protocol header lives here, not in `pj_base`.
diff --git a/VERSION b/VERSION
index 4e8f395f..1b58cc10 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-0.26.0
+0.27.0
diff --git a/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp b/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
new file mode 100644
index 00000000..2cb27951
--- /dev/null
+++ b/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
@@ -0,0 +1,297 @@
+/**
+ * @file timestamp_policy.hpp
+ * @brief Shared timestamp-axis detection, unit persistence, and conversion
+ *        policy for plugins that import columnar data.
+ *
+ * A column name alone is not enough to make a useful epoch axis: narrow
+ * integers overflow near the epoch, float32 loses sub-second resolution, and
+ * expanded list elements are not scalar columns. This header keeps those
+ * decisions consistent without depending on a particular columnar library.
+ */
+// Copyright 2026 Davide Faconti
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <array>
+#include <cmath>
+#include <cstddef>
+#include <cstdint>
+#include <limits>
+#include <nlohmann/json.hpp>
+#include <optional>
+#include <span>
+#include <string_view>
+
+namespace PJ {
+namespace sdk {
+
+/// How a candidate column's storage relates to an epoch-nanosecond axis.
+enum class TimeKind : uint8_t {
+  /// A native timestamp whose unit is known to the caller.
+  kNativeTimestamp,
+  /// A signed 64-bit integer containing nanoseconds.
+  kInt64,
+  /// An unsigned 64-bit integer containing nanoseconds.
+  kUInt64,
+  /// Double-precision seconds, with about 238 ns resolution at the present epoch.
+  kFloat64,
+  /// Unsigned 32-bit nanoseconds, which end about 4.3 seconds after the epoch.
+  kUInt32,
+  /// Signed 8/16/32-bit or unsigned 8/16-bit nanoseconds.
+  kNarrowInt,
+  /// Single-precision seconds, whose spacing reaches one second at 2^23 seconds.
+  kFloat32,
+  /// Storage that cannot serve as a timestamp axis.
+  kOther,
+};
+
+/// Whether a TimeKind can be auto-selected or must be handled explicitly.
+enum class AxisSupport : uint8_t {
+  /// Safe enough for automatic timestamp-axis selection.
+  kPlausible,
+  /// Available only after surfacing explicitAxisWarning().
+  kAcceptedWithWarning,
+  /// Cannot serve as a timestamp axis.
+  kUnsupported,
+};
+
+/// Classifies timestamp-axis support without inspecting a column name.
+[[nodiscard]] constexpr AxisSupport axisSupport(TimeKind kind) noexcept {
+  switch (kind) {
+    case TimeKind::kNativeTimestamp:
+    case TimeKind::kInt64:
+    case TimeKind::kUInt64:
+    case TimeKind::kFloat64:
+      return AxisSupport::kPlausible;
+    case TimeKind::kUInt32:
+    case TimeKind::kNarrowInt:
+    case TimeKind::kFloat32:
+      return AxisSupport::kAcceptedWithWarning;
+    case TimeKind::kOther:
+      return AxisSupport::kUnsupported;
+  }
+  return AxisSupport::kUnsupported;
+}
+
+/// Returns the warning a plugin must surface for an explicitly selected lossy
+/// or short-range axis; all other kinds return an empty view without allocating.
+[[nodiscard]] constexpr std::string_view explicitAxisWarning(TimeKind kind) noexcept {
+  switch (kind) {
+    case TimeKind::kUInt32:
+      return "uint32 can express at most 4294967295 ns since the Unix epoch.";
+    case TimeKind::kNarrowInt:
+      return "Narrow integers can express at most 2147483647 ns since the Unix epoch.";
+    case TimeKind::kFloat32:
+      return "float32 seconds reach 1-second spacing at 2^23 (8388608) seconds from the Unix epoch.";
+    case TimeKind::kNativeTimestamp:
+    case TimeKind::kInt64:
+    case TimeKind::kUInt64:
+    case TimeKind::kFloat64:
+    case TimeKind::kOther:
+      return {};
+  }
+  return {};
+}
+
+/// Arrow-independent description of a flattened column considered for the axis.
+struct TimestampCandidate {
+  /// Flattened leaf path; separators are '/', with source dots already normalized.
+  std::string_view name;
+  /// Storage classification supplied by the importing plugin.
+  TimeKind kind;
+  /// Expanded list elements are never eligible for automatic selection.
+  bool is_list_element = false;
+};
+
+/// Ordered name preferences used after native timestamp-type detection.
+struct TimestampPolicy {
+  /// Candidate names in priority order, with the most specific first.
+  std::span<const std::string_view> names;
+  /// Whether the name pass also accepts ASCII case-folded matches.
+  bool case_insensitive = true;
+};
+
+/// Union of timestamp names used by official plugins, most specific first.
+inline constexpr std::array<std::string_view, 11> kCanonicalTimestampNames = {"timestamp_ns", "recording_timestamp_ns",
+                                                                              "timestamp",    "time",
+                                                                              "ts",           "t",
+                                                                              "time_stamp",   "datetime",
+                                                                              "date_time",    "_timestamp",
+                                                                              "_time"};
+
+/// Default policy shared by official plugins.
+inline constexpr TimestampPolicy kCanonicalPolicy{kCanonicalTimestampNames, true};
+
+namespace detail {
+
+[[nodiscard]] constexpr char foldTimestampAscii(char value) noexcept {
+  if (value >= 'A' && value <= 'Z') {
+    return static_cast<char>(value + ('a' - 'A'));
+  }
+  return value;
+}
+
+[[nodiscard]] constexpr bool timestampNamesEqualFolded(std::string_view left, std::string_view right) noexcept {
+  if (left.size() != right.size()) {
+    return false;
+  }
+  for (std::size_t index = 0; index < left.size(); ++index) {
+    if (foldTimestampAscii(left[index]) != foldTimestampAscii(right[index])) {
+      return false;
+    }
+  }
+  return true;
+}
+
+}  // namespace detail
+
+/// Selects a timestamp column with a native-type pass followed by a plausible
+/// scalar name pass. Exact-case matches win within each preferred name before
+/// allocation-free ASCII case folding is considered.
+[[nodiscard]] constexpr std::optional<std::size_t> detectTimestampColumn(
+    std::span<const TimestampCandidate> candidates, const TimestampPolicy& policy = kCanonicalPolicy) {
+  for (std::size_t index = 0; index < candidates.size(); ++index) {
+    const TimestampCandidate& candidate = candidates[index];
+    if (!candidate.is_list_element && candidate.kind == TimeKind::kNativeTimestamp) {
+      return index;
+    }
+  }
+
+  for (const std::string_view preferred_name : policy.names) {
+    for (std::size_t index = 0; index < candidates.size(); ++index) {
+      const TimestampCandidate& candidate = candidates[index];
+      if (!candidate.is_list_element && axisSupport(candidate.kind) == AxisSupport::kPlausible &&
+          candidate.name == preferred_name) {
+        return index;
+      }
+    }
+
+    if (!policy.case_insensitive) {
+      continue;
+    }
+    for (std::size_t index = 0; index < candidates.size(); ++index) {
+      const TimestampCandidate& candidate = candidates[index];
+      if (!candidate.is_list_element && axisSupport(candidate.kind) == AxisSupport::kPlausible &&
+          detail::timestampNamesEqualFolded(candidate.name, preferred_name)) {
+        return index;
+      }
+    }
+  }
+  return std::nullopt;
+}
+
+/// Converts seconds to nanoseconds without platform-dependent long-double
+/// arithmetic. Fractional nanoseconds round halfway away from zero; non-finite
+/// input and either whole-second or final-addition overflow return nullopt.
+[[nodiscard]] inline std::optional<int64_t> secondsToNanoseconds(double seconds) noexcept {
+  if (!std::isfinite(seconds)) {
+    return std::nullopt;
+  }
+
+  constexpr int64_t kNanosecondsPerSecond = 1'000'000'000;
+  constexpr int64_t kMaximumWholeSeconds = std::numeric_limits<int64_t>::max() / kNanosecondsPerSecond;
+  constexpr int64_t kMinimumWholeSeconds = std::numeric_limits<int64_t>::min() / kNanosecondsPerSecond;
+
+  double whole_seconds = 0.0;
+  const double fractional_seconds = std::modf(seconds, &whole_seconds);
+  if (whole_seconds > static_cast<double>(kMaximumWholeSeconds) ||
+      whole_seconds < static_cast<double>(kMinimumWholeSeconds)) {
+    return std::nullopt;
+  }
+
+  const int64_t whole_nanoseconds = static_cast<int64_t>(whole_seconds) * kNanosecondsPerSecond;
+  const int64_t fractional_nanoseconds =
+      static_cast<int64_t>(std::llround(fractional_seconds * static_cast<double>(kNanosecondsPerSecond)));
+
+  if ((fractional_nanoseconds > 0 &&
+       whole_nanoseconds > std::numeric_limits<int64_t>::max() - fractional_nanoseconds) ||
+      (fractional_nanoseconds < 0 &&
+       whole_nanoseconds < std::numeric_limits<int64_t>::min() - fractional_nanoseconds)) {
+    return std::nullopt;
+  }
+  return whole_nanoseconds + fractional_nanoseconds;
+}
+
+/// Canonical JSON key for the selected timestamp column.
+inline constexpr std::string_view kTimestampColumnKey = "timestamp_column";
+
+/// Canonical JSON key for an integer timestamp column's unit.
+inline constexpr std::string_view kTimestampUnitKey = "timestamp_unit";
+
+/// Units accepted by the shared timestamp-axis configuration contract.
+enum class TimestampUnit : uint8_t {
+  /// Nanoseconds ("ns"); the compatibility default for integer columns.
+  kNanoseconds,
+  /// Microseconds ("us").
+  kMicroseconds,
+  /// Milliseconds ("ms").
+  kMilliseconds,
+  /// Seconds ("s").
+  kSeconds,
+};
+
+/// Returns the integral nanosecond scale for a configured timestamp unit.
+[[nodiscard]] constexpr int64_t nanosecondsPer(TimestampUnit unit) noexcept {
+  switch (unit) {
+    case TimestampUnit::kNanoseconds:
+      return 1;
+    case TimestampUnit::kMicroseconds:
+      return 1'000;
+    case TimestampUnit::kMilliseconds:
+      return 1'000'000;
+    case TimestampUnit::kSeconds:
+      return 1'000'000'000;
+  }
+  return 0;
+}
+
+/// Reads "ns", "us", "ms", or "s" from kTimestampUnitKey. A missing key
+/// preserves the historical nanosecond default; malformed or unknown values
+/// return nullopt so callers can reject the named config field.
+[[nodiscard]] inline std::optional<TimestampUnit> timestampUnitFromJson(const nlohmann::json& object) {
+  const auto unit_it = object.find(kTimestampUnitKey.data());
+  if (unit_it == object.end()) {
+    return TimestampUnit::kNanoseconds;
+  }
+  if (!unit_it->is_string()) {
+    return std::nullopt;
+  }
+
+  const auto& value = unit_it->get_ref<const nlohmann::json::string_t&>();
+  if (value == "ns") {
+    return TimestampUnit::kNanoseconds;
+  }
+  if (value == "us") {
+    return TimestampUnit::kMicroseconds;
+  }
+  if (value == "ms") {
+    return TimestampUnit::kMilliseconds;
+  }
+  if (value == "s") {
+    return TimestampUnit::kSeconds;
+  }
+  return std::nullopt;
+}
+
+/// Writes a TimestampUnit using the canonical short spelling under
+/// kTimestampUnitKey, converting a null JSON value to an object as needed.
+inline void timestampUnitToJson(nlohmann::json& object, TimestampUnit unit) {
+  switch (unit) {
+    case TimestampUnit::kNanoseconds:
+      object[kTimestampUnitKey.data()] = "ns";
+      return;
+    case TimestampUnit::kMicroseconds:
+      object[kTimestampUnitKey.data()] = "us";
+      return;
+    case TimestampUnit::kMilliseconds:
+      object[kTimestampUnitKey.data()] = "ms";
+      return;
+    case TimestampUnit::kSeconds:
+      object[kTimestampUnitKey.data()] = "s";
+      return;
+  }
+}
+
+}  // namespace sdk
+}  // namespace PJ
diff --git a/pj_plugins/tests/plugin_sdk_helpers_test.cpp b/pj_plugins/tests/plugin_sdk_helpers_test.cpp
index 5f62c68f..3de372c7 100644
--- a/pj_plugins/tests/plugin_sdk_helpers_test.cpp
+++ b/pj_plugins/tests/plugin_sdk_helpers_test.cpp
@@ -7,15 +7,21 @@
 
 #include <gtest/gtest.h>
 
+#include <array>
+#include <cstdint>
+#include <limits>
 #include <nlohmann/json.hpp>
+#include <optional>
 #include <set>
 #include <string>
+#include <string_view>
 #include <vector>
 
 #include "pj_plugins/sdk/endpoint.hpp"
 #include "pj_plugins/sdk/parser_array_policy.hpp"
 #include "pj_plugins/sdk/streaming_dialog.hpp"
 #include "pj_plugins/sdk/streaming_source.hpp"
+#include "pj_plugins/sdk/timestamp_policy.hpp"
 
 namespace {
 
@@ -240,4 +246,166 @@ TEST(DelegatedIngestTest, BindingFailureIsANonFatalDisposition) {
   EXPECT_EQ(*result, PJ::sdk::DelegatedIngestDisposition::kBindingUnavailable);
 }
 
+// ---------------------------------------------------------------------------
+// Timestamp policy
+// ---------------------------------------------------------------------------
+
+TEST(TimestampPolicyTest, NativeTimestampTypePassWinsOverPreferredName) {
+  const PJ::sdk::TimestampCandidate candidates[] = {
+      {.name = "timestamp_ns", .kind = PJ::sdk::TimeKind::kInt64},
+      {.name = "foo", .kind = PJ::sdk::TimeKind::kNativeTimestamp},
+  };
+
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
+}
+
+TEST(TimestampPolicyTest, NamePriorityWinsOverCandidateOrder) {
+  const PJ::sdk::TimestampCandidate candidates[] = {
+      {.name = "time", .kind = PJ::sdk::TimeKind::kInt64},
+      {.name = "recording_timestamp_ns", .kind = PJ::sdk::TimeKind::kInt64},
+  };
+
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
+}
+
+TEST(TimestampPolicyTest, NarrowIntegerNameIsSkippedForPlausibleInt64) {
+  const PJ::sdk::TimestampCandidate candidates[] = {
+      {.name = "timestamp_ns", .kind = PJ::sdk::TimeKind::kNarrowInt},
+      {.name = "time", .kind = PJ::sdk::TimeKind::kInt64},
+  };
+
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
+}
+
+TEST(TimestampPolicyTest, UInt32AndFloat32NamesAreSkippedForFloat64) {
+  const PJ::sdk::TimestampCandidate candidates[] = {
+      {.name = "timestamp_ns", .kind = PJ::sdk::TimeKind::kUInt32},
+      {.name = "recording_timestamp_ns", .kind = PJ::sdk::TimeKind::kFloat32},
+      {.name = "time", .kind = PJ::sdk::TimeKind::kFloat64},
+  };
+
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{2});
+}
+
+TEST(TimestampPolicyTest, ListElementIsNeverSelected) {
+  const PJ::sdk::TimestampCandidate candidates[] = {
+      {.name = "timestamp", .kind = PJ::sdk::TimeKind::kNativeTimestamp, .is_list_element = true},
+  };
+
+  EXPECT_FALSE(PJ::sdk::detectTimestampColumn(candidates));
+}
+
+TEST(TimestampPolicyTest, CanonicalNamesMatchAsciiCaseInsensitively) {
+  const PJ::sdk::TimestampCandidate title_case[] = {
+      {.name = "Timestamp", .kind = PJ::sdk::TimeKind::kInt64},
+  };
+  const PJ::sdk::TimestampCandidate upper_case[] = {
+      {.name = "DATETIME", .kind = PJ::sdk::TimeKind::kFloat64},
+  };
+
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(title_case), std::optional<std::size_t>{0});
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(upper_case), std::optional<std::size_t>{0});
+}
+
+TEST(TimestampPolicyTest, ExactCaseWinsWithinPreferredNameRegardlessOfOrder) {
+  const PJ::sdk::TimestampCandidate candidates[] = {
+      {.name = "Timestamp", .kind = PJ::sdk::TimeKind::kInt64},
+      {.name = "timestamp", .kind = PJ::sdk::TimeKind::kInt64},
+  };
+
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
+}
+
+TEST(TimestampPolicyTest, CaseSensitiveCustomPolicyRejectsFoldedMatch) {
+  const std::string_view names[] = {"timestamp"};
+  const PJ::sdk::TimestampPolicy policy{.names = names, .case_insensitive = false};
+  const PJ::sdk::TimestampCandidate candidates[] = {
+      {.name = "Timestamp", .kind = PJ::sdk::TimeKind::kInt64},
+  };
+
+  EXPECT_FALSE(PJ::sdk::detectTimestampColumn(candidates, policy));
+}
+
+TEST(TimestampPolicyTest, SupportAndWarningsCoverEveryTimeKind) {
+  struct SupportCase {
+    PJ::sdk::TimeKind kind;
+    PJ::sdk::AxisSupport support;
+  };
+  constexpr std::array<SupportCase, 8> cases = {{
+      {PJ::sdk::TimeKind::kNativeTimestamp, PJ::sdk::AxisSupport::kPlausible},
+      {PJ::sdk::TimeKind::kInt64, PJ::sdk::AxisSupport::kPlausible},
+      {PJ::sdk::TimeKind::kUInt64, PJ::sdk::AxisSupport::kPlausible},
+      {PJ::sdk::TimeKind::kFloat64, PJ::sdk::AxisSupport::kPlausible},
+      {PJ::sdk::TimeKind::kUInt32, PJ::sdk::AxisSupport::kAcceptedWithWarning},
+      {PJ::sdk::TimeKind::kNarrowInt, PJ::sdk::AxisSupport::kAcceptedWithWarning},
+      {PJ::sdk::TimeKind::kFloat32, PJ::sdk::AxisSupport::kAcceptedWithWarning},
+      {PJ::sdk::TimeKind::kOther, PJ::sdk::AxisSupport::kUnsupported},
+  }};
+
+  for (const SupportCase& test_case : cases) {
+    EXPECT_EQ(PJ::sdk::axisSupport(test_case.kind), test_case.support);
+    EXPECT_EQ(
+        PJ::sdk::explicitAxisWarning(test_case.kind).empty(),
+        test_case.support != PJ::sdk::AxisSupport::kAcceptedWithWarning);
+  }
+}
+
+TEST(TimestampPolicyTest, SecondsToNanosecondsUsesIntegerSplitAndStableRounding) {
+  struct ConversionCase {
+    double seconds;
+    int64_t nanoseconds;
+  };
+  const ConversionCase cases[] = {
+      {1.5, 1'500'000'000}, {1.7e9 + 0.125, 1'700'000'000'125'000'000}, {-1.6e-9, -2}, {1.6e-9, 2}, {2.4e-9, 2},
+  };
+
+  for (const ConversionCase& test_case : cases) {
+    const auto converted = PJ::sdk::secondsToNanoseconds(test_case.seconds);
+    ASSERT_TRUE(converted);
+    EXPECT_EQ(*converted, test_case.nanoseconds);
+  }
+}
+
+TEST(TimestampPolicyTest, SecondsToNanosecondsRejectsNonFiniteAndOverflow) {
+  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(std::numeric_limits<double>::quiet_NaN()));
+  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(std::numeric_limits<double>::infinity()));
+  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(-std::numeric_limits<double>::infinity()));
+  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(9.3e9));
+  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(9223372036.0 + 0.999999999));
+}
+
+TEST(TimestampPolicyTest, TimestampUnitsReadEverySpellingAndRoundTrip) {
+  struct UnitCase {
+    const char* spelling;
+    PJ::sdk::TimestampUnit unit;
+    int64_t nanoseconds_per_unit;
+  };
+  const UnitCase cases[] = {
+      {"ns", PJ::sdk::TimestampUnit::kNanoseconds, 1},
+      {"us", PJ::sdk::TimestampUnit::kMicroseconds, 1'000},
+      {"ms", PJ::sdk::TimestampUnit::kMilliseconds, 1'000'000},
+      {"s", PJ::sdk::TimestampUnit::kSeconds, 1'000'000'000},
+  };
+
+  for (const UnitCase& test_case : cases) {
+    const nlohmann::json input = {{"timestamp_unit", test_case.spelling}};
+    EXPECT_EQ(PJ::sdk::timestampUnitFromJson(input), std::optional<PJ::sdk::TimestampUnit>{test_case.unit});
+    EXPECT_EQ(PJ::sdk::nanosecondsPer(test_case.unit), test_case.nanoseconds_per_unit);
+
+    nlohmann::json output;
+    PJ::sdk::timestampUnitToJson(output, test_case.unit);
+    EXPECT_EQ(output.at("timestamp_unit"), test_case.spelling);
+    EXPECT_EQ(PJ::sdk::timestampUnitFromJson(output), std::optional<PJ::sdk::TimestampUnit>{test_case.unit});
+  }
+}
+
+TEST(TimestampPolicyTest, MissingUnitDefaultsToNanosecondsAndUnknownIsRejected) {
+  EXPECT_EQ(
+      PJ::sdk::timestampUnitFromJson(nlohmann::json::object()),
+      std::optional<PJ::sdk::TimestampUnit>{PJ::sdk::TimestampUnit::kNanoseconds});
+  EXPECT_FALSE(PJ::sdk::timestampUnitFromJson(nlohmann::json{{"timestamp_unit", "minutes"}}));
+  EXPECT_EQ(PJ::sdk::kTimestampColumnKey, "timestamp_column");
+  EXPECT_EQ(PJ::sdk::kTimestampUnitKey, "timestamp_unit");
+}
+
 }  // namespace

From 47610b5c8d6d255995efe9e4f8759535dcaa3f73 Mon Sep 17 00:00:00 2001
From: GNERSIS <gornersisyan4@gmail.com>
Date: Wed, 2 Sep 2026 22:42:48 +0100
Subject: [PATCH 2/4] refactor(pj_base): checked time arithmetic in a C++17
 time_math.hpp; policy header keeps only policy

The absolute time spine (pj_base/time.hpp) forbade hand-rolled 1e9
conversions in its own comment and then provided none, so the host, the
parser modules and every plugin wrote their own. The first cut of this branch
added the arithmetic to the plugin policy header, which the host cannot
include; and the spine itself is C++20 (sys_time), while parser-module headers
are held to C++17 by pj_parser_module's cxx_std_17 and a -std=c++17 compile
check.

pj_base/time_math.hpp is the C++17-clean home: TimeUnit and nanosecondsPer,
checked scaleToNanoseconds, widenUnsignedTicks, the integer-split
secondsToNanoseconds, combineSecondsAndNanos, syntheticInstant,
fitSyntheticInterval and kDefaultSyntheticIntervalNs, all constexpr/inline
with std::optional on overflow. time.hpp includes it so spine users see one
PJ surface; parser_module/time.hpp forwards its combine to it with error texts
unchanged; the new header joins the C++17 compile check.

pj_plugins/sdk/timestamp_policy.hpp keeps detection, axis support and the
config keys, now typed on PJ::TimeUnit, and adds synthetic_interval_ns and
flatten_structs as canonical keys. Header-only, no ABI change;
abi/baseline.abi byte-identical.
---
 CHANGELOG.md                                  |  28 ++--
 CLAUDE.md                                     |   1 +
 pj_base/CMakeLists.txt                        |   1 +
 .../include/pj_base/parser_module/time.hpp    |  12 +-
 pj_base/include/pj_base/time.hpp              |   3 +-
 pj_base/include/pj_base/time_math.hpp         | 132 ++++++++++++++++++
 pj_base/tests/time_spine_test.cpp             |  60 ++++++++
 .../pj_plugins/sdk/timestamp_policy.hpp       |  89 +++---------
 pj_plugins/tests/plugin_sdk_helpers_test.cpp  |  44 ++----
 9 files changed, 246 insertions(+), 124 deletions(-)
 create mode 100644 pj_base/include/pj_base/time_math.hpp

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b72bcf44..4e58871a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,18 +5,22 @@ All notable changes to `plotjuggler_sdk` are recorded here. Versioning policy is
 
 ## [0.27.0]
 
-### Feature: shared timestamp-axis policy for plugins (MINOR)
-
-`pj_plugins/sdk/timestamp_policy.hpp` is meant to replace the five divergent per-plugin
-timestamp-axis detectors inventoried in
-[#186](https://github.com/PlotJuggler/plotjuggler_sdk/issues/186) with one
-header-only contract: native timestamp storage first, then canonical names restricted to
-plausible scalar storage (`TIMESTAMP`, `int64`, `uint64`, or `double`), never expanded list
-elements. Explicit narrow-integer, `uint32`, and `float32` axes carry a shared warning.
-Canonical `timestamp_column` / `timestamp_unit` (`ns` | `us` | `ms` | `s`) JSON keys stop
-unit inference from being private plugin policy, and the integer-split seconds-to-nanoseconds
-helper makes rounding and overflow handling platform-independent. No ABI change;
-`abi/baseline.abi` untouched.
+### Feature: shared timestamp arithmetic and axis policy (MINOR)
+
+`pj_base/time_math.hpp` adds C++17-clean, checked time arithmetic usable by parser modules,
+the host, and plugins: `nanosecondsPer`, `scaleToNanoseconds`, `widenUnsignedTicks`,
+`secondsToNanoseconds`, `combineSecondsAndNanos`, `syntheticInstant`, and
+`fitSyntheticInterval`. The absolute-time spine re-exports this arithmetic through
+`pj_base/time.hpp`.
+
+Layered on that spine, `pj_plugins/sdk/timestamp_policy.hpp` is meant to replace the five
+divergent per-plugin timestamp-axis detectors inventoried in
+[#186](https://github.com/PlotJuggler/plotjuggler_sdk/issues/186) with one header-only
+detection and configuration contract: native timestamp storage first, then canonical names
+restricted to plausible scalar storage (`TIMESTAMP`, `int64`, `uint64`, or `double`), never
+expanded list elements. Explicit narrow-integer, `uint32`, and `float32` axes carry a shared
+warning, while canonical axis configuration keys and `PJ::TimeUnit` stop unit inference from
+being private plugin policy. No ABI change; `abi/baseline.abi` untouched.
 
 ## [0.26.0]
 
diff --git a/CLAUDE.md b/CLAUDE.md
index 5d8508ea..37fddc53 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -23,6 +23,7 @@ not in the PJ4 superproject. This file is the root navigation node for the whole
   standalone C++17 functional parser-module authoring kit (`pj_base/parser_module/`), the host-side
   wasm parser-module manifest custom-section codec, and the test-only static WASI ABI auditor. The
   0.22 authoring helper builds native parser modules only; wasm loading/execution is not present.
+  The absolute-time spine now also carries checked arithmetic shared across those layers.
 - **descriptor_import_support** — a separate compiled component
   (`plotjuggler_sdk::descriptor_import_support`, headers under
   `pj_base/sdk/descriptor_import/`): the callee side of the descriptor-import
diff --git a/pj_base/CMakeLists.txt b/pj_base/CMakeLists.txt
index b10c386a..f35ae922 100644
--- a/pj_base/CMakeLists.txt
+++ b/pj_base/CMakeLists.txt
@@ -293,6 +293,7 @@ if(PJ_BUILD_TESTS)
         -Wl,--export=pj_module_free
         -o "${_pj_wasm_raw}"
       DEPENDS "${_pj_wasm_source}" ${_pj_parser_module_headers}
+        "${CMAKE_CURRENT_SOURCE_DIR}/include/pj_base/time_math.hpp"
       COMMENT "Compiling C++17 parser-module WASI reactor fixture"
       VERBATIM
     )
diff --git a/pj_base/include/pj_base/parser_module/time.hpp b/pj_base/include/pj_base/parser_module/time.hpp
index b0671b4a..c8cb0d61 100644
--- a/pj_base/include/pj_base/parser_module/time.hpp
+++ b/pj_base/include/pj_base/parser_module/time.hpp
@@ -5,24 +5,22 @@
 /** @file time.hpp @brief Checked ROS and protobuf timestamp normalization. */
 
 #include <cstdint>
-#include <limits>
 
 #include "pj_base/parser_module/core.hpp"
+#include "pj_base/time_math.hpp"
 
 namespace pj {
 namespace detail {
 
 inline Expected<int64_t> combineSecondsAndNanos(int64_t seconds, int32_t nanos) {
-  constexpr int64_t kNanosPerSecond = INT64_C(1000000000);
-  if (nanos < 0 || nanos >= kNanosPerSecond) {
+  const auto combined = PJ::combineSecondsAndNanos(seconds, nanos);
+  if (nanos < 0 || nanos >= INT64_C(1000000000)) {
     return Status::error("timestamp nanoseconds are outside [0, 1000000000)");
   }
-  const int64_t positive_room = (std::numeric_limits<int64_t>::max() - nanos) / kNanosPerSecond;
-  const int64_t negative_room = std::numeric_limits<int64_t>::min() / kNanosPerSecond;
-  if (seconds > positive_room || seconds < negative_room) {
+  if (!combined) {
     return Status::error("timestamp is outside the int64 nanosecond range");
   }
-  return seconds * kNanosPerSecond + nanos;
+  return *combined;
 }
 
 }  // namespace detail
diff --git a/pj_base/include/pj_base/time.hpp b/pj_base/include/pj_base/time.hpp
index 99a01b8c..d307d840 100644
--- a/pj_base/include/pj_base/time.hpp
+++ b/pj_base/include/pj_base/time.hpp
@@ -18,7 +18,8 @@
 
 #include <chrono>
 
-#include "pj_base/types.hpp"  // PJ::Timestamp, PJ::Range
+#include "pj_base/time_math.hpp"  // Checked arithmetic shared with C++17 parser modules.
+#include "pj_base/types.hpp"      // PJ::Timestamp, PJ::Range
 
 namespace PJ {
 
diff --git a/pj_base/include/pj_base/time_math.hpp b/pj_base/include/pj_base/time_math.hpp
new file mode 100644
index 00000000..fb76a260
--- /dev/null
+++ b/pj_base/include/pj_base/time_math.hpp
@@ -0,0 +1,132 @@
+// Copyright 2026 Davide Faconti
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+/** @file time_math.hpp @brief C++17 checked arithmetic for nanosecond timestamps. */
+
+#include <cmath>
+#include <cstdint>
+#include <limits>
+#include <optional>
+
+namespace PJ {
+
+/// Units a tick count can be expressed in. Nanoseconds is the spine's native unit.
+enum class TimeUnit : uint8_t { kSeconds, kMilliseconds, kMicroseconds, kNanoseconds };
+
+/// Returns the integral nanosecond scale for a time unit.
+[[nodiscard]] constexpr int64_t nanosecondsPer(TimeUnit unit) noexcept {
+  switch (unit) {
+    case TimeUnit::kSeconds:
+      return 1'000'000'000;
+    case TimeUnit::kMilliseconds:
+      return 1'000'000;
+    case TimeUnit::kMicroseconds:
+      return 1'000;
+    case TimeUnit::kNanoseconds:
+      return 1;
+  }
+  return 0;
+}
+
+/// Converts ticks in the supplied unit to nanoseconds, returning nullopt on overflow.
+[[nodiscard]] constexpr std::optional<int64_t> scaleToNanoseconds(int64_t ticks, TimeUnit unit) noexcept {
+  const int64_t scale = nanosecondsPer(unit);
+  if (scale == 0 || ticks > std::numeric_limits<int64_t>::max() / scale ||
+      ticks < std::numeric_limits<int64_t>::min() / scale) {
+    return std::nullopt;
+  }
+  return ticks * scale;
+}
+
+/// Widens unsigned ticks to the signed timestamp representation, rejecting values above INT64_MAX.
+[[nodiscard]] constexpr std::optional<int64_t> widenUnsignedTicks(uint64_t ticks) noexcept {
+  if (ticks > static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
+    return std::nullopt;
+  }
+  return static_cast<int64_t>(ticks);
+}
+
+/// Converts floating seconds to nanoseconds using an integer split and half-away-from-zero rounding.
+/// Non-finite inputs and either whole-second or final-addition overflow return nullopt.
+[[nodiscard]] inline std::optional<int64_t> secondsToNanoseconds(double seconds) noexcept {
+  if (!std::isfinite(seconds)) {
+    return std::nullopt;
+  }
+
+  constexpr int64_t kNanosecondsPerSecond = 1'000'000'000;
+  constexpr int64_t kMaximumWholeSeconds = std::numeric_limits<int64_t>::max() / kNanosecondsPerSecond;
+  constexpr int64_t kMinimumWholeSeconds = std::numeric_limits<int64_t>::min() / kNanosecondsPerSecond;
+
+  double whole_seconds = 0.0;
+  const double fractional_seconds = std::modf(seconds, &whole_seconds);
+  if (whole_seconds > static_cast<double>(kMaximumWholeSeconds) ||
+      whole_seconds < static_cast<double>(kMinimumWholeSeconds)) {
+    return std::nullopt;
+  }
+
+  const int64_t whole_nanoseconds = static_cast<int64_t>(whole_seconds) * kNanosecondsPerSecond;
+  const int64_t fractional_nanoseconds =
+      static_cast<int64_t>(std::llround(fractional_seconds * static_cast<double>(kNanosecondsPerSecond)));
+  if ((fractional_nanoseconds > 0 &&
+       whole_nanoseconds > std::numeric_limits<int64_t>::max() - fractional_nanoseconds) ||
+      (fractional_nanoseconds < 0 &&
+       whole_nanoseconds < std::numeric_limits<int64_t>::min() - fractional_nanoseconds)) {
+    return std::nullopt;
+  }
+  return whole_nanoseconds + fractional_nanoseconds;
+}
+
+/// Combines seconds and nanoseconds-of-second; nanos must be in [0, 1e9).
+/// Returns nullopt when either the nanos range or the signed timestamp range would be exceeded.
+[[nodiscard]] constexpr std::optional<int64_t> combineSecondsAndNanos(int64_t seconds, int64_t nanos) noexcept {
+  constexpr int64_t kNanosecondsPerSecond = 1'000'000'000;
+  if (nanos < 0 || nanos >= kNanosecondsPerSecond) {
+    return std::nullopt;
+  }
+  const int64_t positive_room = (std::numeric_limits<int64_t>::max() - nanos) / kNanosecondsPerSecond;
+  const int64_t negative_room = std::numeric_limits<int64_t>::min() / kNanosecondsPerSecond;
+  if (seconds > positive_room || seconds < negative_room) {
+    return std::nullopt;
+  }
+  return seconds * kNanosecondsPerSecond + nanos;
+}
+
+/// Computes anchor + row * interval for a non-negative synthetic row, returning nullopt on overflow.
+[[nodiscard]] constexpr std::optional<int64_t> syntheticInstant(
+    int64_t anchor_ns, int64_t interval_ns, int64_t row) noexcept {
+  if (row < 0) {
+    return std::nullopt;
+  }
+  if ((interval_ns > 0 && row > std::numeric_limits<int64_t>::max() / interval_ns) ||
+      (interval_ns < 0 && row > 0 && interval_ns < std::numeric_limits<int64_t>::min() / row)) {
+    return std::nullopt;
+  }
+  const int64_t offset = interval_ns * row;
+  if ((offset > 0 && anchor_ns > std::numeric_limits<int64_t>::max() - offset) ||
+      (offset < 0 && anchor_ns < std::numeric_limits<int64_t>::min() - offset)) {
+    return std::nullopt;
+  }
+  return anchor_ns + offset;
+}
+
+/// Fits an interval across rows over [first, last], using fallback for insufficient rows,
+/// a non-positive span, or an interval that cannot be represented by int64_t.
+[[nodiscard]] constexpr int64_t fitSyntheticInterval(
+    int64_t first_ns, int64_t last_ns, int64_t rows, int64_t fallback_ns) noexcept {
+  if (rows < 2 || last_ns <= first_ns) {
+    return fallback_ns;
+  }
+  const uint64_t span = static_cast<uint64_t>(last_ns) - static_cast<uint64_t>(first_ns);
+  const uint64_t interval = span / static_cast<uint64_t>(rows - 1);
+  if (interval > static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
+    return fallback_ns;
+  }
+  return static_cast<int64_t>(interval);
+}
+
+/// Default cadence for a synthesized axis when nothing better is known (approximately 30 fps).
+inline constexpr int64_t kDefaultSyntheticIntervalNs = 33'333'333;
+
+}  // namespace PJ
diff --git a/pj_base/tests/time_spine_test.cpp b/pj_base/tests/time_spine_test.cpp
index b6edf09a..7e23866d 100644
--- a/pj_base/tests/time_spine_test.cpp
+++ b/pj_base/tests/time_spine_test.cpp
@@ -8,6 +8,9 @@
 
 #include <gtest/gtest.h>
 
+#include <cstdint>
+#include <limits>
+#include <optional>
 #include <type_traits>
 
 #include "pj_base/time.hpp"
@@ -39,4 +42,61 @@ TEST(TimeSpine, FromRawRangeLiftsBothEnds) {
   EXPECT_EQ(PJ::toRaw(lifted.max), 5'000'000'000LL);
 }
 
+TEST(TimeMath, ScalesTicksAndRejectsOverflow) {
+  EXPECT_EQ(PJ::scaleToNanoseconds(1, PJ::TimeUnit::kSeconds), std::optional<int64_t>{1'000'000'000});
+  EXPECT_FALSE(PJ::scaleToNanoseconds(9'223'372'037, PJ::TimeUnit::kSeconds));
+}
+
+TEST(TimeMath, WidensUnsignedTicksWithinSignedRange) {
+  EXPECT_FALSE(PJ::widenUnsignedTicks(std::numeric_limits<uint64_t>::max()));
+  EXPECT_EQ(
+      PJ::widenUnsignedTicks(static_cast<uint64_t>(std::numeric_limits<int64_t>::max())),
+      std::optional<int64_t>{std::numeric_limits<int64_t>::max()});
+}
+
+TEST(TimeMath, ConvertsSecondsUsingIntegerSplitAndStableRounding) {
+  struct ConversionCase {
+    double seconds;
+    int64_t nanoseconds;
+  };
+  const ConversionCase cases[] = {
+      {1.5, 1'500'000'000}, {1.7e9 + 0.125, 1'700'000'000'125'000'000}, {-1.6e-9, -2}, {1.6e-9, 2}, {2.4e-9, 2},
+  };
+
+  for (const ConversionCase& test_case : cases) {
+    const auto converted = PJ::secondsToNanoseconds(test_case.seconds);
+    ASSERT_TRUE(converted);
+    EXPECT_EQ(*converted, test_case.nanoseconds);
+  }
+}
+
+TEST(TimeMath, RejectsNonFiniteAndOverflowingSeconds) {
+  EXPECT_FALSE(PJ::secondsToNanoseconds(std::numeric_limits<double>::quiet_NaN()));
+  EXPECT_FALSE(PJ::secondsToNanoseconds(std::numeric_limits<double>::infinity()));
+  EXPECT_FALSE(PJ::secondsToNanoseconds(-std::numeric_limits<double>::infinity()));
+  EXPECT_FALSE(PJ::secondsToNanoseconds(9.3e9));
+  EXPECT_FALSE(PJ::secondsToNanoseconds(9223372036.0 + 0.999999999));
+}
+
+TEST(TimeMath, CombinesSecondsAndNanosecondsWithChecks) {
+  EXPECT_EQ(PJ::combineSecondsAndNanos(1, 5), std::optional<int64_t>{1'000'000'005});
+  EXPECT_FALSE(PJ::combineSecondsAndNanos(0, 1'000'000'000));
+  EXPECT_FALSE(PJ::combineSecondsAndNanos(0, -1));
+}
+
+TEST(TimeMath, ComputesSyntheticInstantsWithChecks) {
+  EXPECT_EQ(PJ::syntheticInstant(10, 3, 4), std::optional<int64_t>{22});
+  EXPECT_FALSE(PJ::syntheticInstant(std::numeric_limits<int64_t>::max(), 1, 1));
+  EXPECT_FALSE(PJ::syntheticInstant(std::numeric_limits<int64_t>::min(), -1, 1));
+  EXPECT_FALSE(PJ::syntheticInstant(10, 3, -1));
+}
+
+TEST(TimeMath, FitsSyntheticIntervalsOrUsesFallback) {
+  EXPECT_EQ(PJ::fitSyntheticInterval(1000, 4000, 5, 7), 750);
+  EXPECT_EQ(PJ::fitSyntheticInterval(1000, 1000, 5, 7), 7);
+  EXPECT_EQ(PJ::fitSyntheticInterval(1000, 4000, 1, 7), 7);
+  EXPECT_EQ(PJ::fitSyntheticInterval(4000, 1000, 5, 7), 7);
+  EXPECT_EQ(PJ::kDefaultSyntheticIntervalNs, 33'333'333);
+}
+
 }  // namespace
diff --git a/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp b/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
index 2cb27951..067fba61 100644
--- a/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
+++ b/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
@@ -14,15 +14,15 @@
 #pragma once
 
 #include <array>
-#include <cmath>
 #include <cstddef>
 #include <cstdint>
-#include <limits>
 #include <nlohmann/json.hpp>
 #include <optional>
 #include <span>
 #include <string_view>
 
+#include "pj_base/time.hpp"
+
 namespace PJ {
 namespace sdk {
 
@@ -181,78 +181,25 @@ namespace detail {
   return std::nullopt;
 }
 
-/// Converts seconds to nanoseconds without platform-dependent long-double
-/// arithmetic. Fractional nanoseconds round halfway away from zero; non-finite
-/// input and either whole-second or final-addition overflow return nullopt.
-[[nodiscard]] inline std::optional<int64_t> secondsToNanoseconds(double seconds) noexcept {
-  if (!std::isfinite(seconds)) {
-    return std::nullopt;
-  }
-
-  constexpr int64_t kNanosecondsPerSecond = 1'000'000'000;
-  constexpr int64_t kMaximumWholeSeconds = std::numeric_limits<int64_t>::max() / kNanosecondsPerSecond;
-  constexpr int64_t kMinimumWholeSeconds = std::numeric_limits<int64_t>::min() / kNanosecondsPerSecond;
-
-  double whole_seconds = 0.0;
-  const double fractional_seconds = std::modf(seconds, &whole_seconds);
-  if (whole_seconds > static_cast<double>(kMaximumWholeSeconds) ||
-      whole_seconds < static_cast<double>(kMinimumWholeSeconds)) {
-    return std::nullopt;
-  }
-
-  const int64_t whole_nanoseconds = static_cast<int64_t>(whole_seconds) * kNanosecondsPerSecond;
-  const int64_t fractional_nanoseconds =
-      static_cast<int64_t>(std::llround(fractional_seconds * static_cast<double>(kNanosecondsPerSecond)));
-
-  if ((fractional_nanoseconds > 0 &&
-       whole_nanoseconds > std::numeric_limits<int64_t>::max() - fractional_nanoseconds) ||
-      (fractional_nanoseconds < 0 &&
-       whole_nanoseconds < std::numeric_limits<int64_t>::min() - fractional_nanoseconds)) {
-    return std::nullopt;
-  }
-  return whole_nanoseconds + fractional_nanoseconds;
-}
-
 /// Canonical JSON key for the selected timestamp column.
 inline constexpr std::string_view kTimestampColumnKey = "timestamp_column";
 
 /// Canonical JSON key for an integer timestamp column's unit.
 inline constexpr std::string_view kTimestampUnitKey = "timestamp_unit";
 
-/// Units accepted by the shared timestamp-axis configuration contract.
-enum class TimestampUnit : uint8_t {
-  /// Nanoseconds ("ns"); the compatibility default for integer columns.
-  kNanoseconds,
-  /// Microseconds ("us").
-  kMicroseconds,
-  /// Milliseconds ("ms").
-  kMilliseconds,
-  /// Seconds ("s").
-  kSeconds,
-};
+/// Canonical JSON key for a synthesized axis interval in nanoseconds.
+inline constexpr std::string_view kSyntheticIntervalKey = "synthetic_interval_ns";
 
-/// Returns the integral nanosecond scale for a configured timestamp unit.
-[[nodiscard]] constexpr int64_t nanosecondsPer(TimestampUnit unit) noexcept {
-  switch (unit) {
-    case TimestampUnit::kNanoseconds:
-      return 1;
-    case TimestampUnit::kMicroseconds:
-      return 1'000;
-    case TimestampUnit::kMilliseconds:
-      return 1'000'000;
-    case TimestampUnit::kSeconds:
-      return 1'000'000'000;
-  }
-  return 0;
-}
+/// Canonical JSON key controlling whether structured columns are flattened.
+inline constexpr std::string_view kFlattenStructsKey = "flatten_structs";
 
 /// Reads "ns", "us", "ms", or "s" from kTimestampUnitKey. A missing key
 /// preserves the historical nanosecond default; malformed or unknown values
 /// return nullopt so callers can reject the named config field.
-[[nodiscard]] inline std::optional<TimestampUnit> timestampUnitFromJson(const nlohmann::json& object) {
+[[nodiscard]] inline std::optional<PJ::TimeUnit> timestampUnitFromJson(const nlohmann::json& object) {
   const auto unit_it = object.find(kTimestampUnitKey.data());
   if (unit_it == object.end()) {
-    return TimestampUnit::kNanoseconds;
+    return PJ::TimeUnit::kNanoseconds;
   }
   if (!unit_it->is_string()) {
     return std::nullopt;
@@ -260,34 +207,34 @@ enum class TimestampUnit : uint8_t {
 
   const auto& value = unit_it->get_ref<const nlohmann::json::string_t&>();
   if (value == "ns") {
-    return TimestampUnit::kNanoseconds;
+    return PJ::TimeUnit::kNanoseconds;
   }
   if (value == "us") {
-    return TimestampUnit::kMicroseconds;
+    return PJ::TimeUnit::kMicroseconds;
   }
   if (value == "ms") {
-    return TimestampUnit::kMilliseconds;
+    return PJ::TimeUnit::kMilliseconds;
   }
   if (value == "s") {
-    return TimestampUnit::kSeconds;
+    return PJ::TimeUnit::kSeconds;
   }
   return std::nullopt;
 }
 
-/// Writes a TimestampUnit using the canonical short spelling under
+/// Writes a TimeUnit using the canonical short spelling under
 /// kTimestampUnitKey, converting a null JSON value to an object as needed.
-inline void timestampUnitToJson(nlohmann::json& object, TimestampUnit unit) {
+inline void timestampUnitToJson(nlohmann::json& object, PJ::TimeUnit unit) {
   switch (unit) {
-    case TimestampUnit::kNanoseconds:
+    case PJ::TimeUnit::kNanoseconds:
       object[kTimestampUnitKey.data()] = "ns";
       return;
-    case TimestampUnit::kMicroseconds:
+    case PJ::TimeUnit::kMicroseconds:
       object[kTimestampUnitKey.data()] = "us";
       return;
-    case TimestampUnit::kMilliseconds:
+    case PJ::TimeUnit::kMilliseconds:
       object[kTimestampUnitKey.data()] = "ms";
       return;
-    case TimestampUnit::kSeconds:
+    case PJ::TimeUnit::kSeconds:
       object[kTimestampUnitKey.data()] = "s";
       return;
   }
diff --git a/pj_plugins/tests/plugin_sdk_helpers_test.cpp b/pj_plugins/tests/plugin_sdk_helpers_test.cpp
index 3de372c7..679b0c98 100644
--- a/pj_plugins/tests/plugin_sdk_helpers_test.cpp
+++ b/pj_plugins/tests/plugin_sdk_helpers_test.cpp
@@ -350,62 +350,40 @@ TEST(TimestampPolicyTest, SupportAndWarningsCoverEveryTimeKind) {
   }
 }
 
-TEST(TimestampPolicyTest, SecondsToNanosecondsUsesIntegerSplitAndStableRounding) {
-  struct ConversionCase {
-    double seconds;
-    int64_t nanoseconds;
-  };
-  const ConversionCase cases[] = {
-      {1.5, 1'500'000'000}, {1.7e9 + 0.125, 1'700'000'000'125'000'000}, {-1.6e-9, -2}, {1.6e-9, 2}, {2.4e-9, 2},
-  };
-
-  for (const ConversionCase& test_case : cases) {
-    const auto converted = PJ::sdk::secondsToNanoseconds(test_case.seconds);
-    ASSERT_TRUE(converted);
-    EXPECT_EQ(*converted, test_case.nanoseconds);
-  }
-}
-
-TEST(TimestampPolicyTest, SecondsToNanosecondsRejectsNonFiniteAndOverflow) {
-  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(std::numeric_limits<double>::quiet_NaN()));
-  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(std::numeric_limits<double>::infinity()));
-  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(-std::numeric_limits<double>::infinity()));
-  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(9.3e9));
-  EXPECT_FALSE(PJ::sdk::secondsToNanoseconds(9223372036.0 + 0.999999999));
-}
-
 TEST(TimestampPolicyTest, TimestampUnitsReadEverySpellingAndRoundTrip) {
   struct UnitCase {
     const char* spelling;
-    PJ::sdk::TimestampUnit unit;
+    PJ::TimeUnit unit;
     int64_t nanoseconds_per_unit;
   };
   const UnitCase cases[] = {
-      {"ns", PJ::sdk::TimestampUnit::kNanoseconds, 1},
-      {"us", PJ::sdk::TimestampUnit::kMicroseconds, 1'000},
-      {"ms", PJ::sdk::TimestampUnit::kMilliseconds, 1'000'000},
-      {"s", PJ::sdk::TimestampUnit::kSeconds, 1'000'000'000},
+      {"ns", PJ::TimeUnit::kNanoseconds, 1},
+      {"us", PJ::TimeUnit::kMicroseconds, 1'000},
+      {"ms", PJ::TimeUnit::kMilliseconds, 1'000'000},
+      {"s", PJ::TimeUnit::kSeconds, 1'000'000'000},
   };
 
   for (const UnitCase& test_case : cases) {
     const nlohmann::json input = {{"timestamp_unit", test_case.spelling}};
-    EXPECT_EQ(PJ::sdk::timestampUnitFromJson(input), std::optional<PJ::sdk::TimestampUnit>{test_case.unit});
-    EXPECT_EQ(PJ::sdk::nanosecondsPer(test_case.unit), test_case.nanoseconds_per_unit);
+    EXPECT_EQ(PJ::sdk::timestampUnitFromJson(input), std::optional<PJ::TimeUnit>{test_case.unit});
+    EXPECT_EQ(PJ::nanosecondsPer(test_case.unit), test_case.nanoseconds_per_unit);
 
     nlohmann::json output;
     PJ::sdk::timestampUnitToJson(output, test_case.unit);
     EXPECT_EQ(output.at("timestamp_unit"), test_case.spelling);
-    EXPECT_EQ(PJ::sdk::timestampUnitFromJson(output), std::optional<PJ::sdk::TimestampUnit>{test_case.unit});
+    EXPECT_EQ(PJ::sdk::timestampUnitFromJson(output), std::optional<PJ::TimeUnit>{test_case.unit});
   }
 }
 
 TEST(TimestampPolicyTest, MissingUnitDefaultsToNanosecondsAndUnknownIsRejected) {
   EXPECT_EQ(
       PJ::sdk::timestampUnitFromJson(nlohmann::json::object()),
-      std::optional<PJ::sdk::TimestampUnit>{PJ::sdk::TimestampUnit::kNanoseconds});
+      std::optional<PJ::TimeUnit>{PJ::TimeUnit::kNanoseconds});
   EXPECT_FALSE(PJ::sdk::timestampUnitFromJson(nlohmann::json{{"timestamp_unit", "minutes"}}));
   EXPECT_EQ(PJ::sdk::kTimestampColumnKey, "timestamp_column");
   EXPECT_EQ(PJ::sdk::kTimestampUnitKey, "timestamp_unit");
+  EXPECT_EQ(PJ::sdk::kSyntheticIntervalKey, "synthetic_interval_ns");
+  EXPECT_EQ(PJ::sdk::kFlattenStructsKey, "flatten_structs");
 }
 
 }  // namespace

From 955493ba814f5ffe0de8582e3a8c7f3961a984e5 Mon Sep 17 00:00:00 2001
From: GNERSIS <gornersisyan4@gmail.com>
Date: Thu, 3 Sep 2026 08:48:01 +0100
Subject: [PATCH 3/4] feat(sdk): expose the timestamp name pass as
 matchesTimestampName

detectTimestampColumn fuses two questions: does this name look like an axis,
and is this type usable as one. A transport that frames a schema for a
downstream parser needs the first question alone: when nothing plausible
survives framing but the raw schema carried a leaf named like an axis, it
must fail loudly rather than run on a synthetic axis, and it must ask that
without re-implementing the name list privately.

matchesTimestampName(name, policy) returns the priority index of the first
policy name matched, exact-case first and ASCII case-folded when the policy
allows, or nullopt. detectTimestampColumn's name pass is rebuilt on it with
its tie rule unchanged.
---
 CHANGELOG.md                                  |  3 +-
 .../pj_plugins/sdk/timestamp_policy.hpp       | 30 +++++++++++++++++--
 pj_plugins/tests/plugin_sdk_helpers_test.cpp  | 25 ++++++++++++++++
 3 files changed, 54 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4e58871a..7af61232 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -20,7 +20,8 @@ detection and configuration contract: native timestamp storage first, then canon
 restricted to plausible scalar storage (`TIMESTAMP`, `int64`, `uint64`, or `double`), never
 expanded list elements. Explicit narrow-integer, `uint32`, and `float32` axes carry a shared
 warning, while canonical axis configuration keys and `PJ::TimeUnit` stop unit inference from
-being private plugin policy. No ABI change; `abi/baseline.abi` untouched.
+being private plugin policy. `PJ::sdk::matchesTimestampName` exposes the allocation-free name pass
+beside `PJ::sdk::detectTimestampColumn`. No ABI change; `abi/baseline.abi` untouched.
 
 ## [0.26.0]
 
diff --git a/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp b/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
index 067fba61..c06d491b 100644
--- a/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
+++ b/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
@@ -146,6 +146,28 @@ namespace detail {
 
 }  // namespace detail
 
+/// The name pass on its own: the priority index (into `policy.names`) of the first policy name that `name` matches —
+/// exact-case first, then ASCII case-folded when `policy.case_insensitive` — or nullopt. Says nothing about type or
+/// list-ness; pair it with axisSupport() for a full verdict. detectTimestampColumn's name pass is built on this.
+[[nodiscard]] constexpr std::optional<std::size_t> matchesTimestampName(
+    std::string_view name, const TimestampPolicy& policy = kCanonicalPolicy) noexcept {
+  for (std::size_t index = 0; index < policy.names.size(); ++index) {
+    if (name == policy.names[index]) {
+      return index;
+    }
+  }
+
+  if (!policy.case_insensitive) {
+    return std::nullopt;
+  }
+  for (std::size_t index = 0; index < policy.names.size(); ++index) {
+    if (detail::timestampNamesEqualFolded(name, policy.names[index])) {
+      return index;
+    }
+  }
+  return std::nullopt;
+}
+
 /// Selects a timestamp column with a native-type pass followed by a plausible
 /// scalar name pass. Exact-case matches win within each preferred name before
 /// allocation-free ASCII case folding is considered.
@@ -158,11 +180,12 @@ namespace detail {
     }
   }
 
-  for (const std::string_view preferred_name : policy.names) {
+  for (std::size_t name_index = 0; name_index < policy.names.size(); ++name_index) {
+    const TimestampPolicy exact_policy{policy.names.subspan(name_index, 1), false};
     for (std::size_t index = 0; index < candidates.size(); ++index) {
       const TimestampCandidate& candidate = candidates[index];
       if (!candidate.is_list_element && axisSupport(candidate.kind) == AxisSupport::kPlausible &&
-          candidate.name == preferred_name) {
+          matchesTimestampName(candidate.name, exact_policy)) {
         return index;
       }
     }
@@ -170,10 +193,11 @@ namespace detail {
     if (!policy.case_insensitive) {
       continue;
     }
+    const TimestampPolicy folded_policy{policy.names.subspan(name_index, 1), true};
     for (std::size_t index = 0; index < candidates.size(); ++index) {
       const TimestampCandidate& candidate = candidates[index];
       if (!candidate.is_list_element && axisSupport(candidate.kind) == AxisSupport::kPlausible &&
-          detail::timestampNamesEqualFolded(candidate.name, preferred_name)) {
+          matchesTimestampName(candidate.name, folded_policy)) {
         return index;
       }
     }
diff --git a/pj_plugins/tests/plugin_sdk_helpers_test.cpp b/pj_plugins/tests/plugin_sdk_helpers_test.cpp
index 679b0c98..cc626780 100644
--- a/pj_plugins/tests/plugin_sdk_helpers_test.cpp
+++ b/pj_plugins/tests/plugin_sdk_helpers_test.cpp
@@ -326,6 +326,31 @@ TEST(TimestampPolicyTest, CaseSensitiveCustomPolicyRejectsFoldedMatch) {
   EXPECT_FALSE(PJ::sdk::detectTimestampColumn(candidates, policy));
 }
 
+TEST(TimestampPolicyTest, MatchesTimestampNameReturnsCanonicalPriority) {
+  EXPECT_EQ(PJ::sdk::matchesTimestampName("timestamp_ns"), std::optional<std::size_t>{0});
+}
+
+TEST(TimestampPolicyTest, MatchesTimestampNameHonorsCaseSensitivity) {
+  const PJ::sdk::TimestampPolicy case_sensitive_policy{
+      .names = PJ::sdk::kCanonicalTimestampNames,
+      .case_insensitive = false,
+  };
+
+  EXPECT_EQ(PJ::sdk::matchesTimestampName("Timestamp"), std::optional<std::size_t>{2});
+  EXPECT_FALSE(PJ::sdk::matchesTimestampName("Timestamp", case_sensitive_policy));
+}
+
+TEST(TimestampPolicyTest, MatchesTimestampNameRejectsUnrelatedName) {
+  EXPECT_FALSE(PJ::sdk::matchesTimestampName("speed"));
+}
+
+TEST(TimestampPolicyTest, MatchesTimestampNamePrefersExactCaseAcrossPolicyNames) {
+  const std::string_view names[] = {"timestamp", "Timestamp"};
+  const PJ::sdk::TimestampPolicy policy{.names = names, .case_insensitive = true};
+
+  EXPECT_EQ(PJ::sdk::matchesTimestampName("Timestamp", policy), std::optional<std::size_t>{1});
+}
+
 TEST(TimestampPolicyTest, SupportAndWarningsCoverEveryTimeKind) {
   struct SupportCase {
     PJ::sdk::TimeKind kind;

From 8b6dd7ab19919b1fa83b562c62af5ba4e6ab63ce Mon Sep 17 00:00:00 2001
From: Davide Faconti <davide.faconti@gmail.com>
Date: Thu, 3 Sep 2026 21:34:18 +0200
Subject: [PATCH 4/4] feat(sdk): judge timestamp-axis eligibility by storage
 width and configured unit

timestampEligibility(storage, unit) replaces the unit-blind axisSupport(kind):
32-bit integers are eligible for automatic selection when a tick is a second,
8/16-bit integers and float32 are explicit-only at every unit, 64-bit storage
always qualifies. TimestampPolicy carries the unit so detectTimestampColumn
and the warning text agree with the timestamp_unit key instead of assuming
integers are nanoseconds.

Renames while nothing consumes the header yet: TimeKind -> TimestampStorage,
AxisSupport -> TimestampEligibility, matchesTimestampName ->
timestampNamePriority, widenUnsignedTicks -> toSignedTicks.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  11 +-
 pj_base/include/pj_base/time_math.hpp         |   4 +-
 pj_base/tests/time_spine_test.cpp             |   6 +-
 .../pj_plugins/sdk/timestamp_policy.hpp       | 153 +++++++++++-------
 pj_plugins/tests/plugin_sdk_helpers_test.cpp  | 126 ++++++++++-----
 5 files changed, 190 insertions(+), 110 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7af61232..18aa7233 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,7 +8,7 @@ All notable changes to `plotjuggler_sdk` are recorded here. Versioning policy is
 ### Feature: shared timestamp arithmetic and axis policy (MINOR)
 
 `pj_base/time_math.hpp` adds C++17-clean, checked time arithmetic usable by parser modules,
-the host, and plugins: `nanosecondsPer`, `scaleToNanoseconds`, `widenUnsignedTicks`,
+the host, and plugins: `nanosecondsPer`, `scaleToNanoseconds`, `toSignedTicks`,
 `secondsToNanoseconds`, `combineSecondsAndNanos`, `syntheticInstant`, and
 `fitSyntheticInterval`. The absolute-time spine re-exports this arithmetic through
 `pj_base/time.hpp`.
@@ -17,10 +17,11 @@ Layered on that spine, `pj_plugins/sdk/timestamp_policy.hpp` is meant to replace
 divergent per-plugin timestamp-axis detectors inventoried in
 [#186](https://github.com/PlotJuggler/plotjuggler_sdk/issues/186) with one header-only
 detection and configuration contract: native timestamp storage first, then canonical names
-restricted to plausible scalar storage (`TIMESTAMP`, `int64`, `uint64`, or `double`), never
-expanded list elements. Explicit narrow-integer, `uint32`, and `float32` axes carry a shared
-warning, while canonical axis configuration keys and `PJ::TimeUnit` stop unit inference from
-being private plugin policy. `PJ::sdk::matchesTimestampName` exposes the allocation-free name pass
+restricted to eligible scalar storage, never expanded list elements. `PJ::sdk::timestampEligibility`
+judges storage against the configured `timestamp_unit`: 64-bit integers, native timestamps and
+`double` are always eligible, 32-bit integers only when the unit is seconds, and 8/16-bit integers
+and `float32` are explicit-only. Explicitly selected explicit-only storage carries a shared warning, while canonical axis configuration keys and `PJ::TimeUnit` stop unit inference from
+being private plugin policy. `PJ::sdk::timestampNamePriority` exposes the allocation-free name pass
 beside `PJ::sdk::detectTimestampColumn`. No ABI change; `abi/baseline.abi` untouched.
 
 ## [0.26.0]
diff --git a/pj_base/include/pj_base/time_math.hpp b/pj_base/include/pj_base/time_math.hpp
index fb76a260..a5459a6c 100644
--- a/pj_base/include/pj_base/time_math.hpp
+++ b/pj_base/include/pj_base/time_math.hpp
@@ -40,8 +40,8 @@ enum class TimeUnit : uint8_t { kSeconds, kMilliseconds, kMicroseconds, kNanosec
   return ticks * scale;
 }
 
-/// Widens unsigned ticks to the signed timestamp representation, rejecting values above INT64_MAX.
-[[nodiscard]] constexpr std::optional<int64_t> widenUnsignedTicks(uint64_t ticks) noexcept {
+/// Checked uint64 -> int64 tick conversion; values above INT64_MAX return nullopt.
+[[nodiscard]] constexpr std::optional<int64_t> toSignedTicks(uint64_t ticks) noexcept {
   if (ticks > static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
     return std::nullopt;
   }
diff --git a/pj_base/tests/time_spine_test.cpp b/pj_base/tests/time_spine_test.cpp
index 7e23866d..a8aaa02f 100644
--- a/pj_base/tests/time_spine_test.cpp
+++ b/pj_base/tests/time_spine_test.cpp
@@ -47,10 +47,10 @@ TEST(TimeMath, ScalesTicksAndRejectsOverflow) {
   EXPECT_FALSE(PJ::scaleToNanoseconds(9'223'372'037, PJ::TimeUnit::kSeconds));
 }
 
-TEST(TimeMath, WidensUnsignedTicksWithinSignedRange) {
-  EXPECT_FALSE(PJ::widenUnsignedTicks(std::numeric_limits<uint64_t>::max()));
+TEST(TimeMath, ConvertsUnsignedTicksWithinSignedRange) {
+  EXPECT_FALSE(PJ::toSignedTicks(std::numeric_limits<uint64_t>::max()));
   EXPECT_EQ(
-      PJ::widenUnsignedTicks(static_cast<uint64_t>(std::numeric_limits<int64_t>::max())),
+      PJ::toSignedTicks(static_cast<uint64_t>(std::numeric_limits<int64_t>::max())),
       std::optional<int64_t>{std::numeric_limits<int64_t>::max()});
 }
 
diff --git a/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp b/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
index c06d491b..a4b5b26a 100644
--- a/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
+++ b/pj_plugins/include/pj_plugins/sdk/timestamp_policy.hpp
@@ -16,6 +16,7 @@
 #include <array>
 #include <cstddef>
 #include <cstdint>
+#include <limits>
 #include <nlohmann/json.hpp>
 #include <optional>
 #include <span>
@@ -26,72 +27,103 @@
 namespace PJ {
 namespace sdk {
 
-/// How a candidate column's storage relates to an epoch-nanosecond axis.
-enum class TimeKind : uint8_t {
+/// Storage classification of a candidate column. Integer kinds carry ticks in
+/// the policy's TimeUnit; floating kinds carry seconds.
+enum class TimestampStorage : uint8_t {
   /// A native timestamp whose unit is known to the caller.
   kNativeTimestamp,
-  /// A signed 64-bit integer containing nanoseconds.
   kInt64,
-  /// An unsigned 64-bit integer containing nanoseconds.
   kUInt64,
   /// Double-precision seconds, with about 238 ns resolution at the present epoch.
   kFloat64,
-  /// Unsigned 32-bit nanoseconds, which end about 4.3 seconds after the epoch.
+  kInt32,
   kUInt32,
-  /// Signed 8/16/32-bit or unsigned 8/16-bit nanoseconds.
+  /// Signed or unsigned 8/16-bit integers.
   kNarrowInt,
-  /// Single-precision seconds, whose spacing reaches one second at 2^23 seconds.
+  /// Single-precision seconds: 24 significant bits, so spacing exceeds 100 s at the present epoch.
   kFloat32,
   /// Storage that cannot serve as a timestamp axis.
   kOther,
 };
 
-/// Whether a TimeKind can be auto-selected or must be handled explicitly.
-enum class AxisSupport : uint8_t {
-  /// Safe enough for automatic timestamp-axis selection.
-  kPlausible,
-  /// Available only after surfacing explicitAxisWarning().
-  kAcceptedWithWarning,
-  /// Cannot serve as a timestamp axis.
-  kUnsupported,
+/// When a TimestampStorage may become the time axis.
+enum class TimestampEligibility : uint8_t {
+  /// May be picked automatically by detectTimestampColumn().
+  kEligible,
+  /// Only when named explicitly, after surfacing explicitOnlyWarning().
+  kExplicitOnly,
+  /// Never.
+  kIneligible,
 };
 
-/// Classifies timestamp-axis support without inspecting a column name.
-[[nodiscard]] constexpr AxisSupport axisSupport(TimeKind kind) noexcept {
+/// An integer column is eligible for automatic selection when its width can hold instants at
+/// least this far past the Unix epoch at the configured unit: int32 seconds (2038-01-19), the
+/// narrowest storage in common use for absolute time.
+inline constexpr int64_t kEligibleHorizonSeconds = std::numeric_limits<int32_t>::max();
+
+namespace detail {
+
+/// Largest tick count the integer kind can hold; nullopt for non-integer kinds.
+[[nodiscard]] constexpr std::optional<uint64_t> maxIntegerTicks(TimestampStorage kind) noexcept {
   switch (kind) {
-    case TimeKind::kNativeTimestamp:
-    case TimeKind::kInt64:
-    case TimeKind::kUInt64:
-    case TimeKind::kFloat64:
-      return AxisSupport::kPlausible;
-    case TimeKind::kUInt32:
-    case TimeKind::kNarrowInt:
-    case TimeKind::kFloat32:
-      return AxisSupport::kAcceptedWithWarning;
-    case TimeKind::kOther:
-      return AxisSupport::kUnsupported;
+    case TimestampStorage::kInt64:
+      return static_cast<uint64_t>(std::numeric_limits<int64_t>::max());
+    case TimestampStorage::kUInt64:
+      return std::numeric_limits<uint64_t>::max();
+    case TimestampStorage::kInt32:
+      return static_cast<uint64_t>(std::numeric_limits<int32_t>::max());
+    case TimestampStorage::kUInt32:
+      return std::numeric_limits<uint32_t>::max();
+    case TimestampStorage::kNarrowInt:
+      return static_cast<uint64_t>(std::numeric_limits<int8_t>::max());
+    case TimestampStorage::kNativeTimestamp:
+    case TimestampStorage::kFloat64:
+    case TimestampStorage::kFloat32:
+    case TimestampStorage::kOther:
+      return std::nullopt;
   }
-  return AxisSupport::kUnsupported;
+  return std::nullopt;
 }
 
-/// Returns the warning a plugin must surface for an explicitly selected lossy
-/// or short-range axis; all other kinds return an empty view without allocating.
-[[nodiscard]] constexpr std::string_view explicitAxisWarning(TimeKind kind) noexcept {
+}  // namespace detail
+
+/// Whether storage holding ticks of `unit` may become the time axis, without inspecting a column
+/// name: kEligible for automatic selection, kExplicitOnly when a producer must name it and surface
+/// explicitOnlyWarning(), kIneligible never. Integers are eligible when they reach
+/// kEligibleHorizonSeconds at that unit; float32 is always explicit-only because its precision,
+/// not its range, is the problem.
+[[nodiscard]] constexpr TimestampEligibility timestampEligibility(TimestampStorage kind, PJ::TimeUnit unit) noexcept {
   switch (kind) {
-    case TimeKind::kUInt32:
-      return "uint32 can express at most 4294967295 ns since the Unix epoch.";
-    case TimeKind::kNarrowInt:
-      return "Narrow integers can express at most 2147483647 ns since the Unix epoch.";
-    case TimeKind::kFloat32:
-      return "float32 seconds reach 1-second spacing at 2^23 (8388608) seconds from the Unix epoch.";
-    case TimeKind::kNativeTimestamp:
-    case TimeKind::kInt64:
-    case TimeKind::kUInt64:
-    case TimeKind::kFloat64:
-    case TimeKind::kOther:
-      return {};
+    case TimestampStorage::kNativeTimestamp:
+    case TimestampStorage::kFloat64:
+      return TimestampEligibility::kEligible;
+    case TimestampStorage::kFloat32:
+      return TimestampEligibility::kExplicitOnly;
+    case TimestampStorage::kOther:
+      return TimestampEligibility::kIneligible;
+    case TimestampStorage::kInt64:
+    case TimestampStorage::kUInt64:
+    case TimestampStorage::kInt32:
+    case TimestampStorage::kUInt32:
+    case TimestampStorage::kNarrowInt:
+      break;
+  }
+  const uint64_t ticks_per_second = static_cast<uint64_t>(1'000'000'000 / PJ::nanosecondsPer(unit));
+  const uint64_t horizon_ticks = static_cast<uint64_t>(kEligibleHorizonSeconds) * ticks_per_second;
+  return *detail::maxIntegerTicks(kind) >= horizon_ticks ? TimestampEligibility::kEligible
+                                                         : TimestampEligibility::kExplicitOnly;
+}
+
+/// The warning a plugin must surface when a kExplicitOnly column is selected by name;
+/// every other eligibility returns an empty view.
+[[nodiscard]] constexpr std::string_view explicitOnlyWarning(TimestampStorage kind, PJ::TimeUnit unit) noexcept {
+  if (timestampEligibility(kind, unit) != TimestampEligibility::kExplicitOnly) {
+    return {};
+  }
+  if (kind == TimestampStorage::kFloat32) {
+    return "float32 keeps 24 significant bits, so instants near the present epoch are spaced over 100 s apart.";
   }
-  return {};
+  return "Integer storage too narrow to reach present-day instants at the configured timestamp unit.";
 }
 
 /// Arrow-independent description of a flattened column considered for the axis.
@@ -99,7 +131,7 @@ struct TimestampCandidate {
   /// Flattened leaf path; separators are '/', with source dots already normalized.
   std::string_view name;
   /// Storage classification supplied by the importing plugin.
-  TimeKind kind;
+  TimestampStorage kind;
   /// Expanded list elements are never eligible for automatic selection.
   bool is_list_element = false;
 };
@@ -110,6 +142,8 @@ struct TimestampPolicy {
   std::span<const std::string_view> names;
   /// Whether the name pass also accepts ASCII case-folded matches.
   bool case_insensitive = true;
+  /// Unit of integer candidates (the configured kTimestampUnitKey); decides which widths are eligible.
+  PJ::TimeUnit unit = PJ::TimeUnit::kNanoseconds;
 };
 
 /// Union of timestamp names used by official plugins, most specific first.
@@ -121,7 +155,7 @@ inline constexpr std::array<std::string_view, 11> kCanonicalTimestampNames = {"t
                                                                               "_time"};
 
 /// Default policy shared by official plugins.
-inline constexpr TimestampPolicy kCanonicalPolicy{kCanonicalTimestampNames, true};
+inline constexpr TimestampPolicy kCanonicalPolicy{kCanonicalTimestampNames, true, PJ::TimeUnit::kNanoseconds};
 
 namespace detail {
 
@@ -148,8 +182,9 @@ namespace detail {
 
 /// The name pass on its own: the priority index (into `policy.names`) of the first policy name that `name` matches —
 /// exact-case first, then ASCII case-folded when `policy.case_insensitive` — or nullopt. Says nothing about type or
-/// list-ness; pair it with axisSupport() for a full verdict. detectTimestampColumn's name pass is built on this.
-[[nodiscard]] constexpr std::optional<std::size_t> matchesTimestampName(
+/// list-ness; pair it with timestampEligibility() for a full verdict. detectTimestampColumn's name pass is built on
+/// this.
+[[nodiscard]] constexpr std::optional<std::size_t> timestampNamePriority(
     std::string_view name, const TimestampPolicy& policy = kCanonicalPolicy) noexcept {
   for (std::size_t index = 0; index < policy.names.size(); ++index) {
     if (name == policy.names[index]) {
@@ -168,24 +203,25 @@ namespace detail {
   return std::nullopt;
 }
 
-/// Selects a timestamp column with a native-type pass followed by a plausible
-/// scalar name pass. Exact-case matches win within each preferred name before
+/// Selects a timestamp column with a native-type pass followed by a name pass
+/// over kEligible scalars. Exact-case matches win within each preferred name before
 /// allocation-free ASCII case folding is considered.
 [[nodiscard]] constexpr std::optional<std::size_t> detectTimestampColumn(
     std::span<const TimestampCandidate> candidates, const TimestampPolicy& policy = kCanonicalPolicy) {
   for (std::size_t index = 0; index < candidates.size(); ++index) {
     const TimestampCandidate& candidate = candidates[index];
-    if (!candidate.is_list_element && candidate.kind == TimeKind::kNativeTimestamp) {
+    if (!candidate.is_list_element && candidate.kind == TimestampStorage::kNativeTimestamp) {
       return index;
     }
   }
 
   for (std::size_t name_index = 0; name_index < policy.names.size(); ++name_index) {
-    const TimestampPolicy exact_policy{policy.names.subspan(name_index, 1), false};
+    const TimestampPolicy exact_policy{policy.names.subspan(name_index, 1), false, policy.unit};
     for (std::size_t index = 0; index < candidates.size(); ++index) {
       const TimestampCandidate& candidate = candidates[index];
-      if (!candidate.is_list_element && axisSupport(candidate.kind) == AxisSupport::kPlausible &&
-          matchesTimestampName(candidate.name, exact_policy)) {
+      if (!candidate.is_list_element &&
+          timestampEligibility(candidate.kind, policy.unit) == TimestampEligibility::kEligible &&
+          timestampNamePriority(candidate.name, exact_policy)) {
         return index;
       }
     }
@@ -193,11 +229,12 @@ namespace detail {
     if (!policy.case_insensitive) {
       continue;
     }
-    const TimestampPolicy folded_policy{policy.names.subspan(name_index, 1), true};
+    const TimestampPolicy folded_policy{policy.names.subspan(name_index, 1), true, policy.unit};
     for (std::size_t index = 0; index < candidates.size(); ++index) {
       const TimestampCandidate& candidate = candidates[index];
-      if (!candidate.is_list_element && axisSupport(candidate.kind) == AxisSupport::kPlausible &&
-          matchesTimestampName(candidate.name, folded_policy)) {
+      if (!candidate.is_list_element &&
+          timestampEligibility(candidate.kind, policy.unit) == TimestampEligibility::kEligible &&
+          timestampNamePriority(candidate.name, folded_policy)) {
         return index;
       }
     }
diff --git a/pj_plugins/tests/plugin_sdk_helpers_test.cpp b/pj_plugins/tests/plugin_sdk_helpers_test.cpp
index cc626780..12d80100 100644
--- a/pj_plugins/tests/plugin_sdk_helpers_test.cpp
+++ b/pj_plugins/tests/plugin_sdk_helpers_test.cpp
@@ -252,8 +252,8 @@ TEST(DelegatedIngestTest, BindingFailureIsANonFatalDisposition) {
 
 TEST(TimestampPolicyTest, NativeTimestampTypePassWinsOverPreferredName) {
   const PJ::sdk::TimestampCandidate candidates[] = {
-      {.name = "timestamp_ns", .kind = PJ::sdk::TimeKind::kInt64},
-      {.name = "foo", .kind = PJ::sdk::TimeKind::kNativeTimestamp},
+      {.name = "timestamp_ns", .kind = PJ::sdk::TimestampStorage::kInt64},
+      {.name = "foo", .kind = PJ::sdk::TimestampStorage::kNativeTimestamp},
   };
 
   EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
@@ -261,17 +261,17 @@ TEST(TimestampPolicyTest, NativeTimestampTypePassWinsOverPreferredName) {
 
 TEST(TimestampPolicyTest, NamePriorityWinsOverCandidateOrder) {
   const PJ::sdk::TimestampCandidate candidates[] = {
-      {.name = "time", .kind = PJ::sdk::TimeKind::kInt64},
-      {.name = "recording_timestamp_ns", .kind = PJ::sdk::TimeKind::kInt64},
+      {.name = "time", .kind = PJ::sdk::TimestampStorage::kInt64},
+      {.name = "recording_timestamp_ns", .kind = PJ::sdk::TimestampStorage::kInt64},
   };
 
   EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
 }
 
-TEST(TimestampPolicyTest, NarrowIntegerNameIsSkippedForPlausibleInt64) {
+TEST(TimestampPolicyTest, NarrowIntegerNameIsSkippedForEligibleInt64) {
   const PJ::sdk::TimestampCandidate candidates[] = {
-      {.name = "timestamp_ns", .kind = PJ::sdk::TimeKind::kNarrowInt},
-      {.name = "time", .kind = PJ::sdk::TimeKind::kInt64},
+      {.name = "timestamp_ns", .kind = PJ::sdk::TimestampStorage::kNarrowInt},
+      {.name = "time", .kind = PJ::sdk::TimestampStorage::kInt64},
   };
 
   EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
@@ -279,9 +279,9 @@ TEST(TimestampPolicyTest, NarrowIntegerNameIsSkippedForPlausibleInt64) {
 
 TEST(TimestampPolicyTest, UInt32AndFloat32NamesAreSkippedForFloat64) {
   const PJ::sdk::TimestampCandidate candidates[] = {
-      {.name = "timestamp_ns", .kind = PJ::sdk::TimeKind::kUInt32},
-      {.name = "recording_timestamp_ns", .kind = PJ::sdk::TimeKind::kFloat32},
-      {.name = "time", .kind = PJ::sdk::TimeKind::kFloat64},
+      {.name = "timestamp_ns", .kind = PJ::sdk::TimestampStorage::kUInt32},
+      {.name = "recording_timestamp_ns", .kind = PJ::sdk::TimestampStorage::kFloat32},
+      {.name = "time", .kind = PJ::sdk::TimestampStorage::kFloat64},
   };
 
   EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{2});
@@ -289,7 +289,7 @@ TEST(TimestampPolicyTest, UInt32AndFloat32NamesAreSkippedForFloat64) {
 
 TEST(TimestampPolicyTest, ListElementIsNeverSelected) {
   const PJ::sdk::TimestampCandidate candidates[] = {
-      {.name = "timestamp", .kind = PJ::sdk::TimeKind::kNativeTimestamp, .is_list_element = true},
+      {.name = "timestamp", .kind = PJ::sdk::TimestampStorage::kNativeTimestamp, .is_list_element = true},
   };
 
   EXPECT_FALSE(PJ::sdk::detectTimestampColumn(candidates));
@@ -297,10 +297,10 @@ TEST(TimestampPolicyTest, ListElementIsNeverSelected) {
 
 TEST(TimestampPolicyTest, CanonicalNamesMatchAsciiCaseInsensitively) {
   const PJ::sdk::TimestampCandidate title_case[] = {
-      {.name = "Timestamp", .kind = PJ::sdk::TimeKind::kInt64},
+      {.name = "Timestamp", .kind = PJ::sdk::TimestampStorage::kInt64},
   };
   const PJ::sdk::TimestampCandidate upper_case[] = {
-      {.name = "DATETIME", .kind = PJ::sdk::TimeKind::kFloat64},
+      {.name = "DATETIME", .kind = PJ::sdk::TimestampStorage::kFloat64},
   };
 
   EXPECT_EQ(PJ::sdk::detectTimestampColumn(title_case), std::optional<std::size_t>{0});
@@ -309,8 +309,8 @@ TEST(TimestampPolicyTest, CanonicalNamesMatchAsciiCaseInsensitively) {
 
 TEST(TimestampPolicyTest, ExactCaseWinsWithinPreferredNameRegardlessOfOrder) {
   const PJ::sdk::TimestampCandidate candidates[] = {
-      {.name = "Timestamp", .kind = PJ::sdk::TimeKind::kInt64},
-      {.name = "timestamp", .kind = PJ::sdk::TimeKind::kInt64},
+      {.name = "Timestamp", .kind = PJ::sdk::TimestampStorage::kInt64},
+      {.name = "timestamp", .kind = PJ::sdk::TimestampStorage::kInt64},
   };
 
   EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
@@ -320,61 +320,103 @@ TEST(TimestampPolicyTest, CaseSensitiveCustomPolicyRejectsFoldedMatch) {
   const std::string_view names[] = {"timestamp"};
   const PJ::sdk::TimestampPolicy policy{.names = names, .case_insensitive = false};
   const PJ::sdk::TimestampCandidate candidates[] = {
-      {.name = "Timestamp", .kind = PJ::sdk::TimeKind::kInt64},
+      {.name = "Timestamp", .kind = PJ::sdk::TimestampStorage::kInt64},
   };
 
   EXPECT_FALSE(PJ::sdk::detectTimestampColumn(candidates, policy));
 }
 
-TEST(TimestampPolicyTest, MatchesTimestampNameReturnsCanonicalPriority) {
-  EXPECT_EQ(PJ::sdk::matchesTimestampName("timestamp_ns"), std::optional<std::size_t>{0});
+TEST(TimestampPolicyTest, TimestampNamePriorityReturnsCanonicalPriority) {
+  EXPECT_EQ(PJ::sdk::timestampNamePriority("timestamp_ns"), std::optional<std::size_t>{0});
 }
 
-TEST(TimestampPolicyTest, MatchesTimestampNameHonorsCaseSensitivity) {
+TEST(TimestampPolicyTest, TimestampNamePriorityHonorsCaseSensitivity) {
   const PJ::sdk::TimestampPolicy case_sensitive_policy{
       .names = PJ::sdk::kCanonicalTimestampNames,
       .case_insensitive = false,
   };
 
-  EXPECT_EQ(PJ::sdk::matchesTimestampName("Timestamp"), std::optional<std::size_t>{2});
-  EXPECT_FALSE(PJ::sdk::matchesTimestampName("Timestamp", case_sensitive_policy));
+  EXPECT_EQ(PJ::sdk::timestampNamePriority("Timestamp"), std::optional<std::size_t>{2});
+  EXPECT_FALSE(PJ::sdk::timestampNamePriority("Timestamp", case_sensitive_policy));
 }
 
-TEST(TimestampPolicyTest, MatchesTimestampNameRejectsUnrelatedName) {
-  EXPECT_FALSE(PJ::sdk::matchesTimestampName("speed"));
+TEST(TimestampPolicyTest, TimestampNamePriorityRejectsUnrelatedName) {
+  EXPECT_FALSE(PJ::sdk::timestampNamePriority("speed"));
 }
 
-TEST(TimestampPolicyTest, MatchesTimestampNamePrefersExactCaseAcrossPolicyNames) {
+TEST(TimestampPolicyTest, TimestampNamePriorityPrefersExactCaseAcrossPolicyNames) {
   const std::string_view names[] = {"timestamp", "Timestamp"};
   const PJ::sdk::TimestampPolicy policy{.names = names, .case_insensitive = true};
 
-  EXPECT_EQ(PJ::sdk::matchesTimestampName("Timestamp", policy), std::optional<std::size_t>{1});
+  EXPECT_EQ(PJ::sdk::timestampNamePriority("Timestamp", policy), std::optional<std::size_t>{1});
 }
 
-TEST(TimestampPolicyTest, SupportAndWarningsCoverEveryTimeKind) {
-  struct SupportCase {
-    PJ::sdk::TimeKind kind;
-    PJ::sdk::AxisSupport support;
+TEST(TimestampPolicyTest, EligibilityAndWarningsCoverEveryTimeKindAtNanoseconds) {
+  struct EligibilityCase {
+    PJ::sdk::TimestampStorage kind;
+    PJ::sdk::TimestampEligibility eligibility;
   };
-  constexpr std::array<SupportCase, 8> cases = {{
-      {PJ::sdk::TimeKind::kNativeTimestamp, PJ::sdk::AxisSupport::kPlausible},
-      {PJ::sdk::TimeKind::kInt64, PJ::sdk::AxisSupport::kPlausible},
-      {PJ::sdk::TimeKind::kUInt64, PJ::sdk::AxisSupport::kPlausible},
-      {PJ::sdk::TimeKind::kFloat64, PJ::sdk::AxisSupport::kPlausible},
-      {PJ::sdk::TimeKind::kUInt32, PJ::sdk::AxisSupport::kAcceptedWithWarning},
-      {PJ::sdk::TimeKind::kNarrowInt, PJ::sdk::AxisSupport::kAcceptedWithWarning},
-      {PJ::sdk::TimeKind::kFloat32, PJ::sdk::AxisSupport::kAcceptedWithWarning},
-      {PJ::sdk::TimeKind::kOther, PJ::sdk::AxisSupport::kUnsupported},
+  constexpr std::array<EligibilityCase, 9> cases = {{
+      {PJ::sdk::TimestampStorage::kNativeTimestamp, PJ::sdk::TimestampEligibility::kEligible},
+      {PJ::sdk::TimestampStorage::kInt64, PJ::sdk::TimestampEligibility::kEligible},
+      {PJ::sdk::TimestampStorage::kUInt64, PJ::sdk::TimestampEligibility::kEligible},
+      {PJ::sdk::TimestampStorage::kFloat64, PJ::sdk::TimestampEligibility::kEligible},
+      {PJ::sdk::TimestampStorage::kInt32, PJ::sdk::TimestampEligibility::kExplicitOnly},
+      {PJ::sdk::TimestampStorage::kUInt32, PJ::sdk::TimestampEligibility::kExplicitOnly},
+      {PJ::sdk::TimestampStorage::kNarrowInt, PJ::sdk::TimestampEligibility::kExplicitOnly},
+      {PJ::sdk::TimestampStorage::kFloat32, PJ::sdk::TimestampEligibility::kExplicitOnly},
+      {PJ::sdk::TimestampStorage::kOther, PJ::sdk::TimestampEligibility::kIneligible},
   }};
 
-  for (const SupportCase& test_case : cases) {
-    EXPECT_EQ(PJ::sdk::axisSupport(test_case.kind), test_case.support);
+  for (const EligibilityCase& test_case : cases) {
+    EXPECT_EQ(PJ::sdk::timestampEligibility(test_case.kind, PJ::TimeUnit::kNanoseconds), test_case.eligibility);
     EXPECT_EQ(
-        PJ::sdk::explicitAxisWarning(test_case.kind).empty(),
-        test_case.support != PJ::sdk::AxisSupport::kAcceptedWithWarning);
+        PJ::sdk::explicitOnlyWarning(test_case.kind, PJ::TimeUnit::kNanoseconds).empty(),
+        test_case.eligibility != PJ::sdk::TimestampEligibility::kExplicitOnly);
   }
 }
 
+TEST(TimestampPolicyTest, IntegerEligibilityFollowsTheConfiguredUnit) {
+  using PJ::TimeUnit;
+  using PJ::sdk::TimestampEligibility;
+  using PJ::sdk::TimestampStorage;
+
+  // Unix seconds in 32 bits reach 2038 (int32) or 2106 (uint32): eligible.
+  EXPECT_EQ(
+      PJ::sdk::timestampEligibility(TimestampStorage::kInt32, TimeUnit::kSeconds), TimestampEligibility::kEligible);
+  EXPECT_EQ(
+      PJ::sdk::timestampEligibility(TimestampStorage::kUInt32, TimeUnit::kSeconds), TimestampEligibility::kEligible);
+  EXPECT_TRUE(PJ::sdk::explicitOnlyWarning(TimestampStorage::kUInt32, TimeUnit::kSeconds).empty());
+  // The same widths in milliseconds end within weeks of the epoch.
+  EXPECT_EQ(
+      PJ::sdk::timestampEligibility(TimestampStorage::kUInt32, TimeUnit::kMilliseconds),
+      TimestampEligibility::kExplicitOnly);
+  EXPECT_EQ(
+      PJ::sdk::timestampEligibility(TimestampStorage::kInt32, TimeUnit::kMilliseconds),
+      TimestampEligibility::kExplicitOnly);
+  // 8/16-bit storage and float32 never become eligible; 64-bit always is.
+  EXPECT_EQ(
+      PJ::sdk::timestampEligibility(TimestampStorage::kNarrowInt, TimeUnit::kSeconds),
+      TimestampEligibility::kExplicitOnly);
+  EXPECT_EQ(
+      PJ::sdk::timestampEligibility(TimestampStorage::kFloat32, TimeUnit::kSeconds),
+      TimestampEligibility::kExplicitOnly);
+  EXPECT_EQ(
+      PJ::sdk::timestampEligibility(TimestampStorage::kInt64, TimeUnit::kSeconds), TimestampEligibility::kEligible);
+}
+
+TEST(TimestampPolicyTest, DetectorAutoSelectsInt32SecondsOnlyWhenPolicyUnitIsSeconds) {
+  const PJ::sdk::TimestampCandidate candidates[] = {
+      {.name = "timestamp", .kind = PJ::sdk::TimestampStorage::kInt32},
+      {.name = "time", .kind = PJ::sdk::TimestampStorage::kInt64},
+  };
+  const PJ::sdk::TimestampPolicy seconds_policy{
+      .names = PJ::sdk::kCanonicalTimestampNames, .case_insensitive = true, .unit = PJ::TimeUnit::kSeconds};
+
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates), std::optional<std::size_t>{1});
+  EXPECT_EQ(PJ::sdk::detectTimestampColumn(candidates, seconds_policy), std::optional<std::size_t>{0});
+}
+
 TEST(TimestampPolicyTest, TimestampUnitsReadEverySpellingAndRoundTrip) {
   struct UnitCase {
     const char* spelling;