From 6b57a1074ff4002a0153f667435c2418b617f0f3 Mon Sep 17 00:00:00 2001 From: leonoxo Date: Tue, 27 Oct 2026 23:49:01 +0800 Subject: [PATCH] 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 --------- Co-authored-by: leonoxo Co-authored-by: Chin Rock Co-authored-by: multica-agent --- .../pkg/agent/runner/interaction_store.py | 7 +- src/langbot/pkg/agent/runner/run_journal.py | 35 +++++----- .../pkg/agent/runner/run_ledger_store.py | 7 +- .../agent/test_interaction_store.py | 24 +++++++ tests/unit_tests/agent/test_run_journal.py | 65 +++++++++++++++++++ .../unit_tests/agent/test_run_ledger_store.py | 27 +++++++- 6 files changed, 144 insertions(+), 21 deletions(-) diff --git a/src/langbot/pkg/agent/runner/interaction_store.py b/src/langbot/pkg/agent/runner/interaction_store.py index ccd8178b9..22bc21eb6 100644 --- a/src/langbot/pkg/agent/runner/interaction_store.py +++ b/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 diff --git a/src/langbot/pkg/agent/runner/run_journal.py b/src/langbot/pkg/agent/runner/run_journal.py index ed288ff55..e0724f841 100644 --- a/src/langbot/pkg/agent/runner/run_journal.py +++ b/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()), diff --git a/src/langbot/pkg/agent/runner/run_ledger_store.py b/src/langbot/pkg/agent/runner/run_ledger_store.py index 6be6aaab2..ade45de91 100644 --- a/src/langbot/pkg/agent/runner/run_ledger_store.py +++ b/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 diff --git a/tests/unit_tests/agent/test_interaction_store.py b/tests/unit_tests/agent/test_interaction_store.py index c3dcb6f93..4502bd1cb 100644 --- a/tests/unit_tests/agent/test_interaction_store.py +++ b/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) diff --git a/tests/unit_tests/agent/test_run_journal.py b/tests/unit_tests/agent/test_run_journal.py index c229318ca..3b76ffe73 100644 --- a/tests/unit_tests/agent/test_run_journal.py +++ b/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): diff --git a/tests/unit_tests/agent/test_run_ledger_store.py b/tests/unit_tests/agent/test_run_ledger_store.py index 04d41a26a..3f3b2b601 100644 --- a/tests/unit_tests/agent/test_run_ledger_store.py +++ b/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(