From f6f741ba235a52b21f0f4bd5497cba91cfe76571 Mon Sep 17 00:00:00 2001 From: ManishMadan2882 Date: Thu, 25 Jun 2026 19:27:44 +0530 Subject: [PATCH 01/11] (feat:ui) message input, search icon --- frontend/src/Navigation.tsx | 4 ++-- frontend/src/components/MessageInput.tsx | 15 +++++++-------- frontend/src/conversation/Conversation.tsx | 12 +++++++++--- 3 files changed, 18 insertions(+), 13 deletions(-) diff --git a/frontend/src/Navigation.tsx b/frontend/src/Navigation.tsx index 3cd144db..efdc07c3 100644 --- a/frontend/src/Navigation.tsx +++ b/frontend/src/Navigation.tsx @@ -571,14 +571,14 @@ export default function Navigation({ navOpen, setNavOpen }: NavigationProps) { )} {conversations?.data && conversations.data.length > 0 ? (
-
+

{t('chats')}

+ + {showNodeConfig && selectedNode && ( diff --git a/frontend/src/agents/workflow/hooks/useUndoRedo.ts b/frontend/src/agents/workflow/hooks/useUndoRedo.ts new file mode 100644 index 00000000..6c5ed249 --- /dev/null +++ b/frontend/src/agents/workflow/hooks/useUndoRedo.ts @@ -0,0 +1,146 @@ +import { useCallback, useRef, useState } from 'react'; + +import type { Dispatch, SetStateAction } from 'react'; +import type { Edge, Node } from 'reactflow'; + +export type WorkflowSnapshot = { + nodes: Node[]; + edges: Edge[]; +}; + +type UseUndoRedoOptions = { + nodes: Node[]; + edges: Edge[]; + setNodes: Dispatch>; + setEdges: Dispatch>; + onRestore?: (snapshot: WorkflowSnapshot) => void; + maxHistorySize?: number; +}; + +const SNAPSHOT_GROUP_WINDOW_MS = 1000; + +// Selection is not part of history: restored elements keep whatever +// selection the user has right now instead of the flags baked into the +// snapshot when it was taken. +function withCurrentSelection( + restored: T[], + current: T[], +): T[] { + const selectedIds = new Set( + current.filter((el) => el.selected).map((el) => el.id), + ); + return restored.map((el) => + Boolean(el.selected) === selectedIds.has(el.id) + ? el + : { ...el, selected: selectedIds.has(el.id) }, + ); +} + +export function useUndoRedo({ + nodes, + edges, + setNodes, + setEdges, + onRestore, + maxHistorySize = 100, +}: UseUndoRedoOptions) { + const [past, setPast] = useState([]); + const [future, setFuture] = useState([]); + + const stateRef = useRef({ nodes, edges }); + stateRef.current = { nodes, edges }; + const pastRef = useRef(past); + pastRef.current = past; + const futureRef = useRef(future); + futureRef.current = future; + const onRestoreRef = useRef(onRestore); + onRestoreRef.current = onRestore; + + // Consecutive calls sharing a group key (e.g. keystrokes editing one + // node's config) collapse into a single history entry while each call + // stays inside the group window. + const lastGroupKeyRef = useRef(null); + const lastGroupTimeRef = useRef(0); + // One user action can request several snapshots in the same event tick + // (deleting a node also deletes its edges); only the first one counts. + const tickGuardRef = useRef(false); + + const takeSnapshot = useCallback( + (groupKey?: string) => { + const now = Date.now(); + if ( + groupKey && + groupKey === lastGroupKeyRef.current && + now - lastGroupTimeRef.current < SNAPSHOT_GROUP_WINDOW_MS + ) { + lastGroupTimeRef.current = now; + return; + } + lastGroupKeyRef.current = groupKey ?? null; + lastGroupTimeRef.current = now; + + if (tickGuardRef.current) return; + tickGuardRef.current = true; + queueMicrotask(() => { + tickGuardRef.current = false; + }); + + const snapshot: WorkflowSnapshot = { + nodes: stateRef.current.nodes, + edges: stateRef.current.edges, + }; + setPast((prev) => + prev.length >= maxHistorySize + ? [...prev.slice(prev.length - maxHistorySize + 1), snapshot] + : [...prev, snapshot], + ); + setFuture([]); + }, + [maxHistorySize], + ); + + const undo = useCallback(() => { + const previous = pastRef.current[pastRef.current.length - 1]; + if (!previous) return; + const current: WorkflowSnapshot = { + nodes: stateRef.current.nodes, + edges: stateRef.current.edges, + }; + lastGroupKeyRef.current = null; + setPast(pastRef.current.slice(0, -1)); + setFuture([...futureRef.current, current]); + setNodes(withCurrentSelection(previous.nodes, current.nodes)); + setEdges(withCurrentSelection(previous.edges, current.edges)); + onRestoreRef.current?.(previous); + }, [setNodes, setEdges]); + + const redo = useCallback(() => { + const next = futureRef.current[futureRef.current.length - 1]; + if (!next) return; + const current: WorkflowSnapshot = { + nodes: stateRef.current.nodes, + edges: stateRef.current.edges, + }; + lastGroupKeyRef.current = null; + setFuture(futureRef.current.slice(0, -1)); + setPast([...pastRef.current, current]); + setNodes(withCurrentSelection(next.nodes, current.nodes)); + setEdges(withCurrentSelection(next.edges, current.edges)); + onRestoreRef.current?.(next); + }, [setNodes, setEdges]); + + const clearHistory = useCallback(() => { + lastGroupKeyRef.current = null; + setPast([]); + setFuture([]); + }, []); + + return { + takeSnapshot, + undo, + redo, + clearHistory, + canUndo: past.length > 0, + canRedo: future.length > 0, + }; +} From 41affdf1f5d60d0609219ae9f0c9ed507ec1537d Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 7 Jul 2026 18:51:31 +0100 Subject: [PATCH 08/11] feat: improve sse --- application/cache.py | 65 ++++++++++++ application/streaming/broadcast_channel.py | 22 +++- tests/test_events_substrate.py | 10 +- tests/test_pubsub_timeouts.py | 116 +++++++++++++++++++++ 4 files changed, 206 insertions(+), 7 deletions(-) create mode 100644 tests/test_pubsub_timeouts.py diff --git a/application/cache.py b/application/cache.py index 4e2b6719..e5a82db0 100644 --- a/application/cache.py +++ b/application/cache.py @@ -1,6 +1,7 @@ import hashlib import json import logging +import socket as _socket import time from threading import Lock @@ -11,6 +12,12 @@ from application.utils import get_hash logger = logging.getLogger(__name__) +# Upper bound on any single blocking read by a pub/sub subscriber. Must stay +# comfortably above Topic.subscribe's poll_timeout (1 s) — get_message's idle +# wait polls with select() and never trips socket_timeout, but a half-open +# connection's pending read (e.g. the health-check PONG) does. +PUBSUB_SOCKET_TIMEOUT_SECONDS = 10 + def _cache_default(value): # Image attachments arrive inline as bytes (see GoogleLLM.prepare_messages_with_attachments); @@ -51,6 +58,64 @@ def get_redis_instance(): return _redis_instance +_pubsub_redis_instance = None +_pubsub_redis_creation_failed = False + + +def _tcp_keepalive_options(): + """Kernel keepalive knobs for long-lived, mostly-idle pub/sub sockets. + + Probing well inside NAT/IPVS idle-expiry windows (Docker Swarm's IPVS + expires idle flows after ~15 min) keeps the flow-table entry alive and + lets the kernel surface a dead peer instead of leaving the socket + half-open. The constants are Linux-specific, so build the dict from + whatever this platform exposes. + """ + options = {} + for name, value in (("TCP_KEEPIDLE", 300), ("TCP_KEEPINTVL", 60), ("TCP_KEEPCNT", 3)): + const = getattr(_socket, name, None) + if const is not None: + options[const] = value + return options + + +def get_pubsub_redis_instance(): + """Redis client dedicated to pub/sub subscribers. + + Separate from ``get_redis_instance`` because subscribers hold a socket + open for the life of an SSE connection. Without ``socket_timeout``, a + connection silently dropped by NAT/IPVS blocks ``pubsub.get_message`` + forever — including the ``health_check_interval`` PONG read — pinning + the subscriber's WSGI thread until the worker restarts. Bounding every + read lets a dead subscriber fail within seconds and release its thread. + + Returns: + A shared ``redis.Redis`` client, or ``None`` if Redis is + unavailable or ``CACHE_REDIS_URL`` is invalid. + """ + global _pubsub_redis_instance, _pubsub_redis_creation_failed + if _pubsub_redis_instance is None and not _pubsub_redis_creation_failed: + with _instance_lock: + if _pubsub_redis_instance is None and not _pubsub_redis_creation_failed: + try: + _pubsub_redis_instance = redis.Redis.from_url( + settings.CACHE_REDIS_URL, + socket_connect_timeout=2, + socket_timeout=PUBSUB_SOCKET_TIMEOUT_SECONDS, + socket_keepalive=True, + socket_keepalive_options=_tcp_keepalive_options(), + health_check_interval=10, + ) + except ValueError as e: + logger.error(f"Invalid Redis URL: {e}") + _pubsub_redis_creation_failed = True # Stop future attempts + _pubsub_redis_instance = None + except redis.ConnectionError as e: + logger.error(f"Redis connection error: {e}") + _pubsub_redis_instance = None # Keep trying for connection errors + return _pubsub_redis_instance + + def gen_cache_key(messages, model="docgpt", tools=None): if not all(isinstance(msg, dict) for msg in messages): raise ValueError("All messages must be dictionaries.") diff --git a/application/streaming/broadcast_channel.py b/application/streaming/broadcast_channel.py index 07861d5d..a7dcc022 100644 --- a/application/streaming/broadcast_channel.py +++ b/application/streaming/broadcast_channel.py @@ -19,7 +19,9 @@ from __future__ import annotations import logging from typing import Callable, Iterator, Optional -from application.cache import get_redis_instance +import redis as redis_lib + +from application.cache import get_pubsub_redis_instance, get_redis_instance logger = logging.getLogger(__name__) @@ -65,8 +67,13 @@ class Topic: If Redis is unavailable, returns immediately without yielding. Cleanly unsubscribes on ``GeneratorExit`` (client disconnect). + + Uses the pub/sub-dedicated client (bounded ``socket_timeout``): a + subscriber whose connection went half-open must fail within seconds + and release its WSGI thread, not block in ``get_message`` until the + worker restarts. """ - redis = get_redis_instance() + redis = get_pubsub_redis_instance() if redis is None: logger.debug("Redis unavailable; subscribe to %s yielded nothing", self.name) return @@ -86,6 +93,17 @@ class Topic: while True: try: msg = pubsub.get_message(timeout=poll_timeout) + except redis_lib.exceptions.TimeoutError: + # A bounded read timed out mid-message/health-check: the + # connection is half-open (NAT/IPVS dropped the flow). + # End the subscription so the SSE client reconnects on a + # fresh connection; expected during network churn, so no + # stack trace. + logger.info( + "pubsub read timed out for %s; closing subscriber", + self.name, + ) + return except Exception: logger.exception("pubsub.get_message failed for %s", self.name) return diff --git a/tests/test_events_substrate.py b/tests/test_events_substrate.py index 7523f456..ef42b164 100644 --- a/tests/test_events_substrate.py +++ b/tests/test_events_substrate.py @@ -305,7 +305,7 @@ class TestTopic: # Must not raise. assert Topic("user:alice").publish("hi") == 0 - @patch("application.streaming.broadcast_channel.get_redis_instance") + @patch("application.streaming.broadcast_channel.get_pubsub_redis_instance") def test_subscribe_returns_immediately_when_redis_unavailable( self, mock_redis ): @@ -313,7 +313,7 @@ class TestTopic: # Generator should produce nothing, not raise. assert list(Topic("user:alice").subscribe(poll_timeout=0.01)) == [] - @patch("application.streaming.broadcast_channel.get_redis_instance") + @patch("application.streaming.broadcast_channel.get_pubsub_redis_instance") def test_subscribe_yields_none_on_poll_timeout(self, mock_redis): client = MagicMock() pubsub = MagicMock() @@ -331,7 +331,7 @@ class TestTopic: assert third == b"x" gen.close() - @patch("application.streaming.broadcast_channel.get_redis_instance") + @patch("application.streaming.broadcast_channel.get_pubsub_redis_instance") def test_subscribe_fires_on_subscribe_after_ack(self, mock_redis): client = MagicMock() pubsub = MagicMock() @@ -355,7 +355,7 @@ class TestTopic: assert callback_calls == [1] gen.close() - @patch("application.streaming.broadcast_channel.get_redis_instance") + @patch("application.streaming.broadcast_channel.get_pubsub_redis_instance") def test_subscribe_cleans_up_on_generator_close(self, mock_redis): client = MagicMock() pubsub = MagicMock() @@ -373,7 +373,7 @@ class TestTopic: pubsub.unsubscribe.assert_called_once_with("user:alice") pubsub.close.assert_called_once() - @patch("application.streaming.broadcast_channel.get_redis_instance") + @patch("application.streaming.broadcast_channel.get_pubsub_redis_instance") def test_subscribe_skips_unsubscribe_if_subscribe_never_acked( self, mock_redis ): diff --git a/tests/test_pubsub_timeouts.py b/tests/test_pubsub_timeouts.py new file mode 100644 index 00000000..13b28a1c --- /dev/null +++ b/tests/test_pubsub_timeouts.py @@ -0,0 +1,116 @@ +"""Tests for the pub/sub-dedicated Redis client and subscriber timeout handling. + +Regression tests for the 2026-07-05→07 prod incident: idle pub/sub +connections silently dropped by NAT/IPVS never raised inside +``pubsub.get_message`` (the shared cache client has no ``socket_timeout``), +leaving each ``/api/events`` SSE generator blocked in ``_read_from_socket`` +forever. One WSGI thread-pool slot leaked per dead subscriber until the +pool was exhausted and the worker stopped serving Flask requests. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +import redis as redis_lib + +import application.cache as cache +from application.streaming.broadcast_channel import Topic + + +@pytest.fixture(autouse=True) +def _reset_pubsub_singleton(): + cache._pubsub_redis_instance = None + cache._pubsub_redis_creation_failed = False + yield + cache._pubsub_redis_instance = None + cache._pubsub_redis_creation_failed = False + + +class TestGetPubsubRedisInstance: + + @pytest.mark.unit + def test_bounds_blocking_reads_and_enables_keepalive(self): + with patch("application.cache.redis.Redis.from_url") as from_url: + client = cache.get_pubsub_redis_instance() + assert client is from_url.return_value + kwargs = from_url.call_args.kwargs + assert kwargs["socket_timeout"] == cache.PUBSUB_SOCKET_TIMEOUT_SECONDS + assert kwargs["socket_timeout"] > 0 + assert kwargs["socket_keepalive"] is True + assert kwargs["health_check_interval"] == 10 + + @pytest.mark.unit + def test_socket_timeout_exceeds_subscribe_poll_interval(self): + """get_message(timeout=poll) waits ``poll`` seconds while healthy-idle; + socket_timeout must be comfortably larger or idle polling would be + misread as a dead connection.""" + assert cache.PUBSUB_SOCKET_TIMEOUT_SECONDS >= 5 + + @pytest.mark.unit + def test_singleton(self): + with patch("application.cache.redis.Redis.from_url") as from_url: + first = cache.get_pubsub_redis_instance() + second = cache.get_pubsub_redis_instance() + assert first is second + assert from_url.call_count == 1 + + @pytest.mark.unit + def test_invalid_url_marks_failed_and_stops_retrying(self): + with patch( + "application.cache.redis.Redis.from_url", side_effect=ValueError("bad url") + ): + assert cache.get_pubsub_redis_instance() is None + with patch("application.cache.redis.Redis.from_url") as from_url: + assert cache.get_pubsub_redis_instance() is None + from_url.assert_not_called() + + @pytest.mark.unit + def test_connection_error_allows_retry(self): + with patch( + "application.cache.redis.Redis.from_url", + side_effect=redis_lib.ConnectionError("down"), + ): + assert cache.get_pubsub_redis_instance() is None + with patch("application.cache.redis.Redis.from_url") as from_url: + assert cache.get_pubsub_redis_instance() is from_url.return_value + + +class TestSubscribeTimeout: + + @pytest.mark.unit + def test_socket_timeout_ends_subscription(self): + """A read timeout means the connection is half-open/dead: the generator + must exit — freeing its WSGI thread — rather than loop or raise.""" + pubsub = MagicMock() + pubsub.get_message.side_effect = redis_lib.exceptions.TimeoutError( + "Timeout reading from socket" + ) + client = MagicMock() + client.pubsub.return_value = pubsub + with patch( + "application.streaming.broadcast_channel.get_pubsub_redis_instance", + return_value=client, + ): + events = list(Topic("user:u1").subscribe()) + assert events == [] + pubsub.close.assert_called_once() + + @pytest.mark.unit + def test_subscribe_uses_dedicated_pubsub_client(self): + """The shared cache client has no socket_timeout; subscribers must not + fall back to it.""" + pubsub = MagicMock() + pubsub.get_message.side_effect = redis_lib.exceptions.TimeoutError() + client = MagicMock() + client.pubsub.return_value = pubsub + with patch( + "application.streaming.broadcast_channel.get_pubsub_redis_instance", + return_value=client, + ) as get_pubsub, patch( + "application.streaming.broadcast_channel.get_redis_instance" + ) as get_cache: + list(Topic("user:u1").subscribe()) + get_pubsub.assert_called_once() + get_cache.assert_not_called() From e9c377378c493b263f8f8147242c2323d990871c Mon Sep 17 00:00:00 2001 From: ManishMadan2882 Date: Wed, 8 Jul 2026 00:57:58 +0530 Subject: [PATCH 09/11] (fixes)worflow: gracefully handle delete on key event --- .../src/agents/workflow/WorkflowBuilder.tsx | 79 ++++++++++++------- 1 file changed, 50 insertions(+), 29 deletions(-) diff --git a/frontend/src/agents/workflow/WorkflowBuilder.tsx b/frontend/src/agents/workflow/WorkflowBuilder.tsx index adbf82fb..415798ef 100644 --- a/frontend/src/agents/workflow/WorkflowBuilder.tsx +++ b/frontend/src/agents/workflow/WorkflowBuilder.tsx @@ -582,30 +582,49 @@ function WorkflowBuilderInner() { [], ); + const deleteNodesAndEdges = useCallback( + (nodesToDelete: Node[], edgesToDelete: Edge[]) => { + const removableIds = new Set( + nodesToDelete.filter((n) => n.type !== 'start').map((n) => n.id), + ); + const edgeIdsToDelete = new Set(edgesToDelete.map((e) => e.id)); + if (removableIds.size === 0 && edgeIdsToDelete.size === 0) return; + + takeSnapshot(); + setNodes((nds) => nds.filter((n) => !removableIds.has(n.id))); + setEdges((eds) => + eds.filter( + (e) => + !edgeIdsToDelete.has(e.id) && + !removableIds.has(e.source) && + !removableIds.has(e.target), + ), + ); + const dropRemoved = (prev: Record): Record => { + const next = { ...prev }; + let changed = false; + removableIds.forEach((id) => { + if (id in next) { + delete next[id]; + changed = true; + } + }); + return changed ? next : prev; + }; + setAgentJsonSchemaDrafts(dropRemoved); + setAgentJsonSchemaErrors(dropRemoved); + if (selectedNode && removableIds.has(selectedNode.id)) { + setSelectedNode(null); + setShowNodeConfig(false); + } + }, + [selectedNode, takeSnapshot], + ); + const handleDeleteNode = useCallback(() => { - if (!selectedNode || selectedNode.type === 'start') return; - takeSnapshot(); - setNodes((nds) => nds.filter((n) => n.id !== selectedNode.id)); - setEdges((eds) => - eds.filter( - (e) => e.source !== selectedNode.id && e.target !== selectedNode.id, - ), - ); - setAgentJsonSchemaDrafts((prev) => { - if (!(selectedNode.id in prev)) return prev; - const next = { ...prev }; - delete next[selectedNode.id]; - return next; - }); - setAgentJsonSchemaErrors((prev) => { - if (!(selectedNode.id in prev)) return prev; - const next = { ...prev }; - delete next[selectedNode.id]; - return next; - }); - setSelectedNode(null); - setShowNodeConfig(false); - }, [selectedNode, takeSnapshot]); + if (!selectedNode) return; + deleteNodesAndEdges([selectedNode], []); + }, [selectedNode, deleteNodesAndEdges]); const handleUpdateNodeData = useCallback( (data: Record, options?: { snapshot?: boolean }) => { @@ -741,8 +760,12 @@ function WorkflowBuilderInner() { return; } } - if (e.key === 'Delete' && selectedNode && !isEditable) { - handleDeleteNode(); + if ((e.key === 'Delete' || e.key === 'Backspace') && !isEditable) { + e.preventDefault(); + deleteNodesAndEdges( + nodes.filter((n) => n.selected || n.id === selectedNode?.id), + edges.filter((edge) => edge.selected), + ); } if (e.key === 'Escape') { setShowNodeConfig(false); @@ -751,7 +774,7 @@ function WorkflowBuilderInner() { }; window.addEventListener('keydown', handleKeyDown); return () => window.removeEventListener('keydown', handleKeyDown); - }, [selectedNode, handleDeleteNode, undo, redo]); + }, [nodes, edges, selectedNode, deleteNodesAndEdges, undo, redo]); const handlePanelBackdropClick = useCallback(() => { setShowNodeConfig(false); @@ -1817,11 +1840,9 @@ function WorkflowBuilderInner() { onNodeClick={handleNodeClick} onNodeDragStart={snapshotBeforeCanvasChange} onSelectionDragStart={snapshotBeforeCanvasChange} - onNodesDelete={snapshotBeforeCanvasChange} - onEdgesDelete={snapshotBeforeCanvasChange} nodeTypes={nodeTypes} nodeDragThreshold={1} - deleteKeyCode={['Backspace', 'Delete']} + deleteKeyCode={null} fitView > From 83c560f626f0dd60c10f2a3047571047fa8284a6 Mon Sep 17 00:00:00 2001 From: Alex Date: Wed, 8 Jul 2026 00:23:32 +0100 Subject: [PATCH 10/11] feat: mini replay improvements --- application/api/events/routes.py | 87 +++++++++++------- application/asgi.py | 3 +- application/core/settings.py | 13 +++ tests/api/test_events_routes.py | 149 +++++++++++++++++++++++++------ tests/test_asgi.py | 12 +++ 5 files changed, 207 insertions(+), 57 deletions(-) diff --git a/application/api/events/routes.py b/application/api/events/routes.py index 9080a58a..d1097157 100644 --- a/application/api/events/routes.py +++ b/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: '(' 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: '(' 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 diff --git a/application/asgi.py b/application/asgi.py index 1ffb1d6d..787f94ca 100644 --- a/application/asgi.py +++ b/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="/") diff --git a/application/core/settings.py b/application/core/settings.py index a5d5b8ec..52caed41 100644 --- a/application/core/settings.py +++ b/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. diff --git a/tests/api/test_events_routes.py b/tests/api/test_events_routes.py index 629acd39..8eef96c3 100644 --- a/tests/api/test_events_routes.py +++ b/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 diff --git a/tests/test_asgi.py b/tests/test_asgi.py index 556a47a6..da516fa2 100644 --- a/tests/test_asgi.py +++ b/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 From 748975ae957370ef185cf42b67f50fb916cafb3a Mon Sep 17 00:00:00 2001 From: Alex Date: Wed, 8 Jul 2026 00:25:35 +0100 Subject: [PATCH 11/11] fix: clean up comments --- application/core/settings.py | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/application/core/settings.py b/application/core/settings.py index 52caed41..fe44ba7e 100644 --- a/application/core/settings.py +++ b/application/core/settings.py @@ -272,12 +272,6 @@ 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. @@ -295,11 +289,6 @@ 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;