From 9aeede7d4c8ea17ca8072c9789a799d0b79948d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=85=B1=E7=89=9B=E8=82=89?= Date: Tue, 11 Aug 2026 11:08:39 +0800 Subject: [PATCH] Track ReMe job invocation counts in sandbox results --- meta-reme/as_tools/__init__.py | 5 + meta-reme/as_tools/validation_tool.py | 156 ++++++++++++++++++ meta-reme/run.py | 2 + meta-reme/runtime.py | 32 ++++ sandbox/models.py | 5 +- sandbox/worker.py | 14 +- tests/unit_sandbox/test_meta_reme_run.py | 2 + .../test_meta_reme_validation_tool.py | 104 ++++++++++++ tests/unit_sandbox/test_sandbox.py | 20 ++- 9 files changed, 335 insertions(+), 5 deletions(-) create mode 100644 meta-reme/as_tools/__init__.py create mode 100644 meta-reme/as_tools/validation_tool.py create mode 100644 meta-reme/runtime.py create mode 100644 tests/unit_sandbox/test_meta_reme_validation_tool.py diff --git a/meta-reme/as_tools/__init__.py b/meta-reme/as_tools/__init__.py new file mode 100644 index 00000000..778f7aff --- /dev/null +++ b/meta-reme/as_tools/__init__.py @@ -0,0 +1,5 @@ +"""AgentScope tools exposed by the Meta-ReMe orchestration layer.""" + +from .validation_tool import create_validation_tool, validation_tool + +__all__ = ["create_validation_tool", "validation_tool"] diff --git a/meta-reme/as_tools/validation_tool.py b/meta-reme/as_tools/validation_tool.py new file mode 100644 index 00000000..a60687a4 --- /dev/null +++ b/meta-reme/as_tools/validation_tool.py @@ -0,0 +1,156 @@ +"""AgentScope tool for synchronously validating the current Meta-ReMe code.""" + +# Import roots must be installed before importing the hyphenated meta-reme tree. +# pylint: disable=wrong-import-position + +from __future__ import annotations + +import json +from pathlib import Path +import subprocess +import sys +from typing import Any +from uuid import uuid4 + +from agentscope.tool import FunctionTool + +from runtime import TOOL_RUNTIME + +# The meta-reme tree is intentionally not a regular Python package. +META_REME_ROOT = Path(__file__).resolve().parent.parent +PROJECT_ROOT = META_REME_ROOT.parent +for import_root in (META_REME_ROOT, PROJECT_ROOT): + if str(import_root) not in sys.path: + sys.path.insert(0, str(import_root)) + +from validation import run_validation # noqa: E402 + + +def validation_tool(case_ids: list[str] | None = None, fail_fast: bool = False) -> dict[str, Any]: + """Validate selected workspace cases and return only a run-level summary. + + Args: + case_ids: Case IDs installed in the active workspace. Omit or pass an + empty list to validate every installed case. + fail_fast: Stop the validation at the first error when true. + + Returns: + Counts of cases that ran without errors or failed, the mean query + score, and the result directory containing full artifacts. + """ + + if not isinstance(fail_fast, bool): + raise ValueError("fail_fast must be a boolean") + workspace, concurrency = TOOL_RUNTIME.require_configured() + selected_case_ids = _case_ids(workspace, case_ids) + code_id = _current_code_id(workspace) + validation_id = uuid4().hex + result_dir = workspace / "evaluations" / code_id / validation_id + + error: str | None = None + try: + run_validation( + workspace, + selected_case_ids, + code_id, + concurrency, + validation_id=validation_id, + fail_fast=fail_fast, + ) + except Exception as exc: # Validation persists diagnostics before raising. + error = f"{type(exc).__name__}: {exc}" + + return _read_summary(result_dir, len(selected_case_ids), error) + + +def create_validation_tool() -> FunctionTool: + """Create the AgentScope wrapper for the synchronous validation function.""" + + return FunctionTool( + validation_tool, + name="validation_tool", + description=( + "Validate selected cases against the current Meta-ReMe code branch. " + "Omit case_ids to run all workspace cases; fail_fast stops at the first error. " + "The result includes details_path, which points to the complete validation artifacts." + ), + is_concurrency_safe=False, + ) + + +def _case_ids(workspace: Path, requested: list[str] | None) -> list[str]: + if requested is not None and not isinstance(requested, list): + raise ValueError("case_ids must be a list of strings or null") + if requested: + if any(not isinstance(case_id, str) or not case_id for case_id in requested): + raise ValueError("case_ids must contain non-empty strings") + if len(requested) != len(set(requested)): + raise ValueError("case_ids must be unique") + return requested + + installed: list[str] = [] + for path in sorted((workspace / "dataset/cases").glob("*.json")): + payload = json.loads(path.read_text(encoding="utf-8")) + case_id = payload.get("case_id") + if not isinstance(case_id, str) or not case_id: + raise ValueError(f"prepared case has no valid case_id: {path}") + installed.append(case_id) + if not installed: + raise ValueError("active workspace has no prepared cases") + if len(installed) != len(set(installed)): + raise ValueError("active workspace contains duplicate case IDs") + return installed + + +def _current_code_id(workspace: Path) -> str: + repository = workspace / "code/repo/reme" + result = subprocess.run( + ["git", "branch", "--show-current"], + cwd=repository, + capture_output=True, + text=True, + check=False, + ) + code_id = result.stdout.strip() + if result.returncode or not code_id: + detail = (result.stderr or result.stdout).strip() + suffix = f": {detail}" if detail else " (detached HEAD is not supported)" + raise RuntimeError(f"cannot determine the current Meta-ReMe code branch{suffix}") + return code_id + + +def _read_summary(result_dir: Path, requested_count: int, error: str | None) -> dict[str, Any]: + summary_path = result_dir / "summary.json" + if summary_path.is_file(): + summary = json.loads(summary_path.read_text(encoding="utf-8")) + successful = sum(case.get("status") == "completed" for case in summary.get("cases", [])) + failed = int(summary.get("case_count", 0)) - successful + return { + "status": summary.get("status", "completed"), + "requested_cases": requested_count, + "run_cases": int(summary.get("case_count", 0)), + "successful_cases": successful, + "failed_cases": failed, + "mean_query_score": summary.get("mean_query_score"), + "result_dir": str(result_dir), + "details_path": str(result_dir), + "error": error, + } + + case_results = [] + for path in sorted((result_dir / "cases").glob("*/case_result.json")): + case_results.append(json.loads(path.read_text(encoding="utf-8"))) + query_results = [query for case in case_results for query in case.get("queries", [])] + scores = [float(query["score"]) if query.get("score") is not None else 0.0 for query in query_results] + successful = sum(case.get("status") == "completed" for case in case_results) + return { + "status": "failed" if error else "partial", + "requested_cases": requested_count, + "run_cases": len(case_results), + "successful_cases": successful, + "failed_cases": len(case_results) - successful, + "mean_query_score": sum(scores) / len(scores) if scores else None, + "result_dir": str(result_dir), + "details_path": str(result_dir), + "error": error or "validation did not publish summary.json", + } diff --git a/meta-reme/run.py b/meta-reme/run.py index 2984dcab..40ca7897 100644 --- a/meta-reme/run.py +++ b/meta-reme/run.py @@ -31,6 +31,7 @@ from models import ( SandboxSpec, ScopeSpec, ) +from runtime import TOOL_RUNTIME from validation import run_validation from workspace import Workspace @@ -155,6 +156,7 @@ def prepare_and_validate_workspace( dataset_variant=config["dataset_variant"], dataset_source=config["dataset_source"], ) + TOOL_RUNTIME.configure(workspace.root, config["validation_concurrency"]) validation = run_initial_validation( workspace, config["validation_concurrency"], diff --git a/meta-reme/runtime.py b/meta-reme/runtime.py new file mode 100644 index 00000000..8dddea3a --- /dev/null +++ b/meta-reme/runtime.py @@ -0,0 +1,32 @@ +"""Process-local runtime configuration shared by Meta-ReMe tools.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class MetaReMeToolRuntime: + """Runtime values established by ``meta-reme/run.py`` for agent tools.""" + + workspace: Path | None = None + validation_concurrency: int | None = None + + def configure(self, workspace: Path, validation_concurrency: int) -> None: + """Replace the active run values while preserving singleton identity.""" + + if validation_concurrency < 1: + raise ValueError("validation_concurrency must be at least 1") + self.workspace = Path(workspace).resolve() + self.validation_concurrency = validation_concurrency + + def require_configured(self) -> tuple[Path, int]: + """Return configured values or fail clearly outside the main workflow.""" + + if self.workspace is None or self.validation_concurrency is None: + raise RuntimeError("Meta-ReMe tool runtime is not configured; run meta-reme/run.py first") + return self.workspace, self.validation_concurrency + + +TOOL_RUNTIME = MetaReMeToolRuntime() diff --git a/sandbox/models.py b/sandbox/models.py index f2facd9e..f3c73251 100644 --- a/sandbox/models.py +++ b/sandbox/models.py @@ -12,7 +12,8 @@ class JobResult: ``token_usage`` contains the input/output/total token delta for each configured agent wrapper during this invocation. Missing provider usage is - represented by ``None`` at the metric level. + represented by ``None`` at the metric level. ``job_call_counts`` contains + non-zero ReMe job invocation deltas, including nested job tools. """ success: bool @@ -20,6 +21,7 @@ class JobResult: metadata: dict[str, Any] = field(default_factory=dict) error: str | None = None token_usage: dict[str, dict[str, int | None]] = field(default_factory=dict) + job_call_counts: dict[str, int] = field(default_factory=dict) @classmethod def from_dict(cls, value: dict[str, Any]) -> "JobResult": @@ -29,6 +31,7 @@ class JobResult: answer=value.get("answer", ""), metadata=value.get("metadata") if isinstance(value.get("metadata"), dict) else {}, token_usage=value.get("token_usage") if isinstance(value.get("token_usage"), dict) else {}, + job_call_counts=value.get("job_call_counts") if isinstance(value.get("job_call_counts"), dict) else {}, error=value.get("error") if isinstance(value.get("error"), str) else None, ) diff --git a/sandbox/worker.py b/sandbox/worker.py index 3b59f502..ae40aac1 100644 --- a/sandbox/worker.py +++ b/sandbox/worker.py @@ -95,14 +95,19 @@ def _prepare_runtime(request: dict[str, Any]) -> tuple[Path, dict[str, Any]]: async def _run_job_on_app(app: Any, job: str, arguments: dict[str, Any]) -> dict[str, Any]: - """Run one job and return its serializable result with invocation token deltas.""" + """Run one job and return token and ReMe job invocation deltas.""" from reme.enumeration import ComponentEnum - from reme.utils.evaluation_interface import track_agent_token_usage + from reme.utils.evaluation_interface import track_agent_token_usage, track_job_counts agent_names = list(app.context.components.get(ComponentEnum.AGENT_WRAPPER, {})) + job_names = list(app.context.jobs) token_usage: dict[str, dict[str, int | None]] = {} + job_call_counts: dict[str, int] = {} try: - with track_agent_token_usage(agent_names, app.context) as token_usage: + with ( + track_agent_token_usage(agent_names, app.context) as token_usage, + track_job_counts(job_names, app.context) as job_call_counts, + ): response = await app.run_job(job, **arguments) except Exception as exc: # Preserve usage accumulated before a job failure. return { @@ -110,6 +115,7 @@ async def _run_job_on_app(app: Any, job: str, arguments: dict[str, Any]) -> dict "answer": "", "metadata": {}, "token_usage": token_usage, + "job_call_counts": {name: count for name, count in job_call_counts.items() if count}, "error": f"{type(exc).__name__}: {exc}", "traceback": traceback.format_exc(), } @@ -118,6 +124,7 @@ async def _run_job_on_app(app: Any, job: str, arguments: dict[str, Any]) -> dict "answer": response.answer, "metadata": response.metadata, "token_usage": token_usage, + "job_call_counts": {name: count for name, count in job_call_counts.items() if count}, "error": None if response.success else str(response.answer), } @@ -372,6 +379,7 @@ def main() -> int: "answer": "", "metadata": {}, "token_usage": {}, + "job_call_counts": {}, "error": f"{type(exc).__name__}: {exc}", "traceback": traceback.format_exc(), } diff --git a/tests/unit_sandbox/test_meta_reme_run.py b/tests/unit_sandbox/test_meta_reme_run.py index 6e4bd44d..b0058af5 100644 --- a/tests/unit_sandbox/test_meta_reme_run.py +++ b/tests/unit_sandbox/test_meta_reme_run.py @@ -206,6 +206,8 @@ def test_prepare_and_validate_workspace_uses_all_installed_cases(tmp_path: Path) workspace, validation = run.prepare_and_validate_workspace(config_path, validation_runner=fake_validation) assert validation == workspace.path("evaluations/init/initial") + assert run.TOOL_RUNTIME.workspace == workspace.root + assert run.TOOL_RUNTIME.validation_concurrency == 5 assert calls == [(workspace.root, ["case-2", "case-1"], "init", 5, "initial", True)] _, repeated_validation = run.prepare_and_validate_workspace(config_path, validation_runner=fake_validation) diff --git a/tests/unit_sandbox/test_meta_reme_validation_tool.py b/tests/unit_sandbox/test_meta_reme_validation_tool.py new file mode 100644 index 00000000..8b3b4ed4 --- /dev/null +++ b/tests/unit_sandbox/test_meta_reme_validation_tool.py @@ -0,0 +1,104 @@ +"""Focused tests for the Meta-ReMe AgentScope validation tool.""" + +# pylint: disable=missing-function-docstring + +from __future__ import annotations + +import importlib +import json +from pathlib import Path +import subprocess +import sys + +META_REME = Path(__file__).resolve().parents[2] / "meta-reme" +if str(META_REME) not in sys.path: + sys.path.insert(0, str(META_REME)) + +runtime = importlib.import_module("runtime") +tool_module = importlib.import_module("as_tools.validation_tool") + + +def _workspace(tmp_path: Path) -> Path: + workspace = tmp_path / "workspace" + cases = workspace / "dataset/cases" + cases.mkdir(parents=True) + for index, case_id in enumerate(("case-b", "case-a")): + (cases / f"{index}.json").write_text(json.dumps({"case_id": case_id}), encoding="utf-8") + repository = workspace / "code/repo/reme" + repository.mkdir(parents=True) + subprocess.run(["git", "init", "-b", "candidate"], cwd=repository, check=True, capture_output=True) + return workspace + + +def test_validation_tool_uses_runtime_defaults_and_returns_summary(tmp_path: Path, monkeypatch) -> None: + workspace = _workspace(tmp_path) + runtime.TOOL_RUNTIME.configure(workspace, 3) + calls = [] + + def fake_validation(root, case_ids, code_id, concurrency, *, validation_id, fail_fast): + calls.append((root, case_ids, code_id, concurrency, validation_id, fail_fast)) + result_dir = Path(root) / "evaluations" / code_id / validation_id + result_dir.mkdir(parents=True) + (result_dir / "summary.json").write_text( + json.dumps( + { + "status": "completed", + "case_count": 2, + "mean_query_score": 0.75, + "cases": [{"status": "completed"}, {"status": "candidate_failure"}], + }, + ), + encoding="utf-8", + ) + return result_dir + + monkeypatch.setattr(tool_module, "run_validation", fake_validation) + + result = tool_module.validation_tool(fail_fast=True) + + assert calls[0][0:4] == (workspace.resolve(), ["case-b", "case-a"], "candidate", 3) + assert calls[0][-1] is True + assert result == { + "status": "completed", + "requested_cases": 2, + "run_cases": 2, + "successful_cases": 1, + "failed_cases": 1, + "mean_query_score": 0.75, + "result_dir": str(workspace / "evaluations/candidate" / calls[0][4]), + "details_path": str(workspace / "evaluations/candidate" / calls[0][4]), + "error": None, + } + + +def test_validation_tool_summarizes_persisted_results_after_fail_fast(tmp_path: Path, monkeypatch) -> None: + workspace = _workspace(tmp_path) + runtime.TOOL_RUNTIME.configure(workspace, 1) + + def fake_validation(root, case_ids, code_id, concurrency, *, validation_id, fail_fast): + del case_ids, concurrency, fail_fast + case_root = Path(root) / "evaluations" / code_id / validation_id / "cases/case-a" + case_root.mkdir(parents=True) + (case_root / "case_result.json").write_text( + json.dumps({"status": "candidate_failure", "queries": []}), + encoding="utf-8", + ) + raise RuntimeError("construction failed") + + monkeypatch.setattr(tool_module, "run_validation", fake_validation) + + result = tool_module.validation_tool(["case-a"], fail_fast=True) + + assert result["status"] == "failed" + assert result["run_cases"] == 1 + assert result["successful_cases"] == 0 + assert result["failed_cases"] == 1 + assert result["mean_query_score"] is None + assert result["error"] == "RuntimeError: construction failed" + + +def test_create_validation_tool_is_not_concurrency_safe() -> None: + tool = tool_module.create_validation_tool() + + assert tool.name == "validation_tool" + assert tool.is_concurrency_safe is False diff --git a/tests/unit_sandbox/test_sandbox.py b/tests/unit_sandbox/test_sandbox.py index 541c3640..63cd0a92 100644 --- a/tests/unit_sandbox/test_sandbox.py +++ b/tests/unit_sandbox/test_sandbox.py @@ -129,6 +129,7 @@ class FakeBackend: "token_usage": { "bench": {"input_tokens": 12, "output_tokens": 3, "total_tokens": 15}, }, + "job_call_counts": {"answer_judge": 1}, "error": None, } self.files[response_path] = json.dumps(response).encode() @@ -265,6 +266,7 @@ async def test_image_candidate_skips_source_upload_and_runs_jobs(tmp_path): assert result.token_usage == { "bench": {"input_tokens": 12, "output_tokens": 3, "total_tokens": 15}, } + assert result.job_call_counts == {"answer_judge": 1} assert "/workspace/candidate.tar.gz" not in workspaces[0].backend.files assert exported.read_bytes() == b"case-archive" manifest = json.loads(workspaces[0].backend.files["/workspace/case/manifest.json"]) @@ -771,6 +773,7 @@ async def test_worker_calls_application_directly_and_always_closes(monkeypatch, self.context = SimpleNamespace( metadata={}, components={ComponentEnum.AGENT_WRAPPER: {"bench": SimpleNamespace()}}, + jobs={"agentic_answer": SimpleNamespace(), "search": SimpleNamespace()}, ) async def start(self): @@ -783,6 +786,8 @@ async def test_worker_calls_application_directly_and_always_closes(monkeypatch, global_counter_add(self.context.metadata, ["__token_counter", "bench", "input_tokens"], 12) global_counter_add(self.context.metadata, ["__token_counter", "bench", "output_tokens"], 3) global_counter_add(self.context.metadata, ["__token_counter", "bench", "total_tokens"], 15) + global_counter_add(self.context.metadata, ["__job_counter", "agentic_answer"], 1) + global_counter_add(self.context.metadata, ["__job_counter", "search"], 3) return SimpleNamespace(success=True, answer="answer", metadata={"direct": True}) async def close(self): @@ -808,6 +813,7 @@ async def test_worker_calls_application_directly_and_always_closes(monkeypatch, "token_usage": { "bench": {"input_tokens": 12, "output_tokens": 3, "total_tokens": 15}, }, + "job_call_counts": {"agentic_answer": 1, "search": 3}, "error": None, } assert worker.os.environ["TMPDIR"] == str(tmp_path / "tmp") @@ -836,7 +842,16 @@ async def test_worker_reuses_one_application_per_phase_and_isolates_query_logs(m def __init__(self, **_config): self.application_id = len(application_ids) + 1 application_ids.append(self.application_id) - self.context = SimpleNamespace(metadata={}, components={}) + self.context = SimpleNamespace( + metadata={}, + components={}, + jobs={ + "auto_memory": SimpleNamespace(), + "index_update": SimpleNamespace(), + "agentic_answer": SimpleNamespace(), + "answer_judge": SimpleNamespace(), + }, + ) async def start(self): """Start without external resources.""" @@ -926,6 +941,7 @@ async def test_worker_returns_usage_accumulated_before_job_failure(monkeypatch, self.context = SimpleNamespace( metadata={}, components={ComponentEnum.AGENT_WRAPPER: {"bench": SimpleNamespace()}}, + jobs={"agentic_answer": SimpleNamespace()}, ) async def start(self): @@ -936,6 +952,7 @@ async def test_worker_returns_usage_accumulated_before_job_failure(monkeypatch, global_counter_add(self.context.metadata, ["__token_counter", "bench", "input_tokens"], 8) global_counter_add(self.context.metadata, ["__token_counter", "bench", "output_tokens"], 2) global_counter_add(self.context.metadata, ["__token_counter", "bench", "total_tokens"], 10) + global_counter_add(self.context.metadata, ["__job_counter", "agentic_answer"], 1) raise RuntimeError("step failed") async def close(self): @@ -958,3 +975,4 @@ async def test_worker_returns_usage_accumulated_before_job_failure(monkeypatch, assert result["token_usage"] == { "bench": {"input_tokens": 8, "output_tokens": 2, "total_tokens": 10}, } + assert result["job_call_counts"] == {"agentic_answer": 1}