fix(runtime): preserve run event timestamp offsets in SQLite (#6894)

* fix(runtime): preserve run event timestamp offsets in SQLite

* docs: clarify run event timestamp upgrade behavior

---------

Co-authored-by: XY-code1 <231763295+XY-code1@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -133,6 +133,8 @@ That prompt is intended for coding agents. It tells the agent to clone the repo
 
 ### Configuration
 
+With `run_events.backend: db`, caller-supplied ISO 8601 event timestamps are normalized to UTC before storage, preserving their instant when SQLite drops timezone information. Timestamps without an offset are treated as UTC. Previously stored timestamps are not rewritten.
+
 Operators can extend lead-agent, subagent, and DeerMem extraction prompts with
 literal prepend/append configuration without editing source templates. See
 [prompt overlays](backend/docs/CONFIGURATION.md#prompt-overlays).

backend/docs/CONFIGURATION.md

@@ -736,6 +736,17 @@ Notes:
   - **Language:** the owner's web UI language (the `locale` user preference, which the web app keeps in sync), else `channel_connections.notification_locale` (`en-US` or `zh-CN`, default `en-US`; startup-scoped). Auth-disabled installs always use the config value.
   - **Delivery:** channel/transport outages park rows without consuming the retry budget, for up to about a day; platform rejections exhaust ~15 minutes of counted retries then settle `failed`. The worker re-checks the binding right before sending: a target the owner has disconnected since enqueue is dropped as `failed`, never pushed.
 
+### Run event timestamps
+
+With `run_events.backend: db`, caller-supplied ISO 8601 event timestamps are
+normalized to UTC before storage, preserving their instant when SQLite drops
+timezone information. Timestamps without an offset are treated as UTC.
+
+**Upgrade note:** Previously stored timestamps are not rewritten. If callers
+supplied non-UTC offsets before this fix, existing SQLite `created_at` values
+may represent the wrong instant. Upgrading does not correct those values;
+the original offset cannot be recovered from the stored timestamp alone.
+
 ### Deployment topology (multi-instance)
 
 ```yaml

backend/packages/harness/deerflow/runtime/events/store/db.py

@@ -26,6 +26,13 @@ from deerflow.utils.time import coerce_iso
 logger = logging.getLogger(__name__)
 
 
+def _parse_created_at(value: str | None) -> datetime:
+    timestamp = datetime.fromisoformat(value) if value else datetime.now(UTC)
+    # SQLite drops tzinfo, so convert the clock value before it reaches storage.
+    # Preserve the existing convention that timestamps without an offset are UTC.
+    return timestamp.replace(tzinfo=UTC) if timestamp.tzinfo is None else timestamp.astimezone(UTC)
+
+
 class DbRunEventStore(RunEventStore):
     def __init__(self, session_factory: async_sessionmaker[AsyncSession], *, max_trace_content: int = 10240):
         self._sf = session_factory
@@ -227,7 +234,7 @@ class DbRunEventStore(RunEventStore):
                         content=db_content,
                         event_metadata=metadata,
                         seq=seq,
-                        created_at=datetime.fromisoformat(created_at) if created_at else datetime.now(UTC),
+                        created_at=_parse_created_at(created_at),
                     )
                     session.add(row)
                     await self._raise_seq_watermark(session, thread_id, seq)
@@ -264,7 +271,7 @@ class DbRunEventStore(RunEventStore):
                             content=db_content,
                             event_metadata=metadata,
                             seq=seq,
-                            created_at=datetime.fromisoformat(e["created_at"]) if e.get("created_at") else datetime.now(UTC),
+                            created_at=_parse_created_at(e.get("created_at")),
                         )
                         session.add(row)
                         rows.append(row)
@@ -319,7 +326,7 @@ class DbRunEventStore(RunEventStore):
                         content=db_content,
                         event_metadata=metadata,
                         seq=(max_seq or 0) + 1,
-                        created_at=datetime.fromisoformat(created_at) if created_at else datetime.now(UTC),
+                        created_at=_parse_created_at(created_at),
                     )
                     session.add(row)
                     await self._raise_seq_watermark(session, thread_id, row.seq)

backend/tests/test_persistence_timezone.py

@@ -11,6 +11,7 @@ threads by the local UTC offset.
 """
 
 import re
+from datetime import UTC, datetime
 
 import pytest
 
@@ -36,6 +37,46 @@ async def _cleanup():
     await close_engine()
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize("write_method", ["put", "put_batch", "put_if_absent"])
+@pytest.mark.parametrize(
+    ("timestamp", "expected"),
+    [
+        ("2026-10-14T12:34:56.123456+08:00", "2026-10-14T04:34:56.123456+00:00"),
+        ("2026-10-14T23:45:00-05:30", "2026-10-15T05:15:00+00:00"),
+        ("2026-10-14T04:34:56Z", "2026-10-14T04:34:56+00:00"),
+        ("2026-10-14T04:34:56", "2026-10-14T04:34:56+00:00"),
+    ],
+)
+async def test_run_event_timestamp_preserves_instant_after_sqlite_round_trip(tmp_path, write_method, timestamp, expected):
+    from deerflow.runtime.events.store.db import DbRunEventStore
+
+    store = DbRunEventStore(await _init_sqlite(tmp_path))
+    event = dict(thread_id="t-tz", run_id="r-tz", event_type="llm.ai.response", category="message", content="answer", created_at=timestamp)
+    try:
+        if write_method == "put_batch":
+            written = (await store.put_batch([event]))[0]
+        elif write_method == "put_if_absent":
+            written, created = await store.put_if_absent(**event)
+            assert created
+        else:
+            written = await store.put(**event)
+
+        assert written["created_at"] == expected
+        # A new query must read the persisted clock value, not the ORM identity map.
+        fetched = await store.list_events("t-tz", "r-tz", user_id=None)
+        assert len(fetched) == 1
+        assert fetched[0]["created_at"] == expected
+        assert datetime.fromisoformat(fetched[0]["created_at"]).tzinfo == UTC
+
+        if write_method == "put_if_absent":
+            existing, created = await store.put_if_absent(**{**event, "created_at": "2026-10-16T00:00:00Z"})
+            assert not created
+            assert existing["created_at"] == expected
+    finally:
+        await _cleanup()
+
+
 @pytest.mark.anyio
 async def test_thread_meta_emits_tz_aware_timestamps(tmp_path):
     from deerflow.persistence.thread_meta import ThreadMetaRepository