Merge pull request #29 from openai/fix-concurrency-reset-bug

Fix parallel recovery cap resets and reduce duplicate warnings

Author: hemanth-asirvatham

src/gabriel/utils/openai_utils.py

@@ -4481,9 +4481,10 @@ def _log_rate_limit_once(detail: Optional[str] = None) -> None:
             msg = (
                 "Encountered first rate limit error. Future rate limit errors will be silenced and tracked in periodic updates."
             )
-            logger.warning(msg)
             if message_verbose:
                 print(msg)
+            else:
+                logger.warning(msg)
             first_rate_limit_logged = True
         else:
             if detail:
@@ -4497,9 +4498,10 @@ def _log_connection_once(detail: Optional[str] = None) -> None:
             msg = (
                 "Encountered first connection error. Future connection errors will be silenced and tracked in periodic updates."
             )
-            logger.warning(msg)
             if message_verbose:
                 print(msg)
+            else:
+                logger.warning(msg)
             first_connection_logged = True
         else:
             if detail:
@@ -4518,9 +4520,10 @@ def _halt_ramp_up(reason: str) -> None:
         if concurrency_cap > ramp_cap:
             concurrency_cap = ramp_cap
         msg = f"[parallelization] Halting ramp-up at {ramp_cap} due to {reason}."
-        logger.warning(msg)
         if message_verbose:
             print(msg)
+        else:
+            logger.warning(msg)
 
     def _record_timeout_event(now: Optional[float] = None) -> None:
         ts = now if now is not None else time.time()
@@ -4961,7 +4964,6 @@ def maybe_adjust_concurrency() -> None:
                     f"{int(round(error_window))}s (rate-limit={recent_rate_limit_errors}, "
                     f"connection={recent_connection_errors})."
                 )
-                logger.warning(reason)
                 _halt_ramp_up("error recovery")
                 emit_parallelization_status(reason, force=True)
             rate_limit_errors_since_adjust = 0
@@ -4986,7 +4988,7 @@ def maybe_adjust_concurrency() -> None:
             and (now - last_concurrency_scale_up) >= error_window
         ):
             growth_headroom_limit = max(1, int(math.floor(ceiling_cap * 0.9)))
-            success_threshold = max(60, int(math.ceil(concurrency_cap * 2.5)))
+            success_threshold = max(60, int(math.ceil(concurrency_cap * 1.5)))
             if (
                 concurrency_cap < growth_headroom_limit
                 and successes_since_adjust >= success_threshold
@@ -5427,7 +5429,6 @@ async def _handle_rate_limit_error(error_text: str) -> None:
                         dedup_key=("cap-adjustment", type(e).__name__, str(e)),
                     )
                     logger.debug("Throughput tuning failed; retaining existing cap.", exc_info=True)
-                concurrency_cap = new_cap
                 if resps and all((isinstance(r, str) and not r.strip()) for r in resps):
                     if call_timeout is not None:
                         elapsed = time.time() - start