From 6d93a24328464787873b0a5826e03de2c9b73c86 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=85=B1=E7=89=9B=E8=82=89?= Date: Sat, 8 Aug 2026 23:44:43 +0800 Subject: [PATCH] Implement two-phase validation scheduling with reusable workers --- meta-reme/validation/README.md | 24 +- meta-reme/validation/evaluator.py | 401 +++++++++++++++--- meta-reme/validation/scheduler.py | 165 +++++++ sandbox/docker_sandbox.py | 71 +++- sandbox/worker.py | 16 +- .../unit_sandbox/test_meta_reme_validation.py | 361 +++++++++++++++- tests/unit_sandbox/test_sandbox.py | 57 ++- 7 files changed, 1014 insertions(+), 81 deletions(-) create mode 100644 meta-reme/validation/scheduler.py diff --git a/meta-reme/validation/README.md b/meta-reme/validation/README.md index 0d082949..a148198a 100644 --- a/meta-reme/validation/README.md +++ b/meta-reme/validation/README.md @@ -54,9 +54,27 @@ as `reme_workspace/` for direct inspection. Query artifacts use a temporary archive only for transfer and safe extraction; the archive and its redundant `summary.json` are deleted immediately afterward. -Each case uses a fresh sandbox. The concurrency limit covers the complete case -lifecycle: sandbox creation, memory construction, queries, artifact export, and -shutdown. +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 +state do workers begin answering queries. + +Query cases have a preferred owner, while individual queries use fenced leases +so idle workers can steal remaining work. A worker first continues the exact +`case_id` already loaded in its container, then prefers its owned case, an +unowned case, and finally the case with the largest remaining query tail. Case +IDs are unique within one validation and containers do not cross validation +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. + +Containers are named by the factory rather than by their current case and call +`reset_case()` before switching snapshots, retaining only the installed +candidate and harness. All containers are shut down after validation finishes. The same workflow is available as a project API. Use the synchronous entry point from regular code: diff --git a/meta-reme/validation/evaluator.py b/meta-reme/validation/evaluator.py index 4241e3f7..b993a9a0 100644 --- a/meta-reme/validation/evaluator.py +++ b/meta-reme/validation/evaluator.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio from contextlib import contextmanager +from dataclasses import dataclass import hashlib import json import os @@ -13,6 +14,7 @@ import stat import subprocess import tarfile import tempfile +import traceback from typing import Any, Callable, Iterator from uuid import uuid4 @@ -23,6 +25,7 @@ from workspace import Workspace from sandbox import DockerReMeSandboxFactory, SourceCandidate, SourceSnapshot from .adapters import build_jobs, evaluation_queries +from .scheduler import QueryCasePlan, QueryLease, QueryScheduler SANDBOX_ENV_NAMES = ( "LLM_API_KEY", @@ -150,6 +153,12 @@ async def _execute_validation( "commit_sha": commit_sha, "case_ids": [case.case_id for case in cases], "concurrency": concurrency, + "container_reuse": True, + "scheduling": { + "construction_barrier": True, + "query_unit": "single_query_lease", + "workspace_affinity": "case_id", + }, "dataset": domain.dataset.name.value, "fingerprints": fingerprints, "started_at": utc_now().isoformat(), @@ -175,60 +184,178 @@ async def _run_cases( cases: list[CaseSpec], concurrency: int, ) -> list[dict[str, Any]]: - semaphore = asyncio.Semaphore(concurrency) + """Run strict construction and query phases with reusable worker sandboxes.""" - async def run(case: CaseSpec) -> dict[str, Any]: - async with semaphore: - return await _run_case_with_retries(workspace, run_root, factory, domain, case) + queue: asyncio.Queue[tuple[int, CaseSpec]] = asyncio.Queue() + for index, case in enumerate(cases): + queue.put_nowait((index, case)) + results: list[dict[str, Any] | None] = [None] * len(cases) + plans: list[QueryCasePlan] = [] + workers = [_WorkerState(worker_id=index) for index in range(concurrency)] - return list(await asyncio.gather(*(run(case) for case in cases))) + async def run_construction_worker(worker: _WorkerState) -> None: + while True: + try: + index, case = queue.get_nowait() + except asyncio.QueueEmpty: + return + result, plan = await _construct_case_with_retries( + workspace, + run_root, + factory, + domain, + index, + case, + worker, + ) + if result is not None: + results[index] = result + 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, + ) + _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, + ) + _raise_worker_failure(query_outcomes) + + for plan in plans: + context = plan.context + assert isinstance(context, _ConstructedCase) + results[plan.case_index] = _finalize_constructed_case(workspace, plan, context) + finally: + await asyncio.gather( + *(_best_effort_close(worker.sandbox_case) for worker in workers if worker.sandbox_case is not None), + ) + + if any(result is None for result in results): + raise RuntimeError("validation worker exited without publishing every case result") + return [result for result in results if result is not None] -async def _run_case_with_retries( +@dataclass +class _WorkerState: + """Host-side state retained by one reusable sandbox worker.""" + + worker_id: int + sandbox_case: Any | None = None + loaded_case_id: str | None = None + + +@dataclass(frozen=True) +class _ConstructedCase: + """Artifacts needed to run and finalize one successfully built case.""" + + case_spec: CaseSpec + attempt_number: int + attempt_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 _construct_case_with_retries( workspace: Workspace, run_root: Path, factory: DockerReMeSandboxFactory, domain: DomainSpec, + case_index: int, case: CaseSpec, -) -> dict[str, Any]: + 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 = await _run_case(workspace, run_root, factory, domain, case, attempt_number) + result, context = await _construct_case( + workspace, + run_root, + factory, + domain, + case, + attempt_number, + 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, + "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 + return result, None + await _retire_worker_sandbox(worker) assert last_result is not None - return last_result + return last_result, None -async def _run_case( +async def _construct_case( workspace: Workspace, run_root: Path, factory: DockerReMeSandboxFactory, domain: DomainSpec, case_spec: CaseSpec, attempt_number: int, -) -> dict[str, Any]: + 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) - sandbox_case = None completed_stages: list[str] = [] + build: dict[str, Any] | None = None try: - sandbox_case = await factory.create_case(case_spec.case_id) + if worker.sandbox_case is None: + worker.sandbox_case = await factory.create_case(case_spec.case_id) + else: + await worker.sandbox_case.reset_case(case_spec.case_id) + worker.loaded_case_id = None completed_stages.append("prepare") - build = await sandbox_case.run_build(build_jobs(domain.dataset.name, case_spec)) + 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.mkdir(parents=False, exist_ok=False) workspace.atomic_write_json(memory_root.relative_to(workspace.root) / "result.json", build) - memory_workspace = await sandbox_case.export_workspace(memory_root / "reme_workspace.tar.gz") + memory_workspace = await worker.sandbox_case.export_workspace(memory_root / "reme_workspace.tar.gz") _extract_artifacts( memory_workspace, memory_root / "reme_workspace", domain.sandbox.max_artifact_bytes, ) - await sandbox_case.export_build_log(memory_root / "build.log") + await worker.sandbox_case.export_build_log(memory_root / "build.log") + memory_sha256 = _sha256_file(memory_workspace) if not build.get("success"): result = { "case_id": case_spec.case_id, @@ -239,61 +366,212 @@ async def _run_case( "queries": [], "error": "memory construction failed", "artifact_sha256": { - "memory_workspace": _sha256_file(memory_workspace), + "memory_workspace": memory_sha256, }, } - await _best_effort_export(sandbox_case, attempt_root / "full.tar.gz") - return _publish_case_result(workspace, attempt_root, result) + await _best_effort_export(worker.sandbox_case, attempt_root / "full.tar.gz") + return _publish_case_result(workspace, attempt_root, result), None - queries = await sandbox_case.run_queries(evaluation_queries(domain.dataset.name, case_spec)) - completed_stages.append("test") - queries_root = attempt_root / "queries" - queries_root.mkdir(parents=False, exist_ok=False) - workspace.atomic_write_json(queries_root.relative_to(workspace.root) / "result.json", queries) - queries_archive = queries_root / ".artifacts.tar.gz" - try: - await sandbox_case.export_queries(queries_archive) - _extract_artifacts( - queries_archive, - queries_root, - domain.sandbox.max_artifact_bytes, - strip_prefix=PurePosixPath("queries"), - excluded_paths={PurePosixPath("summary.json")}, - ) - finally: - queries_archive.unlink(missing_ok=True) - completed_stages.append("export") - result = { - "case_id": case_spec.case_id, - "attempt": attempt_number, - "status": "completed", - "completed_stages": completed_stages, - "build": build, - "queries": queries.get("queries", []), - "query_summary": queries.get("summary", {}), - "error": None, - "artifact_sha256": { - "memory_workspace": _sha256_file(memory_workspace), - }, - } - return _publish_case_result(workspace, attempt_root, result) + worker.loaded_case_id = case_spec.case_id + return None, _ConstructedCase( + case_spec=case_spec, + attempt_number=attempt_number, + attempt_root=attempt_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 sandbox_case is not None: - await _best_effort_export(sandbox_case, attempt_root / "full.tar.gz") + if worker.sandbox_case is not None: + await _best_effort_export(worker.sandbox_case, attempt_root / "full.tar.gz") result = { "case_id": case_spec.case_id, "attempt": attempt_number, "status": "infra_error", + "stage": "construction", "completed_stages": completed_stages, - "build": None, + "build": build, "queries": [], "error": f"{type(exc).__name__}: {exc}", + "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) + return _publish_case_result(workspace, attempt_root, result), None + + +async def _run_query_worker( + workspace: Workspace, + factory: DockerReMeSandboxFactory, + domain: DomainSpec, + scheduler: QueryScheduler, + worker: _WorkerState, +) -> None: + while True: + lease = await scheduler.claim(worker.worker_id, worker.loaded_case_id) + if lease is None: + return + context = lease.plan.context + assert isinstance(context, _ConstructedCase) + query_id = str(getattr(lease.query, "query_id", "")) + try: + await _load_query_workspace(factory, worker, lease.plan, context) + 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. + failure = { + "query_id": query_id, + "question": getattr(lease.query, "question", None), + "golden_answer": getattr(lease.query, "golden_answer", None), + "answer": None, + "score": None, + "answer_result": None, + "judge_result": None, + "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) + if worker.sandbox_case is not None: + await _best_effort_export(worker.sandbox_case, query_attempt_root / "full.tar.gz") + await scheduler.fail(lease, failure) + await _retire_worker_sandbox(worker) + + +async def _load_query_workspace( + factory: DockerReMeSandboxFactory, + worker: _WorkerState, + plan: QueryCasePlan, + context: _ConstructedCase, +) -> None: + if worker.sandbox_case is not None and worker.loaded_case_id == plan.case_id: + return + if worker.sandbox_case is None: + worker.sandbox_case = await factory.create_case(plan.case_id) + else: + await worker.sandbox_case.reset_case(plan.case_id) + worker.loaded_case_id = None + await worker.sandbox_case.upload_workspace(context.memory_workspace) + worker.loaded_case_id = plan.case_id + + +async def _export_query_artifacts( + workspace: Workspace, + domain: DomainSpec, + sandbox_case: Any, + context: _ConstructedCase, + query_id: str, + lease: QueryLease, +) -> None: + queries_root = context.attempt_root / "queries" + queries_root.mkdir(parents=True, exist_ok=True) + archive = queries_root / f".{query_id}.{lease.token}.tar.gz" + try: + await sandbox_case.export_query(query_id, archive) + with tempfile.TemporaryDirectory(prefix=".query-export-", dir=queries_root) as temporary: + temporary_root = Path(temporary) + _extract_artifacts( + archive, + temporary_root, + domain.sandbox.max_artifact_bytes, + strip_prefix=PurePosixPath("queries"), + ) + source = workspace.entity_path(temporary_root.relative_to(workspace.root), query_id) + target = workspace.entity_path(queries_root.relative_to(workspace.root), query_id) + if not source.is_dir(): + raise ValidationError(f"query archive is missing artifact directory: {query_id!r}") + if target.exists(): + raise ValidationError(f"query artifact directory already exists: {query_id!r}") + source.replace(target) finally: - if sandbox_case is not None: - await sandbox_case.close() + 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 + + +def _finalize_constructed_case( + workspace: Workspace, + plan: QueryCasePlan, + context: _ConstructedCase, +) -> dict[str, Any]: + if any(result is None for result in plan.results): + raise RuntimeError(f"case {plan.case_id!r} has unpublished query results") + query_results = [result for result in plan.results if result is not None] + scores = [float(result["score"]) for result in query_results if result.get("score") is not None] + query_summary = { + "schema_version": 1, + "case_id": plan.case_id, + "query_count": len(query_results), + "scored_count": len(scores), + "mean_score": sum(scores) / len(scores) if scores else None, + "queries": [ + { + "query_id": result.get("query_id"), + "score": result.get("score"), + **({"error": result["error"]} if result.get("error") is not None else {}), + } + for result in query_results + ], + } + query_payload = { + "success": all(result.get("error") is None for result in query_results), + "summary": query_summary, + "queries": query_results, + } + queries_root = context.attempt_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, + "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) + + +async def _retire_worker_sandbox(worker: _WorkerState) -> None: + if worker.sandbox_case is not None: + await _best_effort_close(worker.sandbox_case) + worker.sandbox_case = None + worker.loaded_case_id = None async def _best_effort_export(sandbox_case: Any, destination: Path) -> None: @@ -303,6 +581,13 @@ async def _best_effort_export(sandbox_case: Any, destination: Path) -> None: return +async def _best_effort_close(sandbox_case: Any) -> None: + try: + await sandbox_case.close() + except Exception: + 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) return result diff --git a/meta-reme/validation/scheduler.py b/meta-reme/validation/scheduler.py new file mode 100644 index 00000000..be6c9fef --- /dev/null +++ b/meta-reme/validation/scheduler.py @@ -0,0 +1,165 @@ +"""Deterministic query leasing for two-phase validation runs.""" + +from __future__ import annotations + +import asyncio +from collections import deque +from dataclasses import dataclass, field +from typing import Any +from uuid import uuid4 + + +@dataclass +class QueryCasePlan: + """One constructed case whose queries can be executed independently.""" + + case_index: int + case_id: str + queries: tuple[Any, ...] + 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: + """Count queries that have not yet reached a terminal result.""" + + return len(self._pending) + len(self._leases) + + +@dataclass(frozen=True) +class QueryLease: + """A fenced claim for one query execution attempt.""" + + token: str + worker_id: int + plan: QueryCasePlan + query_index: int + attempt: int + selection: str + + @property + def query(self) -> Any: + """Return the query payload covered by this lease.""" + + return self.plan.queries[self.query_index] + + +class QueryScheduler: + """Assign query leases with workspace affinity and deterministic stealing. + + A case is initially owned as a whole by one worker, but ownership is an + affinity hint rather than an exclusive lock. Idle workers may lease the + 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") + 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: + """Wait for and return the best query for one worker, or ``None`` when done.""" + + async with self._condition: + while True: + 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: + plan.owner_worker_id = worker_id + return QueryLease( + token=token, + 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(): + return None + await self._condition.wait() + + async def complete(self, lease: QueryLease, result: dict[str, Any]) -> None: + """Publish a terminal query result if the lease is still current.""" + + async with self._condition: + self._consume_lease(lease) + 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 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 + self._condition.notify_all() + return requeued + + def _select_plan( + self, + worker_id: int, + loaded_case_id: str | None, + ) -> tuple[QueryCasePlan | None, str]: + available = [plan for plan in self.plans if plan._pending] # pylint: disable=protected-access + if not available: + return None, "" + + if loaded_case_id is not None: + matching = [plan for plan in available if plan.case_id == loaded_case_id] + if matching: + owner = matching[0].owner_worker_id + return matching[0], "affinity" if owner in (None, worker_id) else "affinity_steal" + + owned = [plan for plan in available if plan.owner_worker_id == worker_id] + if owned: + return self._most_remaining(owned), "owned" + + unowned = [plan for plan in available if plan.owner_worker_id is None] + if unowned: + return self._most_remaining(unowned), "unowned" + + return self._most_remaining(available), "steal" + + @staticmethod + def _most_remaining(plans: list[QueryCasePlan]) -> QueryCasePlan: + """Prefer the largest pending tail, with stable input order as a tie-breaker.""" + + return min(plans, key=lambda plan: (-len(plan._pending), plan.case_index)) # pylint: disable=protected-access + + def _consume_lease(self, lease: QueryLease) -> None: + current = lease.plan._leases.get(lease.query_index) # pylint: disable=protected-access + if current != lease.token: + raise RuntimeError("stale or unknown query lease") + del lease.plan._leases[lease.query_index] # pylint: disable=protected-access + + def _is_complete(self) -> bool: + return all(not plan._pending and not plan._leases for plan in self.plans) # pylint: disable=protected-access diff --git a/sandbox/docker_sandbox.py b/sandbox/docker_sandbox.py index 5c4645ef..695814be 100644 --- a/sandbox/docker_sandbox.py +++ b/sandbox/docker_sandbox.py @@ -193,7 +193,7 @@ def _archive_workspace(source: str | Path) -> bytes: return raw.getvalue() -def _default_workspace_builder(case_id: str, image: str, env: dict[str, str]) -> Workspace: +def _default_workspace_builder(workspace_id: str, image: str, env: dict[str, str]) -> Workspace: """Create the exact AgentScope 2.0.4.post1 Docker backend lazily.""" import agentscope @@ -202,7 +202,7 @@ def _default_workspace_builder(case_id: str, image: str, env: dict[str, str]) -> if agentscope.__version__ != "2.0.4.post1": raise RuntimeError(f"sandbox host requires agentscope 2.0.4.post1, got {agentscope.__version__}") return DirectDockerWorkspace( - workspace_id=f"reme-benchmark-{case_id}", + workspace_id=workspace_id, base_image=image, env=env, ) @@ -232,14 +232,23 @@ class DockerReMeSandboxFactory: self.config = config self.command_timeout = command_timeout self._workspace_builder = workspace_builder or _default_workspace_builder + self._workspace_prefix = f"reme-benchmark-{uuid.uuid4().hex[:12]}" + self._workspace_sequence = 0 + + def _next_workspace_id(self) -> str: + """Allocate a factory-scoped container identity independent of case IDs.""" + self._workspace_sequence += 1 + return f"{self._workspace_prefix}-{self._workspace_sequence:04d}" async def create_case(self, case_id: str) -> "DockerReMeSandbox": """Start and prepare a new container for one benchmark case.""" _validate_case_id(case_id) image = self.candidate.base_image if isinstance(self.candidate, SourceCandidate) else self.candidate.image - workspace = self._workspace_builder(case_id, image, self.env) + container_id = self._next_workspace_id() + workspace = self._workspace_builder(container_id, image, self.env) case = DockerReMeSandbox( case_id=case_id, + container_id=container_id, candidate=self.candidate, workspace=workspace, config=self.config, @@ -282,6 +291,7 @@ class DockerReMeSandbox: self, *, case_id: str, + container_id: str, candidate: SourceCandidate | ImageCandidate, workspace: Workspace, config: str, @@ -289,6 +299,7 @@ class DockerReMeSandbox: environment_names: list[str], ) -> None: self.case_id = case_id + self.container_id = container_id self.candidate = candidate self.workspace = workspace self.config = config @@ -504,6 +515,31 @@ class DockerReMeSandbox: async def run_queries(self, queries: list[EvaluationQuery | dict[str, Any]]) -> dict[str, Any]: """Answer and judge multiple queries in one Application with isolated logs.""" + serialized, query_ids = self._serialize_queries(queries) + if not serialized: + raise ValueError("run_queries requires at least one query") + if len(query_ids) != len(set(query_ids)): + raise ValueError("query IDs must be unique") + async with self._operation_lock: + return await self._run_worker_request("queries", {"queries": serialized}) + + async def run_query(self, query: EvaluationQuery | dict[str, Any]) -> dict[str, Any]: + """Answer and judge one append-only query for a host-side lease.""" + + serialized, _ = self._serialize_queries([query]) + async with self._operation_lock: + response = await self._run_worker_request("query", {"queries": serialized}) + results = response.get("queries") + if not isinstance(results, list) or len(results) != 1 or not isinstance(results[0], dict): + raise SandboxCommandError("single-query worker returned an invalid result") + return results[0] + + def _serialize_queries( + self, + queries: list[EvaluationQuery | dict[str, Any]], + ) -> tuple[list[dict[str, Any]], list[str]]: + """Validate and serialize query payloads shared by batch and lease APIs.""" + serialized: list[dict[str, Any]] = [] query_ids: list[str] = [] for query in queries: @@ -516,12 +552,7 @@ class DockerReMeSandbox: raise ValueError(f"query {query_id!r} requires a string question") query_ids.append(query_id) serialized.append(value) - if not serialized: - raise ValueError("run_queries requires at least one query") - if len(query_ids) != len(set(query_ids)): - raise ValueError("query IDs must be unique") - async with self._operation_lock: - return await self._run_worker_request("queries", {"queries": serialized}) + return serialized, query_ids async def ingest_session( self, @@ -750,6 +781,27 @@ class DockerReMeSandbox: target.write_bytes(await self.backend.read_file(EXPORT_ARCHIVE)) return target + async def export_query(self, query_id: str, destination: str | Path) -> Path: + """Download one leased query's logs and result without a case summary.""" + + self._validate_query_artifact_id(query_id) + async with self._operation_lock: + assert self.backend is not None, "sandbox is not initialized" + query_root = f"{CASE_ROOT}/queries/{query_id}" + required = f"{query_root}/result.json" + try: + await self.backend.read_file(required) + except FileNotFoundError as exc: + raise SandboxCommandError(f"query export is missing required artifact: {required}") from exc + await self._exec_checked( + "export_query", + ["tar", "-czf", EXPORT_ARCHIVE, "-C", CASE_ROOT, f"queries/{query_id}"], + ) + target = Path(destination).expanduser().resolve() + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(await self.backend.read_file(EXPORT_ARCHIVE)) + return target + async def export_evaluation(self, destination: str | Path) -> Path: """Download the workspace, build log, and per-query evaluation artifacts.""" async with self._operation_lock: @@ -870,6 +922,7 @@ class DockerReMeSandbox: ) manifest = { "case_id": self.case_id, + "container_id": self.container_id, "candidate_id": candidate_id, "candidate_mode": "source" if isinstance(self.candidate, SourceCandidate) else "image", "config": self.config, diff --git a/sandbox/worker.py b/sandbox/worker.py index c39846b6..3b59f502 100644 --- a/sandbox/worker.py +++ b/sandbox/worker.py @@ -228,8 +228,13 @@ async def _run_build(request: dict[str, Any]) -> dict[str, Any]: } -async def _run_queries(request: dict[str, Any]) -> dict[str, Any]: - """Execute answer-and-judge pairs in one Application with isolated logs.""" +async def _run_queries(request: dict[str, Any], *, write_summary: bool = True) -> dict[str, Any]: + """Execute answer-and-judge pairs in one Application with isolated logs. + + ``write_summary=False`` is the append-only primitive used by the validation + scheduler. It runs exactly one independently leased query without sealing + the case-level artifact directory with ``summary.json``. + """ case_root, app_config = _prepare_runtime(request) from reme.reme import ReMe from reme.utils import get_logger @@ -237,6 +242,8 @@ async def _run_queries(request: dict[str, Any]) -> dict[str, Any]: queries = request.get("queries") if not isinstance(queries, list) or not queries: raise ValueError("query request requires a non-empty queries list") + if not write_summary and len(queries) != 1: + raise ValueError("single-query request requires exactly one query") query_ids = [_validate_query_id(query.get("query_id") if isinstance(query, dict) else None) for query in queries] if len(query_ids) != len(set(query_ids)): raise ValueError("query IDs must be unique") @@ -329,7 +336,8 @@ async def _run_queries(request: dict[str, Any]) -> dict[str, Any]: for result in results ], } - _atomic_write_json(queries_root / "summary.json", summary) + if write_summary: + _atomic_write_json(queries_root / "summary.json", summary) return {"success": all(result["error"] is None for result in results), "summary": summary, "queries": results} @@ -352,6 +360,8 @@ def main() -> int: operation = _run_build(request) elif mode == "queries": operation = _run_queries(request) + elif mode == "query": + operation = _run_queries(request, write_summary=False) else: raise ValueError(f"unknown sandbox worker mode: {mode!r}") result = asyncio.run(operation) diff --git a/tests/unit_sandbox/test_meta_reme_validation.py b/tests/unit_sandbox/test_meta_reme_validation.py index 77b7e415..b88bde1c 100644 --- a/tests/unit_sandbox/test_meta_reme_validation.py +++ b/tests/unit_sandbox/test_meta_reme_validation.py @@ -27,6 +27,7 @@ models = importlib.import_module("models") workspace_module = importlib.import_module("workspace") adapters = importlib.import_module("validation.adapters") evaluator = importlib.import_module("validation.evaluator") +scheduler_module = importlib.import_module("validation.scheduler") def _domain(max_retries: int = 0): @@ -47,7 +48,7 @@ def _domain(max_retries: int = 0): ) -def _case(case_id: str): +def _case(case_id: str, *, query_count: int = 1): return models.CaseSpec( case_id=case_id, sessions=[ @@ -65,18 +66,19 @@ def _case(case_id: str): ], queries=[ models.QuerySpec( - query_id=f"query-{case_id}", + query_id=f"query-{case_id}-{index}" if query_count > 1 else f"query-{case_id}", question="What should be remembered?", golden_answer=case_id, metadata={"question_type": "single-session-user"}, - ), + ) + for index in range(query_count) ], metadata={"dataset": "longmemeval"}, ) -def _prepared_workspace(tmp_path: Path, case_ids: list[str]): - workspace = workspace_module.Workspace.create(tmp_path / "workspace", _domain()) +def _prepared_workspace(tmp_path: Path, case_ids: list[str], *, max_retries: int = 0): + workspace = workspace_module.Workspace.create(tmp_path / "workspace", _domain(max_retries=max_retries)) for index, case_id in enumerate(case_ids): workspace.atomic_write_json(f"dataset/cases/{index:06d}.json", _case(case_id)) workspace.atomic_write_json( @@ -105,10 +107,13 @@ class _FakeFactory: self.active = 0 self.max_active = 0 self.events = [] + self.created = [] + self.resets = [] 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 _FakeCase(self, case_id) @@ -116,6 +121,7 @@ class _FakeCase: def __init__(self, factory, case_id): self.factory = factory self.case_id = case_id + self.container_id = f"container-{len(factory.created)}" async def run_build(self, jobs): self.factory.events.append((self.case_id, "run_build")) @@ -123,6 +129,10 @@ class _FakeCase: await asyncio.sleep(0.01) return {"success": True, "jobs": []} + async def reset_case(self, case_id): + self.factory.resets.append((self.case_id, case_id)) + self.case_id = case_id + async def run_queries(self, queries): self.factory.events.append((self.case_id, "run_queries")) await asyncio.sleep(0.01) @@ -142,6 +152,24 @@ class _FakeCase: ], } + async def run_query(self, query): + self.factory.events.append((self.case_id, f"run_query:{query.query_id}")) + await asyncio.sleep(0.01) + return { + "query_id": query.query_id, + "question": query.question, + "golden_answer": query.golden_answer, + "answer": query.golden_answer, + "score": 1.0, + "answer_result": {"success": True, "answer": query.golden_answer}, + "judge_result": {"success": True, "answer": "yes"}, + "error": None, + } + + async def upload_workspace(self, source): + assert Path(source).is_file() + self.factory.events.append((self.case_id, "upload_workspace")) + async def _export(self, destination, files): destination = Path(destination) destination.parent.mkdir(parents=True, exist_ok=True) @@ -171,6 +199,13 @@ class _FakeCase: ["queries/summary.json", f"{query_root}/answer.log", f"{query_root}/result.json"], ) + async def export_query(self, query_id, destination): + self.factory.events.append((self.case_id, f"export_query:{query_id}")) + return await self._export( + destination, + [f"queries/{query_id}/answer.log", f"queries/{query_id}/result.json"], + ) + async def export_full(self, destination): return await self._export(destination, ["failure.json"]) @@ -202,6 +237,8 @@ def test_validation_uses_immutable_commit_and_bounds_full_case_lifecycle(tmp_pat assert output == workspace.path("evaluations/candidate-1/validation-1") assert created[0].max_active == 2 + assert len(created[0].created) == 2 + assert len(created[0].resets) == 2 assert created[0].kwargs["config"] == "lme.yaml" with tarfile.open(fileobj=io.BytesIO(created[0].candidate.snapshot.archive), mode="r:gz") as archive: assert archive.extractfile("reme/__init__.py").read() == b"VERSION = 'committed'\n" @@ -211,12 +248,19 @@ def test_validation_uses_immutable_commit_and_bounds_full_case_lifecycle(tmp_pat assert summary["commit_sha"] == commit assert manifest["code_id"] == "candidate-1" assert manifest["commit_sha"] == commit + assert manifest["container_reuse"] is True + assert manifest["scheduling"] == { + "construction_barrier": True, + "query_unit": "single_query_lease", + "workspace_affinity": "case_id", + } assert [event for case_id, event in created[0].events if case_id == "case-1"] == [ "run_build", "export_workspace", "export_build_log", - "run_queries", - "export_queries", + "upload_workspace", + "run_query:query-case-1", + "export_query:query-case-1", ] assert summary["status"] == "completed" assert summary["mean_query_score"] == 1.0 @@ -243,6 +287,281 @@ def test_validation_uses_immutable_commit_and_bounds_full_case_lifecycle(tmp_pat assert not list((attempt / "queries").glob(".*.tar.gz")) +def test_validation_builds_every_memory_before_stealing_ordered_queries(tmp_path: Path) -> None: + """Idle workers fan out one case only after the strict construction barrier.""" + + workspace, repository, commit = _prepared_workspace(tmp_path, ["case-many"]) + workspace.atomic_write_json("dataset/cases/000000.json", _case("case-many", query_count=6)) + subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True) + created = [] + + def build_factory(*args, **kwargs): + factory = _FakeFactory(*args, **kwargs) + created.append(factory) + return factory + + output = evaluator.run_validation( + workspace.root, + ["case-many"], + "candidate-1", + 3, + validation_id="query-stealing", + factory_builder=build_factory, + environment={}, + ) + + factory = created[0] + query_events = [event for _, event in factory.events if event.startswith("run_query:")] + assert len(factory.created) == 3 + assert len(query_events) == 6 + assert [event for _, event in factory.events].index(query_events[0]) > max( + index for index, (_, event) in enumerate(factory.events) if event == "export_build_log" + ) + assert sum(event == "upload_workspace" for _, event in factory.events) == 2 + summary = json.loads((output / "summary.json").read_text(encoding="utf-8")) + assert [query["query_id"] for query in summary["cases"][0]["queries"]] == [ + f"query-case-many-{index}" for index in range(6) + ] + + +def test_validation_replaces_an_infra_error_container_before_retrying(tmp_path: Path) -> None: + """A suspect pool member is destroyed; its replacement remains reusable.""" + + class RetryCase(_FakeCase): + """Fail the first build to emulate a broken container.""" + + async def run_build(self, jobs): + if self.factory.fail_next_build: + self.factory.fail_next_build = False + raise RuntimeError("container became unhealthy") + return await super().run_build(jobs) + + class RetryFactory(_FakeFactory): + """Create retry cases while tracking physical replacements.""" + + def __init__(self, candidate, **kwargs): + super().__init__(candidate, **kwargs) + self.fail_next_build = True + + 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 RetryCase(self, case_id) + + workspace, repository, commit = _prepared_workspace(tmp_path, ["case-1", "case-2"], max_retries=1) + subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True) + created = [] + + def build_factory(*args, **kwargs): + factory = RetryFactory(*args, **kwargs) + created.append(factory) + return factory + + output = evaluator.run_validation( + workspace.root, + ["case-1", "case-2"], + "candidate-1", + 1, + validation_id="replace-infra-container", + factory_builder=build_factory, + environment={}, + ) + + 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.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() + + +def test_validation_discards_a_container_when_reset_fails(tmp_path: Path) -> None: + """A failed reset cannot leave the pool member available to later cases.""" + + class ResetFailureCase(_FakeCase): + """Reject assignment of the second logical case.""" + + async def reset_case(self, case_id): + if case_id == "case-2": + raise RuntimeError("reset failed") + await super().reset_case(case_id) + + class ResetFailureFactory(_FakeFactory): + """Create reset-failure cases while tracking replacements.""" + + 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 ResetFailureCase(self, case_id) + + case_ids = ["case-1", "case-2", "case-3"] + workspace, repository, commit = _prepared_workspace(tmp_path, case_ids) + subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True) + created = [] + + def build_factory(*args, **kwargs): + factory = ResetFailureFactory(*args, **kwargs) + created.append(factory) + return factory + + output = evaluator.run_validation( + workspace.root, + case_ids, + "candidate-1", + 1, + validation_id="replace-reset-failure", + factory_builder=build_factory, + environment={}, + ) + + factory = created[0] + summary = json.loads((output / "summary.json").read_text(encoding="utf-8")) + assert factory.created == ["case-1", "case-3"] + assert factory.resets == [("case-3", "case-1")] + assert factory.active == 0 + assert [result["status"] for result in summary["cases"]] == ["completed", "infra_error", "completed"] + + +def test_construction_candidate_failure_skips_case_queries(tmp_path: Path) -> None: + """A structured build failure is terminal and never enters the query queue.""" + + class FailedBuildCase(_FakeCase): + """Return a structured candidate failure from construction.""" + + async def run_build(self, jobs): + self.factory.events.append((self.case_id, "run_build")) + return {"success": False, "jobs": [{"job": jobs[0].job, "result": {"success": False}}]} + + class FailedBuildFactory(_FakeFactory): + """Create cases whose candidate build fails normally.""" + + 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 FailedBuildCase(self, case_id) + + workspace, repository, commit = _prepared_workspace(tmp_path, ["case-1"]) + subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True) + created = [] + + def build_factory(*args, **kwargs): + factory = FailedBuildFactory(*args, **kwargs) + created.append(factory) + return factory + + output = evaluator.run_validation( + workspace.root, + ["case-1"], + "candidate-1", + 2, + validation_id="failed-construction", + factory_builder=build_factory, + environment={}, + ) + + 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() + + +def test_query_infrastructure_failure_is_saved_and_retried_independently(tmp_path: Path) -> None: + """A failed lease retires its container without rebuilding case memory.""" + + class FlakyQueryCase(_FakeCase): + """Raise one infrastructure exception from the query primitive.""" + + async def run_query(self, query): + if self.factory.fail_next_query: + self.factory.fail_next_query = False + raise RuntimeError("query worker disconnected") + return await super().run_query(query) + + class FlakyQueryFactory(_FakeFactory): + """Create query cases while sharing the one-shot failure flag.""" + + def __init__(self, candidate, **kwargs): + super().__init__(candidate, **kwargs) + self.fail_next_query = True + + 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 FlakyQueryCase(self, case_id) + + workspace, repository, commit = _prepared_workspace(tmp_path, ["case-1"], max_retries=1) + subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True) + created = [] + + def build_factory(*args, **kwargs): + factory = FlakyQueryFactory(*args, **kwargs) + created.append(factory) + return factory + + output = evaluator.run_validation( + workspace.root, + ["case-1"], + "candidate-1", + 1, + validation_id="retry-query", + factory_builder=build_factory, + environment={}, + ) + + 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"] + 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.""" + + class BrokenQueryCase(_FakeCase): + """Always lose the query worker response.""" + + async def run_query(self, query): + raise RuntimeError(f"lost {query.query_id}") + + class BrokenQueryFactory(_FakeFactory): + """Create cases with a permanently broken query primitive.""" + + 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 BrokenQueryCase(self, case_id) + + workspace, repository, commit = _prepared_workspace(tmp_path, ["case-1"]) + subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True) + output = evaluator.run_validation( + workspace.root, + ["case-1"], + "candidate-1", + 2, + validation_id="exhaust-query", + factory_builder=BrokenQueryFactory, + environment={}, + ) + + summary = json.loads((output / "summary.json").read_text(encoding="utf-8")) + case_result = summary["cases"][0] + assert summary["status"] == "infra_error" + 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() + + @pytest.mark.asyncio async def test_validation_exposes_native_async_project_api(tmp_path: Path) -> None: workspace, _, _ = _prepared_workspace(tmp_path, ["case-async"]) @@ -315,6 +634,34 @@ def test_adapters_build_dataset_specific_requests() -> None: assert beam_query.score_mapping is None +@pytest.mark.asyncio +async def test_query_scheduler_prefers_loaded_memory_and_fences_retries() -> 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) + + owner = await scheduler.claim(0, first.case_id) + thief = await scheduler.claim(1, first.case_id) + assert owner.plan is first + assert owner.selection == "affinity" + 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 + 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 + await scheduler.complete(remaining, {"query_id": "q3", "score": 1.0}) + assert await scheduler.claim(0, None) is None + + def test_longmemeval_build_filters_future_sessions() -> None: case = _case("lme-filter") case.queries[0].metadata["question_date"] = "2026/08/07 (Fri) 09:30" diff --git a/tests/unit_sandbox/test_sandbox.py b/tests/unit_sandbox/test_sandbox.py index 097ce280..541c3640 100644 --- a/tests/unit_sandbox/test_sandbox.py +++ b/tests/unit_sandbox/test_sandbox.py @@ -104,6 +104,23 @@ class FakeBackend: }, "queries": [], } + elif request.get("mode") == "query": + item = request["queries"][0] + query_root = f"/workspace/case/queries/{item['query_id']}" + result = { + "query_id": item["query_id"], + "question": item["question"], + "golden_answer": item.get("golden_answer"), + "answer": item.get("golden_answer"), + "score": 1.0, + "error": None, + } + self.files[f"{query_root}/result.json"] = json.dumps(result).encode() + response = { + "success": True, + "summary": {"queries": [{"query_id": item["query_id"], "score": 1.0}]}, + "queries": [result], + } else: response = { "success": True, @@ -211,6 +228,13 @@ async def test_one_source_candidate_is_reused_in_independent_case_workspaces(tmp first, second = await factory.create_cases(["case-1", "case-2"], concurrency=2) assert first.workspace is not second.workspace + assert first.container_id != second.container_id + assert first.container_id.startswith("reme-benchmark-") + assert second.container_id.startswith("reme-benchmark-") + assert workspaces[0].case_id == first.container_id + assert workspaces[1].case_id == second.container_id + assert "case-1" not in first.container_id + assert "case-2" not in second.container_id assert workspaces[0].image == workspaces[1].image == "base:test" assert workspaces[0].backend.files["/workspace/candidate.tar.gz"] is candidate.snapshot.archive assert workspaces[1].backend.files["/workspace/candidate.tar.gz"] is candidate.snapshot.archive @@ -243,7 +267,9 @@ async def test_image_candidate_skips_source_upload_and_runs_jobs(tmp_path): } assert "/workspace/candidate.tar.gz" not in workspaces[0].backend.files assert exported.read_bytes() == b"case-archive" - assert json.loads(workspaces[0].backend.files["/workspace/case/manifest.json"])["candidate_mode"] == "image" + manifest = json.loads(workspaces[0].backend.files["/workspace/case/manifest.json"]) + assert manifest["candidate_mode"] == "image" + assert manifest["container_id"] == case.container_id export_command = next(command for command, _, _ in workspaces[0].backend.commands if command[:2] == ["tar", "-czf"]) assert "logs" in export_command assert "results" in export_command @@ -338,6 +364,35 @@ async def test_build_and_queries_use_batch_worker_requests(tmp_path): ] +@pytest.mark.asyncio +async def test_single_query_api_appends_and_exports_one_lease(tmp_path): + """The lease primitive does not require or create a case-level summary.""" + + workspace = FakeWorkspace("case-1", "candidate:test", {}) + factory = DockerReMeSandboxFactory( + ImageCandidate("candidate:test"), + workspace_builder=lambda *_args: workspace, + ) + case = await factory.create_case("case-1") + query = EvaluationQuery(query_id="query:1", question="What?", golden_answer="This") + + result = await case.run_query(query) + exported = await case.export_query(query.query_id, tmp_path / "query.tar.gz") + + assert result["query_id"] == "query:1" + assert result["score"] == 1.0 + assert "/workspace/case/queries/summary.json" not in workspace.backend.files + assert exported.read_bytes() == b"case-archive" + assert workspace.backend.commands[-1][0] == [ + "tar", + "-czf", + "/tmp/reme-case-export.tar.gz", + "-C", + "/workspace/case", + "queries/query:1", + ] + + @pytest.mark.asyncio async def test_query_artifact_ids_reject_path_components(): """Verbatim query directory names cannot escape the queries root."""