rss_generator.py

@@ -9,7 +9,6 @@
 # pylance: disable=reportMissingImports, reportMissingModuleSource
 # mypy: disable-error-code="import-untyped"
 
-import copy
 import logging
 import asyncio
 import re
@@ -46,72 +45,80 @@ class ChannelNotFound(Exception):
         self.channel_identifier = channel_identifier
         super().__init__(str(channel_identifier))
 
-def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]:
-    """
-    Create media groups based on time difference between messages
+def _compute_time_based_group_ids(messages: list[Message], merge_seconds: int = 5) -> dict[int, str | int]:
+    """Return {message.id: effective_media_group_id} WITHOUT mutating messages.
 
     Plain synchronous function (contains no await): runs inside the render thread
     via _render_pipeline. Must not touch asyncio.
+
+    Contract: all messages belong to ONE chat (message.id is unique only per
+    chat); callers must not mix chats in a single call.
+
+    Pure re-implementation of the old _create_time_based_media_groups mutation,
+    reproducing its result for every input the old code survived on PRODUCTION
+    data (naive kurigram dates). Inputs only aware-date test mocks could produce
+    (fully-None and aware+None mixes, where the old code clustered the None-date
+    tail by insertion order incl. truthy-id adoption) are deliberately replaced
+    (registry §3.11):
+      - messages WITHOUT a date do not participate in time clustering and get
+        NO mapping entry (their own media_group_id still applies downstream);
+      - dated messages are ordered ascending by date via a timestamp key
+        (naive-safe: no aware/naive mix once None-date are excluded);
+      - a message joins the current cluster if the gap to the PREVIOUS message
+        is <= merge_seconds; the gap is a NAIVE datetime subtraction
+        (msg.date - prev.date).total_seconds(), exactly as the old code — NOT a
+        timestamp diff (they diverge across a DST fold, and the old behavior is
+        the contract);
+      - the effective id of a cluster is the FIRST TRUTHY media_group_id in
+        cluster order (old code used truthiness, not `is not None`, and
+        overwrote members' own differing ids — kept);
+      - a cluster of >= 2 members with no truthy id gets a synthetic
+        f"time_{min(dates)}" id (exact old format);
+      - singleton clusters and clusters with no effective id produce NO entries;
+        every member of a cluster with an effective id gets one.
     """
-    # Deep-copy the input list to avoid mutating cached Message objects (the cache may
-    # reuse the same objects across calls with different merge_seconds values)
-    messages = copy.deepcopy(messages)
-    # Compute fallback once so all None-date messages get the same sort key (deterministic order)
-    _sort_fallback = datetime.now(timezone.utc)
-    messages_sorted = sorted(messages, key=lambda msg: msg.date or _sort_fallback) # type: ignore
+    dated = sorted((m for m in messages if m.date is not None),
+                   key=lambda m: m.date.timestamp())  # type: ignore
+    group_ids: dict[int, str | int] = {}
+
+    def _flush(cluster: list[Message], effective: str | int | None) -> None:
+        if len(cluster) < 2:
+            return
+        if not effective:
+            # All members are dated here (None-date excluded above), so min() is safe.
+            effective = f"time_{min(m.date for m in cluster)}"  # type: ignore
+        for m in cluster:
+            group_ids[m.id] = effective
+
     cluster: list[Message] = []
-    last_msg_date: datetime = datetime.now(timezone.utc)
-    current_media_group_id: Optional[str] = None
+    effective: str | int | None = None
+    prev_date: Optional[datetime] = None
 
-    for msg in messages_sorted:
-        
+    for msg in dated:
+        mgid = getattr(msg, "media_group_id", None)
         if not cluster:
-            cluster.append(msg)
-            # Use current time as fallback when date is None
-            last_msg_date = msg.date or datetime.now(timezone.utc) # type: ignore
-            current_media_group_id = getattr(msg, "media_group_id", None)
-            continue
-        
-        # Use current time as fallback when date is None to avoid TypeError in subtraction
-        msg_date = msg.date or datetime.now(timezone.utc)
-        time_diff = (msg_date - last_msg_date).total_seconds()
-        
-        msg_media_group_id = getattr(msg, "media_group_id", None)
-        
-        if time_diff <= merge_seconds:
-            if current_media_group_id:
-                msg.media_group_id = current_media_group_id  # type: ignore
-            elif msg_media_group_id:
-                current_media_group_id = msg_media_group_id
-                for m in cluster:
-                    m.media_group_id = current_media_group_id  # type: ignore
-            cluster.append(msg)
-            # Use current time as fallback when date is None
-            last_msg_date = msg.date or datetime.now(timezone.utc) # type: ignore
-        else:
-            if len(cluster) >= 2 and not current_media_group_id:
-                dates = [m.date for m in cluster if m.date is not None]
-                if dates:
-                    min_date = min(dates)
-                    new_group_id = f"time_{min_date}"
-                    for m in cluster:
-                        m.media_group_id = new_group_id  # type: ignore
             cluster = [msg]
-            # Use current time as fallback when date is None
-            last_msg_date = msg.date or datetime.now(timezone.utc) # type: ignore
-            current_media_group_id = msg_media_group_id
-    
-    if len(cluster) >= 2 and not current_media_group_id:
-        dates = [m.date for m in cluster if m.date is not None]
-        if dates:
-            min_date = min(dates)
-            new_group_id = f"time_{min_date}"
-            for m in cluster:
-                m.media_group_id = new_group_id  # type: ignore
+            effective = mgid or None
+            prev_date = msg.date
+            continue
+        time_diff = (msg.date - prev_date).total_seconds()  # type: ignore
+        if time_diff <= merge_seconds:
+            cluster.append(msg)
+            # First truthy id in cluster order wins; keep it even if this member
+            # carries a different truthy id (the old code overwrote it).
+            if not effective and mgid:
+                effective = mgid
+            prev_date = msg.date
+        else:
+            _flush(cluster, effective)
+            cluster = [msg]
+            effective = mgid or None
+            prev_date = msg.date
 
-    return messages_sorted
+    _flush(cluster, effective)
+    return group_ids
 
-def _create_messages_groups(messages: list[Message]) -> list[list[Message]]:
+def _create_messages_groups(messages: list[Message], group_ids: dict[int, str | int] | None = None) -> list[list[Message]]:
     """
     Process messages into formatted posts, handling media groups
 
@@ -119,7 +126,11 @@ def _create_messages_groups(messages: list[Message]) -> list[list[Message]]:
     """
     processing_groups: list[list[Message]] = []
     media_groups: dict[str | int, list[Message]] = {}
-    
+    # Time-clustering supplies effective ids as a PURE mapping (no message mutation,
+    # no deepcopy). Absent an entry, the message's own media_group_id applies
+    # (registry §3.11 / spec Этап 4).
+    group_ids = group_ids or {}
+
     # First pass - collect messages and organize into processing groups
     for message in messages:
         try:
@@ -135,10 +146,11 @@ def _create_messages_groups(messages: list[Message]) -> list[list[Message]]:
                 if 'CHANNEL_CHAT_CREATED'   in str(message.service): continue
                 if 'DELETE_CHAT_PHOTO'      in str(message.service): continue
 
-            if message.media_group_id:
-                if message.media_group_id not in media_groups:
-                    media_groups[message.media_group_id] = []
-                media_groups[message.media_group_id].append(message)
+            effective_group_id = group_ids.get(message.id, message.media_group_id)
+            if effective_group_id:
+                if effective_group_id not in media_groups:
+                    media_groups[effective_group_id] = []
+                media_groups[effective_group_id].append(message)
             else:
                 processing_groups.append([message]) # Single message becomes its own processing group
                 
@@ -152,9 +164,15 @@ def _create_messages_groups(messages: list[Message]) -> list[list[Message]]:
         media_group.sort(key=lambda x: x.id, reverse=False)
         processing_groups.append(media_group)
     
-    # Sort processing groups by date of first message in each group
-    processing_groups.sort(key=lambda group: group[0].date if group[0].date else datetime.now(timezone.utc), reverse=True)
-    
+    # Sort processing groups by date of first message in each group. Timestamp-based key
+    # is naive-safe: kurigram dates are naive-local, and a None-date group used to fall
+    # back to an AWARE datetime.now(timezone.utc) here, raising TypeError on the first
+    # naive-vs-aware comparison — a 500 on ANY feed carrying a None-date post, in the
+    # DEFAULT path (registry §3.12). None-date groups now sort as newest (float('inf'))
+    # and deterministically survive the later [:limit] slice; the final post sort in
+    # _render_messages_groups (0.0 fallback) still places them at the tail of the feed.
+    processing_groups.sort(key=lambda group: group[0].date.timestamp() if group[0].date else float('inf'), reverse=True)
+
     return processing_groups
 
 def _trim_messages_groups(messages_groups: list[list[Message]], limit: int):
@@ -301,15 +319,23 @@ def _render_pipeline(messages: list[Message],
 
     Runs entirely in a worker thread via a single asyncio.to_thread call. It contains
     NO await, NO asyncio primitives, NO create_task/get_running_loop — all the CPU-heavy
-    work (deepcopy, grouping, rendering, bleach) happens here off the event loop.
+    work (grouping, rendering, bleach) happens here off the event loop.
     Media file-id records are accumulated on post_parser._pending_media_ids and flushed
     by the caller after this returns.
 
     `channel` is used only for the sanitize log_context (grep-ability).
     """
     if time_based_merge:
-        messages = _create_time_based_media_groups(messages, merge_seconds)
-    message_groups = _create_messages_groups(messages)
+        # Pure mapping msg.id -> effective_group_id (no message mutation, no deepcopy),
+        # plus a date-ASC pre-sort with the same naive-safe timestamp key (None-date last
+        # via +inf). A stable sorted() on the fetch-order input reproduces the old
+        # `messages_sorted` order — including ties — and does NOT mutate the input, so the
+        # cache-protecting deepcopy is gone (spec Этап 4).
+        group_ids = _compute_time_based_group_ids(messages, merge_seconds)
+        messages = sorted(messages, key=lambda m: m.date.timestamp() if m.date else float('inf'))
+        message_groups = _create_messages_groups(messages, group_ids)
+    else:
+        message_groups = _create_messages_groups(messages)
     message_groups = _trim_messages_groups(message_groups, limit)
     posts = _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text)
     # Sanitize each surviving (post-filter) post exactly once, here in the worker
@@ -407,7 +433,7 @@ async def _prepare_feed_posts(channel: str | int,
         logger.debug(f"{log_prefix}_reply_enrichment_timing: channel {channel}, replies enriched in {enrichment_elapsed:.3f} seconds")
 
     # 5) Process messages into groups and render them. The whole grouping/trimming/
-    # rendering pipeline is CPU-heavy (deepcopy + per-message rendering) and contains no
+    # rendering pipeline is CPU-heavy (per-message rendering) and contains no
     # await, so run it in ONE worker thread to keep the event loop responsive.
     processing_start_time = time.time()
     try:

tests/test_group_ids.py

@@ -0,0 +1,272 @@
+# flake8: noqa
+# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
+# pylint: disable=redefined-outer-name, line-too-long
+# pylance: disable=reportMissingImports, reportMissingModuleSource
+"""Stage 4 (render-pipeline refactor, issue #31 / epic #34) — pure time-clustering.
+
+`_create_time_based_media_groups` deep-copied every cached message per feed request and
+MUTATED media_group_id. It is replaced by `_compute_time_based_group_ids`, a PURE function
+returning {message.id: effective_media_group_id}; `_create_messages_groups` reads that
+mapping instead of a mutated attribute. All tests use NAIVE dates (as kurigram emits on
+prod), not aware-UTC mocks.
+
+Behavior registry items exercised here: §3.11 (None-date excluded from clustering, no
+mapping entry; own media_group_id still applies) and §3.12 (naive-safe sort keys; None-date
+groups survive [:limit] as newest and land at the tail of the feed via the 0.0 final sort).
+"""
+import os
+import time as _time
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import pytest
+
+import rss_generator as rss_module
+from rss_generator import (
+    _compute_time_based_group_ids,
+    _create_messages_groups,
+    generate_channel_html,
+)
+
+
+D = datetime  # naive datetimes throughout
+
+
+class _Str(str):
+    """Str stand-in: .html returns the raw string (mirrors kurigram's Str)."""
+    @property
+    def html(self):
+        return str(self)
+
+
+class Msg:
+    """Minimal message stand-in for the pure clustering function (needs id/date/mgid)."""
+    def __init__(self, mid, date, media_group_id=None):
+        self.id = mid
+        self.date = date
+        self.media_group_id = media_group_id
+        self.service = None
+
+
+def at(sec, minute=0):
+    # Naive local datetime, same shape kurigram's datetime.fromtimestamp() produces.
+    return D(2024, 1, 1, 12, minute, sec)
+
+
+# --------------------------------------------------------------------------- #
+# Core clustering equivalence with the old mutation.
+# --------------------------------------------------------------------------- #
+def test_time_cluster_without_id_gets_synthetic():
+    a, b = Msg(1, at(0)), Msg(2, at(2))
+    mapping = _compute_time_based_group_ids([a, b], merge_seconds=5)
+    synthetic = f"time_{at(0)}"
+    assert mapping == {1: synthetic, 2: synthetic}
+
+
+def test_adoption_backfill_and_overwrite():
+    # First member has no id, second brings "B" (first truthy in cluster order -> wins and
+    # is BACKFILLED onto member 1), third brings "C" which is OVERWRITTEN to "B".
+    a, b, c = Msg(1, at(0)), Msg(2, at(2), "B"), Msg(3, at(4), "C")
+    mapping = _compute_time_based_group_ids([a, b, c], merge_seconds=5)
+    assert mapping == {1: "B", 2: "B", 3: "B"}
+
+
+def test_falsy_id_ignored_like_old_code():
+    # 0 and "" are falsy: they are NOT adopted as the cluster id, exactly as the old
+    # truthiness check. A 2-member cluster with only falsy ids gets a synthetic id.
+    a, b = Msg(1, at(0), 0), Msg(2, at(2), "")
+    mapping = _compute_time_based_group_ids([a, b], merge_seconds=5)
+    synthetic = f"time_{at(0)}"
+    assert mapping == {1: synthetic, 2: synthetic}
+
+
+def test_singleton_gets_no_entry():
+    # A lone message (even one carrying a truthy media_group_id) forms a singleton cluster
+    # and gets NO entry; downstream it falls back to its own media_group_id.
+    lonely = Msg(1, at(0), "MG")
+    far = Msg(2, at(30, minute=1), "OTHER")  # >5s gap -> separate singleton
+    mapping = _compute_time_based_group_ids([lonely, far], merge_seconds=5)
+    assert mapping == {}
+
+
+def test_input_is_not_mutated():
+    a, b, c = Msg(1, at(0)), Msg(2, at(2), "B"), Msg(3, at(4), "C")
+    before = [(m.id, m.date, m.media_group_id) for m in (a, b, c)]
+    _compute_time_based_group_ids([a, b, c], merge_seconds=5)
+    after = [(m.id, m.date, m.media_group_id) for m in (a, b, c)]
+    assert before == after
+    assert a.media_group_id is None and b.media_group_id == "B" and c.media_group_id == "C"
+
+
+def test_ties_equal_dates_cluster_in_fetch_order():
+    # Equal dates -> stable sort keeps fetch order; they cluster (gap 0 <= merge) and the
+    # first truthy id in fetch order wins.
+    a, b = Msg(1, at(5), "FIRST"), Msg(2, at(5), "SECOND")
+    mapping = _compute_time_based_group_ids([a, b], merge_seconds=5)
+    assert mapping == {1: "FIRST", 2: "FIRST"}
+
+
+def test_gap_uses_naive_datetime_subtraction_not_timestamps():
+    # The gap is a NAIVE wall-clock subtraction, exactly as the old code — NOT a timestamp
+    # diff (they DIVERGE across a DST fold, and the old behavior is the contract). Pin a
+    # DST zone and straddle the ambiguous "fall back" hour with the two folds:
+    #   a = 01:30:00 fold=0 -> the FIRST (EDT, UTC-4) occurrence of that wall-clock time;
+    #   b = 01:30:02 fold=1 -> the SECOND (EST, UTC-5) occurrence, ~1h later in REAL time.
+    # Naive subtraction (b.date - a.date) = 2s -> they cluster. A timestamp diff would be
+    # ~3602s -> they would NOT cluster. So mutating the gap to a `.timestamp()` diff turns
+    # this red, locking the naive-subtraction contract.
+    os.environ["TZ"] = "America/New_York"
+    _time.tzset()
+    try:
+        a = Msg(1, D(2024, 11, 3, 1, 30, 0, fold=0))
+        b = Msg(2, D(2024, 11, 3, 1, 30, 2, fold=1))
+        # Sanity-check the fixture itself: naive gap 2s, real (timestamp) gap ~1h.
+        assert (b.date - a.date).total_seconds() == 2
+        assert b.date.timestamp() - a.date.timestamp() > 3600
+        mapping = _compute_time_based_group_ids([a, b], merge_seconds=5)
+        synthetic = f"time_{D(2024, 11, 3, 1, 30, 0)}"
+        assert mapping == {1: synthetic, 2: synthetic}
+    finally:
+        os.environ["TZ"] = "UTC"
+        _time.tzset()
+
+
+# --------------------------------------------------------------------------- #
+# §3.11 — None-date posts: excluded from clustering, own media_group_id still applies.
+# --------------------------------------------------------------------------- #
+def test_none_date_gets_no_mapping_entry():
+    dated, nodate = Msg(1, at(0), "MG"), Msg(2, None, "MG")
+    mapping = _compute_time_based_group_ids([dated, nodate], merge_seconds=5)
+    assert 2 not in mapping  # None-date never participates
+
+
+def test_mixed_none_date_does_not_crash():
+    # The historical prod bug: a single None-date post amid naive-dated posts raised
+    # TypeError. The pure function must simply skip it.
+    msgs = [Msg(1, at(0)), Msg(2, None), Msg(3, at(2))]
+    mapping = _compute_time_based_group_ids(msgs, merge_seconds=5)  # must not raise
+    assert 2 not in mapping
+    # The two dated posts still cluster (2s apart) under a synthetic id.
+    synthetic = f"time_{at(0)}"
+    assert mapping == {1: synthetic, 3: synthetic}
+
+
+def test_fully_none_input_yields_empty_mapping():
+    # Registry §3.11: the old code clustered a fully-None tail by insertion order (only
+    # reachable via aware-date test mocks). New behavior: no clustering at all.
+    msgs = [Msg(1, None, "A"), Msg(2, None), Msg(3, None)]
+    assert _compute_time_based_group_ids(msgs, merge_seconds=5) == {}
+
+
+def test_none_date_media_group_still_assembled_downstream():
+    # None-date posts get no mapping entry but keep their own media_group_id, so a media
+    # group made of None-date members is still assembled by _create_messages_groups.
+    m1, m2 = Msg(10, None, "SHARED"), Msg(11, None, "SHARED")
+    solo = Msg(12, at(0))
+    mapping = _compute_time_based_group_ids([m1, m2, solo], merge_seconds=5)
+    groups = _create_messages_groups([m1, m2, solo], mapping)
+    shared = [g for g in groups if len(g) == 2]
+    assert len(shared) == 1
+    assert {m.id for m in shared[0]} == {10, 11}
+
+
+# --------------------------------------------------------------------------- #
+# §3.12 — naive-safe group sort: None-date groups survive [:limit] and land at the tail.
+# --------------------------------------------------------------------------- #
+def test_create_messages_groups_none_date_does_not_crash_default_path():
+    # This is the LIVE default-path 500: a None-date group used to fall back to an AWARE
+    # now() in the group sort key and blow up against the naive real dates. No mapping /
+    # time_based_merge needed — plain grouping must survive.
+    dated = Msg(1, at(0))
+    nodate = Msg(2, None)
+    groups = _create_messages_groups([dated, nodate])  # must not raise
+    # None-date group sorts as newest (float('inf')) -> first here (reverse=True).
+    assert groups[0][0].id == 2
+
+
+def _make_message(mid, text, date, media_group_id=None):
+    m = SimpleNamespace()
+    m.id = mid
+    m.date = date
+    m.text = _Str(text) if text is not None else None
+    m.caption = None
+    m.media = None
+    m.web_page = None
+    m.poll = None
+    m.service = None
+    m.forward_origin = None
+    m.reply_to_message = None
+    m.reply_to_message_id = None
+    m.sender_chat = None
+    m.from_user = None
+    m.reactions = None
+    m.views = 100
+    m.media_group_id = media_group_id
+    m.show_caption_above_media = False
+    m.chat = SimpleNamespace(id=-1001234567890, username="testchan")
+    for attr in ("photo", "video", "document", "audio", "voice",
+                 "video_note", "animation", "sticker"):
+        setattr(m, attr, None)
+    return m
+
+
+@pytest.mark.asyncio
+async def test_none_date_post_renders_and_lands_at_feed_end(monkeypatch):
+    # END-TO-END regression for the live prod 500: a None-date post in a feed of real-dated
+    # posts, WITH time_based_merge. Old code raised TypeError (naive vs aware) in BOTH the
+    # group sort key (default path) and the time-cluster sort -> HTTP 500 on the default
+    # feed path. New code renders it and, per §3.12, places it at the tail of the feed.
+    posts = [
+        _make_message(1, "OLDEST_REAL", at(0)),
+        _make_message(2, "NEWEST_REAL", at(30)),
+        _make_message(3, "NONE_DATE_POST", None),
+    ]
+
+    async def fake_get_chat(client, channel):
+        return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
+
+    async def fake_get_history(client, channel, limit=20):
+        return posts
+
+    monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
+    monkeypatch.setattr("tg_cache.cached_get_chat_history", fake_get_history, raising=False)
+    monkeypatch.setitem(rss_module.Config, "time_based_merge", True)
+
+    html = await generate_channel_html("testchan", client=SimpleNamespace(), limit=10)
+
+    assert "NONE_DATE_POST" in html
+    assert "NEWEST_REAL" in html and "OLDEST_REAL" in html
+    # None-date post renders LAST (final post sort fallback 0.0 -> tail of feed, §3.12).
+    assert html.index("NONE_DATE_POST") > html.index("OLDEST_REAL")
+    assert html.index("NONE_DATE_POST") > html.index("NEWEST_REAL")
+
+
+@pytest.mark.asyncio
+async def test_none_date_group_survives_limit_slice(monkeypatch):
+    # §3.12 retention: the group sort key gives None-date groups float('inf') so they sort
+    # as NEWEST and deterministically survive the [:limit] slice in _trim_messages_groups.
+    # This test applies REAL limit pressure (limit < number of groups) so the slice actually
+    # drops groups — otherwise reverting 'inf' to 0.0 (or any small key) would go unnoticed.
+    # With 5 dated posts + 1 None-date post and limit=3, the surviving 3 groups must be the
+    # None-date group + the 2 newest dated; the 3 oldest dated are dropped. If 'inf' becomes
+    # 0.0 the None-date group sorts OLDEST and is dropped instead -> this assertion fails.
+    posts = [_make_message(mid, f"DATED_{mid}", at(0, minute=mid)) for mid in range(1, 6)]
+    posts.append(_make_message(99, "NONE_DATE_POST", None))
+
+    async def fake_get_chat(client, channel):
+        return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
+
+    async def fake_get_history(client, channel, limit=20):
+        return posts
+
+    monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
+    monkeypatch.setattr("tg_cache.cached_get_chat_history", fake_get_history, raising=False)
+
+    html = await generate_channel_html("testchan", client=SimpleNamespace(), limit=3)
+
+    # Exactly 3 posts survive the slice; the None-date group is one of them.
+    assert html.count('class="message-body"') == 3
+    assert "NONE_DATE_POST" in html, "None-date group must survive [:limit] via the inf key"
+    # The 2 newest dated posts survive; the 3 oldest are dropped by the slice.
+    assert "DATED_5" in html and "DATED_4" in html
+    assert "DATED_1" not in html and "DATED_2" not in html and "DATED_3" not in html

tests/test_stage4_eventloop.py

@@ -17,7 +17,6 @@ Covers:
       ALL outputs — rss, html-feed, single-post html, and json — each with exactly one pass.
 """
 import re
-import copy
 import pickle
 import asyncio
 import threading
@@ -35,7 +34,7 @@ from rss_generator import (
     generate_channel_rss,
     generate_channel_html,
     _render_pipeline,
-    _create_time_based_media_groups,
+    _compute_time_based_group_ids,
     _create_messages_groups,
     _trim_messages_groups,
     _render_messages_groups,
@@ -92,7 +91,7 @@ def _co_names(func):
 # ---------------------------------------------------------------------------
 
 def test_render_functions_are_sync():
-    for fn in (_create_time_based_media_groups, _create_messages_groups,
+    for fn in (_compute_time_based_group_ids, _create_messages_groups,
                _trim_messages_groups, _render_messages_groups, _render_pipeline):
         assert not asyncio.iscoroutinefunction(fn), f"{fn.__name__} must be a plain sync function"
 
@@ -124,15 +123,27 @@ async def test_pipeline_runs_in_worker_thread(monkeypatch):
     assert seen["ident"] != main_ident, "render pipeline must run in a worker thread, not the loop thread"
 
 
-def test_deepcopy_of_pickled_message_does_not_crash():
+def test_time_clustering_does_not_mutate_pickled_message():
+    # Stage 4: _create_time_based_media_groups (which deep-copied the cached list and
+    # MUTATED media_group_id) is gone. Time-clustering is now a PURE mapping function; a
+    # pickled Message straight from the cache must come out untouched.
     from pyrogram.types import Message, Chat
     from pyrogram.enums import ChatType
-    m = Message(id=7, date=datetime.now(timezone.utc), text="hello",
-                chat=Chat(id=-1001, type=ChatType.CHANNEL, username="testchan"))
-    roundtripped = pickle.loads(pickle.dumps(m))   # mimics the pickle cache
-    clone = copy.deepcopy(roundtripped)            # what _create_time_based_media_groups does
-    assert clone.id == 7
-    assert clone.chat.username == "testchan"
+    base = datetime(2024, 1, 1, 12, 0, 0)  # naive, as kurigram emits
+    a = pickle.loads(pickle.dumps(Message(
+        id=7, date=base, text="hello", media_group_id="orig_A",
+        chat=Chat(id=-1001, type=ChatType.CHANNEL, username="testchan"))))
+    b = pickle.loads(pickle.dumps(Message(
+        id=8, date=base.replace(second=2), text="world", media_group_id=None,
+        chat=Chat(id=-1001, type=ChatType.CHANNEL, username="testchan"))))
+
+    mapping = _compute_time_based_group_ids([a, b], merge_seconds=5)
+
+    # The two adjacent posts are clustered under the first truthy id, but only via the
+    # RETURNED mapping — the input objects keep their original media_group_id.
+    assert mapping == {7: "orig_A", 8: "orig_A"}
+    assert a.media_group_id == "orig_A"
+    assert b.media_group_id is None
 
 
 # ---------------------------------------------------------------------------
@@ -142,7 +153,7 @@ def test_deepcopy_of_pickled_message_does_not_crash():
 def test_render_path_has_no_asyncio_side_effects():
     banned = {"create_task", "get_running_loop", "to_thread", "ensure_future"}
     funcs = [
-        _render_pipeline, _create_time_based_media_groups, _create_messages_groups,
+        _render_pipeline, _compute_time_based_group_ids, _create_messages_groups,
         _trim_messages_groups, _render_messages_groups,
         PostParser.process_message, PostParser._generate_html_body,
         PostParser._generate_html_media, PostParser.generate_html_footer,