fix(events): normalize millisecond timestamps before datetime conversion (LANG-170) (#2722)

* fix(events): normalize millisecond timestamps before datetime conversion

- run_journal: _event_time_to_utc normalizes second/millisecond/
  microsecond epoch values (loop-divide above 1e10) with guards for
  bool/non-numeric/zero/non-finite inputs; OverflowError contained
- run_ledger_store, interaction_store: divide epoch values above 1e11
  by 1000 before datetime conversion; OverflowError added to the
  handled exceptions

Second-based inputs are unchanged. Verified in production on a
self-hosted v4.11.0 deployment.

* fix(events): harden timestamp conversion and cover persistence regressions

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: leonoxo <[email protected]>
Co-authored-by: Chin Rock <[email protected]>
Co-authored-by: multica-agent <[email protected]>

src/langbot/pkg/agent/runner/interaction_store.py

@@ -69,8 +69,11 @@ def _epoch_to_datetime(value: int | float | None) -> datetime.datetime | None:
     if value is None:
         return None
     try:
-        return datetime.datetime.fromtimestamp(float(value), UTC)
-    except (TypeError, ValueError, OSError) as exc:
+        ts = float(value)
+        if ts > 1e11:
+            ts /= 1000.0
+        return datetime.datetime.fromtimestamp(ts, UTC)
+    except (TypeError, ValueError, OSError, OverflowError) as exc:
         raise ValueError('expires_at must be a valid epoch timestamp') from exc
 
 

src/langbot/pkg/agent/runner/run_journal.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import datetime
+import math
 import typing
 
 from ...core import app
@@ -39,6 +41,21 @@ class AgentRunJournal:
             return dict(value)
         return {}
 
+    @staticmethod
+    def _event_time_to_utc(value: typing.Any) -> datetime.datetime | None:
+        """Normalize legacy second/millisecond/microsecond Unix timestamps to UTC datetime."""
+        if isinstance(value, bool) or not isinstance(value, (int, float)):
+            return None
+        try:
+            timestamp = float(value)
+            if timestamp == 0 or not math.isfinite(timestamp):
+                return None
+            while abs(timestamp) > 10_000_000_000:
+                timestamp /= 1000
+            return datetime.datetime.fromtimestamp(timestamp, datetime.timezone.utc)
+        except (TypeError, ValueError, OSError, OverflowError):
+            return None
+
     @classmethod
     def _sanitize_content_item(cls, value: typing.Any) -> typing.Any:
         item = cls._to_plain_dict(value)
@@ -258,8 +275,6 @@ class AgentRunJournal:
         metadata: dict[str, typing.Any] | None = None,
     ) -> str:
         """Write incoming event to EventLog."""
-        import datetime
-
         from .event_log_store import EventLogStore
 
         store = EventLogStore(self.ap.persistence_mgr.get_db_engine())
@@ -292,9 +307,7 @@ class AgentRunJournal:
             input_json=input_json,
             run_id=run_id,
             runner_id=runner_id,
-            event_time=(
-                datetime.datetime.fromtimestamp(event.event_time, datetime.timezone.utc) if event.event_time else None
-            ),
+            event_time=self._event_time_to_utc(event.event_time),
             metadata={**(metadata or {}), 'input_event': self._sanitize_event_data(event.data)},
         )
 
@@ -351,7 +364,6 @@ class AgentRunJournal:
         if not items:
             return
 
-        import datetime
         import uuid
 
         from .event_log_store import EventLogStore
@@ -364,16 +376,7 @@ class AgentRunJournal:
             actor = item.get('actor') if isinstance(item.get('actor'), dict) else {}
             subject = item.get('subject') if isinstance(item.get('subject'), dict) else {}
 
-            event_time = None
-            raw_event_time = event.get('event_time')
-            if raw_event_time:
-                try:
-                    event_time = datetime.datetime.fromtimestamp(
-                        raw_event_time,
-                        datetime.timezone.utc,
-                    )
-                except (TypeError, ValueError, OSError):
-                    event_time = None
+            event_time = self._event_time_to_utc(event.get('event_time'))
 
             await store.append_event(
                 event_id=str(uuid.uuid4()),

src/langbot/pkg/agent/runner/run_ledger_store.py

@@ -50,8 +50,11 @@ def _epoch_to_datetime(value: typing.Any) -> datetime.datetime | None:
     if value is None:
         return None
     try:
-        return as_naive_utc(datetime.datetime.fromtimestamp(float(value), UTC))
-    except (TypeError, ValueError, OSError):
+        ts = float(value)
+        if ts > 1e11:
+            ts /= 1000.0
+        return as_naive_utc(datetime.datetime.fromtimestamp(ts, UTC))
+    except (TypeError, ValueError, OSError, OverflowError):
         return None
 
 

tests/unit_tests/agent/test_interaction_store.py

@@ -10,6 +10,7 @@ import sqlalchemy
 from sqlalchemy.ext.asyncio import create_async_engine
 
 from langbot.pkg.agent.runner.interaction_store import (
+    _epoch_to_datetime,
     DuplicateInteractionError,
     InteractionAlreadySubmittedError,
     InteractionExpiredError,
@@ -24,6 +25,21 @@ from langbot.pkg.entity.persistence.pipeline import LegacyPipeline
 UTC = datetime.timezone.utc
 
 
+@pytest.mark.parametrize('scale', [1, 1000])
+def test_epoch_to_datetime_accepts_seconds_and_milliseconds(scale):
+    assert _epoch_to_datetime(1_790_000_000.125 * scale) == datetime.datetime.fromtimestamp(1_790_000_000.125, UTC)
+
+
+@pytest.mark.parametrize('value', [float('nan'), float('inf'), -float('inf'), 10**400, 'invalid', []])
+def test_epoch_to_datetime_reports_invalid_timestamp(value):
+    with pytest.raises(ValueError, match='expires_at must be a valid epoch timestamp'):
+        _epoch_to_datetime(value)
+
+
+def test_epoch_to_datetime_preserves_absent_expiry():
+    assert _epoch_to_datetime(None) is None
+
+
 @pytest.fixture
 async def db_engine(tmp_path):
     engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "interactions.db"}', echo=False)
@@ -72,6 +88,14 @@ async def _create(store: InteractionStore, **overrides):
     return await store.create_request(**values)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize('scale', [1, 1000])
+async def test_create_request_persists_normalized_expiry(store, scale):
+    expires_at = int((datetime.datetime.now(UTC) + datetime.timedelta(minutes=5)).timestamp())
+    record, _token = await _create(store, expires_at=expires_at * scale)
+    assert record['expires_at'] == expires_at
+
+
 @pytest.mark.asyncio
 async def test_create_request_returns_token_but_persists_only_hash(store, db_engine):
     record, token = await _create(store)

tests/unit_tests/agent/test_run_journal.py

@@ -1,14 +1,79 @@
 """Agent monitoring snapshots retain useful input without duplicating binary payloads."""
 
+import datetime
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
 
 from langbot.pkg.agent.runner.run_journal import AgentRunJournal
+from langbot.pkg.agent.runner import event_log_store
+from langbot.pkg.agent.runner.host_models import AgentEventEnvelope
 from langbot.pkg.storage.media import MediaCache
 
 
+@pytest.mark.parametrize('scale', [1, 1000, 1_000_000])
+@pytest.mark.parametrize('seconds', [1_790_000_000, 1_790_000_000.125, -1_790_000_000, 10_000_000_000])
+def test_event_time_normalizes_current_epoch_units(scale, seconds):
+    assert AgentRunJournal._event_time_to_utc(seconds * scale) == datetime.datetime.fromtimestamp(
+        seconds, datetime.timezone.utc
+    )
+
+
+@pytest.mark.parametrize(
+    'value',
+    [None, True, False, 0, '1790000000', [], {}, float('nan'), float('inf'), -float('inf'), 10**400, -(10**400)],
+)
+def test_event_time_rejects_invalid_values_without_raising(value):
+    assert AgentRunJournal._event_time_to_utc(value) is None
+
+
+@pytest.fixture
+def event_log_writer(monkeypatch):
+    append_event = AsyncMock(return_value='event-log-id')
+    monkeypatch.setattr(event_log_store, 'EventLogStore', lambda engine: SimpleNamespace(append_event=append_event))
+    journal = AgentRunJournal(SimpleNamespace(persistence_mgr=SimpleNamespace(get_db_engine=lambda: None)))
+    return journal, append_event
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('scale', [1, 1000, 1_000_000])
+async def test_write_event_log_normalizes_timestamp(event_log_writer, scale):
+    journal, append_event = event_log_writer
+    event = AgentEventEnvelope(
+        event_id='event',
+        event_type='message.received',
+        source='platform',
+        event_time=1_790_000_000 * scale,
+        input={},
+        delivery={'surface': 'platform'},
+    )
+    assert await journal.write_event_log(event, None, 'run', 'runner') == 'event-log-id'
+    assert append_event.call_args.kwargs['event_time'] == datetime.datetime.fromtimestamp(
+        1_790_000_000, datetime.timezone.utc
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('scale', [1, 1000, 1_000_000])
+async def test_steering_dropped_audit_normalizes_timestamp(event_log_writer, scale):
+    journal, append_event = event_log_writer
+    await journal.write_steering_dropped_audits([{'event': {'event_time': 1_790_000_000 * scale}}], 'run', 'runner')
+    assert append_event.call_args.kwargs['event_type'] == 'steering.dropped'
+    assert append_event.call_args.kwargs['event_time'] == datetime.datetime.fromtimestamp(
+        1_790_000_000, datetime.timezone.utc
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('value', [None, 0, True, 'invalid', float('inf'), 10**400])
+async def test_invalid_steering_time_does_not_prevent_audit(event_log_writer, value):
+    journal, append_event = event_log_writer
+    await journal.write_steering_dropped_audits([{'event': {'event_time': value}}], 'run', 'runner')
+    append_event.assert_awaited_once()
+    assert append_event.call_args.kwargs['event_time'] is None
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize('event_type', ['message.received', 'group.member_joined'])
 async def test_agent_run_snapshot_records_identity_and_redacts_inline_attachments(event_type):

tests/unit_tests/agent/test_run_ledger_store.py

@@ -12,7 +12,7 @@ from sqlalchemy.ext.asyncio import create_async_engine
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import sessionmaker
 
-from langbot.pkg.agent.runner.run_ledger_store import RunLedgerStore
+from langbot.pkg.agent.runner.run_ledger_store import RunLedgerStore, _epoch_to_datetime
 from langbot.pkg.entity.persistence.agent_run import AgentRun
 from langbot.pkg.entity.persistence.base import Base
 
@@ -20,6 +20,18 @@ from langbot.pkg.entity.persistence.base import Base
 UTC = datetime.timezone.utc
 
 
+@pytest.mark.parametrize('scale', [1, 1000])
+def test_epoch_to_datetime_accepts_seconds_and_milliseconds(scale):
+    assert _epoch_to_datetime(1_790_000_000.125 * scale) == datetime.datetime.fromtimestamp(
+        1_790_000_000.125, UTC
+    ).replace(tzinfo=None)
+
+
+@pytest.mark.parametrize('value', [None, float('nan'), float('inf'), -float('inf'), 10**400, 'invalid', []])
+def test_epoch_to_datetime_ignores_invalid_timestamp(value):
+    assert _epoch_to_datetime(value) is None
+
+
 @pytest.fixture
 async def db_engine(tmp_path):
     db_path = tmp_path / 'run_ledger_store.db'
@@ -38,6 +50,19 @@ def store(db_engine):
     return RunLedgerStore(db_engine)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize('scale', [1, 1000])
+async def test_create_run_persists_normalized_deadline(store, scale):
+    record = await store.create_run(
+        run_id='run-timestamp',
+        event_id='event',
+        binding_id='binding',
+        runner_id='runner',
+        deadline_at=1_790_000_000 * scale,
+    )
+    assert record['deadline_at'] == 1_790_000_000
+
+
 @pytest.fixture
 async def legacy_run(store, db_engine):
     await store.create_run(