mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-10-13 04:01:35 +00:00
e4c381c7b1
benchmark.yaml's python_execute job advertised a `timeout` parameter, but `timeout` is one of the CLI's client-selection kwargs (reme/reme.py pops it before the payload is sent), so `reme python_execute code=... timeout=300` never delivered the timeout to the job. The value silently became the HTTP client's request timeout while the subprocess kept the 60s default, so benchmark code that needs longer was killed at 60s with no hint why. HTTP and MCP callers were unaffected. Rename the parameter to `python_timeout`, mirroring how the shell job already avoids the same collision with `shell_timeout`. The step prefers the new name and still accepts the legacy `timeout` context key so existing benchmark configs keep working. A new guard test fails if any job in reme/config/*.yaml or a plugin plugin.yaml declares a client-reserved parameter name again.
102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
"""Execute Python code and return printed stdout."""
|
|
|
|
import asyncio
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from ..base_step import BaseStep
|
|
from ...components import R
|
|
|
|
DEFAULT_TIMEOUT = 60.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _PythonResult:
|
|
stdout: str
|
|
stderr: str
|
|
returncode: int | None
|
|
timed_out: bool = False
|
|
|
|
|
|
@R.register("python_execute_step")
|
|
class PythonExecuteStep(BaseStep):
|
|
"""Run Python code in a subprocess and return stdout as the response answer."""
|
|
|
|
async def execute(self):
|
|
assert self.context is not None
|
|
|
|
code = self.context.get("code", "")
|
|
timeout, timeout_error = self._parse_timeout(
|
|
self.context.get("python_timeout", self.context.get("timeout", DEFAULT_TIMEOUT)),
|
|
)
|
|
if not isinstance(code, str) or not code.strip():
|
|
self.context.response.success = False
|
|
self.context.response.answer = "code is required"
|
|
return self.context.response
|
|
if timeout_error:
|
|
self.context.response.success = False
|
|
self.context.response.answer = timeout_error
|
|
return self.context.response
|
|
|
|
result = await self._run_python(code, timeout)
|
|
if result.timed_out:
|
|
self.context.response.success = False
|
|
self.context.response.answer = f"Python execution timed out after {timeout:g}s"
|
|
self.context.response.metadata.update(
|
|
{
|
|
"returncode": result.returncode,
|
|
"stderr": result.stderr,
|
|
"python_timeout": timeout,
|
|
},
|
|
)
|
|
return self.context.response
|
|
|
|
stdout = result.stdout or ""
|
|
stderr = result.stderr or ""
|
|
self.context.response.success = result.returncode == 0
|
|
self.context.response.answer = stdout if stdout or result.returncode == 0 else stderr
|
|
self.context.response.metadata.update(
|
|
{
|
|
"returncode": result.returncode,
|
|
"stderr": stderr,
|
|
"python_timeout": timeout,
|
|
},
|
|
)
|
|
return self.context.response
|
|
|
|
async def _run_python(self, code: str, timeout: float) -> _PythonResult:
|
|
process = await asyncio.create_subprocess_exec(
|
|
sys.executable,
|
|
"-c",
|
|
code,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
cwd=self.workspace_path,
|
|
)
|
|
try:
|
|
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
|
|
return _PythonResult(
|
|
stdout=stdout.decode(),
|
|
stderr=stderr.decode(),
|
|
returncode=process.returncode,
|
|
)
|
|
except TimeoutError:
|
|
process.kill()
|
|
stdout, stderr = await process.communicate()
|
|
return _PythonResult(
|
|
stdout=stdout.decode(),
|
|
stderr=stderr.decode(),
|
|
returncode=process.returncode,
|
|
timed_out=True,
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_timeout(raw: Any) -> tuple[float, str]:
|
|
try:
|
|
timeout = float(raw)
|
|
except (TypeError, ValueError):
|
|
return DEFAULT_TIMEOUT, "python_timeout must be a positive number"
|
|
if timeout <= 0:
|
|
return DEFAULT_TIMEOUT, "python_timeout must be a positive number"
|
|
return timeout, ""
|