From f14f987c61e7d351c2845870c7fec3e9d91248c6 Mon Sep 17 00:00:00 2001 From: Ethan Date: Fri, 23 Oct 2026 16:01:58 +0800 Subject: [PATCH] feat(cli): sieval run starts an engine only for samples left to infer sieval run resolved every plan and launched every engine before the session planned its resume, so a resume that reused every inference still took a GPU and loaded the model, and could not run on a CPU node at all. A run that may resume now plans first. The session is built as for the run, with each served model bound to the deployment launching it would realize (unlaunched_deployment: the same plan, endpoints and engine, nothing listening), so every identity and binding is the launched one. Each task's TaskRunner.plan_run runs the resume planning arun runs, read-only: both call the one _plan_recorded, so the plan counts the samples the run infers, and a resume= is read where it lies, not copied. Only the models some task still has samples to infer with are launched. With none, no engine starts and the run needs no GPU; when some sample needs one and nvidia-smi finds no GPU, the run is refused before anything starts. A fresh run is not planned. Unlaunched models reach the session as unlaunched_deployments and their runners as unserved_candidate: a run whose directory changed since it was planned is refused before it records anything, and a sample that a rerun feedback sends to another iteration fails at infer, before any request, to be retried by a resume that starts the engine. Judges and other external models are connected lazily already; nothing calls them until a sample is graded. Launched engines keep their logs, infer_plans.yaml and realized_deployments as before. A stub-engine run on main and on this branch records the same inference digest and request digest, fresh and resumed. Closes #213. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 + docs/guide/infer.md | 17 ++ sieval/cli/run.py | 111 ++++++- sieval/cli/session.py | 122 +++++++- sieval/core/runners/__init__.py | 2 + sieval/core/runners/multi_runner.py | 22 +- sieval/core/runners/runner.py | 243 ++++++++++++--- sieval/core/runners/stage_resume.py | 23 ++ sieval/infer/deployer.py | 38 ++- sieval/infer/introspect.py | 14 +- tests/integration/test_context_overflow.py | 41 ++- tests/integration/test_lazy_launch.py | 336 +++++++++++++++++++++ tests/unit/cli/test_cli_model.py | 1 + tests/unit/cli/test_run.py | 20 +- tests/unit/cli/test_session.py | 5 + tests/unit/core/runners/test_plan_run.py | 273 +++++++++++++++++ 16 files changed, 1202 insertions(+), 68 deletions(-) create mode 100644 tests/integration/test_lazy_launch.py create mode 100644 tests/unit/core/runners/test_plan_run.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a3352bb..f9f81a12 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,11 +18,13 @@ Planned as 0.12.0. - **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 - **`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. ## [0.11.0] - 2026-10-23 diff --git a/docs/guide/infer.md b/docs/guide/infer.md index 1b754db1..93d07652 100644 --- a/docs/guide/infer.md +++ b/docs/guide/infer.md @@ -155,6 +155,23 @@ Models with a `path` field (and no `api_base`) or an `infer` section in the YAML A top-level `api_base` turns auto-serve off. An `args.api_base` next to `path` or `infer.checkpoint` is refused before any engine starts: keep one or the other. +A run that may resume (`--resume`, or `resume` in a `runner_config`) plans each +task's resume before it starts any engine, with the same per-sample planning +the run uses, and launches one only for a model some task still has samples to +infer with. A resume that keeps every sample, or reruns only from +`postprocess`, `feedback` or `report`, starts none, so it runs on a node +without GPUs; its `infer_plans.yaml`, the inference identity and `realized_deployments` +are what a launch would record, and no engine log is written. When some sample +needs inference and `nvidia-smi` finds no GPU, the run is refused before +anything starts. A judge or other external model is called only when a sample +is graded. A recorded context overflow the resume scores from its record still +starts the engine: a task that calls the model more than once asks again. + +A sample that a rerun `feedback` sends to another iteration needs the model +again, which planning cannot foresee. With no engine started it fails at +`infer`, before any request, and a later `--resume` that starts the engine +retries it. + ### Anthropic Messages To call the native Anthropic Messages API, select the dialect explicitly and diff --git a/sieval/cli/run.py b/sieval/cli/run.py index e1a930c7..95b9260b 100644 --- a/sieval/cli/run.py +++ b/sieval/cli/run.py @@ -48,6 +48,7 @@ from sieval.cli.resolution import resolve_config_model_types from sieval.cli.services import SERVICE_LOG_DIR from sieval.cli.session import ( + config_resumes, holds_secret, read_persisted_config, redact_secrets, @@ -56,11 +57,16 @@ ) from sieval.cli.signals import terminate_as_interrupt from sieval.core.models import Deployment +from sieval.core.runners import RunPlan from sieval.core.utils.logging import configure_logging, log_user from sieval.infer.backends import get_translator from sieval.infer.backends.translator import BackendCommand, inject_user_env from sieval.infer.config import InferHandle -from sieval.infer.deployer import DEFAULT_READY_TIMEOUT, LocalDeployer +from sieval.infer.deployer import ( + DEFAULT_READY_TIMEOUT, + LocalDeployer, + unlaunched_deployment, +) from sieval.infer.introspect import ( check_launch_gpu, declared_target_gpu, @@ -289,6 +295,81 @@ async def _prepare_launch_batch( return plan_dicts, prepared_launches +def _resumes(resume: bool | str | Path | None, config: Mapping[str, Any]) -> bool: + """Whether some task may resume: ``--resume``, or a ``runner_config`` key.""" + return bool(resume) or config_resumes(config) + + +def _unlaunched( + prepared_launches: Mapping[str, tuple[DeploymentPlan, list[BackendCommand]]], + names: set[str], +) -> dict[str, Deployment]: + return { + name: unlaunched_deployment(*prepared_launches[name]) for name in sorted(names) + } + + +async def _plan_tasks( + config_path: Path, + prepared_launches: Mapping[str, tuple[DeploymentPlan, list[BackendCommand]]], + plan_dicts: dict[str, dict[str, Any]], + *, + model: str | None, + resume: bool | str | Path | None, + result_dir: str | None, + deterministic: bool | None, + invocation: str, + cli_model: CliModel | None, +) -> tuple[dict[str, str], dict[str, RunPlan | Exception]]: + """Each task's served model and resume plan, with no engine launched. + + The one call that turns resume planning into a launch decision. + """ + from sieval.cli.session import aplan_session + + return await aplan_session( + config_path, + model=model, + resume=resume, + result_dir=result_dir, + deterministic=deterministic, + infer_plans=plan_dicts or None, + invocation=invocation, + self_managed_endpoints=frozenset(prepared_launches), + unlaunched_deployments=_unlaunched(prepared_launches, set(prepared_launches)), + cli_model=cli_model, + ) + + +async def _models_to_infer( + config_path: Path, + config: Mapping[str, Any], + prepared_launches: Mapping[str, tuple[DeploymentPlan, list[BackendCommand]]], + plan_dicts: dict[str, dict[str, Any]], + **session: Any, +) -> set[str]: + """The served models some task still has samples to infer with. + + A run where nothing resumes infers everything, so it is not planned. A + task whose plan raises infers nothing: its run raises the same before its + first sample. + """ + if not prepared_launches: + return set() + if not _resumes(session["resume"], config): + return set(prepared_launches) + task_models, plans = await _plan_tasks( + config_path, prepared_launches, plan_dicts, **session + ) + return { + task_models[task] + for task, plan in plans.items() + if not isinstance(plan, Exception) + and plan["to_infer"] + and task_models.get(task) in prepared_launches + } + + async def _run_dry_run( config_path: Path, *, @@ -447,9 +528,27 @@ async def _run_all( f"Model name {SERVICE_LOG_DIR!r} is reserved for " f"/logs/{SERVICE_LOG_DIR}/; rename the model." ) - if prepared_launches: - await check_launch_gpu() + # Plan the resume first: an engine starts only for a model some task + # still has samples to infer with. + to_launch = await _models_to_infer( + config_path, + config, + prepared_launches, + plan_dicts, + model=model, + resume=resume, + result_dir=result_dir, + deterministic=deterministic, + invocation=invocation, + cli_model=cli_model, + ) + if to_launch: + await check_launch_gpu(required=True) + for model_name in sorted(set(prepared_launches) - to_launch): + log_user("No inference needed for model {}; not starting it", model_name) for model_name, (plan, commands) in prepared_launches.items(): + if model_name not in to_launch: + continue log_user("Starting inference for model: {}", model_name) last_status = None @@ -536,10 +635,14 @@ def _progress( resume=resume, result_dir=result_dir, deterministic=deterministic, - self_managed_endpoints=frozenset(realized_deployments), + self_managed_endpoints=frozenset(prepared_launches), infer_plans=plan_dicts or None, invocation=invocation, realized_deployments=realized_deployments or None, + unlaunched_deployments=( + _unlaunched(prepared_launches, set(prepared_launches) - to_launch) + or None + ), cli_model=cli_model, infer_gpu=await _plan_gpu_text() if plan_dicts else None, ) diff --git a/sieval/cli/session.py b/sieval/cli/session.py index 6bb8b21a..9b891720 100644 --- a/sieval/cli/session.py +++ b/sieval/cli/session.py @@ -125,6 +125,7 @@ LegacyStageMatch, MultiTaskRunner, ResumeValue, + RunPlan, TaskRunnerConfig, as_resume, fold_resume_aliases, @@ -1273,6 +1274,15 @@ def _runner_config_blocks(cfg: Mapping[str, Any]) -> dict[str, dict[str, Any]]: return {k: v for k, v in blocks.items() if isinstance(v, dict)} +def config_resumes(cfg: Mapping[str, Any]) -> bool: + """Whether some ``runner_config`` of *cfg* sets ``resume`` or an alias of it.""" + return any( + block.get(key) + for block in _runner_config_blocks(cfg).values() + for key in _RESUME_KEYS + ) + + def _drop_resume_keys(cfg: dict[str, Any]) -> None: """Remove every resume key in place, and a ``runner_config`` left empty.""" for block in _runner_config_blocks(cfg).values(): @@ -1671,6 +1681,7 @@ def __init__( resume_from: str | Path | None = None, cli_model: CliModel | None = None, infer_gpu: str | None = None, + unlaunched_deployments: Mapping[str, Deployment] | None = None, ): self.config_path = Path(config_path) self.model_override = model_override @@ -1695,7 +1706,17 @@ def __init__( self._legacy_endpoint_map: Mapping[str, str] = legacy_endpoint_map self._infer_plans: Mapping[str, dict[str, Any]] | None = infer_plans self._infer_gpu = infer_gpu - self._realized_deployments = realized_deployment_map + unlaunched = dict(unlaunched_deployments or {}) + if set(unlaunched) & set(realized_deployment_map): + raise ValueError( + "a root model cannot have both a realized and an unlaunched " + "deployment: " + + ", ".join(sorted(set(unlaunched) & set(realized_deployment_map))) + ) + # Roots served by no engine: bound like launched ones, so every + # identity is the launched one, but a task never infers through them. + self._unlaunched_roots = frozenset(unlaunched) + self._realized_deployments = {**realized_deployment_map, **unlaunched} for root_name, deployment in self._realized_deployments.items(): if not isinstance(root_name, str) or not root_name: raise TypeError("realized deployment keys must be non-empty strings") @@ -4870,6 +4891,9 @@ def _setup_tasks(self) -> None: inference_recipe=self._inference_recipe_for(candidate.config_name), resume_config_diff=list(self._resume_config_diff), legacy_match=self._legacy_match_for(task_name, runner_config), + unserved_candidate=( + self._root_model(candidate.config_name) in self._unlaunched_roots + ), ) logger.info("Added task '{}' with class '{}'", task_name, task_spec) @@ -4934,6 +4958,25 @@ def _legacy_match_for( == _stage_config(persisted, recorded_name, "inference"), } + def _root_model(self, model_name: str) -> str: + """The root of *model_name*'s ``base`` chain: the model that is served.""" + models_cfg = self._get_named_config_map("models") + if model_name not in models_cfg: + return model_name + return self._model_config_chain(model_name, models_cfg)[0][0] + + def task_models(self) -> dict[str, str]: + """Each task's model under test, as the root model config that serves it. + + Read after :meth:`prepare_prelaunch`. + """ + out: dict[str, str] = {} + for task_name, context in self._task_requirement_contexts.items(): + candidate = context.model_bindings.get("candidate") + if isinstance(candidate, NamedModelBinding): + out[task_name] = self._root_model(candidate.config_name) + return out + def _inference_recipe_for(self, model_name: str) -> dict[str, JSONValue] | None: """The serving plan of *model_name*'s root, for the inference identity. @@ -4947,13 +4990,7 @@ def _inference_recipe_for(self, model_name: str) -> dict[str, JSONValue] | None: infer_plans = self._infer_plans if not infer_plans: return None - models_cfg = self._get_named_config_map("models") - root = ( - self._model_config_chain(model_name, models_cfg)[0][0] - if model_name in models_cfg - else model_name - ) - plan = infer_plans.get(root) + plan = infer_plans.get(self._root_model(model_name)) if plan is None: return None recipe = cast(dict[str, Any], copy.deepcopy(unwrap_proxies(plan))) @@ -5283,6 +5320,34 @@ async def arun(self) -> dict[str, Any]: finally: await self._close_owned_model_resources() + async def aplan(self) -> dict[str, RunPlan | Exception]: + """What :meth:`arun` would do with each task, by task name; writes nothing. + + Prepares the session as :meth:`arun` does, vendored services included, + and asks each task's runner for its :meth:`TaskRunner.plan_run`. A + task whose planning raises maps to the exception its run would raise. + Raises what :meth:`arun` would raise before any task starts. Service + logs go to ``~/.sieval/logs``, never into a result directory. + """ + try: + await run_sync(self.prepare_prelaunch) + self._stamp_deterministic_seed_contract() + result_dir = self._resolve_result_dir() + if self.resume_override is True and result_dir: + # What `_persist_effective_config` reads before it rewrites. + self._recorded_config = await run_sync( + read_persisted_config, Path(result_dir) / "effective_config.yaml" + ) + async with vendored_services_for( + self._required_services, result_dir=lambda: None + ): + await self._prepare_execution() + if self.runner is None: + raise RuntimeError("Runner not initialized") + return await self.runner.plan_run() + finally: + await self._close_owned_model_resources() + async def _close_owned_model_resources(self) -> None: """Drain and close every session-owned root once; never close externals.""" @@ -5331,6 +5396,7 @@ async def arun_session( resume_from: str | Path | None = None, cli_model: CliModel | None = None, infer_gpu: str | None = None, + unlaunched_deployments: Mapping[str, Deployment] | None = None, ) -> dict[str, Any]: """Run tasks defined in a YAML configuration file asynchronously. @@ -5363,6 +5429,10 @@ async def arun_session( infer_gpu: The GPUs *infer_plans* were resolved for, as ``SIEVAL_TARGET_GPU`` spells them; recorded in ``infer_plans.yaml`` as ``gpu``, so a resume can tell whether it may reuse the plans. + unlaunched_deployments: For a root model ``sieval run`` serves but did + not launch, since no task had anything for it to infer: the + deployment launching it would realize, nothing listening. Its + tasks bind and resume as if it were launched, and refuse to infer. Returns: A dictionary mapping task names to their reports. @@ -5381,11 +5451,47 @@ async def arun_session( resume_from=resume_from, cli_model=cli_model, infer_gpu=infer_gpu, + unlaunched_deployments=unlaunched_deployments, ) return await runner.arun() +async def aplan_session( + config: str | Path, + *, + model: str | None = None, + resume: bool | str | Path | None = None, + result_dir: str | None = None, + deterministic: bool | None = None, + infer_plans: Mapping[str, dict[str, Any]] | None = None, + invocation: str | None = None, + self_managed_endpoints: frozenset[str] | set[str] = frozenset(), + unlaunched_deployments: Mapping[str, Deployment] | None = None, + cli_model: CliModel | None = None, +) -> tuple[dict[str, str], dict[str, RunPlan | Exception]]: + """Plan the run :func:`arun_session` would make with the same arguments. + + Returns each task's model under test (its root model config) and its + :meth:`EvalSession.aplan` entry. Every served model passes as one of + *unlaunched_deployments*. Reads ``--resume`` directories, writes nothing. + """ + session = EvalSession( + config_path=config, + model_override=model, + resume=resume, + result_dir_override=result_dir, + deterministic_override=deterministic, + infer_plans=infer_plans, + invocation=invocation, + self_managed_endpoints=self_managed_endpoints, + unlaunched_deployments=unlaunched_deployments, + cli_model=cli_model, + ) + plans = await session.aplan() + return session.task_models(), plans + + def run_session( config: str | Path, model: str | None = None, diff --git a/sieval/core/runners/__init__.py b/sieval/core/runners/__init__.py index 8f1fe449..c4c1514a 100644 --- a/sieval/core/runners/__init__.py +++ b/sieval/core/runners/__init__.py @@ -19,6 +19,7 @@ from sieval.core.runners.stage_resume import ( LegacyStageMatch, ResumePlan, + RunPlan, StageResumeError, plan_resume, task_resume_plan, @@ -31,6 +32,7 @@ "ResumeAction", "ResumePlan", "ResumeValue", + "RunPlan", "StageResumeError", "TaskRunner", "TaskRunnerConfig", diff --git a/sieval/core/runners/multi_runner.py b/sieval/core/runners/multi_runner.py index 65b9e56b..c09b0d13 100644 --- a/sieval/core/runners/multi_runner.py +++ b/sieval/core/runners/multi_runner.py @@ -14,7 +14,7 @@ from sieval.core.types import JSONValue from .runner import TaskRunner, TaskRunnerConfig -from .stage_resume import LegacyStageMatch +from .stage_resume import LegacyStageMatch, RunPlan class MultiTaskRunner: @@ -69,11 +69,12 @@ def add_task( inference_recipe: dict[str, JSONValue] | None = None, resume_config_diff: list[str] | None = None, legacy_match: LegacyStageMatch | None = None, + unserved_candidate: bool = False, ) -> Self: """Register a task for parallel execution. - *inference_recipe*, *resume_config_diff* and *legacy_match* pass - through to :class:`TaskRunner`. + *inference_recipe*, *resume_config_diff*, *legacy_match* and + *unserved_candidate* pass through to :class:`TaskRunner`. Auto-sets ``result_dir`` to ``/`` when a base was provided and the task's config lacks its own. @@ -103,6 +104,7 @@ def add_task( inference_recipe=inference_recipe, resume_config_diff=resume_config_diff, legacy_match=legacy_match, + unserved_candidate=unserved_candidate, ) if runner.root_dir in self._output_dirs: @@ -116,6 +118,20 @@ def add_task( self._output_dirs.add(runner.root_dir) return self + async def plan_run(self) -> dict[str, RunPlan | Exception]: + """Each task's :meth:`TaskRunner.plan_run`, by task name; read-only. + + A task whose planning raises maps to the exception, which its run + would raise before its first sample. Runners that planned are not run. + """ + plans: dict[str, RunPlan | Exception] = {} + for runner in self._runners: + try: + plans[runner._task.name] = await runner.plan_run() + except Exception as exc: + plans[runner._task.name] = exc + return plans + def run(self) -> dict[str, Any]: """Synchronous wrapper around :meth:`arun`.""" return anyio.run(self.arun) diff --git a/sieval/core/runners/runner.py b/sieval/core/runners/runner.py index 751391c0..92213d85 100644 --- a/sieval/core/runners/runner.py +++ b/sieval/core/runners/runner.py @@ -106,6 +106,7 @@ from .resume_option import as_resume, fold_resume_aliases, resume_source from .stage_resume import ( LegacyStageMatch, + RunPlan, StageResumeError, abandon_copy, copy_pending, @@ -490,12 +491,15 @@ def __init__( inference_recipe: dict[str, JSONValue] | None = None, resume_config_diff: list[str] | None = None, legacy_match: LegacyStageMatch | None = None, + unserved_candidate: bool = False, ): """*inference_recipe* is the serving plan of the model under test, without its checkpoint, for the inference identity; ``None`` when the model is not served by SiEval. *resume_config_diff* is the config change a resume made, for ``runs[].config``. *legacy_match* compares a recorded run without identities through its persisted config. + *unserved_candidate* says no engine serves the model under test, since + a plan found nothing to infer: a run that would infer is refused. """ # A malformed judge reference would leave every grader identity # unknown; refuse it before any work. @@ -505,6 +509,7 @@ def __init__( self._inference_recipe = inference_recipe self._resume_config_diff = list(resume_config_diff or []) self._legacy_match = LegacyStageMatch(**(legacy_match or {})) + self._unserved_candidate = unserved_candidate # Per-stage resume: what was copied in, and what the comparison chose. self._run_from: RunFrom | None = None self._copied: list[Path] = [] @@ -898,21 +903,8 @@ async def _begin_run(self, resumed: bool) -> list[str | int]: keeps_recorded_model = False if resumed: # Plan before `begin_run`: a refusal leaves meta.json as it was. - self._contexts = await self._loader.load_initial_state() - if any(_reuses_inference(c) for c in self._contexts.values()): - self._check_inference(inference) - self._grading_change = self._recorded_grading_change() - await self._hydrate_initial(resumed=True) - # Decide per sample what a resume keeps before failures roll back, - # so a sample that failed in grading regrades rather than reinfers. - changed = await self._plan_stage_resume() - self._collect_overflow_replays(inference) - self._loader.prepare_failed_retries(self._contexts, self._record_each_stage) - for sid in self._overflow_replays: - # Scored from the record, not retried: no retry is spent. - ctx = self._contexts[sid] - self._contexts[sid] = replace(ctx, retry_count=ctx.retry_count - 1) - keeps_recorded_model = self._model_differs() and self._check_model_switch() + changed, keeps_recorded_model = await self._plan_recorded(inference) + self._refuse_unserved() # A changed grading stage, and every one after it, is unknown until the # run ends: samples not reached yet hold the recorded verdicts. One # that did not change is written now, upgrading a schema-1 block. @@ -942,6 +934,33 @@ async def _begin_run(self, resumed: bool) -> list[str | int]: ) return changed + async def _plan_recorded( + self, inference: InferenceIdentity | None + ) -> tuple[list[str | int], bool]: + """Load the recorded contexts and decide what a resume does with each. + + The one resume plan, shared by :meth:`_begin_run` and :meth:`plan_run` + so that a plan made before launch counts the samples the run infers. + Writes nothing; raises a refusal before anything is recorded. Returns + the samples it rewinds, and whether the recorded model is kept. + """ + self._contexts = await self._loader.load_initial_state() + if any(_reuses_inference(c) for c in self._contexts.values()): + self._check_inference(inference) + self._grading_change = self._recorded_grading_change() + await self._hydrate_initial(resumed=True) + # Decide per sample what a resume keeps before failures roll back, + # so a sample that failed in grading regrades rather than reinfers. + changed = await self._plan_stage_resume() + self._collect_overflow_replays(inference) + self._loader.prepare_failed_retries(self._contexts, self._record_each_stage) + for sid in self._overflow_replays: + # Scored from the record, not retried: no retry is spent. + ctx = self._contexts[sid] + self._contexts[sid] = replace(ctx, retry_count=ctx.retry_count - 1) + keeps_recorded_model = self._model_differs() and self._check_model_switch() + return changed, keeps_recorded_model + def _collect_overflow_replays(self, inference: InferenceIdentity | None) -> None: """Samples an earlier run failed with a context overflow at ``infer``. @@ -997,6 +1016,41 @@ async def _fast_resume(self) -> JSONValue | None: grader identity reads, so even a fast resume pays for ``setup()``: a task that contacts a service there needs it up. """ + reusable = await self._reusable_report() + if reusable is None: + return None + cached_report, rereport = reusable + if rereport: + return await self._rereport() + # Nothing reruns. The identity is kept, but for grading identities a + # schema-1 block lacked; the views are rebuilt from the shards like + # every other exit. + await self._saver.begin_run( + resumed=True, + model=self._model_name(), + config_diff=self._resume_config_diff, + config_path=self._config_path(), + identity=self._grading_identity_update(), + ) + self._contexts = await self._loader.load_initial_state() + await self._loader.hydrate( + self._contexts, + self._hydrated_ids, + include_stages={TaskStage.FINAL, TaskStage.FAILED}, + prepare_retries=False, + record_each_stage=self._record_each_stage, + ) + # The resume gate pins the dataset, and the report says every sample + # is final, so the recorded samples are the whole run. + self._total_samples = len(self._contexts) + self._resume_plan = _no_outcomes() + self._resume_plan["kept"] = len(self._contexts) + return await self._finalize(cached_report=cached_report) + + async def _reusable_report(self) -> tuple[JSONValue, bool] | None: + """The recorded ``report.json`` a resume finishes on, and whether only + ``report`` changed so it is recomputed from the records; ``None`` when + the run must go on. Read-only.""" if not self._resumes: return None cached_report = await self._loader.load_cached_report() @@ -1038,32 +1092,7 @@ async def _fast_resume(self) -> JSONValue | None: # Failures to retry, samples left, or a report from before the # `sieval` counts: the full path decides. return None - if rereport: - return await self._rereport() - # Nothing reruns. The identity is kept, but for grading identities a - # schema-1 block lacked; the views are rebuilt from the shards like - # every other exit. - await self._saver.begin_run( - resumed=True, - model=self._model_name(), - config_diff=self._resume_config_diff, - config_path=self._config_path(), - identity=self._grading_identity_update(), - ) - self._contexts = await self._loader.load_initial_state() - await self._loader.hydrate( - self._contexts, - self._hydrated_ids, - include_stages={TaskStage.FINAL, TaskStage.FAILED}, - prepare_retries=False, - record_each_stage=self._record_each_stage, - ) - # The resume gate pins the dataset, and the report says every sample - # is final, so the recorded samples are the whole run. - self._total_samples = len(self._contexts) - self._resume_plan = _no_outcomes() - self._resume_plan["kept"] = len(self._contexts) - return await self._finalize(cached_report=cached_report) + return cached_report, rereport async def _rereport(self) -> JSONValue: """Recompute ``report.json`` from a finished run's records: re-report. @@ -1459,6 +1488,7 @@ def _run_stage( self._ran_actions.add(action) try: + self._check_served(action) match action: case TaskAction.PREPROCESS: r = await self._task.preprocess(ctx.raw_sample, ctx) @@ -1970,6 +2000,137 @@ def _check_inference(self, current: InferenceIdentity | None) -> None: def _recorded_dir(self) -> str: return self._run_from["dir"] if self._run_from else str(self._root_dir) + async def plan_run(self) -> RunPlan: + """Decide what :meth:`arun` would do, without doing or writing any of it. + + Runs the resume planning :meth:`arun` runs (:meth:`_plan_recorded`) + against the recorded directory, read where it lies: ``resume=`` is + not copied. Raises what :meth:`arun` would raise before its first + sample, such as a :class:`StageResumeError`. Calls the task's + ``setup()`` and ``shutdown()``, as :meth:`arun` does. A runner that + planned is not run. + """ + await self._task.setup() + try: + return await self._plan_run() + finally: + await self._task.shutdown() + + async def _plan_run(self) -> RunPlan: + test_set = self._task.dataset.test_set + size = len(test_set) if test_set else 0 + if not (self._resumed_from_existing or self._run_from is not None): + return { + "start": "fresh", + "recorded": None, + "samples": size, + "outcomes": None, + "new": size, + "to_infer": size, + "to_grade": size, + } + recorded = self._recorded_dir() + reusable = ( + await self._reusable_report() if self._resumed_from_existing else None + ) + if reusable is not None: + outcomes = _no_outcomes() + outcomes["kept"] = size + if reusable[1]: + outcomes["rereported"] = size + return { + "start": "reuse", + "recorded": recorded, + "samples": size, + "outcomes": outcomes, + "new": 0, + "to_infer": 0, + "to_grade": 0, + } + if self._run_from is not None: + # `arun` copies the directory first; the plan reads it in place. + self._loader = TaskLoader( + task=self._task, + root_dir=Path(self._run_from["dir"]), + shard_read_concurrency=self._config.shard_read_concurrency, + profiler=self._profiler, + ) + await self._plan_recorded(self._inference_identity()) + to_infer, to_grade = self._pending_work() + return { + "start": "resume", + "recorded": recorded, + "samples": size, + "outcomes": self._resume_plan, + "new": sum(1 for sid in range(size) if sid not in self._contexts), + "to_infer": to_infer, + "to_grade": to_grade, + } + + def _pending_work(self) -> tuple[int, int]: + """``(to_infer, to_grade)``: the samples a run starting now would send + to the model under test, and those it would grade. + + Read after :meth:`_plan_recorded`. Counts samples with no record, and + recorded ones that are not terminal and stay within ``max_retries``. A + recorded context overflow the run scores from its record still counts + as inferred, since a task that calls the model more than once infers + it again. A sample ``feedback`` sends to another iteration after a + regrade is not counted: the plan cannot know it (see + :meth:`_check_served`). + """ + test_set = self._task.dataset.test_set + size = len(test_set) if test_set else 0 + to_infer = to_grade = 0 + for sid in range(size): + ctx = self._contexts.get(sid) + if ctx is not None and ( + ctx.is_terminal() + or ( + self._resumes + and self._max_retries is not None + and ctx.retry_count > self._max_retries + ) + ): + continue + to_grade += 1 + if ctx is None or ctx.stage in (TaskStage.INITIAL, TaskStage.PREPROCESSED): + to_infer += 1 + return to_infer, to_grade + + def _refuse_unserved(self) -> None: + """Refuse, before recording anything, a run that would infer with no + engine serving the model under test. + + :meth:`plan_run` counted the same samples through the same + :meth:`_plan_recorded`, so this only fires when the directory changed + between the plan and the run. + """ + if not self._unserved_candidate: + return + to_infer, _ = self._pending_work() + if not to_infer: + return + raise RuntimeError( + f"Task '{self._task.name}' now has {to_infer} sample(s) to infer, but " + "no engine was launched for its model: the plan made before launch " + f"found none. Did {self._recorded_dir()} change since? Run again." + ) + + def _check_served(self, action: TaskAction) -> None: + """Fail a sample about to infer with no engine serving the model. + + Reached when a rerun ``feedback`` sends a sample to another iteration, + which no plan can foresee. The sample fails before any request, and a + resume that launches the engine retries it. + """ + if action is TaskAction.INFER and self._unserved_candidate: + raise RuntimeError( + "infer needs an engine for the model under test, and none was " + "launched: the plan made before launch found nothing to infer. " + "Resume to retry this sample." + ) + def _abandon_copy(self) -> None: """Undo a ``resume=`` copy no run began on.""" if self._run_from is None: diff --git a/sieval/core/runners/stage_resume.py b/sieval/core/runners/stage_resume.py index 1999f303..8f5b5ecd 100644 --- a/sieval/core/runners/stage_resume.py +++ b/sieval/core/runners/stage_resume.py @@ -40,6 +40,7 @@ from sieval.core.tasks.run_meta import ( GRADING_STAGES, STAGES, + ResumeOutcomes, RunIdentity, digest, grader_key, @@ -84,6 +85,28 @@ class LegacyStageMatch(TypedDict, total=False): inference: bool +class RunPlan(TypedDict): + """What a run of one task would do, decided before it does anything. + + Unlike :class:`ResumePlan`, which compares identities alone, this is the + runner's per-sample plan (:meth:`~sieval.core.runners.runner.TaskRunner.plan_run`). + ``start`` is ``fresh`` (nothing recorded), ``resume`` (stage by stage), or + ``reuse`` (the recorded ``report.json`` stands, recomputed when only + ``report`` changed). ``outcomes`` counts the recorded samples as + ``runs[].resume`` will, ``None`` for a fresh run. ``new`` counts samples + with no record. ``to_infer`` counts the samples the run sends to the model + under test, ``to_grade`` those it runs through a grading stage. + """ + + start: Literal["fresh", "resume", "reuse"] + recorded: str | None + samples: int + outcomes: ResumeOutcomes | None + new: int + to_infer: int + to_grade: int + + def recorded_identity(meta: Mapping[str, Any] | None) -> RunIdentity | None: """A recorded ``meta.json``'s ``identity`` block as schema 2. diff --git a/sieval/infer/deployer.py b/sieval/infer/deployer.py index ad68f24c..3dc76509 100644 --- a/sieval/infer/deployer.py +++ b/sieval/infer/deployer.py @@ -231,16 +231,10 @@ async def _launch_one( cwd=cmd.working_dir, ) - # Derive endpoint from health_url - endpoint = None - if cmd.health_url: - parsed = urlparse(cmd.health_url) - endpoint = urlunparse((parsed.scheme, parsed.netloc, "/v1", "", "", "")) - return InferHandle( backend=cmd.backend or cmd.role, # engine name; fallback to role handle_id=str(process.pid), - endpoint=endpoint, + endpoint=command_endpoint(cmd), metadata={ "cmd": cmd.cli_args, "log_file": str(log_file), @@ -567,6 +561,36 @@ def build_deployment( ) +def unlaunched_deployment( + plan: DeploymentPlan, commands: list[BackendCommand] +) -> Deployment: + """The :class:`Deployment` launching *commands* would realize, launching nothing. + + Equal to :meth:`LocalDeployer.build_deployment` of a launch in everything + but ``deployment_id``, which names no process. For planning a run before + deciding whether to launch: nothing listens at its endpoints. + """ + handles = [ + InferHandle( + backend=cmd.backend or cmd.role, + handle_id="unlaunched", + endpoint=command_endpoint(cmd), + metadata={"role": cmd.role}, + ) + for cmd in commands + ] + return LocalDeployer().build_deployment(plan, handles) + + +def command_endpoint(cmd: BackendCommand) -> str | None: + """The OpenAI-compatible endpoint *cmd* serves once launched, from its + health URL; ``None`` without one.""" + if not cmd.health_url: + return None + parsed = urlparse(cmd.health_url) + return urlunparse((parsed.scheme, parsed.netloc, "/v1", "", "", "")) + + def _local_deployment_id( plan_fingerprint: str, handles: list[InferHandle], diff --git a/sieval/infer/introspect.py b/sieval/infer/introspect.py index 91312a8e..80811ed6 100644 --- a/sieval/infer/introspect.py +++ b/sieval/infer/introspect.py @@ -521,19 +521,27 @@ def declared_target_gpu(environ: Mapping[str, str] | None = None) -> GPUInfo | N return parse_target_gpu(raw) if raw.strip() else None -async def check_launch_gpu() -> None: +async def check_launch_gpu(*, required: bool = False) -> None: """Refuse to launch an engine on GPUs other than the declared ones. A plan resolved for :data:`TARGET_GPU_ENV` is only that node's plan on the declared hardware, so every field must match, memory included: two nodes of one SKU can report different ``memory.total`` (ECC, PCIe vs SXM), and the memory decides tensor parallelism. Without a declaration nothing is - checked. + checked, unless *required* (some engine will launch): then a node where + ``nvidia-smi`` finds no GPU is refused before anything starts. """ declared = declared_target_gpu() - if declared is None: + if declared is None and not required: return detected = await detect_local_gpu() + if declared is None: + if detected is None: + raise RuntimeError( + "This run needs an inference engine, and nvidia-smi finds no " + "GPU on this node. Run it on a GPU node." + ) + return if detected == declared: return if detected is None: diff --git a/tests/integration/test_context_overflow.py b/tests/integration/test_context_overflow.py index 362055a9..67ad207a 100644 --- a/tests/integration/test_context_overflow.py +++ b/tests/integration/test_context_overflow.py @@ -20,7 +20,7 @@ import httpx import pytest -from sieval.cli.session import arun_session +from sieval.cli.session import aplan_session, arun_session from sieval.core.runners import TaskRunner from sieval.core.tasks import Task from sieval.core.tasks.records import ( @@ -497,3 +497,42 @@ async def legacy_infer(self, ctx): assert counts["context_overflow"] == 1 # Q2's first turn was answered, and right: kept, not blanked. assert report["t"]["score"] == 1.0 + + @pytest.mark.anyio + @pytest.mark.parametrize("max_retries", [0, 3]) + async def test_the_plan_counts_a_scored_refusal_as_the_run_does( + self, + tmp_path: Path, + asked: _Asked, + monkeypatch: pytest.MonkeyPatch, + max_retries: int, + ) -> None: + # A replay spends no retry, so at max_retries 0 the run still scores Q2. + # A plan that counted the spent retry would find nothing to infer, and + # `sieval run` would refuse the run that launched no engine for it. + asked.answer_with(_with_q3(_refuse("Q2"))) + config = _config(tmp_path, "AnswerTask") + config.write_text( + config.read_text().replace( + "show_progress: false", + f"show_progress: false\n max_retries: {max_retries}", + ) + ) + + async def legacy_infer(self, ctx): + return await self._task.infer(ctx.preprocess_result, ctx), [] + + with monkeypatch.context() as patched: + patched.setattr(TaskRunner, "_infer", legacy_infer) + await arun_session(config) + + _, plans = await aplan_session(config, resume=True) + plan = plans["t"] + assert not isinstance(plan, Exception) + # Counted as inferred: a task calling the model twice asks again. + assert (plan["to_infer"], plan["to_grade"]) == (1, 1) + + report = await arun_session(config, resume=True) + assert asked.count("Q2") == 1, "scored from the record" + assert report["t"]["sieval"]["context_overflow"] == 1 + assert report["t"]["sieval"]["failed"] == 0 diff --git a/tests/integration/test_lazy_launch.py b/tests/integration/test_lazy_launch.py new file mode 100644 index 00000000..a7bb9c1d --- /dev/null +++ b/tests/integration/test_lazy_launch.py @@ -0,0 +1,336 @@ +"""Pattern: ``sieval run`` launches an engine only for samples left to infer. + +A served model goes through the real session, runner and connection factory. +Only the engine is a stub: ``launch_model`` returns the handles a launch of the +translated commands would, and an in-process transport answers at their +endpoint. GPU detection is faked. + +AI-Generated Code - Claude Opus 5.5 (Anthropic) +""" + +import json +from collections.abc import Iterator +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, patch + +import httpx +import orjson +import pytest +import yaml +from openai import AsyncOpenAI + +from sieval.cli.run import _run_all +from sieval.core.models import connection_factory +from sieval.core.models.request_log import RecordingOpenAIClient +from sieval.core.tasks import Task +from sieval.infer.backends.translator import BackendCommand +from sieval.infer.config import InferHandle +from sieval.infer.deployer import command_endpoint +from sieval.infer.introspect import GPUInfo + +H100 = GPUInfo(model="NVIDIA H100 80GB HBM3", count=8, memory_mib=81559) +QUESTIONS = ["Q1", "Q2", "Q3"] + + +class EchoTask(Task): + """Asks each question once; right when the reply names the question.""" + + model_type = "chat" + + def __init__(self, dataset, model, name=None, strict=False): + super().__init__(dataset=dataset, model=model, name=name) + self._strict = strict + + async def preprocess(self, raw, ctx): + return raw["question"] + + async def infer(self, pre, ctx): + return await self.model.agenerate(pre) + + async def postprocess(self, inf, ctx): + return inf.texts[0].strip() + + async def feedback(self, post, ctx): + expected = f"A{ctx.raw_sample['question'][1:]}" + return True, {"correct": not self._strict and post == expected} + + async def report(self, finals, fails): + correct = sum(1 for f in finals if f.feedback_result["correct"]) + return {"correct": correct, "total": len(finals) + len(fails)} + + +def _sse_answer(answer: str) -> httpx.Response: + base = {"id": "c1", "object": "chat.completion.chunk", "created": 0} + events = [ + { + **base, + "model": "Qwen3-4B", + "choices": [ + { + "index": 0, + "delta": {"role": "assistant", "content": answer}, + "finish_reason": None, + } + ], + }, + { + **base, + "model": "Qwen3-4B", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 2, "completion_tokens": 1, "total_tokens": 3}, + }, + ] + body = "".join(f"data: {json.dumps(e)}\n\n" for e in events) + "data: [DONE]\n\n" + return httpx.Response( + 200, content=body.encode(), headers={"content-type": "text/event-stream"} + ) + + +class Engine: + """The stub engine: launches by returning handles, answers over HTTP.""" + + def __init__(self) -> None: + self.launches: list[str] = [] + self.requests: list[str] = [] + + async def launch( + self, model_name: str, commands: list[BackendCommand], **_: Any + ) -> tuple[list[InferHandle], None]: + self.launches.append(model_name) + handles = [ + InferHandle( + backend=cmd.backend, + handle_id="4242", + endpoint=command_endpoint(cmd), + metadata={"role": cmd.role}, + ) + for cmd in commands + ] + return handles, None + + def __call__(self, request: httpx.Request) -> httpx.Response: + question = json.loads(request.content)["messages"][-1]["content"] + self.requests.append(question) + return _sse_answer(f"A{question[1:]}") + + +@pytest.fixture +def engine(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Iterator[Engine]: + stub = Engine() + real_openai = AsyncOpenAI + + def openai_client(**kwargs: Any) -> AsyncOpenAI: + kwargs["http_client"] = RecordingOpenAIClient( + timeout=kwargs["timeout"], transport=httpx.MockTransport(stub) + ) + return real_openai(**kwargs) + + monkeypatch.setattr(connection_factory, "AsyncOpenAI", openai_client) + monkeypatch.setattr("sieval.cli.infer.lifecycle.HANDLE_DIR", tmp_path / "handles") + with ( + patch("sieval.cli.run.launch_model", new=stub.launch), + patch("sieval.cli.run.cleanup_model", new=AsyncMock()), + ): + yield stub + + +def _gpu(gpu: GPUInfo | None): + """This node's GPUs, as ``nvidia-smi`` would report them.""" + return patch( + "sieval.infer.introspect.detect_local_gpu", new=AsyncMock(return_value=gpu) + ) + + +def _write(tmp_path: Path, questions: list[str], strict: bool = False) -> Path: + checkpoint = tmp_path / "Qwen3-4B" + checkpoint.mkdir(exist_ok=True) + (checkpoint / "config.json").write_text( + json.dumps( + { + "architectures": ["Qwen3ForCausalLM"], + "model_type": "qwen3", + "vocab_size": 151936, + "hidden_size": 2560, + "num_hidden_layers": 36, + "num_attention_heads": 32, + "num_key_value_heads": 8, + "intermediate_size": 9216, + "torch_dtype": "bfloat16", + } + ) + ) + config = tmp_path / "eval.yaml" + config.write_text( + yaml.safe_dump( + { + "result_dir": str(tmp_path / "results"), + "models": {"subject": {"path": str(checkpoint)}}, + "datasets": { + "ds": { + "class": "tests.conftest.MockDataset", + "args": {"samples": [{"question": q} for q in questions]}, + } + }, + "tasks": { + "echo": { + "class": "tests.integration.test_lazy_launch.EchoTask", + "dataset": "ds", + "model": "subject", + "args": {"strict": strict}, + "runner_config": { + "show_progress": False, + "dump_progress": False, + }, + } + }, + } + ) + ) + return config + + +def _meta(tmp_path: Path) -> dict: + return orjson.loads((tmp_path / "results" / "echo" / "meta.json").read_bytes()) + + +async def _run(config: Path, **kwargs: Any) -> dict: + # `resolve_infer_config`'s path, through auto_resolve_plan's detection. + with patch( + "sieval.infer.topology.resolver.detect_local_gpu", + new=AsyncMock(return_value=H100), + ): + return await _run_all(config, **kwargs) + + +class TestLazyLaunch: + @pytest.mark.anyio + async def test_a_resume_with_nothing_to_infer_starts_no_engine( + self, tmp_path: Path, engine: Engine + ) -> None: + config = _write(tmp_path, QUESTIONS[:2]) + with _gpu(H100): + first = await _run(config) + assert engine.launches == ["subject"] + assert sorted(engine.requests) == ["Q1", "Q2"] + identity = _meta(tmp_path)["identity"] + + # Every sample is reused, so this resume runs on a node with no GPU. + engine.launches.clear() + engine.requests.clear() + with _gpu(None): + second = await _run(config, resume=True) + + assert engine.launches == [] + assert engine.requests == [] + assert second == first + meta = _meta(tmp_path) + assert meta["identity"]["inference"] == identity["inference"] + assert [run["start"] for run in meta["runs"]] == ["fresh", "resume"] + assert meta["runs"][-1]["reran"] == [] + assert not (tmp_path / "results" / "logs" / "subject").exists() + + @pytest.mark.anyio + async def test_a_regrade_starts_no_engine( + self, tmp_path: Path, engine: Engine + ) -> None: + with _gpu(H100): + await _run(_write(tmp_path, QUESTIONS[:2])) + identity = _meta(tmp_path)["identity"] + engine.launches.clear() + engine.requests.clear() + + # Another grader: both samples regrade on their recorded inference. + with _gpu(None): + report = await _run( + _write(tmp_path, QUESTIONS[:2], strict=True), resume=True + ) + + assert engine.launches == [] + assert engine.requests == [] + assert report["echo"]["correct"] == 0 + meta = _meta(tmp_path) + assert meta["identity"]["inference"] == identity["inference"] + assert meta["identity"]["feedback"] != identity["feedback"] + # A task argument is read by every grading stage. + assert meta["runs"][-1]["reran"] == ["postprocess", "feedback", "report"] + + @pytest.mark.anyio + async def test_a_resume_with_samples_to_infer_starts_the_engine( + self, tmp_path: Path, engine: Engine + ) -> None: + with _gpu(H100): + await _run(_write(tmp_path, QUESTIONS[:2])) + first = _meta(tmp_path)["identity"]["inference"] + engine.launches.clear() + engine.requests.clear() + report = await _run(_write(tmp_path, QUESTIONS), resume=True) + + assert engine.launches == ["subject"] + assert engine.requests == ["Q3"] + assert (report["echo"]["correct"], report["echo"]["total"]) == (3, 3) + inference = _meta(tmp_path)["identity"]["inference"] + assert inference["digest"] == first["digest"] + + # What a fresh run of all three sends, digest for digest. + fresh = tmp_path / "fresh" + fresh.mkdir() + with _gpu(H100): + await _run(_write(fresh, QUESTIONS)) + assert _meta(fresh)["identity"]["inference"] == inference + + @pytest.mark.anyio + async def test_samples_to_infer_on_a_node_without_gpus_are_refused_first( + self, tmp_path: Path, engine: Engine + ) -> None: + with _gpu(H100): + await _run(_write(tmp_path, QUESTIONS[:2])) + before = (tmp_path / "results" / "echo" / "meta.json").read_bytes() + engine.launches.clear() + + with _gpu(None), pytest.raises(RuntimeError, match="no GPU on this node"): + await _run(_write(tmp_path, QUESTIONS), resume=True) + + assert engine.launches == [] + assert (tmp_path / "results" / "echo" / "meta.json").read_bytes() == before + + @pytest.mark.anyio + async def test_a_fresh_run_is_not_planned( + self, tmp_path: Path, engine: Engine + ) -> None: + planned = AsyncMock() + with _gpu(H100), patch("sieval.cli.run._plan_tasks", new=planned): + await _run(_write(tmp_path, QUESTIONS[:1])) + planned.assert_not_awaited() + assert engine.launches == ["subject"] + + @pytest.mark.anyio + async def test_only_the_model_with_samples_to_infer_starts( + self, tmp_path: Path, engine: Engine + ) -> None: + config = _write(tmp_path, QUESTIONS[:2]) + cfg = yaml.safe_load(config.read_text()) + cfg["models"]["other"] = {"path": cfg["models"]["subject"]["path"]} + cfg["tasks"]["echo_other"] = {**cfg["tasks"]["echo"], "model": "other"} + config.write_text(yaml.safe_dump(cfg)) + with _gpu(H100): + await _run(config) + assert sorted(engine.launches) == ["other", "subject"] + engine.launches.clear() + engine.requests.clear() + + # A third question for `other`'s task only. + datasets = cfg["datasets"] + datasets["ds3"] = { + **datasets["ds"], + "args": {"samples": [{"question": q} for q in QUESTIONS]}, + } + cfg["tasks"]["echo_other"]["dataset"] = "ds3" + config.write_text(yaml.safe_dump(cfg)) + with _gpu(H100): + reports = await _run(config, resume=True) + + assert engine.launches == ["other"] + assert engine.requests == ["Q3"] + assert reports["echo"]["total"] == 2 + assert reports["echo_other"]["total"] == 3 diff --git a/tests/unit/cli/test_cli_model.py b/tests/unit/cli/test_cli_model.py index 4fb4b8c5..b681ddcb 100644 --- a/tests/unit/cli/test_cli_model.py +++ b/tests/unit/cli/test_cli_model.py @@ -300,6 +300,7 @@ async def test_checkpoint_is_launched_with_the_default_plan(self, tmp_path: Path "sieval.cli.run.launch_model", new=AsyncMock(return_value=([], None)) ), patch("sieval.cli.run.cleanup_model", new=AsyncMock()), + patch("sieval.cli.run.check_launch_gpu", new=AsyncMock()), patch( "sieval.cli.session.arun_session", new=AsyncMock(return_value={}), diff --git a/tests/unit/cli/test_run.py b/tests/unit/cli/test_run.py index 416df60b..300b95be 100644 --- a/tests/unit/cli/test_run.py +++ b/tests/unit/cli/test_run.py @@ -93,6 +93,20 @@ def capture(plan: DeploymentPlan) -> list[BackendCommand]: return captured, capture +@pytest.fixture(autouse=True) +def _a_gpu_node(monkeypatch: pytest.MonkeyPatch) -> None: + """Every launch here is faked, so fake the GPUs it would need too. + + A test that means a node without GPUs patches ``detect_local_gpu`` itself. + """ + from sieval.infer.introspect import GPUInfo + + monkeypatch.setattr( + "sieval.infer.introspect.detect_local_gpu", + AsyncMock(return_value=GPUInfo(model="NVIDIA H100", count=8, memory_mib=81559)), + ) + + class TestRunCommand: def test_run_help(self): result = runner.invoke(app, ["run", "--help"]) @@ -1657,7 +1671,11 @@ async def test_a_plan_recorded_with_a_secret_is_resolved_again( self._record(tmp_path, config_path, redact_secrets(recorded)) monkeypatch.setenv("SIEVAL_TARGET_GPU", self.H100) resolve = AsyncMock(return_value=ResolveResult(plan=_fake_plan(), steps=())) - with patch("sieval.cli.run.auto_resolve_plan", new=resolve): + # The config has no task; say one still infers with `m`, so it launches. + with ( + patch("sieval.cli.run.auto_resolve_plan", new=resolve), + patch("sieval.cli.run._models_to_infer", new=AsyncMock(return_value={"m"})), + ): launch, _ = await self._run( config_path, gpu=self._gpu(self.H100), resume=True ) diff --git a/tests/unit/cli/test_session.py b/tests/unit/cli/test_session.py index 3efdcab4..9ec54b6e 100644 --- a/tests/unit/cli/test_session.py +++ b/tests/unit/cli/test_session.py @@ -7440,6 +7440,7 @@ async def test_arun_session_delegates_to_runner(self): resume_from=None, cli_model=None, infer_gpu=None, + unlaunched_deployments=None, ) assert result == {"task_a": {"ok": True}} @@ -7474,6 +7475,7 @@ async def test_arun_session_passes_deterministic(self): resume_from=None, cli_model=None, infer_gpu=None, + unlaunched_deployments=None, ) assert result == {"task_a": {"ok": True}} @@ -7537,6 +7539,7 @@ async def test_arun_session_forwards_legacy_external_endpoint_adapter(self): resume_from=None, cli_model=None, infer_gpu=None, + unlaunched_deployments=None, ) def test_run_session_forwards_legacy_external_endpoint_adapter_positionally( @@ -7687,6 +7690,7 @@ def _make_runner(self, tasks_cfg, models=None, datasets=None) -> EvalSession: runner.runner = MultiTaskRunner() runner._infer_plans = None runner._resume_config_diff = [] + runner._unlaunched_roots = frozenset() runner._task_requirement_contexts = { task_name: _task_requirement_context_for_setup_test( task_cfg, @@ -9060,6 +9064,7 @@ def test_per_task_infer_args_temperature_allowed(self): runner.runner = MultiTaskRunner() runner._infer_plans = None runner._resume_config_diff = [] + runner._unlaunched_roots = frozenset() runner.deterministic = True runner._task_requirement_contexts = { "eval_task": RequirementContext( diff --git a/tests/unit/core/runners/test_plan_run.py b/tests/unit/core/runners/test_plan_run.py new file mode 100644 index 00000000..818e07e4 --- /dev/null +++ b/tests/unit/core/runners/test_plan_run.py @@ -0,0 +1,273 @@ +"""Tests for ``TaskRunner.plan_run``: what a run would do, decided read-only. + +AI-Generated Code - Claude Opus 5.5 (Anthropic) +""" + +import hashlib +from pathlib import Path + +import pytest + +from sieval.core.runners import MultiTaskRunner, StageResumeError +from sieval.core.runners.runner import TaskRunner +from sieval.core.tasks.consts import TaskAction +from sieval.core.tasks.loader import TaskLoader +from tests.conftest import ( + MockChatModel, + MockDataset, + MockSelectiveFailModel, + make_config, +) +from tests.integration.resume.conftest import ( + RESUME_SAMPLES, + CountingMockChatModel, + ResumeTask, +) + +ANSWERS = {"Q1": "A1", "Q2": "A2", "Q3": "A3"} + + +class AgainTask(ResumeTask): + """With *again*, ``feedback`` asks once more after the first answer.""" + + def __init__(self, dataset, model, name=None, again=False): + super().__init__(dataset=dataset, model=model, name=name) + self._again = again + + async def feedback(self, post, ctx): + done, result = await super().feedback(post, ctx) + return done and not (self._again and ctx.iteration == 0), result + + +def _tree(root: Path) -> dict[str, str]: + """Every file under *root* with its content digest.""" + return { + str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in sorted(root.rglob("*")) + if p.is_file() + } + + +async def _record(tmp_path: Path, samples=RESUME_SAMPLES, model=None, **config) -> Path: + result_dir = tmp_path / "recorded" + task = ResumeTask( + dataset=MockDataset(samples), + model=model or MockChatModel(answers=ANSWERS), + name="t", + ) + config = make_config(tmp_path, result_dir=str(result_dir), **config) + await TaskRunner(task, config).arun() + return result_dir + + +def _runner(tmp_path: Path, result_dir: Path, samples=RESUME_SAMPLES, **config): + model = CountingMockChatModel(answers=ANSWERS) + task = ResumeTask(dataset=MockDataset(samples), model=model, name="t") + config.setdefault("resume", True) + return TaskRunner( + task, make_config(tmp_path, result_dir=str(result_dir), **config) + ), model + + +class TestPlanResume: + @pytest.mark.anyio + async def test_a_fresh_run_infers_every_sample(self, tmp_path: Path) -> None: + runner, _ = _runner(tmp_path, tmp_path / "none", resume=False) + plan = await runner.plan_run() + assert plan == { + "start": "fresh", + "recorded": None, + "samples": 3, + "outcomes": None, + "new": 3, + "to_infer": 3, + "to_grade": 3, + } + assert not (tmp_path / "none").exists() + + @pytest.mark.anyio + async def test_a_finished_run_is_reused_whole(self, tmp_path: Path) -> None: + result_dir = await _record(tmp_path) + before = _tree(result_dir) + runner, model = _runner(tmp_path, result_dir) + + plan = await runner.plan_run() + + assert plan["start"] == "reuse" + assert plan["outcomes"] is not None + assert (plan["outcomes"]["kept"], plan["outcomes"]["rereported"]) == (3, 0) + assert (plan["to_infer"], plan["to_grade"]) == (0, 0) + assert model.call_count == 0 + assert _tree(result_dir) == before + + @pytest.mark.anyio + async def test_new_samples_are_counted_to_infer(self, tmp_path: Path) -> None: + result_dir = await _record(tmp_path, samples=RESUME_SAMPLES[:2]) + before = _tree(result_dir) + runner, model = _runner(tmp_path, result_dir) + + plan = await runner.plan_run() + + assert plan["start"] == "resume" + assert plan["outcomes"] == { + "kept": 2, + "rereported": 0, + "regraded": 0, + "repostprocessed": 0, + "restarted": 0, + "dropped": 0, + } + assert (plan["new"], plan["to_infer"], plan["to_grade"]) == (1, 1, 1) + assert model.call_count == 0 + assert _tree(result_dir) == before + + @pytest.mark.anyio + async def test_a_failed_inference_is_counted_to_infer(self, tmp_path: Path) -> None: + failing = MockSelectiveFailModel(fail_samples={"Q3"}, answers=ANSWERS) + result_dir = await _record(tmp_path, model=failing) + before = _tree(result_dir) + runner = TaskRunner( + ResumeTask(dataset=MockDataset(RESUME_SAMPLES), model=failing, name="t"), + make_config(tmp_path, result_dir=str(result_dir), resume=True), + ) + + plan = await runner.plan_run() + + assert (plan["new"], plan["to_infer"], plan["to_grade"]) == (0, 1, 1) + assert _tree(result_dir) == before + + @pytest.mark.anyio + async def test_a_failure_past_max_retries_is_not_counted( + self, tmp_path: Path + ) -> None: + failing = MockSelectiveFailModel(fail_samples={"Q3"}, answers=ANSWERS) + result_dir = await _record(tmp_path, model=failing, max_retries=0) + runner = TaskRunner( + ResumeTask(dataset=MockDataset(RESUME_SAMPLES), model=failing, name="t"), + make_config( + tmp_path, result_dir=str(result_dir), resume=True, max_retries=0 + ), + ) + plan = await runner.plan_run() + assert (plan["to_infer"], plan["to_grade"]) == (0, 0) + + @pytest.mark.anyio + async def test_resume_from_another_dir_reads_it_and_copies_nothing( + self, tmp_path: Path + ) -> None: + recorded = await _record(tmp_path, samples=RESUME_SAMPLES[:2]) + before = _tree(recorded) + target = tmp_path / "new" + runner, _ = _runner(tmp_path, target, resume=recorded) + + plan = await runner.plan_run() + + assert plan["start"] == "resume" + assert plan["recorded"] == str(recorded) + assert plan["outcomes"]["kept"] == 2 + assert plan["to_infer"] == 1 + assert _tree(recorded) == before + assert not target.exists() + + @pytest.mark.anyio + async def test_a_resume_that_would_be_refused_raises(self, tmp_path: Path) -> None: + result_dir = await _record(tmp_path, samples=RESUME_SAMPLES[:2]) + task = ResumeTask( + dataset=MockDataset(RESUME_SAMPLES), + # Other request arguments: another inference identity. + model=MockChatModel(answers=ANSWERS).with_args(temperature=0.9), + name="t", + ) + runner = TaskRunner( + task, make_config(tmp_path, result_dir=str(result_dir), resume=True) + ) + with pytest.raises(StageResumeError, match="inference"): + await runner.plan_run() + + @pytest.mark.anyio + async def test_the_multi_runner_maps_a_refusal_to_its_task( + self, tmp_path: Path + ) -> None: + result_dir = await _record(tmp_path, samples=RESUME_SAMPLES[:2]) + multi = MultiTaskRunner() + multi.add_task( + ResumeTask( + dataset=MockDataset(RESUME_SAMPLES), + model=MockChatModel(answers=ANSWERS).with_args(temperature=0.9), + name="t", + ), + make_config(tmp_path, result_dir=str(result_dir), resume=True), + ) + plans = await multi.plan_run() + assert isinstance(plans["t"], StageResumeError) + + +class TestUnservedCandidate: + @pytest.mark.anyio + async def test_a_resume_with_nothing_to_infer_runs(self, tmp_path: Path) -> None: + result_dir = await _record(tmp_path) + (result_dir / "report.json").unlink() # finalize through the full path + model = CountingMockChatModel(answers=ANSWERS) + runner = TaskRunner( + ResumeTask(dataset=MockDataset(RESUME_SAMPLES), model=model, name="t"), + make_config(tmp_path, result_dir=str(result_dir), resume=True), + unserved_candidate=True, + ) + report = await runner.arun() + assert report["completed"] == 3 + assert model.call_count == 0 + + @pytest.mark.anyio + async def test_a_resume_with_samples_to_infer_is_refused_unrecorded( + self, tmp_path: Path + ) -> None: + result_dir = await _record(tmp_path, samples=RESUME_SAMPLES[:2]) + before = _tree(result_dir) + model = CountingMockChatModel(answers=ANSWERS) + runner = TaskRunner( + ResumeTask(dataset=MockDataset(RESUME_SAMPLES), model=model, name="t"), + make_config(tmp_path, result_dir=str(result_dir), resume=True), + unserved_candidate=True, + ) + with pytest.raises(RuntimeError, match="1 sample"): + await runner.arun() + assert model.call_count == 0 + assert _tree(result_dir) == before + + @pytest.mark.anyio + async def test_a_regrade_that_sends_a_sample_back_fails_it_at_infer( + self, tmp_path: Path + ) -> None: + # The plan saw nothing to infer; the new grader's feedback iterates. + result_dir = tmp_path / "recorded" + answers = AgainTask( + dataset=MockDataset(RESUME_SAMPLES), model=MockChatModel(ANSWERS), name="t" + ) + await TaskRunner( + answers, make_config(tmp_path, result_dir=str(result_dir)) + ).arun() + + def resume(model, **runner): + task = AgainTask( + dataset=MockDataset(RESUME_SAMPLES), model=model, name="t", again=True + ) + config = make_config(tmp_path, result_dir=str(result_dir), resume=True) + return TaskRunner(task, config, **runner) + + model = CountingMockChatModel(answers=ANSWERS) + unserved = resume(model, unserved_candidate=True) + assert (await unserved.plan_run())["to_infer"] == 0 + report = await resume(model, unserved_candidate=True).arun() + + # Failed by the guard at infer, before any request: not a dead endpoint. + assert model.call_count == 0 + assert (report["completed"], report["failed"]) == (0, 3) + recorded = await TaskLoader(answers, result_dir).load_initial_state() + assert { + (c.iteration, c.error_action, c.error_reason) for c in recorded.values() + } == {(1, TaskAction.INFER, "exception::RuntimeError")} + + # A resume with the engine up retries them. + report = await resume(model).arun() + assert model.call_count == 3 + assert (report["completed"], report["failed"]) == (3, 0)