Skip to content

Fix parallel recovery cap resets and reduce duplicate warnings #29

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hemanth-asirvatham merged 1 commit into from
Feb 7, 2026
Merged

Fix parallel recovery cap resets and reduce duplicate warnings #29

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 7 additions & 6 deletions src/gabriel/utils/openai_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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()
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
Loading