feat(v1): taskset system_prompt handoff for GEPA → eval/train

docs/v1/evaluation.md

@@ -33,6 +33,8 @@ The output from evaluations are written into `outputs/<taskset>--<model>--<harne
 - `model` — the model id to evaluate, e.g. `nvidia/NVIDIA-Nemotron-3-Ultra-550B-A55B`
 - `sampling` — generation params passed to the model, e.g. `sampling.temperature`
 - `taskset.id` / `harness.id` — pick the taskset and harness
+- `taskset.system_prompt` / `taskset.system_prompt_file` — override the task system
+  prompt (mutually exclusive; e.g. a GEPA `best_system_prompt.txt`)
 - `num_tasks` — how many tasks to evaluate. Not setting a value means all tasks; an
   infinite taskset (a procedural generator, e.g. `wordle-v1`) requires it
 - `num_rollouts` — rollouts per task

docs/v1/gepa.md

@@ -6,7 +6,7 @@ verifiers offers built in support for [GEPA](https://github.com/gepa-ai/gepa), a
 uv run gepa reverse-text-v1
 ```
 
-`gepa` runs GEPA where a number of rollouts are done before a teacher LLM reflects on the results to propose a better `Task.system_prompt` without any gradient based training. It runs against native v1 tasksets.
+`gepa` runs GEPA where a number of rollouts are done before a teacher LLM reflects on the results to propose a better system prompt without any gradient based training. It runs against native v1 tasksets.
 
 GEPA reuses the same `taskset` / `harness` / `client` / `sampling` config as eval, so the `.toml` config remains very similar:
 
@@ -35,10 +35,17 @@ Validate the config by using `uv run gepa @ config.toml --dry-run`. To run GEPA,
 
 ## Output
 
-Results go under `outputs/<taskset>--<model>--<harness>/<uuid>/`, matching `eval`. The best system prompt is printed when the run finishes.
+Results go under `outputs/<taskset>--<model>--<harness>/<uuid>/`, matching `eval`. The best system prompt is written to `best_system_prompt.txt` and printed when the run finishes.
+
+Reuse it in eval or training via the same taskset knobs:
+
+```bash
+uv run eval reverse-text-v1 \
+  --taskset.system-prompt-file outputs/<run>/best_system_prompt.txt
+```
 
 ## Limitations
 
-**Tasksets** — GEPA optimizes `Task.system_prompt`, so the taskset must provide one. Tasksets that bake instructions into the user `prompt` instead (e.g. `gsm8k-v1`) are not supported out of the box.
+**Tasksets** — GEPA seeds from `--taskset.system-prompt` / `-file` when set, otherwise from a baked-in `TaskData.system_prompt`. Tasksets that only put instructions in the user `prompt` (e.g. `gsm8k-v1`) need an explicit config seed to work with GEPA.
 
 **Harnesses** — any eval harness works. With `APPENDS_SYSTEM_PROMPT`, the optimized prompt is used as a system message but otherwise is folded into the user prompt.

skills/evaluate-environments/references/REFERENCE.md

@@ -216,6 +216,8 @@ fields become typed dotted flags such as `--taskset.split test`.
 |---|---|---|---|
 | `id` | `ID` | `""` | Local package or Hub `org/name[@version]`; selects the taskset and its config type. Set via `--taskset.id`. |
 | `task` | `TaskConfig` | `TaskConfig()` | Task-facing config passed to every constructed task. `SerializeAsAny` preserves a narrowed subclass. Set through `--taskset.task.*`. |
+| `system_prompt` | `str \| None` | `None` | Config-layer system prompt: replaces bake-in `TaskData.system_prompt` after `load()` (GEPA optimizes this layer). Mutually exclusive with `system_prompt_file`. |
+| `system_prompt_file` | `Path \| None` | `None` | File override for `system_prompt` (read as UTF-8 text), same mutual-exclusion pattern as `JudgeConfig.prompt` / `prompt_file`. |
 
 `.name` → the package name (id with org / version stripped).
 
@@ -430,7 +432,7 @@ value in the run's `TimeoutConfig` wins; otherwise the corresponding row value i
 | `name` | `str \| None` | `None` | Optional human-readable label used in logs and dashboards. |
 | `description` | `str \| None` | `None` | Optional human-readable description. |
 | `prompt` | `str \| Messages \| None` | — | Initial user input. A string is one user prompt; `Messages` seeds a full initial conversation and requires a harness with `SUPPORTS_MESSAGE_PROMPT`; `None` lets the user simulator open via `respond("")`. |
-| `system_prompt` | `str \| None` | `None` | Optional system prompt. Harnesses with `APPENDS_SYSTEM_PROMPT` emit a real system message; otherwise a string prompt is prefixed with a warning. A separate system prompt cannot be folded into `Messages` or `None`. |
+| `system_prompt` | `str \| None` | `None` | Bake-in task-side system prompt (per-task or taskset default). Config `taskset.system_prompt` / `_file` replaces it after `load()`. Harnesses with `APPENDS_SYSTEM_PROMPT` emit a real system message (harness framing first via `compose_system_prompt` when set); otherwise a string prompt is prefixed with a warning. A separate system prompt cannot be folded into `Messages` or `None`. |
 | `image` | `str \| None` | `None` | Required container/sandbox image for this row. It replaces the base runtime image; subprocess is refused when set. |
 | `workdir` | `str \| None` | `None` | Working directory for harness execution and task hooks. Applied when the runtime supports it and its config remains at the default. |
 | `timeout` | `TaskTimeout` | `TaskTimeout()` | Per-stage timeout requests described above. |

verifiers/v1/__init__.py

@@ -90,7 +90,7 @@
     TaskTimeout,
     WireTaskData,
 )
-from verifiers.v1.taskset import Taskset, TasksetConfig
+from verifiers.v1.taskset import Taskset, TasksetConfig, resolve_system_prompt
 from verifiers.v1.mcp import (
     Toolset,
     SharedToolsetConfig,
@@ -196,6 +196,7 @@
     "Taskset",
     "TaskConfig",
     "TasksetConfig",
+    "resolve_system_prompt",
     "BaseConfig",
     "Harness",
     "HarnessConfig",

verifiers/v1/cli/gepa.py

@@ -1,7 +1,8 @@
 """The GEPA entrypoint: `uv run gepa [<taskset-id>] --model <model> [options]`.
 
-Registered as the `gepa` console script. Optimizes a v1 taskset's `Task.system_prompt` via
-GEPA (Genetic-Pareto): alternating rollouts with a teacher LM reflecting on results — see
+Registered as the `gepa` console script. Optimizes a v1 taskset's config-layer system
+prompt (`--taskset.system-prompt` / `--taskset.system-prompt-file`) via GEPA
+(Genetic-Pareto): alternating rollouts with a teacher LM reflecting on results — see
 `verifiers.v1.gepa`. CLI resolution mirrors `eval`/`serve` (`verifiers.v1.cli.resolve`): a
 leading bare token is the taskset id, the taskset/harness subconfigs are narrowed from their
 `id`s so `--taskset.*` / `--harness.*` stay typed and `-h` renders them, `@ file.toml` loads,

verifiers/v1/gepa/__init__.py

@@ -1,8 +1,9 @@
 """GEPA (Genetic-Pareto) system-prompt optimization for native v1 environments.
 
 `run_gepa(env, config)` drives the third-party `gepa` optimizer against a v1 `Environment`
-via `GEPAAdapter`, seeding from and improving `Task.system_prompt`. The `gepa` console
-script (`verifiers.v1.cli.gepa`) is a thin entrypoint over this package.
+via `GEPAAdapter`, seeding from and improving the taskset config-layer system prompt
+(`--taskset.system-prompt` / `--taskset.system-prompt-file`). The `gepa` console script
+(`verifiers.v1.cli.gepa`) is a thin entrypoint over this package.
 """
 
 from verifiers.v1.gepa.adapter import GEPAAdapter

verifiers/v1/gepa/adapter.py

@@ -26,12 +26,14 @@
 
 @dataclass
 class GEPAAdapter:
-    """Bridges GEPA's optimization loop with a native v1 `Environment`. `tasks` covers only the
-    trainset + valset tasks GEPA was given (not the whole taskset), keyed by `task.data.idx` —
-    GEPA's `batch` is a list of those idxs, and injecting the candidate rebuilds each `Task`
-    around a `data` row carrying the new `system_prompt`. `loop` is the runner's persistent event
-    loop (which holds `env.serving()` open); `evaluate` drives its rollouts on it with
-    `run_until_complete`."""
+    """Bridges GEPA's optimization loop with a native v1 `Environment`.
+
+    `tasks` are bake-in rows (pre config-overlay), keyed by `task.data.idx`. Each
+    candidate is applied as a config-layer override via
+    `Taskset.apply_system_prompt(..., override=...)` — the same path as
+    `--taskset.system-prompt`. `loop` holds `env.serving()` open; `evaluate` drives
+    rollouts with `run_until_complete`.
+    """
 
     env: Environment
     ctx: ModelContext
@@ -53,9 +55,10 @@ def evaluate(
         candidate: Candidate,
         capture_traces: bool = False,
     ) -> EvaluationBatch[Trace, Trace]:
-        """Run `candidate`'s system prompt on the tasks named by `batch` (`Task.idx` values)
-        and score them. Called synchronously by GEPA on the main thread; each batch's rollouts
-        run on the runner's persistent loop via `run_until_complete`."""
+        """Run `candidate`'s config-layer system prompt on the tasks named by `batch`
+        (`Task.idx` values) and score them. Called synchronously by GEPA on the main
+        thread; each batch's rollouts run on the runner's persistent loop via
+        `run_until_complete`."""
         system_prompt = candidate.get("system_prompt", "")
         traces = self.loop.run_until_complete(self._run_batch(batch, system_prompt))
         scores = [trace.reward for trace in traces]
@@ -66,13 +69,11 @@ def evaluate(
         )
 
     async def _run_batch(self, batch: list[int], system_prompt: str) -> list[Trace]:
-        # Inject the candidate by rebuilding each Task around a data row with the new
-        # system_prompt (TaskData is frozen; behavior/config carry over unchanged).
         tasks = [
-            type(t)(
-                t.data.model_copy(update={"system_prompt": system_prompt}), t.config
+            self.env.taskset.apply_system_prompt(
+                self.tasks[idx], override=system_prompt
             )
-            for t in (self.tasks[idx] for idx in batch)
+            for idx in batch
         ]
         episodes = [self.env.episode(task, self.ctx, n=1) for task in tasks]
         results = await asyncio.gather(

verifiers/v1/gepa/config.py

@@ -1,7 +1,8 @@
 """The `GEPAConfig`: the single config object the `gepa` CLI parses.
 
-GEPA optimizes one taskset's `Task.system_prompt` by alternating rollouts (`evaluate`) with a
-teacher LM reflecting on the reflective dataset (`make_reflective_dataset`) — see
+GEPA optimizes the taskset config-layer system prompt (`taskset.system_prompt` /
+`system_prompt_file`) by alternating rollouts (`evaluate`) with a teacher LM reflecting
+on the reflective dataset (`make_reflective_dataset`) — see
 `verifiers.v1.gepa.adapter.GEPAAdapter`. This inherits `EnvConfig`'s fields (`taskset`,
 `harness`, `max_turns`, token limits, timeouts) as top-level flags, the same way `EvalConfig`
 does, and adds the optimization loop's own knobs (model, reflection model, train/val split,
@@ -55,9 +56,6 @@ class GEPAConfig(EnvConfig):
     """Train tasks sampled per reflection step."""
     reflection_columns: list[str] = Field(default_factory=list)
     """Extra per-trace fields (from `trace.info`, else `task`) to surface to the teacher LM."""
-    initial_prompt: str | None = None
-    """Seed system prompt. None = the first loaded task's `Task.system_prompt`, if any task
-    sets one (see `resolve_gepa_seed_prompt`)."""
 
     max_concurrent: int | None = Field(
         128, validation_alias=AliasChoices("max_concurrent", "c")
@@ -67,7 +65,7 @@ class GEPAConfig(EnvConfig):
         None, validation_alias=AliasChoices("output_dir", "o")
     )
     """Where to write results (config.toml + the streamed traces.jsonl, alongside GEPA's own
-    candidates.json / run_log.json). None = a fresh per-run dir under
+    candidates.json / run_log.json / best_system_prompt.txt). None = a fresh per-run dir under
     `outputs/<taskset>--<model>--<harness>/<uuid>` (via `output_path`)."""
     save_results: bool = True
     verbose: bool = Field(False, validation_alias=AliasChoices("verbose", "v"))

verifiers/v1/gepa/dataset.py

@@ -8,6 +8,7 @@
 
 from verifiers.v1.decorators import discover_decorated
 from verifiers.v1.task import Task
+from verifiers.v1.taskset import TasksetConfig, resolve_system_prompt
 
 
 def reject_group_reward_tasksets(tasks: list[Task]) -> None:
@@ -37,23 +38,26 @@ def split_tasks(
     return tasks[:num_train], tasks[num_train : num_train + num_val]
 
 
-def resolve_gepa_seed_prompt(tasks: list[Task], initial_prompt: str | None) -> str:
-    """The system prompt GEPA starts optimizing from: `initial_prompt` if given, else the first
-    task that sets `system_prompt`. Some tasksets (e.g. `gsm8k-v1`) bake instructions into
-    `prompt` rather than `system_prompt` and can't be optimized this way — pass `--initial-prompt`
-    to seed one explicitly.
+def resolve_gepa_seed_prompt(config: TasksetConfig, tasks: list[Task]) -> str:
+    """The config-layer system prompt GEPA starts from.
 
-    How the resolved prompt reaches the model at rollout time — a real system message vs. folded
-    into the user prompt — is the harness's call: `Harness.resolve_prompt` owns that policy and
-    warns when it folds, so it isn't re-policed here."""
-    if initial_prompt is not None:
-        return initial_prompt
+    Prefer `--taskset.system-prompt` / `--taskset.system-prompt-file` when set; otherwise
+    bootstrap from the first bake-in `TaskData.system_prompt` on `tasks` (expected to be
+    pre-overlay / `select(apply_config=False)`). Tasksets that only put instructions in
+    `prompt` (e.g. `gsm8k-v1`) need an explicit config seed.
+
+    How the prompt reaches the model — a real system message vs. folded into the user
+    prompt — is the harness's call (`Harness.resolve_prompt` / `compose_system_prompt`).
+    """
+    if (override := resolve_system_prompt(config)) is not None:
+        return override
     for task in tasks:
         if task.data.system_prompt is not None:
             return task.data.system_prompt
     raise ValueError(
-        "no task in this taskset sets Task.system_prompt — some tasksets bake instructions "
-        "directly into `prompt` instead (e.g. gsm8k-v1) and can't be optimized this way. Pass "
-        "--initial-prompt to seed one explicitly, or pick a taskset whose load() sets "
-        "system_prompt on its task data (e.g. reverse-text-v1, lean, textarena)."
+        "no config system prompt and no task sets TaskData.system_prompt — some "
+        "tasksets bake instructions into `prompt` instead (e.g. gsm8k-v1). Pass "
+        "--taskset.system-prompt or --taskset.system-prompt-file as the GEPA seed, "
+        "or pick a taskset that sets a bake-in system_prompt (e.g. reverse-text-v1, "
+        "lean, textarena)."
     )

verifiers/v1/gepa/runner.py

@@ -38,15 +38,18 @@ def log(self, message: str) -> None:
 
 def run_gepa(env: Environment, config: GEPAConfig) -> GEPAResult:
     logger.info("gepa config:\n%s", config.model_dump_json(indent=2))
-    all_tasks = env.taskset.select(config.num_train + config.num_val, config.shuffle)
+    all_tasks = env.taskset.select(
+        config.num_train + config.num_val,
+        config.shuffle,
+        apply_config=False,
+    )
     train_tasks, val_tasks = split_tasks(all_tasks, config.num_train, config.num_val)
     selected_tasks = [*train_tasks, *val_tasks]
     reject_group_reward_tasksets(
         selected_tasks
     )  # GEPA scores n=1 per task; groups need >=2
-    # Seed from the tasks GEPA actually evaluates (train ∪ val), not the full pre-split pool —
-    # a taskset with per-task system prompts could otherwise seed from a task in neither split.
-    seed_prompt = resolve_gepa_seed_prompt(selected_tasks, config.initial_prompt)
+    # Config-layer seed (explicit override, else bake-in from train ∪ val tasks).
+    seed_prompt = resolve_gepa_seed_prompt(env.taskset.config, selected_tasks)
     tasks_by_idx = {task.data.idx: task for task in selected_tasks}
 
     run_dir = output_path(config) if config.save_results else None
@@ -109,7 +112,13 @@ async def on_complete(trace: Trace) -> None:
                 skip_perfect_score=False,
                 logger=_GEPALog(),
             )
-            return optimize(**optimize_kwargs)
+            result = optimize(**optimize_kwargs)
+            if run_dir is not None:
+                best = result.best_candidate.get("system_prompt", "")
+                path = run_dir / "best_system_prompt.txt"
+                path.write_text(best, encoding="utf-8")
+                logger.info("best system prompt: %s", path)
+            return result
         finally:
             loop.run_until_complete(serving.__aexit__(None, None, None))
     finally:

verifiers/v1/harness.py

@@ -50,14 +50,25 @@ def resolved_env(self) -> dict[str, str]:
 
 class Harness(ABC, Generic[ConfigT]):
     APPENDS_SYSTEM_PROMPT: ClassVar[bool] = False
-    """Emit `TaskData.system_prompt` separately instead of folding it into the user prompt."""
+    """Emit the task-side system prompt as a separate system message instead of folding
+    it into the user prompt. (Harness-owned framing is composed separately via
+    `compose_system_prompt` — harness first, then task.)"""
     SUPPORTS_MCP: ClassVar[bool] = False
     SUPPORTS_USER_SIM: ClassVar[bool] = False
     SUPPORTS_MESSAGE_PROMPT: ClassVar[bool] = False
 
     def __init__(self, config: ConfigT) -> None:
         self.config = config
 
+    def system_prompt(self) -> str | None:
+        """Harness-owned framing (tools/protocol). None if none."""
+        return None
+
+    def compose_system_prompt(self, task_system: str | None) -> str | None:
+        """Harness framing first, then the task-side system prompt."""
+        parts = [p for p in (self.system_prompt(), task_system) if p]
+        return "\n\n".join(parts) or None
+
     def resolve_prompt(
         self, task: TaskData
     ) -> tuple[str | None, str | Messages | None]:

verifiers/v1/harnesses/default/harness.py

@@ -42,6 +42,14 @@ class DefaultHarness(Harness[DefaultHarnessConfig]):
     SUPPORTS_USER_SIM = True
     SUPPORTS_MESSAGE_PROMPT = True
 
+    def system_prompt(self) -> str | None:
+        fragments = [BASH_SYSTEM_PROMPT]
+        if self.config.edit:
+            fragments.append(EDIT_SYSTEM_PROMPT)
+        if self.config.search:
+            fragments.append(SEARCH_PROMPT)
+        return " ".join(fragments)
+
     async def setup(self, runtime: Runtime) -> None:
         await runtime.prepare_uv_script(PROGRAM_SOURCE, self.config.resolved_env)
 
@@ -54,15 +62,8 @@ async def launch(
         secret: str,
         mcp_urls: dict[str, str],
     ) -> ProgramResult:
-        system_prompt, prompt = self.resolve_prompt(trace.task.data)
-        fragments = [BASH_SYSTEM_PROMPT]
-        if self.config.edit:
-            fragments.append(EDIT_SYSTEM_PROMPT)
-        if self.config.search:
-            fragments.append(SEARCH_PROMPT)
-        system_prompt = "\n\n".join(
-            p for p in (" ".join(fragments), system_prompt) if p
-        )
+        task_system, prompt = self.resolve_prompt(trace.task.data)
+        system_prompt = self.compose_system_prompt(task_system)
         env = {**self.config.resolved_env}
         args = [
             f"--base-url={endpoint}",

verifiers/v1/serve/server.py

@@ -40,7 +40,7 @@ def __init__(
         # A finite taskset is materialized up front (its count is served via `info`); an
         # infinite one is pulled off its generator on demand (see `_task`), so
         # `num_tasks=None` on the wire ⟺ the taskset is infinite.
-        self._task_iter = iter(self.env.taskset.load())
+        self._task_iter = iter(self.env.taskset.iter_tasks())
         self._tasks: list = []
         self.num_tasks: int | None = None
         if not type(self.env.taskset).INFINITE:
@@ -88,10 +88,10 @@ def run_server(cls, address_queue=None, **kwargs) -> None:
     def _task(self, idx: int):
         """The task at `idx`; an infinite taskset is generated (and cached) up to `idx`
         on demand. Generation must be deterministic — every pool worker runs its own
-        `load()`, so idx-addressing relies on all of them producing the same sequence.
-        Lazy generation is capped at `MAX_LAZY_TASKS`: an idx that far ahead is a
-        runaway driver, and generating (and caching) toward it would hang the worker
-        and exhaust memory instead of failing the one request."""
+        `iter_tasks()`, so idx-addressing relies on all of them producing the same
+        sequence. Lazy generation is capped at `MAX_LAZY_TASKS`: an idx that far ahead
+        is a runaway driver, and generating (and caching) toward it would hang the
+        worker and exhaust memory instead of failing the one request."""
         while len(self._tasks) <= idx:
             if idx >= MAX_LAZY_TASKS:
                 raise IndexError(