05 · 取消与超时：请求停止以后，还要等待收尾

    async def _next(self):
        take = asyncio.create_task(self.outputs.get())
        try:
            await asyncio.wait((take, self.producer), return_when=asyncio.FIRST_COMPLETED)
            if take.done():
                return take.result()
            self.producer.result()  # Propagate custom generation errors.
            if not self.outputs.empty():
                return await take
            return None
        finally:
            take.cancel()
            await asyncio.gather(take, return_exceptions=True)

    try:
        async with asyncio.timeout(CONFIG.rollout_guard_sec):
            async with boot_agent_sandbox(md["image"], instance_id) as sb:
                await swe.prepare_workspace(sb, md["workdir"], md)
                agent_exit_code = await HARNESS_CLS().run(
                    sb,
                    workdir=md["workdir"],
                    session_id=session_id,
                    adapter_url=state.adapter_url,
                    time_budget_sec=CONFIG.agent_time_budget_sec,
                    prompt=swe.SWE_PROMPT,

async def run_rollout_io(args, function, *values, **kwargs):
    """Bound the executor backlog before submitting encoding/sync work."""
    loop = asyncio.get_running_loop()
    key = (os.getpid(), loop)
    if key not in _async_limits:
        _async_limits[key] = asyncio.Semaphore(getattr(args, "rollout_io_concurrency", 4))
    async with _async_limits[key]:
        pending = asyncio.create_task(asyncio.to_thread(function, *values, **kwargs))
        try:
            return await asyncio.shield(pending)
        except asyncio.CancelledError:
            # Repeated cancellation still cannot stop a filesystem commit.
            # Keep capacity and input ownership until the accepted write finishes.
            while not pending.done():
                try:
                    await asyncio.shield(pending)
                except asyncio.CancelledError:
                    pass
            pending.result()
            raise