From d4501e02f56cb6b323b66894093a6c49d6787fea Mon Sep 17 00:00:00 2001 From: Mahmoud Mabrouk Date: Fri, 24 Jul 2026 23:58:29 +0200 Subject: [PATCH] fix(api): order session records by producer time so turns stop interleaving `get_records` ordered by (created_at, record_index). `record_index` restarts at 0 on every turn, and the ingest worker batches records from adjacent turns into one write, so those rows share a `created_at`. The tiebreak then sorted the NEXT turn's first record ahead of the PREVIOUS turn's later ones, and a conversation rebuilt from the log came back interleaved: a live three-turn run reconstructed as user, user, assistant, user with the replies in the wrong places. Order by the producer-stamped `timestamp` first. It is the only key monotonic across turns, and the runner already sets it on every record. Ingest time and the per-turn index stay as tiebreaks, and rows predating `timestamp` sort last within their batch rather than jumping the queue. Claude-Session: https://claude.ai/code/session_01KM69J7uHafgciiN5zfG7qR --- api/oss/src/dbs/postgres/sessions/records/dao.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/api/oss/src/dbs/postgres/sessions/records/dao.py b/api/oss/src/dbs/postgres/sessions/records/dao.py index 93ff6ed548..7554e48f70 100644 --- a/api/oss/src/dbs/postgres/sessions/records/dao.py +++ b/api/oss/src/dbs/postgres/sessions/records/dao.py @@ -109,7 +109,17 @@ async def get_records( RecordDBE.project_id == project_id, RecordDBE.session_id == session_id, ) - .order_by(RecordDBE.created_at.asc(), RecordDBE.record_index.asc()) + # Producer event time first: it is the only key that is monotonic across + # turns. `record_index` restarts at 0 every turn, and the worker can batch + # records from two turns into one write so they share `created_at` — the old + # (created_at, record_index) order then sorted the NEXT turn's first record + # ahead of the PREVIOUS turn's later ones, interleaving the conversation. + # Rows written before `timestamp` existed sort last within their ingest batch. + .order_by( + RecordDBE.timestamp.asc().nullslast(), + RecordDBE.created_at.asc(), + RecordDBE.record_index.asc(), + ) ) dbes = (await session.execute(stmt)).scalars().all()