mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-11-11 08:38:48 +00:00
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-<n>/ directory level so case results live directly under cases/<case_id>/ - 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
This commit is contained in:
酱牛肉
parent
6d93a24328
commit
9ce29e0c63
15 changed files with 405 additions and 204 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -8,3 +8,4 @@ dataset:
|
|||
|
||||
validation:
|
||||
concurrency: 5
|
||||
fail_fast: false # true stops all cases after the first construction or query error
|
||||
|
|
|
|||
|
|
@ -83,18 +83,17 @@ workspace/
|
|||
summary.json
|
||||
cases/
|
||||
<case_id>/
|
||||
attempt-<n>/
|
||||
case_result.json
|
||||
memory_construction/
|
||||
case_result.json
|
||||
memory_construction/
|
||||
result.json
|
||||
reme_workspace.tar.gz
|
||||
reme_workspace/
|
||||
build.log
|
||||
queries/
|
||||
result.json
|
||||
<query_id>/
|
||||
answer.log
|
||||
result.json
|
||||
reme_workspace.tar.gz
|
||||
reme_workspace/
|
||||
build.log
|
||||
queries/
|
||||
result.json
|
||||
<query_id>/
|
||||
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/<code_id>/<validation_id>/cases/<case>/attempt-<n>/
|
||||
evaluations/<code_id>/<validation_id>/cases/<case>/
|
||||
```
|
||||
|
||||
其中 `memory_construction/` 保存结构化结果、build 日志和可复用的 workspace 压缩包；`queries/` 保存结构化结果及安全解压后的逐 query artifacts。Query 传输使用的临时压缩包必须在解压后删除。解压过程需要拒绝绝对路径、`../`、symlink、设备文件和异常大的压缩包；`reme_workspace.tar.gz` 保存 SHA256，确保可验证、可复用。
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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 <branch-name> \
|
||||
--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:
|
|||
<workspace>/evaluations/<code-id>/<validation-id>/
|
||||
```
|
||||
|
||||
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/<case-id>/attempt-<n>/
|
||||
cases/<case-id>/
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue