diff --git a/AGENTS.md b/AGENTS.md index 004b6965..39d80f4f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,7 +15,7 @@ SiEval is a **model delivery quality verification system**: eval-side knowledge Precise reproducibility is a product contract, not a nicety. - Safety guards (e.g. the resume stage comparison) ship strict-only. No `--force-*` flags, no bypass env vars, no "(future)" escape-hatch hints in errors. -- A resume (`--resume`, `--resume DIR`) compares stage identities, not config text (`sieval/core/runners/stage_resume.py`), and reruns from the earliest changed stage: changed data or preprocess → those samples rerun; changed postprocess, feedback or report → rerun from that stage on the recorded inference (report only: recomputed from the records, no engine or judge); changed inference, or an unknown stage it would reuse → refused. Before that, `meta.json` `result_schema_version` must equal `RESULT_SCHEMA_VERSION` (absent: mapped from `version`, 0.10.0+ is 1); the package version is provenance only. `effective_config.yaml` and `infer_plans.yaml` are audit records; config changes go to `meta.json` `runs[].config`. +- A resume (`--resume`, `--resume DIR`) compares stage identities, not config text (`sieval/core/runners/stage_resume.py`), and reruns from the earliest changed stage: changed data → those samples rerun; changed preprocess (its code and arguments, not the rendered prompts) → every sample reruns; changed postprocess, feedback or report → rerun from that stage on the recorded inference (report only: recomputed from the records, no engine or judge); changed inference, or an unknown stage it would reuse → refused. Before that, `meta.json` `result_schema_version` must equal `RESULT_SCHEMA_VERSION` (absent: mapped from `version`, 0.10.0+ is 1); the package version is provenance only. `effective_config.yaml` and `infer_plans.yaml` are audit records; config changes go to `meta.json` `runs[].config`. - Recovery: "start fresh" or "match the stage the refusal names". Escape-hatch proposals must re-justify the contract, not just add a flag. ## Result Schema Version @@ -25,7 +25,9 @@ Bump `RESULT_SCHEMA_VERSION` (`sieval/core/tasks/run_meta.py`) in the PR that ma - a change to a task result directory's layout (shards, `.idx`, `meta.json`, the `report.json` shape) that older directories cannot be read under; - a core change that alters results. The grading identities do not walk `sieval.core` or `sieval.datasets`, so nothing else catches it. -`IDENTITY_SCHEMA` versions only the `identity` block. Bumping it does not refuse a resume, so a new schema must read the old one: schema 2 reads schema 1's `grader` through `legacy_grader_keys`, which keeps the old walks unchanged. +`IDENTITY_SCHEMA` versions only the `identity` block. Bumping it does not refuse a resume, so a new schema must read the old one: schema 2 reads schema 1's `grader` through `legacy_grader_keys`, which keeps the old walks unchanged; schema 3 reads the rendered-prompt `preprocess` of schemas 1 and 2 by rendering the recorded samples once. + +`Model.extra` is the capability contract between tasks and models (`sieval/core/models/extra.py`): a task reads a key in `preprocess`, an absent key changes nothing, and `core/` only checks the shape of documented keys. ## Project Structure & Module Organization diff --git a/CHANGELOG.md b/CHANGELOG.md index b927f211..c18e4a82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,17 +11,20 @@ Planned as 0.12.0. ### Migrating from 0.11.x -- **`meta.json` `identity` is schema 2**: `data`, `preprocess` (was `prompt`), `inference`, `postprocess`, `feedback`, `report`. The combined `grader` is gone; judges and services move to `feedback`. Readers must accept schema 1 too. `RESULT_SCHEMA_VERSION` stays 1, so 0.10.x and 0.11.0 directories resume, and an unchanged task's schema-1 directory upgrades in place without regrading. +- **`meta.json` `identity` is schema 3** (#211, #218): `data`, `preprocess` (was `prompt`), `inference`, `postprocess`, `feedback`, `report`. The combined `grader` is gone; judges and services move to `feedback`. `preprocess` is `{digest, modules, args, splits?, fallback?}`, the code and arguments that render a prompt, instead of a digest of the rendered prompts. Each `preprocess` record is stamped `{data, preprocess}` instead of `{data, prompt}`. Readers must accept schema 1 too (and schema 2, written only by unreleased `main` builds). `RESULT_SCHEMA_VERSION` stays 1, so 0.10.x and 0.11.0 directories resume: an unchanged task's directory upgrades in place without regrading, after one local `preprocess` call per recorded sample. - `TaskSaver.begin_run(identity={...})` and `set_identity(**stages)` replace the per-stage keyword arguments. ### Changed +- **`preprocess` is identified by code, not by the prompts it renders** (#218): the static walk the grading stages use, the `grader_args()` it reads, and for few-shot a digest of the split it draws from. A model's `extra` reaches the request, which `inference` digests, so a wrapping and a plain model share `preprocess` and differ in `inference`. A run-time name or a file opened at run time falls back to the task's whole code and the data files beside the reader. **A preprocess change now restarts every sample of the task**, where before only the samples whose prompt changed restarted. + - **A resume reruns from the earliest stage that changed** (#210). An extraction change reruns from `postprocess`, a scoring change (judge, prompt, sampling, service, `feedback` code) from `feedback`, and a `report`-only change recomputes `report.json` from the records with no engine or judge call. Before, each of these regraded every sample. Code is attributed to stages by a static walk; a run-time lookup (`getattr(self, )`, `eval`, `importlib`) falls back to all grading code. No built-in task falls back. - `runs[].resume` counts what a resume did with each recorded sample. - **`sieval run` starts an engine only for a model with samples to infer** (closes #213). A run that may resume (`--resume`, or `resume` in a `runner_config`) first plans each task's resume, read-only, through the runner's own resume planning (`TaskRunner.plan_run`), and launches only the engines whose models some task still has samples to infer with. A resume that reruns only grading stages, or re-reports, starts no engine and runs on a node without GPUs; when some sample needs inference and `nvidia-smi` finds no GPU, the run is refused before anything starts. The plans, `infer_plans.yaml`, the inference identity and its request digest are what an eager launch records. A fresh run is not planned and launches as before. A sample that `feedback` sends to another iteration after a rerun from `feedback` fails at `infer` when its engine was not started, and a resume that starts it retries it. ### Added +- **`Model.extra` is the capability contract between tasks and models, starting with `sequence_wrappers`** (#218). A task asks for a capability by key and does nothing different when it is absent. `sequence_wrappers: {: ""}` (types `dna`, `rna`, `protein`, `mol`) gives the special tokens a model was trained with around biological sequences; a task reads it with `sequence_wrapper(self.model.extra, )` in `preprocess`. A malformed template is refused when the model is built and by `--dry-run`. No built-in task injects such sequences yet. - **`SIEVAL_TARGET_GPU` declares the GPUs a serving plan is resolved for**, as `::` with the name and memory `nvidia-smi` reports, e.g. `NVIDIA H100 80GB HBM3:8:81559`. Recipe resolution reads it instead of `nvidia-smi`, on any node, so a CPU node resolves the plan, and the inference identity, a node with those GPUs would. Unset, nothing changes. Declared, an engine launches only on a node with exactly those GPUs. `infer_plans.yaml` gains `gpu`, the GPUs its plans were resolved for, outside every identity; a `sieval run --resume` declaring the same GPUs reuses the recorded plan of each served model whose `path`, `infer`, capability layer and `deterministic` are unchanged, unless the plan held an engine secret, which `infer_plans.yaml` records redacted (#215, closes #212). - `sieval.core.runners.task_resume_plan` / `plan_resume`: the per-stage resume plan, without running anything. - `TaskRunner.plan_run` / `MultiTaskRunner.plan_run` (`RunPlan`): the runner's per-sample plan of a run, `to_infer` included, read-only; the one `sieval run` launches by. diff --git a/docs/guide/configuration.md b/docs/guide/configuration.md index 1124e205..e88e459e 100644 --- a/docs/guide/configuration.md +++ b/docs/guide/configuration.md @@ -191,6 +191,49 @@ true, top_logprobs: N}`), each input token's top-N lands in the generated tokens'. Served by `sglang_legacy` and `openai_completions`. The reply grows by N entries per input token. +## Model Capabilities (`extra`) + +`extra` is a dict on a model that tasks read and the API never sees: a task +asks for a capability by key and does nothing different when it is absent. +SiEval records it in the inference identity and checks the shape of the keys +below, so a typo fails before the run. Set it on a model (top level or +`args`) or per task in `infer_args`; a task's `infer_args.extra` replaces the +model's. + +```yaml +models: + naturelm: + name: org/naturelm + api_base: http://127.0.0.1:30000/v1 + extra: + sequence_wrappers: + dna: "{seq}" + rna: "{seq}" +``` + +| Key | Value | Read by | +| --- | --- | --- | +| `sequence_wrappers` | `{: ""}`; types `dna`, `rna`, `protein`, `mol` (SMILES) | Tasks that put a biological sequence in a prompt | + +`sequence_wrappers` is for models trained with special tokens around +biological sequences (NatureLM, domain-tokenizer models); without them the +score degrades silently. Each value is a format string whose only field is +`{seq}` (literal braces as `{{` and `}}`). + +A task wraps each sequence it injects in `preprocess`: + +```python +from sieval.core.models import sequence_wrapper + +async def preprocess(self, raw, ctx): + wrap = sequence_wrapper(self.model.extra, "dna") + seq = wrap.format(seq=raw["sequence"]) if wrap else raw["sequence"] + return build_prompt_record([{"role": "user", "content": PROMPT.format(seq=seq)}]) +``` + +With no wrapper the prompt is unchanged. A wrapping model and a plain one +share `data` and `preprocess` and differ in `inference` (see [Resume](#resume)). + ## Task Pipeline Each task implements a typed 5-stage pipeline: @@ -261,7 +304,8 @@ run's, and reruns each sample from the earliest stage that changed: | Differs | Resume | | --- | --- | -| `data` (the selected samples) or `preprocess` (the prompt) | Those samples rerun from the start; the rest are kept | +| `data` (the selected samples) | Those samples rerun from the start; the rest are kept | +| `preprocess` (prompt rendering: its code and the arguments it reads) | Every sample reruns from the start | | `inference` | Refused, naming the stage | | `postprocess` (answer extraction) | `postprocess`, `feedback` and `report` rerun on the recorded inference | | `feedback` (scoring: judges, their prompts and sampling, services) | `feedback` and `report` rerun on the recorded `postprocess` | @@ -276,15 +320,34 @@ class and `grader_args()`; `feedback` also holds every judge and service. those in `GRADER_ARGS_EXCLUDED`; a task extends it with arguments that cannot change a verdict: `Task.GRADER_ARGS_EXCLUDED | {"grade_timeout"}`. +The `preprocess` identity is the code `preprocess` reaches, the +`grader_args()` it reads (all of them when the task overrides `grader_args`) +and, for few-shot, a digest of the `fewshot_split` it draws from (every split +but `test` when it names none). It is computed from code at setup, not from +the prompts: the model's `extra` reaches the request, which `inference` +digests. A preprocess change restarts every sample of the task. + +The walk is static. A prompt that reads an environment variable, or a file +through a reader it cannot see (`pd.read_csv`), is covered only through the +argument naming it: keep such inputs in task arguments. + Code is attributed to stages by a static walk from each stage method; code two stages reach counts in both. A stage that reaches code by a name only known at run time (`getattr(self, name)`, `eval`, `importlib`) counts all grading code, and so does every stage after it; `stage_code()` names the construct. +`preprocess` falls back alone, and also when its code opens a file at run time +(`open`, `read_text`, `importlib.resources`, a directory listing, a database +connection): it then counts the task's whole code and the data files beside +the module that reads one. A schema-1 `meta.json` (0.10.0 to 0.11.x) folds the grading stages into one `grader`. If it still matches the task's code, every verdict is kept and the block is upgraded in place; otherwise every sample regrades. +Schemas 1 and 2 recorded `preprocess` as rendered-prompt digests. A resume +renders each recorded sample once: if all match, the block is upgraded to +schema 3 in place; otherwise every sample reruns. + A same-directory resume reruns a sample with no recorded identity, such as one interrupted before `preprocess`. Under `record_meta: false` no sample has one, so every recorded sample reruns, with a warning. @@ -349,11 +412,12 @@ in `SIEVAL_TARGET_GPU` ([Resolving for declared GPUs](infer.md#resolving-for-dec "samples": 3, "resume_plan": {"kept": 2, "rereported": 0, "regraded": 0, "repostprocessed": 0, "restarted": 0, "dropped": 0}, "stages": { - "identity_schema": 2, - "stages": {"data": "differs", "inference": "same", "postprocess": "same", "feedback": "same", "report": "same"}, + "identity_schema": 3, + "stages": {"data": "differs", "preprocess": "same", "inference": "same", "postprocess": "same", "feedback": "same", "report": "same"}, "rerun_from": null, "refused": null, - "legacy_grader": null + "legacy_grader": null, + "legacy_preprocess": false }, "new": 1, "to_infer": 1, @@ -378,10 +442,12 @@ in `SIEVAL_TARGET_GPU` ([Resolving for declared GPUs](infer.md#resolving-for-dec counts the samples sent to the model under test, `to_grade` those run through a grading stage; `needs_inference` is whether `to_infer` is above zero. - `tasks..stages` compares each stage's recorded identity with this - run's: `same`, `differs` or `unknown`. `rerun_from` is the grading stage the - resume reruns from (`report` for a re-report), else `null`; a data change - shows in `resume_plan` as `restarted`. `legacy_grader` is `kept` or `regrade` - for a schema-1 directory. `stages` is `null` for a fresh start. + run's: `same`, `differs` or `unknown`. `rerun_from` is the stage the resume + reruns from (`preprocess` restarts every sample, `report` is a re-report), + else `null`; a data change shows in `resume_plan` as `restarted`. + `legacy_grader` is `kept` or `regrade` for a schema-1 directory; + `legacy_preprocess` is `true` for a schema-1 or schema-2 directory, compared + by rendering its recorded samples once. `stages` is `null` for a fresh start. - `needs_inference` is whether the run starts any engine. A task that infers through an `api_base` model counts in its `to_infer` but starts none. diff --git a/sieval/cli/validation.py b/sieval/cli/validation.py index 3d13d42d..dfefac41 100644 --- a/sieval/cli/validation.py +++ b/sieval/cli/validation.py @@ -33,6 +33,7 @@ validate_task_config_args, ) from sieval.cli.session import RootConfigDict +from sieval.core.models import check_extra from sieval.core.models.capabilities import ( CAPABILITY_KEYS, CapabilityConfigError, @@ -209,6 +210,18 @@ def _validate_models(cfg: dict, result: ValidationResult) -> None: except (TypeError, ValueError) as exc: result.errors.append(str(exc)) + raw_model_args = mcfg.get("args") + for where, extra in ( + (f"Model '{name}' extra", mcfg.get("extra")), + ( + f"Model '{name}' args.extra", + raw_model_args.get("extra") + if isinstance(raw_model_args, dict) + else None, + ), + ): + _check_model_extra(extra, where, result) + if has_base: base_ref = mcfg["base"] if not isinstance(base_ref, str) or not base_ref: @@ -523,6 +536,19 @@ def _validate_tasks(cfg: dict, result: ValidationResult) -> None: f"Task '{name}' infer_args cannot change binding resources: " + ", ".join(misplaced) ) + _check_model_extra( + infer_args.get("extra"), f"Task '{name}' infer_args.extra", result + ) + + +def _check_model_extra(extra: object, where: str, result: ValidationResult) -> None: + """Report an ``extra`` whose documented keys are malformed + (:func:`~sieval.core.models.check_extra`), such as a wrapper without + ``{seq}``.""" + try: + check_extra(extra, where) + except ValueError as exc: + result.errors.append(str(exc)) def _validate_runner_config(cfg: dict, result: ValidationResult) -> None: diff --git a/sieval/core/models/__init__.py b/sieval/core/models/__init__.py index bdd0776a..08280753 100644 --- a/sieval/core/models/__init__.py +++ b/sieval/core/models/__init__.py @@ -50,6 +50,7 @@ get_dialect_spec, ) from .exceptions import CapabilityError +from .extra import SEQUENCE_TYPES, SEQUENCE_WRAPPERS, check_extra, sequence_wrapper from .gen_model import GenModel from .ir import ( FINISH_SOURCE_CANDIDATE, @@ -145,6 +146,8 @@ "FINISH_SOURCE_CANDIDATE", "REQUEST_ENVELOPE_HTTP", "REQUEST_ENVELOPE_KEY", + "SEQUENCE_TYPES", + "SEQUENCE_WRAPPERS", "FINISH_SOURCE_PROMPT_BLOCK", "FunctionToolCall", "GenModel", @@ -198,6 +201,8 @@ "TopKEntry", "Transport", "UsageStats", + "check_extra", + "sequence_wrapper", "ASYNC_HTTP_JSON_FACTORY", "AsyncHTTPJSONConnection", "BindingReconcileInput", diff --git a/sieval/core/models/extra.py b/sieval/core/models/extra.py new file mode 100644 index 00000000..9482e6b9 --- /dev/null +++ b/sieval/core/models/extra.py @@ -0,0 +1,94 @@ +"""``Model.extra``: what a model offers the tasks that evaluate it. + +A dict on the model, never sent to the API. A task asks for a capability by +key and does nothing different when it is absent. Core passes the dict through +and records it in the inference identity; it only checks the shape of the +keys documented here, so a typo is refused before a run. + +====================== ====================================================== +``sequence_wrappers`` ``{: ""}``: the special + tokens a model was trained with around biological + sequences. Read with :func:`sequence_wrapper`. +====================== ====================================================== + +AI-Generated Code - Claude Opus 5.5 (Anthropic) +""" + +import string +from collections.abc import Mapping +from typing import cast + +from sieval.core.types import JSONValue + +#: ``extra`` key of the sequence wrappers a model needs. +SEQUENCE_WRAPPERS = "sequence_wrappers" + +#: Sequence types a task reads from ``sequence_wrappers``: DNA, RNA (mRNA and +#: oligonucleotides included), protein, and molecules as SMILES. +SEQUENCE_TYPES: tuple[str, ...] = ("dna", "rna", "protein", "mol") + + +def sequence_wrapper(extra: Mapping[str, JSONValue], seq_type: str) -> str | None: + """The model's template for *seq_type* in ``Model.extra``, or ``None``. + + ``wrap.format(seq=...)`` wraps a sequence; with ``None`` a task renders + what it renders for a plain model. + """ + wrappers = extra.get(SEQUENCE_WRAPPERS) + if not isinstance(wrappers, Mapping): + return None + template = cast(Mapping[str, object], wrappers).get(seq_type) + return template if isinstance(template, str) else None + + +def check_extra(extra: object, where: str = "extra") -> None: + """Raise :class:`ValueError` when a key documented here has the wrong shape; + other keys are not checked.""" + if extra is None: + return + if not isinstance(extra, Mapping): + raise ValueError(f"{where} must be a mapping, got {type(extra).__name__}") + keyed = cast(Mapping[str, object], extra) + if SEQUENCE_WRAPPERS in keyed: + wrappers = keyed[SEQUENCE_WRAPPERS] + check_sequence_wrappers(wrappers, f"{where}.{SEQUENCE_WRAPPERS}") + + +def check_sequence_wrappers(wrappers: object, where: str = SEQUENCE_WRAPPERS) -> None: + """Refuse ``sequence_wrappers`` that are not ``{type: template}``: each type + in :data:`SEQUENCE_TYPES`, each template a format string whose only field + is ``{seq}``.""" + if not isinstance(wrappers, Mapping): + raise ValueError( + f"{where} must map a sequence type to a template, " + f"got {type(wrappers).__name__}" + ) + for seq_type, template in cast(Mapping[object, object], wrappers).items(): + key = f"{where}.{seq_type}" + if seq_type not in SEQUENCE_TYPES: + raise ValueError( + f"{key}: unknown sequence type {seq_type!r}; " + f"expected one of {', '.join(SEQUENCE_TYPES)}" + ) + if not isinstance(template, str): + raise ValueError( + f"{key} must be a format string, got {type(template).__name__}" + ) + try: + fields = [f for _, f, _, _ in string.Formatter().parse(template)] + except ValueError as exc: + raise ValueError( + f"{key}: {template!r} is not a format string ({exc})" + ) from None + named = [f for f in fields if f is not None] + if "seq" not in named: + raise ValueError( + f"{key}: {template!r} has no {{seq}}, so the sequence would be " + "dropped; write e.g. '{seq}'" + ) + others = sorted({f for f in named if f != "seq"}) + if others: + raise ValueError( + f"{key}: {template!r} may only use {{seq}}, not " + + ", ".join(f"{{{f}}}" for f in others) + ) diff --git a/sieval/core/models/model.py b/sieval/core/models/model.py index 03133665..b01897db 100644 --- a/sieval/core/models/model.py +++ b/sieval/core/models/model.py @@ -60,6 +60,7 @@ get_dialect_spec, ) from .exceptions import CapabilityError +from .extra import check_extra from .ir import ( CapabilityEvidence, ChatInput, @@ -368,6 +369,7 @@ def _initialize( self._model = runtime_plan.requested_model_id self._api_base = api_base self._kwargs = _checked_builder_defaults(builder_defaults) + check_extra(extra) self._extra = dict(extra) if extra is not None else None self._limiter = local_limiter self._parent_limiter = parent_limiter @@ -442,6 +444,7 @@ def with_args( new_model = copy.copy(self) new_model._kwargs = _checked_builder_defaults({**self._kwargs, **kwargs}) if extra is not None: + check_extra(extra) new_model._extra = dict(extra) if concurrency_limit is not None: new_model._limiter = anyio.CapacityLimiter(concurrency_limit) @@ -659,6 +662,8 @@ def get_quota_info(self) -> ModelQuotaInfo: @property def extra(self) -> dict[str, JSONValue]: + """What this model offers the tasks evaluating it, by key; never sent + (:mod:`sieval.core.models.extra`).""" return dict(self._extra) if self._extra else {} def meta(self) -> ModelMeta: diff --git a/sieval/core/runners/runner.py b/sieval/core/runners/runner.py index cd904647..5c3f8fc2 100644 --- a/sieval/core/runners/runner.py +++ b/sieval/core/runners/runner.py @@ -4,7 +4,7 @@ import re import time from collections import defaultdict -from collections.abc import Callable +from collections.abc import Callable, Iterable from dataclasses import dataclass, replace from datetime import datetime from pathlib import Path @@ -65,6 +65,7 @@ GradingIdentity, InferenceIdentity, JudgeIdentity, + PreprocessIdentity, ResumeOutcomes, RunFrom, RunRequests, @@ -76,8 +77,10 @@ grader_task, grading_identities, inference_identity, + is_code_identity, judge_identities, legacy_grader_keys, + preprocess_key, prompt_of, request_digest, stage_code, @@ -120,6 +123,7 @@ recorded_grader, recorded_identity, recorded_sample_digests, + task_preprocess_identity, ) #: Key of the run's own counts in ``report.json``. @@ -522,6 +526,12 @@ def __init__( self._grading: dict[str, GradingIdentity] | None = None self._stage_code: dict | None = None self._legacy_keys: set[str] | None = None + # This run's preprocess identity (`False` until computed); whether a + # recorded sample's preprocess changed; for a schema-1/2 directory, + # whether every recorded sample still renders its recorded prompt. + self._preprocess: PreprocessIdentity | None | Literal[False] = False + self._preprocess_restarted = False + self._legacy_prompts_match: bool | None = None self._stage_stamps: dict[str, str | None] = dict.fromkeys(GRADING_STAGES) # Context overflows an earlier run recorded as failed inference, by # sample: a resume scores them from the recorded refusal (`_infer`). @@ -529,13 +539,10 @@ def __init__( self._data_match: bool | None = None self._resume_plan: ResumeOutcomes | None = None self._unverified_inference = 0 - # Run history: actions executed this run, each sample's data and prompt - # digests as of its first preprocess, and judge versions the API - # reported, by role. + # Run history: actions executed this run, each sample's data digest as + # of its first preprocess, and judge versions the API reported, by role. self._ran_actions: set[TaskAction] = set() - self._sample_digests: dict[ - str | int, tuple[int | None, str | None, str | None] - ] = {} + self._sample_digests: dict[str | int, tuple[int | None, str | None]] = {} self._input_identity_written = False self._judge_versions: dict[str, str] = {} self._judge_roles: dict[str, list[str]] | None = None @@ -896,11 +903,13 @@ async def _begin_run(self, resumed: bool) -> list[str | int]: Recorded before any work, so an interrupted run is resumed and the resume gates have a ``meta.json`` to read. Inference and grader - identities are known now; data and prompt once every sample is past + identities are known now, and so is preprocess unless a recorded + sample's changed (then at the end); data once every sample is past preprocess. Returns the samples a resume rewinds. """ inference = self._inference_identity() grading = self._grading_identities() + preprocess = self._preprocess_identity() changed: list[str | int] = [] keeps_recorded_model = False if resumed: @@ -924,7 +933,9 @@ async def _begin_run(self, resumed: bool) -> list[str | int]: config_path=self._config_path(), identity={ "data": None, - "preprocess": None, + # Unknown while a sample stamped with another preprocess may + # remain: `_sample_preprocess_changed` trusts a recorded one. + "preprocess": None if self._preprocess_restarted else preprocess, "inference": inference, **{ stage: None @@ -1032,7 +1043,7 @@ async def _fast_resume(self) -> JSONValue | None: model=self._model_name(), config_diff=self._resume_config_diff, config_path=self._config_path(), - identity=self._grading_identity_update(), + identity=self._resume_identity_update(), ) self._contexts = await self._loader.load_initial_state() await self._loader.hydrate( @@ -1080,6 +1091,8 @@ async def _reusable_report(self) -> tuple[JSONValue, bool] | None: rereport = False if cached_report is not None: match = self._recorded_stages_match() + if match is not None and not await self._recorded_preprocess_matches(): + match = None if match is None: logger.info( "report.json found, but its run's stage identities differ " @@ -1108,7 +1121,7 @@ async def _rereport(self) -> JSONValue: model=self._model_name(), config_diff=self._resume_config_diff, config_path=self._config_path(), - identity=self._grading_identity_update(report_unknown=True), + identity=self._resume_identity_update(report_unknown=True), ) self._contexts = await self._loader.load_initial_state() await self._loader.hydrate( @@ -1437,10 +1450,7 @@ def _run_stage( if overflows: meta[CONTEXT_OVERFLOW] = list(overflows) if stage == TaskStage.PREPROCESSED: - meta["identity"] = { - "data": self._safe_digest(ctx.raw_sample), - "prompt": self._safe_digest(prompt_of(inner)), - } + meta["identity"] = self._input_stamp(self._safe_digest(ctx.raw_sample)) elif stage in (TaskStage.POSTPROCESSED, TaskStage.FEEDBACK): # Per sample, so a resume can tell which code graded it. meta["identity"] = self._grading_stamps(stage) @@ -1955,21 +1965,85 @@ def _compute_legacy_grader_keys(self) -> set[str]: logger.warning("Legacy grader identity unavailable: {}", exc) return set() - def _grading_identity_update(self, *, report_unknown: bool = False) -> dict: - """Grading identities a run that grades nothing writes at its start. - - Each stage identity this run computes, so a schema-1 block becomes - schema 2 in place; ``report`` stays unknown until a re-report writes - it. Empty when they cannot be computed: the carried ones stay. - """ + def _resume_identity_update(self, *, report_unknown: bool = False) -> dict: + """This run's preprocess and grading identities, which a fast resume or + re-report writes at its start to upgrade an older block in place. + ``report`` stays unknown until a re-report writes it.""" + update: dict = {"preprocess": self._preprocess_identity()} grading = self._grading_identities() - if grading is None: - return {} - update: dict = dict(grading) - if report_unknown: - update["report"] = None + if grading is not None: + update.update(grading) + if report_unknown: + update["report"] = None return update + async def _recorded_preprocess_matches(self) -> bool: + """Whether a finished run's ``preprocess`` is this run's: by + :func:`preprocess_key`, or for rendered prompts (schemas 1 and 2) by + rendering every recorded sample once.""" + identity = recorded_identity(self._recorded) + if identity is None: + return True + recorded = identity["preprocess"] + if is_code_identity(recorded): + mine = preprocess_key(self._preprocess_identity()) + return mine is not None and mine == preprocess_key(recorded) + if self._legacy_prompts_match is not None: + return self._legacy_prompts_match + contexts = await self._loader.load_initial_state() + await self._loader.hydrate( + contexts, + set(), + include_stages={TaskStage.FINAL, TaskStage.FAILED}, + prepare_retries=False, + record_each_stage=self._record_each_stage, + ) + self._legacy_prompts_match = await self._legacy_prompts_render( + contexts.values() + ) + return self._legacy_prompts_match + + async def _legacy_prompts_render(self, contexts: Iterable[TaskContext]) -> bool: + """Whether every unstamped recorded sample of *contexts* renders the + prompt it recorded: one ``preprocess`` call each. Samples whose data + changed rerun anyway and are skipped.""" + checked = 0 + for ctx in contexts: + if ctx.preprocess_result is None: + continue + ctx = self._ensure_raw_sample(ctx) + stamps = recorded_sample_digests(ctx) + if stamps["preprocess"] is not None: + continue + data = stamps["data"] + if stamps["source"] == "derived": + data = ( + self._safe_digest(ctx.raw_sample) + if self._recorded_data_matches() + else None + ) + if data is None or data != self._safe_digest(ctx.raw_sample): + continue + checked += 1 + if stamps["prompt"] is None or ( + stamps["prompt"] != await self._current_prompt_digest(ctx) + ): + logger.info( + "{}: sample {} renders another prompt than it recorded; " + "preprocess changed, every sample reruns", + self._task.name, + ctx.sample_id, + ) + return False + if checked: + logger.info( + "{}: {} recorded prompt(s) re-rendered unchanged; preprocess " + "identity upgraded in place", + self._task.name, + checked, + ) + return True + def _check_inference(self, current: InferenceIdentity | None) -> None: """Refuse a resume whose inference differs from the recorded run's.""" identity = recorded_identity(self._recorded) @@ -2075,18 +2149,28 @@ async def _plan_run(self) -> RunPlan: def _stage_plan(self) -> ResumePlan: """:func:`plan_resume` over this runner's identities, with its decisions. - ``preprocess`` is decided per sample (``restarted``), so it is left - out. ``rerun_from`` is the runner's, which trusts a directory with no - identities; ``refused`` is ``None``, as a refusal raises before this. + ``preprocess`` is compared as code; a schema-1/2 one takes the verdict + of the rendering pass planning made. ``rerun_from`` is the runner's, + which trusts a directory with no identities; ``refused`` is ``None``, + as a refusal raises before this. """ current: dict[str, object] = { "data": self._dataset_identity(), + "preprocess": self._preprocess_identity(), "inference": self._inference_identity(), **(self._grading_identities() or {}), } plan = plan_resume(self._recorded, current, self._legacy_grader_keys()) - del plan["stages"]["preprocess"] - plan["rerun_from"] = self._recorded_grading_change() + if plan["legacy_preprocess"] and self._legacy_prompts_match is not None: + plan["stages"]["preprocess"] = ( + "same" if self._legacy_prompts_match else "differs" + ) + changed = plan["stages"]["preprocess"] == "differs" + plan["rerun_from"] = ( + "preprocess" + if changed or self._preprocess_restarted + else self._recorded_grading_change() + ) plan["refused"] = None return plan @@ -2168,8 +2252,8 @@ def _recorded_stages_match(self) -> Literal["all", "report"] | None: ``report`` all but the report, ``None`` otherwise. Same data, inference, postprocess and feedback reuse every sample. - Preprocess is covered by the code the stages digest: a prompt - template lives in the task module or the arguments it digests. + Preprocess is compared after, by :meth:`_recorded_preprocess_matches`, + which may render the recorded samples. """ identity = recorded_identity(self._recorded) if identity is None: @@ -2283,7 +2367,8 @@ async def _plan_stage_resume(self) -> list[str | int]: """Decide, per recorded sample, what this resume keeps. Each sample rolls back to the earliest stage that changed for it: - changed data or preprocess restarts it; a changed postprocess rolls it + changed data or preprocess restarts it (a preprocess change restarts + every sample: all were stamped alike); a changed postprocess rolls it back to its recorded inference, a changed feedback to its recorded postprocess; a sample past iteration 0 restarts instead, since its later inference saw the old feedback. A changed report keeps every @@ -2294,6 +2379,19 @@ async def _plan_stage_resume(self) -> list[str | int]: size = len(test_set) if test_set else 0 changed: list[str | int] = [] counts = _no_outcomes() + if ( + self._legacy_prompts_match is None + and not self._trusts_unrecorded() + and not is_code_identity( + (recorded_identity(self._recorded) or {}).get("preprocess") + ) + ): + in_range = [ + c + for sid, c in self._contexts.items() + if isinstance(sid, int) and 0 <= sid < size + ] + self._legacy_prompts_match = await self._legacy_prompts_render(in_range) for sid, ctx in list(self._contexts.items()): if not isinstance(sid, int) or not 0 <= sid < size: @@ -2341,8 +2439,9 @@ async def _plan_sample( ctx = self._ensure_raw_sample(ctx) restart = self._task.make_context(ctx.sample_id) current_data = self._safe_digest(ctx.raw_sample) - data, prompt, source = recorded_sample_digests(ctx) - derived = source == "derived" + stamps = recorded_sample_digests(ctx) + data = stamps["data"] + derived = stamps["source"] == "derived" if derived: # No per-sample identity on record (a run before them, or one with # `record_meta` off): the recorded data identity of the whole set, @@ -2365,11 +2464,11 @@ async def _plan_sample( if data is None or data != current_data: return "restarted", restart if ctx.preprocess_result is not None: - current_prompt = await self._current_prompt_digest(ctx) - if prompt is None or prompt != current_prompt: + if self._sample_preprocess_changed(stamps["preprocess"]): + self._preprocess_restarted = True return "restarted", restart if derived: - ctx = self._stamp_input(ctx, current_data, current_prompt) + ctx = self._stamp_input(ctx, current_data) if not has_graded(ctx): return "kept", ctx change = self._sample_grading_change(ctx) @@ -2381,13 +2480,29 @@ async def _plan_sample( return "regraded", self._to_refeed(ctx) return "repostprocessed", self._to_regrade(ctx) - @staticmethod - def _stamp_input(ctx: TaskContext, data: str | None, prompt: str | None): - """Add the data and prompt digests to *ctx*'s latest ``preprocess`` meta.""" + def _sample_preprocess_changed(self, stamp: str | None) -> bool: + """Whether a recorded sample's ``preprocess`` differs from this run's. + + *stamp* is the :func:`preprocess_key` its record carries. Without one, + the directory's code identity decides if it has one (an upgrade + verified the sample), else the rendering pass. Unknown differs. + """ + current = preprocess_key(self._preprocess_identity()) + if stamp is not None: + return current is None or stamp != current + identity = recorded_identity(self._recorded) + recorded = identity["preprocess"] if identity is not None else None + if is_code_identity(recorded): + return current is None or preprocess_key(recorded) != current + return self._legacy_prompts_match is not True + + def _stamp_input(self, ctx: TaskContext, data: str | None): + """Add the data digest and this run's preprocess key to *ctx*'s latest + ``preprocess`` meta.""" key = TaskStage.PREPROCESSED.value history = list(ctx.stage_meta.get(key) or [{}]) last = dict(history[-1]) - last["identity"] = {"data": data, "prompt": prompt} + last["identity"] = self._input_stamp(data) history[-1] = cast(TaskStageMeta, last) return replace(ctx, stage_meta={**ctx.stage_meta, key: history}) @@ -2525,20 +2640,9 @@ def _grading_identities(self) -> dict[str, GradingIdentity] | None: versions=self._judge_versions, ) task_cls = type(self._task) - if self._stage_code is None: - self._stage_code = stage_code( - task_cls, [j["role"] for j in judges], task_cls.services - ) - for stage, code in self._stage_code.items(): - if "fallback" in code: - logger.info( - "{}: {} counts all grading code ({})", - self._task.name, - stage, - code["fallback"], - ) + code = self._stage_code_once([j["role"] for j in judges]) grading = grading_identities( - self._stage_code, + code, judges, service_identities(task_cls.services), self._task.grader_args(), @@ -2552,6 +2656,47 @@ def _grading_identities(self) -> dict[str, GradingIdentity] | None: self._grading = grading return grading + def _stage_code_once(self, judge_roles: list[str]) -> dict: + """:func:`stage_code` for this task, walked once per runner.""" + if self._stage_code is None: + task_cls = type(self._task) + self._stage_code = stage_code( + task_cls, + judge_roles, + task_cls.services, + list(self._task.grader_args()), + ) + for stage, code in self._stage_code.items(): + if "fallback" in code: + logger.info( + "{}: {} counts all of its task code ({})", + self._task.name, + stage, + code["fallback"], + ) + return self._stage_code + + def _preprocess_identity(self) -> PreprocessIdentity | None: + """This run's ``preprocess`` identity, computed once: no sample is + preprocessed for it. ``None`` when it cannot be computed.""" + if self._preprocess is False: + try: + roles = [j["role"] for j in self._current_judges()] + code = self._stage_code_once(roles)["preprocess"] + self._preprocess = task_preprocess_identity(self._task, code) + except IDENTITY_ERRORS as exc: + logger.warning("Preprocess identity unavailable: {}", exc) + self._preprocess = None + return self._preprocess + + def _input_stamp(self, data: str | None) -> dict[str, str | None]: + """What a ``preprocess`` record is stamped with: its sample's data + digest and this run's :func:`preprocess_key`.""" + return { + "data": data, + "preprocess": preprocess_key(self._preprocess_identity()), + } + def _seed_judge_versions(self, meta: dict | None) -> None: """Carry judge versions a resumed run's ``meta.json`` recorded. @@ -2620,10 +2765,10 @@ def _observe_judge_versions(self, model_calls: list) -> None: self._judge_versions.setdefault(role, reported) def _note_sample_digests(self, ctx: TaskContext) -> None: - """Record *ctx*'s data and prompt digests once, as of its first preprocess. + """Record *ctx*'s data digest once, as of its first preprocess. - Prefers the digests its ``preprocess`` record carries -- what that run - saw -- and falls back to digesting the sample and the recorded prompt. + Prefers the digest its ``preprocess`` record carries -- what that run + saw -- and falls back to digesting the sample. """ if ctx.sample_id in self._sample_digests: return @@ -2637,19 +2782,10 @@ def _note_sample_digests(self, ctx: TaskContext) -> None: data = recorded.get("data") if recorded else None if data is None: data = self._safe_digest(ctx.raw_sample) - prompt = recorded.get("prompt") if recorded else None - if prompt is None: - # No prompt at all is a fact (preprocess failed or never ran); a - # prompt that would not digest is unknown. - prompt = ( - ABSENT - if ctx.preprocess_result is None - else self._safe_digest(prompt_of(ctx.preprocess_result)) - ) - self._sample_digests[ctx.sample_id] = (ctx.repeat_index, data, prompt) + self._sample_digests[ctx.sample_id] = (ctx.repeat_index, data) async def _maybe_write_input_identity(self, force: bool = False) -> None: - """Write the data and prompt identities once every sample is past preprocess. + """Write the data identity once every sample is past preprocess. Not a barrier: samples already past preprocess keep going while the rest catch up. *force* writes whatever is known at the end of a run, so a @@ -2666,8 +2802,7 @@ async def _maybe_write_input_identity(self, force: bool = False) -> None: self._input_identity_written = True entries = self._sample_digests.items() await self._saver.set_identity( - data=aggregate_digest((sid, rep, d) for sid, (rep, d, _) in entries), - preprocess=aggregate_digest((sid, rep, p) for sid, (rep, _, p) in entries), + data=aggregate_digest((sid, rep, d) for sid, (rep, d) in entries) ) def _requests_identity(self) -> DigestedStage: @@ -2707,12 +2842,13 @@ async def _end_run(self, *, reported: bool, rules_hash: str | None = None) -> No *reported* says ``report.json`` was written; ``report()`` returning ``None`` writes none. - Writes the data and preprocess identities if a sample never reached - preprocess, folds what the candidate was sent into the inference - identity, and rewrites the grading ones with the judge versions - reported. + Writes the data identity if a sample never reached preprocess, and the + preprocess one if a resume left it unknown; folds what the candidate + was sent into the inference identity, and rewrites the grading ones + with the judge versions reported. """ await self._maybe_write_input_identity(force=True) + await self._saver.set_identity(preprocess=self._preprocess_identity()) inference = self._saver.inference_identity if inference is not None: await self._saver.set_identity( diff --git a/sieval/core/runners/stage_resume.py b/sieval/core/runners/stage_resume.py index fa8e0987..41d68897 100644 --- a/sieval/core/runners/stage_resume.py +++ b/sieval/core/runners/stage_resume.py @@ -4,7 +4,8 @@ changed, and reruns from there: ============================== ============================================== -Data or preprocess differ That sample reruns from the start. +Data differs That sample reruns from the start. +Preprocess differs Every sample reruns from the start. Inference differs Refused: :class:`StageResumeError` names it. Postprocess differs ``postprocess`` onward reruns on the recorded inference. @@ -20,11 +21,15 @@ A schema-1 ``grader`` (0.10.0 to 0.11.x) matches all grading stages or none (:func:`~sieval.core.tasks.run_meta.legacy_grader_keys`). -A sample's data and prompt digests come from its ``preprocess`` record. For a -directory recorded before them, the prompt is digested from the ``preprocess`` -output, and data and inference come from the caller's -:class:`LegacyStageMatch`. A stage neither can tell is unknown, and a resume -that would reuse it is refused. Nothing here writes to the recorded directory. +Each ``preprocess`` record is stamped with the task's ``preprocess`` key. A +directory that recorded rendered prompts (identity schemas 1 and 2) is +upgraded by rendering its recorded samples once. + +A sample's data digest comes from its ``preprocess`` record. For a directory +recorded before them, the prompt is digested from the ``preprocess`` output, +and data and inference come from the caller's :class:`LegacyStageMatch`. A +stage neither can tell is unknown, and a resume that would reuse it is +refused. Nothing here writes to the recorded directory. AI-Generated Code - Claude Opus 5.5 (Anthropic) """ @@ -40,17 +45,23 @@ from sieval.core.tasks.run_meta import ( GRADING_STAGES, STAGES, + PreprocessIdentity, ResumeOutcomes, RunIdentity, + StageCode, digest, grader_key, grader_task, grading_identities, inference_identity, + is_code_identity, judge_identities, legacy_grader_keys, + preprocess_identity, + preprocess_key, prompt_of, read_identity, + split_digests, stage_code, stage_key, ) @@ -121,7 +132,7 @@ def recorded_identity(meta: Mapping[str, Any] | None) -> RunIdentity | None: def recorded_grader(meta: Mapping[str, Any] | None) -> Mapping[str, Any] | None: - """A schema-1 block's combined ``grader`` identity; ``None`` for schema 2.""" + """A schema-1 block's combined ``grader`` identity; ``None`` for later ones.""" read = read_identity((meta or {}).get("identity")) return read[1] if read is not None else None @@ -156,11 +167,13 @@ def earliest_grading_change( class ResumePlan(TypedDict): """What a resume of one task directory would do, from identities alone. - ``rerun_from`` is the earliest grading stage a finished sample reruns - from, ``None`` if none changed; data and preprocess are decided per - sample, so only compared here. ``refused`` is ``inference`` when reusing - the recorded inference would be refused. ``legacy_grader`` is set for a - schema-1 directory: ``kept`` (upgraded in place) or ``regrade``. + ``rerun_from`` is the earliest stage a finished sample reruns from + (``preprocess`` restarts every sample), ``None`` if none changed; data is + decided per sample, so only compared here. ``refused`` is ``inference`` + when reusing the recorded inference would be refused. ``legacy_grader`` is + set for a schema-1 directory: ``kept`` or ``regrade``. + ``legacy_preprocess`` marks a schema-1 or schema-2 directory, whose + ``preprocess`` compares ``unknown`` until its samples are re-rendered. """ identity_schema: int | None @@ -168,6 +181,7 @@ class ResumePlan(TypedDict): rerun_from: str | None refused: str | None legacy_grader: Literal["kept", "regrade"] | None + legacy_preprocess: bool def plan_resume( @@ -177,9 +191,9 @@ def plan_resume( ) -> ResumePlan: """Compare a recorded ``meta.json`` with this run's identities, running nothing. - A pure entry point for a dry run. *current* is this run's identity as - schema 2; ``data`` and ``preprocess`` may be ``None`` before anything is - preprocessed and then compare as ``unknown``. *legacy_grader_keys* is + A pure entry point for a dry run. *current* is this run's identity; + ``data`` may be ``None`` before anything is preprocessed and then + compares as ``unknown``. *legacy_grader_keys* is :func:`~sieval.core.tasks.run_meta.legacy_grader_keys` for this task, read only for a schema-1 directory. The runner plans per sample and is authoritative; this says what a resume of a finished directory does. @@ -204,15 +218,49 @@ def plan_resume( stages[stage] = "same" if legacy_match else "unknown" if not legacy_match: stages[GRADING_STAGES[0]] = "differs" + recorded_preprocess = recorded.get("preprocess") return { "identity_schema": schema if isinstance(schema, int) else None, "stages": stages, - "rerun_from": earliest_grading_change(recorded, current, legacy_match), + "rerun_from": "preprocess" + if stages["preprocess"] == "differs" + else earliest_grading_change(recorded, current, legacy_match), "refused": "inference" if stages["inference"] != "same" else None, "legacy_grader": legacy, + "legacy_preprocess": recorded_preprocess is not None + and not is_code_identity(recorded_preprocess), } +#: The task argument naming the split few-shot exemplars are drawn from. +FEWSHOT_SPLIT_ARG = "fewshot_split" + + +def task_preprocess_identity(task: "Task", code: StageCode) -> PreprocessIdentity: + """:func:`~sieval.core.tasks.run_meta.preprocess_identity` for *task*. + + *code* is ``stage_code(...)["preprocess"]``. When that code reads a + dataset split, the one its ``fewshot_split`` argument names is digested, + else every split but ``test`` (which ``data`` covers). Raises + :class:`ValueError` when one cannot be. + """ + splits: dict[str, str] | None = None + if code.get("splits"): + dataset_dict = task.dataset.dataset_dict + args = task.grader_args() + named = args.get(FEWSHOT_SPLIT_ARG) + if FEWSHOT_SPLIT_ARG in code.get("args", ()) and named in dataset_dict: + names = [named] + else: + names = [name for name in dataset_dict if name != "test"] + splits = split_digests( + {str(name): cast(Iterable[object], dataset_dict[name]) for name in names} + ) + if splits is None: + raise ValueError("a dataset split the prompt reads cannot be digested") + return preprocess_identity(code, task.grader_args(), splits) + + def task_resume_plan( task: "Task", recorded_meta: Mapping[str, Any] | None, @@ -233,14 +281,12 @@ def task_resume_plan( judges = judge_identities(task.judge_models(), task.judge_prompts()) services = service_identities(task_cls.services) args = task.grader_args() - grading = grading_identities( - stage_code(task_cls, [j["role"] for j in judges], task_cls.services), - judges, - services, - args, - grader_task(task_cls), + code = stage_code( + task_cls, [j["role"] for j in judges], task_cls.services, list(args) ) + grading = grading_identities(code, judges, services, args, grader_task(task_cls)) current: dict[str, Any] = { + "preprocess": task_preprocess_identity(task, code["preprocess"]), "inference": inference_identity( model=task.model, task_args=task.inference_args(), @@ -263,6 +309,9 @@ def _compare(stage: str, recorded: object, current: object) -> StageComparison: mine = cast(Mapping[str, Any], current) if isinstance(current, Mapping) else None if stage in GRADING_STAGES: a, b = stage_key(theirs), stage_key(mine) + elif stage == "preprocess": + # A recorded digest of rendered prompts compares as unknown. + a, b = preprocess_key(theirs), preprocess_key(mine) else: # Inference compares as a resume refuses it: by its configuration. a = theirs.get("digest") if theirs is not None else None @@ -272,28 +321,50 @@ def _compare(stage: str, recorded: object, current: object) -> StageComparison: return "same" if a == b else "differs" -def recorded_sample_digests( - ctx: TaskContext, -) -> tuple[str | None, str | None, Literal["record", "derived"]]: - """A recorded sample's ``(data, prompt)`` digests and where they came from. +class SampleDigests(TypedDict): + """A recorded sample's input stamps (:func:`recorded_sample_digests`). + + ``preprocess`` is the :func:`~sieval.core.tasks.run_meta.preprocess_key` + its record carries; ``prompt`` the digest of its rendered prompt, which + only a record written under identity schema 1 or 2 carries. + """ + + data: str | None + preprocess: str | None + prompt: str | None + source: Literal["record", "derived"] + + +def recorded_sample_digests(ctx: TaskContext) -> SampleDigests: + """A recorded sample's data and preprocess stamps, and where they came from. - ``record`` when its ``preprocess`` record carries them. Otherwise - ``derived``: the prompt is digested from the recorded ``preprocess`` output - and the data is ``None``, since shards do not hold the raw sample. + ``record`` when its ``preprocess`` record carries a data digest and a + ``preprocess`` key, or, under schema 1 or 2, a ``prompt`` digest. + Otherwise ``derived``: the prompt is digested from the recorded + ``preprocess`` output and the data is ``None``, since shards do not hold + the raw sample. """ for entry in ctx.stage_meta.get(TaskStage.PREPROCESSED.value, ()): if isinstance(entry, dict) and isinstance(entry.get("identity"), dict): identity = entry["identity"] - data, prompt = identity.get("data"), identity.get("prompt") - if isinstance(data, str) and isinstance(prompt, str): - return data, prompt, "record" - prompt: str | None = None + data = identity.get("data") + preprocess, prompt = identity.get("preprocess"), identity.get("prompt") + if isinstance(data, str) and ( + isinstance(preprocess, str) or isinstance(prompt, str) + ): + return { + "data": data, + "preprocess": preprocess if isinstance(preprocess, str) else None, + "prompt": prompt if isinstance(prompt, str) else None, + "source": "record", + } + rendered: str | None = None if ctx.preprocess_result is not None: try: - prompt = digest(prompt_of(ctx.preprocess_result)) + rendered = digest(prompt_of(ctx.preprocess_result)) except (TypeError, ValueError): - prompt = None - return None, prompt, "derived" + rendered = None + return {"data": None, "preprocess": None, "prompt": rendered, "source": "derived"} def has_graded(ctx: TaskContext) -> bool: diff --git a/sieval/core/tasks/run_meta.py b/sieval/core/tasks/run_meta.py index c8612c3f..32423a11 100644 --- a/sieval/core/tasks/run_meta.py +++ b/sieval/core/tasks/run_meta.py @@ -10,8 +10,10 @@ =============== ============================================================ ``data`` Every selected sample, keyed by ``(sample_id, repeat_index)`` and digested over its whole raw content. -``preprocess`` Every sample's ``PromptRecord.prompt``, before the model's - chat template. +``preprocess`` The code ``preprocess`` reaches (:func:`stage_code`), the + task arguments it reads and, for few-shot, the split it draws + exemplars from (:func:`preprocess_identity`). Computed from + code once at setup. ``inference`` The candidate's resolved request arguments, dialect and engine, the task's own inference arguments (``n``), the result-changing runner keys and the serving recipe, plus @@ -27,11 +29,22 @@ ``report`` The same for ``report`` and ``problem_groups``. =============== ============================================================ -A change to one stage reruns it and every stage after it. +A change to one stage reruns it and every stage after it; a ``preprocess`` +change restarts every sample. + +``preprocess`` is code, not the prompts it renders: a prompt is code, +arguments, data and the model's ``extra``, and ``extra`` reaches the request, +which ``inference`` digests per sample. So a wrapping model and a plain one +share ``data`` and ``preprocess`` and differ in ``inference``. The walk is +static: a prompt that reads an environment variable, or a file through a +reader outside :data:`_FILE_READS` (``pd.read_csv``), is covered only through +the argument naming it. Identity schema 1 (0.10.0 to 0.11.x) named ``preprocess`` ``prompt`` and folded the grading stages into one ``grader``. :func:`legacy_grader_keys` recomputes it with the walk that wrote it: equal means nothing changed. +Schemas 1 and 2 hold ``preprocess`` as a digest of rendered prompts; a resume +renders the recorded samples once to compare (:func:`is_code_identity`). Every digest is ``sha256:`` over canonical JSON (sorted keys, no whitespace). A stage whose input could not be read is ``None``: unknown, never a guess. A @@ -45,10 +58,12 @@ import codecs import hashlib import importlib.util +import inspect import json import os import secrets import sys +import textwrap import warnings from collections.abc import Iterable, Mapping, Sequence from datetime import UTC, datetime @@ -72,7 +87,7 @@ #: Version of the ``identity`` block's layout. A digest computed differently is #: not a layout change: it reruns one stage, a new schema would discard them all. #: Schema 1 is still read (:func:`read_identity`). -IDENTITY_SCHEMA = 2 +IDENTITY_SCHEMA = 3 #: A per-sample digest of a value that was legitimately absent: the prompt of a #: sample whose ``preprocess`` failed, the requests of one never inferred. Kept @@ -203,17 +218,33 @@ class GradingIdentity(TypedDict): services: NotRequired[list[ServiceIdentity]] +class PreprocessIdentity(TypedDict): + """The ``preprocess`` stage: what renders a prompt, but the sample. + + ``modules`` is the code it reaches, ``args`` the ``grader_args()`` it + reads, ``splits`` a digest of each split it draws exemplars from, and + ``fallback`` why ``modules`` holds all the task's code. + """ + + digest: str + modules: dict[str, str] + args: dict[str, JSONValue] + splits: NotRequired[dict[str, str]] + fallback: NotRequired[str] + + class RunIdentity(TypedDict): """The ``identity`` block. A stage is ``None`` while unknown. Two runs share a stage only when every digest in it matches: for ``inference`` that is ``digest`` *and* ``requests.digest``, and a ``None`` - digest on either side matches nothing. + digest on either side matches nothing. A schema-1 or schema-2 + ``preprocess`` is a :class:`DigestedStage` of rendered prompts. """ schema: int data: DigestedStage | None - preprocess: DigestedStage | None + preprocess: PreprocessIdentity | DigestedStage | None inference: InferenceIdentity | None postprocess: GradingIdentity | None feedback: GradingIdentity | None @@ -1023,6 +1054,44 @@ def grader_task_v0_10_0(task_cls: type) -> str: #: Attribute lookups that name the attribute in an argument. _ATTR_CALLS = frozenset({"getattr", "setattr", "delattr", "hasattr"}) +#: Names and attributes that read a file whose path only the run knows. +_FILE_READS = frozenset( + {"open", "read_text", "read_bytes", "files", "iterdir", "glob", "rglob"} + | {"listdir", "scandir", "connect"} +) + + +def _loaded_unfollowed(task_cls: type) -> set[str] | None: + """The attributes a ``preprocess`` outside the followed packages (a task + in a test or a script) loads, by name from its classes' source; ``None`` + when a class's source cannot be read.""" + methods: dict[str, list[ast.AST]] = {} + for klass in task_cls.__mro__: + top = klass.__module__.partition(".")[0] + if top in ("sieval", "builtins") or top in sys.stdlib_module_names: + continue + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", SyntaxWarning) + tree = ast.parse(textwrap.dedent(inspect.getsource(klass))) + except (OSError, TypeError, SyntaxError): + return None + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + methods.setdefault(node.name, []).append(node) + pending = ["preprocess", *SHARED_ENTRY_METHODS] + seen: set[str] = set() + loaded: set[str] = set() + while pending: + name = pending.pop() + if name in seen: + continue + seen.add(name) + refs = _Refs(methods.get(name, ())) + loaded |= refs.loaded + pending.extend(a for a in refs.attrs if a in methods) + return loaded + class _Unattributable(Exception): """A construct the static per-method walk cannot attribute to a stage.""" @@ -1181,6 +1250,8 @@ def __init__(self, nodes: Iterable[ast.AST]): #: ``(attribute, attributes read, strings)`` for each ``. = #: ``: what the stored value was built from. self.stores: list[tuple[str, frozenset[str], frozenset[str]]] = [] + #: Constructs that read a file at run time (:data:`_FILE_READS`). + self.files: list[str] = [] #: String constants outside such a stored value: an endpoint variable #: read where it is used. self.strings: set[str] = set() @@ -1216,12 +1287,16 @@ def __init__(self, nodes: Iterable[ast.AST]): self.names.add(node.id) if node.id in _DYNAMIC_NAMES: self.dynamic.append(node.id) + if node.id in _FILE_READS: + self.files.append(node.id) elif isinstance(node, ast.Attribute): self.attrs.add(node.attr) if isinstance(node.ctx, ast.Load): self.loaded.add(node.attr) if node.attr in _DYNAMIC_ATTRS: self.dynamic.append(node.attr) + if node.attr in _FILE_READS: + self.files.append(node.attr) elif isinstance(node, ast.Call): self._attr_call(node) elif isinstance(node, (ast.Import, ast.ImportFrom)): @@ -1431,6 +1506,13 @@ def __init__(self, walker: _StageWalker): self.loaded: set[str] = set() self.stores: list[tuple[str, frozenset[str], frozenset[str]]] = [] self.strings: set[str] = set() + #: ``(module, where)`` of each function or method reached that reads a + #: file at run time. + self.file_units: set[tuple[str, str]] = set() + #: Modules that read a file at import, or whole modules that read one. + self.file_modules: set[str] = set() + #: Attributes a whole module reads anywhere: any of its code may run. + self.whole_loaded: set[str] = set() self._pending: list[tuple[str, str]] = [] self._pending_shells: list[str] = [] self._pending_whole: list[str] = [] @@ -1494,7 +1576,14 @@ def reach_name(self, parsed: _Module, name: str) -> None: (parsed.name, q) for q in parsed.units if q.startswith(prefix) ) - def read(self, parsed: _Module, refs: _Refs, where: str, local: _Bindings) -> None: + def read( + self, + parsed: _Module, + refs: _Refs, + where: str, + local: _Bindings, + unit: bool = False, + ) -> None: """Follow what *refs* read in *parsed*. *local* holds the reading unit's own imports: they run whenever it does, so all are reached, and a name they bind resolves to them as well as to the module's.""" @@ -1514,6 +1603,11 @@ def read(self, parsed: _Module, refs: _Refs, where: str, local: _Bindings) -> No ) for attr in refs.attrs: self.reach_attr(attr) + if refs.files: + if unit: + self.file_units.add((parsed.name, where)) + else: + self.file_modules.add(parsed.name) self.loaded |= refs.loaded self.stores.extend(refs.stores) self.strings |= refs.strings @@ -1551,6 +1645,7 @@ def run(self) -> None: _Refs([parsed.units[qualname]]), f"{module}.{qualname}", parsed.unit_bindings(qualname), + unit=True, ) else: self._whole_step(self._pending_whole.pop()) @@ -1565,7 +1660,11 @@ def _whole_step(self, name: str) -> None: with warnings.catch_warnings(): warnings.simplefilter("ignore", SyntaxWarning) tree = ast.parse(tree_source) - loads = sorted({d for d in _Refs([tree]).dynamic if d in _LOADERS}) + refs = _Refs([tree]) + if refs.files: + self.file_modules.add(name) + self.whole_loaded |= refs.loaded + loads = sorted({d for d in refs.dynamic if d in _LOADERS}) if loads and name.partition(".")[0] != "sieval": # It may load any file of its package, which no import shows and # another stage may reach. Under `sieval` the walk follows imports @@ -1641,12 +1740,16 @@ class StageCode(TypedDict): judges: list[str] services: bool fallback: NotRequired[str] + args: NotRequired[list[str]] + splits: NotRequired[bool] + files: NotRequired[list[str]] def stage_code( task_cls: type, judge_roles: Iterable[str] = (), services: Iterable[str] = (), + args: Iterable[str] = (), ) -> dict[str, StageCode]: """The code each stage of *task_cls* reaches, by stage. @@ -1664,8 +1767,20 @@ def stage_code( computed name, ``eval``, ``importlib``, ``vars()``, a method with no source) makes that grading stage and every later one count all of :func:`grader_modules`, every judge and service; ``fallback`` names it. - ``preprocess`` falls back on its own: its prompts are digested per sample. + + ``preprocess`` falls back on its own, and also when a function or method + it reaches reads a file (:data:`_FILE_READS`): it then counts all of + :func:`grader_modules` and the data files of the reading module's package + (:func:`_package_data`). A module that reads a file at import adds its + package's data files without falling back, named in ``files``. + + ``args`` names the *args* ``preprocess`` reads as ``self.``, + ``self._`` or an attribute stored from one; all of them under a + fallback, or when the task overrides ``grader_args``, which may return an + argument kept under another name. ``splits`` is whether it reads a split + beyond the sample (:data:`_SPLIT_READS`). """ + arg_names = sorted(set(args)) roles = sorted(set(judge_roles)) markers = {SERVICE_ENDPOINT_ENV[s] for s in services if s in SERVICE_ENDPOINT_ENV} unknown_services = any(s not in SERVICE_ENDPOINT_ENV for s in services) @@ -1691,12 +1806,15 @@ def stage_code( for top in sorted(tops): modules = unreached.get(top, {}) | modules judges = [r for r in roles if own_judges or walk.reads((r, f"_{r}"))] - out[stage] = { + entry: StageCode = { "modules": dict(sorted(modules.items())), "judges": judges, "services": has_services and (unknown_services or walk.reads((), markers)), } + if stage == "preprocess": + _preprocess_entry(entry, walk, walker, arg_names) + out[stage] = entry continue if stage != "preprocess": grading_reason = why @@ -1708,6 +1826,158 @@ def stage_code( "services": has_services, "fallback": cast(str, why), } + if stage == "preprocess": + out[stage]["args"] = list(arg_names) + out[stage]["splits"] = True + return out + + +def _preprocess_entry( + entry: StageCode, walk: _StageWalk, walker: _StageWalker, args: list[str] +) -> None: + """Add what :func:`stage_code` records for ``preprocess`` alone to *entry*.""" + if _overrides_grader_args(walker.task_cls): + entry["args"] = list(args) + entry["splits"] = bool((walk.loaded | walk.whole_loaded) & _SPLIT_READS) + elif "preprocess" in walker.methods: + entry["args"] = [a for a in args if walk.reads((a, f"_{a}"))] + entry["splits"] = bool((walk.loaded | walk.whole_loaded) & _SPLIT_READS) + else: + loaded = _loaded_unfollowed(walker.task_cls) + if loaded is None: + entry["args"], entry["splits"] = list(args), True + else: + entry["args"] = [a for a in args if {a, f"_{a}"} & loaded] + entry["splits"] = bool(loaded & _SPLIT_READS) + own = walker.own_package + if walk.file_modules: + readers = sorted(walk.file_modules) + entry["modules"] = dict( + sorted((entry["modules"] | _package_data(readers, own)).items()) + ) + entry["files"] = readers + if walk.file_units: + units = sorted(walk.file_units) + readers = sorted({module for module, _ in units}) + whole = grader_modules(walker.task_cls) | _package_data(readers, own) + entry["modules"] = dict(sorted((entry["modules"] | whole).items())) + entry["args"] = list(args) + entry["fallback"] = f"{', '.join(w for _, w in units)} read a file at run time" + + +def _overrides_grader_args(task_cls: type) -> bool: + """Whether a class of *task_cls* defines its own ``grader_args``: the + base reads each argument back under its own name, an override need not.""" + return any( + "grader_args" in vars(klass) and klass.__module__ != "sieval.core.tasks.task" + for klass in task_cls.__mro__ + ) + + +#: Attributes through which a task reads a dataset split other than the +#: evaluated one, such as few-shot exemplars. +_SPLIT_READS = frozenset({"dataset_dict", "train_set", "retrieve_samples"}) + +#: Files beside a module's source that are never data it reads: code and the +#: agent instruction files. Other Markdown may be a prompt template. +_NOT_DATA_SUFFIXES = (".py", ".pyc", ".pyi") +_NOT_DATA_NAMES = frozenset({"AGENTS.md", "CLAUDE.md", "README.md"}) + + +def _package_data(modules: Iterable[str], own: str | None) -> dict[str, str]: + """``{file key: digest}`` of every data file under each of *modules*' + package directories: a module that reads a file the walk cannot name most + likely reads one shipped beside it (a template, few-shot examples).""" + out: dict[str, str] = {} + for module in modules: + located = _locate(module) + if located is None or located[0] is None: + continue + root = os.path.dirname(located[0]) + for dirpath, dirnames, filenames in os.walk(root): + dirnames[:] = sorted( + d for d in dirnames if not d.startswith((".", "__pycache__")) + ) + for filename in sorted(filenames): + if ( + filename.startswith(".") + or filename.endswith(_NOT_DATA_SUFFIXES) + or filename in _NOT_DATA_NAMES + ): + continue + path = os.path.join(dirpath, filename) + out[_file_key(module, path, own)] = ( + "sha256:" + hashlib.sha256(_read_source(path)).hexdigest() + ) + return out + + +def is_code_identity(stage: object) -> bool: + """Whether a recorded ``preprocess`` identity is the code-based one. + + Schemas 1 (as ``prompt``) and 2 recorded a digest over every sample's + rendered prompt (:class:`DigestedStage`), which holds no ``modules``. + """ + if not isinstance(stage, Mapping): + return False + return isinstance(cast(Mapping[str, Any], stage).get("modules"), Mapping) + + +def preprocess_key(stage: object) -> str | None: + """What a resume compares of a ``preprocess`` identity, and stamps on each + ``preprocess`` record: every field but ``digest`` and ``fallback``. + + ``fallback`` only explains ``modules``, and spells import names. ``None`` + for an identity that is not code-based or is malformed. + """ + if not is_code_identity(stage): + return None + payload = { + k: v + for k, v in cast(Mapping[str, Any], stage).items() + if k not in ("digest", "fallback") + } + try: + return digest(payload) + except (TypeError, ValueError): + return None + + +def preprocess_identity( + code: StageCode, + args: Mapping[str, JSONValue], + splits: Mapping[str, str] | None = None, +) -> PreprocessIdentity: + """Build the ``preprocess`` identity from ``stage_code()["preprocess"]``. + + *args* is :meth:`~sieval.core.tasks.task.Task.grader_args`; only those the + code reads count. *splits* digests each dataset split + (:func:`split_digests`), used when the code reads one. + """ + read = set(code.get("args", args)) + identity: PreprocessIdentity = { + "digest": "", + "modules": dict(code["modules"]), + "args": {k: v for k, v in args.items() if k in read}, + } + if code.get("splits"): + identity["splits"] = dict(splits or {}) + if "fallback" in code: + identity["fallback"] = code["fallback"] + identity["digest"] = digest({k: v for k, v in identity.items() if k != "digest"}) + return identity + + +def split_digests(splits: Mapping[str, Iterable[object]]) -> dict[str, str] | None: + """``{split: digest}`` over each split's rows in order, as ``data`` digests + them; ``None`` when a row cannot be digested.""" + out: dict[str, str] = {} + for name in sorted(splits): + try: + rows = [digest(row) for row in splits[name]] + except (TypeError, ValueError): + return None + out[name] = digest(rows) return out @@ -2005,20 +2275,22 @@ def empty_identity() -> RunIdentity: def read_identity( block: object, ) -> tuple[RunIdentity, Mapping[str, Any] | None] | None: - """A recorded ``identity`` block as schema 2, with a schema-1 ``grader``. - - Schema 2 is read as written. Schema 1 maps ``data`` and ``inference`` - across and ``prompt`` to ``preprocess`` (:data:`SCHEMA_1_STAGES`); its - ``grader`` cannot be split, so the grading stages read as unknown and the - ``grader`` is returned beside them, for the caller to compare whole - (:func:`legacy_grader_keys`). ``None`` for any other schema. + """A recorded ``identity`` block as this schema, with a schema-1 ``grader``. + + Schemas 2 and 3 are read as written; schema 2's ``preprocess`` is a digest + of rendered prompts, which :func:`is_code_identity` tells apart. Schema 1 + maps ``data`` and ``inference`` across and ``prompt`` to ``preprocess`` + (:data:`SCHEMA_1_STAGES`); its ``grader`` cannot be split, so the grading + stages read as unknown and the ``grader`` is returned beside them, for the + caller to compare whole (:func:`legacy_grader_keys`). ``None`` for any + other schema. """ if not isinstance(block, Mapping): return None block = cast(Mapping[str, Any], block) out = empty_identity() schema = block.get("schema") - if schema == IDENTITY_SCHEMA: + if schema in (2, IDENTITY_SCHEMA): for stage in STAGES: if isinstance(block.get(stage), dict): out[stage] = block[stage] # type: ignore[literal-required] @@ -2033,7 +2305,7 @@ def read_identity( def carried_identity(existing: Mapping[str, Any] | None) -> RunIdentity: - """The previous run's identity as schema 2 (:func:`read_identity`), else empty. + """The previous run's identity as this schema (:func:`read_identity`), else empty. A schema-1 block's grading stages are unknown until the run writes them. """ diff --git a/tests/integration/resume/test_across_releases.py b/tests/integration/resume/test_across_releases.py index 8203eb7d..717aff90 100644 --- a/tests/integration/resume/test_across_releases.py +++ b/tests/integration/resume/test_across_releases.py @@ -4,7 +4,8 @@ 0.10.2 and 0.11.0 wrote for five HumanEval problems against a fake model. ``human_eval-5.jsonl`` holds those five problems, so the data identity they recorded matches. All three recorded identity schema 1; a resume reads it and -writes schema 2. +writes schema 3, rendering the five recorded prompts once to check that +``preprocess`` is unchanged. The grading tests point ``SIEVAL_CODE_EVAL_API`` at a fake code evaluator and declare it the vendored service each fixture recorded, so the service identity @@ -38,6 +39,7 @@ GRADING_STAGES, IDENTITY_SCHEMA, RESULT_SCHEMA_VERSION, + is_code_identity, ) from sieval.core.tasks.services import VENDORED_SERVICES_ENV from sieval.tasks import HumanEvalZeroShotGenTask @@ -224,12 +226,15 @@ def test_an_unchanged_task_keeps_every_verdict_and_upgrades_the_identity( assert resumed["reran"] == [] assert resumed["resume"]["kept"] == 5 assert resumed["resume"]["repostprocessed"] == 0 + assert resumed["resume"]["restarted"] == 0 identity = meta["identity"] - assert identity["schema"] == IDENTITY_SCHEMA == 2 + assert identity["schema"] == IDENTITY_SCHEMA == 3 assert "grader" not in identity and "prompt" not in identity old = recorded["identity"] assert identity["data"] == old["data"] - assert identity["preprocess"] == old["prompt"] + # The rendered prompts became the code identity a fresh run writes. + assert is_code_identity(identity["preprocess"]) + assert identity["preprocess"] != old["prompt"] assert identity["inference"] == old["inference"] assert all(identity[stage] is not None for stage in GRADING_STAGES) assert identity["feedback"]["services"] == old["grader"]["services"] @@ -259,6 +264,40 @@ def test_the_upgraded_block_is_what_a_fresh_run_writes( assert fresh is not None for stage in GRADING_STAGES: assert upgraded[stage] == fresh[stage], stage + assert upgraded["preprocess"] == runner._preprocess_identity() + + def test_a_changed_prompt_restarts_every_sample( + self, release, tmp_path, evaluator, monkeypatch + ): + """The recorded prompts no longer render: preprocess changed, so the + upgrade restarts every sample instead of keeping its inference.""" + evaluator(_recorded_service(release)) + root = _copy(release, tmp_path) + original = HumanEvalZeroShotGenTask.preprocess + + async def preprocess(self, raw, ctx): + record = await original(self, raw, ctx) + record["prompt"][0]["content"] += "\n" + return record + + monkeypatch.setattr(HumanEvalZeroShotGenTask, "preprocess", preprocess) + result = CliRunner().invoke( + app, + [ + "eval", + str(_config(tmp_path)), + "--result-dir", + str(root), + "--resume", + "-o", + "json", + ], + ) + # Nothing serves the model, so the restarted samples fail at infer. + assert result.exception is None, result.output + resumed = _meta(root)["runs"][-1] + assert resumed["resume"]["restarted"] == 5 + assert resumed["resume"]["kept"] == 0 def test_a_changed_grader_regrades(self, release, tmp_path, evaluator): served = evaluator(_recorded_service(release)) @@ -272,8 +311,11 @@ def test_a_changed_grader_regrades(self, release, tmp_path, evaluator): assert resumed["reran"] == ["postprocess", "feedback", "report"] assert resumed["resume"]["repostprocessed"] == 5 identity = _meta(root)["identity"] - assert identity["schema"] == 2 + assert identity["schema"] == 3 assert identity["feedback"]["args"]["timeout"] == 4.0 + # `timeout` is a grading argument, never read by preprocess. + assert "timeout" not in identity["preprocess"]["args"] + assert resumed["resume"]["restarted"] == 0 assert report["score"] == 20.0 def test_a_changed_report_rereports_only( diff --git a/tests/integration/test_lazy_launch.py b/tests/integration/test_lazy_launch.py index c26aa9ab..8f3b0bfa 100644 --- a/tests/integration/test_lazy_launch.py +++ b/tests/integration/test_lazy_launch.py @@ -429,11 +429,14 @@ def test_a_resume_reports_its_plan_and_reads_dirs_read_only( "needs_inference": True, } stages = data["tasks"]["echo"]["stages"] - assert stages["identity_schema"] == 2 + assert stages["identity_schema"] == 3 assert stages["rerun_from"] is None # A third sample changes the set's data identity; each recorded sample # still matches its own digest, so both are kept. assert stages["stages"]["data"] == "differs" + # Preprocess is code: the same task compares the same. + assert stages["stages"]["preprocess"] == "same" + assert stages["legacy_preprocess"] is False assert stages["stages"]["inference"] == "same" assert engine.launches == [] assert _tree(tmp_path / "results") == recorded diff --git a/tests/unit/cli/test_validation.py b/tests/unit/cli/test_validation.py index 84829cee..7be5e1f4 100644 --- a/tests/unit/cli/test_validation.py +++ b/tests/unit/cli/test_validation.py @@ -1027,6 +1027,53 @@ def test_task_infer_args_cannot_contain_binding_resource(self): assert not result.ok assert any("binding resources" in e and "api_key" in e for e in result.errors) + @pytest.mark.parametrize( + ("model", "infer_args", "where"), + [ + ( + {"name": "m", "extra": {"sequence_wrappers": {"dna": ""}}}, + None, + "Model 'm' extra.sequence_wrappers.dna", + ), + ( + {"name": "m", "args": {"extra": {"sequence_wrappers": {"x": "{seq}"}}}}, + None, + "Model 'm' args.extra.sequence_wrappers.x", + ), + ( + {"name": "m"}, + {"extra": {"sequence_wrappers": {"rna": 1}}}, + "Task 't' infer_args.extra.sequence_wrappers.rna", + ), + ], + ) + def test_malformed_sequence_wrappers_are_rejected(self, model, infer_args, where): + task: dict = {"class": "X", "dataset": "d", "model": "m"} + if infer_args is not None: + task["infer_args"] = infer_args + cfg = { + "models": {"m": model}, + "datasets": {"d": {"class": "X"}}, + "tasks": {"t": task}, + } + result = validate_eval_config(cfg) + assert any(e.startswith(where) for e in result.errors), result.errors + + def test_well_formed_sequence_wrappers_pass(self): + cfg = { + "models": { + "m": { + "name": "m", + "extra": {"sequence_wrappers": {"dna": "{seq}"}}, + } + }, + "datasets": {"d": {"class": "X"}}, + "tasks": {"t": {"class": "X", "dataset": "d", "model": "m"}}, + } + assert not any( + "sequence_wrappers" in e for e in validate_eval_config(cfg).errors + ) + def test_model_role_unsupported_by_task_constructor_is_rejected(self): # `sieval eval` must reject the same role config `sieval run --dry-run` # rejects; the two entry points disagreeing is what let the missing diff --git a/tests/unit/core/runners/test_runner_run_meta.py b/tests/unit/core/runners/test_runner_run_meta.py index b1ab5b20..10d0b433 100644 --- a/tests/unit/core/runners/test_runner_run_meta.py +++ b/tests/unit/core/runners/test_runner_run_meta.py @@ -29,7 +29,13 @@ ) from sieval.core.tasks.consts import TaskStage from sieval.core.tasks.loader import TaskLoader -from sieval.core.tasks.run_meta import RESULT_SCHEMA_VERSION, aggregate_digest, digest +from sieval.core.tasks.run_meta import ( + RESULT_SCHEMA_VERSION, + aggregate_digest, + digest, + preprocess_identity, + preprocess_key, +) from sieval.core.tasks.saver import TaskSaver from sieval.core.tasks.task import Task from sieval.core.utils import atomic as saver_module @@ -179,9 +185,12 @@ async def test_writes_every_field(self, tmp_path): assert meta["deterministic"] is False assert meta["model"] == "mock-chat" identity = meta["identity"] - assert identity["schema"] == 2 + assert identity["schema"] == 3 assert identity["data"]["samples"] == 3 - assert identity["preprocess"]["samples"] == 3 + # A test task lives outside the followed packages: no code, and `n` + # is an inference argument, not a preprocess one. + assert identity["preprocess"]["modules"] == {} + assert identity["preprocess"]["args"] == {} assert identity["inference"]["digest"].startswith("sha256:") # A test task lives outside the followed packages: no modules, no judges. for stage in ("postprocess", "feedback", "report"): @@ -204,7 +213,7 @@ async def test_writes_every_field(self, tmp_path): assert "from" not in run @pytest.mark.anyio - async def test_data_and_prompt_digests_are_what_the_samples_hold(self, tmp_path): + async def test_data_is_the_samples_and_preprocess_the_code(self, tmp_path): runner = _runner(tmp_path) await runner.arun() identity = _meta(runner.root_dir)["identity"] @@ -212,12 +221,10 @@ async def test_data_and_prompt_digests_are_what_the_samples_hold(self, tmp_path) expected_data = aggregate_digest( (i, None, digest(sample)) for i, sample in enumerate(SAMPLES) ) - expected_prompt = aggregate_digest( - (i, None, digest([{"role": "user", "content": s["question"]}])) - for i, s in enumerate(SAMPLES) - ) assert identity["data"] == expected_data - assert identity["preprocess"] == expected_prompt + assert identity["preprocess"] == preprocess_identity( + {"modules": {}, "judges": [], "services": False, "args": []}, {} + ) @pytest.mark.anyio async def test_per_sample_digests_land_in_the_preprocess_record(self, tmp_path): @@ -235,7 +242,9 @@ async def test_per_sample_digests_land_in_the_preprocess_record(self, tmp_path): [entry] = ctx.stage_meta[TaskStage.PREPROCESSED.value] assert entry["identity"] == { "data": digest(sample), - "prompt": digest([{"role": "user", "content": sample["question"]}]), + "preprocess": preprocess_key( + _meta(runner.root_dir)["identity"]["preprocess"] + ), } @pytest.mark.anyio @@ -279,7 +288,7 @@ async def preprocess(self, raw, ctx): await runner.arun() identity = _meta(runner.root_dir)["identity"] assert identity["data"]["samples"] == 3 - assert identity["preprocess"]["samples"] == 3 + assert identity["data"]["digest"] is not None class TestInferenceIdentityFromRunner: @@ -519,7 +528,9 @@ async def infer(self, pre, ctx): runner = _runner(tmp_path, task=task, record_meta=False) await runner.arun() identity = _meta(runner.root_dir)["identity"] - assert identity["preprocess"] == aggregate_digest([(0, None, None)]) + # Preprocess is code, so a prompt that will not digest leaves it known. + assert identity["preprocess"]["digest"].startswith("sha256:") + assert identity["data"]["digest"] is not None @pytest.mark.anyio async def test_a_broken_judge_lookup_leaves_the_grader_unknown(self, tmp_path): @@ -926,12 +937,12 @@ async def test_resume_from_a_pre_identity_meta(self, tmp_path): meta = _meta(root) assert "task" not in meta assert [r["start"] for r in meta["runs"]] == ["resume"] - # Every sample was already final: only `report()` ran, and data and - # prompt are digested from what the shards hold. + # Every sample was already final: only `report()` ran; data is + # digested from the samples, preprocess from the code. assert meta["runs"][0]["reran"] == ["report"] assert meta["runs"][0]["reported"] is True assert meta["identity"]["data"]["samples"] == 3 - assert meta["identity"]["preprocess"]["samples"] == 3 + assert meta["identity"]["preprocess"]["digest"].startswith("sha256:") # =================================================================== diff --git a/tests/unit/core/runners/test_sequence_wrappers.py b/tests/unit/core/runners/test_sequence_wrappers.py new file mode 100644 index 00000000..c573a0ac --- /dev/null +++ b/tests/unit/core/runners/test_sequence_wrappers.py @@ -0,0 +1,351 @@ +""" +A task reading ``model.extra["sequence_wrappers"]`` against two models. + +``WrapTask`` is the pattern a sequence task follows: in ``preprocess`` it asks +the model for a wrapper by sequence type and, when there is none, renders +exactly what it rendered before. A wrapping model and a plain one then share +``data`` and ``preprocess`` and differ only in ``inference``, whose +per-sample ``requests`` digest the wrapped prompt as sent. + +AI-Generated Code - Claude Opus 5.5 (Anthropic) +""" + +from dataclasses import replace +from pathlib import Path + +import orjson +import pytest + +from sieval.core.models import check_extra, sequence_wrapper +from sieval.core.runners import task_resume_plan +from sieval.core.runners.runner import TaskRunner +from sieval.core.runners.stage_resume import StageResumeError, task_preprocess_identity +from sieval.core.tasks import build_prompt_record +from sieval.core.tasks.run_meta import stage_code +from sieval.core.tasks.task import Task +from tests.conftest import MockChatModel, MockDataset, make_config, prompt_of + +SAMPLES = [ + {"sequence": "ACGT", "partner": "TTGA", "answer": "yes"}, + {"sequence": "GGCC", "partner": "AATT", "answer": "no"}, +] + +WRAPPERS = {"dna": "{seq}"} + + +class WrapTask(Task): + """A dual-sequence DNA task: each sequence is wrapped on its own.""" + + model_type = "chat" + + async def preprocess(self, raw, ctx): + wrap = sequence_wrapper(self.model.extra, "dna") + first, second = raw["sequence"], raw["partner"] + if wrap: + first, second = wrap.format(seq=first), wrap.format(seq=second) + return build_prompt_record( + [{"role": "user", "content": f"Do {first} and {second} bind?"}], + reference=raw["answer"], + ) + + async def infer(self, pre, ctx): + return await self.model.agenerate(pre["prompt"]) + + async def postprocess(self, inf, ctx): + return inf.texts[0].strip() + + async def feedback(self, post, ctx): + return True, {"correct": post == ctx.raw_sample["answer"]} + + async def report(self, finals, fails): + return {"total": len(finals) + len(fails)} + + +class PlainTask(WrapTask): + """The same prompt with no wrapping code: what the task rendered before.""" + + async def preprocess(self, raw, ctx): + return build_prompt_record( + [ + { + "role": "user", + "content": f"Do {raw['sequence']} and {raw['partner']} bind?", + } + ], + reference=raw["answer"], + ) + + +class EchoModel(MockChatModel): + """Echoes the prompt into ``request_params``, as a dialect records the body.""" + + async def _stub_arun(self, req): + response = await super()._stub_arun(req) + return replace(response, request_params={"messages": prompt_of(req)}) + + +def _meta(root: Path) -> dict: + return orjson.loads((root / "meta.json").read_bytes()) + + +def _model(wrappers=None) -> EchoModel: + model = EchoModel(default_answer="yes") + return model.with_args(extra={"sequence_wrappers": wrappers}) if wrappers else model + + +async def _run(tmp_path, name, model, cls=WrapTask, **cfg): + task = cls(dataset=MockDataset(SAMPLES), model=model, name="seq") + runner = TaskRunner( + task, make_config(tmp_path, result_dir=str(tmp_path / name), **cfg) + ) + await runner.arun() + return runner.root_dir + + +def _sent(root: Path) -> list[str]: + """Every prompt the candidate was sent, from the ``infer`` records.""" + out = [] + for shard in sorted(root.glob("0/*/*.jsonl")): + for raw in shard.read_bytes().splitlines(): + record = orjson.loads(raw) + history = (record.get("stage_meta") or {}).get("inferred", []) + if record.get("stage") == "inferred" and record.get("meta_last"): + history = [*history, record["meta_last"]] + for entry in history: + out += [c["request_params"]["messages"] for c in entry["model_calls"]] + return sorted(set(out)) + + +class TestWrappingChangesOnlyTheRequest: + @pytest.mark.anyio + async def test_data_and_preprocess_are_shared_inference_differs(self, tmp_path): + plain = _meta(await _run(tmp_path, "plain", _model()))["identity"] + wrapped = _meta(await _run(tmp_path, "wrapped", _model(WRAPPERS)))["identity"] + + assert wrapped["data"] == plain["data"] + assert wrapped["preprocess"] == plain["preprocess"] + assert wrapped["inference"]["digest"] != plain["inference"]["digest"] + assert ( + wrapped["inference"]["requests"]["digest"] + != plain["inference"]["requests"]["digest"] + ) + + @pytest.mark.anyio + async def test_the_request_bodies_show_the_wrapping(self, tmp_path): + plain = _sent(await _run(tmp_path, "plain", _model())) + wrapped = _sent(await _run(tmp_path, "wrapped", _model(WRAPPERS))) + assert plain == ["Do ACGT and TTGA bind?", "Do GGCC and AATT bind?"] + assert wrapped == [ + "Do ACGT and TTGA bind?", + "Do GGCC and AATT bind?", + ] + + @pytest.mark.anyio + async def test_another_type_s_wrapper_leaves_the_prompt(self, tmp_path): + """A model wrapping only RNA sends a DNA task's prompt unchanged.""" + plain = _sent(await _run(tmp_path, "plain", _model())) + rna_only = _sent( + await _run(tmp_path, "rna", _model({"rna": "{seq}"})) + ) + assert rna_only == plain + + @pytest.mark.anyio + async def test_absent_key_renders_today_s_prompts_byte_for_byte(self, tmp_path): + """No `sequence_wrappers`: the reading task sends what a task without + the reading code sends.""" + before = _sent(await _run(tmp_path, "before", _model(), cls=PlainTask)) + after = _sent(await _run(tmp_path, "after", _model())) + assert [p.encode() for p in after] == [p.encode() for p in before] + + @pytest.mark.anyio + async def test_a_resume_with_the_wrapping_model_is_refused(self, tmp_path): + """Per-sample requests and the model's `extra` are inference: a resume + that would reuse a plain model's inference for a wrapping one is + refused, as for any inference change.""" + await _run(tmp_path, "run", _model()) + task = WrapTask( + dataset=MockDataset(SAMPLES), model=_model(WRAPPERS), name="seq" + ) + runner = TaskRunner( + task, make_config(tmp_path, result_dir=str(tmp_path / "run"), resume=True) + ) + with pytest.raises(StageResumeError, match="stage `inference`"): + await runner.arun() + + @pytest.mark.anyio + async def test_the_dry_run_plan_shares_preprocess(self, tmp_path): + """`task_resume_plan`, which a dry run reads, compares a wrapping model + with a plain run's directory: preprocess is the same, inference is not.""" + root = await _run(tmp_path, "plain", _model()) + recorded = _meta(root) + task = WrapTask( + dataset=MockDataset(SAMPLES), model=_model(WRAPPERS), name="seq" + ) + keys = {"deterministic": False, "max_iterations": 1, "max_retries": 3} + plan = task_resume_plan(task, recorded, runner_keys=keys) + assert plan["stages"]["preprocess"] == "same" + assert plan["stages"]["inference"] == "differs" + assert plan["refused"] == "inference" + assert plan["legacy_preprocess"] is False + + @pytest.mark.anyio + async def test_plan_run_stages_share_preprocess_across_models(self, tmp_path): + """The dry-run's per-task `stages` (`plan_run`) for the wrapping model + over a plain model's directory: preprocess is the same. Only the + preprocess records are kept, so no inference is reused and the plan is + not refused.""" + root = await _run(tmp_path, "plain", _model()) + (root / "report.json").unlink() + for stage_dir in root.glob("0/*"): + if stage_dir.name != "preprocessed": + for f in stage_dir.iterdir(): + f.unlink() + task = WrapTask( + dataset=MockDataset(SAMPLES), model=_model(WRAPPERS), name="seq" + ) + runner = TaskRunner( + task, make_config(tmp_path, result_dir=str(root), resume=True) + ) + plan = await runner.plan_run() + stages = plan["stages"] + assert stages is not None + assert stages["stages"]["preprocess"] == "same" + assert stages["stages"]["data"] == "same" + assert stages["stages"]["inference"] == "differs" + assert stages["rerun_from"] != "preprocess" + assert plan["outcomes"] is not None + assert plan["outcomes"]["restarted"] == 0 + assert plan["outcomes"]["kept"] == 2 + + @pytest.mark.anyio + async def test_plan_run_refuses_the_wrapping_model_as_the_run_does(self, tmp_path): + await _run(tmp_path, "run", _model()) + task = WrapTask( + dataset=MockDataset(SAMPLES), model=_model(WRAPPERS), name="seq" + ) + runner = TaskRunner( + task, make_config(tmp_path, result_dir=str(tmp_path / "run"), resume=True) + ) + with pytest.raises(StageResumeError, match="stage `inference`"): + await runner.plan_run() + + +class TestValidation: + def test_a_template_without_seq_is_refused(self): + with pytest.raises(ValueError, match=r"has no \{seq\}"): + _model({"dna": ""}) + + @pytest.mark.parametrize( + ("wrappers", "match"), + [ + ({"dna": 3}, "must be a format string"), + ({"genome": "{seq}"}, "unknown sequence type 'genome'"), + ({"dna": "<{kind}>{seq}"}, r"may only use \{seq\}"), + ({"dna": "{seq"}, "is not a format string"), + (["dna"], "must map a sequence type"), + ], + ) + def test_malformed_wrappers_are_refused(self, wrappers, match): + with pytest.raises(ValueError, match=match): + check_extra({"sequence_wrappers": wrappers}) + + def test_literal_braces_and_every_type_are_accepted(self): + check_extra( + { + "sequence_wrappers": { + "dna": "{{dna}}{seq}", + "rna": "{seq}", + "protein": "{seq}", + "mol": "{seq}", + }, + "anything_else": [1, 2], + } + ) + + def test_extra_without_wrappers_is_not_checked(self): + check_extra(None) + check_extra({"note": "free-form"}) + + @pytest.mark.parametrize( + ("extra", "expected"), + [ + ({"sequence_wrappers": WRAPPERS}, "{seq}"), + ({"sequence_wrappers": {"rna": "{seq}"}}, None), + ({"sequence_wrappers": "{seq}"}, None), + ({}, None), + ], + ) + def test_sequence_wrapper_reads_one_type(self, extra, expected): + assert sequence_wrapper(extra, "dna") == expected + + +class ShotTask(WrapTask): + """Prefixes each prompt with the ``train`` split's first row: a few-shot + task, whose ``preprocess`` reads beyond the sample it renders.""" + + async def preprocess(self, raw, ctx): + shot = self.dataset.dataset_dict["train"][0]["sequence"] + record = await super().preprocess(raw, ctx) + record["prompt"][0]["content"] = f"{shot}\n" + record["prompt"][0]["content"] + return record + + +class NamedShotTask(WrapTask): + """Draws its exemplar from the split ``fewshot_split`` names.""" + + def __init__(self, *args, fewshot_split: str = "dev", **kwargs): + super().__init__(*args, **kwargs) + self._fewshot_split = fewshot_split + + async def preprocess(self, raw, ctx): + split = self.dataset.dataset_dict[self._fewshot_split] + record = await super().preprocess(raw, ctx) + record["prompt"][0]["content"] = ( + f"{split[0]['sequence']}\n" + (record["prompt"][0]["content"]) + ) + return record + + +class TestFewShotSplits: + @pytest.mark.anyio + async def test_a_changed_exemplar_split_changes_preprocess(self, tmp_path): + from datasets import Dataset as HFDataset + + def task(train_seq): + dataset = MockDataset(SAMPLES) + dataset.dataset_dict["train"] = HFDataset.from_list( + [{"sequence": train_seq, "partner": "", "answer": ""}] + ) + return ShotTask(dataset=dataset, model=_model(), name="seq") + + keys = {"deterministic": False, "max_iterations": 1, "max_retries": 3} + first = TaskRunner( + task("AAAA"), make_config(tmp_path, result_dir=str(tmp_path / "r")) + ) + await first.arun() + recorded = _meta(first.root_dir) + assert "train" in recorded["identity"]["preprocess"]["splits"] + assert "test" not in recorded["identity"]["preprocess"]["splits"] + + same = task_resume_plan(task("AAAA"), recorded, runner_keys=keys) + assert same["stages"]["preprocess"] == "same" + other = task_resume_plan(task("CCCC"), recorded, runner_keys=keys) + assert other["stages"]["preprocess"] == "differs" + assert other["rerun_from"] == "preprocess" + + def test_only_the_named_fewshot_split_is_digested(self): + from datasets import Dataset as HFDataset + + def task(validation_seq): + dataset = MockDataset(SAMPLES) + for split, seq in (("dev", "AAAA"), ("validation", validation_seq)): + dataset.dataset_dict[split] = HFDataset.from_list( + [{"sequence": seq, "partner": "", "answer": ""}] + ) + return NamedShotTask(dataset=dataset, model=_model(), name="seq") + + code = stage_code(NamedShotTask, args=["fewshot_split"])["preprocess"] + one = task_preprocess_identity(task("CCCC"), code) + assert list(one["splits"]) == ["dev"] + # An unread split's rows do not move the identity. + assert task_preprocess_identity(task("GGGG"), code) == one diff --git a/tests/unit/core/runners/test_stage_resume.py b/tests/unit/core/runners/test_stage_resume.py index 8616d632..25d68661 100644 --- a/tests/unit/core/runners/test_stage_resume.py +++ b/tests/unit/core/runners/test_stage_resume.py @@ -755,7 +755,7 @@ async def test_a_refused_resume_appends_no_run(self, tmp_path): assert (root / "meta.json").read_bytes() == before -def test_recorded_identity_reads_schema_1_and_2(): +def test_recorded_identity_reads_schemas_1_to_3(): assert recorded_identity(None) is None assert recorded_identity({"identity": {"schema": 99}}) is None identity = recorded_identity({"identity": {"schema": 2, "data": {"digest": "x"}}}) @@ -769,7 +769,7 @@ def test_recorded_identity_reads_schema_1_and_2(): } identity = recorded_identity({"identity": old}) assert identity is not None - assert identity["schema"] == 2 + assert identity["schema"] == 3 assert identity["preprocess"] == {"digest": "p"} assert identity["postprocess"] is None assert recorded_grader({"identity": old}) == old["grader"] @@ -1300,7 +1300,7 @@ async def test_a_data_change_restarts_and_infers(self, tmp_path, plugin): stages = plan["stages"] assert stages is not None assert stages["stages"]["data"] == "differs" - assert "preprocess" not in stages["stages"] + assert stages["stages"]["preprocess"] == "same" assert stages["rerun_from"] is None assert (plan["to_infer"], plan["to_grade"]) == (1, 1) @@ -1381,3 +1381,207 @@ def test_plan_resume_rolls_back_to_the_earliest_changed_stage(): other = plan_resume(recorded, {**same, "inference": {"digest": "sha256:j"}}) assert other["refused"] == "inference" assert plan_resume(None, same)["identity_schema"] is None + + +# =================================================================== +# preprocess is code: a change restarts every sample, and a directory that +# recorded rendered prompts (identity schema 1 or 2) is upgraded in place +# =================================================================== +def _as_schema_2(root: Path) -> None: + """Rewrite *root* as a schema-2 run left it: ``preprocess`` a digest over + every sample's rendered prompt, each ``preprocess`` record stamped with + its prompt digest instead of a ``preprocess`` key.""" + from sieval.core.tasks.run_meta import aggregate_digest, digest, prompt_of + + prompts: list[tuple[int, None, str]] = [] + for path in root.rglob("*.jsonl"): + lines = [] + for line in path.read_bytes().splitlines(): + record = orjson.loads(line) + entries = list((record.get("stage_meta") or {}).values()) + if isinstance(record.get("meta_last"), dict): + entries.append([record["meta_last"]]) + pre = record.get("preprocess_result") + for history in entries: + for entry in history: + stamp = entry.get("identity") + if isinstance(stamp, dict) and "preprocess" in stamp: + stamp.pop("preprocess") + stamp["prompt"] = digest(prompt_of(pre)) + if record.get("stage") == "preprocessed": + prompts.append((record["sample_id"], None, digest(prompt_of(pre)))) + lines.append(orjson.dumps(record)) + path.write_bytes(b"\n".join(lines) + b"\n") + idx = path.with_suffix(".idx") + offset, out = 0, [] + for raw, entry in zip(lines, idx.read_text().splitlines(), strict=True): + parts = entry.split("\t") + parts[3], parts[4] = str(offset), str(len(raw) + 1) + out.append("\t".join(parts)) + offset += len(raw) + 1 + idx.write_text("\n".join(out) + "\n") + meta = _meta(root) + meta["identity"]["schema"] = 2 + meta["identity"]["preprocess"] = aggregate_digest(prompts) + (root / "meta.json").write_bytes(orjson.dumps(meta)) + + +class TestPreprocessIsCode: + @pytest.mark.anyio + async def test_a_preprocess_arg_change_restarts_every_sample(self, tmp_path): + """Before, only the samples whose rendered prompt changed restarted; + now every sample of the task does, as the identity is per task.""" + root = await _first(tmp_path) + task, model = _task(prefix="Q: ") + await _run(tmp_path, "first", task, resume=True) + assert model.call_count == 3 + run = _meta(root)["runs"][-1] + assert run["resume"]["restarted"] == 3 + assert _meta(root)["identity"]["preprocess"]["args"]["prefix"] == "Q: " + + @pytest.mark.anyio + async def test_preprocess_is_computed_without_preprocessing(self, tmp_path): + """A fast resume of an unchanged run calls no `preprocess`.""" + await _first(tmp_path) + task, model = _task() + await _run(tmp_path, "first", task, resume=True) + assert task.calls["preprocess"] == 0 + assert model.call_count == 0 + + @pytest.mark.anyio + async def test_a_grading_arg_leaves_preprocess(self, tmp_path): + root = await _first(tmp_path) + before = _meta(root)["identity"]["preprocess"] + task, model = _task(threshold=0.9) + await _run(tmp_path, "first", task, resume=True) + assert model.call_count == 0 + assert _meta(root)["identity"]["preprocess"] == before + + @pytest.mark.anyio + async def test_an_interrupted_restart_keeps_preprocess_unknown(self, tmp_path): + """While samples recorded under the old preprocess may remain, the + directory's preprocess is unknown, and each stamp decides.""" + root = await _first(tmp_path) + scope = anyio.CancelScope() + + class Stops(QATask): + async def postprocess(self, inf, ctx): + scope.cancel() + await anyio.sleep_forever() + + task, _ = _task(cls=Stops, prefix="Q: ") + with scope: + await _run(tmp_path, "first", task, resume=True) + assert _meta(root)["identity"]["preprocess"] is None + inferred_new = await _inferred_under(root, _new_key()) + + task, model = _task(prefix="Q: ") + await _run(tmp_path, "first", task, resume=True) + assert _meta(root)["identity"]["preprocess"]["args"]["prefix"] == "Q: " + # Those are kept by their stamp; the rest, recorded under the old + # preprocess, are inferred again. + assert model.call_count == 3 - len(inferred_new) + + +def _new_key() -> str | None: + """The ``preprocess`` key of the ``prefix="Q: "`` task these tests resume.""" + from sieval.core.runners.stage_resume import task_preprocess_identity + from sieval.core.tasks.run_meta import preprocess_key, stage_code + + task, _ = _task(prefix="Q: ") + code = stage_code(type(task), args=list(task.grader_args()))["preprocess"] + return preprocess_key(task_preprocess_identity(task, code)) + + +async def _inferred_under(root: Path, key: str | None) -> set: + """Samples on disk past ``infer`` whose ``preprocess`` record has *key*.""" + from sieval.core.runners.stage_resume import has_graded, recorded_sample_digests + from sieval.core.tasks.consts import TaskStage + from sieval.core.tasks.loader import TaskLoader + + task, _ = _task() + loader = TaskLoader(task=task, root_dir=root) + contexts = await loader.load_initial_state() + await loader.hydrate( + contexts, + set(), + include_stages=set(TaskStage), + record_each_stage=True, + ) + return { + sid + for sid, ctx in contexts.items() + if (ctx.infer_result is not None or has_graded(ctx)) + and recorded_sample_digests(ctx)["preprocess"] == key + } + + +class TestSchema2Upgrade: + @pytest.mark.anyio + async def test_unchanged_prompts_upgrade_in_place(self, tmp_path): + """Rendering the recorded samples once shows nothing changed: every + sample is kept, and the block is rewritten as schema 3.""" + root = await _first(tmp_path) + _as_schema_2(root) + task, model = _task() + await _run(tmp_path, "first", task, resume=True) + assert model.call_count == 0 + assert task.calls["preprocess"] == 3 # the one upgrade pass + assert task.calls["postprocess"] == 0 + identity = _meta(root)["identity"] + assert identity["schema"] == 3 + assert "modules" in identity["preprocess"] + + # Upgraded: the next resume renders nothing. + task, model = _task() + await _run(tmp_path, "first", task, resume=True) + assert task.calls["preprocess"] == 0 + assert model.call_count == 0 + + @pytest.mark.anyio + async def test_unchanged_prompts_upgrade_an_unfinished_run(self, tmp_path): + root = await _first(tmp_path) + _as_schema_2(root) + (root / "report.json").unlink() + task, model = _task() + await _run(tmp_path, "first", task, resume=True) + assert model.call_count == 0 + run = _meta(root)["runs"][-1] + assert run["resume"]["kept"] == 3 + assert run["resume"]["restarted"] == 0 + + @pytest.mark.anyio + async def test_a_changed_prompt_restarts_every_sample(self, tmp_path): + root = await _first(tmp_path) + _as_schema_2(root) + task, model = _task(prefix="Q: ") + await _run(tmp_path, "first", task, resume=True) + assert model.call_count == 3 + assert _meta(root)["runs"][-1]["resume"]["restarted"] == 3 + + @pytest.mark.anyio + async def test_the_plan_reads_the_upgrade(self, tmp_path): + from sieval.core.runners import task_resume_plan + + root = await _first(tmp_path) + _as_schema_2(root) + task, _ = _task() + keys = {"deterministic": False, "max_iterations": 1, "max_retries": 3} + plan = task_resume_plan(task, _meta(root), runner_keys=keys) + assert plan["identity_schema"] == 2 + assert plan["legacy_preprocess"] is True + assert plan["stages"]["preprocess"] == "unknown" + assert plan["rerun_from"] is None + + # plan_run renders the recorded samples as the run would. + runner = TaskRunner( + task, make_config(tmp_path, result_dir=str(root), resume=True) + ) + planned = await runner.plan_run() + assert planned["to_infer"] == 0 + changed, _ = _task(prefix="Q: ") + runner = TaskRunner( + changed, make_config(tmp_path, result_dir=str(root), resume=True) + ) + planned = await runner.plan_run() + assert planned["to_infer"] == 3 diff --git a/tests/unit/core/tasks/test_run_meta.py b/tests/unit/core/tasks/test_run_meta.py index 18a674bd..89260cd4 100644 --- a/tests/unit/core/tasks/test_run_meta.py +++ b/tests/unit/core/tasks/test_run_meta.py @@ -23,6 +23,7 @@ ABSENT, GRADER_PACKAGES, GRADING_STAGES, + IDENTITY_SCHEMA, JUDGE_MODEL_REF_ENV, STAGES, JudgeIdentity, @@ -40,6 +41,7 @@ grader_modules, grader_task, inference_identity, + is_code_identity, judge_identities, judge_model_id, judge_model_refs, @@ -1137,7 +1139,7 @@ def test_carried_identity_keeps_known_stages(self): assert carried["data"] == data assert carried["report"] is None - def test_a_schema_1_block_reads_as_schema_2(self): + def test_a_schema_1_block_reads_as_this_schema(self): data = {"digest": "sha256:d", "samples": 1} prompt = {"digest": "sha256:p", "samples": 1} grader = {"digest": "sha256:g", "modules": {}, "judges": []} @@ -1151,7 +1153,7 @@ def test_a_schema_1_block_reads_as_schema_2(self): read = read_identity(block) assert read is not None identity, legacy = read - assert identity["schema"] == 2 + assert identity["schema"] == IDENTITY_SCHEMA == 3 assert identity["data"] == data assert identity["preprocess"] == prompt assert identity["inference"] == {"digest": "sha256:i"} @@ -1159,10 +1161,20 @@ def test_a_schema_1_block_reads_as_schema_2(self): assert legacy == grader assert carried_identity({"identity": block}) == identity + def test_a_schema_2_block_is_read_as_written(self): + prompts = {"digest": "sha256:p", "samples": 1} + read = read_identity({"schema": 2, "preprocess": prompts}) + assert read is not None + identity, legacy = read + assert identity["schema"] == 3 + assert identity["preprocess"] == prompts + assert not is_code_identity(identity["preprocess"]) + assert legacy is None + def test_another_schema_is_not_read(self): - assert read_identity({"schema": 3, "data": {"digest": "x"}}) is None + assert read_identity({"schema": 4, "data": {"digest": "x"}}) is None assert read_identity("nope") is None - assert carried_identity({"identity": {"schema": 3}}) == empty_identity() + assert carried_identity({"identity": {"schema": 4}}) == empty_identity() def test_carried_runs_tolerates_old_and_malformed(self): assert carried_runs(None) == [] diff --git a/tests/unit/core/tasks/test_run_meta_preprocess.py b/tests/unit/core/tasks/test_run_meta_preprocess.py new file mode 100644 index 00000000..73ac88e2 --- /dev/null +++ b/tests/unit/core/tasks/test_run_meta_preprocess.py @@ -0,0 +1,252 @@ +""" +Tests for the code-based ``preprocess`` identity in ``run_meta``. + +Each test writes a throwaway ``sieval_*`` plugin package, so the walk reads +real source files, and edits it to see whether ``preprocess`` changes. + +AI-Generated Code - Claude Opus 5.5 (Anthropic) +""" + +import importlib +import sys +import textwrap + +import pytest + +from sieval.core.tasks.run_meta import ( + is_code_identity, + preprocess_identity, + preprocess_key, + split_digests, + stage_code, +) + +PACKAGE = "sieval_preprocess_plugin" + +TASK = """ +from sieval.core.tasks.task import Task + +from .render import render + +HEADER = "Classify the sequence." + + +class SeqTask(Task): + def __init__(self, dataset, model, name=None, label="promoter", k=1): + super().__init__(dataset, model, name) + self._label = label + self._k = k + + async def preprocess(self, raw, ctx): + return render(HEADER, self._label, raw["sequence"]) + + async def infer(self, pre, ctx): + return pre + + async def postprocess(self, inf, ctx): + return inf.strip() + + async def feedback(self, post, ctx): + return True, post == "yes" + + async def report(self, finals, fails): + return {"k": self._k} +""" + +FILES = { + "__init__.py": "", + "task.py": TASK, + "render.py": ( + "def render(header, label, seq):\n" + " return f'{header} Is it a {label}?\\n{seq}'\n" + ), +} + +ARGS = {"label": "promoter", "k": 1} + + +@pytest.fixture +def plugin(tmp_path, monkeypatch): + """Write the plugin package; ``load()`` imports it afresh, after edits.""" + monkeypatch.setattr(sys, "dont_write_bytecode", True) + root = tmp_path / "plugins" + pkg = root / PACKAGE + pkg.mkdir(parents=True) + for name, text in FILES.items(): + (pkg / name).write_text(textwrap.dedent(text)) + monkeypatch.syspath_prepend(str(root)) + + def drop(): + for name in list(sys.modules): + if name == PACKAGE or name.startswith(PACKAGE + "."): + sys.modules.pop(name, None) + + def load(): + drop() + return importlib.import_module(f"{PACKAGE}.task").SeqTask + + yield pkg, load + drop() + + +def _edit(pkg, name, old, new): + path = pkg / name + text = path.read_text() + assert old in text + path.write_text(text.replace(old, new)) + + +def _identity(load, args=ARGS): + code = stage_code(load(), args=list(args))["preprocess"] + return preprocess_identity(code, args) + + +class TestCodeChanges: + def test_preprocess_holds_its_code_and_the_args_it_reads(self, plugin): + _, load = plugin + identity = _identity(load) + assert is_code_identity(identity) + assert "./task.py:SeqTask.preprocess" in identity["modules"] + assert "./render.py" in identity["modules"] + # `k` is read only by `report`. + assert identity["args"] == {"label": "promoter"} + assert "fallback" not in identity + + def test_an_edit_to_preprocess_code_changes_it(self, plugin): + pkg, load = plugin + before = _identity(load) + _edit(pkg, "task.py", 'raw["sequence"])', 'raw["sequence"].upper())') + assert preprocess_key(_identity(load)) != preprocess_key(before) + + def test_an_edit_to_a_helper_it_calls_changes_it(self, plugin): + pkg, load = plugin + before = _identity(load) + _edit(pkg, "render.py", "Is it a", "Is this a") + assert preprocess_key(_identity(load)) != preprocess_key(before) + + def test_a_module_constant_changes_it(self, plugin): + pkg, load = plugin + before = _identity(load) + _edit(pkg, "task.py", "Classify the sequence.", "Classify it.") + assert preprocess_key(_identity(load)) != preprocess_key(before) + + def test_an_edit_to_an_unrelated_method_does_not(self, plugin): + pkg, load = plugin + before = _identity(load) + _edit(pkg, "task.py", 'return {"k": self._k}', 'return {"k": self._k + 1}') + _edit(pkg, "task.py", 'post == "yes"', 'post == "no"') + assert _identity(load) == before + + def test_an_arg_preprocess_reads_changes_it(self, plugin): + _, load = plugin + before = _identity(load) + after = _identity(load, {**ARGS, "label": "enhancer"}) + assert preprocess_key(after) != preprocess_key(before) + + def test_an_arg_preprocess_does_not_read_leaves_it(self, plugin): + _, load = plugin + assert _identity(load, {**ARGS, "k": 5}) == _identity(load) + + def test_an_arg_kept_under_another_name_counts_with_an_override(self, plugin): + """A ``grader_args`` override may return ``label`` while ``preprocess`` + reads ``self._kind``: every argument it returns counts.""" + pkg, load = plugin + _edit(pkg, "task.py", "self._label = label", "self._kind = label") + _edit(pkg, "task.py", "self._label, raw", "self._kind, raw") + _edit( + pkg, + "task.py", + " async def infer(", + " def grader_args(self):\n" + ' return {"label": self._kind, "k": self._k}\n\n' + " async def infer(", + ) + identity = _identity(load) + assert identity["args"] == ARGS + after = _identity(load, {**ARGS, "label": "enhancer"}) + assert preprocess_key(after) != preprocess_key(identity) + + def test_the_key_ignores_why_it_fell_back(self, plugin): + _, load = plugin + identity = _identity(load) + explained = {**identity, "fallback": "a reason"} + assert preprocess_key(explained) == preprocess_key(identity) + + +class TestDynamicCode: + def test_a_file_read_at_run_time_counts_the_package_and_its_data(self, plugin): + """A template opened by path is a file no import names: the stage + counts all of the task's code and the data files beside it.""" + pkg, load = plugin + (pkg / "template.txt").write_text("Is it a {label}?") + _edit( + pkg, + "task.py", + ' return render(HEADER, self._label, raw["sequence"])\n', + " here = __file__.rpartition('/')[0]\n" + " with open(here + '/template.txt') as f:\n" + " template = f.read()\n" + ' return template.format(label=self._label) + raw["sequence"]\n', + ) + identity = _identity(load) + assert "read a file at run time" in identity["fallback"] + assert "./template.txt" in identity["modules"] + # Under the fallback every argument counts. + assert identity["args"] == ARGS + + before = preprocess_key(identity) + (pkg / "template.txt").write_text("Is this a {label}?") + assert preprocess_key(_identity(load)) != before + + def test_a_markdown_template_is_data_but_agent_notes_are_not(self, plugin): + pkg, load = plugin + (pkg / "prompt.md").write_text("Is it a {label}?") + (pkg / "CLAUDE.md").write_text("notes") + _edit( + pkg, + "task.py", + ' return render(HEADER, self._label, raw["sequence"])\n', + " here = __file__.rpartition('/')[0]\n" + " with open(here + '/prompt.md') as f:\n" + ' return f.read() + raw["sequence"]\n', + ) + modules = _identity(load)["modules"] + assert "./prompt.md" in modules + assert "./CLAUDE.md" not in modules + + def test_a_run_time_name_falls_back(self, plugin): + pkg, load = plugin + _edit( + pkg, + "task.py", + ' return render(HEADER, self._label, raw["sequence"])\n', + ' return getattr(self, "_" + "render")(raw)\n', + ) + identity = _identity(load) + assert "run time" in identity["fallback"] + assert identity["args"] == ARGS + + def test_a_dataset_split_read_counts_every_split(self, plugin): + pkg, load = plugin + _edit( + pkg, + "task.py", + ' return render(HEADER, self._label, raw["sequence"])\n', + " shots = self.dataset.dataset_dict['train']\n" + ' return str(list(shots)) + raw["sequence"]\n', + ) + code = stage_code(load(), args=list(ARGS))["preprocess"] + assert code["splits"] is True + one = preprocess_identity(code, ARGS, split_digests({"train": [{"s": "A"}]})) + two = preprocess_identity(code, ARGS, split_digests({"train": [{"s": "C"}]})) + assert preprocess_key(one) != preprocess_key(two) + + +def test_a_recorded_digest_of_prompts_is_not_a_code_identity(): + assert not is_code_identity({"digest": "sha256:p", "samples": 3}) + assert not is_code_identity(None) + assert preprocess_key({"digest": "sha256:p", "samples": 3}) is None + + +def test_split_digests_is_unknown_for_an_undigestable_row(): + assert split_digests({"train": [object()]}) is None diff --git a/tests/unit/core/tasks/test_saver.py b/tests/unit/core/tasks/test_saver.py index 185ce962..39ddd552 100644 --- a/tests/unit/core/tasks/test_saver.py +++ b/tests/unit/core/tasks/test_saver.py @@ -392,7 +392,7 @@ async def test_fresh_run_creates_dir_and_file(self, tmp_path): assert meta["task"] == _IDENTITY assert meta["model"] == "candidate" assert meta["identity"] == { - "schema": 2, + "schema": 3, "data": None, "preprocess": None, "inference": None, @@ -565,7 +565,7 @@ async def test_identity_of_another_schema_is_not_carried(self, tmp_path): ) await TaskSaver(root_dir=root).begin_run(resumed=True) meta = _read_meta(root) - assert meta["identity"]["schema"] == 2 + assert meta["identity"]["schema"] == 3 assert meta["identity"]["data"] is None # Malformed entries are dropped, well-formed ones kept. assert [r["start"] for r in meta["runs"]] == ["fresh", "resume"] @@ -596,7 +596,7 @@ async def test_an_unknown_stage_is_refused(self, saver): await saver.set_identity(grader=None) @pytest.mark.anyio - async def test_a_schema_1_identity_is_carried_as_schema_2(self, tmp_path): + async def test_a_schema_1_identity_is_carried_as_schema_3(self, tmp_path): """`prompt` becomes `preprocess`; the unsplittable `grader` is dropped, leaving the grading stages unknown until the run writes them.""" root = tmp_path / "old" @@ -618,7 +618,7 @@ async def test_a_schema_1_identity_is_carried_as_schema_2(self, tmp_path): ) await TaskSaver(root_dir=root).begin_run(resumed=True) assert _read_meta(root)["identity"] == { - "schema": 2, + "schema": 3, "data": {"digest": "sha256:d", "samples": 1}, "preprocess": {"digest": "sha256:p", "samples": 1}, "inference": {"digest": "sha256:i"},