diff --git a/benchmark/longmemeval/config_meta_reme.yaml b/benchmark/longmemeval/config_meta_reme.yaml index 0ad548ce..83e61804 100644 --- a/benchmark/longmemeval/config_meta_reme.yaml +++ b/benchmark/longmemeval/config_meta_reme.yaml @@ -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 diff --git a/meta-reme/design.md b/meta-reme/design.md index 1419fc36..6846d8ca 100644 --- a/meta-reme/design.md +++ b/meta-reme/design.md @@ -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 - / - manifest.json - summary.md + dataset/ + manifest.json + cases/ weaknesses/ proposals/ evaluations/ - / - search.json - debug/ - cases/ + / + / + manifest.json + summary.json + cases/ + / + attempt-/ + case_result.json + memory_construction/ + result.json + reme_workspace.tar.gz + reme_workspace/ + build.log + queries/ + result.json + / + 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///cases//attempt-/ +evaluations///cases//attempt-/ ``` -解压过程需要拒绝绝对路径、`../`、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//screening/.json`；完整 search validation 仍发布为 harness 级 `evaluations//search.json`。两类结果均保存： +每次 validation 均写入 `evaluations///`；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//search.json`。随后记录 `validation_completed` 事件。若聚合阶段中断，可从具有相同选择 fingerprint 的 case 结果重新生成，不重复执行已验证的选择范围。 +`ValidationSpec` 选中的全部 case/query 达到可聚合状态后，编排器在对应 validation 目录确定性地生成 summary，随后记录 `validation_completed` 事件。若聚合阶段中断，可从具有相同选择 fingerprint 的 case 结果重新生成，不重复执行已验证的选择范围。 ### 12.3 搜索、Proposal 与 Git 续跑 diff --git a/meta-reme/git_manager.py b/meta-reme/git_manager.py index 6237b4c8..6f7d8d1b 100644 --- a/meta-reme/git_manager.py +++ b/meta-reme/git_manager.py @@ -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" diff --git a/meta-reme/run.py b/meta-reme/run.py index bc408d41..0555cbf3 100644 --- a/meta-reme/run.py +++ b/meta-reme/run.py @@ -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}", ) diff --git a/meta-reme/validation/README.md b/meta-reme/validation/README.md new file mode 100644 index 00000000..0d082949 --- /dev/null +++ b/meta-reme/validation/README.md @@ -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 \ + --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 +/evaluations/// +``` + +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//attempt-/ + case_result.json + memory_construction/ + result.json + reme_workspace.tar.gz + reme_workspace/ + build.log + queries/ + result.json + / + 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) +``` diff --git a/meta-reme/validation/__init__.py b/meta-reme/validation/__init__.py new file mode 100644 index 00000000..e9db6f65 --- /dev/null +++ b/meta-reme/validation/__init__.py @@ -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"] diff --git a/meta-reme/validation/__main__.py b/meta-reme/validation/__main__.py new file mode 100644 index 00000000..5debaff9 --- /dev/null +++ b/meta-reme/validation/__main__.py @@ -0,0 +1,6 @@ +"""Allow ``python -m validation`` to invoke the validation CLI.""" + +from .run import main + +if __name__ == "__main__": + main() diff --git a/meta-reme/validation/adapters.py b/meta-reme/validation/adapters.py new file mode 100644 index 00000000..80b4670b --- /dev/null +++ b/meta-reme/validation/adapters.py @@ -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") diff --git a/meta-reme/validation/evaluator.py b/meta-reme/validation/evaluator.py new file mode 100644 index 00000000..4241e3f7 --- /dev/null +++ b/meta-reme/validation/evaluator.py @@ -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() diff --git a/meta-reme/validation/run.py b/meta-reme/validation/run.py new file mode 100644 index 00000000..40884ba4 --- /dev/null +++ b/meta-reme/validation/run.py @@ -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() diff --git a/meta-reme/workspace.py b/meta-reme/workspace.py index ced78e8f..3f81eae3 100644 --- a/meta-reme/workspace.py +++ b/meta-reme/workspace.py @@ -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) diff --git a/sandbox/README.md b/sandbox/README.md index b12a6639..eb0f9023 100644 --- a/sandbox/README.md +++ b/sandbox/README.md @@ -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 diff --git a/sandbox/README_ZH.md b/sandbox/README_ZH.md index f9ed1871..a09ef035 100644 --- a/sandbox/README_ZH.md +++ b/sandbox/README_ZH.md @@ -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 diff --git a/sandbox/docker_sandbox.py b/sandbox/docker_sandbox.py index d382694d..5c4645ef 100644 --- a/sandbox/docker_sandbox.py +++ b/sandbox/docker_sandbox.py @@ -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: diff --git a/tests/unit_sandbox/test_meta_reme_models_workspace.py b/tests/unit_sandbox/test_meta_reme_models_workspace.py index 40b907bb..9ee49a31 100644 --- a/tests/unit_sandbox/test_meta_reme_models_workspace.py +++ b/tests/unit_sandbox/test_meta_reme_models_workspace.py @@ -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: diff --git a/tests/unit_sandbox/test_meta_reme_run.py b/tests/unit_sandbox/test_meta_reme_run.py index b024ce6c..148f4eb1 100644 --- a/tests/unit_sandbox/test_meta_reme_run.py +++ b/tests/unit_sandbox/test_meta_reme_run.py @@ -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 diff --git a/tests/unit_sandbox/test_meta_reme_validation.py b/tests/unit_sandbox/test_meta_reme_validation.py new file mode 100644 index 00000000..77b7e415 --- /dev/null +++ b/tests/unit_sandbox/test_meta_reme_validation.py @@ -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"] diff --git a/tests/unit_sandbox/test_sandbox.py b/tests/unit_sandbox/test_sandbox.py index dc471e5f..097ce280 100644 --- a/tests/unit_sandbox/test_sandbox.py +++ b/tests/unit_sandbox/test_sandbox.py @@ -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]