Function _slow_runner

tests/algorithm/test_baseline.py:186–226

A slow runner that creates backpressure by processing rollouts with delays.

()

    artifacts: List[_RolloutArtifacts] = []

    async def _slow_runner() -> None:
        """A slow runner that creates backpressure by processing rollouts with delays."""
        processed = 0
        while processed < expected_rollouts:
            attempted = await store.dequeue_rollout()
            if attempted is None:
                await asyncio.sleep(0.01)
                continue

            attempt = attempted.attempt
            rollout_id = attempted.rollout_id
            rollout = await store.get_rollout_by_id(rollout_id)

            # Track the sample that was enqueued
            if rollout:
                enqueued_samples.append(rollout.input)

            await store.update_attempt(
                rollout_id,
                attempt.attempt_id,
                status="running",
                worker_id="slow-runner",
            )

            # Add a delay to create backpressure and cause queue to fill up
            await asyncio.sleep(0.05)

            span = _build_span(rollout_id, attempt.attempt_id, sequence_id=1, index=processed + 1)
            await store.add_span(span)
            await store.update_attempt(rollout_id, attempt.attempt_id, status="succeeded")
            await store.update_rollout(rollout_id, status="succeeded")

            artifacts.append(
                _RolloutArtifacts(
                    rollout_id=rollout_id,
                    attempt_id=attempt.attempt_id,
                    attempt_sequence=attempt.sequence_id,
                    span=span,
                )
            )
            processed += 1

    runner_task = asyncio.create_task(_slow_runner())
    try: