fix(runner): keep blocking card interactions alive

This commit is contained in:
fdc310
2026-11-03 06:54:19 +08:00
parent 0f017efdfc
commit 7de0df45d9
6 changed files with 62 additions and 33 deletions
@@ -49,12 +49,18 @@ class RunCancelledError(Exception):
class RunDeadline:
"""Wall-clock deadline shared by assembly, model calls, and tool calls."""
def __init__(self, timeout_seconds: float):
def __init__(self, timeout_seconds: float, runtime: Any = None):
self.timeout_seconds = max(0.001, float(timeout_seconds))
self._deadline = time.monotonic() + self.timeout_seconds
self._runtime = runtime
self._initial_host_deadline = getattr(runtime, "deadline_at", None)
def remaining(self) -> float:
return max(0.0, self._deadline - time.monotonic())
extension = 0.0
current_host_deadline = getattr(self._runtime, "deadline_at", None)
if self._initial_host_deadline is not None and current_host_deadline is not None:
extension = float(current_host_deadline) - float(self._initial_host_deadline)
return max(0.0, self._deadline + extension - time.monotonic())
class RunInterruptChecker:
@@ -148,16 +154,19 @@ class RunInterruptChecker:
cancellation = asyncio.create_task(self._cancelled.wait())
temporary_watcher = asyncio.create_task(self._watch()) if self.available and self._watcher is None else None
try:
await asyncio.wait(
{task, cancellation},
timeout=deadline.remaining() if deadline else None,
return_when=asyncio.FIRST_COMPLETED,
)
if self._cancelled.is_set():
raise RunCancelledError(CANCELLED_ERROR)
if not task.done():
raise asyncio.TimeoutError(TIMEOUT_ERROR)
return task.result()
while True:
if self._cancelled.is_set():
raise RunCancelledError(CANCELLED_ERROR)
if task.done():
return task.result()
remaining = deadline.remaining() if deadline else None
if remaining is not None and remaining <= 0:
raise asyncio.TimeoutError(TIMEOUT_ERROR)
await asyncio.wait(
{task, cancellation},
timeout=min(remaining, self.interval_seconds) if remaining is not None else None,
return_when=asyncio.FIRST_COMPLETED,
)
finally:
pending = [task, cancellation] + ([temporary_watcher] if temporary_watcher else [])
for future in pending:
@@ -222,7 +231,7 @@ class DefaultRunner(Runner):
api = self.get_run_api(ctx)
config = ctx.config if isinstance(ctx.config, dict) else {}
timeout_seconds = get_run_timeout_seconds(config)
deadline = RunDeadline(timeout_seconds) if timeout_seconds is not None else None
deadline = RunDeadline(timeout_seconds, ctx.runtime) if timeout_seconds is not None else None
usage_tracker = RunUsageTracker()
if await interrupt_checker.is_cancelled(force=True):
@@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
import time
import pytest
@@ -31,6 +32,25 @@ async def test_slow_ledger_does_not_delay_chunks_or_deadline():
assert checker._watcher is None
@pytest.mark.asyncio
async def test_interaction_extension_keeps_local_runner_wait_alive():
context = make_context()
context.runtime.deadline_at = time.time() + 0.03
deadline = RunDeadline(0.03, context.runtime)
checker = RunInterruptChecker(None, context, interval_seconds=0.01)
async def interaction():
await asyncio.sleep(0.01)
context.runtime.deadline_at += 0.1
await asyncio.sleep(0.06)
return "submitted"
assert await checker.wait_for(interaction(), deadline=deadline) == "submitted"
assert deadline.remaining() > 0
assert await checker.wait_for(interaction(), deadline=deadline) == "submitted"
assert deadline.remaining() > 0
@pytest.mark.asyncio
async def test_operation_timeout_is_not_mistaken_for_poll_timeout():
checker = RunInterruptChecker(None, make_context())