fix(slack): backport stream pacing and honor shared cooldowns

contributors/emails/jake@withvariable.com

@@ -0,0 +1 @@
+jakeoliver-withvariable

contributors/emails/josh@withvariable.com

@@ -0,0 +1 @@
+JoshSnider

contributors/emails/tars@withvariable.com

@@ -0,0 +1 @@
+tars-withvariable

contributors/emails/zikaclawbot@gmail.com

@@ -0,0 +1 @@
+zikaclawbot

gateway/stream_consumer.py

@@ -999,11 +999,11 @@ async def run(self) -> None:
                     should_edit = should_edit or (
                         (elapsed >= self._current_edit_interval
                             and self._accumulated)
-                        # buffer_threshold is intentionally codepoint-based:
-                        # it's a debounce heuristic ("send updates roughly
-                        # every N visible characters"), not a platform-limit
-                        # check. _len_fn is reserved for overflow detection.
-                        or len(self._accumulated) >= self.cfg.buffer_threshold
+                        # The size threshold may accelerate the first output,
+                        # never repeated updates: the cumulative buffer stays
+                        # large, even during flood backoff.
+                        or (self._last_edit_time == 0.0
+                            and len(self._accumulated) >= self.cfg.buffer_threshold)
                     )
 
                 current_update_visible = False
@@ -2602,9 +2602,11 @@ async def _send_or_edit(
                         # edits after _MAX_FLOOD_STRIKES consecutive failures.
                         if self._is_flood_error(result):
                             self._flood_strikes += 1
-                            self._current_edit_interval = min(
-                                self._current_edit_interval * 2, 10.0,
-                            )
+                            backoff = min(self._current_edit_interval * 2, 10.0)
+                            retry_after = getattr(result, "retry_after", None)
+                            if isinstance(retry_after, (int, float)) and retry_after > 0:
+                                backoff = max(backoff, min(float(retry_after), 30.0))
+                            self._current_edit_interval = backoff
                             logger.debug(
                                 "Flood control on edit (strike %d/%d), "
                                 "backoff interval → %.1fs",

plugins/platforms/slack/adapter.py

@@ -22,6 +22,8 @@
 
 import aiohttp
 
+from agent.retry_utils import parse_retry_after_seconds
+
 try:
     from slack_bolt.async_app import AsyncApp
     from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
@@ -954,6 +956,10 @@ def __init__(self, config: PlatformConfig):
         self._socket_mode_task: Optional[asyncio.Task] = None
         # Multi-workspace support
         self._team_clients: Dict[str, Any] = {}  # team_id → WebClient
+        # chat.update's quota is shared by all threads in a workspace. Clients
+        # are workspace-scoped; serialize edits and share any server cooldown.
+        self._edit_locks: Dict[Any, asyncio.Lock] = {}
+        self._edit_next_at: Dict[Any, float] = {}
         self._team_bot_user_ids: Dict[str, str] = {}  # team_id → bot_user_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
@@ -2971,6 +2977,34 @@ async def send_or_update_status(
             self._status_message_ids[key] = str(result.message_id)
         return result
 
+    @staticmethod
+    def _retry_after_from_exc(e: BaseException) -> Optional[float]:
+        """Retry-After from an SDK error response (upstream Slack helper)."""
+        return parse_retry_after_seconds(getattr(getattr(e, "response", None), "headers", None))
+
+    async def _chat_update(self, client: Any, **kwargs: Any) -> Any:
+        """Pace chat.update across this workspace, including final edits.
+
+        1.25s leaves a little headroom under Slack's documented 50/minute
+        tier. A server-requested cooldown always takes precedence.
+        """
+        lock = self._edit_locks.setdefault(client, asyncio.Lock())
+        async with lock:
+            delay = self._edit_next_at.get(client, 0.0) - time.monotonic()
+            if delay > 0:
+                await asyncio.sleep(delay)
+            try:
+                return await client.chat_update(**kwargs)
+            except Exception as exc:
+                retry_after = self._retry_after_from_exc(exc)
+                if getattr(getattr(exc, "response", None), "status_code", None) == 429:
+                    self._edit_next_at[client] = time.monotonic() + (retry_after or 1.25)
+                raise
+            finally:
+                self._edit_next_at[client] = max(
+                    self._edit_next_at.get(client, 0.0), time.monotonic() + 1.25,
+                )
+
     async def edit_message(
         self,
         chat_id: str,
@@ -3011,9 +3045,10 @@ async def edit_message(
                 if blocks:
                     update_kwargs["blocks"] = blocks
             try:
-                await self._get_client(
-                    chat_id, team_id=self._metadata_team_id(metadata)
-                ).chat_update(**update_kwargs)
+                await self._chat_update(
+                    self._get_client(chat_id, team_id=self._metadata_team_id(metadata)),
+                    **update_kwargs,
+                )
             except Exception as e:
                 if update_kwargs.get("blocks") and self._is_block_payload_rejection(e):
                     retry_kwargs = dict(update_kwargs)
@@ -3025,9 +3060,10 @@ async def edit_message(
                         "[Slack] Block Kit payload rejected; retrying edit without blocks: %s",
                         e,
                     )
-                    await self._get_client(
-                        chat_id, team_id=self._metadata_team_id(metadata)
-                    ).chat_update(**retry_kwargs)
+                    await self._chat_update(
+                        self._get_client(chat_id, team_id=self._metadata_team_id(metadata)),
+                        **retry_kwargs,
+                    )
                 else:
                     raise
             if finalize:
@@ -3080,7 +3116,13 @@ async def edit_message(
                 e,
                 exc_info=True,
             )
-            return SendResult(success=False, error=str(e))
+            rate_limited = getattr(getattr(e, "response", None), "status_code", None) == 429
+            return SendResult(
+                success=False,
+                error=f"rate_limited: {e}" if rate_limited else str(e),
+                retryable=self._is_retryable_upload_error(e),
+                retry_after=self._retry_after_from_exc(e),
+            )
 
     async def delete_message(self, chat_id: str, message_id: str) -> bool:
         """Delete a Slack message previously sent by this bot.

tests/gateway/test_slack_edit_pacing.py

@@ -0,0 +1,163 @@
+"""Slack edits share a workspace budget and preserve cooldown/recovery."""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from slack_sdk.errors import SlackApiError
+from slack_sdk.web.async_client import AsyncWebClient
+from slack_sdk.web.async_slack_response import AsyncSlackResponse
+
+from gateway.config import PlatformConfig
+from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
+from plugins.platforms.slack import adapter as slack_module
+from plugins.platforms.slack.adapter import SlackAdapter
+
+
+def slack_error(status, headers):
+    return SlackApiError(
+        "Slack refused the edit",
+        AsyncSlackResponse(
+            client=None,
+            http_verb="POST",
+            api_url="https://slack.com/api/chat.update",
+            req_args={},
+            data={
+                "ok": False,
+                "error": "ratelimited" if status == 429 else "cant_update_message",
+            },
+            headers=headers,
+            status_code=status,
+        ),
+    )
+
+
+@pytest.fixture
+def setup(monkeypatch):
+    clock = SimpleNamespace(now=100.0)
+    sleeps = []
+
+    async def sleep(delay):
+        sleeps.append(delay)
+        clock.now += delay
+        await asyncio.sleep(0)
+
+    monkeypatch.setattr(
+        slack_module, "time", SimpleNamespace(monotonic=lambda: clock.now)
+    )
+    monkeypatch.setattr(
+        slack_module, "asyncio", SimpleNamespace(Lock=asyncio.Lock, sleep=sleep)
+    )
+    adapter = SlackAdapter(PlatformConfig(enabled=True, token="xoxb-test"))
+    adapter._app = MagicMock()
+    adapter._clear_thread_status_quietly = AsyncMock()
+    client = MagicMock(spec=AsyncWebClient)
+    client.chat_update = AsyncMock(return_value={"ok": True})
+    adapter._app.client = client
+    return adapter, client, clock, sleeps
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "headers,expected",
+    [
+        ({"Retry-After": "45"}, 45.0),
+        ({"retry-after": "9"}, 9.0),
+        ({"Retry-After": "invalid"}, None),
+    ],
+)
+async def test_edit_preserves_slack_cooldown(setup, headers, expected):
+    adapter, client, _, _ = setup
+    client.chat_update.side_effect = slack_error(429, headers)
+    result = await adapter.edit_message("C1", "m1", "streaming text")
+    assert not result.success
+    assert result.retryable is True
+    assert result.retry_after == expected
+    assert "rate" in result.error
+
+
+@pytest.mark.asyncio
+async def test_concurrent_threads_share_budget_and_workspaces_are_independent(setup):
+    adapter, client, clock, sleeps = setup
+    updates = []
+
+    async def update(**kwargs):
+        updates.append((clock.now, kwargs["ts"]))
+        return {"ok": True}
+
+    client.chat_update.side_effect = update
+    results = await asyncio.gather(
+        adapter.edit_message("C1", "thread-1", "one"),
+        adapter.edit_message("C2", "thread-2", "two"),
+        adapter.edit_message("C1", "thread-3", "three", finalize=True),
+    )
+    assert all(result.success for result in results)
+    assert [when for when, _ in updates] == [100.0, 101.25, 102.5]
+    other = MagicMock(spec=AsyncWebClient)
+    other.chat_update = AsyncMock(return_value={"ok": True})
+    adapter._team_clients["T2"] = other
+    adapter._channel_team["C3"] = "T2"
+    assert (await adapter.edit_message("C3", "other", "other workspace")).success
+    assert sleeps == [1.25, 1.25]
+
+
+@pytest.mark.asyncio
+async def test_cooldown_blocks_other_threads_and_final_edits_until_full_retry_after(
+    setup,
+):
+    adapter, client, clock, sleeps = setup
+    client.chat_update.side_effect = [
+        slack_error(429, {"Retry-After": "45"}),
+        {"ok": True},
+    ]
+    assert not (await adapter.edit_message("C1", "m1", "preview")).success
+    result = await adapter.edit_message("C2", "m2", "completed answer", finalize=True)
+    assert result.success
+    assert sleeps == [45.0]  # never truncate Slack's cooldown to the consumer's 30s cap
+    assert clock.now == 145.0
+
+
+@pytest.mark.asyncio
+async def test_consumer_recovers_after_real_slack_429_without_fallback(
+    setup, monkeypatch
+):
+    from gateway import stream_consumer
+
+    adapter, client, clock, sleeps = setup
+    monkeypatch.setattr(
+        stream_consumer, "time", SimpleNamespace(monotonic=lambda: clock.now)
+    )
+    client.chat_update.side_effect = [
+        slack_error(429, {"Retry-After": "9"}),
+        {"ok": True},
+        {"ok": True},
+    ]
+    client.chat_postMessage = AsyncMock(return_value={"ok": True, "ts": "preview"})
+    consumer = GatewayStreamConsumer(
+        adapter, "C1", StreamConsumerConfig(edit_interval=0.8)
+    )
+    assert await consumer._send_or_edit("start")
+    assert not await consumer._send_or_edit("start plus more")
+    assert consumer._current_edit_interval == 9.0
+    clock.now += 9.0
+    assert await consumer._send_or_edit("start plus more and recovered")
+    assert await consumer._send_or_edit(
+        "start plus more and recovered final", finalize=True
+    )
+    assert consumer._edit_supported
+    assert not consumer._fallback_final_send
+    assert consumer._flood_strikes == 0
+    assert client.chat_postMessage.await_count == 1
+    assert client.chat_update.await_count == 3
+    assert sleeps == [1.25]
+
+
+@pytest.mark.asyncio
+async def test_permanent_edit_errors_remain_permanent(setup):
+    adapter, client, _, _ = setup
+    client.chat_update.side_effect = slack_error(403, {})
+    result = await adapter.edit_message("C1", "m1", "text")
+    assert not result.success
+    assert not result.retryable
+    assert result.retry_after is None

tests/gateway/test_stream_consumer_throttle.py

@@ -0,0 +1,83 @@
+"""Intermediate updates obey the interval even after the cumulative buffer fills."""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway import stream_consumer
+from gateway.platforms.base import BasePlatformAdapter, SendResult
+from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("transport,flood", [("edit", False), ("edit", True), ("draft", False)])
+@pytest.mark.parametrize("boundary", [False, True], ids=["done", "segment"])
+async def test_run_throttles_cumulative_updates_but_not_finalization(monkeypatch, transport, flood, boundary):
+    clock = SimpleNamespace(now=0.1)
+    monkeypatch.setattr(stream_consumer, "time", SimpleNamespace(monotonic=lambda: clock.now))
+    adapter = MagicMock(spec=BasePlatformAdapter)
+    adapter.MAX_MESSAGE_LENGTH = 4096
+    adapter.message_len_fn_for_chat.return_value = len
+    adapter.max_message_length_for_chat.return_value = 4096
+    adapter.streaming_overflow_limit.return_value = 4096
+    adapter.supports_draft_streaming.return_value = True
+    adapter.prefers_fresh_final_streaming.return_value = False
+    updates = []
+    finals = []
+
+    async def send(*, content, metadata=None, **kwargs):
+        (finals if metadata and metadata.get("notify") else updates).append((clock.now, content))
+        return SendResult(success=True, message_id="preview")
+
+    async def edit(*, content, finalize=False, **kwargs):
+        (finals if finalize else updates).append((clock.now, content))
+        if flood and not finalize and len(updates) == 2:
+            return SendResult(success=False, error="flood_control:6", retry_after=6)
+        return SendResult(success=True, message_id="preview")
+
+    async def draft(*, content, **kwargs):
+        updates.append((clock.now, content))
+        return SendResult(success=True)
+
+    adapter.send = AsyncMock(side_effect=send)
+    adapter.edit_message = AsyncMock(side_effect=edit)
+    adapter.send_draft = AsyncMock(side_effect=draft)
+    consumer = GatewayStreamConsumer(adapter, "chat", StreamConsumerConfig(
+        edit_interval=1.0, buffer_threshold=32, cursor=" ▉", transport=transport,
+    ))
+    text = "A" * 64
+    consumer.on_delta(text)
+    times = iter([0.15, 0.2, 1.2, 1.3, 2.3, 3.3, 7.3, 7.35])
+    final_text = None
+
+    async def next_tick(delay):
+        nonlocal text, final_text
+        if final_text is not None:
+            consumer.finish()
+            return
+        clock.now = next(times)
+        text += " more"
+        consumer.on_delta(" more")
+        if clock.now == 7.35:
+            if boundary:
+                final_text = text
+                consumer.on_segment_break()
+            else:
+                final_text = text + " authoritative final"
+                consumer.finish(final_text)
+
+    # Drive real queue drains and transport calls without wall-clock races.
+    monkeypatch.setattr(stream_consumer, "asyncio", SimpleNamespace(
+        sleep=next_tick, CancelledError=asyncio.CancelledError,
+    ))
+    await consumer.run()
+
+    expected_times = [0.1, 1.2, 7.3] if flood else [0.1, 1.2, 2.3, 3.3, 7.3]
+    assert [when for when, _ in updates] == expected_times
+    assert finals == [(7.35, final_text)]
+    assert all(len(content) >= consumer.cfg.buffer_threshold for _, content in updates)
+    if not boundary:
+        assert consumer.final_content_delivered
+        assert consumer.delivered_final_matches(final_text) is True