Track ReMe job invocation counts in sandbox results

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"]

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",
+    }

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"],

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()

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,
         )
 

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(),
         }

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)

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

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}