feat(v1): lazy task resolution + forced group dispatch hooks for tasksets#1919
feat(v1): lazy task resolution + forced group dispatch hooks for tasksets#1919faresobeid wants to merge 5 commits into
Conversation
…sets Add Taskset.resolve_task(idx, tasks) — called by the env server per run_rollout/run_group request, defaulting to the eager list — so a taskset can materialize tasks lazily behind a fixed index range (e.g. a replay buffer reading ~1MB saved-rollout lines on demand). run_group resolves one task per request, so every rollout of a group shares the binding. Add Taskset.REQUIRES_GROUP_ROLLOUTS, OR-ed into the server's advertised requires_group_scoring, for tasksets whose resolve_task binds per-request state the whole group must share even without @group_reward s. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- null/default harnesses: a seeded conversation over ~100KB no longer ships as the INITIAL_MESSAGES env var — execve caps a single env string at MAX_ARG_STRLEN (128KiB on Linux), which E2BIG-crashed every launch — but as a workspace file named by INITIAL_MESSAGES_FILE (majority of real replay recheck seeds exceed the limit). - eval runner + validate CLI: resolve tasks through Taskset.resolve_task so lazy tasksets serve real tasks in-process, not stubs; eager tasksets get the identical objects back. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- resolve_task takes the loaded task (stubs carry their idx) instead of (idx, tasks) — callers no longer thread the eager list back to the taskset that produced it, which deletes the select-and-resolve dance from the eval runner and validate CLI. - NEEDS_CONTAINER / REQUIRES_GROUP_ROLLOUTS become plain (non-ClassVar) attributes: they are read off instances and settable per instance (a taskset may derive them from its config), so the annotation now says so. - The INITIAL_MESSAGES / INITIAL_MESSAGES_FILE staging moves into a shared Harness.stage_message_prompt — the threshold, filename, and env-var names live in one place instead of two copies (the program-side read stays duplicated: standalone uv scripts can't import verifiers). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
| async def _run_rollout(self, req: RunRolloutRequest) -> RunRolloutResponse: | ||
| ctx = self._context(req.client, req.model, req.sampling) | ||
| episode = self.env.episode(self.tasks[req.task_idx], ctx, n=1) | ||
| task = await self.env.taskset.resolve_task(self.tasks[req.task_idx]) | ||
| episode = self.env.episode(task, ctx, n=1) |
There was a problem hiding this comment.
🟡 Medium serve/server.py:122
_run_rollout serves single-rollout requests unconditionally, even for tasksets that set REQUIRES_GROUP_ROLLOUTS. A caller that sends run_rollout directly bypasses the group-only contract and calls resolve_task() once per rollout, so rollouts that should share a single resolved task instead get independently sampled task state. Consider rejecting run_rollout with an error when self.requires_group_scoring is true, so callers are forced to use run_group.
async def _run_rollout(self, req: RunRolloutRequest) -> RunRolloutResponse:
+ if self.requires_group_scoring:
+ raise ValueError("this taskset requires group rollouts; use run_group")
ctx = self._context(req.client, req.model, req.sampling)
task = await self.env.taskset.resolve_task(self.tasks[req.task_idx])🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/serve/server.py around lines 122-125:
`_run_rollout` serves single-rollout requests unconditionally, even for tasksets that set `REQUIRES_GROUP_ROLLOUTS`. A caller that sends `run_rollout` directly bypasses the group-only contract and calls `resolve_task()` once per rollout, so rollouts that should share a single resolved task instead get independently sampled task state. Consider rejecting `run_rollout` with an error when `self.requires_group_scoring` is true, so callers are forced to use `run_group`.
| # Lazy tasksets bind task content at request time (Taskset.resolve_task); resolve each | ||
| # selected task once, mirroring the env server's one-bind-per-episode semantics. The | ||
| # default resolve returns the identical objects, so eager tasksets are unchanged. | ||
| tasks = [await env.taskset.resolve_task(task) for task in tasks] |
There was a problem hiding this comment.
🟠 High eval/runner.py:35
run_eval resolves each task once and passes the same object to all config.num_rollouts rollouts, but lazy tasksets bind fresh task content per resolve_task call. The env server resolves a new task for each run_rollout request, so the in-process path now shares one sampled payload across every rollout for a task, silently changing evaluation semantics. Resolve each task once per rollout instead of once per task.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/cli/eval/runner.py around line 35:
`run_eval` resolves each task once and passes the same object to all `config.num_rollouts` rollouts, but lazy tasksets bind fresh task content per `resolve_task` call. The env server resolves a new task for each `run_rollout` request, so the in-process path now shares one sampled payload across every rollout for a task, silently changing evaluation semantics. Resolve each task once per rollout instead of once per task.
…outs Add verifiers.v1.tasksets.replay, a base taskset (like harbor/textarena) that reads saved rollout files (Trace.to_record jsonl lines, the prime-rl orchestrator's rollout output) as a lazy buffer and serves derived tasks from them. A derivation subclasses ReplayTaskset (binding its narrowed config in the generic, so the loader picks it up) and implements two hooks: record_anchors (the resume points one saved rollout offers) and build_prompt (the seeded conversation for one anchor). The base owns everything shared and hard: the (file, offset) candidate index over barrier-complete steps, online semantics (throttled rescans, fresh sampling with forced group dispatch, capacity-as-recency-eviction, fail-fast on empty), forest-aware trace surgery (main-tree isolation, compaction-fork and resumable-tool-result enumerators, recheck seeds), scoring/tools/setup/finalize delegation to the original (`inner`) taskset via a task-swapped trace view, and lineage so replay-derived records aren't re-replayed by default (chains opt in via source_envs). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
|
||
| # ---------------------------------------------------- original-world hooks | ||
|
|
||
| def tools(self, task: ReplayTask) -> list[Toolset]: |
There was a problem hiding this comment.
🟡 Medium replay/taskset.py:233
tools() returns [] whenever task.source_task is empty, but Environment.shared_tools() calls taskset.tools(tasks[0]) at startup to decide which shared MCP servers to start — and tasks[0] is an unresolved stub with source_task={}. So for an inner taskset whose Toolset.config.shared=True, the shared servers are never started at eval level; every rollout re-launches them per-rollout in serve_tools(...) instead of reusing one instance. Consider returning the inner taskset's shared toolsets for stub tasks so shared_tools() sees them at startup.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/tasksets/replay/taskset.py around line 233:
`tools()` returns `[]` whenever `task.source_task` is empty, but `Environment.shared_tools()` calls `taskset.tools(tasks[0])` at startup to decide which shared MCP servers to start — and `tasks[0]` is an unresolved stub with `source_task={}`. So for an inner taskset whose `Toolset.config.shared=True`, the shared servers are never started at eval level; every rollout re-launches them per-rollout in `serve_tools(...)` instead of reusing one instance. Consider returning the inner taskset's shared toolsets for stub tasks so `shared_tools()` sees them at startup.
| return await asyncio.to_thread(self._materialize, task.idx, candidate) | ||
| # An online source line can vanish under us (its run resumed and cleaned future | ||
| # steps): drop the dangling candidate and draw again instead of failing forever. | ||
| for _ in range(8): |
There was a problem hiding this comment.
🟠 High replay/taskset.py:192
When a sampled candidate's file has vanished, the FileNotFoundError handler discards only that one candidate — but a single rollout line contributes one candidate per anchor node, so all sibling candidates pointing at the same missing file stay in the pool and keep being resampled. If the deleted rollout produced more than eight anchors, resolve_task exhausts its retry loop and raises RuntimeError even though other valid candidates still exist. Consider discarding all candidates that share the vanished file's path, not just the one that was sampled.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/tasksets/replay/taskset.py around line 192:
When a sampled candidate's file has vanished, the `FileNotFoundError` handler discards only that one candidate — but a single rollout line contributes one candidate per anchor node, so all sibling candidates pointing at the same missing file stay in the pool and keep being resampled. If the deleted rollout produced more than eight anchors, `resolve_task` exhausts its retry loop and raises `RuntimeError` even though other valid candidates still exist. Consider discarding all candidates that share the vanished file's path, not just the one that was sampled.
| Rollouts routinely end truncated mid-tool-call (timeouts, context length); a trailing | ||
| assistant's pending ``tool_calls`` must be stripped or the seed is API-malformed, and | ||
| if that leaves it with no content it is dropped entirely.""" | ||
| path = path_to_root(nodes, final_leaf(children, tree)) |
There was a problem hiding this comment.
🟡 Medium replay/surgery.py:149
recheck_seed only strips tool_calls from the last message when it is an assistant turn. When a rollout is truncated after some but not all tool results arrive, the final leaf is a tool node and the earlier assistant message retains its pending tool_calls — leaving the seed with an incomplete tool-call sequence and then appending a new user instruction. The resulting seed is API-malformed for any truncated multi-tool trace. Consider scanning all assistant messages on the path and removing tool_calls that have no corresponding tool result, rather than only checking the trailing message.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/tasksets/replay/surgery.py around line 149:
`recheck_seed` only strips `tool_calls` from the last message when it is an assistant turn. When a rollout is truncated after some but not all tool results arrive, the final leaf is a `tool` node and the earlier assistant message retains its pending `tool_calls` — leaving the seed with an incomplete tool-call sequence and then appending a new user instruction. The resulting seed is API-malformed for any truncated multi-tool trace. Consider scanning all assistant messages on the path and removing `tool_calls` that have no corresponding `tool` result, rather than only checking the trailing message.
| if fresh: | ||
| # Newest steps win the cap; this evicts the oldest candidates. The sorted | ||
| # fresh list is swapped in atomically (rescans run in a thread while the | ||
| # event loop keeps sampling). | ||
| retained.sort(key=lambda c: c.step, reverse=True) | ||
| del retained[MAX_CANDIDATES:] | ||
| self._all = retained |
There was a problem hiding this comment.
🟡 Medium replay/buffer.py:137
scan() snapshots self._all into retained at line 125 and later overwrites self._all with that snapshot at line 143. If another request calls discard() while the rescan thread is running (for example after _materialize() hits FileNotFoundError on a resumed/cleaned run), the discarded candidate is removed from the live list but then gets written back from retained, so the buffer can keep re-serving a vanished source file and make online resolution fail repeatedly.
if fresh:
# Newest steps win the cap; this evicts the oldest candidates. The sorted
# fresh list is swapped in atomically (rescans run in a thread while the
# event loop keeps sampling).
- retained.sort(key=lambda c: c.step, reverse=True)
- del retained[MAX_CANDIDATES:]
- self._all = retained
+ retained = [c for c in self._all if c not in retained] + retained
+ retained.sort(key=lambda c: c.step, reverse=True)
+ del retained[MAX_CANDIDATES:]
+ self._all = retained🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/tasksets/replay/buffer.py around lines 137-143:
`scan()` snapshots `self._all` into `retained` at line 125 and later overwrites `self._all` with that snapshot at line 143. If another request calls `discard()` while the rescan thread is running (for example after `_materialize()` hits `FileNotFoundError` on a resumed/cleaned run), the discarded candidate is removed from the live list but then gets written back from `retained`, so the buffer can keep re-serving a vanished source file and make online resolution fail repeatedly.
| self.rollout_dir = resolve_rollout_dir(buffer_dir) | ||
| self.anchors = anchors # the derivation's resume points for one record (see ReplayTaskset) | ||
| self.online = online | ||
| self.source_envs = set(source_envs) if source_envs else None |
There was a problem hiding this comment.
🟡 Medium replay/buffer.py:106
source_envs=[] is collapsed to None at line 106, so an explicitly empty filter is silently ignored. In _candidates_from, None takes the default branch and replays every non-replay env instead of matching nothing. Consider preserving the empty list (e.g. self.source_envs = set(source_envs) if source_envs is not None else None) so an explicit empty filter matches no envs.
| self.source_envs = set(source_envs) if source_envs else None | |
| self.source_envs = set(source_envs) if source_envs is not None else None |
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/tasksets/replay/buffer.py around line 106:
`source_envs=[]` is collapsed to `None` at line 106, so an explicitly empty filter is silently ignored. In `_candidates_from`, `None` takes the default branch and replays every non-replay env instead of matching nothing. Consider preserving the empty list (e.g. `self.source_envs = set(source_envs) if source_envs is not None else None`) so an explicit empty filter matches no envs.
| if type(self.inner).user is not Taskset.user: | ||
| raise ValueError(f"taskset {inner_config.id!r} defines a user simulator; replay does not support one") | ||
| self.inner_task_type = task_type(inner_config.id) | ||
| self.NEEDS_CONTAINER = self.inner.NEEDS_CONTAINER |
There was a problem hiding this comment.
🟡 Medium replay/taskset.py:135
__init__ unconditionally overwrites self.NEEDS_CONTAINER with self.inner.NEEDS_CONTAINER, so a replay subclass that sets NEEDS_CONTAINER = True on itself has that flag silently erased when the inner taskset doesn't also need a container. The env then runs in a subprocess on the host instead of refusing the unsupported runtime. Consider OR-ing the two flags (e.g. self.NEEDS_CONTAINER = self.inner.NEEDS_CONTAINER or type(self).NEEDS_CONTAINER) so a subclass container requirement is preserved.
| self.NEEDS_CONTAINER = self.inner.NEEDS_CONTAINER | |
| self.NEEDS_CONTAINER = self.inner.NEEDS_CONTAINER or type(self).NEEDS_CONTAINER |
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/tasksets/replay/taskset.py around line 135:
`__init__` unconditionally overwrites `self.NEEDS_CONTAINER` with `self.inner.NEEDS_CONTAINER`, so a replay subclass that sets `NEEDS_CONTAINER = True` on itself has that flag silently erased when the inner taskset doesn't also need a container. The env then runs in a subprocess on the host instead of refusing the unsupported runtime. Consider OR-ing the two flags (e.g. `self.NEEDS_CONTAINER = self.inner.NEEDS_CONTAINER or type(self).NEEDS_CONTAINER`) so a subclass container requirement is preserved.
| retained.extend(candidates) | ||
| fresh += len(candidates) | ||
| self._scanned_steps.add(step) | ||
| if fresh >= MAX_CANDIDATES: |
There was a problem hiding this comment.
🟡 Medium replay/buffer.py:132
When scan() fills the MAX_CANDIDATES cap from newer steps, line 135 adds every older step to _scanned_steps permanently. If those newer source files later disappear (the online path already expects this via discard()), the older steps that were skipped are never scanned, so _all can become empty and sample() fails with a RuntimeError even though replayable rollouts still exist in older step directories. Consider not marking unscanned older steps as scanned, so a later rescan can promote them when newer candidates are discarded.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/tasksets/replay/buffer.py around line 132:
When `scan()` fills the `MAX_CANDIDATES` cap from newer steps, line 135 adds every older step to `_scanned_steps` permanently. If those newer source files later disappear (the online path already expects this via `discard()`), the older steps that were skipped are never scanned, so `_all` can become empty and `sample()` fails with a `RuntimeError` even though replayable rollouts still exist in older step directories. Consider not marking unscanned older steps as scanned, so a later rescan can promote them when newer candidates are discarded.
Whole-rollout derivations implement a single hook (build_prompt); only mid-rollout derivations override record_anchors. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
| # inner tasksets with shared-placement toolsets are therefore unsupported. | ||
| if not task.source_task: | ||
| return [] | ||
| return self.inner.tools(self._inner_task(task)) |
There was a problem hiding this comment.
🟡 Medium replay/taskset.py:239
ReplayTaskset.__init__ only rejects inner tasksets whose taskset-level State is custom (state_cls(type(self.inner)) is not State), but it does not guard against inner toolsets that declare their own custom State parameter (Toolset[..., MyState]). Replay rollouts build the base State (empty dict), and the interception server validates every tool PUT /state against state_cls(type(self.taskset)), which is the base State. When an inner toolset writes a custom state field, the PUT /state is rejected with 400 and the tool call fails, breaking the rollout. Consider extending the constructor validation to inspect each inner toolset's State parameter and reject tasksets whose toolsets use a custom State.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @verifiers/v1/tasksets/replay/taskset.py around line 239:
`ReplayTaskset.__init__` only rejects inner tasksets whose taskset-level `State` is custom (`state_cls(type(self.inner)) is not State`), but it does not guard against inner toolsets that declare their own custom `State` parameter (`Toolset[..., MyState]`). Replay rollouts build the base `State` (empty dict), and the interception server validates every tool `PUT /state` against `state_cls(type(self.taskset))`, which is the base `State`. When an inner toolset writes a custom state field, the `PUT /state` is rejected with 400 and the tool call fails, breaking the rollout. Consider extending the constructor validation to inspect each inner toolset's `State` parameter and reject tasksets whose toolsets use a custom `State`.
2 participants
Hooks for lazily-bound tasksets, a replay taskset base built on them, and two fixes surfaced while building the first consumer (prime-rl's replay envs, PrimeIntellect-ai/prime-rl#2941).
Changes
Taskset.resolve_task(task): called by the env server once perrun_rollout/run_grouprequest; defaults to returning the loaded task, so existing tasksets are byte-for-byte unchanged. Overriding it letsload_tasks()return lightweight stubs while the real task content binds per request.run_groupresolves one task per request, so every rollout of a GRPO group binds the same resolved task. The in-process eval runner andvalidateCLI resolve through the same hook.Taskset.REQUIRES_GROUP_ROLLOUTS: OR-ed into the server's advertisedrequires_group_scoring, for tasksets whoseresolve_taskbinds per-request state the whole group must share, even without@group_rewards. Both this andNEEDS_CONTAINERare plain (non-ClassVar) attributes — read off instances, settable per instance from config.verifiers.v1.tasksets.replay— a base taskset in the harbor/textarena mold: reads saved rollout files (Trace.to_record()jsonl lines, the prime-rl orchestrator's rollout output) as a lazy buffer and serves derived tasks. A derivation subclassesReplayTasksetand implementsrecord_anchors(the resume points one saved rollout offers) +build_prompt(the seeded conversation for one anchor); the base owns the (file, offset) candidate index over barrier-complete steps, online semantics (throttled rescans, fresh sampling with forced group dispatch, capacity-as-recency eviction, fail-fast on empty buffers), forest-aware trace surgery (main-tree isolation, compaction-fork and resumable-tool-result enumerators, recheck seeds), scoring/tools/setup/finalize delegation to the original (inner) taskset via a task-swapped trace view, and lineage so replay-derived records aren't re-replayed by default (chains opt in viasource_envs). First derivations:replay-continue-v1/replay-recheck-v1in prime-rl.INITIAL_MESSAGES_FILE) instead of theINITIAL_MESSAGESenv var — execve caps a single env string atMAX_ARG_STRLEN(128KiB on Linux), which crashed every launch for the majority of real resumed-conversation seeds. The staging lives in a sharedHarness.stage_message_prompt.Verification
tests/v1/test_replay.py(16 tests: graph surgery on synthetic forests, buffer scan/barrier/eviction/pick determinism, lineage skipping). Empirically reproduced the E2BIG failure and the fix path. Trace-surgery algorithms validated against real production rollout files. End-to-end resolution exercised by the prime-rl derivation packages. The legacy v0 bridge overrides both server request paths and is unaffected.🤖 Generated with Claude Code