From 78217b2f52069eb73fb64a78e812f59bb4f77bcc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=85=B1=E7=89=9B=E8=82=89?= Date: Thu, 6 Aug 2026 16:30:32 +0800 Subject: [PATCH] Refactor core implementation and simplify related code --- .gitignore | 3 +- meta-reme/build_bundle.yaml | 80 ++- meta-reme/bundle_builder.py | 172 +++++- meta-reme/design.md | 64 ++- meta-reme/mermaid-diagram.svg | 1 + meta-reme/models.py | 507 ++++++++++++++++++ meta-reme/workspace.py | 317 +++++++++++ sandbox/README.md | 31 ++ sandbox/docker_sandbox.py | 156 +++++- tests/unit_sandbox/test_bundle_builder.py | 68 +++ .../test_meta_reme_models_workspace.py | 177 ++++++ tests/{unit => unit_sandbox}/test_sandbox.py | 94 ++++ 12 files changed, 1660 insertions(+), 10 deletions(-) create mode 100644 meta-reme/mermaid-diagram.svg create mode 100644 meta-reme/models.py create mode 100644 meta-reme/workspace.py create mode 100644 tests/unit_sandbox/test_bundle_builder.py create mode 100644 tests/unit_sandbox/test_meta_reme_models_workspace.py rename tests/{unit => unit_sandbox}/test_sandbox.py (85%) diff --git a/.gitignore b/.gitignore index 725c9c0b..7475e136 100644 --- a/.gitignore +++ b/.gitignore @@ -72,7 +72,8 @@ benchmark/*/results/ # integration tests outputs tests/integration/logs/ +sandbox/tmp_test_trajectories/ daily/ #meta-reme -meta-reme/bundles/ \ No newline at end of file +meta-reme/bundles/ diff --git a/meta-reme/build_bundle.yaml b/meta-reme/build_bundle.yaml index 5070dc35..e4a22248 100644 --- a/meta-reme/build_bundle.yaml +++ b/meta-reme/build_bundle.yaml @@ -5,8 +5,6 @@ version: 1 # copied. common: include: - - LICENSE - - README.md - pyproject.toml - reme/__init__.py - reme/application.py @@ -24,6 +22,32 @@ common: - reme/steps/file_io - reme/steps/index - reme/steps/transfer + exclude: + - reme/components/client/http_client.py + - reme/components/client/mcp_client.py + - reme/components/file_graph/neo4j_file_graph.py + - reme/components/file_graph/nx_file_graph.py + - reme/components/file_store/faiss_local_file_store.py + - reme/components/file_store/zvec_local_file_store.py + - reme/components/service/http_service.py + - reme/components/service/mcp_service.py + - reme/components/tokenizer/jieba_tokenizer.py + - reme/schema/auto_fin.py + - reme/schema/daily_paper.py + - reme/steps/common/add.py + - reme/steps/common/demo.py + - reme/steps/common/llm_demo.py + - reme/steps/common/stream_demo.py + - reme/steps/common/stream_llm_demo.py + - reme/steps/common/version.py + - reme/steps/file_io/read_image.py + - reme/utils/arxiv.py + - reme/utils/huggingface_papers.py + config_transform: + service_backend: cli + remove_jobs: + - read_image + - version step_packages: - common - evolve @@ -38,8 +62,50 @@ common: - auto_memory_step - search_step - search_v2_step + forbidden_backends: + - as_embedding:dashscope_multimodal + - client:http + - client:mcp + - file_graph:neo4j + - file_graph:nx + - file_store:faiss + - file_store:zvec + - service:http + - service:mcp + - step:add_step + - step:demo_echo_step1 + - step:demo_echo_step2 + - step:llm_demo_step + - step:read_image_step + - step:stream_demo_step1 + - step:stream_demo_step2 + - step:stream_llm_demo_step + - step:version_step + - tokenizer:jieba forbidden_paths: + - LICENSE + - README.md - reme/steps/cookbook + - reme/components/client/http_client.py + - reme/components/client/mcp_client.py + - reme/components/file_graph/neo4j_file_graph.py + - reme/components/file_graph/nx_file_graph.py + - reme/components/file_store/faiss_local_file_store.py + - reme/components/file_store/zvec_local_file_store.py + - reme/components/service/http_service.py + - reme/components/service/mcp_service.py + - reme/components/tokenizer/jieba_tokenizer.py + - reme/schema/auto_fin.py + - reme/schema/daily_paper.py + - reme/steps/common/add.py + - reme/steps/common/demo.py + - reme/steps/common/llm_demo.py + - reme/steps/common/stream_demo.py + - reme/steps/common/stream_llm_demo.py + - reme/steps/common/version.py + - reme/steps/file_io/read_image.py + - reme/utils/arxiv.py + - reme/utils/huggingface_papers.py targets: default: @@ -61,6 +127,8 @@ targets: - reme/steps/benchmark/base - reme/steps/benchmark/lme benchmark_package: lme + exclude: + - reme/steps/evolve/auto_memory_cc.py validation: imports: - reme.steps.benchmark.lme @@ -68,7 +136,10 @@ targets: - lme_auto_memory_step - lme_agentic_answer_step - lme_answer_judge_step + forbidden_backends: + - step:auto_memory_cc_step forbidden_paths: + - reme/steps/evolve/auto_memory_cc.py - reme/steps/benchmark/beam beam: @@ -77,6 +148,8 @@ targets: - reme/steps/benchmark/base - reme/steps/benchmark/beam benchmark_package: beam + exclude: + - reme/steps/evolve/auto_memory_cc.py validation: imports: - reme.steps.benchmark.beam @@ -84,5 +157,8 @@ targets: - beam_auto_memory_step - beam_agentic_answer_step - beam_rubric_judge_step + forbidden_backends: + - step:auto_memory_cc_step forbidden_paths: + - reme/steps/evolve/auto_memory_cc.py - reme/steps/benchmark/lme diff --git a/meta-reme/bundle_builder.py b/meta-reme/bundle_builder.py index 572ff635..20ea1314 100644 --- a/meta-reme/bundle_builder.py +++ b/meta-reme/bundle_builder.py @@ -10,6 +10,7 @@ from __future__ import annotations import argparse import os from pathlib import Path +import re import shutil import subprocess import sys @@ -80,13 +81,25 @@ def _resolve_source_path(source_repo: Path, relative: str) -> Path: return resolved -def _iter_files(source_repo: Path, entries: Iterable[str]) -> list[tuple[Path, Path]]: +def _is_within(relative: Path, parent: Path) -> bool: + """Return whether ``relative`` is ``parent`` or one of its descendants.""" + return relative == parent or parent in relative.parents + + +def _iter_files( + source_repo: Path, + entries: Iterable[str], + exclusions: Iterable[str] = (), +) -> list[tuple[Path, Path]]: + excluded = [Path(value) for value in exclusions] files: dict[Path, Path] = {} for entry in entries: source = _resolve_source_path(source_repo, entry) candidates = [source] if source.is_file() else sorted(path for path in source.rglob("*") if path.is_file()) for candidate in candidates: relative = candidate.relative_to(source_repo) + if any(_is_within(relative, path) for path in excluded): + continue if any(part in _IGNORED_PARTS for part in relative.parts) or candidate.suffix in _IGNORED_SUFFIXES: continue if candidate.is_symlink(): @@ -95,6 +108,145 @@ def _iter_files(source_repo: Path, entries: Iterable[str]) -> list[tuple[Path, P return [(relative, files[relative]) for relative in sorted(files)] +def _write_pruned_package_initializers(bundle_root: Path) -> None: + """Write registration imports for packages whose optional modules were pruned.""" + initializers = { + "reme/components/client/__init__.py": '''"""Client abstractions included in this generated bundle.""" + +from .base_client import BaseClient + +__all__ = ["BaseClient"] +''', + "reme/components/file_graph/__init__.py": '''"""File graph implementations included in this generated bundle.""" + +from .base_file_graph import BaseFileGraph +from .local_file_graph import LocalFileGraph + +__all__ = ["BaseFileGraph", "LocalFileGraph"] +''', + "reme/components/file_store/__init__.py": '''"""File store implementations included in this generated bundle.""" + +from .base_file_store import BaseFileStore +from .local_file_store import LocalFileStore + +__all__ = ["BaseFileStore", "LocalFileStore"] +''', + "reme/components/job/__init__.py": '''"""Job implementations included in this generated bundle.""" + +from .background_job import BackgroundJob +from .base_job import BaseJob +from .cron_job import CronJob +from .stream_job import StreamJob + +__all__ = ["BackgroundJob", "BaseJob", "CronJob", "StreamJob"] +''', + "reme/components/service/__init__.py": '''"""Service implementations included in this generated bundle.""" + +from .base_service import BaseService +from .cli_service import CliService + +__all__ = ["BaseService", "CliService"] +''', + "reme/components/tokenizer/__init__.py": '''"""Tokenizer implementations included in this generated bundle.""" + +from .base_tokenizer import BaseTokenizer +from .regex_tokenizer import RegexTokenizer + +__all__ = ["BaseTokenizer", "RegexTokenizer"] +''', + } + for relative, content in initializers.items(): + (bundle_root / relative).write_text(content, encoding="utf-8") + + file_io_init = bundle_root / "reme/steps/file_io/__init__.py" + content = file_io_init.read_text(encoding="utf-8") + content = content.replace("from .read_image import ReadImageStep\n", "") + content = content.replace(' "ReadImageStep",\n', "") + file_io_init.write_text(content, encoding="utf-8") + + common_init = bundle_root / "reme/steps/common/__init__.py" + content = common_init.read_text(encoding="utf-8") + for import_line in ( + "from .add import AddStep\n", + "from .demo import DemoEchoStep1, DemoEchoStep2\n", + "from .llm_demo import LLMDemoStep\n", + "from .stream_demo import StreamDemoStep1, StreamDemoStep2\n", + "from .stream_llm_demo import StreamLLMDemoStep\n", + "from .version import VersionStep\n", + ): + content = content.replace(import_line, "") + for exported_name in ( + "AddStep", + "DemoEchoStep1", + "DemoEchoStep2", + "LLMDemoStep", + "StreamDemoStep1", + "StreamDemoStep2", + "StreamLLMDemoStep", + "VersionStep", + ): + content = content.replace(f' "{exported_name}",\n', "") + common_init.write_text(content, encoding="utf-8") + + auto_memory_cc = bundle_root / "reme/steps/evolve/auto_memory_cc.py" + if not auto_memory_cc.exists(): + evolve_init = bundle_root / "reme/steps/evolve/__init__.py" + content = evolve_init.read_text(encoding="utf-8") + content = content.replace("from .auto_memory_cc import AutoMemoryCCStep\n", "") + content = content.replace(' "AutoMemoryCCStep",\n', "") + evolve_init.write_text(content, encoding="utf-8") + + schema_init = bundle_root / "reme/schema/__init__.py" + content = schema_init.read_text(encoding="utf-8") + auto_fin_import = re.search(r"from \.auto_fin import \((.*?)\)\n", content, flags=re.DOTALL) + if auto_fin_import is None: + raise BundleBuildError("Could not prune auto-fin schema exports") + auto_fin_names = re.findall(r"\b[A-Za-z_][A-Za-z0-9_]*\b", auto_fin_import.group(1)) + content = content[: auto_fin_import.start()] + content[auto_fin_import.end() :] + daily_paper_import = re.search(r"from \.daily_paper import ([^\n]+)\n", content) + if daily_paper_import is None: + raise BundleBuildError("Could not prune daily-paper schema exports") + daily_paper_names = [name.strip() for name in daily_paper_import.group(1).split(",")] + content = content[: daily_paper_import.start()] + content[daily_paper_import.end() :] + for exported_name in auto_fin_names + daily_paper_names: + content = content.replace(f' "{exported_name}",\n', "") + schema_init.write_text(content, encoding="utf-8") + + embedding_init = bundle_root / "reme/components/as_embedding/__init__.py" + content = embedding_init.read_text(encoding="utf-8") + content, count = re.subn( + r'\n@R\.register\("dashscope_multimodal"\)\nclass DashScopeMultiModalAsEmbedding\b.*?' + r'(?=\n@R\.register\("gemini"\))', + "", + content, + flags=re.DOTALL, + ) + if count != 1: + raise BundleBuildError("Could not prune the multimodal embedding backend") + content = content.replace(' "DashScopeMultiModalAsEmbedding",\n', "") + embedding_init.write_text(content, encoding="utf-8") + + +def _transform_target_config(bundle_root: Path, entries: list[str], transform: Mapping[str, Any]) -> None: + """Apply the sealed-bundle service and job restrictions to its one target config.""" + config_entries = [Path(value) for value in entries if Path(value).parent == Path("reme/config")] + config_entries = [value for value in config_entries if value.suffix in {".yaml", ".yml"}] + if len(config_entries) != 1: + raise BundleBuildError(f"Expected exactly one target YAML config, found: {config_entries}") + + config_path = bundle_root / config_entries[0] + config = _mapping(yaml.safe_load(config_path.read_text(encoding="utf-8")), "target config") + config = dict(config) + service = dict(_mapping(config.get("service", {}), "target config.service")) + service["backend"] = transform.get("service_backend", service.get("backend")) + config["service"] = service + jobs = dict(_mapping(config.get("jobs", {}), "target config.jobs")) + for name in _string_list(transform.get("remove_jobs"), "common.config_transform.remove_jobs"): + jobs.pop(name, None) + config["jobs"] = jobs + config_path.write_text(yaml.safe_dump(config, sort_keys=False, allow_unicode=True), encoding="utf-8") + + def _write_step_initializers(bundle_root: Path, step_packages: list[str], benchmark_package: str | None) -> None: steps_dir = bundle_root / "reme" / "steps" imports = list(step_packages) @@ -134,7 +286,7 @@ def _merged_validation(common: Mapping[str, Any], target_spec: Mapping[str, Any] common_validation = _mapping(common.get("validation", {}), "common.validation") target_validation = _mapping(target_spec.get("validation", {}), "target.validation") result: dict[str, list[str]] = {} - for field in ("imports", "required_steps", "forbidden_paths"): + for field in ("imports", "required_steps", "forbidden_backends", "forbidden_paths"): result[field] = _string_list(common_validation.get(field), f"common.validation.{field}") + _string_list( target_validation.get(field), f"target.validation.{field}", @@ -143,6 +295,9 @@ def _merged_validation(common: Mapping[str, Any], target_spec: Mapping[str, Any] def _validate_bundle(bundle_root: Path, validation: Mapping[str, list[str]]) -> None: + cached = [path.relative_to(bundle_root) for path in bundle_root.rglob("*") if path.name == "__pycache__"] + if cached: + raise BundleBuildError(f"Python cache directories were included in bundle: {cached}") for relative in validation["forbidden_paths"]: if (bundle_root / relative).exists(): raise BundleBuildError(f"Forbidden path was included in bundle: {relative}") @@ -154,10 +309,14 @@ def _validate_bundle(bundle_root: Path, validation: Mapping[str, list[str]]) -> "from reme.enumeration import ComponentEnum\n" f"imports = {validation['imports']!r}\n" f"required = {validation['required_steps']!r}\n" + f"forbidden = {validation['forbidden_backends']!r}\n" "for name in imports: importlib.import_module(name)\n" "registered = R.get_all(ComponentEnum.STEP)\n" "missing = [name for name in required if name not in registered]\n" "if missing: raise RuntimeError(f'Missing registered steps: {missing}')\n" + "unexpected = [entry for entry in forbidden " + "if R.get(ComponentEnum(entry.split(':', 1)[0]), entry.split(':', 1)[1]) is not None]\n" + "if unexpected: raise RuntimeError(f'Forbidden registered backends: {unexpected}')\n" ) environment = os.environ.copy() environment["PYTHONPATH"] = str(bundle_root) @@ -192,12 +351,17 @@ def build_bundle( target_spec.get("include"), f"targets.{target}.include", ) + exclusions = _string_list(common.get("exclude"), "common.exclude") + _string_list( + target_spec.get("exclude"), + f"targets.{target}.exclude", + ) step_packages = _string_list(common.get("step_packages"), "common.step_packages") benchmark_package = target_spec.get("benchmark_package") if benchmark_package is not None and not isinstance(benchmark_package, str): raise BundleBuildError(f"'targets.{target}.benchmark_package' must be a string") - files = _iter_files(source_repo, entries) + files = _iter_files(source_repo, entries, exclusions) validation = _merged_validation(common, target_spec) + config_transform = _mapping(common.get("config_transform", {}), "common.config_transform") output_dir.mkdir(parents=True, exist_ok=True) temporary = Path(tempfile.mkdtemp(prefix=f".{target}-bundle-", dir=output_dir)) @@ -209,6 +373,8 @@ def build_bundle( destination.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, destination) _write_step_initializers(staged_root, step_packages, benchmark_package) + _write_pruned_package_initializers(staged_root) + _transform_target_config(staged_root, entries, config_transform) if validate: _validate_bundle(staged_root, validation) diff --git a/meta-reme/design.md b/meta-reme/design.md index 6a2808c3..83e4cc97 100644 --- a/meta-reme/design.md +++ b/meta-reme/design.md @@ -51,6 +51,7 @@ meta-reme/ - auto-memory 正式修改白名单； - debug 阶段允许使用的临时目录； - sandbox 镜像、超时、并发和重试策略； +- validation 的默认抽样策略、允许指定的 case/query 以及升级为全量评测的条件； - proposer 模型、搜索预算和 top-K 数量； - 唯一目标 `mean_query_score`。 @@ -185,6 +186,8 @@ prepare → install → ingest → index → answer → export Agent 可以自主选择完整 case、指定 session、session prefix 或少量 query 进行 debug。部分 session 结果只用于排查语法、配置、记忆写入和运行错误，不进入正式评分。 +除 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、已完成阶段和宿主异常。每次运行结束后，宿主下载原始压缩包，并安全解压到： ```text @@ -195,13 +198,18 @@ evaluations///cases//attempt-/ ## 10. Validation 与结果记录 -正式 validation 必须从 frozen commit 创建全新 SourceCandidate。不同 harness、case 之间使用独立 runtime workspace；同一 case 的多个 query 应从相同的 post-ingestion memory snapshot 启动，避免 query 顺序造成状态污染。 +Validation 分为 targeted screening 和 full validation。两者都必须从 frozen commit 创建全新 SourceCandidate，并使用同一套 sandbox、失败分类和结果 schema。不同 harness、case 之间使用独立 runtime workspace；同一 case 本次选中的多个 query 应从相同的 post-ingestion memory snapshot 启动，避免 query 顺序造成状态污染。 -每个 harness 的 `search.json` 保存： +`ValidationSpec` 明确记录目标 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`。两类结果均保存： - commit、parents、changed paths 和 scope 结果； - dataset、代码、配置、模型和镜像 fingerprint； - 总平均分、query 数、失败数和运行状态； +- validation spec、选择理由、case/query 覆盖数量、覆盖率和是否具有正式可比性； - 每个 case/session 的处理结果； - 每个 query 的问题、golden answer、模型回答、score、Judge 结论、异常、token 和耗时； - artifacts 路径及 hash。 @@ -219,10 +227,60 @@ evaluations///cases//attempt-/ → 修改和 sandbox debug → 冻结 commit → 白名单及接口检查 -→ 完整 search validation +→ 可选：指定 case/query 的 targeted screening +→ 通过预设升级条件后执行完整 search validation → 更新 registry、summary 和 leaderboard ``` +Targeted screening 是可选的成本控制阶段。未通过筛选的 candidate 记录结果和淘汰原因，但不消耗“完整 validation 数”预算；其实际 token、时间和费用仍计入对应总预算。筛选规则、样本范围和升级阈值必须在 domain spec 或 proposal 执行前确定，避免根据中间结果反复挑选有利样本。 + 达到 proposal 数、完整 validation 数、token 或时间预算后，系统选择 search set 平均分最高的候选。完全同分时保留较早的 incumbent，不引入其他优化指标。 最终 winner 必须从 commit SHA 重新构建，并在完整 search set 上进行 clean replay。只有 replay 成功后才加载 test set，分别评测 baseline 和 winner，生成最终报告。整个过程中 test 结果不反馈给 agent，从而保证 test set 只用于衡量泛化效果。 + +## 12. 中断与恢复 + +搜索可能持续数小时乃至数天，进程崩溃、机器重启或用户主动终止都不应破坏已完成的成果。恢复机制必须满足三个要求：持久化状态可校验、恢复操作幂等、同一 workspace 同时只有一个编排器写入。 + +### 12.1 持久化与一致性边界 + +`events.jsonl` 是搜索状态的事实来源，`run.json` 是由事件日志生成的加速快照。每条事件包含单调递增的 `seq`、唯一 `event_id`、时间戳和关联的 round、proposal、commit、validation、case、attempt 标识。写入事件后再原子替换 `run.json`；恢复时从快照记录的 `last_event_seq` 继续回放。若快照缺失、落后或校验失败，则完全从事件日志重建，不以快照覆盖较新的事件。 + +`run.json`、case 结果、聚合结果和 registry 均采用“写临时文件 → flush/fsync → 同目录原子 rename”的方式发布。`events.jsonl` 尾部若存在崩溃造成的不完整 JSON 行，恢复时忽略该行并记录诊断信息；中间行损坏则停止恢复并明确报错，避免静默跳过历史。启动时通过 workspace lock 拒绝第二个并发编排器，lock 中记录 PID、主机和启动时间，并允许在确认所属进程已不存在后接管 stale lock。 + +### 12.2 Case 原子性与结果复用 + +每个 case 的完整 pipeline（`prepare → install → ingest → index → answer → export`）是最小恢复单元。中断时不尝试复用半写入的记忆、索引或回答；下一次 attempt 在全新的 runtime workspace 中从头执行。中断或基础设施失败的 attempt 目录及原始 artifacts 保留用于审计，不原地覆盖，也不作为有效结果参与聚合；可归因于候选本身的确定性失败仍按第 3 节记为 0 分。 + +完成的 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 结果重新生成，不重复执行已验证的选择范围。 + +### 12.3 搜索、Proposal 与 Git 续跑 + +恢复以最后一个已提交的阶段事件为准，而不是无条件进入“下一轮”： + +- **搜索循环**：已记录 `round_completed` 的轮次不再执行；进行中的轮次从其最后一个完成阶段继续。只有尚未产生可恢复产物的 agent 调用才重新发起。 +- **Proposal**：已落盘的 weakness 和 proposal 文件必须具有唯一 ID、fingerprint 和对应的完成事件。仅有临时文件或缺少完成事件的 agent 输出视为未提交，可保留为诊断材料，但不进入搜索状态。 +- **Harness**：commit 只有通过 scope、import、smoke 检查并写入 registry 后才成为 frozen harness。存在 commit 但尚未登记时，恢复流程根据事件和检查结果决定继续冻结或放弃，不能仅凭分支名推断状态。 +- **Git 管理**：恢复时先核对 registry 中的 commit 和 snapshot hash，再清理可证明由本次运行创建且不再被引用的 worktree。未登记分支和 debug commit 默认保留或归档；不得自动删除用户分支、受保护分支或仍被事件、proposal、registry 引用的对象。 +- **Validation**：进行中的 validation 只调度缺少有效完成标志的 case/query 选择；恢复时必须使用原 `ValidationSpec`，不能重新抽样。部分 validation 完成后记录覆盖范围和筛选结论；只有 full validation 才更新正式 leaderboard。 + +预算只由去重后的完成事件结算，`event_id` 重放不得重复计费。已发起但没有产生完成事件的模型调用或 sandbox attempt 仍记录实际可观测的 token、费用和耗时；若外部系统无法确认消耗，则记为 `unknown` 并按 domain spec 的保守策略处理。时间预算使用累计运行时，不包含进程停止期间的墙钟时间，除非 domain spec 明确配置绝对截止时间。 + +### 12.4 恢复流程 + +`run.py` 启动时执行以下恢复流程： + +```text +1. 获取 workspace lock；若已有存活 owner，则拒绝启动 +2. 校验 events.jsonl，并以 run.json 的 last_event_seq 为起点回放；必要时从头重建快照 +3. 校验 registry 中的 commit、snapshot hash 和 frozen harness 状态 +4. 根据阶段事件定位进行中的 round、proposal 或 validation +5. 校验 case 的 complete.json、fingerprint 和 artifacts hash，复用有效结果 +6. 为未完成 case 创建新的 attempt；从有效 case 结果重新执行缺失的聚合 +7. 清理无引用且可安全识别的临时 worktree，保留或归档其他 Git 状态 +8. 原子写入新快照并追加 resume_completed 事件，从最后一个未完成阶段继续 +``` + +`resume_started` 和 `resume_completed` 事件记录恢复前后的状态摘要及本次采取的动作。上述流程可重复执行：再次中断不会重复登记 harness、重复结算预算或覆盖已有 attempt。无法验证的状态应停止并给出可操作的错误，不得通过猜测或静默删除数据来推进搜索。 diff --git a/meta-reme/mermaid-diagram.svg b/meta-reme/mermaid-diagram.svg new file mode 100644 index 00000000..c3085cbc --- /dev/null +++ b/meta-reme/mermaid-diagram.svg @@ -0,0 +1 @@ +⑦ 中断恢复路径⑥ Winner 与最终测试⑤ Harness 搜索循环④ Candidate 评测流水线③ Dataset 层② Baseline 建立① 初始化与恢复使用 schema计算 NextAction首次运行选择 default / lme / beam规范化、生成 fingerprint最终阶段才加载首次完整评测全部有效 case仅提供 Search Set选择 Parent不通过：记录失败通过：冻结完整 Search Validation否：下一轮是一致且成功恢复启动跳过已有 complete.json 的Caserun.py解析命令、装配依赖、持有Workspace Lockconfig.py加载并校验 DomainSpecmodels.pyPydantic 公共契约workspace.py目录、原子写入、Workspace Lockresult_store.py事件日志、快照、Registry、Leaderboardrecovery.py / state.py事件回放与状态机search_loop.py执行状态机返回的下一动作bundle_builder.py生成最小 BenchmarkBundlegit_manager.pyGit仓库、Commit、Worktree不可变 Baseline Commitdatasets/base.py统一 Dataset 接口datasets/longmemeval.pydatasets/beam.pySearch SetAgent 可见Test Set搜索阶段隔离evaluator.py调度 Sandbox 和重试SourceSnapshotSourceCandidate独立 Sandbox Caseprepare → install → ingest→ index → answer →exportcase_result.jsoncomplete.json + SHA256Validation 聚合evaluations/commit/search.json历史结果与 Top-Kagent.pyWeakness Mining +Proposalweaknesses/round.jsonproposals/proposal.json受控 Git Worktree修改、Debug、Commitscope_guard.py路径、Python、Prompt、YAML 语义检查Frozen Harness CommitLeaderboard 更新预算是否耗尽？选择 Search Set平均分最高的 Harness从 Commit 重建Clean Replay Search Set隔离评测Baseline vs Winner最终结构化报告与可读报告events.jsonl事实来源run.json可重建快照校验Registry、Commit、Fingerprint、Hash定位最后一个未完成阶段 \ No newline at end of file diff --git a/meta-reme/models.py b/meta-reme/models.py new file mode 100644 index 00000000..8829eb74 --- /dev/null +++ b/meta-reme/models.py @@ -0,0 +1,507 @@ +"""Public data contracts shared by the Meta-ReMe orchestration modules. + +The models in this module describe persisted state and values crossing module +boundaries. File I/O, event reduction, scoring, and orchestration belong in +their respective modules rather than here. +""" + +# Persisted schema classes are intentionally named and self-describing. +# pylint: disable=missing-class-docstring,missing-function-docstring + +from __future__ import annotations + +from datetime import datetime, timezone +from enum import StrEnum +import hashlib +import json +from pathlib import Path +from typing import Any, Mapping +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +SCHEMA_VERSION = 1 + + +def utc_now() -> datetime: + """Return an aware UTC timestamp.""" + + return datetime.now(timezone.utc) + + +class ContractModel(BaseModel): + """Base class for strict, JSON-serializable persisted contracts.""" + + model_config = ConfigDict(extra="forbid", use_enum_values=False) + + +def canonical_json_bytes(value: Any) -> bytes: + """Serialize a supported value deterministically for hashing.""" + + if isinstance(value, BaseModel): + value = value.model_dump(mode="json", exclude_none=False) + return json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + default=_json_default, + ).encode("utf-8") + + +def _json_default(value: Any) -> Any: + if isinstance(value, Path): + return value.as_posix() + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, StrEnum): + return value.value + raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable") + + +def fingerprint(value: Any) -> str: + """Return the SHA256 of a canonical JSON representation.""" + + return hashlib.sha256(canonical_json_bytes(value)).hexdigest() + + +class DatasetName(StrEnum): + LONGMEMEVAL = "longmemeval" + BEAM = "beam" + + +class RunStatus(StrEnum): + INITIALIZING = "initializing" + SEARCHING = "searching" + REPLAYING = "replaying" + TESTING = "testing" + COMPLETED = "completed" + FAILED = "failed" + + +class EvaluationMode(StrEnum): + DEBUG = "debug" + SCREENING = "screening" + SEARCH = "search" + REPLAY = "replay" + TEST = "test" + + +class AttemptStatus(StrEnum): + COMPLETED = "completed" + CANDIDATE_FAILURE = "candidate_failure" + INFRA_ERROR = "infra_error" + INTERRUPTED = "interrupted" + + +class ValidationStatus(StrEnum): + RUNNING = "running" + COMPLETED = "completed" + INFRA_ERROR = "infra_error" + + +class HarnessStatus(StrEnum): + DEBUG = "debug" + FROZEN = "frozen" + REJECTED = "rejected" + + +class EventType(StrEnum): + RUN_INITIALIZED = "run_initialized" + RESUME_STARTED = "resume_started" + RESUME_COMPLETED = "resume_completed" + ROUND_STARTED = "round_started" + WEAKNESS_COMPLETED = "weakness_completed" + PROPOSAL_COMPLETED = "proposal_completed" + HARNESS_COMMITTED = "harness_committed" + HARNESS_FROZEN = "harness_frozen" + HARNESS_REJECTED = "harness_rejected" + VALIDATION_STARTED = "validation_started" + CASE_ATTEMPT_COMPLETED = "case_attempt_completed" + VALIDATION_COMPLETED = "validation_completed" + LEADERBOARD_UPDATED = "leaderboard_updated" + ROUND_COMPLETED = "round_completed" + WINNER_SELECTED = "winner_selected" + RUN_COMPLETED = "run_completed" + RUN_FAILED = "run_failed" + + +class DatasetSpec(ContractModel): + name: DatasetName + source: str + fingerprint: str = Field(min_length=1) + + +class ScopeSpec(ContractModel): + harness_paths: list[str] = Field(default_factory=list) + debug_paths: list[str] = Field(default_factory=lambda: ["meta_debug"]) + config_paths: dict[str, list[str]] = Field(default_factory=dict) + + +class SandboxSpec(ContractModel): + image: str = Field(min_length=1) + timeout_seconds: float = Field(gt=0) + concurrency: int = Field(default=1, ge=1) + max_retries: int = Field(default=2, ge=0) + max_artifact_bytes: int = Field(default=1_073_741_824, gt=0) + + +class ProposerSpec(ContractModel): + model: str = Field(min_length=1) + top_k: int = Field(default=5, ge=1) + + +class ValidationPolicySpec(ContractModel): + screening_enabled: bool = True + default_case_ids: list[str] | None = None + default_query_ids: dict[str, list[str]] = Field(default_factory=dict) + promotion_min_mean_query_score: float | None = None + require_full_for_leaderboard: bool = True + + @field_validator("require_full_for_leaderboard") + @classmethod + def require_full_results(cls, value: bool) -> bool: + if not value: + raise ValueError("partial validations cannot be admitted to the leaderboard") + return value + + @model_validator(mode="after") + def validate_default_selection(self) -> "ValidationPolicySpec": + ValidationSelection( + case_ids=self.default_case_ids, + query_ids=self.default_query_ids, + reason="domain default screening selection", + ) + return self + + +class BudgetSpec(ContractModel): + max_proposals: int | None = Field(default=None, ge=1) + max_validations: int | None = Field(default=None, ge=1) + max_tokens: int | None = Field(default=None, ge=1) + max_runtime_seconds: float | None = Field(default=None, gt=0) + deadline: datetime | None = None + + @model_validator(mode="after") + def require_a_limit(self) -> "BudgetSpec": + if all( + value is None + for value in ( + self.max_proposals, + self.max_validations, + self.max_tokens, + self.max_runtime_seconds, + self.deadline, + ) + ): + raise ValueError("at least one search budget limit must be configured") + return self + + +class DomainSpec(ContractModel): + schema_version: int = Field(default=SCHEMA_VERSION, ge=1) + dataset: DatasetSpec + bundle_target: str = Field(pattern=r"^(default|lme|beam)$") + benchmark_runner: str = Field(min_length=1) + scorer: str = Field(min_length=1) + objective: str = Field(default="mean_query_score", pattern=r"^mean_query_score$") + baseline_files: list[str] = Field(default_factory=list) + scope: ScopeSpec + sandbox: SandboxSpec + proposer: ProposerSpec + validation: ValidationPolicySpec = Field(default_factory=ValidationPolicySpec) + budget: BudgetSpec + + +class QuerySpec(ContractModel): + query_id: str = Field(min_length=1) + question: str + golden_answer: Any + metadata: dict[str, Any] = Field(default_factory=dict) + + +class SessionSpec(ContractModel): + session_id: str = Field(min_length=1) + messages: list[dict[str, Any]] + metadata: dict[str, Any] = Field(default_factory=dict) + + +class CaseSpec(ContractModel): + case_id: str = Field(min_length=1) + sessions: list[SessionSpec] + queries: list[QuerySpec] + metadata: dict[str, Any] = Field(default_factory=dict) + + +class DatasetManifest(ContractModel): + schema_version: int = Field(default=SCHEMA_VERSION, ge=1) + dataset: DatasetName + source_fingerprint: str = Field(min_length=1) + normalized_fingerprint: str = Field(min_length=1) + case_count: int = Field(ge=0) + query_count: int = Field(ge=0) + created_at: datetime = Field(default_factory=utc_now) + + +class Fingerprints(ContractModel): + dataset: str = Field(min_length=1) + code: str = Field(min_length=1) + config: str = Field(min_length=1) + model: str = Field(min_length=1) + image: str = Field(min_length=1) + + +class ValidationSelection(ContractModel): + """A fixed case/query subset; no selection means the complete dataset.""" + + case_ids: list[str] | None = None + query_ids: dict[str, list[str]] = Field(default_factory=dict) + reason: str = Field(default="full dataset", min_length=1) + + @model_validator(mode="after") + def validate_selection(self) -> "ValidationSelection": + if self.case_ids is not None: + if not self.case_ids: + raise ValueError("case_ids cannot be an empty selection") + if len(self.case_ids) != len(set(self.case_ids)): + raise ValueError("case_ids must be unique") + unknown_cases = set(self.query_ids) - set(self.case_ids) + if unknown_cases: + raise ValueError(f"query selections reference unselected cases: {sorted(unknown_cases)}") + for case_id, query_ids in self.query_ids.items(): + if not case_id or not query_ids: + raise ValueError("query_ids must contain non-empty case and query selections") + if len(query_ids) != len(set(query_ids)): + raise ValueError(f"query IDs must be unique for case {case_id!r}") + return self + + @property + def requests_full_dataset(self) -> bool: + return self.case_ids is None and not self.query_ids + + +class ValidationSpec(ContractModel): + validation_id: str = Field(default_factory=lambda: uuid4().hex) + commit_sha: str = Field(min_length=1) + mode: EvaluationMode + selection: ValidationSelection = Field(default_factory=ValidationSelection) + selection_fingerprint: str = "" + fingerprints: Fingerprints + + @model_validator(mode="after") + def validate_selection_fingerprint(self) -> "ValidationSpec": + expected = model_fingerprint(self.selection) + if not self.selection_fingerprint: + self.selection_fingerprint = expected + elif self.selection_fingerprint != expected: + raise ValueError("selection_fingerprint does not match validation selection") + return self + + +class ValidationCoverage(ContractModel): + selected_cases: int = Field(ge=0) + total_cases: int = Field(ge=0) + selected_queries: int = Field(ge=0) + total_queries: int = Field(ge=0) + is_full: bool + + @model_validator(mode="after") + def validate_coverage(self) -> "ValidationCoverage": + if self.selected_cases > self.total_cases or self.selected_queries > self.total_queries: + raise ValueError("selected validation coverage cannot exceed dataset totals") + expected_full = self.selected_cases == self.total_cases and self.selected_queries == self.total_queries + if self.is_full != expected_full: + raise ValueError("is_full does not match selected and total coverage") + return self + + +class BudgetUsage(ContractModel): + proposals: int = Field(default=0, ge=0) + validations: int = Field(default=0, ge=0) + tokens: int | None = Field(default=0, ge=0) + runtime_seconds: float = Field(default=0, ge=0) + has_unknown_usage: bool = False + + +class RunState(ContractModel): + schema_version: int = Field(default=SCHEMA_VERSION, ge=1) + run_id: str = Field(default_factory=lambda: uuid4().hex) + status: RunStatus = RunStatus.INITIALIZING + last_event_seq: int = Field(default=0, ge=0) + current_round: int = Field(default=0, ge=0) + baseline_commit: str | None = None + best_commit: str | None = None + usage: BudgetUsage = Field(default_factory=BudgetUsage) + created_at: datetime = Field(default_factory=utc_now) + updated_at: datetime = Field(default_factory=utc_now) + + +class SearchEvent(ContractModel): + schema_version: int = Field(default=SCHEMA_VERSION, ge=1) + seq: int = Field(ge=1) + event_id: str = Field(default_factory=lambda: uuid4().hex) + event_type: EventType + timestamp: datetime = Field(default_factory=utc_now) + round_id: int | None = Field(default=None, ge=0) + proposal_id: str | None = None + commit_sha: str | None = None + validation_id: str | None = None + case_id: str | None = None + attempt_id: str | None = None + payload: dict[str, Any] = Field(default_factory=dict) + + +class WeaknessEvidence(ContractModel): + query_id: str + observation: str + + +class WeaknessReport(ContractModel): + weakness_id: str = Field(default_factory=lambda: uuid4().hex) + round_id: int = Field(ge=0) + patterns: list[str] + evidence: list[WeaknessEvidence] + possible_causes: list[str] + regression_risks: list[str] = Field(default_factory=list) + input_fingerprint: str + + +class Proposal(ContractModel): + proposal_id: str = Field(default_factory=lambda: uuid4().hex) + round_id: int = Field(ge=0) + parent_commits: list[str] = Field(min_length=1) + hypothesis: str + expected_improvements: list[str] + merge_rationale: str | None = None + input_fingerprint: str + + +class ScopeCheckResult(ContractModel): + passed: bool + changed_paths: list[str] + violations: list[str] = Field(default_factory=list) + + +class HarnessManifest(ContractModel): + commit_sha: str = Field(min_length=1) + parent_commits: list[str] + snapshot_sha256: str = Field(min_length=1) + config_fingerprint: str = Field(min_length=1) + proposal_id: str | None = None + status: HarnessStatus + scope: ScopeCheckResult + created_at: datetime = Field(default_factory=utc_now) + + +class QueryResult(ContractModel): + query_id: str + question: str + golden_answer: Any + answer: str | None = None + score: float + judge: dict[str, Any] = Field(default_factory=dict) + error: str | None = None + tokens: int | None = Field(default=None, ge=0) + duration_seconds: float | None = Field(default=None, ge=0) + + +class CaseResult(ContractModel): + case_id: str + attempt_id: str + status: AttemptStatus + selected_query_ids: list[str] = Field(default_factory=list) + queries: list[QueryResult] = Field(default_factory=list) + completed_stages: list[str] = Field(default_factory=list) + error: str | None = None + artifacts: dict[str, str] = Field(default_factory=dict) + + @model_validator(mode="after") + def validate_terminal_result(self) -> "CaseResult": + if self.status is AttemptStatus.COMPLETED and not self.queries: + raise ValueError("completed case result must contain at least one query") + if len(self.selected_query_ids) != len(set(self.selected_query_ids)): + raise ValueError("selected_query_ids must be unique") + result_query_ids = {query.query_id for query in self.queries} + if self.queries and result_query_ids != set(self.selected_query_ids): + raise ValueError("case query results must match selected_query_ids") + return self + + +class AttemptCompletion(ContractModel): + schema_version: int = Field(default=SCHEMA_VERSION, ge=1) + case_id: str + attempt_id: str + status: AttemptStatus + fingerprints: Fingerprints + selection_fingerprint: str = Field(min_length=1) + result_file: str = "case_result.json" + result_sha256: str = Field(min_length=1) + artifact_sha256: dict[str, str] = Field(default_factory=dict) + completed_at: datetime = Field(default_factory=utc_now) + + @field_validator("status") + @classmethod + def require_reusable_status(cls, value: AttemptStatus) -> AttemptStatus: + if value not in (AttemptStatus.COMPLETED, AttemptStatus.CANDIDATE_FAILURE): + raise ValueError("complete marker requires a reusable terminal status") + return value + + +class ValidationResult(ContractModel): + validation_id: str + commit_sha: str + mode: EvaluationMode + status: ValidationStatus + fingerprints: Fingerprints + selection: ValidationSelection + coverage: ValidationCoverage + comparable: bool = False + mean_query_score: float | None = None + query_count: int = Field(default=0, ge=0) + failure_count: int = Field(default=0, ge=0) + cases: list[CaseResult] = Field(default_factory=list) + created_at: datetime = Field(default_factory=utc_now) + + @model_validator(mode="after") + def validate_score(self) -> "ValidationResult": + if self.status is ValidationStatus.COMPLETED and self.mean_query_score is None: + raise ValueError("completed validation must contain mean_query_score") + if self.status is ValidationStatus.INFRA_ERROR and self.mean_query_score is not None: + raise ValueError("infra_error validation cannot have a comparable score") + expected_comparable = ( + self.status is ValidationStatus.COMPLETED + and self.mode in (EvaluationMode.SEARCH, EvaluationMode.REPLAY) + and self.coverage.is_full + ) + if self.comparable != expected_comparable: + raise ValueError("comparable requires a completed full search or replay validation") + return self + + +class LeaderboardEntry(ContractModel): + commit_sha: str + validation_id: str + mean_query_score: float + completed_at: datetime + proposal_id: str | None = None + + +class WorkspaceManifest(ContractModel): + schema_version: int = Field(default=SCHEMA_VERSION, ge=1) + workspace_id: str = Field(default_factory=lambda: uuid4().hex) + domain_fingerprint: str + created_at: datetime = Field(default_factory=utc_now) + + +class WorkspaceLockOwner(ContractModel): + token: str = Field(default_factory=lambda: uuid4().hex) + pid: int = Field(gt=0) + hostname: str + started_at: datetime = Field(default_factory=utc_now) + + +def model_fingerprint(model: BaseModel | Mapping[str, Any]) -> str: + """Convenience alias documenting fingerprints derived from contracts.""" + + return fingerprint(model) diff --git a/meta-reme/workspace.py b/meta-reme/workspace.py new file mode 100644 index 00000000..ced78e8f --- /dev/null +++ b/meta-reme/workspace.py @@ -0,0 +1,317 @@ +"""Filesystem boundary and lifecycle management for a Meta-ReMe workspace.""" + +from __future__ import annotations + +from contextlib import AbstractContextManager +import errno +import json +import os +from pathlib import Path +import shutil +import socket +import stat +import tempfile +from typing import Any + +from pydantic import BaseModel, ValidationError +import yaml + +from models import DomainSpec, WorkspaceLockOwner, WorkspaceManifest, canonical_json_bytes, model_fingerprint + +WORKSPACE_MANIFEST = ".meta-reme-workspace.json" +WORKSPACE_LOCK = ".meta-reme.lock" +_DIRECTORIES = ( + "code/repo", + "code/worktrees", + "datasets/search/cases", + "harnesses", + "weaknesses", + "proposals", + "evaluations", + "logs", +) + + +class WorkspaceError(RuntimeError): + """Base error for invalid or unsafe workspace operations.""" + + +class WorkspaceLockedError(WorkspaceError): + """Raised when another process owns a workspace lock.""" + + +class WorkspaceFormatError(WorkspaceError): + """Raised when a directory is not a valid compatible workspace.""" + + +class Workspace: + """A validated Meta-ReMe workspace rooted at an absolute path.""" + + def __init__(self, root: Path, manifest: WorkspaceManifest) -> None: + self.root = Path(root).resolve() + self.manifest = manifest + + @classmethod + def create(cls, root: Path, domain_spec: DomainSpec) -> "Workspace": + """Create a new workspace without overwriting an existing directory.""" + + root = Path(root).resolve() + if root.exists() and (not root.is_dir() or any(root.iterdir())): + raise WorkspaceError(f"Refusing to initialize non-empty workspace: {root}") + + root.mkdir(parents=True, exist_ok=True) + manifest = WorkspaceManifest(domain_fingerprint=model_fingerprint(domain_spec)) + workspace = cls(root, manifest) + try: + for relative in _DIRECTORIES: + (root / relative).mkdir(parents=True, exist_ok=False) + workspace.atomic_write_text( + "domain_spec.yaml", + yaml.safe_dump( + domain_spec.model_dump(mode="json", exclude_none=False), + allow_unicode=True, + sort_keys=True, + ), + ) + # The manifest is the publication marker and must be written last. + workspace.atomic_write_json(WORKSPACE_MANIFEST, manifest) + return workspace + except BaseException: + # Only remove a directory created by this call and only while it has + # not yet become a published workspace. + if not (root / WORKSPACE_MANIFEST).exists(): + shutil.rmtree(root, ignore_errors=True) + raise + + @classmethod + def open(cls, root: Path, domain_spec: DomainSpec | None = None) -> "Workspace": + """Open and validate an existing workspace.""" + + root = Path(root).resolve() + manifest_path = root / WORKSPACE_MANIFEST + if not root.is_dir() or not manifest_path.is_file(): + raise WorkspaceFormatError(f"Not a Meta-ReMe workspace: {root}") + try: + 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 + for relative in _DIRECTORIES: + if not (root / relative).is_dir(): + raise WorkspaceFormatError(f"Workspace directory is missing: {relative}") + if domain_spec is not None: + actual = model_fingerprint(domain_spec) + if actual != manifest.domain_fingerprint: + raise WorkspaceFormatError("Domain spec fingerprint does not match this workspace") + return cls(root, manifest) + + def path(self, relative: str | Path) -> Path: + """Resolve a safe workspace-relative path.""" + + relative = Path(relative) + if relative.is_absolute() or ".." in relative.parts: + raise WorkspaceError(f"Path must be workspace-relative: {relative}") + resolved = (self.root / relative).resolve() + try: + resolved.relative_to(self.root) + except ValueError as exc: + raise WorkspaceError(f"Path escapes workspace: {relative}") from exc + return resolved + + def entity_path(self, base: str | Path, identifier: str) -> Path: + """Resolve a path for an externally supplied case/commit/attempt ID.""" + + if not identifier or identifier in {".", ".."} or Path(identifier).name != identifier: + raise WorkspaceError(f"Unsafe workspace identifier: {identifier!r}") + if any(character in identifier for character in ("/", "\\", "\x00")): + raise WorkspaceError(f"Unsafe workspace identifier: {identifier!r}") + return self.path(Path(base) / identifier) + + def case_attempt_dir( + self, + commit_sha: str, + mode: str, + case_id: str, + attempt_id: str, + *, + create: bool = False, + ) -> 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(current.relative_to(self.root), identifier) + if create: + current.mkdir(parents=True, exist_ok=False) + return current + + def atomic_write_json(self, relative: str | Path, value: BaseModel | Any) -> Path: + """Atomically publish JSON in the workspace.""" + + if isinstance(value, BaseModel): + value = value.model_dump(mode="json", exclude_none=False) + content = json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True, allow_nan=False) + "\n" + return self.atomic_write_text(relative, content) + + def atomic_write_text(self, relative: str | Path, content: str) -> Path: + """Atomically publish UTF-8 text in the workspace.""" + + return self.atomic_write_bytes(relative, content.encode("utf-8")) + + def atomic_write_bytes(self, relative: str | Path, content: bytes) -> Path: + """Write, fsync, and atomically replace a file in its destination directory.""" + + destination = self.path(relative) + destination.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{destination.name}.", dir=destination.parent) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(content) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, destination) + _fsync_directory(destination.parent) + return destination + except BaseException: + temporary.unlink(missing_ok=True) + raise + + def acquire_lock(self) -> "WorkspaceLock": + """Acquire the single-writer workspace lock.""" + + lock = WorkspaceLock(self.path(WORKSPACE_LOCK)) + 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.""" + + source = Path(source).resolve() + destination = self.path("datasets/search") + 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}") + 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") + if temporary.exists(): + raise WorkspaceError(f"Incomplete dataset installation requires inspection: {temporary}") + shutil.copytree(source, temporary, symlinks=True) + try: + shutil.rmtree(destination) + os.replace(temporary, destination) + _make_tree_read_only(destination) + _fsync_directory(destination.parent) + return destination + except BaseException: + if not destination.exists(): + shutil.rmtree(temporary, ignore_errors=True) + destination.mkdir(parents=True, exist_ok=True) + raise + + +class WorkspaceLock(AbstractContextManager["WorkspaceLock"]): + """An atomic, ownership-checked filesystem lock.""" + + def __init__(self, path: Path) -> None: + self.path = path + self.owner: WorkspaceLockOwner | None = None + + def acquire(self) -> None: + """Acquire this lock, taking over only a confirmed stale local owner.""" + + if self.owner is not None: + raise WorkspaceLockedError(f"Lock is already held by this object: {self.path}") + owner = WorkspaceLockOwner(pid=os.getpid(), hostname=socket.gethostname()) + for _ in range(2): + try: + descriptor = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError as exc: + existing = self._read_owner() + if not _is_confirmed_stale(existing): + raise WorkspaceLockedError(_lock_message(self.path, existing)) from exc + try: + self.path.unlink() + except FileNotFoundError: + pass + continue + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(canonical_json_bytes(owner) + b"\n") + stream.flush() + os.fsync(stream.fileno()) + _fsync_directory(self.path.parent) + except BaseException: + self.path.unlink(missing_ok=True) + raise + self.owner = owner + return + raise WorkspaceLockedError(f"Could not acquire workspace lock: {self.path}") + + def _read_owner(self) -> WorkspaceLockOwner | None: + """Read the current lock owner, returning None for an invalid lock.""" + + try: + return WorkspaceLockOwner.model_validate_json(self.path.read_text(encoding="utf-8")) + except (OSError, ValidationError): + return None + + def release(self) -> None: + """Release the lock only if its ownership token is unchanged.""" + + if self.owner is None: + return + existing = self._read_owner() + if existing is None or existing.token != self.owner.token: + raise WorkspaceLockedError(f"Workspace lock ownership changed: {self.path}") + self.path.unlink() + _fsync_directory(self.path.parent) + self.owner = None + + def __enter__(self) -> "WorkspaceLock": + return self + + def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None: + self.release() + + +def _is_confirmed_stale(owner: WorkspaceLockOwner | None) -> bool: + if owner is None or owner.hostname != socket.gethostname(): + return False + try: + os.kill(owner.pid, 0) + except ProcessLookupError: + return True + except PermissionError: + return False + except OSError as exc: + return exc.errno == errno.ESRCH + return False + + +def _lock_message(path: Path, owner: WorkspaceLockOwner | None) -> str: + if owner is None: + return f"Workspace has an unreadable lock requiring manual inspection: {path}" + return f"Workspace is locked by pid {owner.pid} on {owner.hostname} since {owner.started_at.isoformat()}" + + +def _fsync_directory(directory: Path) -> None: + flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + descriptor = os.open(directory, flags) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _make_tree_read_only(root: Path) -> None: + for path in sorted(root.rglob("*"), reverse=True): + mode = stat.S_IMODE(path.stat().st_mode) + if path.is_dir(): + path.chmod(mode & ~0o222 | 0o555) + else: + path.chmod(mode & ~0o222 | 0o444) + mode = stat.S_IMODE(root.stat().st_mode) + root.chmod(mode & ~0o222 | 0o555) diff --git a/sandbox/README.md b/sandbox/README.md index 2d8534a2..172d3909 100644 --- a/sandbox/README.md +++ b/sandbox/README.md @@ -150,6 +150,37 @@ and initializes a fresh repository for the next case. For an ephemeral Docker workspace, always call `export()` or `export_full()` before `close()`. +## Reuse one built memory for parallel queries + +Use `export_workspace()` after constructing the memory, then give the snapshot +to each independent query case with `upload_workspace()`. Upload clears the +target runtime workspace by default, so the cases start from the same memory +state but never share mutable files or indexes. + +```python +memory_case = await factory.create_case("build-memory") +try: + await memory_case.ingest_session(session_id="source", messages=[...]) + await memory_case.run_job("index_update") + snapshot = await memory_case.export_workspace("artifacts/built-memory.tar.gz") +finally: + await memory_case.close() + +query_cases = await factory.create_cases(["query-1", "query-2"]) +try: + await asyncio.gather(*(case.upload_workspace(snapshot) for case in query_cases)) + answers = await asyncio.gather( + *(case.answer(query=query) for case, query in zip(query_cases, queries)) + ) +finally: + await asyncio.gather(*(case.close() for case in query_cases)) +``` + +`upload_workspace()` also accepts a host directory. It validates and repacks +the directory or input archive before extraction; symbolic links, special +files, and traversal paths are rejected. Pass `clear=False` only when an +intentional merge into the target workspace is required. + ## Reuse one container for sequential cases When container startup and source installation dominate the benchmark, one diff --git a/sandbox/docker_sandbox.py b/sandbox/docker_sandbox.py index bde0b23c..a5fd8eef 100644 --- a/sandbox/docker_sandbox.py +++ b/sandbox/docker_sandbox.py @@ -5,10 +5,13 @@ from __future__ import annotations import asyncio from dataclasses import asdict from datetime import datetime, timezone +import gzip +import io import json from pathlib import Path from pathlib import PurePosixPath import re +import tarfile from typing import Any, Callable, Literal, Protocol import uuid @@ -21,6 +24,8 @@ CANDIDATE_VENV = "/workspace/candidate-venv" HARNESS_ROOT = "/workspace/harness" SOURCE_ARCHIVE = "/workspace/candidate.tar.gz" EXPORT_ARCHIVE = "/tmp/reme-case-export.tar.gz" +WORKSPACE_ARCHIVE = "/tmp/reme-workspace-upload.tar.gz" +WORKSPACE_EXPORT_ARCHIVE = "/tmp/reme-workspace-export.tar.gz" RUNTIME_LAYOUT = f"{CASE_ROOT}/runtime-layout.json" _CASE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$") _GIT_AUTHOR_NAME = "ReMe Sandbox" @@ -90,6 +95,107 @@ def _decode(value: bytes | str | None) -> str: return value or "" +def _workspace_member_path(name: str) -> PurePosixPath | None: + """Validate and normalize one relative path in a workspace archive.""" + path = PurePosixPath(name) + if not name or path.is_absolute() or ".." in path.parts or "\\" in name: + raise ValueError(f"workspace archive contains unsafe path: {name!r}") + if path == PurePosixPath("."): + return None + return path + + +def _archive_workspace(source: str | Path) -> bytes: + """Create a safe, deterministic gzip tar archive from a host workspace. + + ``source`` may be a workspace directory or an archive previously returned + by :meth:`DockerReMeSandbox.export_workspace`. Incoming archives are read + and rebuilt so links, special files, and traversal paths never reach the + sandbox extractor. + """ + supplied_path = Path(source).expanduser() + if supplied_path.is_symlink(): + raise ValueError("workspace source must not be a symbolic link") + source_path = supplied_path.resolve(strict=True) + entries: list[tuple[PurePosixPath, tarfile.TarInfo, bytes | None]] = [] + if source_path.is_dir(): + for item in sorted(source_path.rglob("*"), key=lambda value: value.relative_to(source_path).as_posix()): + relative = item.relative_to(source_path) + if item.is_symlink(): + raise ValueError(f"workspace source contains a symbolic link: {relative}") + path = _workspace_member_path(relative.as_posix()) + assert path is not None + if item.is_dir(): + info = tarfile.TarInfo(path.as_posix()) + info.type = tarfile.DIRTYPE + info.mode = item.stat().st_mode & 0o777 + entries.append((path, info, None)) + elif item.is_file(): + info = tarfile.TarInfo(path.as_posix()) + info.size = item.stat().st_size + info.mode = item.stat().st_mode & 0o777 + entries.append((path, info, item.read_bytes())) + else: + raise ValueError(f"workspace source contains unsupported file: {relative}") + elif source_path.is_file(): + with tarfile.open(source_path, mode="r:*") as archive: + members = archive.getmembers() + normalized_paths = [ + path for member in members if (path := _workspace_member_path(member.name)) is not None + ] + # ``export()`` archives the whole case, while ``export_workspace`` + # archives only the workspace contents. Accept both forms so an + # existing case artifact can seed independent query sandboxes. + case_export = ( + PurePosixPath("manifest.json") in normalized_paths + and any(path.parts and path.parts[0] == "reme_workspace" for path in normalized_paths) + ) + seen: set[PurePosixPath] = set() + for member in members: + path = _workspace_member_path(member.name) + if path is None: + continue + if case_export: + if not path.parts or path.parts[0] != "reme_workspace": + continue + if len(path.parts) == 1: + if not member.isdir(): + raise ValueError("case export has a non-directory reme_workspace entry") + continue + path = PurePosixPath(*path.parts[1:]) + if path in seen: + raise ValueError(f"workspace archive contains duplicate path: {path}") + seen.add(path) + if member.isdir(): + info = tarfile.TarInfo(path.as_posix()) + info.type = tarfile.DIRTYPE + info.mode = member.mode & 0o777 + entries.append((path, info, None)) + elif member.isfile(): + content = archive.extractfile(member) + if content is None: + raise ValueError(f"workspace archive cannot read file: {path}") + data = content.read() + info = tarfile.TarInfo(path.as_posix()) + info.size = len(data) + info.mode = member.mode & 0o777 + entries.append((path, info, data)) + else: + raise ValueError(f"workspace archive contains unsupported entry: {member.name!r}") + else: + raise ValueError(f"workspace source is neither a directory nor a file: {source_path}") + + raw = io.BytesIO() + with gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as compressed: + with tarfile.open(fileobj=compressed, mode="w") as archive: + for _path, info, data in entries: + info.mtime = 0 + info.uid = info.gid = 0 + info.uname = info.gname = "" + archive.addfile(info, io.BytesIO(data) if data is not None else None) + return raw.getvalue() + + def _default_workspace_builder(case_id: str, image: str, env: dict[str, str]) -> Workspace: """Create the exact AgentScope 2.0.4.post1 Docker backend lazily.""" import agentscope @@ -492,6 +598,54 @@ class DockerReMeSandbox: """Download every file under the disposable case root.""" return await self.export(destination, profile="full", include_candidate=include_candidate) + async def export_workspace(self, destination: str | Path) -> Path: + """Download a portable snapshot of this runtime workspace to the host. + + The resulting gzip tar archive can be passed directly to + :meth:`upload_workspace` on any initialized case. It contains the + workspace contents (including its local Git history), not the enclosing + case directory, logs, or candidate source. + """ + async with self._operation_lock: + assert self.backend is not None, "sandbox is not initialized" + await self._exec_checked( + "export_workspace", + ["tar", "-czf", WORKSPACE_EXPORT_ARCHIVE, "-C", self.runtime_workspace, "."], + ) + target = Path(destination).expanduser().resolve() + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(await self.backend.read_file(WORKSPACE_EXPORT_ARCHIVE)) + return target + + async def upload_workspace(self, source: str | Path, *, clear: bool = True) -> None: + """Copy a host workspace directory or exported archive into this case. + + By default the existing runtime workspace is deleted first, ensuring a + query case receives only the supplied memory state. Set ``clear=False`` + only when intentionally merging files into the current workspace. + The copy is made through a validated archive, so each sandbox receives + an independent filesystem tree and can run queries in parallel. + """ + archive = _archive_workspace(source) + async with self._operation_lock: + assert self.backend is not None, "sandbox is not initialized" + if clear: + await self._exec_checked("clear_runtime_workspace", ["rm", "-rf", self.runtime_workspace]) + await self._exec_checked("create_runtime_workspace", ["mkdir", "-p", self.runtime_workspace]) + await self.backend.write_file(WORKSPACE_ARCHIVE, archive) + await self._exec_checked( + "upload_workspace", + ["tar", "-xzf", WORKSPACE_ARCHIVE, "-C", self.runtime_workspace], + ) + # A directory copied from the host may not have Git metadata. Git + # init is idempotent and preserves history when the snapshot has it. + await self._exec_checked( + "initialize_uploaded_memory_history", + ["git", "init", "--quiet"], + cwd=self.runtime_workspace, + ) + await self._exec_checked("remove_workspace_upload", ["rm", "-f", WORKSPACE_ARCHIVE]) + @staticmethod def _validated_workspace_path(value: Any, workspace_root: PurePosixPath, *, allow_root: bool) -> str: """Validate one case-relative path received from candidate-controlled state.""" @@ -612,7 +766,7 @@ class DockerReMeSandbox: assert self.backend is not None, "sandbox is not initialized" await self._exec_checked( "clear_case", - ["rm", "-rf", CASE_ROOT, EXPORT_ARCHIVE], + ["rm", "-rf", CASE_ROOT, EXPORT_ARCHIVE, WORKSPACE_ARCHIVE, WORKSPACE_EXPORT_ARCHIVE], ) await self._create_case_layout() self.case_id = case_id diff --git a/tests/unit_sandbox/test_bundle_builder.py b/tests/unit_sandbox/test_bundle_builder.py new file mode 100644 index 00000000..6d53a204 --- /dev/null +++ b/tests/unit_sandbox/test_bundle_builder.py @@ -0,0 +1,68 @@ +"""Focused coverage for the sealed benchmark bundle.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import yaml + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +BUILDER_PATH = PROJECT_ROOT / "meta-reme/bundle_builder.py" + + +def _load_builder(): + spec = importlib.util.spec_from_file_location("reme_bundle_builder", BUILDER_PATH) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_bundle_prunes_unused_runtime_backends(tmp_path: Path) -> None: + """Generated benchmark bundles omit backends outside the sealed runtime contract.""" + builder = _load_builder() + bundle = builder.build_bundle("default", tmp_path, validate=True) + + forbidden = ( + "reme/components/client/http_client.py", + "reme/components/client/mcp_client.py", + "reme/components/file_graph/neo4j_file_graph.py", + "reme/components/file_graph/nx_file_graph.py", + "reme/components/file_store/faiss_local_file_store.py", + "reme/components/file_store/zvec_local_file_store.py", + "reme/components/service/http_service.py", + "reme/components/service/mcp_service.py", + "reme/components/tokenizer/jieba_tokenizer.py", + "reme/schema/auto_fin.py", + "reme/schema/daily_paper.py", + "reme/steps/common/add.py", + "reme/steps/common/demo.py", + "reme/steps/common/llm_demo.py", + "reme/steps/common/stream_demo.py", + "reme/steps/common/stream_llm_demo.py", + "reme/steps/common/version.py", + "reme/steps/file_io/read_image.py", + "reme/utils/arxiv.py", + "reme/utils/huggingface_papers.py", + ) + assert all(not (bundle / path).exists() for path in forbidden) + + config = yaml.safe_load((bundle / "reme/config/default.yaml").read_text(encoding="utf-8")) + assert config["service"]["backend"] == "cli" + assert "read_image" not in config["jobs"] + assert "version" not in config["jobs"] + assert {"dream_cron", "optimize_index_cron"}.issubset(config["jobs"]) + assert not list(bundle.rglob("__pycache__")) + assert not (bundle / "README.md").exists() + assert not (bundle / "LICENSE").exists() + assert (bundle / "pyproject.toml").is_file() + + embedding_source = (bundle / "reme/components/as_embedding/__init__.py").read_text(encoding="utf-8") + assert "dashscope_multimodal" not in embedding_source + + assert (bundle / "reme/steps/evolve/auto_memory_cc.py").is_file() + benchmark_bundle = builder.build_bundle("lme", tmp_path / "lme", validate=True) + assert not (benchmark_bundle / "reme/steps/evolve/auto_memory_cc.py").exists() + assert "AutoMemoryCCStep" not in (benchmark_bundle / "reme/steps/evolve/__init__.py").read_text(encoding="utf-8") + assert not list(benchmark_bundle.rglob("__pycache__")) diff --git a/tests/unit_sandbox/test_meta_reme_models_workspace.py b/tests/unit_sandbox/test_meta_reme_models_workspace.py new file mode 100644 index 00000000..40b907bb --- /dev/null +++ b/tests/unit_sandbox/test_meta_reme_models_workspace.py @@ -0,0 +1,177 @@ +"""Focused contract and filesystem tests for Meta-ReMe's initial foundation.""" + +# pylint: disable=missing-function-docstring + +from __future__ import annotations + +import importlib +import json +import os +from pathlib import Path +import socket +import sys + +import pytest +from pydantic import ValidationError + +META_REME = Path(__file__).resolve().parents[2] / "meta-reme" +sys.path.insert(0, str(META_REME)) +models = importlib.import_module("models") +workspace_module = importlib.import_module("workspace") + + +def domain_spec(): + return models.DomainSpec( + dataset=models.DatasetSpec(name="longmemeval", source="dataset.json", fingerprint="dataset-sha"), + bundle_target="lme", + benchmark_runner="benchmark.longmemeval.run", + scorer="lme_answer_judge_step", + scope=models.ScopeSpec(harness_paths=["reme/steps/benchmark/lme/auto_memory.py"]), + sandbox=models.SandboxSpec(image="reme:test", timeout_seconds=60), + proposer=models.ProposerSpec(model="test-model"), + budget=models.BudgetSpec(max_proposals=2), + ) + + +def test_fingerprint_is_stable_for_mapping_order() -> None: + assert models.fingerprint({"a": 1, "b": [2, 3]}) == models.fingerprint({"b": [2, 3], "a": 1}) + + +def test_contracts_reject_unknown_fields_and_invalid_terminal_states() -> None: + with pytest.raises(ValidationError): + models.DatasetSpec(name="longmemeval", source="x", fingerprint="sha", typo=True) + with pytest.raises(ValidationError, match="at least one search budget"): + models.BudgetSpec() + with pytest.raises(ValidationError, match="at least one query"): + models.CaseResult(case_id="case", attempt_id="attempt", status="completed") + with pytest.raises(ValidationError, match="reusable terminal status"): + models.AttemptCompletion( + case_id="case", + attempt_id="attempt", + status="infra_error", + fingerprints=models.Fingerprints(dataset="d", code="c", config="f", model="m", image="i"), + selection_fingerprint="selection-sha", + result_sha256="sha", + ) + + +def test_partial_validation_selection_is_not_comparable() -> None: + selection = models.ValidationSelection( + case_ids=["case-1"], + query_ids={"case-1": ["query-1"]}, + reason="target known failure", + ) + coverage = models.ValidationCoverage( + selected_cases=1, + total_cases=10, + selected_queries=1, + total_queries=100, + is_full=False, + ) + common = { + "validation_id": "validation-1", + "commit_sha": "commit", + "mode": "screening", + "status": "completed", + "fingerprints": models.Fingerprints(dataset="d", code="c", config="f", model="m", image="i"), + "selection": selection, + "coverage": coverage, + "mean_query_score": 1.0, + "query_count": 1, + } + + result = models.ValidationResult(**common) + assert not result.comparable + spec = models.ValidationSpec( + commit_sha="commit", + mode="screening", + selection=selection, + fingerprints=common["fingerprints"], + ) + assert spec.selection_fingerprint == models.model_fingerprint(selection) + with pytest.raises(ValidationError, match="comparable requires"): + models.ValidationResult(**common, comparable=True) + + +def test_validation_selection_rejects_queries_from_unselected_cases() -> None: + with pytest.raises(ValidationError, match="unselected cases"): + models.ValidationSelection(case_ids=["case-1"], query_ids={"case-2": ["query-1"]}, reason="targeted") + + +def test_workspace_create_open_and_domain_fingerprint(tmp_path: Path) -> None: + root = tmp_path / "workspace" + spec = domain_spec() + workspace = workspace_module.Workspace.create(root, spec) + + assert workspace.path("code/repo").is_dir() + assert workspace.path("datasets/search/cases").is_dir() + assert workspace_module.Workspace.open(root, spec).manifest == workspace.manifest + + changed = spec.model_copy(update={"scorer": "different"}) + with pytest.raises(workspace_module.WorkspaceFormatError, match="fingerprint"): + workspace_module.Workspace.open(root, changed) + with pytest.raises(workspace_module.WorkspaceError, match="non-empty"): + workspace_module.Workspace.create(root, 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): + workspace.path("../outside") + with pytest.raises(workspace_module.WorkspaceError): + workspace.entity_path("evaluations", "../case") + + +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()) + destination = workspace.atomic_write_json("run.json", {"value": 1}) + workspace.atomic_write_json("run.json", {"value": 2}) + + assert json.loads(destination.read_text(encoding="utf-8")) == {"value": 2} + assert not list(destination.parent.glob(".run.json.*")) + + +def test_install_search_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) + + 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) + + +def test_workspace_lock_rejects_live_owner_and_releases(tmp_path: Path) -> None: + workspace = workspace_module.Workspace.create(tmp_path / "workspace", domain_spec()) + lock = workspace.acquire_lock() + with pytest.raises(workspace_module.WorkspaceLockedError, match="locked by pid"): + workspace.acquire_lock() + lock.release() + with workspace.acquire_lock(): + assert workspace.path(workspace_module.WORKSPACE_LOCK).exists() + assert not workspace.path(workspace_module.WORKSPACE_LOCK).exists() + + +def test_workspace_lock_recovers_confirmed_stale_local_owner(tmp_path: Path) -> None: + workspace = workspace_module.Workspace.create(tmp_path / "workspace", domain_spec()) + stale = models.WorkspaceLockOwner(pid=_missing_pid(), hostname=socket.gethostname()) + workspace.atomic_write_json(workspace_module.WORKSPACE_LOCK, stale) + + with workspace.acquire_lock() as lock: + assert lock.owner is not None + assert lock.owner.token != stale.token + + +def _missing_pid() -> int: + candidate = max(os.getpid() + 1_000_000, 10_000_000) + while True: + try: + os.kill(candidate, 0) + except ProcessLookupError: + return candidate + candidate += 1 diff --git a/tests/unit/test_sandbox.py b/tests/unit_sandbox/test_sandbox.py similarity index 85% rename from tests/unit/test_sandbox.py rename to tests/unit_sandbox/test_sandbox.py index 1ee77869..cc8a7e09 100644 --- a/tests/unit/test_sandbox.py +++ b/tests/unit_sandbox/test_sandbox.py @@ -230,6 +230,96 @@ async def test_image_candidate_skips_source_upload_and_runs_jobs(tmp_path): assert "--exclude=reme_workspace/session" not in export_command +@pytest.mark.asyncio +async def test_workspace_can_be_cleared_and_uploaded_from_host(tmp_path): + """A completed memory can be copied into an independent query case.""" + source = tmp_path / "built-memory" + (source / "daily").mkdir(parents=True) + (source / "daily" / "2026-08-06.md").write_text("remember this\n") + workspace = FakeWorkspace("query-1", "candidate:test", {}) + factory = DockerReMeSandboxFactory( + ImageCandidate("candidate:test"), + workspace_builder=lambda *_args: workspace, + ) + case = await factory.create_case("query-1") + old_path = f"{case.runtime_workspace}/daily/old.md" + workspace.backend.files[old_path] = b"old memory" + + await case.upload_workspace(source) + + payload = workspace.backend.files["/tmp/reme-workspace-upload.tar.gz"] + with tarfile.open(fileobj=io.BytesIO(payload), mode="r:gz") as archive: + assert archive.getnames() == ["daily", "daily/2026-08-06.md"] + assert archive.extractfile("daily/2026-08-06.md").read() == b"remember this\n" + clear_index = next( + index + for index, (command, _, _) in enumerate(workspace.backend.commands) + if command == ["rm", "-rf", case.runtime_workspace] + ) + upload_index = next( + index + for index, (command, _, _) in enumerate(workspace.backend.commands) + if command == ["tar", "-xzf", "/tmp/reme-workspace-upload.tar.gz", "-C", case.runtime_workspace] + ) + assert clear_index < upload_index + assert old_path not in workspace.backend.files + assert (["git", "init", "--quiet"], case.runtime_workspace, case.command_timeout) in workspace.backend.commands + + +@pytest.mark.asyncio +async def test_workspace_upload_rejects_unsafe_host_archive(tmp_path): + """Host archives cannot use traversal paths to write outside the workspace.""" + source = tmp_path / "unsafe.tar.gz" + with tarfile.open(source, mode="w:gz") as archive: + info = tarfile.TarInfo("../outside.md") + info.size = 4 + archive.addfile(info, io.BytesIO(b"nope")) + workspace = FakeWorkspace("query-1", "candidate:test", {}) + factory = DockerReMeSandboxFactory( + ImageCandidate("candidate:test"), + workspace_builder=lambda *_args: workspace, + ) + case = await factory.create_case("query-1") + commands_before = list(workspace.backend.commands) + + with pytest.raises(ValueError, match="unsafe path"): + await case.upload_workspace(source) + + assert workspace.backend.commands == commands_before + + +@pytest.mark.asyncio +async def test_workspace_upload_accepts_a_complete_case_export(tmp_path): + """A previous ``export()`` archive restores only its runtime workspace.""" + source = tmp_path / "case-export.tar.gz" + with tarfile.open(source, mode="w:gz") as archive: + manifest = b'{"case_id": "source"}' + manifest_info = tarfile.TarInfo("manifest.json") + manifest_info.size = len(manifest) + archive.addfile(manifest_info, io.BytesIO(manifest)) + note = b"remember this\n" + note_info = tarfile.TarInfo("reme_workspace/daily/2023-05-30.md") + note_info.size = len(note) + archive.addfile(note_info, io.BytesIO(note)) + log = b"do not import case logs" + log_info = tarfile.TarInfo("logs/actions.jsonl") + log_info.size = len(log) + archive.addfile(log_info, io.BytesIO(log)) + workspace = FakeWorkspace("query-1", "candidate:test", {}) + factory = DockerReMeSandboxFactory( + ImageCandidate("candidate:test"), + workspace_builder=lambda *_args: workspace, + ) + case = await factory.create_case("query-1") + + await case.upload_workspace(source) + + payload = workspace.backend.files["/tmp/reme-workspace-upload.tar.gz"] + with tarfile.open(fileobj=io.BytesIO(payload), mode="r:gz") as archive: + assert archive.getnames() == ["daily/2023-05-30.md"] + assert archive.extractfile("daily/2023-05-30.md").read() == note + + @pytest.mark.asyncio async def test_host_selects_when_to_commit_configured_daily_directory(): """Several ingested sessions can be grouped into one explicit host checkpoint.""" @@ -329,6 +419,8 @@ async def test_one_container_can_reset_and_run_another_case(tmp_path): backend.files["/workspace/candidate/reme/__init__.py"] = b"candidate-code" old_git_object = "/workspace/case/reme_workspace/.git/objects/old-case-object" backend.files[old_git_object] = b"old case history" + backend.files["/tmp/reme-workspace-upload.tar.gz"] = b"uploaded workspace" + backend.files["/tmp/reme-workspace-export.tar.gz"] = b"exported workspace" assert any(path.startswith("/workspace/case/") for path in backend.files) await case.reset_case("case-2") @@ -341,6 +433,8 @@ async def test_one_container_can_reset_and_run_another_case(tmp_path): assert backend.files["/workspace/case/reme_workspace/.git/HEAD"].startswith(b"ref:") assert not any(path.startswith("/workspace/case/results/") for path in backend.files) assert "/tmp/reme-case-export.tar.gz" not in backend.files + assert "/tmp/reme-workspace-upload.tar.gz" not in backend.files + assert "/tmp/reme-workspace-export.tar.gz" not in backend.files git_init_positions = [ index for index, (command, _, _) in enumerate(backend.commands) if command == ["git", "init", "--quiet"] ]