diff --git a/src/iai_mcp/migrate/_timestamps.py b/src/iai_mcp/migrate/_timestamps.py index 11574f2..70573db 100644 --- a/src/iai_mcp/migrate/_timestamps.py +++ b/src/iai_mcp/migrate/_timestamps.py @@ -70,6 +70,12 @@ def _find_transcript_ts( # the hash matches what was actually stored as literal_surface. msg = obj.get("message") msg = msg if isinstance(msg, dict) else obj + # Internal events (e.g. queue-operation) can carry the same + # text as the real user/assistant turn but appear earlier + # in the file; skip them so the real turn's timestamp wins. + role = obj.get("type") or msg.get("role", "") + if role not in {"user", "assistant"}: + continue content = msg.get("content", "") if isinstance(content, list): parts = [] diff --git a/tests/test_migrate.py b/tests/test_migrate.py index 1d42914..79e3747 100644 --- a/tests/test_migrate.py +++ b/tests/test_migrate.py @@ -529,6 +529,59 @@ def test_migrate_rederive_content_hash_fallback_matches_nested_message(tmp_path) assert ts != collapsed_ts +def test_migrate_rederive_skips_internal_event_duplicate(tmp_path): + """An internal queue-operation line earlier in the transcript must not win + the content-hash match over the real user turn with the same text — + otherwise rederive re-derives the same already-collapsed timestamp. + Mirrors a real collapsed group: several duplicate episodic records sharing + both the collapsed created_at and identical literal_surface content.""" + from iai_mcp.migrate import migrate_rederive_collapsed_timestamps + from iai_mcp.store import MemoryStore + + store = MemoryStore(path=tmp_path) + session_id = "sess-internal-dup" + transcript_root = tmp_path / "transcripts" + + collapsed_ts = datetime(2026, 6, 20, 21, 11, 0, 188000, tzinfo=timezone.utc) + text = "Duplicate task-notification payload" + + # No source_uuid recorded in provenance — forces the content-hash fallback. + # Group size >= 3 to qualify as a collapsed-timestamp candidate. + records = [_make_episodic_record(text, session_id, "", collapsed_ts) for _ in range(3)] + for r in records: + store.insert(r) + + real_ts_str = "2026-06-20T21:11:00.723000Z" + transcript_entries = [ + # Internal event carrying the same text, earlier in the file, at the + # collapsed timestamp — must be skipped. + { + "type": "queue-operation", + "timestamp": "2026-06-20T21:11:00.188000Z", + "sessionId": session_id, + "message": {"content": text}, + }, + # The real user turn, a moment later, at the correct timestamp. + { + "type": "user", + "timestamp": real_ts_str, + "sessionId": session_id, + "uuid": str(uuid4()), + "message": {"role": "user", "content": text}, + }, + ] + _write_fake_transcript(transcript_root, session_id, transcript_entries) + + result = migrate_rederive_collapsed_timestamps(store, transcript_root=transcript_root) + + assert result["records_updated"] == 3 + expected_ts = datetime(2026, 6, 20, 21, 11, 0, 723000, tzinfo=timezone.utc) + for r in records: + updated = store.get(r.id) + assert updated.created_at != collapsed_ts + assert updated.created_at == expected_ts + + def test_migrate_rederive_writes_event(tmp_path): """A migration_rederive_timestamps event is written after a non-dry run.""" from iai_mcp.events import query_events