fix(streams): cut SSE error-log volume and add Redis pool headroom

The task-event SSE endpoint holds one blocking XREAD connection per
connected client. When concurrent streams exceed the Redis connection
pool size, every client's read fails on each cycle. Combined with
plain-text logging that splits each traceback into one log entry per
line, and a flat 1s retry, a sustained pool stall becomes a
log-ingestion firehose.

Changes:

- Streaming error path: replace the flat 1s retry with capped
  exponential backoff (1->2->4->8->16->30s), reset on a healthy read,
  so a tight per-client loop can't hammer Redis or flood logs. Full
  tracebacks are still logged on every failure (nothing swallowed) with
  a failure counter for context; volume is bounded by the backoff and by
  single-entry JSON logging below.

- Default to structured JSON logs in all deployed environments
  (ENVIRONMENT != "development"). JSON keeps a multi-line traceback as a
  single log entry (newlines live inside the quoted exc_info field)
  instead of one cluster-log entry per line. Local development keeps
  plain text for readable console output. The JSON formatter already
  existed but was gated behind Datadog configuration.

- Bump the in-code REDIS_MAX_CONNECTIONS default (50 -> 200) for
  peak-concurrency headroom. This is a mitigation, not a fix:
  connections still scale 1:1 with clients. The durable fix is a shared
  per-pod reader that fans out to in-process queues. Deployed
  environments set the real cap via the REDIS_MAX_CONNECTIONS env var.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: danielmillerp

agentex/src/config/environment_variables.py

@@ -96,7 +96,13 @@ class EnvironmentVariables(BaseModel):
     MONGODB_DATABASE_NAME: str | None = "agentex"
     MONGODB_MAX_POOL_SIZE: int = 50
     MONGODB_MIN_POOL_SIZE: int = 5
-    REDIS_MAX_CONNECTIONS: int = 50  # Increased for SSE streaming
+    # SSE streaming currently holds one blocking XREAD connection per connected
+    # client, so the pool needs headroom for peak concurrent streams per pod.
+    # NOTE: this is only the in-code default — deployed environments override it
+    # via the REDIS_MAX_CONNECTIONS env var, which is the real cap. Bumping this
+    # buys headroom but does NOT change the 1-connection-per-client scaling; the
+    # durable fix is a shared per-pod reader that fans out to in-process queues.
+    REDIS_MAX_CONNECTIONS: int = 200
     REDIS_CONNECTION_TIMEOUT: int = 60  # Connection timeout in seconds
     REDIS_SOCKET_TIMEOUT: int = 30  # Socket timeout in seconds
     REDIS_STREAM_MAXLEN: int = (
@@ -164,7 +170,7 @@ def refresh(cls, force_refresh: bool = False) -> EnvironmentVariables | None:
                 os.environ.get(EnvVarKeys.MONGODB_MIN_POOL_SIZE, "5")
             ),
             REDIS_MAX_CONNECTIONS=int(
-                os.environ.get(EnvVarKeys.REDIS_MAX_CONNECTIONS, "100")
+                os.environ.get(EnvVarKeys.REDIS_MAX_CONNECTIONS, "200")
             ),
             REDIS_CONNECTION_TIMEOUT=int(
                 os.environ.get(EnvVarKeys.REDIS_CONNECTION_TIMEOUT, "20")

agentex/src/domain/use_cases/streams_use_case.py

@@ -114,6 +114,11 @@ async def stream_task_events(
         ping_interval = float(
             self.environment_variables.SSE_KEEPALIVE_PING_INTERVAL
         )  # Configurable keepalive ping interval
+        # Track consecutive read failures so we can back off and avoid a
+        # tight error loop. When the Redis pool is exhausted, every connected
+        # client's read fails on each cycle; without backoff this turns into a
+        # log-ingestion firehose (one failure per client per cycle, ~once/sec).
+        consecutive_errors = 0
         try:
             # Application-level control loop
             while True:
@@ -133,6 +138,10 @@ async def stream_task_events(
                         last_message_time = asyncio.get_running_loop().time()
                         await asyncio.sleep(0.02)
 
+                    # A read cycle completed without raising — the stream is
+                    # healthy again, so reset the backoff/error counter.
+                    consecutive_errors = 0
+
                     # If we didn't get any messages, add a small pause
                     # to prevent tight loops and send keepalive ping if needed
                     if message_count == 0:
@@ -151,13 +160,25 @@ async def stream_task_events(
                     )
                     raise
                 except Exception as e:
+                    consecutive_errors += 1
+                    # Always log the full traceback — nothing is swallowed.
+                    # Volume is controlled two ways instead of by dropping
+                    # diagnostics: structured JSON logging keeps each traceback
+                    # to a single log entry (see utils.logging), and the
+                    # exponential backoff below caps how often a sustained
+                    # failure can repeat. The failure counter gives context on
+                    # how long a stream has been erroring.
                     logger.error(
-                        f"Error processing events for task {task_id}: {e}",
+                        f"Error processing events for task {task_id} "
+                        f"(failure #{consecutive_errors}): {e}",
                         exc_info=True,
                     )
                     yield f"data: {TaskStreamErrorEventEntity(type='error', message=str(e)).model_dump_json()}\n\n"
-                    # Add a small delay before continuing
-                    await asyncio.sleep(1)
+                    # Exponential backoff (capped) so a sustained failure (e.g.
+                    # Redis pool exhaustion) doesn't spin a tight per-client
+                    # loop hammering Redis and flooding logs.
+                    backoff = min(2.0 ** min(consecutive_errors - 1, 5), 30.0)
+                    await asyncio.sleep(backoff)
 
         except asyncio.CancelledError:
             # Just exit the generator on cancellation

agentex/src/utils/logging.py

@@ -14,6 +14,17 @@
 # Check if Datadog is configured
 _is_datadog_configured = bool(os.environ.get("DD_AGENT_HOST"))
 
+# Emit structured JSON logs in all deployed environments. JSON keeps a
+# multi-line traceback (exc_info=True) as a single log entry — the newlines
+# live inside the quoted `exc_info` field — instead of fanning out into one
+# cluster-log entry per traceback line. Splitting tracebacks per line was a
+# primary multiplier behind a log-ingestion spike on a plain-text-logging
+# cluster. Local development keeps plain text for readable console output;
+# JSON is the default everywhere else (including when ENVIRONMENT is unset)
+# so a deployed cluster can never silently fall back to per-line tracebacks.
+_is_local_dev = os.environ.get("ENVIRONMENT", "").lower() == "development"
+_use_json_logs = not _is_local_dev
+
 # Include Datadog trace IDs only when Datadog is configured
 if _is_datadog_configured:
     LOG_FORMAT: str = (
@@ -134,7 +145,7 @@ def make_logger(name: str) -> logging.Logger:
 
     logger = logging.getLogger(name)
     stream_handler = logging.StreamHandler()
-    if _is_datadog_configured:
+    if _use_json_logs:
         stream_handler.setFormatter(CustomJSONFormatter())
     else:
         stream_handler.setFormatter(logging.Formatter(LOG_FORMAT))