Skip to content

fix(v1,serve): discard delivered intercepts; trim worker arenas at heartbeat cadence #1610

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
hubert-marek merged 1 commit into from
Jun 10, 2026
Merged

fix(v1,serve): discard delivered intercepts; trim worker arenas at heartbeat cadence #1610

Show file tree
Hide file tree
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
17 changes: 17 additions & 0 deletions verifiers/serve/server/env_worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,9 +257,26 @@ async def send_error_response(error: str) -> None:

async def stats_loop(self, interval: float = 10.0) -> None:
"""Loop to push worker stats to the router."""
libc = None
try:
import ctypes

libc = ctypes.CDLL("libc.so.6")
except OSError:
pass
while True:
await asyncio.sleep(interval)

# Return freed arena pages to the OS at the heartbeat cadence.
# Long multimodal rollouts churn hundreds of MB of boxed objects
# per rollout; without trim the worker's RSS ratchets to its
# high-water mark (~3x the live set, measured). ctypes calls
# release the GIL, so this never stalls the loop — unlike
# gc.collect(), which must NOT be called here (full collections
# on fat heaps are what caused worker heartbeat timeouts).
if libc is not None:
libc.malloc_trim(0)

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Move malloc_trim off the event loop

Every EnvWorker already starts the configure_runtime_native_threads() daemon trim loop from _cap_native_threads() during construction, so this adds a second trim path that runs synchronously on the asyncio event-loop thread. Releasing the GIL does not let this same event loop continue processing requests or heartbeats while malloc_trim(0) is executing (or waiting on allocator locks held by the daemon trim), so on the large fragmented heaps this targets it can introduce exactly the worker lag/heartbeat delays the stats loop is meant to report; rely on the existing background trim or offload this call instead.

Useful? React with 👍 / 👎.


stats = EnvWorkerStats(
worker_id=self.worker_id,
timestamp=time.time(),
Expand Down
31 changes: 23 additions & 8 deletions verifiers/v1/utils/endpoint_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,6 +224,18 @@ def rollout_queue(self, rollout_key: str) -> asyncio.Queue[str]:
def get_request(self, request_id: str) -> ConfigData:
return cast(ConfigData, self.server.intercepts[request_id])

def discard_request(self, request_id: str) -> None:
"""Drop a delivered intercept from the server's per-request store.

Each intercept retains the raw request body — the full message
history including in-sandbox base64 screenshots — and the server
only sweeps them at rollout unregister, so without per-delivery
discard a long browser rollout holds every turn's request body
simultaneously (~74% of env-worker memory measured). The HTTP
handler keeps its own local reference, so delivery is unaffected.
"""
self.server.intercepts.pop(request_id, None)

def request_context(
self, request_id: str, request: ConfigData
) -> ModelRequestContext:
Expand Down Expand Up @@ -450,14 +462,17 @@ async def forward_request(
state._set_error(error_info(e))
raise
finally:
if bool(request.get("stream")):
if request.get("protocol") != "openai_chat_completions":
raise NotImplementedError(
"Streaming interception is currently supported for OpenAI Chat Completions."
)
await synthesize_stream(request, response, error)
else:
deliver_response(request, response, error)
try:
if bool(request.get("stream")):
if request.get("protocol") != "openai_chat_completions":
raise NotImplementedError(
"Streaming interception is currently supported for OpenAI Chat Completions."
)
await synthesize_stream(request, response, error)
else:
deliver_response(request, response, error)
finally:
endpoint.discard_request(request_id)

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep failed streaming intercepts until cleanup can unblock them

When an intercepted request has stream: true for any non-chat protocol, this block raises before synthesize_stream() can put an EOF sentinel or resolve the response_future, but the new unconditional discard removes the only entry that InterceptionServer.unregister_rollout() uses to cancel that future and signal the chunk queue. In that failure path (and any synthesize_stream() error before delivery), the aiohttp handler is left waiting/keepalive-looping instead of being unblocked during rollout cleanup; only discard after successful delivery or explicitly unblock the intercept before popping it.

Useful? React with 👍 / 👎.



def normalize_endpoint_prompt(request: ConfigData) -> Messages:
Expand Down
Loading