Fix/episodic capture timestamp and dedup guard

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 = []

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