feat(v1): streaming tasksets - Taskset.stream() and a windowed runner

[faresobeid]

What

Taskset.stream() — the read path the eval runner consumes — and the windowed primitive behind it, vf.run_stream, public.

class Taskset(ABC, Generic[TaskT, TasksetConfigT]):
    async def stream(self) -> AsyncIterator[TaskT]:
        """The tasks as they become available. Default: `iter(self)` (system prompt, `head`/`shuffle` view applied).
        A taskset whose tasks appear over time overrides this: an async generator that awaits its source,
        yields each task complete, and returns when the source is drained."""

    @property
    def streaming(self) -> bool:
        """Whether `stream()` is overridden."""

# verifiers/v1/utils/aio.py — exported as `vf.run_stream`; replaces the runner's one `gather_rollouts(...)` call
async def run_stream(source: AsyncIterable[T], run: Callable[[T], Awaitable[R]], *, window: int | None) -> list[R]:
    """Consume `source` — an async iterable of tasks, typically a feed that waits for its next item — running each
    through `run` with at most `window` in flight (None: no bound); return every result in submission order once the
    source has ended and the last run is done. Back-pressure: the next item is pulled only once a run has freed a
    place. One run failing cancels the rest, waits for them to unwind, re-raises; the source is `aclose()`d on any
    exit. The caller decides when the source ends."""

Scope

vf eval is bounded; long-running consumption is the caller's job. An eval entrypoint is a report on a finite set — it never waits open-endedly for new tasks. On a streaming taskset vf eval therefore requires -n/--num-examples: it takes the stream's next n tasks (pulling each only while fewer than -c rollouts are in flight, and never an (n+1)th the feed may still be waiting for), then ends and reports like any other run. Without -n it refuses (ValueError naming the flag — the existing infinite-taskset rule, extended); --shuffle/--resume are refused too (no whole set to sample, no keys to resume against).

A service that consumes a feed for as long as it runs owns that loop itself — its shutdown, where the results go — and is built on the same primitive the runner uses, not on the entrypoint:

async with env.serving():  # env-level resources, up for the life of the service
    ctx = vf.ModelContext(client=client, model=model, sampling=sampling)
    await vf.run_stream(env.taskset.stream(), lambda task: env.run_slot(vf.RunSlot(task), ctx), window=32)

Why the split: the eval entrypoint carries a run (open_run/finish_run), a dashboard with a total, --resume, an exit code — all of which presuppose a finite set. What a service reuses is the runner primitive (bounded concurrency, cancel-the-rest, source close-out) and env.run_slot (whole-episode retries, on_complete), so those are the public surface; run_stream is generic over the item so the in-process path, the served path (one client.run per slot) and a service's env.run_slot all go through the one implementation. (This is the reshape after review: the first cut let vf eval run until a stream drained.)

Why

Production is a feed: an owed seat turn appears when the records change, not from a list. The data-flywheel environment carries its own scheduler for exactly this — run.Loop.sweep/schedule/reconcile/_seat, the Place/Places capacities, the feed-wait loop, settle and SHUTDOWN_GRACE — ~330 lines whose only job is "which task is due now, bounded how, stopped how", re-implementing what the runner already owns (bounded concurrency, run_episode_with_retry, append_episode, shutdown). The runner could not consume a feed because Taskset.load() is sync (a feed cannot be awaited without blocking the loop) and run_eval materialised the taskset (tasks = list(selected)) and planned every slot before the first rollout. An async-capable load() was rejected (breaks __iter__, head/shuffle, and every static taskset's contract). stream() is a separate hook so load stays the small thing AGENTS.md wants, and the override is the declaration (no flag that can disagree with the code). With run_stream public, the service keeps its loop and drops its scheduler; vf eval -n 20 on the same taskset is the bounded spot-check.

What is preserved

• A static taskset (no stream() override) takes the same run_stream path with every slot planned up front, before the dashboard's first frame, and window=None: the semaphore alone bounds what runs, exactly as before — same admission order, same dashboard total from the start, same cancel-the-rest / wait-to-unwind semantics gather_rollouts had (folded into run_stream; it had no other callers). INFINITE, -n, --shuffle, --resume behave as today for it.
• Taskset.load, __iter__, head, shuffle, view unchanged; the default stream() yields iter(self), so head(2).stream() == iter(head(2)).
• The served path is unchanged below the slot: one client.run per slot; the streamed taskset lives in the main process as before.
• open_run(num_examples=) stays an int: a stream's count is its -n.
• Public API additions: vf.run_stream, vf.RunSlot (already the type Env.run_slot/Env.slots take and return).

Tests

• tests/v1/test_taskset.py (deterministic, no model): a fixture taskset streaming off an asyncio.Queue, driven through vf.run_stream with tasks directly: the default stream() equals iteration (head(2) included) and streaming is False/True; _take(…, n) yields the next n, fewer if the feed ends first, never pulls an (n+1)th, and closes a started source; with window=2 the 3rd task is not pulled while two runs are in flight (asserted on the source's cursor with the runs gated); results come back in submission order once the stream ends; a failing run cancels the others, re-raises, and aclose()s a stream whose feed is still waiting — also when it fails while the feed is quiet (the pull is raced against the runs' first failure, so a quiet feed cannot hide it); a task's -r rollouts are windowed one by one (window=1, 50 rollouts, never more than one alive); cancelling the run while stream() is parked on its feed closes it.
• tests/v1/test_e2e.py::test_streaming_taskset[run_v1|run_v1_server]: the fixture taskset (tests/v1/fixtures/echo_stream_v1.py) is a live feed that never drains (the echo phrases every 0.5 s, cycling); with -n 3 the run ends after the 3rd task, in-process and served → 3 ok episodes, 3 rows in traces.jsonl (a pull past the 3rd would land as a 4th episode, not a hang). test_streaming_taskset_requires_num_tasks (offline): no -n → ValueError naming -n/--num-examples before any rollout. test_streaming_taskset_refuses_shuffle_and_resume[shuffle|resume] (offline): ValueError before any rollout.
• Ran locally: uv run pytest tests/v1 -m "not e2e" -n auto: 98 passed; the two streaming e2e cases plus test_single_turn[null-harness-in-subprocess], test_env_id_best_of_n, test_multi_agent_env_server (static path, in-process and served): passed. The documented service pattern (env.serving() / vf.run_stream / env.run_slot) run for real against the fixture feed with window=2: 4 ok episodes, each published through on_complete. uv run ruff check, ruff format --check, pre-commit run --all-files, ty check verifiers: clean.

---

Note

Medium Risk
Changes core eval scheduling and concurrency for all runs (via run_stream), not only streaming tasksets; mistakes in windowing or source closure could affect rollout counts or leave feeds hanging.

Overview
Adds streaming tasksets: subclasses override async Taskset.stream() (detected by streaming) so tasks can arrive over time while load() may stay empty. The public vf.run_stream helper runs an async source with optional window back-pressure, ordered results, fail-fast cancellation, and aclose() on the feed.

run_eval now drives all rollouts through stream(): streaming runs _take(..., -n) → _plan_stream → run_stream (dashboard slots grow as tasks arrive; max_concurrent is the pull window); static tasksets still plan every slot upfront and use window=None (semaphore-only, same behavior as the old gather_rollouts path). Streaming and infinite tasksets require -n; shuffle and resume are rejected on streams.

Docs cover streaming vs bounded eval vs long-lived run_stream services; EchoStreamTaskset and unit/e2e tests exercise back-pressure, bounds, failures, and config errors.

^{Reviewed by Cursor Bugbot for commit 0982604. Bugbot is set up for automated code reviews on this repo. Configure here.}

Note

Add Taskset.stream() async hook and windowed run_stream runner for v1

• Adds Taskset.stream() default async hook and Taskset.streaming property in taskset.py; subclasses override stream() to emit tasks lazily.
• Adds run_stream concurrency utility in aio.py that pulls items with back-pressure, limits pending runs via a window, returns results in submission order, and cancels outstanding work plus closes the source on failure or cancellation.
• Refactors run_eval in runner.py to consume the next num_tasks streamed tasks via _take and _plan_stream, execute both static and streaming evals through run_stream, and reject missing bounds, shuffle, and resume for streaming tasksets.
• Exports RunSlot and run_stream from the verifiers.v1 public API.
• Risk: static evaluations now route through run_stream instead of the old gather_rollouts helper; any out-of-tree callers of gather_rollouts will break since it was removed from runner.py.

^{Macroscope summarized 0982604.}

[macroscopeapp]

🟠 High eval/runner.py:352

Streaming evals hang when a scheduled rollout fails before the concurrency window fills, instead of aborting and propagating the failure. run_stream waits for the next group from selected.stream() before checking pending tasks, so with -c 2 a failed first rollout can leave the runner blocked indefinitely if the feed does not yield another task; update it to wait on either the next group or an existing task completion and close the stream on failure.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @verifiers/v1/cli/eval/runner.py around line 352:

Streaming evals hang when a scheduled rollout fails before the concurrency window fills, instead of aborting and propagating the failure. `run_stream` waits for the next group from `selected.stream()` before checking pending tasks, so with `-c 2` a failed first rollout can leave the runner blocked indefinitely if the feed does not yield another task; update it to wait on either the next group or an existing task completion and close the stream on failure.

[faresobeid]

Fixed in 4090736. run_stream no longer awaits the next group directly: the pull is its own task, raced (asyncio.wait(..., FIRST_COMPLETED)) against a future that the first failing slot resolves via a done-callback (so the wait stays O(1) for a static run's thousands of in-flight slots). A failure while the feed is quiet raises at once; the pull is cancelled and awaited before the stream is aclose()d, and the existing cancel-the-rest semantics are unchanged. Test: test_failing_slot_is_seen_while_the_feed_is_quiet — one failing slot then a feed that never yields again raises promptly (with and without a window) and the stream is closed.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit ddf1e74. Configure here.}

[macroscopeapp]

Approvability

Verdict: Not approved

Macroscope's review found this PR not approvable — This PR introduces a new streaming taskset capability and shared windowed execution infrastructure that changes production evaluation and service lifecycle behavior. Unresolved high-severity concerns include failure handling and missing system-prompt propagation for overridden streams.

Not approved because:

• 2 blocking correctness issues found at or above your repo's Minimum Blocking Severity

^{Adjust the Minimum Blocking Severity for this repo — including turning it Off — in Settings. You can add or adjust custom eligibility rules. Learn more.}

[macroscopeapp]

🟠 High v1/taskset.py:71

Streaming tasksets ignore config.system_prompt, so --env.taskset.system-prompt <file> sends their yielded tasks without the configured prompt and changes evaluation results. An overridden stream() bypasses Taskset.__iter__, where task.with_system_prompt(...) is applied; ensure the streaming path applies the same transformation before yielding tasks.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @verifiers/v1/taskset.py around line 71:

Streaming tasksets ignore `config.system_prompt`, so `--env.taskset.system-prompt <file>` sends their yielded tasks without the configured prompt and changes evaluation results. An overridden `stream()` bypasses `Taskset.__iter__`, where `task.with_system_prompt(...)` is applied; ensure the streaming path applies the same transformation before yielding tasks.