diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 79014cc..1ae578f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -21,5 +21,6 @@ jobs:
           set -euo pipefail
           bash tests/test_install_2fa_bot_watcher.sh
           bash tests/test_wait_for_ib_gateway_ready.sh
+          python3 tests/test_gateway_epoch_value.py
           bash tests/test_workflow_shared_config.sh
           bash tests/test_docker_compose_ports.sh
diff --git a/scripts/gateway_epoch_value.py b/scripts/gateway_epoch_value.py
new file mode 100644
index 0000000..9733fc6
--- /dev/null
+++ b/scripts/gateway_epoch_value.py
@@ -0,0 +1,37 @@
+"""Validated canonical epoch values; intentionally independent from recovery logic."""
+from __future__ import annotations
+import calendar,re
+from dataclasses import dataclass
+from datetime import datetime
+class EpochValueError(ValueError):
+ def __init__(self): super().__init__('invalid epoch value')
+STAMP=re.compile(r'^(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)(?:\.(\d{1,9}))?(Z|\+00:00)$')
+MAX_EPOCH_NS=253402300799999999999
+CONTAINER_ID=re.compile(r'^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$')
+def parse_epoch_ns(value: object)->int:
+ try:
+  if not isinstance(value,str): raise EpochValueError()
+  m=STAMP.fullmatch(value)
+  if not m: raise EpochValueError()
+  year,month,day,hour,minute,second=map(int,m.groups()[:6])
+  if year<1970: raise EpochValueError()
+  datetime(year,month,day,hour,minute,second)
+  return _valid_ns(calendar.timegm((year,month,day,hour,minute,second))*1_000_000_000+int(((m.group(7)or'')+'0'*9)[:9]))
+ except (ValueError,OverflowError,TypeError,EpochValueError): raise EpochValueError() from None
+def _valid_ns(value: object)->int:
+ if type(value) is not int or value<0 or value>MAX_EPOCH_NS: raise EpochValueError()
+ return value
+def _valid_id(value: object)->str:
+ if not isinstance(value,str) or not CONTAINER_ID.fullmatch(value): raise EpochValueError()
+ return value
+@dataclass(frozen=True,order=True)
+class EpochIdentity:
+ container_id:str
+ started_at_epoch_ns:int
+ def __post_init__(self): object.__setattr__(self,'container_id',_valid_id(self.container_id)); object.__setattr__(self,'started_at_epoch_ns',_valid_ns(self.started_at_epoch_ns))
+ def serialize(self)->tuple[str,int]: return (self.container_id,self.started_at_epoch_ns)
+@dataclass(frozen=True,order=True)
+class EpochCursor:
+ lower_bound_epoch_ns:int
+ def __post_init__(self): object.__setattr__(self,'lower_bound_epoch_ns',_valid_ns(self.lower_bound_epoch_ns))
+ def serialize(self)->int:return self.lower_bound_epoch_ns
diff --git a/tests/test_gateway_epoch_value.py b/tests/test_gateway_epoch_value.py
new file mode 100644
index 0000000..193730d
--- /dev/null
+++ b/tests/test_gateway_epoch_value.py
@@ -0,0 +1,22 @@
+from __future__ import annotations
+import unittest,sys
+from dataclasses import replace
+from pathlib import Path
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
+from scripts.gateway_epoch_value import EpochCursor,EpochIdentity,EpochValueError,parse_epoch_ns
+class T(unittest.TestCase):
+ def test_equivalence_order_and_separation(self):
+  a=parse_epoch_ns('2026-07-15T16:00:00.1Z'); b=parse_epoch_ns('2026-07-15T16:00:00.100000000+00:00'); self.assertEqual(a,b); self.assertLess(a,parse_epoch_ns('2026-07-15T16:00:00.100000001Z'))
+  identity=EpochIdentity('c',a); self.assertEqual(identity,replace(identity)); self.assertNotEqual(identity,EpochIdentity('c',a+1)); self.assertNotEqual(EpochCursor(a),EpochCursor(a+1))
+ def test_invalid_is_sanitized(self):
+  for value in ('','2026-02-30T00:00:00Z','2026-01-01T00:00:00+01:00','2026-01-01T00:00:00.1234567890Z',None,1):
+   with self.subTest(value=value):
+    with self.assertRaisesRegex(EpochValueError,'invalid epoch value'): parse_epoch_ns(value)
+ def test_constructors_revalidate(self):
+  with self.assertRaises(EpochValueError): EpochIdentity('',0)
+  with self.assertRaises(EpochValueError): EpochCursor(-1)
+  for bad in ('a/b','a\\b','a\n', ' ' * 129):
+   with self.subTest(bad=bad):
+    with self.assertRaises(EpochValueError): EpochIdentity(bad,0)
+  with self.assertRaises(EpochValueError): EpochCursor(253402300800000000000)
+if __name__=='__main__':unittest.main()