From 9ce29e0c63e124f98b34b59c194f085caff799e3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=85=B1=E7=89=9B=E8=82=89?= Date: Sun, 9 Aug 2026 18:30:54 +0800 Subject: [PATCH] feat(meta-reme): adopt single-execution validation with fail-fast option - Remove retry/attempt model: QueryScheduler failures are now terminal instead of requeued; QueryLease no longer carries an attempt number; SandboxSpec.max_retries defaults to 0 and is deprecated - Flatten workspace layout: drop the attempt-/ directory level so case results live directly under cases// - Add fail-fast: new --fail-fast CLI flag, validation.fail_fast config key, and ValidationFailFastError that cancels sibling workers and writes a run-level failure.json on the first construction or query error - Clarify module boundaries: meta-reme/run.py prepares workspace and runs initial validation; meta-reme/validation/run.py validates an already prepared workspace - Update design.md, validation README, and both benchmark configs - Update tests for single-execution semantics and add fail-fast tests --- benchmark/beam/config_meta_reme.yaml | 4 + benchmark/longmemeval/config_meta_reme.yaml | 1 + meta-reme/design.md | 23 +- meta-reme/models.py | 6 +- meta-reme/run.py | 19 +- meta-reme/validation/README.md | 33 ++- meta-reme/validation/__init__.py | 4 +- meta-reme/validation/__main__.py | 8 +- meta-reme/validation/evaluator.py | 244 ++++++++++-------- meta-reme/validation/run.py | 17 +- meta-reme/validation/scheduler.py | 30 +-- meta-reme/workspace.py | 9 +- .../test_meta_reme_models_workspace.py | 4 +- tests/unit_sandbox/test_meta_reme_run.py | 11 +- .../unit_sandbox/test_meta_reme_validation.py | 196 +++++++++++--- 15 files changed, 405 insertions(+), 204 deletions(-) diff --git a/benchmark/beam/config_meta_reme.yaml b/benchmark/beam/config_meta_reme.yaml index 78290e9f..f3206415 100644 --- a/benchmark/beam/config_meta_reme.yaml +++ b/benchmark/beam/config_meta_reme.yaml @@ -6,3 +6,7 @@ dataset: source: "benchmark/beam/dataset/BEAM" variant: "1M" # 100K | 500K | 1M | 10M train_case_ids: [] # empty = all cases; or ["1", "2", "3"] + +validation: + concurrency: 1 + fail_fast: false # true stops all cases after the first construction or query error diff --git a/benchmark/longmemeval/config_meta_reme.yaml b/benchmark/longmemeval/config_meta_reme.yaml index 83e61804..c1530a56 100644 --- a/benchmark/longmemeval/config_meta_reme.yaml +++ b/benchmark/longmemeval/config_meta_reme.yaml @@ -8,3 +8,4 @@ dataset: validation: concurrency: 5 + fail_fast: false # true stops all cases after the first construction or query error diff --git a/meta-reme/design.md b/meta-reme/design.md index 6846d8ca..efdaeb29 100644 --- a/meta-reme/design.md +++ b/meta-reme/design.md @@ -83,18 +83,17 @@ workspace/ summary.json cases/ / - attempt-/ - case_result.json - memory_construction/ + case_result.json + memory_construction/ + result.json + reme_workspace.tar.gz + reme_workspace/ + build.log + queries/ + result.json + / + answer.log result.json - reme_workspace.tar.gz - reme_workspace/ - build.log - queries/ - result.json - / - answer.log - result.json logs/ ``` @@ -203,7 +202,7 @@ Agent 可以自主选择完整 case、指定 session、session prefix 或少量 Sandbox 在 memory construction 结束、query 开始前，先导出 construction 结果、`build.log` 和当时完整、可直接通过 `upload_workspace()` 回灌的 `reme_workspace.tar.gz`；query 阶段结束后再单独导出 query summary、逐 query 日志和结构化结果。两个阶段不得合并成一个仅在所有 query 结束后生成的归档。失败时应 best-effort 导出 `failure.json`、stdout、stderr、Python traceback、action log、已完成阶段和宿主异常。每次运行的宿主目录为： ```text -evaluations///cases//attempt-/ +evaluations///cases// ``` 其中 `memory_construction/` 保存结构化结果、build 日志和可复用的 workspace 压缩包；`queries/` 保存结构化结果及安全解压后的逐 query artifacts。Query 传输使用的临时压缩包必须在解压后删除。解压过程需要拒绝绝对路径、`../`、symlink、设备文件和异常大的压缩包；`reme_workspace.tar.gz` 保存 SHA256，确保可验证、可复用。 diff --git a/meta-reme/models.py b/meta-reme/models.py index 8829eb74..9f2b35c0 100644 --- a/meta-reme/models.py +++ b/meta-reme/models.py @@ -143,7 +143,11 @@ class SandboxSpec(ContractModel): image: str = Field(min_length=1) timeout_seconds: float = Field(gt=0) concurrency: int = Field(default=1, ge=1) - max_retries: int = Field(default=2, ge=0) + max_retries: int = Field( + default=0, + ge=0, + description="Deprecated compatibility field; validation operations are not retried.", + ) max_artifact_bytes: int = Field(default=1_073_741_824, gt=0) diff --git a/meta-reme/run.py b/meta-reme/run.py index 0555cbf3..2984dcab 100644 --- a/meta-reme/run.py +++ b/meta-reme/run.py @@ -1,4 +1,14 @@ -"""Initialize a Meta-ReMe workspace from a benchmark YAML configuration.""" +"""Prepare a Meta-ReMe workspace and run its initial validation. + +This is the end-to-end Meta-ReMe entry point: it reads the benchmark YAML, +installs the selected dataset cases, builds the initial candidate repository, +and then calls :func:`validation.run_validation` for the ``init`` branch. + +It is distinct from ``meta-reme/validation/run.py``, whose CLI only validates +an already prepared workspace and does not install data or create candidate +code. The validation scheduling implementation itself lives in +``validation/evaluator.py``. +""" from __future__ import annotations @@ -101,6 +111,7 @@ def run_initial_validation( workspace: Workspace, concurrency: int, *, + fail_fast: bool = False, validation_runner: ValidationRunner = run_validation, ) -> Path: """Validate the initial code branch once against every installed case.""" @@ -125,6 +136,7 @@ def run_initial_validation( INITIAL_CODE_ID, concurrency, validation_id=INITIAL_VALIDATION_ID, + fail_fast=fail_fast, ) @@ -146,6 +158,7 @@ def prepare_and_validate_workspace( validation = run_initial_validation( workspace, config["validation_concurrency"], + fail_fast=config["validation_fail_fast"], validation_runner=validation_runner, ) return workspace, validation @@ -186,6 +199,9 @@ def load_config(config_path: Path) -> dict[str, Any]: or validation_concurrency < 1 ): raise ValueError("validation.concurrency must be a positive integer") + validation_fail_fast = validation_config.get("fail_fast", False) + if not isinstance(validation_fail_fast, bool): + raise ValueError("validation.fail_fast must be a boolean") return { "meta_workspace": _project_path(raw.get("meta_workspace"), "meta_workspace"), "dataset": dataset, @@ -193,6 +209,7 @@ def load_config(config_path: Path) -> dict[str, Any]: "dataset_variant": dataset_config.get("variant"), "dataset_source": source, "validation_concurrency": validation_concurrency, + "validation_fail_fast": validation_fail_fast, } diff --git a/meta-reme/validation/README.md b/meta-reme/validation/README.md index a148198a..15196497 100644 --- a/meta-reme/validation/README.md +++ b/meta-reme/validation/README.md @@ -3,8 +3,10 @@ Running `meta-reme/run.py` prepares the configured dataset and initial bundle, then automatically validates the `init` branch against every installed case. The first result is stored under `evaluations/init/initial/`; a completed result -is reused on later invocations. Configure the full-case concurrency with -`validation.concurrency` in `config_meta_reme.yaml`. +is reused on later invocations. Configure concurrency with +`validation.concurrency` in `config_meta_reme.yaml`. Set +`validation.fail_fast: true` to abort the run on its first case or query error; +the default keeps other cases running. Validate prepared workspace cases against an immutable code revision: @@ -14,7 +16,8 @@ python meta-reme/validation/run.py \ --case-id case-1 \ --case-id case-2 \ --code-id \ - --concurrency 2 + --concurrency 2 \ + --fail-fast ``` `--case-id` may be repeated. The code ID is the name of a local Git branch; commit @@ -28,12 +31,20 @@ Results are written without overwriting earlier runs: /evaluations/// ``` -Each case attempt stores the two execution phases independently. Memory -construction is exported before queries begin, so its workspace is an exact +`--fail-fast` is optional. When enabled, a construction failure, query +answer/judge error, or query infrastructure error is persisted before sibling +workers are cancelled and their containers are closed. The run raises an error +and writes a run-level `failure.json`; it does not publish a completed +`summary.json`. Without this flag, errors remain isolated to their cases. + +Each case stores the two execution phases independently. Validation executes +each construction and query only once; retry policy belongs to the sandbox +implementation. Memory construction is exported before queries begin, so its +workspace is an exact post-construction snapshot rather than the workspace left after evaluation: ```text -cases//attempt-/ +cases// case_result.json memory_construction/ result.json @@ -57,7 +68,7 @@ archive only for transfer and safe extraction; the archive and its redundant Validation uses a strict two-phase schedule. Reusable workers first construct all case memories. A successful construction is exported as the case's immutable query snapshot; a candidate construction failure is recorded and its -queries are skipped. Only after every construction attempt reaches a terminal +queries are skipped. Only after every construction reaches a terminal state do workers begin answering queries. Query cases have a preferred owner, while individual queries use fenced leases @@ -69,8 +80,10 @@ runs, so the exported snapshot hash remains an audit field rather than part of the cache identity. This keeps workspace uploads uncommon without making a long case an indivisible scheduling unit. Results are stored in the original case and query slots, so completion order does not change summaries. Infrastructure -failures retire the affected container and retry only the failed stage; -structured construction or answer/judge failures remain candidate results. +failures are persisted as terminal validation results. The affected container +is retired so it cannot contaminate later independent work, but the failed +construction or query is not requeued. Structured construction or answer/judge +failures likewise remain candidate results. Containers are named by the factory rather than by their current case and call `reset_case()` before switching snapshots, retaining only the installed @@ -82,7 +95,7 @@ from regular code: ```python from validation import run_validation -result_dir = run_validation(workspace, case_ids, code_id, concurrency=2) +result_dir = run_validation(workspace, case_ids, code_id, concurrency=2, fail_fast=True) ``` Applications that already run an event loop should use the native async entry diff --git a/meta-reme/validation/__init__.py b/meta-reme/validation/__init__.py index e9db6f65..a13b293f 100644 --- a/meta-reme/validation/__init__.py +++ b/meta-reme/validation/__init__.py @@ -1,5 +1,5 @@ """Run reproducible Meta-ReMe validations in isolated case sandboxes.""" -from .evaluator import ValidationError, run_validation, run_validation_async +from .evaluator import ValidationError, ValidationFailFastError, run_validation, run_validation_async -__all__ = ["ValidationError", "run_validation", "run_validation_async"] +__all__ = ["ValidationError", "ValidationFailFastError", "run_validation", "run_validation_async"] diff --git a/meta-reme/validation/__main__.py b/meta-reme/validation/__main__.py index 5debaff9..87dbc4cc 100644 --- a/meta-reme/validation/__main__.py +++ b/meta-reme/validation/__main__.py @@ -1,4 +1,10 @@ -"""Allow ``python -m validation`` to invoke the validation CLI.""" +"""Forward ``python -m validation`` to ``validation.run.main``. + +This module contains no scheduling or validation logic. The CLI adapter is in +``validation/run.py`` and the implementation is in ``validation/evaluator.py``. +The separate top-level ``meta-reme/run.py`` prepares a complete workspace +before invoking that same validation implementation. +""" from .run import main diff --git a/meta-reme/validation/evaluator.py b/meta-reme/validation/evaluator.py index b993a9a0..2ac008f7 100644 --- a/meta-reme/validation/evaluator.py +++ b/meta-reme/validation/evaluator.py @@ -47,6 +47,10 @@ class ValidationError(RuntimeError): """Raised when validation inputs or workspace state are invalid.""" +class ValidationFailFastError(ValidationError): + """Raised after fail-fast validation persists the first observed case error.""" + + def run_validation( workspace_root: Path, case_ids: list[str], @@ -54,6 +58,7 @@ def run_validation( concurrency: int, *, validation_id: str | None = None, + fail_fast: bool = False, factory_builder: FactoryBuilder = DockerReMeSandboxFactory, environment: dict[str, str] | None = None, ) -> Path: @@ -66,6 +71,7 @@ def run_validation( code_id, concurrency, validation_id=validation_id, + fail_fast=fail_fast, factory_builder=factory_builder, environment=environment, ), @@ -79,6 +85,7 @@ async def run_validation_async( concurrency: int, *, validation_id: str | None = None, + fail_fast: bool = False, factory_builder: FactoryBuilder = DockerReMeSandboxFactory, environment: dict[str, str] | None = None, ) -> Path: @@ -86,6 +93,8 @@ async def run_validation_async( if concurrency < 1: raise ValidationError("concurrency must be at least 1") + if not isinstance(fail_fast, bool): + raise ValidationError("fail_fast must be a boolean") if not case_ids: raise ValidationError("at least one case_id is required") if len(case_ids) != len(set(case_ids)): @@ -113,6 +122,7 @@ async def run_validation_async( commit_sha, run_id, concurrency, + fail_fast, factory_builder, environment, snapshot, @@ -128,6 +138,7 @@ async def _execute_validation( commit_sha: str, run_id: str, concurrency: int, + fail_fast: bool, factory_builder: FactoryBuilder, environment: dict[str, str] | None, snapshot: SourceSnapshot, @@ -154,6 +165,7 @@ async def _execute_validation( "case_ids": [case.case_id for case in cases], "concurrency": concurrency, "container_reuse": True, + "fail_fast": fail_fast, "scheduling": { "construction_barrier": True, "query_unit": "single_query_lease", @@ -170,7 +182,22 @@ async def _execute_validation( config=f"{domain.bundle_target}.yaml", command_timeout=domain.sandbox.timeout_seconds, ) - results = await _run_cases(workspace, run_root, factory, domain, cases, concurrency) + try: + results = await _run_cases(workspace, run_root, factory, domain, cases, concurrency, fail_fast=fail_fast) + except ValidationFailFastError as exc: + workspace.atomic_write_json( + run_root.relative_to(workspace.root) / "failure.json", + { + "schema_version": 1, + "validation_id": run_id, + "status": "aborted", + "fail_fast": True, + "error": f"{type(exc).__name__}: {exc}", + "traceback": traceback.format_exc(), + "aborted_at": utc_now().isoformat(), + }, + ) + raise summary = _summarize(run_id, code_id, commit_sha, [case.case_id for case in cases], results, fingerprints) workspace.atomic_write_json(run_root.relative_to(workspace.root) / "summary.json", summary) return run_root @@ -183,6 +210,8 @@ async def _run_cases( domain: DomainSpec, cases: list[CaseSpec], concurrency: int, + *, + fail_fast: bool, ) -> list[dict[str, Any]]: """Run strict construction and query phases with reusable worker sandboxes.""" @@ -199,7 +228,7 @@ async def _run_cases( index, case = queue.get_nowait() except asyncio.QueueEmpty: return - result, plan = await _construct_case_with_retries( + result, plan = await _construct_case_once( workspace, run_root, factory, @@ -210,22 +239,34 @@ async def _run_cases( ) if result is not None: results[index] = result + if fail_fast and result["status"] != "completed": + raise ValidationFailFastError( + f"case {case.case_id!r} failed during construction: {result.get('error')}", + ) if plan is not None: plans.append(plan) try: - construction_outcomes = await asyncio.gather( - *(run_construction_worker(worker) for worker in workers), - return_exceptions=True, + await _run_worker_group( + [run_construction_worker(worker) for worker in workers], + fail_fast=fail_fast, ) - _raise_worker_failure(construction_outcomes) - scheduler = QueryScheduler(plans, max_retries=domain.sandbox.max_retries) - query_outcomes = await asyncio.gather( - *(_run_query_worker(workspace, factory, domain, scheduler, worker) for worker in workers), - return_exceptions=True, + scheduler = QueryScheduler(plans) + await _run_worker_group( + [ + _run_query_worker( + workspace, + factory, + domain, + scheduler, + worker, + fail_fast=fail_fast, + ) + for worker in workers + ], + fail_fast=fail_fast, ) - _raise_worker_failure(query_outcomes) for plan in plans: context = plan.context @@ -255,20 +296,33 @@ class _ConstructedCase: """Artifacts needed to run and finalize one successfully built case.""" case_spec: CaseSpec - attempt_number: int - attempt_root: Path + case_root: Path build: dict[str, Any] memory_workspace: Path memory_sha256: str -def _raise_worker_failure(outcomes: list[Any]) -> None: - failure = next((outcome for outcome in outcomes if isinstance(outcome, BaseException)), None) - if failure is not None: - raise failure +async def _run_worker_group(workers: list[Any], *, fail_fast: bool) -> None: + """Run one phase, cancelling sibling workers when fail-fast raises.""" + + tasks = [asyncio.create_task(worker) for worker in workers] + if not fail_fast: + outcomes = await asyncio.gather(*tasks, return_exceptions=True) + failure = next((outcome for outcome in outcomes if isinstance(outcome, BaseException)), None) + if failure is not None: + raise failure + return + try: + await asyncio.gather(*tasks) + except BaseException: + for task in tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise -async def _construct_case_with_retries( +async def _construct_case_once( workspace: Workspace, run_root: Path, factory: DockerReMeSandboxFactory, @@ -277,49 +331,44 @@ async def _construct_case_with_retries( case: CaseSpec, worker: _WorkerState, ) -> tuple[dict[str, Any] | None, QueryCasePlan | None]: - last_result: dict[str, Any] | None = None - for attempt_number in range(1, domain.sandbox.max_retries + 2): - result, context = await _construct_case( - workspace, - run_root, - factory, - domain, - case, - attempt_number, - worker, + """Construct a case once and prepare its query plan when successful.""" + + result, context = await _construct_case( + workspace, + run_root, + factory, + domain, + case, + worker, + ) + if context is not None: + try: + queries = tuple(evaluation_queries(domain.dataset.name, case)) + except Exception as exc: + result = { + "case_id": case.case_id, + "status": "infra_error", + "stage": "schedule_queries", + "completed_stages": ["prepare", "construct_memory"], + "build": context.build, + "queries": [], + "error": f"{type(exc).__name__}: {exc}", + "traceback": traceback.format_exc(), + "container_id": getattr(worker.sandbox_case, "container_id", None), + "artifact_sha256": {"memory_workspace": context.memory_sha256}, + } + workspace.atomic_write_json(context.case_root.relative_to(workspace.root) / "failure.json", result) + return _publish_case_result(workspace, context.case_root, result), None + return None, QueryCasePlan( + case_index=case_index, + case_id=case.case_id, + queries=queries, + context=context, ) - if context is not None: - try: - queries = tuple(evaluation_queries(domain.dataset.name, case)) - except Exception as exc: - result = { - "case_id": case.case_id, - "attempt": context.attempt_number, - "status": "infra_error", - "stage": "schedule_queries", - "completed_stages": ["prepare", "construct_memory"], - "build": context.build, - "queries": [], - "error": f"{type(exc).__name__}: {exc}", - "traceback": traceback.format_exc(), - "container_id": getattr(worker.sandbox_case, "container_id", None), - "artifact_sha256": {"memory_workspace": context.memory_sha256}, - } - workspace.atomic_write_json(context.attempt_root.relative_to(workspace.root) / "failure.json", result) - return _publish_case_result(workspace, context.attempt_root, result), None - return None, QueryCasePlan( - case_index=case_index, - case_id=case.case_id, - queries=queries, - context=context, - ) - assert result is not None - last_result = result - if result["status"] != "infra_error": - return result, None + assert result is not None + if result["status"] == "infra_error": await _retire_worker_sandbox(worker) - assert last_result is not None - return last_result, None + return result, None async def _construct_case( @@ -328,12 +377,10 @@ async def _construct_case( factory: DockerReMeSandboxFactory, domain: DomainSpec, case_spec: CaseSpec, - attempt_number: int, worker: _WorkerState, ) -> tuple[dict[str, Any] | None, _ConstructedCase | None]: case_root = workspace.entity_path((run_root / "cases").relative_to(workspace.root), case_spec.case_id) - attempt_root = workspace.entity_path(case_root.relative_to(workspace.root), f"attempt-{attempt_number}") - attempt_root.mkdir(parents=True, exist_ok=False) + case_root.mkdir(parents=True, exist_ok=False) completed_stages: list[str] = [] build: dict[str, Any] | None = None try: @@ -345,7 +392,7 @@ async def _construct_case( completed_stages.append("prepare") build = await worker.sandbox_case.run_build(build_jobs(domain.dataset.name, case_spec)) completed_stages.append("construct_memory") - memory_root = attempt_root / "memory_construction" + memory_root = case_root / "memory_construction" memory_root.mkdir(parents=False, exist_ok=False) workspace.atomic_write_json(memory_root.relative_to(workspace.root) / "result.json", build) memory_workspace = await worker.sandbox_case.export_workspace(memory_root / "reme_workspace.tar.gz") @@ -359,7 +406,6 @@ async def _construct_case( if not build.get("success"): result = { "case_id": case_spec.case_id, - "attempt": attempt_number, "status": "candidate_failure", "completed_stages": completed_stages, "build": build, @@ -369,24 +415,22 @@ async def _construct_case( "memory_workspace": memory_sha256, }, } - await _best_effort_export(worker.sandbox_case, attempt_root / "full.tar.gz") - return _publish_case_result(workspace, attempt_root, result), None + await _best_effort_export(worker.sandbox_case, case_root / "full.tar.gz") + return _publish_case_result(workspace, case_root, result), None worker.loaded_case_id = case_spec.case_id return None, _ConstructedCase( case_spec=case_spec, - attempt_number=attempt_number, - attempt_root=attempt_root, + case_root=case_root, build=build, memory_workspace=memory_workspace, memory_sha256=memory_sha256, ) except Exception as exc: # A failed container/worker must not cancel sibling cases. if worker.sandbox_case is not None: - await _best_effort_export(worker.sandbox_case, attempt_root / "full.tar.gz") + await _best_effort_export(worker.sandbox_case, case_root / "full.tar.gz") result = { "case_id": case_spec.case_id, - "attempt": attempt_number, "status": "infra_error", "stage": "construction", "completed_stages": completed_stages, @@ -396,8 +440,8 @@ async def _construct_case( "traceback": traceback.format_exc(), "container_id": getattr(worker.sandbox_case, "container_id", None), } - workspace.atomic_write_json(attempt_root.relative_to(workspace.root) / "failure.json", result) - return _publish_case_result(workspace, attempt_root, result), None + workspace.atomic_write_json(case_root.relative_to(workspace.root) / "failure.json", result) + return _publish_case_result(workspace, case_root, result), None async def _run_query_worker( @@ -406,6 +450,8 @@ async def _run_query_worker( domain: DomainSpec, scheduler: QueryScheduler, worker: _WorkerState, + *, + fail_fast: bool, ) -> None: while True: lease = await scheduler.claim(worker.worker_id, worker.loaded_case_id) @@ -419,7 +465,13 @@ async def _run_query_worker( result = await worker.sandbox_case.run_query(lease.query) await _export_query_artifacts(workspace, domain, worker.sandbox_case, context, query_id, lease) await scheduler.complete(lease, result) - except Exception as exc: # Retry only infrastructure failures around the structured query operation. + if fail_fast and result.get("error") is not None: + raise ValidationFailFastError( + f"case {lease.plan.case_id!r} query {query_id!r} failed: {result['error']}", + ) + except ValidationFailFastError: + raise + except Exception as exc: # Infrastructure failures are terminal at the validation layer. failure = { "query_id": query_id, "question": getattr(lease.query, "question", None), @@ -431,16 +483,19 @@ async def _run_query_worker( "error": f"{type(exc).__name__}: {exc}", "traceback": traceback.format_exc(), "infrastructure_error": True, - "attempt": lease.attempt, "lease_token": lease.token, "worker_id": worker.worker_id, "container_id": getattr(worker.sandbox_case, "container_id", None), } - query_attempt_root = _write_query_failure(workspace, context, query_id, lease.attempt, failure) + query_failure_root = _write_query_failure(workspace, context, query_id, failure) if worker.sandbox_case is not None: - await _best_effort_export(worker.sandbox_case, query_attempt_root / "full.tar.gz") + await _best_effort_export(worker.sandbox_case, query_failure_root / "full.tar.gz") await scheduler.fail(lease, failure) await _retire_worker_sandbox(worker) + if fail_fast: + raise ValidationFailFastError( + f"case {lease.plan.case_id!r} query {query_id!r} had an infrastructure error: {exc}", + ) from exc async def _load_query_workspace( @@ -468,7 +523,7 @@ async def _export_query_artifacts( query_id: str, lease: QueryLease, ) -> None: - queries_root = context.attempt_root / "queries" + queries_root = context.case_root / "queries" queries_root.mkdir(parents=True, exist_ok=True) archive = queries_root / f".{query_id}.{lease.token}.tar.gz" try: @@ -492,29 +547,17 @@ async def _export_query_artifacts( archive.unlink(missing_ok=True) -def _query_attempt_root( - workspace: Workspace, - context: _ConstructedCase, - query_id: str, - attempt: int, -) -> Path: - attempts_root = context.attempt_root / "queries" / ".attempts" - query_root = workspace.entity_path(attempts_root.relative_to(workspace.root), query_id) - attempt_root = workspace.entity_path(query_root.relative_to(workspace.root), f"attempt-{attempt}") - attempt_root.mkdir(parents=True, exist_ok=True) - return attempt_root - - def _write_query_failure( workspace: Workspace, context: _ConstructedCase, query_id: str, - attempt: int, failure: dict[str, Any], ) -> Path: - attempt_root = _query_attempt_root(workspace, context, query_id, attempt) - workspace.atomic_write_json(attempt_root.relative_to(workspace.root) / "failure.json", failure) - return attempt_root + queries_root = context.case_root / "queries" + query_root = workspace.entity_path(queries_root.relative_to(workspace.root), query_id) + query_root.mkdir(parents=True, exist_ok=True) + workspace.atomic_write_json(query_root.relative_to(workspace.root) / "failure.json", failure) + return query_root def _finalize_constructed_case( @@ -546,25 +589,24 @@ def _finalize_constructed_case( "summary": query_summary, "queries": query_results, } - queries_root = context.attempt_root / "queries" + queries_root = context.case_root / "queries" queries_root.mkdir(parents=True, exist_ok=True) workspace.atomic_write_json(queries_root.relative_to(workspace.root) / "result.json", query_payload) has_infra_error = any(result.get("infrastructure_error") for result in query_results) result = { "case_id": plan.case_id, - "attempt": context.attempt_number, "status": "infra_error" if has_infra_error else "completed", "completed_stages": ["prepare", "construct_memory", "test", *([] if has_infra_error else ["export"])], "build": context.build, "queries": query_results, "query_summary": query_summary, - "error": "one or more query infrastructure attempts were exhausted" if has_infra_error else None, + "error": "one or more queries had an infrastructure failure" if has_infra_error else None, "artifact_sha256": {"memory_workspace": context.memory_sha256}, } if has_infra_error: - workspace.atomic_write_json(context.attempt_root.relative_to(workspace.root) / "failure.json", result) - return _publish_case_result(workspace, context.attempt_root, result) + workspace.atomic_write_json(context.case_root.relative_to(workspace.root) / "failure.json", result) + return _publish_case_result(workspace, context.case_root, result) async def _retire_worker_sandbox(worker: _WorkerState) -> None: @@ -588,8 +630,8 @@ async def _best_effort_close(sandbox_case: Any) -> None: return -def _publish_case_result(workspace: Workspace, attempt_root: Path, result: dict[str, Any]) -> dict[str, Any]: - workspace.atomic_write_json(attempt_root.relative_to(workspace.root) / "case_result.json", result) +def _publish_case_result(workspace: Workspace, case_root: Path, result: dict[str, Any]) -> dict[str, Any]: + workspace.atomic_write_json(case_root.relative_to(workspace.root) / "case_result.json", result) return result diff --git a/meta-reme/validation/run.py b/meta-reme/validation/run.py index 40884ba4..30938c32 100644 --- a/meta-reme/validation/run.py +++ b/meta-reme/validation/run.py @@ -1,4 +1,13 @@ -"""CLI entry point for validating prepared Meta-ReMe cases.""" +"""CLI entry point for validating an already prepared Meta-ReMe workspace. + +Unlike the top-level ``meta-reme/run.py`` workflow, this module does not +prepare a dataset or create the initial candidate repository. It only parses +explicit workspace, case, and code-branch arguments, then delegates execution +to ``validation.evaluator.run_validation``. + +``validation/__main__.py`` forwards ``python -m validation`` here; it is an +alternate invocation form, not another validation implementation. +""" # Import roots must be installed before importing the hyphenated meta-reme tree. # pylint: disable=wrong-import-position @@ -33,6 +42,11 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--code-id", required=True, help="Path-safe local Git branch name identifying the code") parser.add_argument("--concurrency", type=int, required=True, help="Maximum concurrent case sandboxes") parser.add_argument("--validation-id", help="Optional non-overwriting validation run ID") + parser.add_argument( + "--fail-fast", + action="store_true", + help="Stop and cancel all remaining cases after the first construction or query error", + ) return parser.parse_args() @@ -47,6 +61,7 @@ def main() -> None: args.code_id, args.concurrency, validation_id=args.validation_id, + fail_fast=args.fail_fast, ) except Exception as exc: raise SystemExit(f"validation failed: {exc}") from exc diff --git a/meta-reme/validation/scheduler.py b/meta-reme/validation/scheduler.py index be6c9fef..e90c85a0 100644 --- a/meta-reme/validation/scheduler.py +++ b/meta-reme/validation/scheduler.py @@ -19,15 +19,12 @@ class QueryCasePlan: context: Any = None owner_worker_id: int | None = None results: list[dict[str, Any] | None] = field(init=False) - infrastructure_failures: list[dict[str, Any]] = field(default_factory=list) _pending: deque[int] = field(init=False, repr=False) - _attempts: list[int] = field(init=False, repr=False) _leases: dict[int, str] = field(default_factory=dict, init=False, repr=False) def __post_init__(self) -> None: self.results = [None] * len(self.queries) self._pending = deque(range(len(self.queries))) - self._attempts = [0] * len(self.queries) @property def remaining_count(self) -> int: @@ -38,13 +35,12 @@ class QueryCasePlan: @dataclass(frozen=True) class QueryLease: - """A fenced claim for one query execution attempt.""" + """A fenced claim for one query execution.""" token: str worker_id: int plan: QueryCasePlan query_index: int - attempt: int selection: str @property @@ -62,14 +58,11 @@ class QueryScheduler: remaining individual queries, which is what makes work stealing possible. """ - def __init__(self, plans: list[QueryCasePlan], *, max_retries: int) -> None: - if max_retries < 0: - raise ValueError("max_retries must not be negative") + def __init__(self, plans: list[QueryCasePlan]) -> None: indexes = [plan.case_index for plan in plans] if len(indexes) != len(set(indexes)): raise ValueError("query case indexes must be unique") self.plans = sorted(plans, key=lambda plan: plan.case_index) - self.max_retries = max_retries self._condition = asyncio.Condition() async def claim(self, worker_id: int, loaded_case_id: str | None) -> QueryLease | None: @@ -80,7 +73,6 @@ class QueryScheduler: plan, selection = self._select_plan(worker_id, loaded_case_id) if plan is not None: query_index = plan._pending.popleft() # pylint: disable=protected-access - plan._attempts[query_index] += 1 # pylint: disable=protected-access token = uuid4().hex plan._leases[query_index] = token # pylint: disable=protected-access if plan.owner_worker_id is None: @@ -90,7 +82,6 @@ class QueryScheduler: worker_id=worker_id, plan=plan, query_index=query_index, - attempt=plan._attempts[query_index], # pylint: disable=protected-access selection=selection, ) if self._is_complete(): @@ -105,24 +96,13 @@ class QueryScheduler: lease.plan.results[lease.query_index] = result self._condition.notify_all() - async def fail(self, lease: QueryLease, failure: dict[str, Any]) -> bool: - """Record an infrastructure failure and requeue when retries remain. - - Returns ``True`` when the query was requeued and ``False`` when the - supplied failure became its terminal result. - """ + async def fail(self, lease: QueryLease, failure: dict[str, Any]) -> None: + """Publish an infrastructure failure as the query's terminal result.""" async with self._condition: self._consume_lease(lease) - lease.plan.infrastructure_failures.append(failure) - if lease.attempt <= self.max_retries: - lease.plan._pending.appendleft(lease.query_index) # pylint: disable=protected-access - requeued = True - else: - lease.plan.results[lease.query_index] = failure - requeued = False + lease.plan.results[lease.query_index] = failure self._condition.notify_all() - return requeued def _select_plan( self, diff --git a/meta-reme/workspace.py b/meta-reme/workspace.py index 3f81eae3..d75feebb 100644 --- a/meta-reme/workspace.py +++ b/meta-reme/workspace.py @@ -122,7 +122,7 @@ class Workspace: return resolved def entity_path(self, base: str | Path, identifier: str) -> Path: - """Resolve a path for an externally supplied code/case/attempt ID.""" + """Resolve a path for an externally supplied entity ID.""" if not identifier or identifier in {".", ".."} or Path(identifier).name != identifier: raise WorkspaceError(f"Unsafe workspace identifier: {identifier!r}") @@ -130,19 +130,18 @@ class Workspace: raise WorkspaceError(f"Unsafe workspace identifier: {identifier!r}") return self.path(Path(base) / identifier) - def case_attempt_dir( + def validation_case_dir( self, code_id: str, validation_id: str, case_id: str, - attempt_id: str, *, create: bool = False, ) -> Path: - """Return the canonical attempt directory described by the design.""" + """Return the canonical validation result directory for one case.""" current = self.entity_path("evaluations", code_id) - for identifier in (validation_id, "cases", case_id, attempt_id): + for identifier in (validation_id, "cases", case_id): current = self.entity_path(current.relative_to(self.root), identifier) if create: current.mkdir(parents=True, exist_ok=False) diff --git a/tests/unit_sandbox/test_meta_reme_models_workspace.py b/tests/unit_sandbox/test_meta_reme_models_workspace.py index 9ee49a31..804dc360 100644 --- a/tests/unit_sandbox/test_meta_reme_models_workspace.py +++ b/tests/unit_sandbox/test_meta_reme_models_workspace.py @@ -132,8 +132,8 @@ def test_workspace_rejects_escaping_paths_and_ids(tmp_path: Path) -> None: with pytest.raises(workspace_module.WorkspaceError): workspace.entity_path("evaluations", "../case") - attempt = workspace.case_attempt_dir("init", "validation-1", "case-1", "attempt-1", create=True) - assert attempt == workspace.path("evaluations/init/validation-1/cases/case-1/attempt-1") + case_result = workspace.validation_case_dir("init", "validation-1", "case-1", create=True) + assert case_result == workspace.path("evaluations/init/validation-1/cases/case-1") def test_atomic_write_replaces_content_and_leaves_no_temporary_file(tmp_path: Path) -> None: diff --git a/tests/unit_sandbox/test_meta_reme_run.py b/tests/unit_sandbox/test_meta_reme_run.py index 148f4eb1..6e4bd44d 100644 --- a/tests/unit_sandbox/test_meta_reme_run.py +++ b/tests/unit_sandbox/test_meta_reme_run.py @@ -159,7 +159,7 @@ def test_load_config_reads_all_preparation_arguments(tmp_path: Path) -> None: "variant": "100K", "train_case_ids": [1, "2"], }, - "validation": {"concurrency": 5}, + "validation": {"concurrency": 5, "fail_fast": True}, }, ), encoding="utf-8", @@ -172,6 +172,7 @@ def test_load_config_reads_all_preparation_arguments(tmp_path: Path) -> None: assert config["dataset_variant"] == "100K" assert config["train_case_ids"] == ["1", "2"] assert config["validation_concurrency"] == 5 + assert config["validation_fail_fast"] is True def test_prepare_and_validate_workspace_uses_all_installed_cases(tmp_path: Path) -> None: @@ -188,15 +189,15 @@ def test_prepare_and_validate_workspace_uses_all_installed_cases(tmp_path: Path) "source": str(source), "train_case_ids": [], }, - "validation": {"concurrency": 5}, + "validation": {"concurrency": 5, "fail_fast": True}, }, ), encoding="utf-8", ) calls = [] - def fake_validation(workspace, case_ids, code_id, concurrency, *, validation_id): - calls.append((Path(workspace), case_ids, code_id, concurrency, validation_id)) + def fake_validation(workspace, case_ids, code_id, concurrency, *, validation_id, fail_fast): + calls.append((Path(workspace), case_ids, code_id, concurrency, validation_id, fail_fast)) output = Path(workspace) / f"evaluations/{code_id}/{validation_id}" output.mkdir(parents=True) (output / "summary.json").write_text("{}\n", encoding="utf-8") @@ -205,7 +206,7 @@ def test_prepare_and_validate_workspace_uses_all_installed_cases(tmp_path: Path) workspace, validation = run.prepare_and_validate_workspace(config_path, validation_runner=fake_validation) assert validation == workspace.path("evaluations/init/initial") - assert calls == [(workspace.root, ["case-2", "case-1"], "init", 5, "initial")] + assert calls == [(workspace.root, ["case-2", "case-1"], "init", 5, "initial", True)] _, repeated_validation = run.prepare_and_validate_workspace(config_path, validation_runner=fake_validation) diff --git a/tests/unit_sandbox/test_meta_reme_validation.py b/tests/unit_sandbox/test_meta_reme_validation.py index b88bde1c..80cfaa05 100644 --- a/tests/unit_sandbox/test_meta_reme_validation.py +++ b/tests/unit_sandbox/test_meta_reme_validation.py @@ -265,26 +265,26 @@ def test_validation_uses_immutable_commit_and_bounds_full_case_lifecycle(tmp_pat assert summary["status"] == "completed" assert summary["mean_query_score"] == 1.0 assert summary["case_count"] == 3 - attempt = output / "cases/case-1/attempt-1" - assert (attempt / "memory_construction/result.json").is_file() - assert (attempt / "memory_construction/build.log").is_file() - memory_workspace = attempt / "memory_construction/reme_workspace.tar.gz" + case_output = output / "cases/case-1" + assert (case_output / "memory_construction/result.json").is_file() + assert (case_output / "memory_construction/build.log").is_file() + memory_workspace = case_output / "memory_construction/reme_workspace.tar.gz" assert memory_workspace.is_file() with tarfile.open(memory_workspace, mode="r:gz") as archive: assert archive.getnames() == ["memory.json"] - assert (attempt / "memory_construction/reme_workspace/memory.json").is_file() - case_result = json.loads((attempt / "case_result.json").read_text(encoding="utf-8")) + assert (case_output / "memory_construction/reme_workspace/memory.json").is_file() + case_result = json.loads((case_output / "case_result.json").read_text(encoding="utf-8")) assert case_result["artifact_sha256"] == { "memory_workspace": evaluator._sha256_file(memory_workspace), # pylint: disable=protected-access } - assert (attempt / "queries/result.json").is_file() - assert (attempt / "queries/query-case-1/answer.log").is_file() - assert (attempt / "queries/query-case-1/result.json").is_file() - assert not (attempt / "queries/summary.json").exists() - assert not (attempt / "queries/artifacts").exists() - assert {path.name for path in (attempt / "queries").iterdir()} == {"result.json", "query-case-1"} - assert not list((attempt / "queries").glob("*.tar.gz")) - assert not list((attempt / "queries").glob(".*.tar.gz")) + assert (case_output / "queries/result.json").is_file() + assert (case_output / "queries/query-case-1/answer.log").is_file() + assert (case_output / "queries/query-case-1/result.json").is_file() + assert not (case_output / "queries/summary.json").exists() + assert not (case_output / "queries/artifacts").exists() + assert {path.name for path in (case_output / "queries").iterdir()} == {"result.json", "query-case-1"} + assert not list((case_output / "queries").glob("*.tar.gz")) + assert not list((case_output / "queries").glob(".*.tar.gz")) def test_validation_builds_every_memory_before_stealing_ordered_queries(tmp_path: Path) -> None: @@ -324,8 +324,8 @@ def test_validation_builds_every_memory_before_stealing_ordered_queries(tmp_path ] -def test_validation_replaces_an_infra_error_container_before_retrying(tmp_path: Path) -> None: - """A suspect pool member is destroyed; its replacement remains reusable.""" +def test_validation_records_construction_infra_error_without_retrying(tmp_path: Path) -> None: + """A failed construction is terminal while later cases continue on a fresh container.""" class RetryCase(_FakeCase): """Fail the first build to emulate a broken container.""" @@ -370,12 +370,12 @@ def test_validation_replaces_an_infra_error_container_before_retrying(tmp_path: factory = created[0] summary = json.loads((output / "summary.json").read_text(encoding="utf-8")) - assert factory.created == ["case-1", "case-1"] - assert factory.resets == [("case-1", "case-2"), ("case-2", "case-1")] + assert factory.created == ["case-1", "case-2"] + assert factory.resets == [] assert factory.active == 0 - assert [result["status"] for result in summary["cases"]] == ["completed", "completed"] - assert (output / "cases/case-1/attempt-1/failure.json").is_file() - assert (output / "cases/case-1/attempt-2/case_result.json").is_file() + assert [result["status"] for result in summary["cases"]] == ["infra_error", "completed"] + assert (output / "cases/case-1/failure.json").is_file() + assert not list((output / "cases/case-1").glob("attempt-*")) def test_validation_discards_a_container_when_reset_fails(tmp_path: Path) -> None: @@ -467,11 +467,11 @@ def test_construction_candidate_failure_skips_case_queries(tmp_path: Path) -> No summary = json.loads((output / "summary.json").read_text(encoding="utf-8")) assert summary["cases"][0]["status"] == "candidate_failure" assert not any(event.startswith("run_query:") for _, event in created[0].events) - assert not (output / "cases/case-1/attempt-1/queries").exists() + assert not (output / "cases/case-1/queries").exists() -def test_query_infrastructure_failure_is_saved_and_retried_independently(tmp_path: Path) -> None: - """A failed lease retires its container without rebuilding case memory.""" +def test_query_infrastructure_failure_is_saved_without_retrying(tmp_path: Path) -> None: + """A failed query is terminal and its container is retired.""" class FlakyQueryCase(_FakeCase): """Raise one infrastructure exception from the query primitive.""" @@ -515,16 +515,16 @@ def test_query_infrastructure_failure_is_saved_and_retried_independently(tmp_pat ) summary = json.loads((output / "summary.json").read_text(encoding="utf-8")) - failure = output / "cases/case-1/attempt-1/queries/.attempts/query-case-1/attempt-1/failure.json" - assert summary["cases"][0]["status"] == "completed" - assert summary["cases"][0]["queries"][0]["score"] == 1.0 - assert created[0].created == ["case-1", "case-1"] + failure = output / "cases/case-1/queries/query-case-1/failure.json" + assert summary["cases"][0]["status"] == "infra_error" + assert summary["cases"][0]["queries"][0]["score"] is None + assert created[0].created == ["case-1"] assert failure.is_file() assert json.loads(failure.read_text(encoding="utf-8"))["infrastructure_error"] is True -def test_exhausted_query_failure_is_a_terminal_ordered_case_result(tmp_path: Path) -> None: - """An exhausted lease cannot strand waiting workers or disappear from aggregation.""" +def test_query_failure_is_a_terminal_ordered_case_result(tmp_path: Path) -> None: + """A failed lease cannot strand waiting workers or disappear from aggregation.""" class BrokenQueryCase(_FakeCase): """Always lose the query worker response.""" @@ -559,7 +559,131 @@ def test_exhausted_query_failure_is_a_terminal_ordered_case_result(tmp_path: Pat assert case_result["status"] == "infra_error" assert [query["query_id"] for query in case_result["queries"]] == ["query-case-1"] assert case_result["queries"][0]["infrastructure_error"] is True - assert (output / "cases/case-1/attempt-1/failure.json").is_file() + assert (output / "cases/case-1/failure.json").is_file() + assert (output / "cases/case-1/queries/query-case-1/failure.json").is_file() + + +def test_fail_fast_construction_error_cancels_sibling_cases(tmp_path: Path) -> None: + """The first terminal construction error cancels in-progress siblings.""" + + class FailFastBuildCase(_FakeCase): + """Fail one build after allowing its sibling to start.""" + + async def run_build(self, jobs): + if self.case_id == "case-fail": + await asyncio.sleep(0.02) + return {"success": False, "jobs": []} + self.factory.sibling_started = True + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + self.factory.sibling_cancelled = True + raise + return await super().run_build(jobs) + + class FailFastBuildFactory(_FakeFactory): + """Track cancellation of the non-failing construction sandbox.""" + + def __init__(self, candidate, **kwargs): + super().__init__(candidate, **kwargs) + self.sibling_started = False + self.sibling_cancelled = False + + async def create_case(self, case_id): + self.active += 1 + self.max_active = max(self.max_active, self.active) + self.created.append(case_id) + return FailFastBuildCase(self, case_id) + + workspace, repository, commit = _prepared_workspace(tmp_path, ["case-fail", "case-slow"]) + subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True) + created = [] + + def build_factory(*args, **kwargs): + factory = FailFastBuildFactory(*args, **kwargs) + created.append(factory) + return factory + + with pytest.raises(evaluator.ValidationFailFastError, match="case-fail"): + evaluator.run_validation( + workspace.root, + ["case-fail", "case-slow"], + "candidate-1", + 2, + validation_id="fail-fast-build", + fail_fast=True, + factory_builder=build_factory, + environment={}, + ) + + output = workspace.path("evaluations/candidate-1/fail-fast-build") + failure = json.loads((output / "failure.json").read_text(encoding="utf-8")) + assert created[0].sibling_started is True + assert created[0].sibling_cancelled is True + assert created[0].active == 0 + assert failure["status"] == "aborted" + assert failure["fail_fast"] is True + assert not (output / "summary.json").exists() + + +def test_fail_fast_query_error_does_not_run_later_queries(tmp_path: Path) -> None: + """A structured answer/judge error stops later leases without reclassifying it as infrastructure.""" + + class FailedQueryCase(_FakeCase): + """Return one normal structured query failure.""" + + async def run_query(self, query): + self.factory.events.append((self.case_id, f"run_query:{query.query_id}")) + return { + "query_id": query.query_id, + "question": query.question, + "golden_answer": query.golden_answer, + "answer": None, + "score": None, + "answer_result": {"success": False}, + "judge_result": None, + "error": "answer job failed", + } + + class FailedQueryFactory(_FakeFactory): + """Create cases whose first query returns a structured error.""" + + async def create_case(self, case_id): + self.active += 1 + self.max_active = max(self.max_active, self.active) + self.created.append(case_id) + return FailedQueryCase(self, case_id) + + workspace, repository, commit = _prepared_workspace(tmp_path, ["case-1"]) + workspace.atomic_write_json("dataset/cases/000000.json", _case("case-1", query_count=2)) + subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True) + created = [] + + def build_factory(*args, **kwargs): + factory = FailedQueryFactory(*args, **kwargs) + created.append(factory) + return factory + + with pytest.raises(evaluator.ValidationFailFastError, match="answer job failed"): + evaluator.run_validation( + workspace.root, + ["case-1"], + "candidate-1", + 1, + validation_id="fail-fast-query", + fail_fast=True, + factory_builder=build_factory, + environment={}, + ) + + output = workspace.path("evaluations/candidate-1/fail-fast-query") + assert [event for _, event in created[0].events if event.startswith("run_query:")] == [ + "run_query:query-case-1-0", + ] + assert (output / "cases/case-1/queries/query-case-1-0/result.json").is_file() + assert not (output / "cases/case-1/queries/query-case-1-1").exists() + assert (output / "failure.json").is_file() + assert not (output / "summary.json").exists() @pytest.mark.asyncio @@ -635,10 +759,10 @@ def test_adapters_build_dataset_specific_requests() -> None: @pytest.mark.asyncio -async def test_query_scheduler_prefers_loaded_memory_and_fences_retries() -> None: +async def test_query_scheduler_prefers_loaded_memory_and_terminally_records_failures() -> None: first = scheduler_module.QueryCasePlan(0, "case-1", ("q1", "q2")) second = scheduler_module.QueryCasePlan(1, "case-2", ("q3",)) - scheduler = scheduler_module.QueryScheduler([first, second], max_retries=1) + scheduler = scheduler_module.QueryScheduler([first, second]) owner = await scheduler.claim(0, first.case_id) thief = await scheduler.claim(1, first.case_id) @@ -647,14 +771,10 @@ async def test_query_scheduler_prefers_loaded_memory_and_fences_retries() -> Non assert thief.plan is first assert thief.selection == "affinity_steal" - assert await scheduler.fail(owner, {"query_id": "q1", "error": "infra"}) is True - retry = await scheduler.claim(0, first.case_id) - assert retry.query_index == owner.query_index - assert retry.attempt == 2 + await scheduler.fail(owner, {"query_id": "q1", "error": "infra"}) with pytest.raises(RuntimeError, match="stale"): await scheduler.complete(owner, {"query_id": "q1"}) - await scheduler.complete(retry, {"query_id": "q1", "score": 1.0}) await scheduler.complete(thief, {"query_id": "q2", "score": 1.0}) remaining = await scheduler.claim(1, first.case_id) assert remaining.plan is second