diff --git a/contributors/emails/2418548+markoub@users.noreply.github.com b/contributors/emails/2418548+markoub@users.noreply.github.com new file mode 100644 index 000000000000..dab69cb41098 --- /dev/null +++ b/contributors/emails/2418548+markoub@users.noreply.github.com @@ -0,0 +1,2 @@ +markoub +# PR #51097 C16 salvage diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index bf76bd85021a..78c19d50e84c 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -17,7 +17,7 @@ import re import time from dataclasses import dataclass, field -from typing import Dict, Optional, Any, Tuple, List +from typing import Callable, Dict, Optional, Any, Tuple, List try: from slack_bolt.async_app import AsyncApp @@ -675,11 +675,18 @@ def __init__(self, config: PlatformConfig): # so a multi-workspace Socket Mode process never reuses another # tenant's display name. self._user_name_cache: Dict[Tuple[str, str], str] = {} + self._USER_NAME_CACHE_MAX = 5000 self._socket_mode_task: Optional[asyncio.Task] = None # Multi-workspace support self._team_clients: Dict[str, Any] = {} # team_id → WebClient self._team_bot_user_ids: Dict[str, str] = {} # team_id → bot_user_id - self._channel_team: Dict[str, str] = {} # channel_id → team_id + # channel_id → team_id. Grows with every channel AND every DM the bot + # sees (DM channel IDs are per-user), so it must be bounded on busy + # multi-workspace installs. Eviction is safe: entries are re-learned + # from the next event on that channel, and _get_client falls back to + # the primary client meanwhile. + self._channel_team: Dict[str, str] = {} + self._CHANNEL_TEAM_MAX = 10000 # Dedup cache: prevents duplicate bot responses when Socket Mode # reconnects redeliver events (#4777). The TTL must outlast Slack's # worst-case reconnect-redelivery gap, not just a few seconds — the @@ -688,18 +695,21 @@ def __init__(self, config: PlatformConfig): # is safe. self._dedup = MessageDeduplicator(ttl_seconds=_slack_dedup_ttl_seconds()) # Track pending approval message_ts → resolved flag to prevent - # double-clicks on approval buttons. + # double-clicks on approval buttons. Bounded: an approval prompt the + # user never clicks would otherwise leak its entry forever. self._approval_resolved: Dict[str, bool] = {} + self._APPROVAL_RESOLVED_MAX = 1000 # Same guard for clarify prompts (interactive multiple-choice # buttons); mirrors _approval_resolved. self._clarify_resolved: Dict[str, bool] = {} + self._CLARIFY_RESOLVED_MAX = 1000 # Track timestamps of messages sent by the bot so we can respond # to thread replies even without an explicit @mention. - self._bot_message_ts: set = set() + self._bot_message_ts: set[str] = set() self._BOT_TS_MAX = 5000 # cap to avoid unbounded growth # Track threads where the bot has been @mentioned — once mentioned, # respond to ALL subsequent messages in that thread automatically. - self._mentioned_threads: set = set() + self._mentioned_threads: set[str] = set() self._MENTIONED_THREADS_MAX = 5000 # Assistant thread metadata keyed by (team_id, channel_id, thread_ts). # Slack's AI Assistant lifecycle events can arrive before/alongside @@ -715,6 +725,7 @@ def __init__(self, config: PlatformConfig): # Cache for _fetch_thread_context results: cache_key → _ThreadContextCache self._thread_context_cache: Dict[str, _ThreadContextCache] = {} self._THREAD_CACHE_TTL = 60.0 + self._THREAD_CACHE_MAX = 2500 # Persistent sessions survive gateway restarts, but messages that # arrived while the gateway was DOWN never reached the session. # Track which threads have been rehydration-checked this process so @@ -724,10 +735,17 @@ def __init__(self, config: PlatformConfig): self._thread_rehydration_checked: set = set() self._THREAD_REHYDRATION_CHECKED_MAX = 5000 # Track message IDs that should get reaction lifecycle (DMs / @mentions). + # Entries are normally removed when the reaction completes, but an + # exception between add and finalize would leak them — keep it bounded. self._reacting_message_ids: set = set() + self._REACTING_MESSAGE_IDS_MAX = 5000 # Track active Assistant statuses by (team_id, channel_id, thread_ts) # so cleanup cannot clear an overlapping Slack Connect workspace. + # Entries are popped when the status clears, but statuses abandoned + # by an error path would accumulate — bound with oldest-thread-first + # eviction (key[2] is the thread ts). self._active_status_threads: Dict[Tuple[str, str, str], Dict[str, str]] = {} + self._ACTIVE_STATUS_THREADS_MAX = 1000 # Best-effort guard so automatic Slack AI thread titles are set once # per visible DM thread instead of on every reply. self._titled_assistant_threads: set = set() @@ -780,6 +798,86 @@ async def _close_workspace_clients(self) -> None: await result break + @staticmethod + def _slack_timestamp_sort_key(ts: str) -> Tuple[int, int, str]: + """Return a chronological, deterministic sort key for Slack timestamps.""" + seconds, _, fraction = str(ts).partition(".") + try: + seconds_int = int(seconds) + except ValueError: + seconds_int = 0 + try: + fraction_int = int((fraction + "000000")[:6] or "0") + except ValueError: + fraction_int = 0 + return seconds_int, fraction_int, str(ts) + + @classmethod + def _discard_oldest_slack_timestamps( + cls, timestamps: set[str], count: int + ) -> None: + """Discard the oldest Slack timestamps from a bounded tracking set.""" + if count <= 0: + return + for old_ts in sorted(timestamps, key=cls._slack_timestamp_sort_key)[:count]: + timestamps.discard(old_ts) + + def _trim_bot_message_timestamps(self) -> None: + if len(self._bot_message_ts) <= self._BOT_TS_MAX: + return + excess = len(self._bot_message_ts) - self._BOT_TS_MAX // 2 + self._discard_oldest_slack_timestamps(self._bot_message_ts, excess) + + def _trim_mentioned_threads(self) -> None: + if len(self._mentioned_threads) <= self._MENTIONED_THREADS_MAX: + return + self._discard_oldest_slack_timestamps( + self._mentioned_threads, self._MENTIONED_THREADS_MAX // 2 + ) + + @staticmethod + def _trim_oldest_dict_entries(mapping: Dict[Any, Any], max_size: int) -> None: + """Evict the oldest-inserted entries once *mapping* exceeds *max_size*. + + Python dicts preserve insertion order, so ``list(mapping)[:excess]`` + is genuinely oldest-first (unlike sets, whose iteration order is + arbitrary — see #51019). Evicts down to half the cap so eviction + runs amortized-once per max_size//2 writes, matching the sibling + tracking structures. + """ + if len(mapping) <= max_size: + return + excess = len(mapping) - max_size // 2 + for old_key in list(mapping)[:excess]: + del mapping[old_key] + + @classmethod + def _discard_oldest_by_thread_ts( + cls, entries: set, count: int, ts_getter: Callable[[Any], str] + ) -> None: + """Discard the *count* entries with the oldest embedded Slack ts. + + For bounded tracking sets whose members are keys CONTAINING a Slack + timestamp (tuples or colon-joined strings) rather than bare ts + values. Sets iterate in arbitrary order, so a plain + ``list(entries)[:count]`` can evict the most ACTIVE entry (#51019); + sort chronologically by the embedded thread ts instead. + """ + if count <= 0: + return + oldest = sorted( + entries, key=lambda e: cls._slack_timestamp_sort_key(ts_getter(e)) + )[:count] + for entry in oldest: + entries.discard(entry) + + def _remember_channel_team(self, channel_id: str, team_id: str) -> None: + """Record which workspace owns *channel_id*, bounded oldest-first.""" + if not channel_id or not team_id: + return + self._channel_team[str(channel_id)] = str(team_id) + self._trim_oldest_dict_entries(self._channel_team, self._CHANNEL_TEAM_MAX) + def _start_socket_mode_handler(self) -> None: """Start the Slack Socket Mode background task.""" if not self._app or not self._app_token: @@ -1088,6 +1186,8 @@ def _describe_slack_download_failure( _SLASH_CTX_TTL = 120.0 # seconds — response_url is valid for 30 min; # we use a much shorter TTL to avoid routing unrelated messages # as ephemeral if the command handler was slow or dropped. + _SLASH_CTX_MAX = 1000 # hard cap: TTL cleanup only runs on lookup, so + # contexts whose replies never arrive would otherwise accumulate. def _pop_slash_context( self, @@ -1861,10 +1961,7 @@ async def send( # Also register the thread root so replies-to-my-replies work if thread_ts: self._bot_message_ts.add(thread_ts) - if len(self._bot_message_ts) > self._BOT_TS_MAX: - excess = len(self._bot_message_ts) - self._BOT_TS_MAX // 2 - for old_ts in list(self._bot_message_ts)[:excess]: - self._bot_message_ts.discard(old_ts) + self._trim_bot_message_timestamps() return SendResult( success=True, @@ -2012,6 +2109,19 @@ async def send_typing(self, chat_id: str, metadata=None) -> None: "thread_ts": str(thread_ts), "team_id": str(team_id) if team_id else "", } + if len(self._active_status_threads) > self._ACTIVE_STATUS_THREADS_MAX: + # Evict abandoned statuses oldest-thread-first (key[2] is the + # thread ts) so an eviction never clears the newest status. + excess = ( + len(self._active_status_threads) + - self._ACTIVE_STATUS_THREADS_MAX // 2 + ) + oldest = sorted( + self._active_status_threads, + key=lambda k: self._slack_timestamp_sort_key(k[2]), + )[:excess] + for old_key in oldest: + self._active_status_threads.pop(old_key, None) try: _status = ( getattr(self, "_status_text", {}).get(str(chat_id)) @@ -2388,10 +2498,7 @@ def _record_uploaded_file_thread( if not thread_ts: return self._bot_message_ts.add(thread_ts) - if len(self._bot_message_ts) > self._BOT_TS_MAX: - excess = len(self._bot_message_ts) - self._BOT_TS_MAX // 2 - for old_ts in list(self._bot_message_ts)[:excess]: - self._bot_message_ts.discard(old_ts) + self._trim_bot_message_timestamps() def _is_retryable_upload_error(self, exc: Exception) -> bool: """Best-effort detection for transient Slack upload failures.""" @@ -2732,12 +2839,16 @@ async def _resolve_user_name( or user.get("name") or user_id ) - self._user_name_cache[cache_key] = name - return name except Exception as e: logger.debug("[Slack] users.info failed for %s: %s", user_id, e) - self._user_name_cache[cache_key] = user_id - return user_id + name = user_id + + self._user_name_cache[cache_key] = name + if len(self._user_name_cache) > self._USER_NAME_CACHE_MAX: + excess = len(self._user_name_cache) - self._USER_NAME_CACHE_MAX // 2 + for old_key in list(self._user_name_cache)[:excess]: + del self._user_name_cache[old_key] + return name async def _humanize_user_mentions( self, text: str, chat_id: str = "", team_id: str = "" @@ -3247,7 +3358,7 @@ def _cache_assistant_thread_metadata(self, metadata: Dict[str, str]) -> None: del self._assistant_threads[old_key] if team_id and channel_id: - self._channel_team[channel_id] = team_id + self._remember_channel_team(channel_id, team_id) def _lookup_assistant_thread_metadata( self, @@ -3393,8 +3504,11 @@ async def _set_assistant_thread_title( len(self._titled_assistant_threads) - self._TITLED_ASSISTANT_THREADS_MAX // 2 ) - for old_key in list(self._titled_assistant_threads)[:excess]: - self._titled_assistant_threads.discard(old_key) + # Keys are (team_id, channel_id, thread_ts) — evict the oldest + # threads first so recently titled threads keep their guard. + self._discard_oldest_by_thread_ts( + self._titled_assistant_threads, excess, lambda e: e[2] + ) def _seed_assistant_thread_session(self, metadata: Dict[str, str]) -> None: """Prime the session store so assistant threads get stable user scoping.""" @@ -3511,7 +3625,7 @@ async def _handle_app_home_opened( context_channel_id = self._context_channel_id(context) if team_id and channel_id: - self._channel_team[str(channel_id)] = str(team_id) + self._remember_channel_team(channel_id, team_id) metadata = { "channel_id": str(channel_id) if channel_id else "", @@ -3615,12 +3729,7 @@ def _register_mentioned_thread(self, thread_ts: str) -> None: if not thread_ts: return self._mentioned_threads.add(thread_ts) - if len(self._mentioned_threads) > self._MENTIONED_THREADS_MAX: - to_remove = list(self._mentioned_threads)[ - : self._MENTIONED_THREADS_MAX // 2 - ] - for t in to_remove: - self._mentioned_threads.discard(t) + self._trim_mentioned_threads() async def _bot_authored_thread_root( self, channel_id: str, thread_ts: str, team_id: str = "" @@ -3916,7 +4025,7 @@ async def _handle_slack_message( # Track which workspace owns this channel if team_id and channel_id: - self._channel_team[channel_id] = team_id + self._remember_channel_team(channel_id, team_id) # Determine if this is a DM or channel message channel_type = event.get("channel_type", "") @@ -4540,6 +4649,13 @@ async def _handle_slack_message( _should_react = (is_one_to_one_dm or is_mentioned) and self._reactions_enabled() if _should_react: self._reacting_message_ids.add(ts) + if len(self._reacting_message_ids) > self._REACTING_MESSAGE_IDS_MAX: + # Entries are bare Slack message ts values — evict oldest first. + self._discard_oldest_slack_timestamps( + self._reacting_message_ids, + len(self._reacting_message_ids) + - self._REACTING_MESSAGE_IDS_MAX // 2, + ) # App-context is per-turn, user-controlled Slack UI state. Surface it # with the inbound user message rather than storing it on SessionSource: @@ -4648,6 +4764,9 @@ async def send_exec_approval( msg_ts = result.get("ts", "") if msg_ts: self._approval_resolved[msg_ts] = False + self._trim_oldest_dict_entries( + self._approval_resolved, self._APPROVAL_RESOLVED_MAX + ) return SendResult(success=True, message_id=msg_ts, raw_response=result) except Exception as e: @@ -4825,6 +4944,9 @@ async def send_clarify( # Mark unresolved so the action handler's atomic-pop guard can # reject double-clicks (mirrors _approval_resolved). self._clarify_resolved[msg_ts] = False + self._trim_oldest_dict_entries( + self._clarify_resolved, self._CLARIFY_RESOLVED_MAX + ) return SendResult(success=True, message_id=msg_ts, raw_response=result) except Exception as e: @@ -5468,6 +5590,14 @@ async def _fetch_thread_context( parent_user_id=parent_user_id, messages=list(messages), ) + if len(self._thread_context_cache) > self._THREAD_CACHE_MAX: + stale_keys = [ + k + for k, v in self._thread_context_cache.items() + if now - v.fetched_at >= self._THREAD_CACHE_TTL + ] + for k in stale_keys: + del self._thread_context_cache[k] if after_ts: delta, _ = await self._format_thread_context( messages, @@ -5697,7 +5827,7 @@ async def _handle_slash_command(self, command: dict) -> None: # Track which workspace owns this channel if team_id and channel_id: - self._channel_team[channel_id] = team_id + self._remember_channel_team(channel_id, team_id) if slash_name in {"hermes", ""}: # Legacy /hermes [args] routing + free-form questions. @@ -5760,6 +5890,27 @@ async def _handle_slash_command(self, command: dict) -> None: "response_url": response_url, "ts": time.monotonic(), } + if len(self._slash_command_contexts) > self._SLASH_CTX_MAX: + # TTL cleanup normally runs on lookup, but contexts stashed + # for replies that never happen (agent error, ephemeral-only + # command) are never looked up — purge expired entries, then + # fall back to oldest-stash-first eviction if still over cap. + now_ts = time.monotonic() + for stale_key in [ + k + for k, v in self._slash_command_contexts.items() + if now_ts - v["ts"] > self._SLASH_CTX_TTL + ]: + del self._slash_command_contexts[stale_key] + if len(self._slash_command_contexts) > self._SLASH_CTX_MAX: + excess = ( + len(self._slash_command_contexts) - self._SLASH_CTX_MAX // 2 + ) + for old_key in sorted( + self._slash_command_contexts, + key=lambda k: self._slash_command_contexts[k]["ts"], + )[:excess]: + del self._slash_command_contexts[old_key] # Set the ContextVar so send() can match the correct stashed # response_url even when multiple users slash concurrently. @@ -5859,8 +6010,15 @@ def _mark_thread_rehydration_checked( len(self._thread_rehydration_checked) - self._THREAD_REHYDRATION_CHECKED_MAX // 2 ) - for old_key in list(self._thread_rehydration_checked)[:excess]: - self._thread_rehydration_checked.discard(old_key) + # Keys are "team:channel:thread_ts[:user]" — evict the oldest + # threads first. Evicting an ACTIVE thread's key would re-run its + # rehydration check and re-inject the missed delta (#51019-style + # arbitrary eviction), so never pop in set order. + self._discard_oldest_by_thread_ts( + self._thread_rehydration_checked, + excess, + lambda e: e.split(":")[2] if e.count(":") >= 2 else "", + ) def _get_thread_watermark( self, diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py index f4fdae41e6bd..bee64c9400a1 100644 --- a/tests/gateway/test_slack.py +++ b/tests/gateway/test_slack.py @@ -5685,3 +5685,255 @@ def capture_fatal(code, message, *, retryable): assert "SLACK_APP_TOKEN" in fatal_errors[0]["message"] assert "hermes gateway setup" in fatal_errors[0]["message"].lower() or ".env" in fatal_errors[0]["message"] + + +# --------------------------------------------------------------------------- +# TestThreadContextCacheBounded +# --------------------------------------------------------------------------- + + +class TestThreadContextCacheBounded: + """_thread_context_cache must evict expired entries when it exceeds + _THREAD_CACHE_MAX, symmetric with _bot_message_ts / _mentioned_threads / + _assistant_threads which all enforce their respective MAX constants.""" + + @pytest.mark.asyncio + async def test_expired_entries_evicted_when_cache_exceeds_max(self, adapter): + from plugins.platforms.slack.adapter import _ThreadContextCache + + adapter._THREAD_CACHE_MAX = 2 + + stale_ts = time.monotonic() - 120.0 # 120 s ago, past TTL of 60 s + for i in range(3): + adapter._thread_context_cache[f"C_stale:{i}:"] = _ThreadContextCache( + content=f"old {i}", fetched_at=stale_ts + ) + assert len(adapter._thread_context_cache) == 3 + + # Pre-load user name so _resolve_user_name skips the API call + adapter._user_name_cache[("", "U1")] = "Alice" + + adapter._app.client.conversations_replies = AsyncMock( + return_value={ + "messages": [{"ts": "msg-a", "user": "U1", "text": "hello"}] + } + ) + + # Fetch a fresh key — triggers cache write → eviction fires + await adapter._fetch_thread_context( + channel_id="C_fresh", thread_ts="ts-new", current_ts="ts-new" + ) + + assert len(adapter._thread_context_cache) <= adapter._THREAD_CACHE_MAX + + @pytest.mark.asyncio + async def test_fresh_entries_not_evicted(self, adapter): + from plugins.platforms.slack.adapter import _ThreadContextCache + + adapter._THREAD_CACHE_MAX = 2 + + fresh_ts = time.monotonic() + for i in range(2): + adapter._thread_context_cache[f"C_fresh:{i}:"] = _ThreadContextCache( + content=f"fresh {i}", fetched_at=fresh_ts + ) + + adapter._user_name_cache[("", "U2")] = "Bob" + adapter._app.client.conversations_replies = AsyncMock( + return_value={ + "messages": [{"ts": "msg-b", "user": "U2", "text": "hi"}] + } + ) + + await adapter._fetch_thread_context( + channel_id="C_extra", thread_ts="ts-extra", current_ts="ts-extra" + ) + + # Fresh entries must survive — only stale entries are evicted + for i in range(2): + assert f"C_fresh:{i}:" in adapter._thread_context_cache + + +# --------------------------------------------------------------------------- +# TestTrackingStructureBounds (cluster C16 — unbounded/mis-evicting caches) +# --------------------------------------------------------------------------- + + +class TestTrackingStructureBounds: + """Every per-message/per-user tracking structure must be bounded, and + eviction must remove the OLDEST entries — arbitrary (set-order) eviction + can silently drop the most active thread (#51019).""" + + def test_user_name_cache_cap_holds_under_churn(self, adapter): + adapter._USER_NAME_CACHE_MAX = 10 + # Simulate the post-resolution write + trim path directly. + for i in range(50): + adapter._user_name_cache[("T1", f"U{i}")] = f"user{i}" + if len(adapter._user_name_cache) > adapter._USER_NAME_CACHE_MAX: + excess = ( + len(adapter._user_name_cache) + - adapter._USER_NAME_CACHE_MAX // 2 + ) + for old_key in list(adapter._user_name_cache)[:excess]: + del adapter._user_name_cache[old_key] + assert len(adapter._user_name_cache) <= adapter._USER_NAME_CACHE_MAX + # Newest entry survives; oldest was evicted. + assert ("T1", "U49") in adapter._user_name_cache + assert ("T1", "U0") not in adapter._user_name_cache + + @pytest.mark.asyncio + async def test_user_name_cache_bounded_through_resolve(self, adapter): + """End-to-end: _resolve_user_name enforces the cap.""" + adapter._USER_NAME_CACHE_MAX = 4 + adapter._app.client.users_info = AsyncMock( + side_effect=lambda user: { + "user": {"profile": {"display_name": f"name-{user}"}} + } + ) + for i in range(10): + await adapter._resolve_user_name(f"U{i}") + assert len(adapter._user_name_cache) <= adapter._USER_NAME_CACHE_MAX + assert ("", "U9") in adapter._user_name_cache + + def test_trim_oldest_dict_entries_evicts_insertion_order(self, adapter): + d = {f"k{i}": i for i in range(6)} + adapter._trim_oldest_dict_entries(d, 5) + # 6 > 5 → excess = 6 - 2 = 4 → oldest four evicted + assert "k0" not in d and "k3" not in d + assert "k4" in d and "k5" in d + + def test_approval_and_clarify_resolved_bounded(self, adapter): + adapter._APPROVAL_RESOLVED_MAX = 4 + adapter._CLARIFY_RESOLVED_MAX = 4 + for i in range(10): + adapter._approval_resolved[f"{1000 + i}.0"] = False + adapter._trim_oldest_dict_entries( + adapter._approval_resolved, adapter._APPROVAL_RESOLVED_MAX + ) + adapter._clarify_resolved[f"{1000 + i}.0"] = False + adapter._trim_oldest_dict_entries( + adapter._clarify_resolved, adapter._CLARIFY_RESOLVED_MAX + ) + assert len(adapter._approval_resolved) <= 4 + assert len(adapter._clarify_resolved) <= 4 + # The most recent prompt (the one the user is about to click) survives. + assert "1009.0" in adapter._approval_resolved + assert "1009.0" in adapter._clarify_resolved + + def test_titled_assistant_threads_evicts_oldest_thread_first(self, adapter): + adapter._TITLED_ASSISTANT_THREADS_MAX = 4 + keys = [ + ("T1", "D1", "1000.000002"), + ("T1", "D1", "999.999999"), + ("T1", "D1", "1000.000004"), + ("T1", "D1", "1000.000001"), + ("T1", "D1", "1000.000003"), + ] + adapter._titled_assistant_threads.update(keys) + excess = ( + len(adapter._titled_assistant_threads) + - adapter._TITLED_ASSISTANT_THREADS_MAX // 2 + ) + adapter._discard_oldest_by_thread_ts( + adapter._titled_assistant_threads, excess, lambda e: e[2] + ) + assert adapter._titled_assistant_threads == { + ("T1", "D1", "1000.000003"), + ("T1", "D1", "1000.000004"), + } + + def test_rehydration_checked_evicts_oldest_thread_first(self, adapter): + """Regression shape for #51019: the ACTIVE (newest) thread key must + survive eviction pressure so its rehydration check does not re-run.""" + adapter._THREAD_REHYDRATION_CHECKED_MAX = 4 + for ts in [ + "1000.000002", + "999.999999", + "1000.000004", + "1000.000001", + "1000.000003", + ]: + adapter._mark_thread_rehydration_checked("C1", ts, "U1", "T1") + assert adapter._thread_rehydration_checked == { + "T1:C1:1000.000003", + "T1:C1:1000.000004", + } + + def test_active_status_threads_evicts_oldest_and_keeps_newest(self, adapter): + adapter._ACTIVE_STATUS_THREADS_MAX = 4 + adapter._app.client.assistant_threads_setStatus = AsyncMock() + for i, ts in enumerate( + ["1000.000002", "999.999999", "1000.000004", "1000.000001", "1000.000003"] + ): + adapter._active_status_threads[("T1", f"D{i}", ts)] = { + "thread_ts": ts, + "team_id": "T1", + } + # Simulate the overflow trim from send_typing_indicator. + excess = ( + len(adapter._active_status_threads) + - adapter._ACTIVE_STATUS_THREADS_MAX // 2 + ) + oldest = sorted( + adapter._active_status_threads, + key=lambda k: adapter._slack_timestamp_sort_key(k[2]), + )[:excess] + for old_key in oldest: + adapter._active_status_threads.pop(old_key, None) + remaining_ts = {k[2] for k in adapter._active_status_threads} + assert remaining_ts == {"1000.000003", "1000.000004"} + + def test_reacting_message_ids_evicts_oldest_timestamps(self, adapter): + adapter._REACTING_MESSAGE_IDS_MAX = 4 + adapter._reacting_message_ids.update( + {"1000.000002", "999.999999", "1000.000004", "1000.000001", "1000.000003"} + ) + adapter._discard_oldest_slack_timestamps( + adapter._reacting_message_ids, + len(adapter._reacting_message_ids) + - adapter._REACTING_MESSAGE_IDS_MAX // 2, + ) + assert adapter._reacting_message_ids == {"1000.000003", "1000.000004"} + + def test_channel_team_bounded_via_remember_helper(self, adapter): + adapter._CHANNEL_TEAM_MAX = 4 + for i in range(10): + adapter._remember_channel_team(f"C{i}", "T1") + assert len(adapter._channel_team) <= adapter._CHANNEL_TEAM_MAX + # Most recently seen channel survives. + assert "C9" in adapter._channel_team + assert "C0" not in adapter._channel_team + + @pytest.mark.asyncio + async def test_slash_command_contexts_bounded(self, adapter): + adapter._SLASH_CTX_MAX = 4 + adapter.handle_hermes_command = AsyncMock(return_value=None) + for i in range(10): + command = { + "command": "/hermes", + "text": "/status", + "user_id": f"U{i}", + "channel_id": "C1", + "team_id": "T1", + "response_url": f"https://hooks.slack.com/commands/{i}", + } + respond = AsyncMock() # noqa: F841 — kept for shape clarity + await adapter._handle_slash_command(command) + assert len(adapter._slash_command_contexts) <= adapter._SLASH_CTX_MAX + # Newest stash survives. + assert ("C1", "U9") in adapter._slash_command_contexts + + def test_bot_message_ts_active_thread_survives_churn(self, adapter): + """#51019 regression: an active thread registered early must survive + heavy churn of NEWER one-off messages... it will eventually age out, + but eviction must never remove the newest entries while older ones + remain (no arbitrary set-order pops).""" + adapter._BOT_TS_MAX = 100 + for i in range(500): + adapter._bot_message_ts.add(f"{2000 + i}.000000") + adapter._trim_bot_message_timestamps() + assert len(adapter._bot_message_ts) <= adapter._BOT_TS_MAX + # The newest 50 timestamps must all be present (oldest-first eviction + # can never remove a newer entry while an older one remains). + for i in range(450, 500): + assert f"{2000 + i}.000000" in adapter._bot_message_ts diff --git a/tests/gateway/test_slack_approval_buttons.py b/tests/gateway/test_slack_approval_buttons.py index fa0256c4f631..8553caf829bd 100644 --- a/tests/gateway/test_slack_approval_buttons.py +++ b/tests/gateway/test_slack_approval_buttons.py @@ -1000,18 +1000,21 @@ async def test_send_tracks_bot_message_ts(self): # Thread root should also be tracked assert "8000.0" in adapter._bot_message_ts - @pytest.mark.asyncio - async def test_bot_message_ts_cap(self): - """Verify memory is bounded when many messages are sent.""" + def test_bot_message_ts_cap_evicts_oldest_timestamps(self): + """Bot thread tracking evicts the oldest Slack timestamps first.""" adapter = _make_adapter() - adapter._BOT_TS_MAX = 10 # low cap for testing - mock_client = adapter._team_clients["T1"] + adapter._BOT_TS_MAX = 4 - for i in range(20): - mock_client.chat_postMessage = AsyncMock(return_value={"ts": f"{i}.0"}) - await adapter.send(chat_id="C1", content=f"msg {i}") + for ts in [ + "1000.000002", + "999.999999", + "1000.000004", + "1000.000001", + "1000.000003", + ]: + adapter._record_uploaded_file_thread("C1", ts) - assert len(adapter._bot_message_ts) <= 10 + assert adapter._bot_message_ts == {"1000.000003", "1000.000004"} def test_mentioned_threads_populated_on_mention(self): """When bot is @mentioned in a thread, that thread is tracked.""" @@ -1020,14 +1023,24 @@ def test_mentioned_threads_populated_on_mention(self): adapter._mentioned_threads.add("1000.0") assert "1000.0" in adapter._mentioned_threads - def test_mentioned_threads_cap(self): - """Verify _mentioned_threads is bounded.""" + def test_mentioned_threads_cap_evicts_oldest_timestamps(self): + """Mentioned-thread tracking evicts the oldest Slack timestamps first.""" adapter = _make_adapter() - adapter._MENTIONED_THREADS_MAX = 10 - for i in range(15): - adapter._mentioned_threads.add(f"{i}.0") - if len(adapter._mentioned_threads) > adapter._MENTIONED_THREADS_MAX: - to_remove = list(adapter._mentioned_threads)[:adapter._MENTIONED_THREADS_MAX // 2] - for t in to_remove: - adapter._mentioned_threads.discard(t) - assert len(adapter._mentioned_threads) <= 10 + adapter._MENTIONED_THREADS_MAX = 4 + adapter._mentioned_threads.update( + { + "1000.000002", + "999.999999", + "1000.000004", + "1000.000001", + "1000.000003", + } + ) + + adapter._trim_mentioned_threads() + + assert adapter._mentioned_threads == { + "1000.000002", + "1000.000003", + "1000.000004", + }