feat: mini replay improvements

application/api/events/routes.py

@@ -103,33 +103,18 @@ def _allow_replay(
 ) -> bool:
     """Per-user sliding-window snapshot-replay budget.
 
-    Fails open on Redis errors or when the budget is disabled. Empty-backlog
-    no-cursor connects skip INCR so dev double-mounts don't trip 429.
+    Fails open on Redis errors or when the budget is disabled. No-cursor
+    connects never consume budget: fresh sessions start live and do no
+    snapshot work, so counting them only starved real reconnects (a burst
+    of fresh tabs could 429 a user's cursor-bearing reconnect).
     """
     budget = int(settings.EVENTS_REPLAY_BUDGET_REQUESTS_PER_WINDOW)
     if budget <= 0:
         return True
     if redis_client is None:
         return True
-
-    # Cheap pre-check: only INCR when we might actually replay. XLEN
-    # is one Redis op; the alternative (INCR every connect) is two
-    # ops AND wrongly counts no-op probes. The check is conservative:
-    # if ``last_event_id`` is set we always INCR, even if the cursor
-    # has already overtaken the latest entry — that case is rare and
-    # short-lived, and probing further would mean a redundant XRANGE.
     if last_event_id is None:
-        try:
-            if int(redis_client.xlen(stream_key(user_id))) == 0:
-                return True
-        except Exception:
-            # XLEN probe failed; fall through to the INCR path so a
-            # transient Redis hiccup can't bypass the budget.
-            logger.debug(
-                "XLEN probe failed for replay budget check user=%s; "
-                "proceeding to INCR",
-                user_id,
-            )
+        return True
 
     window = max(1, int(settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS))
     key = replay_budget_key(user_id)
@@ -154,7 +139,7 @@ def _normalize_last_event_id(raw: Optional[str]) -> Optional[str]:
 
     Returns the value unchanged when it parses as a Redis Streams id,
     otherwise ``None`` — callers treat ``None`` as "client has nothing"
-    and replay from the start of the retained window. Invalid ids would
+    and start the stream live (no snapshot). Invalid ids would
     otherwise pass straight to XRANGE and surface as a quiet replay
     failure plus broken truncation detection.
     """
@@ -166,6 +151,21 @@ def _normalize_last_event_id(raw: Optional[str]) -> Optional[str]:
     return raw
 
 
+def _replay_floor_id() -> Optional[str]:
+    """Oldest stream id the snapshot replay may reach, or ``None``.
+
+    Streams ids are millisecond timestamps, so an age ceiling maps
+    directly to an id floor. MAXLEN caps the stream by count, not time —
+    for a low-traffic user 1000 entries can span weeks, and shipping
+    that on reconnect helps no one.
+    """
+    max_age_hours = int(settings.EVENTS_REPLAY_MAX_AGE_HOURS)
+    if max_age_hours <= 0:
+        return None
+    floor_ms = int(time.time() * 1000) - max_age_hours * 3600 * 1000
+    return f"{floor_ms}-0"
+
+
 def _replay_backlog(
     redis_client, user_id: str, last_event_id: Optional[str], max_count: int
 ) -> Iterator[tuple[str, str]]:
@@ -175,8 +175,16 @@ def _replay_backlog(
     Parse failures are skipped; the Streams id is injected into the
     envelope so replay matches live-tail shape.
     """
-    # Exclusive start: '(<id>' skips the already-delivered entry.
-    start = f"({last_event_id}" if last_event_id else "-"
+    floor = _replay_floor_id()
+    if last_event_id is None:
+        start = floor or "-"
+    elif floor and stream_id_compare(last_event_id, floor) < 0:
+        # Cursor is past the age ceiling: clamp to the floor (inclusive).
+        # The caller emits ``backlog.truncated`` for this case.
+        start = floor
+    else:
+        # Exclusive start: '(<id>' skips the already-delivered entry.
+        start = f"({last_event_id}"
     try:
         entries = redis_client.xrange(
             stream_key(user_id), min=start, max="+", count=max_count
@@ -385,15 +393,34 @@ def stream_events() -> Response:
                 try:
                     if redis_client is None:
                         return
+                    if last_event_id is None:
+                        # Fresh session: start live. A no-cursor connect has
+                        # no state to catch up on — replaying the whole
+                        # retained window shipped weeks-old entries on every
+                        # tab-open (MAXLEN caps by count, not age). Clients
+                        # that had state present a cursor; the malformed-
+                        # cursor case already got its truncation notice
+                        # before the subscribe loop.
+                        return
                     oldest = _oldest_retained_id(redis_client, user_id)
-                    if (
-                        last_event_id
-                        and oldest
-                        and stream_id_compare(last_event_id, oldest) < 0
+                    floor = _replay_floor_id()
+                    # The snapshot can't reach past whichever is newer:
+                    # the MAXLEN'd window edge or the age floor.
+                    effective_oldest = oldest
+                    if floor and (
+                        effective_oldest is None
+                        or stream_id_compare(floor, effective_oldest) > 0
                     ):
-                        # The Last-Event-ID has slid off the MAXLEN window.
-                        # Tell the client so it can fetch full state.
-                        replay_lines.append(_truncation_notice_line(oldest))
+                        effective_oldest = floor
+                    if (
+                        effective_oldest
+                        and stream_id_compare(last_event_id, effective_oldest) < 0
+                    ):
+                        # The Last-Event-ID has slid off the replayable
+                        # window. Tell the client so it can fetch full state.
+                        replay_lines.append(
+                            _truncation_notice_line(effective_oldest)
+                        )
                     replay_cap = int(settings.EVENTS_REPLAY_MAX_PER_REQUEST)
                     for entry_id, sse_line in _replay_backlog(
                         redis_client, user_id, last_event_id, replay_cap

application/asgi.py

@@ -10,9 +10,10 @@ from starlette.routing import Mount
 
 from application.api.async_sse import async_sse_routes
 from application.app import app as flask_app
+from application.core.settings import settings
 from application.mcp_server import mcp
 
-_WSGI_THREADPOOL = 32
+_WSGI_THREADPOOL = int(settings.WSGI_THREADPOOL_WORKERS)
 
 mcp_app = mcp.http_app(path="/")
 

application/core/settings.py

@@ -272,6 +272,13 @@ class Settings(BaseSettings):
     # --graceful-timeout). Keep below the gunicorn --timeout (180) watchdog.
     # Used by gunicorn_worker.BoundedDrainUvicornWorker.
     GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS: int = 30
+    # a2wsgi thread-pool size per worker process — the hard ceiling on
+    # concurrent Flask requests (SSE holders + in-flight chats included).
+    # Threads parked on idle SSE reads are nearly free; exhaustion is
+    # not (every Flask request on that worker queues behind LLM
+    # generations). Sized so saturation needs hundreds of concurrent
+    # holders per worker, not 32.
+    WSGI_THREADPOOL_WORKERS: int = 96
     # SSE keepalive comment cadence. Must sit under Cloudflare's 100s idle
     # close and iOS Safari's ~60s — 15s gives generous headroom.
     SSE_KEEPALIVE_SECONDS: int = 15
@@ -288,6 +295,12 @@ class Settings(BaseSettings):
     # connection cap above and the windowed budget below, total
     # enumeration throughput is bounded.
     EVENTS_REPLAY_MAX_PER_REQUEST: int = 200
+    # Age ceiling on snapshot replay. MAXLEN caps the stream by COUNT,
+    # so a low-traffic user's window can reach weeks back — replaying
+    # month-old ingest ticks on reconnect helps no one. Entries older
+    # than this never replay (the client gets ``backlog.truncated`` and
+    # refetches state instead). 0 disables the floor.
+    EVENTS_REPLAY_MAX_AGE_HOURS: int = 48
     # Sliding-window cap on snapshot replays per user. Once the budget
     # is exhausted the route returns HTTP 429 with the cursor pinned;
     # the client backs off and retries after the window rolls over.

tests/api/test_events_routes.py

@@ -450,11 +450,10 @@ class TestReplayRateLimit:
             mock_settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS = 60
             assert _allow_replay(None, "alice", "1735682400000-0") is True
 
-    def test_allow_replay_skips_incr_when_no_cursor_and_empty_backlog(self):
-        """Fresh client with no cursor and an empty user stream cannot
-        do snapshot work — INCR'ing the counter would needlessly
-        burn budget. Catches the React-StrictMode dev-burst case where
-        double-mounted components would otherwise 429 in 5 connects.
+    def test_allow_replay_never_consumes_budget_without_cursor(self):
+        """No cursor ⇒ no replay work ⇒ no budget. Fresh sessions start
+        live and must never 429 on the replay budget, no matter how many
+        tabs open at once (30 fresh connects/min used to exhaust it).
         """
         from application.api.events.routes import _allow_replay
 
@@ -462,31 +461,13 @@ class TestReplayRateLimit:
             mock_settings.EVENTS_REPLAY_BUDGET_REQUESTS_PER_WINDOW = 3
             mock_settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS = 60
             redis = MagicMock()
-            redis.xlen.return_value = 0
+            redis.xlen.return_value = 42
 
-            # 5 connects in a row, all with no cursor — none consume
-            # budget because the backlog is empty.
-            for _ in range(5):
+            for _ in range(50):
                 assert _allow_replay(redis, "alice", None) is True
 
-            redis.xlen.assert_called()
             redis.incr.assert_not_called()
-
-    def test_allow_replay_incrs_when_no_cursor_but_backlog_present(self):
-        """A no-cursor connect against a non-empty backlog *will* do
-        snapshot work, so it consumes budget normally.
-        """
-        from application.api.events.routes import _allow_replay
-
-        with patch("application.api.events.routes.settings") as mock_settings:
-            mock_settings.EVENTS_REPLAY_BUDGET_REQUESTS_PER_WINDOW = 5
-            mock_settings.EVENTS_REPLAY_BUDGET_WINDOW_SECONDS = 60
-            redis = MagicMock()
-            redis.xlen.return_value = 42
-            redis.incr.return_value = 1
-
-            assert _allow_replay(redis, "alice", None) is True
-            redis.incr.assert_called_once()
+            redis.xlen.assert_not_called()
 
     def test_allow_replay_passes_until_budget_exhausted(self):
         from application.api.events.routes import _allow_replay
@@ -647,3 +628,119 @@ class TestFormatHelpers:
         from application.api.events.routes import _normalize_last_event_id
 
         assert _normalize_last_event_id(candidate) == expected
+
+
+# ── replay policy: fresh sessions start live, snapshots are age-capped ──
+
+
+class TestReplayPolicy:
+    def test_replay_floor_id_disabled_when_setting_zero(self):
+        from application.api.events.routes import _replay_floor_id
+
+        with patch("application.api.events.routes.settings") as mock_settings:
+            mock_settings.EVENTS_REPLAY_MAX_AGE_HOURS = 0
+            assert _replay_floor_id() is None
+
+    def test_replay_floor_id_is_ms_stream_id(self):
+        from application.api.events.routes import _replay_floor_id
+
+        with patch("application.api.events.routes.settings") as mock_settings:
+            mock_settings.EVENTS_REPLAY_MAX_AGE_HOURS = 48
+            with patch(
+                "application.api.events.routes.time.time",
+                return_value=1_800_000_000.0,
+            ):
+                floor = _replay_floor_id()
+        assert floor == f"{(1_800_000_000 - 48 * 3600) * 1000}-0"
+
+    def test_replay_backlog_uses_cursor_when_newer_than_floor(self):
+        from application.api.events.routes import _replay_backlog
+
+        redis = MagicMock()
+        redis.xrange.return_value = []
+        with patch(
+            "application.api.events.routes._replay_floor_id",
+            return_value="1000-0",
+        ):
+            list(_replay_backlog(redis, "alice", "2000-0", 200))
+        assert redis.xrange.call_args.kwargs["min"] == "(2000-0"
+
+    def test_replay_backlog_clamps_start_to_age_floor(self):
+        from application.api.events.routes import _replay_backlog
+
+        redis = MagicMock()
+        redis.xrange.return_value = []
+        with patch(
+            "application.api.events.routes._replay_floor_id",
+            return_value="5000-0",
+        ):
+            list(_replay_backlog(redis, "alice", "2000-0", 200))
+        # Cursor is older than the floor: replay starts at the floor
+        # (inclusive), not the cursor — entries past their shelf life
+        # are never shipped.
+        assert redis.xrange.call_args.kwargs["min"] == "5000-0"
+
+    def test_no_cursor_connect_never_replays(self):
+        """A fresh session (no Last-Event-ID) starts live: no XRANGE, no
+        weeks-old backlog on every tab-open."""
+        from application.api.events import routes as events_module
+
+        app = _make_app()
+        redis_client = MagicMock()
+        redis_client.incr.return_value = 1
+
+        def _fake_subscribe(self, on_subscribe=None, poll_timeout=1.0):
+            if on_subscribe is not None:
+                on_subscribe()
+            yield None
+
+        with patch.object(
+            events_module, "get_redis_instance", return_value=redis_client
+        ), patch.object(
+            events_module.Topic, "subscribe", _fake_subscribe, create=False
+        ):
+            with app.test_client() as c:
+                r = c.get("/api/events", headers={"X-Test-Sub": "alice"})
+                body = _drain_until(
+                    r, lambda b: b": connected" in b, max_chunks=40
+                )
+        assert b": connected" in body
+        redis_client.xrange.assert_not_called()
+
+    def test_cursor_older_than_floor_emits_truncation_notice(self):
+        """An age-clamped snapshot has a gap the client can't see from
+        entry ids alone — it must get the truncation notice so it
+        refetches full state instead of trusting a partial replay."""
+        from application.api.events import routes as events_module
+
+        app = _make_app()
+        redis_client = MagicMock()
+        redis_client.incr.return_value = 1
+        # No retained-oldest available: the floor alone must trigger it.
+        redis_client.xinfo_stream.side_effect = Exception("nope")
+        redis_client.xrange.return_value = []
+
+        def _fake_subscribe(self, on_subscribe=None, poll_timeout=1.0):
+            if on_subscribe is not None:
+                on_subscribe()
+            yield None
+
+        with patch.object(
+            events_module, "get_redis_instance", return_value=redis_client
+        ), patch.object(
+            events_module.Topic, "subscribe", _fake_subscribe, create=False
+        ), patch.object(
+            events_module, "_replay_floor_id", return_value="9999999999999-0"
+        ):
+            with app.test_client() as c:
+                r = c.get(
+                    "/api/events",
+                    headers={
+                        "X-Test-Sub": "alice",
+                        "Last-Event-ID": "1735682400000-0",
+                    },
+                )
+                body = _drain_until(
+                    r, lambda b: b"backlog.truncated" in b, max_chunks=40
+                )
+        assert b"backlog.truncated" in body

tests/test_asgi.py

@@ -156,3 +156,15 @@ def test_cors_preflight_on_mcp_route():
     assert r.status_code in (200, 204)
     assert r.headers.get("access-control-allow-origin") == "*"
     assert "Mcp-Session-Id" in r.headers.get("access-control-allow-headers", "")
+
+
+@pytest.mark.unit
+def test_wsgi_threadpool_sized_from_settings():
+    """The Flask thread pool is the app's request-capacity ceiling —
+    it must be operator-tunable (prod incident 2026-07-05→07: 32 slots
+    exhausted by SSE holders starved all other requests)."""
+    from application import asgi
+    from application.core.settings import settings
+
+    assert asgi._WSGI_THREADPOOL == int(settings.WSGI_THREADPOOL_WORKERS)
+    assert asgi._WSGI_THREADPOOL >= 64