feat(sandbox): add two-phase build/queries evaluation architecture

- Add meta-reme/git_manager.py for immutable baseline Git initialization
- Enhance prepare_workspace to build and init benchmark code bundle
- Refactor sandbox worker into build/queries modes with shared Application
  lifecycle and per-query log isolation
- Add run_build, run_queries, export_evaluation to DockerReMeSandbox
- Add JobRequest and EvaluationQuery data models
- Support log_filepath parameter in get_logger for isolated query logs
- Add safety validation for case IDs and query artifact IDs
- Update tests for build/queries modes, log isolation, and workspace init
- Add sandbox README_ZH.md and update design docs

meta-reme/design.md

@@ -95,7 +95,7 @@ search set 被转换为统一 JSON 格式复制到 workspace，并设置为只
 
 `bundle_builder.py` 从完整 ReMe 仓库构建 benchmark 所需的最小代码仓库，包括 application、schema、registry、必要组件、auto-memory、search、answer、judge、配置和 packaging 文件。DingTalk、cookbook 等无关代码不进入 bundle。
 
-Bundle 不能依赖人工随意删除文件，而应由 domain spec 中的文件清单确定性生成。生成后执行 import、组件注册、安装和固定 smoke case 检查，并验证精简版本与完整 ReMe baseline 行为一致。随后在 workspace 中初始化 Git，创建不可变 baseline commit。
+Bundle 不能依赖人工随意删除文件，而应由 domain spec 中的文件清单确定性生成。生成后执行 import、组件注册、安装和固定 smoke case 检查，并验证精简版本与完整 ReMe baseline 行为一致。随后在 workspace 中初始化 Git，在 `main` 分支创建 message 为 `Initial version` 的不可变 baseline commit。
 
 ### 5.1 Bundle Target
 

meta-reme/git_manager.py

@@ -0,0 +1,62 @@
+"""Git lifecycle operations for Meta-ReMe code repositories."""
+
+from __future__ import annotations
+
+from pathlib import Path
+import subprocess
+
+INITIAL_BRANCH = "main"
+INITIAL_COMMIT_MESSAGE = "Initial version"
+GIT_AUTHOR_NAME = "Meta-ReMe"
+GIT_AUTHOR_EMAIL = "meta-reme@localhost"
+
+
+class GitManagerError(RuntimeError):
+    """Raised when a managed Git operation fails."""
+
+
+def _git(repository: Path, *arguments: str, check: bool = True) -> subprocess.CompletedProcess[str]:
+    result = subprocess.run(
+        ["git", *arguments],
+        cwd=repository,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if check and result.returncode:
+        detail = (result.stderr or result.stdout).strip()
+        raise GitManagerError(f"git {' '.join(arguments)} failed in {repository}: {detail}")
+    return result
+
+
+def initialize_repository(repository: Path) -> str:
+    """Create the immutable initial commit, or return an existing repository's HEAD."""
+
+    repository = Path(repository).resolve()
+    if not repository.is_dir():
+        raise GitManagerError(f"Code repository does not exist: {repository}")
+
+    if (repository / ".git").exists():
+        _git(repository, "rev-parse", "--is-inside-work-tree")
+        head = _git(repository, "rev-parse", "--verify", "HEAD", check=False)
+        if head.returncode == 0:
+            return head.stdout.strip()
+        _git(repository, "symbolic-ref", "HEAD", f"refs/heads/{INITIAL_BRANCH}")
+    else:
+        _git(repository, "init", "--initial-branch", INITIAL_BRANCH)
+
+    _git(repository, "add", "--all")
+    _git(
+        repository,
+        "-c",
+        f"user.name={GIT_AUTHOR_NAME}",
+        "-c",
+        f"user.email={GIT_AUTHOR_EMAIL}",
+        "commit",
+        "--message",
+        INITIAL_COMMIT_MESSAGE,
+    )
+    branch = _git(repository, "branch", "--show-current").stdout.strip()
+    if branch != INITIAL_BRANCH:
+        raise GitManagerError(f"Initial repository branch is {branch!r}, expected {INITIAL_BRANCH!r}")
+    return _git(repository, "rev-parse", "HEAD").stdout.strip()

meta-reme/run.py

@@ -10,7 +10,9 @@ from typing import Any, Iterable
 
 import yaml
 
+from bundle_builder import build_bundle
 from data_preparation import prepare_dataset
+from git_manager import initialize_repository
 from models import (
     BudgetSpec,
     DatasetSpec,
@@ -26,6 +28,7 @@ DEFAULT_SOURCES = {
     "beam": PROJECT_ROOT / "benchmark/beam/dataset/BEAM",
     "longmemeval": PROJECT_ROOT / "benchmark/longmemeval/dataset/longmemeval_s_reme_cleaned.json",
 }
+INITIAL_CODE_DIR = Path("code/repo/reme")
 
 
 def _domain_spec(dataset: str, source: Path, source_fingerprint: str) -> DomainSpec:
@@ -48,7 +51,7 @@ def prepare_workspace(
     dataset_variant: str | None = None,
     dataset_source: Path | None = None,
 ) -> Workspace:
-    """Create/open a workspace and install only the selected training cases."""
+    """Create/open a workspace and install its training data and initial code bundle."""
 
     meta_workspace = Path(meta_workspace).resolve()
     source = Path(dataset_source or DEFAULT_SOURCES.get(dataset, "")).resolve()
@@ -71,9 +74,25 @@ def prepare_workspace(
         with workspace.acquire_lock():
             if not workspace.path("datasets/search/manifest.json").exists():
                 workspace.install_search_dataset(normalized)
+            _prepare_initial_code(workspace, spec.bundle_target)
     return workspace
 
 
+def _prepare_initial_code(workspace: Workspace, bundle_target: str) -> Path:
+    """Build the benchmark bundle once without overwriting an existing repository."""
+
+    bundle = workspace.path(INITIAL_CODE_DIR)
+    repository_parent = bundle.parent
+    if bundle.is_dir() and (bundle / "pyproject.toml").is_file():
+        initialize_repository(bundle)
+        return bundle
+    if any(repository_parent.iterdir()):
+        raise RuntimeError(f"Initial code directory is not empty or complete: {repository_parent}")
+    bundle = build_bundle(bundle_target, repository_parent, source_repo=PROJECT_ROOT)
+    initialize_repository(bundle)
+    return bundle
+
+
 def _mapping(value: Any, label: str) -> dict[str, Any]:
     if not isinstance(value, dict):
         raise ValueError(f"{label} must be a YAML mapping")
@@ -118,7 +137,7 @@ def parse_args() -> argparse.Namespace:
 
 
 def main() -> None:
-    """Prepare the configured workspace and print a short dataset summary."""
+    """Prepare the configured workspace, including its training data and initial code."""
 
     args = parse_args()
     workspace = prepare_workspace(**load_config(args.config))

reme/utils/logger_utils.py

@@ -51,7 +51,13 @@ def _enable_loguru() -> bool:
     return os.getenv("REME_DISABLE_LOGURU", "").lower() != "true"
 
 
-def _init_loguru(log_dir: str, level: str, log_to_console: bool, log_to_file: bool):
+def _init_loguru(
+    log_dir: str,
+    level: str,
+    log_to_console: bool,
+    log_to_file: bool,
+    log_filepath: str | None = None,
+):
     from loguru import logger
 
     logger.remove()
@@ -66,26 +72,35 @@ def _init_loguru(log_dir: str, level: str, log_to_console: bool, log_to_file: bo
 
     if log_to_file:
         try:
-            os.makedirs(log_dir, exist_ok=True)
-            current_ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
-            log_filepath = os.path.join(log_dir, f"{current_ts}_{os.getpid()}.log")
-
-            logger.add(
-                log_filepath,
-                level=level,
-                rotation="00:00",
-                retention="7 days",
-                compression="zip",
-                encoding="utf-8",
-                format=_LOGURU_FORMAT,
-            )
+            if log_filepath is None:
+                os.makedirs(log_dir, exist_ok=True)
+                current_ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+                log_filepath = os.path.join(log_dir, f"{current_ts}_{os.getpid()}.log")
+                logger.add(
+                    log_filepath,
+                    level=level,
+                    rotation="00:00",
+                    retention="7 days",
+                    compression="zip",
+                    encoding="utf-8",
+                    format=_LOGURU_FORMAT,
+                )
+            else:
+                os.makedirs(os.path.dirname(os.path.abspath(log_filepath)), exist_ok=True)
+                logger.add(log_filepath, level=level, encoding="utf-8", format=_LOGURU_FORMAT)
         except Exception as e:
             logger.error(f"Error configuring file logging: {e}")
 
     return logger
 
 
-def _init_stdlib(log_dir: str, level: str, log_to_console: bool, log_to_file: bool):
+def _init_stdlib(
+    log_dir: str,
+    level: str,
+    log_to_console: bool,
+    log_to_file: bool,
+    log_filepath: str | None = None,
+):
     logger = logging.getLogger("reme")
     logger.propagate = False
 
@@ -94,7 +109,7 @@ def _init_stdlib(log_dir: str, level: str, log_to_console: bool, log_to_file: bo
         handler.close()
 
     qwenpaw_logger = logging.getLogger(_QWENPAW_LOGGER_NAME)
-    if qwenpaw_logger.handlers:
+    if qwenpaw_logger.handlers and log_filepath is None:
         # QwenPaw owns the screen and file handlers. Forwarding keeps ReMe's
         # logger object stable for modules that cache it at import time, while
         # allowing future QwenPaw handlers (for example qwenpaw.log) to take
@@ -115,18 +130,21 @@ def _init_stdlib(log_dir: str, level: str, log_to_console: bool, log_to_file: bo
 
     if log_to_file:
         try:
-            os.makedirs(log_dir, exist_ok=True)
-            # Keep stdlib file naming aligned with QwenPaw logging.
-            current_ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
-            log_filepath = os.path.join(log_dir, f"{current_ts}.log")
-
-            file_handler = TimedRotatingFileHandler(
-                log_filepath,
-                when="midnight",
-                backupCount=7,
-                encoding="utf-8",
-                delay=True,
-            )
+            if log_filepath is None:
+                os.makedirs(log_dir, exist_ok=True)
+                # Keep stdlib file naming aligned with QwenPaw logging.
+                current_ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+                log_filepath = os.path.join(log_dir, f"{current_ts}.log")
+                file_handler = TimedRotatingFileHandler(
+                    log_filepath,
+                    when="midnight",
+                    backupCount=7,
+                    encoding="utf-8",
+                    delay=True,
+                )
+            else:
+                os.makedirs(os.path.dirname(os.path.abspath(log_filepath)), exist_ok=True)
+                file_handler = logging.FileHandler(log_filepath, encoding="utf-8", delay=True)
             file_handler.setLevel(level)
             file_handler.setFormatter(formatter)
             logger.addHandler(file_handler)
@@ -142,6 +160,7 @@ def get_logger(
     log_to_console: bool = True,
     log_to_file: bool = True,
     force_init: bool = False,
+    log_filepath: str | None = None,
 ):
     """Return the global logger, initializing sinks on first call (or when force_init)."""
     global _logger
@@ -156,7 +175,7 @@ def get_logger(
             return _logger
 
         if _enable_loguru():
-            _logger = _init_loguru(log_dir, level, log_to_console, log_to_file)
+            _logger = _init_loguru(log_dir, level, log_to_console, log_to_file, log_filepath)
         else:
-            _logger = _init_stdlib(log_dir, level, log_to_console, log_to_file)
+            _logger = _init_stdlib(log_dir, level, log_to_console, log_to_file, log_filepath)
         return _logger

sandbox/README.md

@@ -1,5 +1,7 @@
 # ReMe Docker sandbox
 
+[简体中文](README_ZH.md)
+
 This directory runs benchmark cases in AgentScope `DockerWorkspace` containers.
 It never starts the ReMe HTTP service: the uploaded worker constructs a ReMe
 `Application` and calls `run_job()` directly.
@@ -25,6 +27,9 @@ pip install -e ".[sandbox]"
 - Source-candidate bytes are snapshotted once and reused unchanged across all
   cases created by the same factory.
 - Export happens before container shutdown.
+- Case IDs must start with an ASCII letter or digit, contain only letters,
+  digits, `.`, `_`, or `-`, and be at most 128 characters. IDs passed to one
+  `create_cases()` call must be unique.
 
 ## Build the dependency-only base image
 
@@ -87,10 +92,12 @@ asyncio.run(main())
 ```
 
 Source archives exclude VCS state, `.reme`, virtual environments, caches,
-credentials, benchmark datasets, existing memory workspaces, logs and build
-output. Dataset inputs should be passed to cases explicitly rather than copied
-into every candidate. Symbolic links are rejected so a snapshot cannot
-accidentally capture files outside the candidate root.
+`.env` credential files, benchmark datasets, existing memory workspaces, logs
+and build output. Other secrets are not discovered automatically and must not
+be placed in the candidate tree. Dataset inputs should be passed to cases
+explicitly rather than copied into every candidate. Symbolic links are
+rejected so a snapshot cannot accidentally capture files outside the
+candidate root.
 
 ## Mode 2: candidate preinstalled in an image
 
@@ -117,25 +124,138 @@ factory = DockerReMeSandboxFactory(
 case = await factory.create_case("session-001")
 ```
 
-## Direct job and artifact contract
+## Direct single-job and artifact contract
 
-The high-level methods map to the existing `lme.yaml` jobs:
+The legacy convenience methods map to the existing `lme.yaml` jobs:
 
 - `ingest_session()` → `auto_memory`, then optionally `index_update`
 - `answer()` → `agentic_answer`
 - `judge()` → `answer_judge`, with `yes=1`, `no=0`
 - `run_job()` → any explicitly named ReMe job
 
-`export()` defaults to an analysis-focused gzip archive containing all ReMe log
-files, the command audit log, per-job JSON results, canonical answer and score
-files, persisted agent sessions, and the user-owned workspace files, including
-files written outside the expected daily/digest paths. It also includes a
-validated runtime layout and a manifest. It omits request copies, temporary
-files, source resources, and rebuildable indexes/caches. Raw dialog sessions
-under the configured `session_dir`/`dialog_dir` are included. Use
+`export()` defaults to an analysis-focused gzip archive for this single-job
+flow. It contains the accumulated files under the legacy `logs/` directory,
+the command audit log, per-job JSON results, canonical answer and score files,
+persisted agent sessions, and the user-owned workspace files, including files
+written outside the expected daily/digest paths. It also includes a validated
+runtime layout and a manifest. It omits request copies, temporary files, source
+resources, and rebuildable indexes/caches. Raw dialog sessions under the
+configured `session_dir/dialog` path are included. Use
 `export(profile="full")` or `export_full()` to download the complete disposable
-case tree. Environment variable names are recorded, but their secret values
-are never written to the manifest.
+case tree. `include_candidate=True` additionally appends an uploaded source
+candidate; it is invalid for an image candidate. Environment variable names
+are recorded, but their secret values are never written to the manifest.
+
+The batch artifacts described below live under `build_log/` and `queries/` and
+are intentionally not part of the legacy analysis profile. Use
+`export_evaluation()` for them, or the full profile when the entire case is
+needed.
+
+## Build and multi-query batches
+
+For multi-query evaluation, use the batch APIs instead of starting one
+Application per job. `run_build()` executes every construction job in one
+Application. `run_queries()` then starts one new Application and reuses it for
+all answer-and-judge pairs. Query execution is sequential so token deltas and
+the process-global log sink remain attributable to one query. The restart at
+the build/query boundary also guarantees that evaluation depends on persisted
+workspace state rather than construction-only in-memory state.
+
+Each method publishes one phase for the active case. `run_build()` requires at
+least one job, stops at the first failed job, and refuses to append to an
+existing `build_log/build.log`. `run_queries()` requires at least one unique
+query ID, continues after individual query failures, and refuses to append if
+`queries/summary.json` or any requested query directory already exists. Call
+`reset_case()` before starting another evaluation in the same container.
+
+```python
+from sandbox import EvaluationQuery, JobRequest
+
+build = await case.run_build(
+    [
+        JobRequest("auto_memory", {"session_id": "session-1", "messages": messages}),
+        JobRequest("index_update"),
+    ],
+)
+assert build["success"]
+await case.commit_memory_history("constructed memory")
+
+evaluation = await case.run_queries(
+    [
+        EvaluationQuery(
+            query_id="question-1",
+            question="What should be remembered?",
+            golden_answer="Remember this.",
+            judge_arguments={
+                "query": "What should be remembered?",
+                "golden_answer": "Remember this.",
+            },
+        ),
+    ],
+)
+assert evaluation["success"]
+await case.export_evaluation("artifacts/case-1.tar.gz")
+```
+
+`export_evaluation()` produces exactly three top-level directories:
+
+```text
+reme_workspace/
+build_log/
+  build.log
+queries/
+  summary.json
+  <query-id>/
+    answer.log
+    result.json
+```
+
+Unlike the analysis profile, this export includes the complete
+`reme_workspace/`, including local Git history and rebuildable metadata,
+resources, indexes, or caches that are present. Export requires both
+`build_log/build.log` and `queries/summary.json`; call it only after both batch
+phases have published their artifacts.
+
+The query ID is used verbatim as its directory name and must therefore be one
+safe path component: it cannot be empty, `.`, `..`, `summary.json`, contain
+`/`, `\`, or NUL, or exceed 255 UTF-8 bytes. Each `answer.log` contains the
+answer and judge logs for that query only; Application startup and shutdown
+logs are excluded. Each `result.json` contains the question, golden answer,
+answer, normalized score, raw answer/judge results, token deltas, and any
+error. `queries/summary.json` records `case_id`, counts, mean score, and each
+query ID and score without a redundant directory field. A failed query
+receives a `null` score and an error, while later queries still run.
+
+Because IDs remain verbatim, callers are responsible for extraction-platform
+portability. For example, `:` is valid in the Linux container and archive but
+is not a valid Windows filename character.
+
+`EvaluationQuery` defaults to the LongMemEval convention: the answer is
+injected into the judge argument `agent_answer`, and judge answers `yes` and
+`no` map to `1.0` and `0.0`. `answer_arguments` and `judge_arguments` hold the
+remaining job arguments. `answer_job`, `judge_job`, `judge_answer_argument`,
+`score_path`, and `score_mapping` can override those conventions. Scores must
+normalize to the inclusive range `[0, 1]`.
+
+BEAM-style judges can select a numeric score from judge metadata:
+
+```python
+EvaluationQuery(
+    query_id="information_extraction:1",
+    question=question,
+    golden_answer=rubric,
+    judge_answer_argument="llm_response",
+    judge_arguments={
+        "rubric": rubric,
+        "probing_question": question,
+        "question_type": "information_extraction",
+    },
+    score_path="metadata.llm_judge_score",
+    score_mapping=None,
+)
+```
+
+## Local Git history for memory
 
 The runtime workspace is also a local Git repository. Session ingestion never
 commits implicitly: the host chooses checkpoint boundaries by calling
@@ -147,8 +267,9 @@ construction history can be inspected offline without a remote or push. A
 `reset_case()` removes this repository together with the old runtime workspace
 and initializes a fresh repository for the next case.
 
-For an ephemeral Docker workspace, always call `export()` or `export_full()`
-before `close()`.
+For an ephemeral Docker workspace, download every artifact that must survive
+before `close()`: use `export()`, `export_full()`, `export_evaluation()`, or
+`export_workspace()` according to the required contract.
 
 ## Reuse one built memory for parallel queries
 
@@ -176,10 +297,15 @@ finally:
     await asyncio.gather(*(case.close() for case in query_cases))
 ```
 
-`upload_workspace()` also accepts a host directory. It validates and repacks
-the directory or input archive before extraction; symbolic links, special
-files, and traversal paths are rejected. Pass `clear=False` only when an
-intentional merge into the target workspace is required.
+`upload_workspace()` also accepts a host directory or a legacy `export()`
+archive containing both `manifest.json` and `reme_workspace/`. It validates
+and repacks the directory or input archive before extraction; symbolic links,
+special files, duplicate archive paths, and traversal paths are rejected. An
+`export_evaluation()` archive is not an upload source because it intentionally
+has no manifest; use `export_workspace()` for portable snapshots. Pass
+`clear=False` only when an intentional merge into the target workspace is
+required. Upload always runs `git init`; existing history is preserved, while
+a directory without `.git` receives a fresh repository.
 
 ## Reuse one container for sequential cases
 
@@ -204,9 +330,10 @@ finally:
     await case.close()
 ```
 
-`reset_case()` deletes the ReMe runtime workspace, requests, logs, results,
-manifest, case-scoped temporary files, and temporary export archive. It retains
-the installed candidate, candidate virtual environment, and benchmark worker.
-Jobs, exports, and resets share one lock, so they never overlap within a
+`reset_case()` deletes the ReMe runtime workspace, requests, legacy logs,
+results, `build_log/`, `queries/`, manifest, case-scoped temporary files, and
+temporary export archives. It retains the installed candidate, candidate
+virtual environment, and benchmark worker. Jobs, batch phases, exports,
+uploads, commits, and resets share one lock, so they never overlap within a
 container. Do not retain paths or state from the previous case after reset
 returns.

sandbox/README_ZH.md

@@ -0,0 +1,321 @@
+# ReMe Docker 沙箱
+
+[English](README.md)
+
+本目录用于在 AgentScope `DockerWorkspace` 容器中运行基准测试 case。它不会启动
+ReMe HTTP 服务；上传到容器的 worker 会直接构造 ReMe `Application` 并调用
+`run_job()`。
+
+测试框架使用一个轻量的 `DockerWorkspace` 子类：保留 AgentScope 的 Docker
+生命周期和 `DockerBackend`，但跳过 MCP gateway 的构建与启动。直接执行 ReMe job
+不需要 gateway；省略它可以避免每个基础镜像额外进行一次依赖网络的镜像构建并启动
+后台进程。
+
+安装宿主机侧的 Docker 依赖：
+
+```bash
+pip install -e ".[sandbox]"
+```
+
+## 隔离模型
+
+- 一个 candidate 可以创建多个容器沙箱。
+- `create_cases()` 默认为每个 case 创建独立容器；也可以用 `reset_case()` 复用一个
+  容器，按顺序执行多个 case。
+- 当前 case 写入 `/workspace/case/reme_workspace`；容器交给下一个 case 前会删除该
+  目录。
+- 同一个 factory 创建的所有 case 复用同一份不可变 candidate 源码快照字节。
+- 必须在关闭容器前完成导出。
+- Case ID 必须以 ASCII 字母或数字开头，只能包含字母、数字、`.`、`_` 或 `-`，
+  最长 128 个字符。一次 `create_cases()` 调用中的 ID 必须唯一。
+
+## 构建仅包含依赖的基础镜像
+
+在仓库根目录运行：
+
+```bash
+docker build \
+  -f sandbox/Dockerfile.base \
+  -t reme-sandbox-base:agentscope-2.0.4-post1 \
+  .
+```
+
+该镜像根据 `pyproject.toml` 安装依赖，验证 AgentScope 版本为
+`2.0.4.post1`，随后卸载用于安装依赖的占位 ReMe 包。因此，准备 case 后，candidate
+源码会成为唯一可导入的 ReMe 实现。
+
+## 模式一：源码持续变化的 candidate
+
+```python
+import asyncio
+import os
+
+from sandbox import DockerReMeSandboxFactory, SourceCandidate, SourceSnapshot
+
+
+async def main():
+    # 每个 candidate 只创建一次快照，不要为每个 case 重复创建。
+    snapshot = SourceSnapshot.from_directory(".")
+    factory = DockerReMeSandboxFactory(
+        SourceCandidate(snapshot),
+        env={
+            name: os.environ[name]
+            for name in ("LLM_API_KEY", "LLM_BASE_URL")
+            if name in os.environ
+        },
+    )
+
+    # 每个 ID 对应独立容器和运行时工作区，但收到完全相同的快照字节。
+    cases = await factory.create_cases(["session-001", "session-002"])
+    try:
+        for case in cases:
+            await case.ingest_session(
+                session_id=case.case_id,
+                messages=[{"role": "user", "content": "Remember this."}],
+            )
+            await case.commit_memory_history(f"session: {case.case_id}")
+            answer = await case.answer(query="What should you remember?")
+            await case.judge(
+                query="What should you remember?",
+                agent_answer=str(answer.answer),
+                golden_answer="Remember this.",
+            )
+            await case.export(f"artifacts/{case.case_id}.tar.gz")
+    finally:
+        await asyncio.gather(*(case.close() for case in cases))
+
+
+asyncio.run(main())
+```
+
+源码归档会排除版本控制状态、`.reme`、虚拟环境、缓存、`.env` 凭据文件、基准
+数据集、已有 memory workspace、日志和构建产物。其他 secret 不会被自动识别，
+因此不能放入 candidate 目录树。数据集输入应显式传给 case，不应复制进每个
+candidate。源码快照拒绝符号链接，避免意外收录 candidate 根目录之外的文件。
+
+## 模式二：candidate 已预装在镜像中
+
+对于稳定 candidate，使用仓库的 `.dockerignore`（`sandbox/.dockerignore.example`
+保留了一份等价示例），然后运行：
+
+```bash
+docker build \
+  -f sandbox/Dockerfile.candidate \
+  --build-arg BASE_IMAGE=reme-sandbox-base:agentscope-2.0.4-post1 \
+  -t reme-candidate:my-candidate \
+  .
+```
+
+使用该 candidate 时，每个 case 不再上传源码：
+
+```python
+from sandbox import DockerReMeSandboxFactory, ImageCandidate
+
+factory = DockerReMeSandboxFactory(
+    ImageCandidate("reme-candidate:my-candidate", candidate_id="git-or-content-hash"),
+    env={...},
+)
+case = await factory.create_case("session-001")
+```
+
+## 单 job 直接调用与产物契约
+
+原有便捷方法映射到 `lme.yaml` 中的 job：
+
+- `ingest_session()` → `auto_memory`，随后可选执行 `index_update`
+- `answer()` → `agentic_answer`
+- `judge()` → `answer_judge`，其中 `yes=1`、`no=0`
+- `run_job()` → 任意显式指定的 ReMe job
+
+`export()` 默认生成面向上述单 job 流程的分析型 gzip 归档。它包含旧版 `logs/`
+目录中累积的日志、命令审计日志、每个 job 的 JSON 结果、标准 answer/score 文件、
+持久化的 agent session，以及用户拥有的 workspace 文件——包括写在预期
+daily/digest 路径之外的文件。归档还包含经过验证的运行时布局和 manifest；请求
+副本、临时文件、源码资源以及可重建的索引/缓存会被排除。配置的
+`session_dir/dialog` 路径下的原始对话 session 会被保留。
+
+使用 `export(profile="full")` 或 `export_full()` 可下载完整的临时 case 目录树。
+`include_candidate=True` 还会把上传的源码 candidate 附加到归档中；image candidate
+不支持这个参数。Manifest 只记录环境变量名，不会记录其秘密值。
+
+下文介绍的批处理产物位于 `build_log/` 和 `queries/`，有意不纳入旧版 analysis
+profile。需要这些产物时使用 `export_evaluation()`；需要整个 case 时使用 full
+profile。
+
+## Build 与多 Query 批处理
+
+多 query 评测应使用批处理 API，避免每个 job 都重新启动 Application。
+`run_build()` 在一个 Application 中执行全部 construction job；随后
+`run_queries()` 启动一个新的 Application，并让所有 answer/judge 对复用它。Query
+按顺序执行，因此 token 增量和进程级全局日志 sink 都能准确归属于单个 query。
+Build/query 边界上的重启还保证评测只依赖已持久化的 workspace 状态，而不依赖
+construction 阶段独有的内存状态。
+
+每个方法只为当前 case 发布一个阶段：
+
+- `run_build()` 至少需要一个 job；遇到第一个失败 job 后停止；如果
+  `build_log/build.log` 已存在，则拒绝追加。
+- `run_queries()` 至少需要一个 query，query ID 必须唯一；单个 query 失败后仍继续
+  后续 query；如果 `queries/summary.json` 或任一待运行 query 的目录已经存在，则
+  拒绝追加。
+- 若要在同一容器中启动另一轮评测，应先调用 `reset_case()`。
+
+```python
+from sandbox import EvaluationQuery, JobRequest
+
+build = await case.run_build(
+    [
+        JobRequest("auto_memory", {"session_id": "session-1", "messages": messages}),
+        JobRequest("index_update"),
+    ],
+)
+assert build["success"]
+await case.commit_memory_history("constructed memory")
+
+evaluation = await case.run_queries(
+    [
+        EvaluationQuery(
+            query_id="question-1",
+            question="What should be remembered?",
+            golden_answer="Remember this.",
+            judge_arguments={
+                "query": "What should be remembered?",
+                "golden_answer": "Remember this.",
+            },
+        ),
+    ],
+)
+assert evaluation["success"]
+await case.export_evaluation("artifacts/case-1.tar.gz")
+```
+
+`export_evaluation()` 生成且只生成三个顶层目录：
+
+```text
+reme_workspace/
+build_log/
+  build.log
+queries/
+  summary.json
+  <query-id>/
+    answer.log
+    result.json
+```
+
+与 analysis profile 不同，该导出包含完整的 `reme_workspace/`，包括本地 Git 历史，
+以及当时存在的可重建 metadata、resource、索引或缓存。导出要求
+`build_log/build.log` 和 `queries/summary.json` 都已存在，因此应在两个批处理阶段
+均完成产物发布后调用。
+
+Query ID 会原样用作目录名，因此必须是一个安全的路径组件：不能为空，不能是
+`.`、`..` 或 `summary.json`，不能包含 `/`、`\` 或 NUL，UTF-8 编码后不能超过
+255 字节。
+
+每个 `answer.log` 只包含对应 query 的 answer 和 judge 日志，不包含 Application
+启动和关闭日志。每个 `result.json` 包含问题、标准答案、实际回答、归一化 score、
+原始 answer/judge 结果、token 增量和错误信息。`queries/summary.json` 包含
+`case_id`、数量统计、平均分，以及每个 query 的 ID 和 score，不包含冗余的目录
+字段。失败 query 的 score 为 `null` 并带有 error，后续 query 仍会继续执行。
+
+由于 ID 原样保留，调用方需要自行保证解压平台的文件名兼容性。例如，`:` 在 Linux
+容器和归档中合法，但不能作为 Windows 文件名字符。
+
+`EvaluationQuery` 默认使用 LongMemEval 约定：answer 会注入 judge 参数
+`agent_answer`；judge 返回的 `yes` 和 `no` 分别映射为 `1.0` 和 `0.0`。
+`answer_arguments` 和 `judge_arguments` 用于提供其余 job 参数；可以通过
+`answer_job`、`judge_job`、`judge_answer_argument`、`score_path` 和
+`score_mapping` 覆盖默认约定。最终 score 必须落在闭区间 `[0, 1]`。
+
+BEAM 风格的 judge 可以从 judge metadata 中选择数值 score：
+
+```python
+EvaluationQuery(
+    query_id="information_extraction:1",
+    question=question,
+    golden_answer=rubric,
+    judge_answer_argument="llm_response",
+    judge_arguments={
+        "rubric": rubric,
+        "probing_question": question,
+        "question_type": "information_extraction",
+    },
+    score_path="metadata.llm_judge_score",
+    score_mapping=None,
+)
+```
+
+## Memory 的本地 Git 历史
+
+运行时 workspace 同时也是一个本地 Git 仓库。Session ingestion 不会隐式提交；
+宿主机通过 `commit_memory_history(message)` 自行选择 checkpoint 边界。可以在每个
+session 后提交，也可以将多个 session 合并为一次提交，并完全控制 commit message。
+
+每个 checkpoint 只提交配置的 `daily_dir`。即使没有变化，也会保留空提交，作为
+显式边界。导出的 workspace 包含 `.git`，因此无需 remote 或 push，也能离线检查
+daily memory 的构建历史。`reset_case()` 会连同旧运行时 workspace 一起删除该
+仓库，并为下一个 case 初始化新的仓库。
+
+对于临时 Docker workspace，必须在 `close()` 前下载所有需要保留的产物。根据所需
+契约选择 `export()`、`export_full()`、`export_evaluation()` 或
+`export_workspace()`。
+
+## 复用一次构建的 Memory，并行执行 Query
+
+完成 memory 构建后调用 `export_workspace()`，再用 `upload_workspace()` 把该快照
+交给多个独立 query case。默认情况下，上传前会清空目标运行时 workspace，因此
+所有 case 从同一 memory 状态开始，但不会共享可变文件或索引。
+
+```python
+memory_case = await factory.create_case("build-memory")
+try:
+    await memory_case.ingest_session(session_id="source", messages=[...])
+    await memory_case.run_job("index_update")
+    snapshot = await memory_case.export_workspace("artifacts/built-memory.tar.gz")
+finally:
+    await memory_case.close()
+
+query_cases = await factory.create_cases(["query-1", "query-2"])
+try:
+    await asyncio.gather(*(case.upload_workspace(snapshot) for case in query_cases))
+    answers = await asyncio.gather(
+        *(case.answer(query=query) for case, query in zip(query_cases, queries))
+    )
+finally:
+    await asyncio.gather(*(case.close() for case in query_cases))
+```
+
+`upload_workspace()` 也接受宿主机目录，或同时包含 `manifest.json` 和
+`reme_workspace/` 的旧版 `export()` 归档。它会先验证并重新打包目录或归档，再
+提取到容器；符号链接、特殊文件、重复归档路径和路径穿越都会被拒绝。
+`export_evaluation()` 归档有意不包含 manifest，因此不能直接用于上传；需要可移植
+快照时应使用 `export_workspace()`。仅当明确需要合并到当前 workspace 时才传入
+`clear=False`。上传后总会执行 `git init`：已有历史会被保留，没有 `.git` 的目录
+则会得到一个新仓库。
+
+## 顺序执行多个 Case 时复用一个容器
+
+当容器启动和源码安装占据大部分评测时间时，可以让一个容器按顺序处理多个 case。
+先完成并按需导出当前 case，然后清理临时 case 目录，再上传下一个 case 的 session：
+
+```python
+case = await factory.create_case("session-001")
+try:
+    await case.ingest_session(session_id="session-001", messages=[...])
+    await case.commit_memory_history("session: session-001")
+    first = await case.answer(query="...")
+    await case.export("artifacts/session-001.tar.gz")
+
+    await case.reset_case("session-002")
+    await case.ingest_session(session_id="session-002", messages=[...])
+    await case.commit_memory_history("session: session-002")
+    second = await case.answer(query="...")
+finally:
+    await case.close()
+```
+
+`reset_case()` 会删除 ReMe 运行时 workspace、请求、旧版日志、results、
+`build_log/`、`queries/`、manifest、case 级临时文件和临时导出归档；已安装的
+candidate、candidate 虚拟环境和 benchmark worker 会被保留。
+
+Job、批处理阶段、导出、上传、Git 提交和 reset 共用同一把锁，因此不会在同一容器
+内重叠执行。`reset_case()` 返回后，不要继续持有或使用前一个 case 的路径和状态。

sandbox/__init__.py

@@ -6,12 +6,14 @@ infrastructure, not part of ReMe's runtime contract.
 
 from .candidate import ImageCandidate, SourceCandidate, SourceSnapshot
 from .docker_sandbox import DockerReMeSandbox, DockerReMeSandboxFactory
-from .models import JobResult
+from .models import EvaluationQuery, JobRequest, JobResult
 
 __all__ = [
     "DockerReMeSandbox",
     "DockerReMeSandboxFactory",
     "ImageCandidate",
+    "EvaluationQuery",
+    "JobRequest",
     "JobResult",
     "SourceCandidate",
     "SourceSnapshot",

sandbox/docker_sandbox.py

@@ -16,7 +16,7 @@ from typing import Any, Callable, Literal, Protocol
 import uuid
 
 from .candidate import ImageCandidate, SourceCandidate
-from .models import ActionRecord, JobResult
+from .models import ActionRecord, EvaluationQuery, JobRequest, JobResult
 
 CASE_ROOT = "/workspace/case"
 CANDIDATE_ROOT = "/workspace/candidate"
@@ -140,15 +140,12 @@ def _archive_workspace(source: str | Path) -> bytes:
     elif source_path.is_file():
         with tarfile.open(source_path, mode="r:*") as archive:
             members = archive.getmembers()
-            normalized_paths = [
-                path for member in members if (path := _workspace_member_path(member.name)) is not None
-            ]
+            normalized_paths = [path for member in members if (path := _workspace_member_path(member.name)) is not None]
             # ``export()`` archives the whole case, while ``export_workspace``
             # archives only the workspace contents. Accept both forms so an
             # existing case artifact can seed independent query sandboxes.
-            case_export = (
-                PurePosixPath("manifest.json") in normalized_paths
-                and any(path.parts and path.parts[0] == "reme_workspace" for path in normalized_paths)
+            case_export = PurePosixPath("manifest.json") in normalized_paths and any(
+                path.parts and path.parts[0] == "reme_workspace" for path in normalized_paths
             )
             seen: set[PurePosixPath] = set()
             for member in members:
@@ -343,6 +340,8 @@ class DockerReMeSandbox:
                 self.runtime_workspace,
                 self.logs_dir,
                 self.results_dir,
+                f"{CASE_ROOT}/build_log",
+                f"{CASE_ROOT}/queries",
                 HARNESS_ROOT,
             ],
         )
@@ -445,11 +444,85 @@ class DockerReMeSandbox:
             raise SandboxCommandError(f"job {job!r} did not produce {response_path}") from exc
         return JobResult.from_dict(payload)
 
+    async def _run_worker_request(self, mode: str, payload: dict[str, Any]) -> dict[str, Any]:
+        """Run one structured worker request and return its JSON response."""
+        assert self.backend is not None, "sandbox is not initialized"
+        request_id = uuid.uuid4().hex
+        request_path = f"{CASE_ROOT}/inbox/{request_id}.json"
+        response_path = f"{self.results_dir}/{request_id}.json"
+        request = {
+            "mode": mode,
+            "config": self.config,
+            "case_id": self.case_id,
+            "case_root": CASE_ROOT,
+            "workspace_dir": self.runtime_workspace,
+            **payload,
+        }
+        await self.backend.write_file(request_path, json.dumps(request, ensure_ascii=False).encode("utf-8"))
+        await self._exec(
+            f"worker:{mode}",
+            [self.python, f"{HARNESS_ROOT}/worker.py", "--request", request_path, "--response", response_path],
+            cwd=CASE_ROOT,
+        )
+        try:
+            result = json.loads((await self.backend.read_file(response_path)).decode("utf-8"))
+        except FileNotFoundError as exc:
+            raise SandboxCommandError(f"worker mode {mode!r} did not produce {response_path}") from exc
+        if not isinstance(result, dict):
+            raise SandboxCommandError(f"worker mode {mode!r} returned a non-object response")
+        return result
+
     async def run_job(self, job: str, arguments: dict[str, Any] | None = None) -> JobResult:
         """Run one ReMe job without overlapping another operation."""
         async with self._operation_lock:
             return await self._run_job(job, arguments)
 
+    async def run_build(self, jobs: list[JobRequest | dict[str, Any]]) -> dict[str, Any]:
+        """Run all construction jobs in one Application and capture ``build.log``."""
+        serialized: list[dict[str, Any]] = []
+        for specification in jobs:
+            value = asdict(specification) if isinstance(specification, JobRequest) else dict(specification)
+            if not isinstance(value.get("job"), str) or not value["job"]:
+                raise ValueError("every build job requires a non-empty job name")
+            if not isinstance(value.get("arguments", {}), dict):
+                raise ValueError("build job arguments must be an object")
+            serialized.append(value)
+        if not serialized:
+            raise ValueError("run_build requires at least one job")
+        async with self._operation_lock:
+            return await self._run_worker_request("build", {"jobs": serialized})
+
+    @staticmethod
+    def _validate_query_artifact_id(query_id: str) -> None:
+        """Keep verbatim query-directory names inside the artifact root."""
+        if not query_id or len(query_id.encode("utf-8")) > 255:
+            raise ValueError("query_id must be non-empty and no longer than 255 UTF-8 bytes")
+        if query_id in {".", ".."} or "/" in query_id or "\\" in query_id or "\x00" in query_id:
+            raise ValueError(f"query_id is not a safe directory name: {query_id!r}")
+        if query_id == "summary.json":
+            raise ValueError("query_id conflicts with the queries summary: 'summary.json'")
+
+    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: list[dict[str, Any]] = []
+        query_ids: list[str] = []
+        for query in queries:
+            value = asdict(query) if isinstance(query, EvaluationQuery) else dict(query)
+            query_id = value.get("query_id")
+            if not isinstance(query_id, str):
+                raise ValueError("every evaluation query requires a string query_id")
+            self._validate_query_artifact_id(query_id)
+            if not isinstance(value.get("question"), str):
+                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})
+
     async def ingest_session(
         self,
         *,
@@ -617,6 +690,33 @@ class DockerReMeSandbox:
             target.write_bytes(await self.backend.read_file(WORKSPACE_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:
+            assert self.backend is not None, "sandbox is not initialized"
+            for required in (f"{CASE_ROOT}/build_log/build.log", f"{CASE_ROOT}/queries/summary.json"):
+                try:
+                    await self.backend.read_file(required)
+                except FileNotFoundError as exc:
+                    raise SandboxCommandError(f"evaluation export is missing required artifact: {required}") from exc
+            await self._exec_checked(
+                "export_evaluation",
+                [
+                    "tar",
+                    "-czf",
+                    EXPORT_ARCHIVE,
+                    "-C",
+                    CASE_ROOT,
+                    "reme_workspace",
+                    "build_log",
+                    "queries",
+                ],
+            )
+            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 upload_workspace(self, source: str | Path, *, clear: bool = True) -> None:
         """Copy a host workspace directory or exported archive into this case.
 

sandbox/models.py

@@ -43,3 +43,27 @@ class ActionRecord:
     stderr: str
     started_at: str
     finished_at: str
+
+
+@dataclass(frozen=True, slots=True)
+class JobRequest:
+    """One job executed as part of a shared build Application lifecycle."""
+
+    job: str
+    arguments: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class EvaluationQuery:
+    """One answer-and-judge request executed in a shared query Application."""
+
+    query_id: str
+    question: str
+    golden_answer: Any
+    answer_arguments: dict[str, Any] = field(default_factory=dict)
+    judge_arguments: dict[str, Any] = field(default_factory=dict)
+    answer_job: str = "agentic_answer"
+    judge_job: str = "answer_judge"
+    judge_answer_argument: str = "agent_answer"
+    score_path: str = "answer"
+    score_mapping: dict[str, float] | None = field(default_factory=lambda: {"yes": 1.0, "no": 0.0})

sandbox/worker.py

@@ -38,7 +38,11 @@ def _write_runtime_layout(case_root: Path, app_config: Any) -> None:
 
     resolved_paths: dict[str, str | None] = {}
     for field in _WORKSPACE_PATH_FIELDS:
-        configured = str(getattr(app_config, field)).strip()
+        configured = (
+            str(Path(app_config.session_dir) / "dialog")
+            if field == "dialog_dir"
+            else str(getattr(app_config, field)).strip()
+        )
         if not configured:
             resolved_paths[field] = None
             continue
@@ -65,12 +69,8 @@ def _write_runtime_layout(case_root: Path, app_config: Any) -> None:
     )
 
 
-async def _run(request: dict[str, Any]) -> dict[str, Any]:
-    """Construct an Application, run one job, and close it deterministically."""
-    job = request["job"]
-    job_args = request.get("arguments") or {}
-    config = request.get("config") or "lme.yaml"
-    workspace_dir = request["workspace_dir"]
+def _prepare_runtime(request: dict[str, Any]) -> tuple[Path, dict[str, Any]]:
+    """Resolve one request's config after confining process-local output."""
     case_root = Path(request["case_root"])
     case_tmp = case_root / "tmp"
     case_tmp.mkdir(parents=True, exist_ok=True)
@@ -80,47 +80,259 @@ async def _run(request: dict[str, Any]) -> dict[str, Any]:
     # Import after setting TMPDIR so ReMe and its subprocesses cannot leave
     # case-specific temporary files outside the disposable case root.
     from reme.config import resolve_app_config
-    from reme.enumeration import ComponentEnum
-    from reme.reme import ReMe
     from reme.schema import ApplicationConfig
-    from reme.utils.evaluation_interface import track_agent_token_usage
 
     app_config = resolve_app_config(
-        config=config,
-        workspace_dir=workspace_dir,
+        config=request.get("config") or "lme.yaml",
+        workspace_dir=request["workspace_dir"],
         log_to_console=False,
         log_to_file=True,
     )
     app_config["environment"] = dict(os.environ)
     resolved_config = ApplicationConfig.model_validate(app_config)
     _write_runtime_layout(case_root, resolved_config)
+    return case_root, app_config
+
+
+async def _run_job_on_app(app: Any, job: str, arguments: dict[str, Any]) -> dict[str, Any]:
+    """Run one job and return its serializable result with invocation token deltas."""
+    from reme.enumeration import ComponentEnum
+    from reme.utils.evaluation_interface import track_agent_token_usage
+
+    agent_names = list(app.context.components.get(ComponentEnum.AGENT_WRAPPER, {}))
+    token_usage: dict[str, dict[str, int | None]] = {}
+    try:
+        with track_agent_token_usage(agent_names, app.context) as token_usage:
+            response = await app.run_job(job, **arguments)
+    except Exception as exc:  # Preserve usage accumulated before a job failure.
+        return {
+            "success": False,
+            "answer": "",
+            "metadata": {},
+            "token_usage": token_usage,
+            "error": f"{type(exc).__name__}: {exc}",
+            "traceback": traceback.format_exc(),
+        }
+    return {
+        "success": bool(response.success),
+        "answer": response.answer,
+        "metadata": response.metadata,
+        "token_usage": token_usage,
+        "error": None if response.success else str(response.answer),
+    }
+
+
+def _atomic_write_json(path: Path, value: Any) -> None:
+    """Publish one JSON artifact without exposing a partially written file."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8")
+    temporary.replace(path)
+
+
+def _validate_query_id(query_id: Any) -> str:
+    """Validate an ID used verbatim as a directory name in the artifact."""
+    if not isinstance(query_id, str) or not query_id or len(query_id.encode("utf-8")) > 255:
+        raise ValueError("query_id must be a non-empty string no longer than 255 UTF-8 bytes")
+    if query_id in {".", ".."} or "/" in query_id or "\\" in query_id or "\x00" in query_id:
+        raise ValueError(f"query_id is not a safe directory name: {query_id!r}")
+    if query_id == "summary.json":
+        raise ValueError("query_id conflicts with the queries summary: 'summary.json'")
+    return query_id
+
+
+def _path_value(value: Any, dotted_path: str) -> Any:
+    """Read a dotted path from a nested result mapping."""
+    current = value
+    for part in dotted_path.split("."):
+        if not isinstance(current, dict) or part not in current:
+            raise ValueError(f"score path is missing from judge result: {dotted_path!r}")
+        current = current[part]
+    return current
+
+
+def _query_score(judge_result: dict[str, Any], query: dict[str, Any]) -> float:
+    """Normalize a configured judge result into one floating-point score."""
+    raw_score = _path_value(judge_result, str(query.get("score_path") or "answer"))
+    mapping = query.get("score_mapping")
+    if mapping is not None:
+        if not isinstance(mapping, dict):
+            raise ValueError("score_mapping must be an object or null")
+        key = str(raw_score).strip().lower()
+        normalized_mapping = {str(name).strip().lower(): score for name, score in mapping.items()}
+        if key not in normalized_mapping:
+            raise ValueError(f"judge score {raw_score!r} is not present in score_mapping")
+        raw_score = normalized_mapping[key]
+    score = float(raw_score)
+    if not 0.0 <= score <= 1.0:
+        raise ValueError(f"query score must be between 0 and 1, got {score}")
+    return score
+
+
+def _clear_query_state(app: Any) -> None:
+    """Drop request-scoped metadata that must not leak into the next query."""
+    app.context.metadata.pop("tool_contexts", None)
+
+
+async def _run(request: dict[str, Any]) -> dict[str, Any]:
+    """Construct an Application, run one job, and close it deterministically."""
+    job = request["job"]
+    job_args = request.get("arguments") or {}
+    _, app_config = _prepare_runtime(request)
+    from reme.reme import ReMe
+
     app = ReMe(**app_config)
     try:
         await app.start()
-        agent_names = list(app.context.components.get(ComponentEnum.AGENT_WRAPPER, {}))
-        try:
-            with track_agent_token_usage(agent_names, app.context) as token_usage:
-                response = await app.run_job(job, **job_args)
-        except Exception as exc:  # Preserve usage accumulated before a job failure.
-            return {
-                "success": False,
-                "answer": "",
-                "metadata": {},
-                "token_usage": token_usage,
-                "error": f"{type(exc).__name__}: {exc}",
-                "traceback": traceback.format_exc(),
-            }
-        return {
-            "success": bool(response.success),
-            "answer": response.answer,
-            "metadata": response.metadata,
-            "token_usage": token_usage,
-            "error": None if response.success else str(response.answer),
-        }
+        return await _run_job_on_app(app, job, job_args)
     finally:
         await app.close()
 
 
+async def _run_build(request: dict[str, Any]) -> dict[str, Any]:
+    """Execute all construction jobs in one Application and one build log."""
+    case_root, app_config = _prepare_runtime(request)
+    from reme.reme import ReMe
+    from reme.utils import get_logger
+
+    jobs = request.get("jobs")
+    if not isinstance(jobs, list) or not jobs:
+        raise ValueError("build request requires a non-empty jobs list")
+    build_log = case_root / "build_log" / "build.log"
+    if build_log.exists():
+        raise FileExistsError(f"build artifact already exists: {build_log}")
+
+    app = ReMe(**app_config)
+    logger = get_logger(log_to_console=False, log_to_file=True, force_init=True, log_filepath=str(build_log))
+    logger.info("[sandbox] build phase started")
+    results: list[dict[str, Any]] = []
+    try:
+        await app.start()
+        for index, specification in enumerate(jobs):
+            if not isinstance(specification, dict) or not isinstance(specification.get("job"), str):
+                raise ValueError(f"invalid build job at index {index}")
+            arguments = specification.get("arguments") or {}
+            if not isinstance(arguments, dict):
+                raise ValueError(f"build job arguments must be an object at index {index}")
+            result = await _run_job_on_app(app, specification["job"], arguments)
+            results.append({"job": specification["job"], "result": result})
+            if not result["success"]:
+                break
+    finally:
+        logger.info("[sandbox] build phase finished")
+        await app.close()
+        get_logger(log_to_console=False, log_to_file=False, force_init=True)
+    return {
+        "success": len(results) == len(jobs) and all(item["result"]["success"] for item in results),
+        "jobs": results,
+    }
+
+
+async def _run_queries(request: dict[str, Any]) -> dict[str, Any]:
+    """Execute answer-and-judge pairs in one Application with isolated logs."""
+    case_root, app_config = _prepare_runtime(request)
+    from reme.reme import ReMe
+    from reme.utils import get_logger
+
+    queries = request.get("queries")
+    if not isinstance(queries, list) or not queries:
+        raise ValueError("query request requires a non-empty queries list")
+    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")
+    queries_root = case_root / "queries"
+    if (queries_root / "summary.json").exists() or any((queries_root / query_id).exists() for query_id in query_ids):
+        raise FileExistsError("query artifacts already exist for this case")
+    queries_root.mkdir(parents=True, exist_ok=True)
+
+    app = ReMe(**app_config)
+    results: list[dict[str, Any]] = []
+    try:
+        await app.start()
+        for query, query_id in zip(queries, query_ids):
+            query_dir = queries_root / query_id
+            query_dir.mkdir(parents=False, exist_ok=False)
+            log_path = query_dir / "answer.log"
+            logger = get_logger(log_to_console=False, log_to_file=True, force_init=True, log_filepath=str(log_path))
+            logger.info(f"[sandbox] query started query_id={query_id!r}")
+            answer_result: dict[str, Any] | None = None
+            judge_result: dict[str, Any] | None = None
+            score: float | None = None
+            error: str | None = None
+            try:
+                question = query.get("question")
+                if not isinstance(question, str):
+                    raise ValueError(f"query {query_id!r} requires a string question")
+                answer_arguments = dict(query.get("answer_arguments") or {})
+                answer_arguments.setdefault("query", question)
+                answer_result = await _run_job_on_app(
+                    app,
+                    str(query.get("answer_job") or "agentic_answer"),
+                    answer_arguments,
+                )
+                if not answer_result["success"]:
+                    error = answer_result.get("error") or "answer job failed"
+                else:
+                    judge_arguments = dict(query.get("judge_arguments") or {})
+                    judge_answer_argument = str(query.get("judge_answer_argument") or "agent_answer")
+                    judge_arguments[judge_answer_argument] = answer_result["answer"]
+                    judge_result = await _run_job_on_app(
+                        app,
+                        str(query.get("judge_job") or "answer_judge"),
+                        judge_arguments,
+                    )
+                    if not judge_result["success"]:
+                        error = judge_result.get("error") or "judge job failed"
+                    else:
+                        score = _query_score(judge_result, query)
+            except Exception as exc:  # Keep later queries runnable and preserve this query's log.
+                error = f"{type(exc).__name__}: {exc}"
+                traceback_text = traceback.format_exc()
+            else:
+                traceback_text = None
+            finally:
+                logger.info(f"[sandbox] query finished query_id={query_id!r}")
+                _clear_query_state(app)
+                get_logger(log_to_console=False, log_to_file=False, force_init=True)
+
+            result = {
+                "query_id": query_id,
+                "question": query.get("question"),
+                "golden_answer": query.get("golden_answer"),
+                "answer": answer_result.get("answer") if answer_result is not None else None,
+                "score": score,
+                "answer_result": answer_result,
+                "judge_result": judge_result,
+                "error": error,
+            }
+            if traceback_text is not None:
+                result["traceback"] = traceback_text
+            _atomic_write_json(query_dir / "result.json", result)
+            results.append(result)
+    finally:
+        get_logger(log_to_console=False, log_to_file=False, force_init=True)
+        await app.close()
+
+    scores = [result["score"] for result in results if result["score"] is not None]
+    summary = {
+        "schema_version": 1,
+        "case_id": request.get("case_id"),
+        "query_count": len(results),
+        "scored_count": len(scores),
+        "mean_score": sum(scores) / len(scores) if scores else None,
+        "queries": [
+            {
+                "query_id": result["query_id"],
+                "score": result["score"],
+                **({"error": result["error"]} if result["error"] is not None else {}),
+            }
+            for result in results
+        ],
+    }
+    _atomic_write_json(queries_root / "summary.json", summary)
+    return {"success": all(result["error"] is None for result in results), "summary": summary, "queries": results}
+
+
 def main() -> int:
     """Read one request file and always emit a structured response file."""
     parser = argparse.ArgumentParser()
@@ -133,7 +345,16 @@ def main() -> int:
     response_path.parent.mkdir(parents=True, exist_ok=True)
     try:
         request = json.loads(request_path.read_text(encoding="utf-8"))
-        result = asyncio.run(_run(request))
+        mode = request.get("mode", "job")
+        if mode == "job":
+            operation = _run(request)
+        elif mode == "build":
+            operation = _run_build(request)
+        elif mode == "queries":
+            operation = _run_queries(request)
+        else:
+            raise ValueError(f"unknown sandbox worker mode: {mode!r}")
+        result = asyncio.run(operation)
         exit_code = 0 if result["success"] else 1
     except Exception as exc:  # The host needs an artifact even for broken candidates.
         result = {

tests/integration/test_sandbox_longmemeval.py

@@ -23,7 +23,7 @@ from dotenv import load_dotenv
 import pytest
 
 from benchmark.longmemeval.run import format_messages_for_reme, parse_haystack_date, sessions_sorted_by_time, to_iso
-from sandbox import DockerReMeSandboxFactory, SourceCandidate, SourceSnapshot
+from sandbox import DockerReMeSandboxFactory, EvaluationQuery, JobRequest, SourceCandidate, SourceSnapshot
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 DATASET_PATH = PROJECT_ROOT / "benchmark/datasets/longmemeval/longmemeval_s_reme_cleaned.json"
@@ -56,8 +56,8 @@ def _archive_members(payload: bytes) -> tuple[set[str], dict[str, bytes]]:
 
 
 @pytest.mark.asyncio
-async def test_first_longmemeval_case_analysis_export(tmp_path):
-    """Ingest two sessions, answer once, and verify selective download paths."""
+async def test_first_longmemeval_case_evaluation_export(tmp_path):
+    """Build once, reuse one query Application, and verify evaluation artifacts."""
     load_dotenv(PROJECT_ROOT / ".env", override=False)
     missing = [name for name in ("LLM_API_KEY", "EMBEDDING_API_KEY") if not os.environ.get(name)]
     if missing:
@@ -72,7 +72,7 @@ async def test_first_longmemeval_case_analysis_export(tmp_path):
     candidate = SourceCandidate(SourceSnapshot.from_directory(PROJECT_ROOT))
     factory = DockerReMeSandboxFactory(candidate, env=env, command_timeout=1800.0)
     case = await factory.create_case(f"longmemeval-{item['question_id']}-smoke")
-    archive_path = tmp_path / "analysis.tar.gz"
+    archive_path = tmp_path / "evaluation.tar.gz"
     try:
         assert case.backend is not None
         dependency_check = await case.backend.exec_shell(
@@ -92,37 +92,55 @@ async def test_first_longmemeval_case_analysis_export(tmp_path):
             "polars": False,
         }
 
+        build_jobs = []
         for _, session_dt, session_id, messages in sessions:
-            result = await case.ingest_session(
-                session_id=session_id,
-                messages=format_messages_for_reme(messages, session_dt),
-                date=session_dt.strftime("%Y-%m-%d"),
+            build_jobs.extend(
+                [
+                    JobRequest(
+                        "auto_memory",
+                        {
+                            "session_id": session_id,
+                            "messages": format_messages_for_reme(messages, session_dt),
+                            "date": session_dt.strftime("%Y-%m-%d"),
+                        },
+                    ),
+                    JobRequest("index_update"),
+                ],
             )
-            assert result.success, result.error or result.answer
-            await case.commit_memory_history(f"session: {session_id}")
+        build = await case.run_build(build_jobs)
+        assert build["success"], build
+        await case.commit_memory_history("constructed LongMemEval memory")
 
-        answer = await case.answer(
-            query=item["question"],
-            query_time=to_iso(parse_haystack_date(item["question_date"])),
+        evaluation = await case.run_queries(
+            [
+                EvaluationQuery(
+                    query_id=str(item["question_id"]),
+                    question=item["question"],
+                    golden_answer=item["answer"],
+                    answer_arguments={"query_time": to_iso(parse_haystack_date(item["question_date"]))},
+                    judge_arguments={
+                        "query": item["question"],
+                        "golden_answer": item["answer"],
+                        "question_type": item["question_type"],
+                    },
+                ),
+            ],
         )
-        assert answer.success, answer.error or answer.answer
-        await case.export(archive_path)
+        assert evaluation["success"], evaluation
+        await case.export_evaluation(archive_path)
     finally:
         await case.close()
 
     names, files = _archive_members(archive_path.read_bytes())
-    assert {"manifest.json", "runtime-layout.json", "logs", "results", "reme_workspace"} <= names
-    assert "results/answer.json" in names
-    assert "logs/actions.jsonl" in names
-    assert any(name.startswith("logs/") and name.endswith(".log") for name in names)
-
-    manifest = json.loads(files["manifest.json"])
-    layout = json.loads(files["runtime-layout.json"])
-    assert manifest["case_id"] == f"longmemeval-{item['question_id']}-smoke"
-    assert manifest["export_profile"] == "analysis"
-    assert layout["workspace_root"] == "reme_workspace"
-    assert layout["configured_paths"]["daily_dir"] == "reme_workspace/daily"
-    assert layout["configured_paths"]["mem_session_dir"] == "reme_workspace/mem_session"
+    query_root = f"queries/{item['question_id']}"
+    assert {"reme_workspace", "build_log", "queries"} <= names
+    assert "build_log/build.log" in names
+    assert f"{query_root}/answer.log" in names
+    assert f"{query_root}/result.json" in names
+    assert "queries/summary.json" in names
+    summary = json.loads(files["queries/summary.json"])
+    assert summary["queries"][0]["query_id"] == str(item["question_id"])
+    assert "directory" not in summary["queries"][0]
 
     assert any(name.startswith("reme_workspace/daily/") and name.endswith(".md") for name in names)
     assert "reme_workspace/.git/HEAD" in names
@@ -132,8 +150,8 @@ async def test_first_longmemeval_case_analysis_export(tmp_path):
     excluded_prefixes = (
         "inbox",
         "tmp",
-        "reme_workspace/metadata",
-        "reme_workspace/resource",
+        "logs",
+        "results",
     )
     assert not any(name == prefix or name.startswith(f"{prefix}/") for prefix in excluded_prefixes for name in names)
     assert not any(".reme" in PurePosixPath(name).parts for name in names)

tests/unit/test_logging_config.py

@@ -85,6 +85,24 @@ def test_stdlib_file_creation_is_delayed_until_first_record(monkeypatch, tmp_pat
             handler.close()
 
 
+@pytest.mark.parametrize("use_loguru", [True, False])
+def test_explicit_log_filepath_uses_exact_non_rotating_file(monkeypatch, tmp_path, use_loguru):
+    """Sandbox phase logging can switch one shared logger to an exact artifact path."""
+    monkeypatch.setenv("REME_DISABLE_LOGURU", "false" if use_loguru else "true")
+    path = tmp_path / "queries" / "query:1" / "answer.log"
+
+    logger = logger_utils.get_logger(
+        log_to_console=False,
+        log_to_file=True,
+        force_init=True,
+        log_filepath=str(path),
+    )
+    logger.info("query-only")
+    logger_utils.get_logger(log_to_console=False, log_to_file=False, force_init=True)
+
+    assert path.read_text(encoding="utf-8").endswith("query-only\n")
+
+
 def test_stdlib_forwards_screen_and_file_logs_to_qwenpaw(monkeypatch, tmp_path):
     """Embedded stdlib logging should reuse QwenPaw's active sinks."""
     monkeypatch.chdir(tmp_path)

tests/unit_sandbox/test_meta_reme_run.py

@@ -7,6 +7,7 @@ from __future__ import annotations
 import importlib.util
 import json
 from pathlib import Path
+import subprocess
 import sys
 
 import yaml
@@ -21,6 +22,16 @@ SPEC.loader.exec_module(run)
 from data_preparation import beam  # noqa: E402  pylint: disable=wrong-import-position
 
 
+def _git(repository: Path, *arguments: str) -> str:
+    return subprocess.run(
+        ["git", *arguments],
+        cwd=repository,
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+
+
 def _lme_item(case_id: str) -> dict:
     return {
         "question_id": case_id,
@@ -58,6 +69,12 @@ def test_prepare_workspace_stores_only_selected_longmemeval_cases(tmp_path: Path
     stored_text = "".join(path.read_text(encoding="utf-8") for path in case_files)
     assert "held-out" not in stored_text
     assert not case_files[0].stat().st_mode & 0o222
+    assert workspace.path("code/repo/reme/pyproject.toml").is_file()
+    assert workspace.path("code/repo/reme/reme/config/lme.yaml").is_file()
+    code_repository = workspace.path("code/repo/reme")
+    assert _git(code_repository, "branch", "--show-current") == "main"
+    assert _git(code_repository, "log", "-1", "--pretty=%s") == "Initial version"
+    assert _git(code_repository, "status", "--short") == ""
 
     assert (
         run.prepare_workspace(
@@ -80,6 +97,22 @@ def test_prepare_workspace_initializes_an_existing_empty_directory(tmp_path: Pat
 
     assert workspace.path(".meta-reme-workspace.json").is_file()
     assert workspace.path("datasets/search/manifest.json").is_file()
+    assert workspace.path("code/repo/reme/reme/steps/benchmark/lme").is_dir()
+
+
+def test_prepare_workspace_does_not_overwrite_existing_initial_code(tmp_path: Path) -> None:
+    source = tmp_path / "longmemeval.json"
+    source.write_text(json.dumps([_lme_item("train")]), encoding="utf-8")
+    workspace = run.prepare_workspace(tmp_path / "meta-workspace", "longmemeval", dataset_source=source)
+    initial_head = _git(workspace.path("code/repo/reme"), "rev-parse", "HEAD")
+    marker = workspace.path("code/repo/reme/user-change.txt")
+    marker.write_text("keep me", encoding="utf-8")
+
+    reopened = run.prepare_workspace(workspace.root, "longmemeval", dataset_source=source)
+
+    assert reopened.root == workspace.root
+    assert _git(workspace.path("code/repo/reme"), "rev-parse", "HEAD") == initial_head
+    assert marker.read_text(encoding="utf-8") == "keep me"
 
 
 def test_load_beam_cases_respects_variant_and_selection(tmp_path: Path) -> None:

tests/unit_sandbox/test_sandbox.py

@@ -13,7 +13,14 @@ import pytest
 from reme.enumeration import ComponentEnum
 from reme.schema import ApplicationConfig
 from reme.utils import global_counter_add
-from sandbox import DockerReMeSandboxFactory, ImageCandidate, SourceCandidate, SourceSnapshot
+from sandbox import (
+    DockerReMeSandboxFactory,
+    EvaluationQuery,
+    ImageCandidate,
+    JobRequest,
+    SourceCandidate,
+    SourceSnapshot,
+)
 from sandbox.direct_docker_workspace import DirectDockerWorkspace
 from sandbox import worker
 
@@ -64,7 +71,9 @@ class FakeBackend:
             stdout = b'{"reme": "/workspace/candidate/reme/__init__.py", "agentscope": "2.0.4.post1"}\n'
             return FakeExecResult(stdout=stdout)
         if "/workspace/harness/worker.py" in command:
+            request_path = command[command.index("--request") + 1]
             response_path = command[command.index("--response") + 1]
+            request = json.loads(self.files[request_path])
             self.files["/workspace/case/runtime-layout.json"] = json.dumps(
                 {
                     "workspace_root": "reme_workspace",
@@ -80,8 +89,23 @@ class FakeBackend:
                     "analysis_excludes": ["reme_workspace/metadata", "reme_workspace/resource"],
                 },
             ).encode()
-            self.files[response_path] = json.dumps(
-                {
+            if request.get("mode") == "build":
+                self.files["/workspace/case/build_log/build.log"] = b"build"
+                response = {
+                    "success": True,
+                    "jobs": [{"job": item["job"], "result": {"success": True}} for item in request["jobs"]],
+                }
+            elif request.get("mode") == "queries":
+                self.files["/workspace/case/queries/summary.json"] = b'{"queries": []}'
+                response = {
+                    "success": True,
+                    "summary": {
+                        "queries": [{"query_id": item["query_id"], "score": 1.0} for item in request["queries"]],
+                    },
+                    "queries": [],
+                }
+            else:
+                response = {
                     "success": True,
                     "answer": "yes",
                     "metadata": {"job": "ok"},
@@ -89,8 +113,8 @@ class FakeBackend:
                         "bench": {"input_tokens": 12, "output_tokens": 3, "total_tokens": 15},
                     },
                     "error": None,
-                },
-            ).encode()
+                }
+            self.files[response_path] = json.dumps(response).encode()
         if command[:2] == ["tar", "-czf"]:
             self.files[command[2]] = b"case-archive"
         return FakeExecResult()
@@ -230,6 +254,85 @@ async def test_image_candidate_skips_source_upload_and_runs_jobs(tmp_path):
     assert "--exclude=reme_workspace/session" not in export_command
 
 
+@pytest.mark.asyncio
+async def test_build_and_queries_use_batch_worker_requests(tmp_path):
+    """Build and query batches each use one worker process and preserve verbatim query IDs."""
+    workspace = FakeWorkspace("case-1", "candidate:test", {})
+    factory = DockerReMeSandboxFactory(
+        ImageCandidate("candidate:test"),
+        workspace_builder=lambda *_args: workspace,
+    )
+    case = await factory.create_case("case-1")
+
+    build = await case.run_build(
+        [
+            JobRequest("auto_memory", {"session_id": "session-1"}),
+            JobRequest("index_update"),
+        ],
+    )
+    queries = await case.run_queries(
+        [
+            EvaluationQuery(
+                query_id="information_extraction:1",
+                question="What happened?",
+                golden_answer="An event",
+                judge_arguments={"query": "What happened?", "golden_answer": "An event"},
+            ),
+            EvaluationQuery(
+                query_id="information_extraction:2",
+                question="When?",
+                golden_answer="Yesterday",
+                judge_arguments={"query": "When?", "golden_answer": "Yesterday"},
+            ),
+        ],
+    )
+
+    assert build["success"] is True
+    assert [item["query_id"] for item in queries["summary"]["queries"]] == [
+        "information_extraction:1",
+        "information_extraction:2",
+    ]
+    worker_commands = [
+        command for command, _, _ in workspace.backend.commands if "/workspace/harness/worker.py" in command
+    ]
+    assert len(worker_commands) == 2
+
+    exported = await case.export_evaluation(tmp_path / "evaluation.tar.gz")
+    assert exported.read_bytes() == b"case-archive"
+    export_command = workspace.backend.commands[-1][0]
+    assert export_command == [
+        "tar",
+        "-czf",
+        "/tmp/reme-case-export.tar.gz",
+        "-C",
+        "/workspace/case",
+        "reme_workspace",
+        "build_log",
+        "queries",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_query_artifact_ids_reject_path_components():
+    """Verbatim query directory names cannot escape the queries root."""
+    workspace = FakeWorkspace("case-1", "candidate:test", {})
+    factory = DockerReMeSandboxFactory(
+        ImageCandidate("candidate:test"),
+        workspace_builder=lambda *_args: workspace,
+    )
+    case = await factory.create_case("case-1")
+
+    with pytest.raises(ValueError, match="safe directory"):
+        await case.run_queries(
+            [EvaluationQuery(query_id="../escape", question="q", golden_answer="a")],
+        )
+
+    with pytest.raises(ValueError, match="conflicts"):
+        await case.run_queries(
+            [EvaluationQuery(query_id="summary.json", question="q", golden_answer="a")],
+        )
+
+
 @pytest.mark.asyncio
 async def test_workspace_can_be_cleared_and_uploaded_from_host(tmp_path):
     """A completed memory can be copied into an independent query case."""
@@ -641,6 +744,96 @@ async def test_worker_calls_application_directly_and_always_closes(monkeypatch,
     ]
 
 
+@pytest.mark.asyncio
+async def test_worker_reuses_one_application_per_phase_and_isolates_query_logs(monkeypatch, tmp_path):
+    """Construction and all queries use two Applications while every query gets only its own log."""
+    application_ids = []
+
+    class FakeReMe:
+        """Minimal multi-job Application whose log messages expose phase leakage."""
+
+        def __init__(self, **_config):
+            self.application_id = len(application_ids) + 1
+            application_ids.append(self.application_id)
+            self.context = SimpleNamespace(metadata={}, components={})
+
+        async def start(self):
+            """Start without external resources."""
+
+        async def run_job(self, name, **arguments):
+            """Log a unique marker and return deterministic answer/judge results."""
+            from reme.utils import get_logger
+
+            marker = arguments.get("query", arguments.get("session_id", "judge"))
+            get_logger().info(f"job={name} marker={marker}")
+            answer = "yes" if name == "answer_judge" else f"answer:{marker}"
+            return SimpleNamespace(success=True, answer=answer, metadata={"job": name})
+
+        async def close(self):
+            """Close without external resources."""
+
+    monkeypatch.setattr("reme.config.resolve_app_config", lambda **kwargs: kwargs)
+    monkeypatch.setattr("reme.reme.ReMe", FakeReMe)
+    common = {
+        "config": "lme.yaml",
+        "case_id": "case-1",
+        "case_root": str(tmp_path),
+        "workspace_dir": str(tmp_path / "workspace"),
+    }
+
+    build = await worker._run_build(  # pylint: disable=protected-access
+        {
+            **common,
+            "jobs": [
+                {"job": "auto_memory", "arguments": {"session_id": "construction-only"}},
+                {"job": "index_update", "arguments": {}},
+            ],
+        },
+    )
+    queries = await worker._run_queries(  # pylint: disable=protected-access
+        {
+            **common,
+            "queries": [
+                {
+                    "query_id": "query:1",
+                    "question": "first-only",
+                    "golden_answer": "a",
+                    "judge_arguments": {},
+                    "score_mapping": {"yes": 1.0, "no": 0.0},
+                },
+                {
+                    "query_id": "query:2",
+                    "question": "second-only",
+                    "golden_answer": "b",
+                    "judge_arguments": {},
+                    "score_mapping": {"yes": 1.0, "no": 0.0},
+                },
+            ],
+        },
+    )
+
+    assert build["success"] is True
+    assert queries["success"] is True
+    assert application_ids == [1, 2]
+    build_log = (tmp_path / "build_log/build.log").read_text(encoding="utf-8")
+    first_log = (tmp_path / "queries/query:1/answer.log").read_text(encoding="utf-8")
+    second_log = (tmp_path / "queries/query:2/answer.log").read_text(encoding="utf-8")
+    assert "construction-only" in build_log
+    assert "first-only" not in build_log
+    assert "first-only" in first_log
+    assert "second-only" not in first_log
+    assert "construction-only" not in first_log
+    assert "second-only" in second_log
+    assert "first-only" not in second_log
+
+    summary = json.loads((tmp_path / "queries/summary.json").read_text(encoding="utf-8"))
+    assert summary["queries"] == [
+        {"query_id": "query:1", "score": 1.0},
+        {"query_id": "query:2", "score": 1.0},
+    ]
+    assert all("directory" not in item for item in summary["queries"])
+
+
 @pytest.mark.asyncio
 async def test_worker_returns_usage_accumulated_before_job_failure(monkeypatch, tmp_path):
     """A failed step still reports tokens spent before the exception."""