Fix Home evaluation recovery and cancellation boundaries

This commit is contained in:
酱牛肉 2026-11-03 19:16:25 +08:00
parent 5c7c7b15d1
commit 7183890099
8 changed files with 354 additions and 62 deletions

View file

@ -3,7 +3,7 @@
from pathlib import Path
from ..evaluation.artifacts import contained, read_json
from ..preparation.code import git
from ..preparation.code import git, git_tree
from ..preparation.spec import relative_path
INITIAL_BRANCH = "init"
@ -135,11 +135,7 @@ def validate_tree(repository: Path, record: dict, revision: str) -> None:
path = Path(relative_path(name))
if ".git" in path.parts or not any(path.is_relative_to(root) for root in roots):
raise ValueError(f"Change outside candidate_paths: {name}")
for entry in filter(None, git(repository, "ls-tree", "-r", "-z", revision, strip=False).split("\0")):
header, name = entry.split("\t", 1)
mode, kind, _oid = header.split()
if mode not in {"100644", "100755"} or kind != "blob" or ".git" in Path(relative_path(name)).parts:
raise ValueError(f"Candidate must contain regular files only: {name}")
git_tree(repository, revision)
def merge_branch(repository: Path, record: dict, *, branch_name: str) -> dict:

View file

@ -6,7 +6,7 @@ from pathlib import Path
import subprocess
import sys
from ..preparation.code import git, git_bytes
from ..preparation.code import git, git_bytes, git_tree
from ..preparation.spec import relative_path
from .repository import MAX_OUTPUT, resolve_revision
@ -25,20 +25,13 @@ def code_path(path: str) -> str:
def tree(repository: Path, revision: str, path: str | None = None) -> list[tuple[str, str]]:
"""Read regular-file object IDs with NUL-delimited names and no archive filters."""
# ls-tree does not support pathspec magic. Filter exact names/prefixes ourselves.
entries = git(repository, "ls-tree", "-r", "-z", "--full-tree", revision, strip=False)
prefix = code_path(path) if path is not None else None
result = []
for entry in filter(None, entries.split("\0")):
header, name = entry.split("\t", 1)
if prefix is not None and name != prefix and not name.startswith(prefix + "/"):
continue
mode, kind, oid = header.split()
code_path(name)
if mode not in {"100644", "100755"} or kind != "blob":
raise ValueError(f"Code snapshot must contain regular files only: {name}")
result.append((name, oid))
return result
def selected(name):
# Browse literal names/prefixes before validating selected paths.
return prefix is None or name == prefix or name.startswith(prefix + "/")
return [(name.as_posix(), oid) for name, _mode, oid in git_tree(repository, revision, path_filter=selected)]
def blob(repository: Path, oid: str) -> bytes:

View file

@ -9,6 +9,7 @@ from contextlib import nullcontext
from deploy.benchmark_docker import BenchmarkDockerPool, CasePlan, DockerReMeSandboxFactory, SourceCandidate
from reme.components.base_component import BaseComponent
from reme.utils.async_utils import finish_cleanup
from ..preparation.data.models import utc_now
from .adapters import accepts_build, build_jobs, evaluation_query
@ -36,12 +37,7 @@ from .reuse import automatic_reuse, construction_fingerprints, reuse_context
async def blocking(function, *args, **kwargs):
"""Await owned filesystem work even when the calling job is cancelled."""
task = asyncio.create_task(asyncio.to_thread(function, *args, **kwargs))
try:
return await asyncio.shield(task)
except asyncio.CancelledError:
await asyncio.gather(task, return_exceptions=True)
raise
return await finish_cleanup(asyncio.to_thread(function, *args, **kwargs))
class HomeEvaluation(BaseComponent):
@ -81,7 +77,10 @@ class HomeEvaluation(BaseComponent):
return await asyncio.shield(task)
except asyncio.CancelledError:
task.cancel()
await asyncio.gather(task, return_exceptions=True)
try:
await finish_cleanup(asyncio.gather(task, return_exceptions=True))
except asyncio.CancelledError:
pass
raise
finally:
self._pending.discard(task)
@ -408,10 +407,16 @@ class HomeEvaluation(BaseComponent):
success=summary["case_status_counts"]["infra_error"] == 0,
finished_at=utc_now().isoformat(),
)
write_json(run / "manifest.json", manifest)
write_json(run / "summary.json", summary, compact_leaf_containers=True)
if not spec.debug and spec.record_history:
await blocking(self._finish_sampling, run)
await blocking(self.history.publish_completed, run, manifest, summary)
try:
await blocking(self._finish_sampling, run)
except Exception:
await blocking(self.history.invalidate)
self.logger.exception("Sampling registration failed; completed evaluation retained")
else:
write_json(run / "manifest.json", manifest)
write_json(run / "summary.json", summary, compact_leaf_containers=True)
return {
"result_dir": str(run),
"commit_sha": commit,
@ -423,11 +428,10 @@ class HomeEvaluation(BaseComponent):
**{key: value for key, value in summary.items() if key != "cases"},
}
except BaseException as error:
# A full summary is the durable completion boundary. Cancellation
# during drained history bookkeeping cancels the caller, not the
# already completed evaluation or its successful history entry.
registration_failed = run is not None and manifest.get("status") == "completed"
if isinstance(error, asyncio.CancelledError) and registration_failed:
# A full summary is the durable completion boundary. Bookkeeping
# errors or cancellation must not demote an already completed run.
completed = run is not None and manifest.get("status") == "completed" and (run / "summary.json").is_file()
if completed:
raise
if run is not None:
manifest.update(
@ -451,11 +455,6 @@ class HomeEvaluation(BaseComponent):
scoring_required=not spec.debug,
)
write_json(run / "partial_summary.json", partial, compact_leaf_containers=True)
if registration_failed and not spec.debug and spec.record_history:
try:
await blocking(self.history.invalidate)
except Exception as cache_error:
error.add_note(f"History cache invalidation failed: {cache_error}")
if isinstance(error, asyncio.CancelledError) or run is None:
raise
raise RuntimeError(f"Evaluation failed; artifacts: {run}: {error}") from error

View file

@ -167,6 +167,9 @@ def update_sampling_state(home: Path, run: Path, *, _state: dict | None = None)
state = read_state(home) if _state is None else _state
key = plan["validation_key"]
if key in state["processed_validations"]:
# A missing cache was reconstructed above, including this completion.
if _state is None and not contained(home, "sampling/state.json").exists():
write_json(contained(home, "sampling/state.json"), state)
return
round_id = state["successful_round"] + 1
cases, queries = state["cases"], state["queries"]
@ -271,11 +274,11 @@ class HomeEvaluationHistory(BaseComponent):
super().__init__(**kwargs)
self.workspace = self.bind("default", HomeWorkspace, optional=False)
def _history_lock(self):
def _history_lock(self, *, wait: bool = False):
if not self.is_started:
raise RuntimeError("Start the Home application before reading evaluation history")
_validate_directory(self.workspace_path.resolve(), self.workspace.state_path)
return _initialization_lock(self.workspace.state_path / "workspace.lock")
return _initialization_lock(self.workspace.state_path / "workspace.lock", wait=wait)
def _run_path(self, run: Path) -> Path:
home = self.workspace.home_path
@ -317,6 +320,18 @@ class HomeEvaluationHistory(BaseComponent):
update_sampling_state(home, run)
return record
def publish_completed(self, run: Path, manifest: dict, summary: dict) -> None:
"""Invalidate observations before publishing the durable completion boundary.
A process exit before registration leaves a missing cache, so the next
reader reconstructs all observations from source artifacts.
"""
with self._history_lock(wait=True):
run = self._run_path(run)
contained(self.workspace.home_path, "sampling/state.json").unlink(missing_ok=True)
write_json(run / "manifest.json", manifest)
write_json(run / "summary.json", summary, compact_leaf_containers=True)
def list_results(self) -> list[EvaluationRecord]:
"""Return successful results in chronological order, including low-scoring candidates."""
with self._history_lock():

View file

@ -54,13 +54,26 @@ def runtime_roots(plugins: list[str]) -> list[Path]:
]
def git_tree(repository: Path, revision: str, roots: list[Path] | None = None) -> list[tuple[Path, str, str]]:
"""Select regular Git files using exact NUL-delimited paths, retaining modes and object IDs."""
def git_tree(
repository: Path,
revision: str,
roots: list[Path] | None = None,
*,
path_filter: Callable[[str], bool] | None = None,
) -> list[tuple[Path, str, str]]:
"""Select regular Git files with exact NUL-delimited names, modes and object IDs.
Browsing may filter raw names before validation; export roots retain strict
path validation across the full tree.
"""
entries = []
raw_tree = git_bytes(repository, "ls-tree", "-r", "-z", "--full-tree", revision)
for entry in filter(None, raw_tree.split(b"\0")):
header, raw_name = entry.split(b"\t", 1)
path = Path(relative_path(raw_name.decode("utf-8")))
name = raw_name.decode("utf-8")
if path_filter is not None and not path_filter(name):
continue
path = Path(relative_path(name))
if roots is not None and not any(path.is_relative_to(root) for root in roots):
continue
mode, kind, oid = header.decode().split()

View file

@ -1,8 +1,10 @@
"""Initialize Home's layout inside an existing ReMe workspace."""
import json
import errno
import os
import tempfile
import time
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
@ -48,8 +50,8 @@ def _validate_directory(root: Path, path: Path) -> None:
@contextmanager
def _initialization_lock(path: Path):
"""Hold a process lock; the persistent lock file is not an ownership marker."""
def _initialization_lock(path: Path, *, wait: bool = False):
"""Hold a process lock, optionally waiting for contention in an owned worker thread."""
if path.is_symlink() or (path.exists() and not path.is_file()):
raise ValueError(f"Invalid Home workspace lock: {path}")
with path.open("a+b") as stream:
@ -77,10 +79,15 @@ def _initialization_lock(path: Path):
def release():
fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
try:
acquire()
except OSError as exc:
raise RuntimeError(f"Home workspace initialization is already in progress: {path}") from exc
while True:
try:
acquire()
break
except OSError as exc:
if wait and exc.errno in {errno.EACCES, errno.EAGAIN, errno.EDEADLK}:
time.sleep(0.05)
continue
raise RuntimeError(f"Home workspace initialization is already in progress: {path}") from exc
try:
yield
finally:

View file

@ -3,7 +3,13 @@
# pylint: disable=missing-function-docstring,redefined-outer-name,unused-argument,protected-access
import json
import asyncio
import os
from pathlib import Path
import subprocess
import sys
from threading import Event
from unittest.mock import AsyncMock
import pytest
from agentscope.message import ToolResultState
@ -39,6 +45,59 @@ def manifest(response):
return json.loads((Path(response.answer["result_dir"]) / "manifest.json").read_text())
@pytest.mark.asyncio
@pytest.mark.parametrize("cancelled", [False, True])
async def test_completion_waits_for_concurrent_code_browsing(app, prepared, monkeypatch, cancelled):
from reme_home.code.component import OPERATIONS
manager = history(app)
publishing, browsing, release = Event(), Event(), Event()
original_publish, original_status = manager.publish_completed, OPERATIONS["status"]
def publish(*args):
publishing.set()
assert browsing.wait(5)
return original_publish(*args)
def status(*args, **kwargs):
browsing.set() # HomeCode owns workspace.lock before dispatching here.
assert release.wait(5)
return original_status(*args, **kwargs)
monkeypatch.setattr(manager, "publish_completed", publish)
monkeypatch.setitem(OPERATIONS, "status", status)
evaluation = asyncio.create_task(app.run_job("home_evaluate"))
code_status = None
try:
assert await asyncio.to_thread(publishing.wait, 5)
code_status = asyncio.create_task(app.run_job("home_code_status"))
assert await asyncio.to_thread(browsing.wait, 5)
await asyncio.sleep(0.1)
assert not evaluation.done()
if cancelled:
evaluation.cancel("original cancellation")
await asyncio.sleep(0)
evaluation.cancel("repeated cancellation")
await asyncio.sleep(0)
assert not evaluation.done()
release.set()
if cancelled:
with pytest.raises(asyncio.CancelledError, match="original cancellation"):
await evaluation
else:
response = await evaluation
assert response.success, response.answer
assert manifest(response)["status"] == "completed"
assert (await code_status).success
assert len(manager.list_results()) == 1
state = read_state(prepared)
manager.rebuild()
assert read_state(prepared) == state and state["successful_round"] == 1
finally:
release.set()
await asyncio.gather(evaluation, *([code_status] if code_status is not None else []), return_exceptions=True)
@pytest.mark.asyncio
async def test_start_cache_failure_keeps_start_evidence_and_exhausts_budget(app, prepared, monkeypatch):
commit_candidate(prepared)
@ -423,9 +482,6 @@ async def test_legacy_cache_rebuild_and_explicit_null_query_budget(app, prepared
@pytest.mark.asyncio
async def test_cancellation_after_durable_completion_keeps_history_consistent(app, prepared, monkeypatch):
import asyncio
from threading import Event
manager = history(app)
original = manager.record
entered, release = Event(), Event()
@ -451,23 +507,183 @@ async def test_cancellation_after_durable_completion_keeps_history_consistent(ap
@pytest.mark.asyncio
async def test_registration_failure_invalidates_only_caches(app, prepared, monkeypatch):
@pytest.mark.parametrize("after_record", [False, True])
async def test_registration_failure_invalidates_only_caches(app, prepared, monkeypatch, after_record):
manager = history(app)
original = manager.record
def fail(*args):
if after_record:
original(*args)
raise OSError("sampling observations could not be saved")
monkeypatch.setattr(manager, "record", fail)
response = await app.run_job("home_evaluate")
assert not response.success
assert response.success, response.answer
runs = list((prepared / "evaluations").glob("*/*/*/manifest.json"))
assert len(runs) == 1
run = runs[0].parent
failed = json.loads(runs[0].read_text())
assert failed["status"] == "failed" and failed["success"] is False
completed = json.loads(runs[0].read_text())
assert completed["status"] == "completed" and completed["success"] is True
assert (run / "summary.json").is_file() and (run / "cases/1/queries/result.json").is_file()
assert not (prepared / "sampling/state.json").exists()
assert read_state(prepared)["round"] == 1 and read_state(prepared)["successful_round"] == 0
assert read_state(prepared)["round"] == 1 and read_state(prepared)["successful_round"] == 1
monkeypatch.setattr(manager, "record", original)
assert manager.list_results() == [] and not manager.sampling_state()["cases"]
assert len(manager.list_results()) == 1
observations = manager.sampling_state()
repeat = await app.run_job("home_evaluate")
assert not repeat.success and repeat.answer["error_code"] == "UNCHANGED_CANDIDATE"
assert manager.rebuild() == manager.list_results()
assert manager.sampling_state() == observations
commit_candidate(prepared)
assert (await app.run_job("home_evaluate")).success
state = read_state(prepared)
manager.rebuild()
assert read_state(prepared) == state
assert state["successful_round"] == 2
@pytest.mark.asyncio
async def test_process_exit_after_completion_recovers_before_next_evaluation(app, prepared):
from reme.application import Application
config = app.config.model_dump(mode="json")
await app.close()
script = """
import asyncio
from importlib.metadata import EntryPoint
import json
import os
import sys
from unittest.mock import AsyncMock
sys.path[:] = json.loads(sys.argv[1])
from reme.application import Application
from reme_home.evaluation import component
from test_evaluation import FakeFactory
import reme.plugin
entries = {name: EntryPoint(name=name, value=value, group="reme.plugins") for name, value in
[("home", "reme_home"), ("beam-judge", "judge_beam")]}
reme.plugin.find_entry_points = lambda group, name: [entries[name]]
component.DockerReMeSandboxFactory = FakeFactory
async def run():
app = Application(**json.loads(sys.argv[2]))
await app.start()
app.context.components["agent_wrapper"]["judge"].reply = AsyncMock(
return_value={"result": '{"score": 0.5}'})
app.context.components["home.evaluation"]["default"]._finish_sampling = lambda run: os._exit(23)
await app.run_job("home_evaluate")
asyncio.run(run())
"""
result = await asyncio.to_thread(
subprocess.run,
[sys.executable, "-c", script, json.dumps(sys.path), json.dumps(config)],
capture_output=True,
text=True,
timeout=30,
env=os.environ.copy(),
)
assert result.returncode == 23, result.stderr
assert not (prepared / "sampling/state.json").exists()
restarted = Application(**config)
await restarted.start()
try:
judge = restarted.context.components["agent_wrapper"]["judge"]
judge.reply = AsyncMock(return_value={"result": '{"score": 0.5}'})
manager = history(restarted)
assert len(manager.list_results()) == 1
assert read_state(prepared)["successful_round"] == 1
commit_candidate(prepared)
following = await restarted.run_job("home_evaluate")
assert following.success, following.answer
assert len(manager.list_results()) == 2
state = read_state(prepared)
assert state["successful_round"] == 2
manager.rebuild()
assert read_state(prepared) == state
finally:
await restarted.close()
@pytest.mark.asyncio
async def test_repeated_cancellation_drains_preparation_before_application_close(app, prepared, monkeypatch):
evaluator = app.context.components["home.evaluation"]["default"]
entered, release, finished = Event(), Event(), Event()
output = prepared / "thread-finished.txt"
def prepare(*args):
entered.set()
assert release.wait(5)
output.write_text("finished")
finished.set()
monkeypatch.setattr(evaluator, "_prepare", prepare)
task = asyncio.create_task(app.run_job("home_evaluate"))
close = None
try:
assert await asyncio.to_thread(entered.wait, 5)
task.cancel("first cancellation")
await asyncio.sleep(0)
task.cancel("second cancellation")
await asyncio.sleep(0)
close = asyncio.create_task(app.close())
await asyncio.sleep(0.05)
assert evaluator._pending and not close.done() and not task.done()
release.set()
with pytest.raises(asyncio.CancelledError, match="first cancellation"):
await task
await close
assert finished.is_set() and output.read_text() == "finished"
assert not evaluator._pending
finally:
release.set()
await asyncio.gather(task, *([close] if close is not None else []), return_exceptions=True)
@pytest.mark.asyncio
@pytest.mark.parametrize("fail", [False, True])
async def test_blocking_preserves_kwargs_and_thread_errors(fail):
from reme_home.evaluation.component import blocking
error = OSError("thread operation failed")
def operation(value, *, suffix):
if fail:
raise error
return value + suffix
if fail:
with pytest.raises(OSError) as raised:
await blocking(operation, "value", suffix="-suffix")
assert raised.value is error
else:
assert await blocking(operation, "value", suffix="-suffix") == "value-suffix"
@pytest.mark.asyncio
async def test_blocking_drains_thread_through_repeated_cancellation(tmp_path):
from reme_home.evaluation.component import blocking
entered, release, finished = Event(), Event(), Event()
output = tmp_path / "thread-finished.txt"
def write(*, content):
entered.set()
assert release.wait(5)
output.write_text(content)
finished.set()
task = asyncio.create_task(blocking(write, content="finished"))
try:
assert await asyncio.to_thread(entered.wait, 5)
task.cancel("first cancellation")
await asyncio.sleep(0)
task.cancel("second cancellation")
await asyncio.sleep(0)
assert not task.done() and not finished.is_set()
release.set()
with pytest.raises(asyncio.CancelledError, match="first cancellation"):
await task
assert finished.is_set() and output.read_text() == "finished"
finally:
release.set()
await asyncio.gather(task, return_exceptions=True)

View file

@ -6,6 +6,7 @@ from unittest.mock import Mock
import pytest
from reme_home.preparation import code
from reme_home.code import repository as candidate_repository, snapshot
def test_batch_export_bypasses_filters_and_checkout(tmp_path, monkeypatch):
@ -42,8 +43,19 @@ def test_batch_export_bypasses_filters_and_checkout(tmp_path, monkeypatch):
target = destination / name
assert target.read_bytes() == data
assert target.stat().st_mode & 0o777 == (0o755 if name == "脚本" else 0o644)
# Browsing and single-file export use the same committed objects while
# retaining their own size limits and streaming behavior.
objects = dict(snapshot.tree(repository, "HEAD"))
assert set(objects) == set(contents)
assert snapshot.blob(repository, objects['quotes"\tand\nnewline']) == contents['quotes"\tand\nnewline']
with pytest.raises(ValueError, match="exceeds 5 MB"):
snapshot.blob(repository, objects["large"])
streamed = tmp_path / "streamed"
snapshot.export_blob(repository, objects["large"], streamed)
assert streamed.read_bytes() == contents["large"]
@pytest.mark.parametrize("reader", ["preparation", "snapshot", "candidate"])
@pytest.mark.parametrize(
"mode,kind,name",
[
@ -54,13 +66,24 @@ def test_batch_export_bypasses_filters_and_checkout(tmp_path, monkeypatch):
("100644", "blob", "/escape"),
("100644", "blob", "~escape"),
("100644", "blob", "folder\\escape"),
("100644", "blob", "other:bad.py"),
],
)
def test_tree_rejects_unsafe_paths_and_nonregular_files(monkeypatch, mode, kind, name):
def test_tree_rejects_unsafe_paths_and_nonregular_files(monkeypatch, mode, kind, name, reader):
"""Reject unsafe object metadata before any file is exported."""
monkeypatch.setattr(code, "git_bytes", Mock(return_value=f"{mode} {kind} {'a' * 40}\t{name}\0".encode()))
monkeypatch.setattr(candidate_repository, "git", Mock(return_value=""))
with pytest.raises(ValueError):
code.git_tree(Path("unused"), "HEAD")
if reader == "preparation":
code.git_tree(Path("unused"), "HEAD")
elif reader == "snapshot":
snapshot.tree(Path("unused"), "HEAD")
else:
candidate_repository.validate_tree(
Path("unused"),
{"code": {"candidate_paths": ["plugin"], "initial_commit": "init"}},
"HEAD",
)
def test_tree_selection_uses_literal_roots(monkeypatch):
@ -72,3 +95,33 @@ def test_tree_selection_uses_literal_roots(monkeypatch):
).encode()
monkeypatch.setattr(code, "git_bytes", Mock(return_value=raw))
assert code.git_tree(Path("unused"), "HEAD", [Path("plugin")]) == [(Path("plugin/file"), "100644", "a" * 40)]
assert snapshot.tree(Path("unused"), "HEAD", "plugin") == [("plugin/file", "a" * 40)]
def test_snapshot_selection_treats_metacharacters_literally(monkeypatch):
"""A browsing prefix selects exact names and descendants, never glob matches."""
names = ['literal[ab]/quotes"\tand\nnewline', "literala/unselected", "literal[ab]-other/unselected"]
raw = "".join(f"100755 blob {'a' * 40}\t{name}\0" for name in names).encode()
monkeypatch.setattr(code, "git_bytes", Mock(return_value=raw))
assert snapshot.tree(Path("unused"), "HEAD", "literal[ab]") == [(names[0], "a" * 40)]
assert snapshot.tree(Path("unused"), "HEAD", names[0]) == [(names[0], "a" * 40)]
def test_snapshot_browsing_ignores_invalid_names_outside_selected_path(tmp_path):
"""Unrelated Git names cannot prevent browsing a valid literal directory."""
repository = tmp_path / "repo"
repository.mkdir()
code.git(repository, "init", "--template=")
(repository / "good").mkdir()
(repository / "good/file.py").write_text("value = 1\n")
(repository / "other:bad.py").write_text("outside browsing prefix\n")
code.git(repository, "add", "--all")
code.git(repository, "-c", "user.name=Test", "-c", "user.email=test@localhost", "commit", "-m", "files")
assert snapshot.list_code_tree(repository, {}, path="good")["paths"] == ["good/file.py"]
assert snapshot.read_code_file(repository, {}, path="good/file.py")["content"] == "value = 1\n"
assert snapshot.search_code(repository, {}, path="good", pattern="value")["occurrence_count"] == 1
with pytest.raises(ValueError, match="Invalid relative path"):
snapshot.tree(repository, "HEAD")
with pytest.raises(ValueError, match="Invalid relative path"):
code.git_tree(repository, "HEAD", [Path("good")])