diff --git a/api/models.py b/api/models.py index cfac4492..771b7d75 100644 --- a/api/models.py +++ b/api/models.py @@ -3863,6 +3863,29 @@ def _session_message_merge_key(msg: dict): ) +def _session_message_dedup_key(msg: dict): + """Like _session_message_merge_key but preserves full-precision timestamp. + + Two messages are true duplicates only if role, content, AND exact + timestamp all match. Sub-second timestamp differences indicate + legitimately distinct messages (e.g. two assistant turns within the + same wall-clock second). + """ + if not isinstance(msg, dict): + return ("non_dict", repr(msg)) + message_identity = msg.get("id") or msg.get("message_id") + if message_identity: + return ("message_id", str(message_identity)) + return ( + "legacy", + str(msg.get("role") or ""), + str(msg.get("content") or ""), + str(msg.get("timestamp") or ""), + str(msg.get("tool_call_id") or ""), + str(msg.get("tool_name") or msg.get("name") or ""), + ) + + def _normalized_session_message_content(msg: dict) -> str: if not isinstance(msg, dict): return repr(msg) @@ -4000,17 +4023,34 @@ def merge_session_messages_append_only( return sidecar_messages if not sidecar_messages: if watermark_timestamp is not None: - return [ + filtered = [ msg for msg in state_messages if ( (timestamp := _message_timestamp_as_float(msg)) is not None and timestamp <= watermark_timestamp ) ] - return state_messages + else: + filtered = state_messages + # Deduplicate true duplicates (same role, content, exact timestamp) + # without collapsing legitimately-repeated identical turns (#3346). + # Note: rows whose timestamps were mutated by compaction/recovery to + # microsecond-different values will not be folded — only byte-identical + # timestamps are treated as the same message. This is intentional; + # collapsing same-second distinct turns would be worse than retaining + # a compaction-restamped duplicate. + seen = set() + deduped = [] + for msg in filtered: + key = _session_message_dedup_key(msg) + if key not in seen: + seen.add(key) + deduped.append(msg) + return deduped merged_messages = [] seen_message_keys = set() + seen_dedup_keys = set() seen_content_keys = set() seen_visible_keys = set() sidecar_visible_sequence = [] @@ -4023,6 +4063,7 @@ def merge_session_messages_append_only( max_sidecar_timestamp = timestamp if max_sidecar_timestamp is None else max(max_sidecar_timestamp, timestamp) key = _session_message_merge_key(msg) seen_message_keys.add(key) + seen_dedup_keys.add(_session_message_dedup_key(msg)) seen_content_keys.add(_session_message_content_key(msg)) visible_key = _session_message_visible_key(msg) seen_visible_keys.add(visible_key) @@ -4055,6 +4096,9 @@ def merge_session_messages_append_only( skipped_state_visible_counts[matched_visible_key] = ( skipped_state_visible_counts.get(matched_visible_key, 0) + 1 ) + # Record dedup key so later duplicates of this replayed message + # are caught by the dedup guard (#3346). + seen_dedup_keys.add(_session_message_dedup_key(msg)) continue # Skip rows ABOVE the watermark only while the sidecar has NOT advanced # past the watermark. Because Session.save() no longer auto-clears the @@ -4093,12 +4137,24 @@ def merge_session_messages_append_only( and _session_message_content_key(msg) not in seen_content_keys ): continue + # Check for true duplicates using full-precision timestamp (#3346). + # Must run before the merge-key guards so that legitimately distinct + # sub-second messages with the same second-level merge key are not + # collapsed. The merge key truncates to seconds; the dedup key does + # not. + dedup_key = _session_message_dedup_key(msg) + if dedup_key in seen_dedup_keys: + continue if max_sidecar_timestamp is not None and timestamp is not None and timestamp <= max_sidecar_timestamp: - if key in seen_message_keys: + # For message_id keys the merge key is authoritative — skip if + # already seen. For legacy keys the dedup check above already + # handled true duplicates; same-second distinct messages must + # fall through. + if key in seen_message_keys and key[0] == "message_id": continue if not (isinstance(key, tuple) and key[:1] == ("message_id",)): continue - if key in seen_message_keys: + if key in seen_message_keys and key[0] == "message_id": continue matched_visible_key = _matching_visible_duplicate( visible_key, @@ -4128,8 +4184,8 @@ def merge_session_messages_append_only( and timestamp <= max_sidecar_timestamp ): continue - if key[0] == "message_id": - seen_message_keys.add(key) + seen_message_keys.add(key) + seen_dedup_keys.add(dedup_key) seen_content_keys.add(_session_message_content_key(msg)) seen_visible_keys.add(visible_key) merged_messages.append(msg) diff --git a/tests/test_issue3346_legacy_dedup.py b/tests/test_issue3346_legacy_dedup.py new file mode 100644 index 00000000..003b2969 --- /dev/null +++ b/tests/test_issue3346_legacy_dedup.py @@ -0,0 +1,85 @@ +"""Regression tests for #3346: merge_session_messages_append_only fails to +deduplicate legacy state messages (messages without explicit id/message_id). + +Three repro rows from the bug report: + + Row A — Gap 1: empty sidecar, duplicate state (no timestamps) + Row B — Gap 2: non-empty sidecar, duplicate legacy state + Row C — Gap 2 (replay variant): sidecar has explicit id, state has + two copies of the same legacy message; first is consumed by + the replay-prefix branch, second must still be deduped. +""" +from __future__ import annotations + +import pytest + +from api.models import merge_session_messages_append_only + + +def _legacy(role: str, content: str, timestamp=None) -> dict: + msg = {"role": role, "content": content} + if timestamp is not None: + msg["timestamp"] = timestamp + return msg + + +def _identified(role: str, content: str, msg_id: str, timestamp=None) -> dict: + msg = {"id": msg_id, "role": role, "content": content} + if timestamp is not None: + msg["timestamp"] = timestamp + return msg + + +# ── Row A: Gap 1 — empty sidecar, early-return path ────────────────────────── + +@pytest.mark.parametrize("use_watermark", [False, True]) +def test_empty_sidecar_deduplicates_identical_legacy_state(use_watermark): + """Empty sidecar with duplicate state rows must return a single message.""" + a = _legacy("user", "hello") + state = [a, a] # true duplicate — same role, content, no timestamp + watermark = "2030-01-01T00:00:00Z" if use_watermark else None + result = merge_session_messages_append_only([], state, truncation_watermark=watermark) + assert len(result) == 1, f"expected 1 (deduped), got {len(result)}" + assert result[0] == a + + +# ── Row B: Gap 2 — non-empty sidecar, legacy dup in state ──────────────────── + +def test_nonempty_sidecar_deduplicates_identical_legacy_state(): + """Non-empty sidecar: duplicate legacy state rows must not both appear.""" + sidecar = [_identified("system", "sys", msg_id="s1")] + a = _legacy("user", "hello") + state = [a, a] + result = merge_session_messages_append_only(sidecar, state) + contents = [m["content"] for m in result] + assert contents == ["sys", "hello"], f"expected [sys, hello], got {contents}" + + +# ── Row C: Gap 2 (replay) — sidecar id-keyed, state has two legacy copies ──── + +def test_replay_then_legacy_dup_is_deduped(): + """Sidecar has {id, a}; state has [a, a]. + First state 'a' is consumed by the replay-prefix branch; second must be + caught by the dedup guard rather than appended. + """ + sidecar = [_identified("user", "hello", msg_id="a")] + a = _legacy("user", "hello") + state = [a, a] + result = merge_session_messages_append_only(sidecar, state) + assert len(result) == 1, f"expected 1, got {len(result)}: {result}" + assert result[0]["id"] == "a", "sidecar message should win" + + +# ── Same-second distinct turns must be preserved ───────────────────────────── + +def test_same_second_distinct_turns_preserved(): + """Two messages with the same role+content but different sub-second + timestamps are legitimately distinct turns and must not be collapsed. + """ + sidecar = [_legacy("user", "start", timestamp=1779300508)] + state = [ + _legacy("assistant", "Still working", timestamp=1779300509.12663), + _legacy("assistant", "Still working", timestamp=1779300509.82718), + ] + result = merge_session_messages_append_only(sidecar, state) + assert len(result) == 3, f"expected 3 (distinct sub-second turns), got {len(result)}"