Add initial validation and canonicalize workspace layout

This commit is contained in:
酱牛肉 2026-08-08 13:22:40 +08:00
parent 8513aaf591
commit 55d9812321
18 changed files with 1409 additions and 57 deletions

View file

@ -5,3 +5,6 @@ dataset:
name: "longmemeval"
source: "benchmark/longmemeval/dataset/longmemeval_s_reme_cleaned.json"
train_case_ids: [] # empty = all cases; or a list of question IDs
validation:
concurrency: 5

View file

@ -57,45 +57,58 @@ meta-reme/
## 4. Workspace 设计
一次搜索的全部状态保存在用户指定 workspace:
一次搜索的全部状态保存在用户指定 workspace。下图是 canonical 结构;标为“搜索编排发布”的文件由尚未接入当前 preparation/validation 路径的搜索循环生成,其余目录由 `Workspace.create()` 或 validation 按需创建:
```text
workspace/
.meta-reme-workspace.json
.meta-reme.lock # 仅持锁期间存在
domain_spec.yaml
run.json
events.jsonl
leaderboard.json
run.json # 搜索编排发布
events.jsonl # 搜索编排发布
leaderboard.json # 搜索编排发布
code/
repo/
reme/ # Git 仓库;初始 branch/code ID 为 init
worktrees/
datasets/
search/
manifest.json
cases/
harnesses/
registry.jsonl
<commit_sha>/
manifest.json
summary.md
dataset/
manifest.json
cases/
weaknesses/
proposals/
evaluations/
<commit_sha>/
search.json
debug/
cases/
<code_id>/
<validation_id>/
manifest.json
summary.json
cases/
<case_id>/
attempt-<n>/
case_result.json
memory_construction/
result.json
reme_workspace.tar.gz
reme_workspace/
build.log
queries/
result.json
<query_id>/
answer.log
result.json
logs/
```
search set 被转换为统一 JSON 格式复制到 workspace,并设置为只读。Test set 不复制到该目录,也不向 agent 暴露;只有最佳 harness 冻结后,外部 evaluator 才加载 test set。
`meta-reme/run.py` 完成 dataset 安装和初始 bundle 创建后,立即使用 `init` branch 对已安装的全部 case 执行第一次 validation。结果固定写入 `evaluations/init/initial/`,并发数由配置中的 `validation.concurrency` 控制;该 summary 已存在时不重复执行。
`run.json` 保存运行状态、baseline、当前轮次、预算和最佳候选。`events.jsonl` 采用 append-only 方式记录 proposal、commit、debug、validation、失败和 selection 等事件,使搜索可以在进程中断后恢复。
## 5. Baseline Bundle
`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,在 `main` 分支创建 message 为 `Initial version` 的不可变 baseline commit。
Bundle 不能依赖人工随意删除文件,而应由 domain spec 中的文件清单确定性生成。生成后执行 import、组件注册、安装和固定 smoke case 检查,并验证精简版本与完整 ReMe baseline 行为一致。随后在 workspace 中初始化 Git,在 code ID 为 `init` 的同名分支创建 message 为 `Initial version` 的不可变 baseline commit。
### 5.1 Bundle Target
@ -145,7 +158,8 @@ build_bundle(
## 6. Harness 与 Git 管理
Branch 表示一条可继续演化的 proposal 线,commit SHA 表示一个不可变 harness。Agent 可以:
Branch 表示一条可继续演化的 proposal 线,branch name 就是 code ID;commit SHA 表示该 code ID 在某次 validation
开始时冻结的不可变 harness。用于结果目录的 code ID 必须是无 `/` 的路径安全 branch name。Agent 可以:
- 从任意历史 commit fork;
- 在独立 Git worktree 中修改;
@ -186,23 +200,23 @@ Agent 可以自主选择完整 case、指定 session、session prefix 或少量
除 debug 外,validation 还支持显式传入 case ID 列表,并可为每个 case 指定 query ID 子集。该能力用于在全量评测代价过高时先执行 targeted screening,例如优先验证 weakness report 中的失败 query 和少量回归 case。筛选范围必须在运行前固化为 `ValidationSpec`,记录选择理由及其 fingerprint;不得在看到本次得分后追加或替换 query 来美化结果。
现有 sandbox 需要补充失败导出能力:即使在安装、初始化或 ingest 阶段失败,也应 best-effort 导出 `failure.json`、stdout、stderr、Python traceback、action log、已完成阶段和宿主异常。每次运行结束后,宿主下载原始压缩包,并安全解压到:
Sandbox 在 memory construction 结束、query 开始前,先导出 construction 结果、`build.log` 和当时完整、可直接通过 `upload_workspace()` 回灌的 `reme_workspace.tar.gz`;query 阶段结束后再单独导出 query summary、逐 query 日志和结构化结果。两个阶段不得合并成一个仅在所有 query 结束后生成的归档。失败时应 best-effort 导出 `failure.json`、stdout、stderr、Python traceback、action log、已完成阶段和宿主异常。每次运行的宿主目录为:
```text
evaluations/<commit>/<mode>/cases/<case>/attempt-<n>/
evaluations/<code_id>/<validation_id>/cases/<case>/attempt-<n>/
```
解压过程需要拒绝绝对路径、`../`、symlink、设备文件和异常大的压缩包,同时保存 tar 包和 SHA256,确保 artifacts 可验证、可复查。
其中 `memory_construction/` 保存结构化结果、build 日志和可复用的 workspace 压缩包;`queries/` 保存结构化结果及安全解压后的逐 query artifacts。Query 传输使用的临时压缩包必须在解压后删除。解压过程需要拒绝绝对路径、`../`、symlink、设备文件和异常大的压缩包;`reme_workspace.tar.gz` 保存 SHA256,确保可验证、可复用。
## 10. Validation 与结果记录
Validation 分为 targeted screening 和 full validation。两者都必须从 frozen commit 创建全新 SourceCandidate,并使用同一套 sandbox、失败分类和结果 schema。不同 harness、case 之间使用独立 runtime workspace;同一 case 本次选中的多个 query 应从相同的 post-ingestion memory snapshot 启动,避免 query 顺序造成状态污染。
`ValidationSpec` 明确记录目标 commit、mode、选中的 case/query、选择理由和 dataset/code/config/model/image fingerprint。未指定 case 表示覆盖全部 case,某个 case 未指定 query 子集表示覆盖该 case 的全部 query。系统在执行前根据 dataset manifest 展开并固化选择结果;空集合、未知 ID、重复 ID,以及不属于所选 case 的 query 均应直接报错。
`ValidationSpec` 明确记录目标 code ID、validation 开始时解析出的 commit、mode、选中的 case/query、选择理由和 dataset/code/config/model/image fingerprint。未指定 case 表示覆盖全部 case,某个 case 未指定 query 子集表示覆盖该 case 的全部 query。系统在执行前根据 dataset manifest 展开并固化选择结果;空集合、未知 ID、重复 ID,以及不属于所选 case 的 query 均应直接报错。
部分 validation 的聚合结果必须同时记录已选数量、数据集总数量、覆盖率和 `is_full=false`。其平均分只描述本次固定子集,不得与其他选择范围的分数直接比较,也不得进入正式 leaderboard 或用于选择 winner。编排器可以依据预先配置的筛选规则决定淘汰候选或将其升级为 full validation;只有覆盖完整 search set 且没有 `infra_error` 的结果才具有正式可比性。Baseline 至少执行一次 full validation,最终 winner 仍需进行第 11 节规定的 clean replay。
每次 targeted screening 的聚合结果写入 `evaluations/<commit>/screening/<validation_id>.json`;完整 search validation 仍发布为 harness 级 `evaluations/<commit>/search.json`。两类结果均保存:
每次 validation 均写入 `evaluations/<code_id>/<validation_id>/`;manifest 记录本次运行的输入与 fingerprint,summary 保存该次固定选择范围的聚合结果。需要提供 code 级快捷索引或 leaderboard 时,只能从已完成且可比较的 summary 确定性生成,不得作为另一份事实来源。所有结果均保存 code ID 和执行时冻结的 commit SHA:
- commit、parents、changed paths 和 scope 结果;
- dataset、代码、配置、模型和镜像 fingerprint;
@ -252,7 +266,7 @@ Targeted screening 是可选的成本控制阶段。未通过筛选的 candidate
完成的 case 结果发布到第 9 节约定的 attempt 目录,并包含 `case_result.json` 和 `complete.json`。`complete.json` 最后写入,记录结果文件及 artifacts 的 SHA256、dataset/code/config/model/image fingerprint、case ID、attempt ID、所选 query 集合的 fingerprint 和完成状态。仅当完成标志可解析、hash 正确、状态为成功或确定性的候选失败,且所有 fingerprint 与当前 validation 完全一致时,结果才可复用;部分 query 的结果不能冒充该 case 的全量结果。目录或某个 `search.json` 文件的存在本身不能作为完成依据。基础设施故障耗尽重试后仍保留为 `infra_error`,但不把 validation 错误地标记为可聚合完成。
`ValidationSpec` 选中的全部 case/query 达到可聚合状态后,编排器确定性地生成对应聚合结果;targeted screening 写入其 validation ID 对应的结果文件,只有 full validation 才生成 harness 级 `evaluations/<commit>/search.json`。随后记录 `validation_completed` 事件。若聚合阶段中断,可从具有相同选择 fingerprint 的 case 结果重新生成,不重复执行已验证的选择范围。
`ValidationSpec` 选中的全部 case/query 达到可聚合状态后,编排器在对应 validation 目录确定性地生成 summary,随后记录 `validation_completed` 事件。若聚合阶段中断,可从具有相同选择 fingerprint 的 case 结果重新生成,不重复执行已验证的选择范围。
### 12.3 搜索、Proposal 与 Git 续跑

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from pathlib import Path
import subprocess
INITIAL_BRANCH = "main"
INITIAL_BRANCH = "init"
INITIAL_COMMIT_MESSAGE = "Initial version"
GIT_AUTHOR_NAME = "Meta-ReMe"
GIT_AUTHOR_EMAIL = "meta-reme@localhost"

View file

@ -6,7 +6,7 @@ import argparse
import json
from pathlib import Path
import tempfile
from typing import Any, Iterable
from typing import Any, Callable, Iterable
import yaml
@ -21,6 +21,7 @@ from models import (
SandboxSpec,
ScopeSpec,
)
from validation import run_validation
from workspace import Workspace
PROJECT_ROOT = Path(__file__).resolve().parent.parent
@ -29,6 +30,9 @@ DEFAULT_SOURCES = {
"longmemeval": PROJECT_ROOT / "benchmark/longmemeval/dataset/longmemeval_s_reme_cleaned.json",
}
INITIAL_CODE_DIR = Path("code/repo/reme")
INITIAL_CODE_ID = "init"
INITIAL_VALIDATION_ID = "initial"
ValidationRunner = Callable[..., Path]
def _domain_spec(dataset: str, source: Path, source_fingerprint: str) -> DomainSpec:
@ -38,7 +42,7 @@ def _domain_spec(dataset: str, source: Path, source_fingerprint: str) -> DomainS
benchmark_runner=f"benchmark.{'beam' if dataset == 'beam' else 'longmemeval'}.run",
scorer="mean_query_score",
scope=ScopeSpec(),
sandbox=SandboxSpec(image="reme:latest", timeout_seconds=3600),
sandbox=SandboxSpec(image="reme-sandbox-base:agentscope-2.0.4-post1", timeout_seconds=3600),
proposer=ProposerSpec(model="not-configured"),
budget=BudgetSpec(max_proposals=1),
)
@ -72,8 +76,8 @@ def prepare_workspace(
else Workspace.open(meta_workspace, spec)
)
with workspace.acquire_lock():
if not workspace.path("datasets/search/manifest.json").exists():
workspace.install_search_dataset(normalized)
if not workspace.path("dataset/manifest.json").exists():
workspace.install_dataset(normalized)
_prepare_initial_code(workspace, spec.bundle_target)
return workspace
@ -93,6 +97,60 @@ def _prepare_initial_code(workspace: Workspace, bundle_target: str) -> Path:
return bundle
def run_initial_validation(
workspace: Workspace,
concurrency: int,
*,
validation_runner: ValidationRunner = run_validation,
) -> Path:
"""Validate the initial code branch once against every installed case."""
if concurrency < 1:
raise ValueError("validation.concurrency must be at least 1")
output = workspace.path(f"evaluations/{INITIAL_CODE_ID}/{INITIAL_VALIDATION_ID}")
if (output / "summary.json").is_file():
return output
case_ids = []
for case_path in sorted(workspace.path("dataset/cases").glob("*.json")):
case = json.loads(case_path.read_text(encoding="utf-8"))
case_id = case.get("case_id")
if not isinstance(case_id, str) or not case_id:
raise ValueError(f"Prepared case has no valid case_id: {case_path}")
case_ids.append(case_id)
if not case_ids:
raise ValueError("Prepared dataset has no cases to validate")
return validation_runner(
workspace.root,
case_ids,
INITIAL_CODE_ID,
concurrency,
validation_id=INITIAL_VALIDATION_ID,
)
def prepare_and_validate_workspace(
config_path: Path,
*,
validation_runner: ValidationRunner = run_validation,
) -> tuple[Workspace, Path]:
"""Prepare the configured workspace and run its first full validation."""
config = load_config(config_path)
workspace = prepare_workspace(
meta_workspace=config["meta_workspace"],
dataset=config["dataset"],
train_case_ids=config["train_case_ids"],
dataset_variant=config["dataset_variant"],
dataset_source=config["dataset_source"],
)
validation = run_initial_validation(
workspace,
config["validation_concurrency"],
validation_runner=validation_runner,
)
return workspace, validation
def _mapping(value: Any, label: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise ValueError(f"{label} must be a YAML mapping")
@ -119,12 +177,22 @@ def load_config(config_path: Path) -> dict[str, Any]:
raise ValueError("dataset.train_case_ids must be a list of strings or integers")
source_value = dataset_config.get("source")
source = DEFAULT_SOURCES[dataset] if source_value is None else _project_path(source_value, "dataset.source")
validation_config = raw.get("validation", {})
validation_config = _mapping(validation_config, "validation")
validation_concurrency = validation_config.get("concurrency", 1)
if (
not isinstance(validation_concurrency, int)
or isinstance(validation_concurrency, bool)
or validation_concurrency < 1
):
raise ValueError("validation.concurrency must be a positive integer")
return {
"meta_workspace": _project_path(raw.get("meta_workspace"), "meta_workspace"),
"dataset": dataset,
"train_case_ids": [str(case_id) for case_id in case_ids],
"dataset_variant": dataset_config.get("variant"),
"dataset_source": source,
"validation_concurrency": validation_concurrency,
}
@ -137,14 +205,15 @@ def parse_args() -> argparse.Namespace:
def main() -> None:
"""Prepare the configured workspace, including its training data and initial code."""
"""Prepare the configured workspace and validate all data with its initial code."""
args = parse_args()
workspace = prepare_workspace(**load_config(args.config))
manifest = json.loads(workspace.path("datasets/search/manifest.json").read_text(encoding="utf-8"))
workspace, validation = prepare_and_validate_workspace(args.config)
manifest = json.loads(workspace.path("dataset/manifest.json").read_text(encoding="utf-8"))
print(
f"Meta-ReMe workspace ready: {workspace.root} "
f"({manifest['case_count']} training cases, {manifest['query_count']} queries)",
f"({manifest['case_count']} training cases, {manifest['query_count']} queries); "
f"initial validation: {validation}",
)

View file

@ -0,0 +1,77 @@
# Meta-ReMe validation
Running `meta-reme/run.py` prepares the configured dataset and initial bundle,
then automatically validates the `init` branch against every installed case.
The first result is stored under `evaluations/init/initial/`; a completed result
is reused on later invocations. Configure the full-case concurrency with
`validation.concurrency` in `config_meta_reme.yaml`.
Validate prepared workspace cases against an immutable code revision:
```bash
python meta-reme/validation/run.py \
--workspace /path/to/workspace \
--case-id case-1 \
--case-id case-2 \
--code-id <branch-name> \
--concurrency 2
```
`--case-id` may be repeated. The code ID is the name of a local Git branch; commit
hashes and tags are not accepted. The branch is resolved to a full Git commit before
execution, and the candidate snapshot is built from that commit rather than the
current working tree. Code IDs must be path-safe branch names (for example,
`candidate-001`, without `/`) because the same value keys the result directory.
Results are written without overwriting earlier runs:
```text
<workspace>/evaluations/<code-id>/<validation-id>/
```
Each case attempt stores the two execution phases independently. Memory
construction is exported before queries begin, so its workspace is an exact
post-construction snapshot rather than the workspace left after evaluation:
```text
cases/<case-id>/attempt-<n>/
case_result.json
memory_construction/
result.json
reme_workspace.tar.gz
reme_workspace/
build.log
queries/
result.json
<query-id>/
answer.log
result.json
```
`reme_workspace.tar.gz` is produced by the sandbox workspace export API and can
be passed directly to `upload_workspace()` on a fresh sandbox case before
running more queries. The same snapshot is safely extracted beside the archive
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.
The same workflow is available as a project API. Use the synchronous entry point
from regular code:
```python
from validation import run_validation
result_dir = run_validation(workspace, case_ids, code_id, concurrency=2)
```
Applications that already run an event loop should use the native async entry
point instead:
```python
from validation import run_validation_async
result_dir = await run_validation_async(workspace, case_ids, code_id, concurrency=2)
```

View file

@ -0,0 +1,5 @@
"""Run reproducible Meta-ReMe validations in isolated case sandboxes."""
from .evaluator import ValidationError, run_validation, run_validation_async
__all__ = ["ValidationError", "run_validation", "run_validation_async"]

View file

@ -0,0 +1,6 @@
"""Allow ``python -m validation`` to invoke the validation CLI."""
from .run import main
if __name__ == "__main__":
main()

View file

@ -0,0 +1,110 @@
"""Translate normalized Meta-ReMe cases into sandbox requests."""
from __future__ import annotations
from datetime import datetime
import re
from models import CaseSpec, DatasetName, QuerySpec, SessionSpec
from sandbox import EvaluationQuery, JobRequest
class AdapterError(ValueError):
"""Raised when normalized benchmark data cannot form sandbox requests."""
def build_jobs(dataset: DatasetName, case: CaseSpec) -> list[JobRequest]:
"""Construct memory in time order and refresh the index after every session."""
jobs: list[JobRequest] = []
sessions = sorted(case.sessions, key=_session_timestamp)
if dataset is DatasetName.LONGMEMEVAL:
question_dates = [query.metadata.get("question_date") for query in case.queries]
if any(question_dates):
if len(case.queries) != 1 or not question_dates[0]:
raise AdapterError(f"LongMemEval case {case.case_id!r} requires one question_date")
query_time = _longmemeval_iso(str(question_dates[0]))
sessions = [session for session in sessions if _session_timestamp(session) <= query_time]
for session in sessions:
jobs.extend(
[
JobRequest(
"auto_memory",
{
"session_id": session.session_id,
"messages": session.messages,
"date": _session_timestamp(session)[:10],
},
),
JobRequest("index_update"),
],
)
if not jobs:
raise AdapterError(f"case {case.case_id!r} has no sessions to construct")
return jobs
def _session_timestamp(session: SessionSpec) -> str:
timestamps = [message.get("created_at") for message in session.messages if message.get("created_at")]
if not timestamps or not isinstance(timestamps[0], str):
raise AdapterError(f"session {session.session_id!r} has no created_at timestamp")
return timestamps[0]
def evaluation_queries(dataset: DatasetName, case: CaseSpec) -> list[EvaluationQuery]:
"""Create dataset-specific answer and judge requests for one case."""
if dataset is DatasetName.LONGMEMEVAL:
return [_longmemeval_query(query) for query in case.queries]
if dataset is DatasetName.BEAM:
return [_beam_query(query) for query in case.queries]
raise AdapterError(f"unsupported validation dataset: {dataset}")
def _longmemeval_query(query: QuerySpec) -> EvaluationQuery:
metadata = query.metadata
question_type = str(metadata.get("question_type", ""))
answer_arguments: dict[str, str] = {}
question_date = metadata.get("question_date")
if question_date:
answer_arguments["query_time"] = _longmemeval_iso(str(question_date))
return EvaluationQuery(
query_id=query.query_id,
question=query.question,
golden_answer=query.golden_answer,
answer_arguments=answer_arguments,
judge_arguments={
"query": query.question,
"golden_answer": query.golden_answer,
"question_type": question_type,
},
)
def _beam_query(query: QuerySpec) -> EvaluationQuery:
metadata = query.metadata
rubric = metadata.get("rubric", query.golden_answer)
if not isinstance(rubric, list):
raise AdapterError(f"BEAM query {query.query_id!r} has a non-list rubric")
question_type = str(metadata.get("question_type", ""))
return EvaluationQuery(
query_id=query.query_id,
question=query.question,
golden_answer=query.golden_answer,
judge_answer_argument="llm_response",
judge_arguments={
"rubric": rubric,
"probing_question": query.question,
"question_type": question_type,
},
score_path="metadata.llm_judge_score",
score_mapping=None,
)
def _longmemeval_iso(value: str) -> str:
match = re.match(r"(\d{4}/\d{2}/\d{2})\s+\(\w+\)\s+(\d{2}:\d{2})", value)
if not match:
raise AdapterError(f"invalid LongMemEval question_date: {value!r}")
parsed = datetime.strptime(f"{match.group(1)} {match.group(2)}", "%Y/%m/%d %H:%M")
return parsed.strftime("%Y-%m-%dT%H:%M:%S")

View file

@ -0,0 +1,523 @@
"""Concurrent validation orchestration for prepared Meta-ReMe workspaces."""
from __future__ import annotations
import asyncio
from contextlib import contextmanager
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import shutil
import stat
import subprocess
import tarfile
import tempfile
from typing import Any, Callable, Iterator
from uuid import uuid4
import yaml
from models import CaseSpec, DomainSpec, fingerprint, model_fingerprint, utc_now
from workspace import Workspace
from sandbox import DockerReMeSandboxFactory, SourceCandidate, SourceSnapshot
from .adapters import build_jobs, evaluation_queries
SANDBOX_ENV_NAMES = (
"LLM_API_KEY",
"LLM_BASE_URL",
"LLM_BACKEND",
"LLM_MODEL_NAME",
"BENCH_MODEL_NAME",
"JUDGE_MODEL_NAME",
"EMBEDDING_API_KEY",
"EMBEDDING_BASE_URL",
"EMBEDDING_BACKEND",
"EMBEDDING_MODEL_NAME",
)
FactoryBuilder = Callable[..., DockerReMeSandboxFactory]
class ValidationError(RuntimeError):
"""Raised when validation inputs or workspace state are invalid."""
def run_validation(
workspace_root: Path,
case_ids: list[str],
code_id: str,
concurrency: int,
*,
validation_id: str | None = None,
factory_builder: FactoryBuilder = DockerReMeSandboxFactory,
environment: dict[str, str] | None = None,
) -> Path:
"""Synchronously validate cases; use ``run_validation_async`` in async applications."""
return asyncio.run(
run_validation_async(
workspace_root,
case_ids,
code_id,
concurrency,
validation_id=validation_id,
factory_builder=factory_builder,
environment=environment,
),
)
async def run_validation_async(
workspace_root: Path,
case_ids: list[str],
code_id: str,
concurrency: int,
*,
validation_id: str | None = None,
factory_builder: FactoryBuilder = DockerReMeSandboxFactory,
environment: dict[str, str] | None = None,
) -> Path:
"""Asynchronously validate one immutable code revision against prepared cases."""
if concurrency < 1:
raise ValidationError("concurrency must be at least 1")
if not case_ids:
raise ValidationError("at least one case_id is required")
if len(case_ids) != len(set(case_ids)):
raise ValidationError("case_ids must be unique")
_validate_code_id(code_id)
root = Path(workspace_root).resolve()
domain = _load_domain(root)
workspace = Workspace.open(root, domain)
cases = _load_cases(workspace, case_ids)
repository = workspace.path("code/repo/reme")
run_id = validation_id or uuid4().hex
run_root = _validation_root(workspace, code_id, run_id)
with workspace.acquire_lock():
commit_sha = _resolve_branch_commit(repository, code_id)
snapshot_context = _source_snapshot(repository, commit_sha)
with snapshot_context as snapshot:
return await _execute_validation(
workspace,
run_root,
domain,
cases,
code_id,
commit_sha,
run_id,
concurrency,
factory_builder,
environment,
snapshot,
)
async def _execute_validation(
workspace: Workspace,
run_root: Path,
domain: DomainSpec,
cases: list[CaseSpec],
code_id: str,
commit_sha: str,
run_id: str,
concurrency: int,
factory_builder: FactoryBuilder,
environment: dict[str, str] | None,
snapshot: SourceSnapshot,
) -> Path:
"""Run validation after resolving and freezing the branch-backed code revision."""
candidate = SourceCandidate(snapshot, base_image=domain.sandbox.image)
env = dict(environment) if environment is not None else _sandbox_environment()
fingerprints = {
"dataset": _dataset_fingerprint(workspace),
"code": snapshot.sha256,
"config": model_fingerprint(domain),
"model": fingerprint({name: env.get(name) for name in SANDBOX_ENV_NAMES if "MODEL" in name}),
"image": fingerprint(domain.sandbox.image),
}
run_root.mkdir(parents=True, exist_ok=False)
workspace.atomic_write_json(
run_root.relative_to(workspace.root) / "manifest.json",
{
"schema_version": 1,
"validation_id": run_id,
"code_id": code_id,
"commit_sha": commit_sha,
"case_ids": [case.case_id for case in cases],
"concurrency": concurrency,
"dataset": domain.dataset.name.value,
"fingerprints": fingerprints,
"started_at": utc_now().isoformat(),
},
)
factory = factory_builder(
candidate,
env=env,
config=f"{domain.bundle_target}.yaml",
command_timeout=domain.sandbox.timeout_seconds,
)
results = await _run_cases(workspace, run_root, factory, domain, cases, concurrency)
summary = _summarize(run_id, code_id, commit_sha, [case.case_id for case in cases], results, fingerprints)
workspace.atomic_write_json(run_root.relative_to(workspace.root) / "summary.json", summary)
return run_root
async def _run_cases(
workspace: Workspace,
run_root: Path,
factory: DockerReMeSandboxFactory,
domain: DomainSpec,
cases: list[CaseSpec],
concurrency: int,
) -> list[dict[str, Any]]:
semaphore = asyncio.Semaphore(concurrency)
async def run(case: CaseSpec) -> dict[str, Any]:
async with semaphore:
return await _run_case_with_retries(workspace, run_root, factory, domain, case)
return list(await asyncio.gather(*(run(case) for case in cases)))
async def _run_case_with_retries(
workspace: Workspace,
run_root: Path,
factory: DockerReMeSandboxFactory,
domain: DomainSpec,
case: CaseSpec,
) -> dict[str, Any]:
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)
last_result = result
if result["status"] != "infra_error":
return result
assert last_result is not None
return last_result
async def _run_case(
workspace: Workspace,
run_root: Path,
factory: DockerReMeSandboxFactory,
domain: DomainSpec,
case_spec: CaseSpec,
attempt_number: int,
) -> dict[str, Any]:
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] = []
try:
sandbox_case = await factory.create_case(case_spec.case_id)
completed_stages.append("prepare")
build = await 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")
_extract_artifacts(
memory_workspace,
memory_root / "reme_workspace",
domain.sandbox.max_artifact_bytes,
)
await sandbox_case.export_build_log(memory_root / "build.log")
if not build.get("success"):
result = {
"case_id": case_spec.case_id,
"attempt": attempt_number,
"status": "candidate_failure",
"completed_stages": completed_stages,
"build": build,
"queries": [],
"error": "memory construction failed",
"artifact_sha256": {
"memory_workspace": _sha256_file(memory_workspace),
},
}
await _best_effort_export(sandbox_case, attempt_root / "full.tar.gz")
return _publish_case_result(workspace, attempt_root, result)
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)
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")
result = {
"case_id": case_spec.case_id,
"attempt": attempt_number,
"status": "infra_error",
"completed_stages": completed_stages,
"build": None,
"queries": [],
"error": f"{type(exc).__name__}: {exc}",
}
workspace.atomic_write_json(attempt_root.relative_to(workspace.root) / "failure.json", result)
return _publish_case_result(workspace, attempt_root, result)
finally:
if sandbox_case is not None:
await sandbox_case.close()
async def _best_effort_export(sandbox_case: Any, destination: Path) -> None:
try:
await sandbox_case.export_full(destination)
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
def _summarize(
validation_id: str,
code_id: str,
commit_sha: str,
requested_case_ids: list[str],
results: list[dict[str, Any]],
fingerprints: dict[str, str],
) -> dict[str, Any]:
query_results = [query for result in results for query in result.get("queries", [])]
scores = [float(query["score"]) if query.get("score") is not None else 0.0 for query in query_results]
infra_errors = sum(result["status"] == "infra_error" for result in results)
candidate_failures = sum(result["status"] == "candidate_failure" for result in results)
return {
"schema_version": 1,
"validation_id": validation_id,
"code_id": code_id,
"commit_sha": commit_sha,
"status": "infra_error" if infra_errors else "completed",
"case_ids": requested_case_ids,
"case_count": len(results),
"query_count": len(query_results),
"mean_query_score": sum(scores) / len(scores) if scores else None,
"infra_error_count": infra_errors,
"candidate_failure_count": candidate_failures,
"fingerprints": fingerprints,
"cases": results,
"completed_at": utc_now().isoformat(),
}
def _load_domain(root: Path) -> DomainSpec:
path = root / "domain_spec.yaml"
try:
return DomainSpec.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
except (OSError, ValueError) as exc:
raise ValidationError(f"cannot load workspace domain spec: {exc}") from exc
def _load_cases(workspace: Workspace, requested_ids: list[str]) -> list[CaseSpec]:
found: dict[str, CaseSpec] = {}
for path in sorted(workspace.path("dataset/cases").glob("*.json")):
case = CaseSpec.model_validate_json(path.read_text(encoding="utf-8"))
if case.case_id in requested_ids:
if case.case_id in found:
raise ValidationError(f"duplicate prepared case_id: {case.case_id}")
found[case.case_id] = case
unknown = [case_id for case_id in requested_ids if case_id not in found]
if unknown:
raise ValidationError(f"unknown prepared case_ids: {', '.join(unknown)}")
return [found[case_id] for case_id in requested_ids]
def _resolve_branch_commit(repository: Path, code_id: str) -> str:
_validate_code_id(code_id)
branch_check = subprocess.run(
["git", "check-ref-format", "--branch", code_id],
cwd=repository,
capture_output=True,
text=True,
check=False,
)
if branch_check.returncode:
raise ValidationError(
f"invalid Git branch code_id {code_id!r}: {(branch_check.stderr or branch_check.stdout).strip()}",
)
result = subprocess.run(
["git", "rev-parse", "--verify", "--end-of-options", f"refs/heads/{code_id}^{{commit}}"],
cwd=repository,
capture_output=True,
text=True,
check=False,
)
if result.returncode:
raise ValidationError(f"unknown local branch code_id {code_id!r}: {(result.stderr or result.stdout).strip()}")
return result.stdout.strip()
def _validate_code_id(code_id: str) -> None:
if (
not code_id
or code_id in {".", ".."}
or Path(code_id).name != code_id
or any(character in code_id for character in ("/", "\\", "\x00"))
):
raise ValidationError(f"code_id must be a path-safe Git branch name: {code_id!r}")
@contextmanager
def _source_snapshot(repository: Path, commit_sha: str) -> Iterator[SourceSnapshot]:
with tempfile.TemporaryDirectory(prefix="meta-reme-validation-code-") as temporary:
destination = Path(temporary) / "source"
destination.mkdir()
archive = subprocess.run(
["git", "archive", "--format=tar", commit_sha],
cwd=repository,
capture_output=True,
check=False,
)
if archive.returncode:
raise ValidationError(f"cannot archive code commit {commit_sha}: {archive.stderr.decode(errors='replace')}")
archive_path = Path(temporary) / "source.tar"
archive_path.write_bytes(archive.stdout)
_extract_source_archive(archive_path, destination)
yield SourceSnapshot.from_directory(destination)
def _extract_source_archive(archive_path: Path, destination: Path) -> None:
seen: set[str] = set()
with tarfile.open(archive_path, mode="r:") as archive:
for member in archive.getmembers():
path = PurePosixPath(member.name)
if path.is_absolute() or ".." in path.parts or member.name in seen:
raise ValidationError(f"unsafe path in code archive: {member.name!r}")
seen.add(member.name)
target = destination.joinpath(*path.parts)
if member.isdir():
target.mkdir(parents=True, exist_ok=True)
continue
if not member.isfile():
raise ValidationError(f"unsupported entry in code archive: {member.name!r}")
target.parent.mkdir(parents=True, exist_ok=True)
source = archive.extractfile(member)
if source is None:
raise ValidationError(f"cannot read code archive entry: {member.name!r}")
with target.open("wb") as stream:
shutil.copyfileobj(source, stream)
target.chmod(member.mode & 0o777)
def _extract_artifacts(
archive_path: Path,
destination: Path,
max_bytes: int,
*,
strip_prefix: PurePosixPath | None = None,
excluded_paths: set[PurePosixPath] | None = None,
) -> None:
_validate_archive_size(archive_path, max_bytes)
destination.mkdir(parents=True, exist_ok=True)
total = 0
seen: set[PurePosixPath] = set()
excluded = excluded_paths or set()
with tarfile.open(archive_path, mode="r:gz") as archive:
for member in archive.getmembers():
path = PurePosixPath(member.name)
if path.is_absolute() or ".." in path.parts:
raise ValidationError(f"unsafe artifact path: {member.name!r}")
if member.issym() or member.islnk() or member.isdev():
raise ValidationError(f"unsupported artifact entry: {member.name!r}")
total += member.size
if total > max_bytes:
raise ValidationError(f"expanded artifacts exceed {max_bytes} bytes")
if strip_prefix is not None:
if path == strip_prefix:
continue
try:
path = path.relative_to(strip_prefix)
except ValueError as exc:
raise ValidationError(f"artifact is outside {strip_prefix}: {member.name!r}") from exc
if not path.parts or path in excluded:
continue
if path in seen:
raise ValidationError(f"duplicate artifact path: {member.name!r}")
seen.add(path)
target = destination.joinpath(*path.parts)
if target.exists():
raise ValidationError(f"artifact path already exists: {member.name!r}")
if member.isdir():
target.mkdir(parents=True, exist_ok=True)
elif member.isfile():
target.parent.mkdir(parents=True, exist_ok=True)
source = archive.extractfile(member)
if source is None:
raise ValidationError(f"cannot read artifact entry: {member.name!r}")
with target.open("wb") as stream:
shutil.copyfileobj(source, stream)
target.chmod(member.mode & (stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO))
else:
raise ValidationError(f"unsupported artifact entry: {member.name!r}")
def _validate_archive_size(archive_path: Path, max_bytes: int) -> None:
if archive_path.stat().st_size > max_bytes:
raise ValidationError(f"artifact archive exceeds {max_bytes} bytes")
def _validation_root(workspace: Workspace, code_id: str, validation_id: str) -> Path:
if not validation_id or validation_id in {".", ".."} or Path(validation_id).name != validation_id:
raise ValidationError(f"unsafe validation_id: {validation_id!r}")
return workspace.entity_path(
workspace.entity_path("evaluations", code_id).relative_to(workspace.root),
validation_id,
)
def _dataset_fingerprint(workspace: Workspace) -> str:
manifest = json.loads(workspace.path("dataset/manifest.json").read_text(encoding="utf-8"))
value = manifest.get("normalized_fingerprint")
if not isinstance(value, str) or not value:
raise ValidationError("dataset manifest has no normalized_fingerprint")
return value
def _sandbox_environment() -> dict[str, str]:
return {name: os.environ[name] for name in SANDBOX_ENV_NAMES if os.environ.get(name)}
def _sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()

View file

@ -0,0 +1,57 @@
"""CLI entry point for validating prepared Meta-ReMe cases."""
# Import roots must be installed before importing the hyphenated meta-reme tree.
# pylint: disable=wrong-import-position
from __future__ import annotations
import argparse
from pathlib import Path
import sys
META_REME_ROOT = Path(__file__).resolve().parent.parent
PROJECT_ROOT = META_REME_ROOT.parent
for import_root in (META_REME_ROOT, PROJECT_ROOT):
if str(import_root) not in sys.path:
sys.path.insert(0, str(import_root))
from validation.evaluator import run_validation # noqa: E402
def parse_args() -> argparse.Namespace:
"""Parse validation inputs without consulting mutable workspace state."""
parser = argparse.ArgumentParser(description="Construct memory and validate immutable ReMe code in sandboxes")
parser.add_argument("--workspace", type=Path, required=True, help="Prepared Meta-ReMe workspace")
parser.add_argument(
"--case-id",
action="append",
required=True,
dest="case_ids",
help="Prepared case ID; repeat this option to validate multiple cases",
)
parser.add_argument("--code-id", required=True, help="Path-safe local Git branch name identifying the code")
parser.add_argument("--concurrency", type=int, required=True, help="Maximum concurrent case sandboxes")
parser.add_argument("--validation-id", help="Optional non-overwriting validation run ID")
return parser.parse_args()
def main() -> None:
"""Run validation and report the immutable result directory."""
args = parse_args()
try:
output = run_validation(
args.workspace,
args.case_ids,
args.code_id,
args.concurrency,
validation_id=args.validation_id,
)
except Exception as exc:
raise SystemExit(f"validation failed: {exc}") from exc
print(f"Validation results: {output}")
if __name__ == "__main__":
main()

View file

@ -23,8 +23,7 @@ WORKSPACE_LOCK = ".meta-reme.lock"
_DIRECTORIES = (
"code/repo",
"code/worktrees",
"datasets/search/cases",
"harnesses",
"dataset/cases",
"weaknesses",
"proposals",
"evaluations",
@ -95,6 +94,11 @@ class Workspace:
manifest = WorkspaceManifest.model_validate_json(manifest_path.read_text(encoding="utf-8"))
except (OSError, ValidationError) as exc:
raise WorkspaceFormatError(f"Invalid workspace manifest: {manifest_path}") from exc
legacy_dataset = root / "datasets/search"
if legacy_dataset.is_dir() and not (root / "dataset").exists():
raise WorkspaceFormatError(
"Legacy dataset layout detected at datasets/search; move that directory to dataset before opening",
)
for relative in _DIRECTORIES:
if not (root / relative).is_dir():
raise WorkspaceFormatError(f"Workspace directory is missing: {relative}")
@ -118,7 +122,7 @@ class Workspace:
return resolved
def entity_path(self, base: str | Path, identifier: str) -> Path:
"""Resolve a path for an externally supplied case/commit/attempt ID."""
"""Resolve a path for an externally supplied code/case/attempt ID."""
if not identifier or identifier in {".", ".."} or Path(identifier).name != identifier:
raise WorkspaceError(f"Unsafe workspace identifier: {identifier!r}")
@ -128,8 +132,8 @@ class Workspace:
def case_attempt_dir(
self,
commit_sha: str,
mode: str,
code_id: str,
validation_id: str,
case_id: str,
attempt_id: str,
*,
@ -137,8 +141,8 @@ class Workspace:
) -> Path:
"""Return the canonical attempt directory described by the design."""
current = self.entity_path("evaluations", commit_sha)
for identifier in (mode, "cases", case_id, attempt_id):
current = self.entity_path("evaluations", code_id)
for identifier in (validation_id, "cases", case_id, attempt_id):
current = self.entity_path(current.relative_to(self.root), identifier)
if create:
current.mkdir(parents=True, exist_ok=False)
@ -183,19 +187,19 @@ class Workspace:
lock.acquire()
return lock
def install_search_dataset(self, source: Path) -> Path:
"""Copy a normalized search dataset into an empty workspace and make it read-only."""
def install_dataset(self, source: Path) -> Path:
"""Copy a normalized dataset into the workspace and make it read-only."""
source = Path(source).resolve()
destination = self.path("datasets/search")
destination = self.path("dataset")
if not source.is_dir():
raise WorkspaceError(f"Normalized search dataset is not a directory: {source}")
if any(path.is_file() or path.is_symlink() for path in destination.rglob("*")):
raise WorkspaceError(f"Search dataset has already been installed: {destination}")
raise WorkspaceError(f"Dataset has already been installed: {destination}")
for path in source.rglob("*"):
if path.is_symlink():
raise WorkspaceError(f"Dataset may not contain symbolic links: {path.relative_to(source)}")
temporary = self.path("datasets/.search-installing")
temporary = self.path(".dataset-installing")
if temporary.exists():
raise WorkspaceError(f"Incomplete dataset installation requires inspection: {temporary}")
shutil.copytree(source, temporary, symlinks=True)

View file

@ -197,6 +197,11 @@ assert evaluation["success"]
await case.export_evaluation("artifacts/case-1.tar.gz")
```
Call `export_memory_construction()` immediately after `run_build()` to capture
`reme_workspace/` and `build_log/` before queries run. After `run_queries()`,
call `export_queries()` to capture only `queries/`. `export_evaluation()` remains
available when a combined archive is intentionally needed.
`export_evaluation()` produces exactly three top-level directories:
```text
@ -268,8 +273,9 @@ construction history can be inspected offline without a remote or push. A
and initializes a fresh repository for the next case.
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.
before `close()`: use `export()`, `export_full()`,
`export_build_log()`, `export_memory_construction()`, `export_queries()`,
`export_evaluation()`, or `export_workspace()` according to the required contract.
## Reuse one built memory for parallel queries

View file

@ -189,6 +189,11 @@ assert evaluation["success"]
await case.export_evaluation("artifacts/case-1.tar.gz")
```
如需分别保存两个阶段,应在 `run_build()` 后立即调用
`export_memory_construction()`,冻结 query 执行前的 `reme_workspace/` 和
`build_log/`;在 `run_queries()` 后调用 `export_queries()`,只导出
`queries/`。仅在明确需要合并归档时使用 `export_evaluation()`。
`export_evaluation()` 生成且只生成三个顶层目录:
```text
@ -256,7 +261,8 @@ daily memory 的构建历史。`reset_case()` 会连同旧运行时 workspace
仓库,并为下一个 case 初始化新的仓库。
对于临时 Docker workspace,必须在 `close()` 前下载所有需要保留的产物。根据所需
契约选择 `export()`、`export_full()`、`export_evaluation()` 或
契约选择 `export()`、`export_full()`、`export_build_log()`、
`export_memory_construction()`、`export_queries()`、`export_evaluation()` 或
`export_workspace()`。
## 复用一次构建的 Memory,并行执行 Query

View file

@ -690,6 +690,66 @@ class DockerReMeSandbox:
target.write_bytes(await self.backend.read_file(WORKSPACE_EXPORT_ARCHIVE))
return target
async def export_build_log(self, destination: str | Path) -> Path:
"""Download the memory-construction log without wrapping it in an archive."""
async with self._operation_lock:
assert self.backend is not None, "sandbox is not initialized"
source = f"{CASE_ROOT}/build_log/build.log"
try:
content = await self.backend.read_file(source)
except FileNotFoundError as exc:
raise SandboxCommandError(f"build log export is missing required artifact: {source}") from exc
target = Path(destination).expanduser().resolve()
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(content)
return target
async def export_memory_construction(self, destination: str | Path) -> Path:
"""Download the post-construction workspace and its build log."""
async with self._operation_lock:
assert self.backend is not None, "sandbox is not initialized"
required = f"{CASE_ROOT}/build_log/build.log"
try:
await self.backend.read_file(required)
except FileNotFoundError as exc:
raise SandboxCommandError(
f"memory construction export is missing required artifact: {required}",
) from exc
await self._exec_checked(
"export_memory_construction",
[
"tar",
"-czf",
EXPORT_ARCHIVE,
"-C",
CASE_ROOT,
"reme_workspace",
"build_log",
],
)
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_queries(self, destination: str | Path) -> Path:
"""Download per-query logs, results, and their aggregate summary."""
async with self._operation_lock:
assert self.backend is not None, "sandbox is not initialized"
required = f"{CASE_ROOT}/queries/summary.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_queries",
["tar", "-czf", EXPORT_ARCHIVE, "-C", CASE_ROOT, "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 export_evaluation(self, destination: str | Path) -> Path:
"""Download the workspace, build log, and per-query evaluation artifacts."""
async with self._operation_lock:

View file

@ -104,7 +104,8 @@ def test_workspace_create_open_and_domain_fingerprint(tmp_path: Path) -> None:
workspace = workspace_module.Workspace.create(root, spec)
assert workspace.path("code/repo").is_dir()
assert workspace.path("datasets/search/cases").is_dir()
assert workspace.path("dataset/cases").is_dir()
assert not workspace.path("harnesses").exists()
assert workspace_module.Workspace.open(root, spec).manifest == workspace.manifest
changed = spec.model_copy(update={"scorer": "different"})
@ -114,6 +115,16 @@ def test_workspace_create_open_and_domain_fingerprint(tmp_path: Path) -> None:
workspace_module.Workspace.create(root, spec)
def test_workspace_reports_legacy_dataset_layout(tmp_path: Path) -> None:
root = tmp_path / "workspace"
workspace = workspace_module.Workspace.create(root, domain_spec())
workspace.path("datasets").mkdir()
workspace.path("dataset").rename(workspace.path("datasets/search"))
with pytest.raises(workspace_module.WorkspaceFormatError, match="move that directory to dataset"):
workspace_module.Workspace.open(root, domain_spec())
def test_workspace_rejects_escaping_paths_and_ids(tmp_path: Path) -> None:
workspace = workspace_module.Workspace.create(tmp_path / "workspace", domain_spec())
with pytest.raises(workspace_module.WorkspaceError):
@ -121,6 +132,9 @@ def test_workspace_rejects_escaping_paths_and_ids(tmp_path: Path) -> None:
with pytest.raises(workspace_module.WorkspaceError):
workspace.entity_path("evaluations", "../case")
attempt = workspace.case_attempt_dir("init", "validation-1", "case-1", "attempt-1", create=True)
assert attempt == workspace.path("evaluations/init/validation-1/cases/case-1/attempt-1")
def test_atomic_write_replaces_content_and_leaves_no_temporary_file(tmp_path: Path) -> None:
workspace = workspace_module.Workspace.create(tmp_path / "workspace", domain_spec())
@ -131,19 +145,19 @@ def test_atomic_write_replaces_content_and_leaves_no_temporary_file(tmp_path: Pa
assert not list(destination.parent.glob(".run.json.*"))
def test_install_search_dataset_replaces_empty_skeleton_and_is_read_only(tmp_path: Path) -> None:
def test_install_dataset_replaces_empty_skeleton_and_is_read_only(tmp_path: Path) -> None:
workspace = workspace_module.Workspace.create(tmp_path / "workspace", domain_spec())
source = tmp_path / "normalized"
(source / "cases").mkdir(parents=True)
(source / "manifest.json").write_text("{}\n", encoding="utf-8")
(source / "cases/case-1.json").write_text("{}\n", encoding="utf-8")
installed = workspace.install_search_dataset(source)
installed = workspace.install_dataset(source)
assert (installed / "cases/case-1.json").is_file()
assert not (installed / "manifest.json").stat().st_mode & 0o222
with pytest.raises(workspace_module.WorkspaceError, match="already been installed"):
workspace.install_search_dataset(source)
workspace.install_dataset(source)
def test_workspace_lock_rejects_live_owner_and_releases(tmp_path: Path) -> None:

View file

@ -60,8 +60,8 @@ def test_prepare_workspace_stores_only_selected_longmemeval_cases(tmp_path: Path
dataset_source=source,
)
manifest = json.loads(workspace.path("datasets/search/manifest.json").read_text(encoding="utf-8"))
case_files = list(workspace.path("datasets/search/cases").glob("*.json"))
manifest = json.loads(workspace.path("dataset/manifest.json").read_text(encoding="utf-8"))
case_files = list(workspace.path("dataset/cases").glob("*.json"))
stored_case = json.loads(case_files[0].read_text(encoding="utf-8"))
assert manifest["case_count"] == 1
assert manifest["query_count"] == 1
@ -72,7 +72,7 @@ def test_prepare_workspace_stores_only_selected_longmemeval_cases(tmp_path: Path
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, "branch", "--show-current") == "init"
assert _git(code_repository, "log", "-1", "--pretty=%s") == "Initial version"
assert _git(code_repository, "status", "--short") == ""
@ -96,7 +96,7 @@ def test_prepare_workspace_initializes_an_existing_empty_directory(tmp_path: Pat
workspace = run.prepare_workspace(workspace_root, "longmemeval", dataset_source=source)
assert workspace.path(".meta-reme-workspace.json").is_file()
assert workspace.path("datasets/search/manifest.json").is_file()
assert workspace.path("dataset/manifest.json").is_file()
assert workspace.path("code/repo/reme/reme/steps/benchmark/lme").is_dir()
@ -159,6 +159,7 @@ def test_load_config_reads_all_preparation_arguments(tmp_path: Path) -> None:
"variant": "100K",
"train_case_ids": [1, "2"],
},
"validation": {"concurrency": 5},
},
),
encoding="utf-8",
@ -170,3 +171,43 @@ def test_load_config_reads_all_preparation_arguments(tmp_path: Path) -> None:
assert config["dataset_source"] == run.PROJECT_ROOT / "benchmark/beam/dataset/BEAM"
assert config["dataset_variant"] == "100K"
assert config["train_case_ids"] == ["1", "2"]
assert config["validation_concurrency"] == 5
def test_prepare_and_validate_workspace_uses_all_installed_cases(tmp_path: Path) -> None:
source = tmp_path / "longmemeval.json"
source.write_text(json.dumps([_lme_item("case-2"), _lme_item("case-1")]), encoding="utf-8")
workspace_root = tmp_path / "meta-workspace"
config_path = tmp_path / "config_meta_reme.yaml"
config_path.write_text(
yaml.safe_dump(
{
"meta_workspace": str(workspace_root),
"dataset": {
"name": "longmemeval",
"source": str(source),
"train_case_ids": [],
},
"validation": {"concurrency": 5},
},
),
encoding="utf-8",
)
calls = []
def fake_validation(workspace, case_ids, code_id, concurrency, *, validation_id):
calls.append((Path(workspace), case_ids, code_id, concurrency, validation_id))
output = Path(workspace) / f"evaluations/{code_id}/{validation_id}"
output.mkdir(parents=True)
(output / "summary.json").write_text("{}\n", encoding="utf-8")
return output
workspace, validation = run.prepare_and_validate_workspace(config_path, validation_runner=fake_validation)
assert validation == workspace.path("evaluations/init/initial")
assert calls == [(workspace.root, ["case-2", "case-1"], "init", 5, "initial")]
_, repeated_validation = run.prepare_and_validate_workspace(config_path, validation_runner=fake_validation)
assert repeated_validation == validation
assert len(calls) == 1

View file

@ -0,0 +1,331 @@
"""Focused tests for Meta-ReMe validation orchestration."""
# Test doubles intentionally keep their methods compact.
# pylint: disable=missing-function-docstring
from __future__ import annotations
import asyncio
import importlib
import io
import json
from pathlib import Path
import subprocess
import sys
import tarfile
import pytest
META_REME = Path(__file__).resolve().parents[2] / "meta-reme"
PROJECT_ROOT = META_REME.parent
for import_root in (META_REME, PROJECT_ROOT):
if str(import_root) not in sys.path:
sys.path.insert(0, str(import_root))
git_manager = importlib.import_module("git_manager")
models = importlib.import_module("models")
workspace_module = importlib.import_module("workspace")
adapters = importlib.import_module("validation.adapters")
evaluator = importlib.import_module("validation.evaluator")
def _domain(max_retries: int = 0):
return models.DomainSpec(
dataset=models.DatasetSpec(name="longmemeval", source="dataset.json", fingerprint="source-sha"),
bundle_target="lme",
benchmark_runner="benchmark.longmemeval.run",
scorer="mean_query_score",
scope=models.ScopeSpec(),
sandbox=models.SandboxSpec(
image="sandbox-base:test",
timeout_seconds=60,
concurrency=2,
max_retries=max_retries,
),
proposer=models.ProposerSpec(model="test-model"),
budget=models.BudgetSpec(max_proposals=1),
)
def _case(case_id: str):
return models.CaseSpec(
case_id=case_id,
sessions=[
models.SessionSpec(
session_id=f"session-{case_id}",
messages=[
{
"role": "user",
"content": f"remember {case_id}",
"created_at": "2026-08-07T08:00:00",
},
],
metadata={"date": "2026-08-07"},
),
],
queries=[
models.QuerySpec(
query_id=f"query-{case_id}",
question="What should be remembered?",
golden_answer=case_id,
metadata={"question_type": "single-session-user"},
),
],
metadata={"dataset": "longmemeval"},
)
def _prepared_workspace(tmp_path: Path, case_ids: list[str]):
workspace = workspace_module.Workspace.create(tmp_path / "workspace", _domain())
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(
"dataset/manifest.json",
{
"schema_version": 1,
"dataset": "longmemeval",
"source_fingerprint": "source-sha",
"normalized_fingerprint": "normalized-sha",
"case_count": len(case_ids),
"query_count": len(case_ids),
},
)
repository = workspace.path("code/repo/reme")
(repository / "reme").mkdir(parents=True)
(repository / "reme/__init__.py").write_text("VERSION = 'committed'\n", encoding="utf-8")
(repository / "pyproject.toml").write_text("[project]\nname='fixture'\nversion='0.0.0'\n", encoding="utf-8")
commit = git_manager.initialize_repository(repository)
return workspace, repository, commit
class _FakeFactory:
def __init__(self, candidate, **kwargs):
self.candidate = candidate
self.kwargs = kwargs
self.active = 0
self.max_active = 0
self.events = []
async def create_case(self, case_id):
self.active += 1
self.max_active = max(self.max_active, self.active)
return _FakeCase(self, case_id)
class _FakeCase:
def __init__(self, factory, case_id):
self.factory = factory
self.case_id = case_id
async def run_build(self, jobs):
self.factory.events.append((self.case_id, "run_build"))
assert [job.job for job in jobs] == ["auto_memory", "index_update"]
await asyncio.sleep(0.01)
return {"success": True, "jobs": []}
async def run_queries(self, queries):
self.factory.events.append((self.case_id, "run_queries"))
await asyncio.sleep(0.01)
return {
"success": True,
"summary": {"mean_score": 1.0, "query_count": len(queries)},
"queries": [
{
"query_id": query.query_id,
"question": query.question,
"golden_answer": query.golden_answer,
"answer": query.golden_answer,
"score": 1.0,
"error": None,
}
for query in queries
],
}
async def _export(self, destination, files):
destination = Path(destination)
destination.parent.mkdir(parents=True, exist_ok=True)
with tarfile.open(destination, mode="w:gz") as archive:
for name in files:
payload = json.dumps({"case_id": self.case_id, "artifact": name}).encode()
info = tarfile.TarInfo(name)
info.size = len(payload)
archive.addfile(info, io.BytesIO(payload))
return destination
async def export_workspace(self, destination):
self.factory.events.append((self.case_id, "export_workspace"))
return await self._export(destination, ["memory.json"])
async def export_build_log(self, destination):
self.factory.events.append((self.case_id, "export_build_log"))
destination = Path(destination)
destination.write_text("build log\n", encoding="utf-8")
return destination
async def export_queries(self, destination):
self.factory.events.append((self.case_id, "export_queries"))
query_root = f"queries/query-{self.case_id}"
return await self._export(
destination,
["queries/summary.json", f"{query_root}/answer.log", f"{query_root}/result.json"],
)
async def export_full(self, destination):
return await self._export(destination, ["failure.json"])
async def close(self):
self.factory.active -= 1
def test_validation_uses_immutable_commit_and_bounds_full_case_lifecycle(tmp_path: Path) -> None:
case_ids = ["case-1", "case-2", "case-3"]
workspace, repository, commit = _prepared_workspace(tmp_path, case_ids)
(repository / "reme/__init__.py").write_text("VERSION = 'dirty'\n", encoding="utf-8")
created = []
def build_factory(*args, **kwargs):
factory = _FakeFactory(*args, **kwargs)
created.append(factory)
return factory
subprocess.run(["git", "branch", "candidate-1", commit], cwd=repository, check=True)
output = evaluator.run_validation(
workspace.root,
case_ids,
"candidate-1",
2,
validation_id="validation-1",
factory_builder=build_factory,
environment={},
)
assert output == workspace.path("evaluations/candidate-1/validation-1")
assert created[0].max_active == 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"
summary = json.loads((output / "summary.json").read_text(encoding="utf-8"))
manifest = json.loads((output / "manifest.json").read_text(encoding="utf-8"))
assert summary["code_id"] == "candidate-1"
assert summary["commit_sha"] == commit
assert manifest["code_id"] == "candidate-1"
assert manifest["commit_sha"] == commit
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",
]
assert summary["status"] == "completed"
assert summary["mean_query_score"] == 1.0
assert summary["case_count"] == 3
attempt = output / "cases/case-1/attempt-1"
assert (attempt / "memory_construction/result.json").is_file()
assert (attempt / "memory_construction/build.log").is_file()
memory_workspace = attempt / "memory_construction/reme_workspace.tar.gz"
assert memory_workspace.is_file()
with tarfile.open(memory_workspace, mode="r:gz") as archive:
assert archive.getnames() == ["memory.json"]
assert (attempt / "memory_construction/reme_workspace/memory.json").is_file()
case_result = json.loads((attempt / "case_result.json").read_text(encoding="utf-8"))
assert case_result["artifact_sha256"] == {
"memory_workspace": evaluator._sha256_file(memory_workspace), # pylint: disable=protected-access
}
assert (attempt / "queries/result.json").is_file()
assert (attempt / "queries/query-case-1/answer.log").is_file()
assert (attempt / "queries/query-case-1/result.json").is_file()
assert not (attempt / "queries/summary.json").exists()
assert not (attempt / "queries/artifacts").exists()
assert {path.name for path in (attempt / "queries").iterdir()} == {"result.json", "query-case-1"}
assert not list((attempt / "queries").glob("*.tar.gz"))
assert not list((attempt / "queries").glob(".*.tar.gz"))
@pytest.mark.asyncio
async def test_validation_exposes_native_async_project_api(tmp_path: Path) -> None:
workspace, _, _ = _prepared_workspace(tmp_path, ["case-async"])
output = await evaluator.run_validation_async(
workspace.root,
["case-async"],
"init",
1,
validation_id="async-api",
factory_builder=_FakeFactory,
environment={},
)
assert output == workspace.path("evaluations/init/async-api")
assert (output / "summary.json").is_file()
def test_validation_rejects_commit_sha_as_code_id(tmp_path: Path) -> None:
workspace, _, commit = _prepared_workspace(tmp_path, ["case-1"])
with pytest.raises(evaluator.ValidationError, match="unknown local branch code_id"):
evaluator.run_validation(
workspace.root,
["case-1"],
commit,
1,
validation_id="commit-is-not-code-id",
factory_builder=_FakeFactory,
environment={},
)
with pytest.raises(evaluator.ValidationError, match="path-safe Git branch name"):
evaluator.run_validation(
workspace.root,
["case-1"],
"feature/candidate",
1,
validation_id="unsafe-code-id",
factory_builder=_FakeFactory,
environment={},
)
def test_adapters_build_dataset_specific_requests() -> None:
lme_case = _case("lme")
lme_case.queries[0].metadata["question_date"] = "2026/08/07 (Fri) 09:30"
jobs = adapters.build_jobs(models.DatasetName.LONGMEMEVAL, lme_case)
lme_query = adapters.evaluation_queries(models.DatasetName.LONGMEMEVAL, lme_case)[0]
assert jobs[0].arguments["date"] == "2026-08-07"
assert lme_query.answer_arguments == {"query_time": "2026-08-07T09:30:00"}
assert lme_query.judge_answer_argument == "agent_answer"
beam_case = models.CaseSpec(
case_id="beam",
sessions=lme_case.sessions,
queries=[
models.QuerySpec(
query_id="event_ordering:1",
question="What happened first?",
golden_answer=["Mentions the first event"],
metadata={"question_type": "event_ordering", "rubric": ["Mentions the first event"]},
),
],
)
beam_query = adapters.evaluation_queries(models.DatasetName.BEAM, beam_case)[0]
assert beam_query.judge_answer_argument == "llm_response"
assert beam_query.score_path == "metadata.llm_judge_score"
assert beam_query.score_mapping 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"
case.sessions.append(
models.SessionSpec(
session_id="future",
messages=[{"role": "user", "content": "future", "created_at": "2026-08-08T08:00:00"}],
metadata={"date": "2026/08/08 (Sat) 08:00"},
),
)
jobs = adapters.build_jobs(models.DatasetName.LONGMEMEVAL, case)
assert [job.arguments.get("session_id") for job in jobs if job.job == "auto_memory"] == ["session-lme-filter"]

View file

@ -297,6 +297,32 @@ async def test_build_and_queries_use_batch_worker_requests(tmp_path):
]
assert len(worker_commands) == 2
construction = await case.export_memory_construction(tmp_path / "memory-construction.tar.gz")
assert construction.read_bytes() == b"case-archive"
assert workspace.backend.commands[-1][0] == [
"tar",
"-czf",
"/tmp/reme-case-export.tar.gz",
"-C",
"/workspace/case",
"reme_workspace",
"build_log",
]
build_log = await case.export_build_log(tmp_path / "build.log")
assert build_log.read_bytes() == b"build"
query_artifacts = await case.export_queries(tmp_path / "queries.tar.gz")
assert query_artifacts.read_bytes() == b"case-archive"
assert workspace.backend.commands[-1][0] == [
"tar",
"-czf",
"/tmp/reme-case-export.tar.gz",
"-C",
"/workspace/case",
"queries",
]
exported = await case.export_evaluation(tmp_path / "evaluation.tar.gz")
assert exported.read_bytes() == b"case-archive"
export_command = workspace.backend.commands[-1][0]