Implement two-phase validation scheduling with reusable workers

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:

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

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

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,

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)

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"

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."""