Skip to content

fix: run local python in session workspace #8792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Soulter merged 1 commit into from
Jun 15, 2026
+60 −1
Merged

fix: run local python in session workspace #8792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions astrbot/core/computer/booters/local.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,13 +143,16 @@ async def exec(
kernel_id: str | None = None,
timeout: int = 30,
silent: bool = False,
cwd: str | None = None,
) -> dict[str, Any]:
def _run() -> dict[str, Any]:
try:
working_dir = os.path.abspath(cwd) if cwd else get_astrbot_root()
result = subprocess.run(
[os.environ.get("PYTHON", sys.executable), "-c", code],
timeout=timeout,
capture_output=True,
cwd=working_dir,
)
stdout = "" if silent else _decode_shell_output(result.stdout)
stderr = (
Expand Down
1 change: 1 addition & 0 deletions astrbot/core/computer/olayer/python.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ async def exec(
kernel_id: str | None = None,
timeout: int = 30,
silent: bool = False,
cwd: str | None = None,
) -> dict[str, Any]:
"""Execute Python code"""
...
7 changes: 6 additions & 1 deletion astrbot/core/tools/computer_tools/python.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
from astrbot.core.message.message_event_result import MessageChain

from ..registry import builtin_tool
from .util import check_admin_permission
from .util import check_admin_permission, workspace_root

_OS_NAME = platform.system()
_SANDBOX_PYTHON_TOOL_CONFIG = {
Expand Down Expand Up @@ -137,10 +137,15 @@ async def call(
else context.tool_call_timeout
)
try:
current_workspace_root = workspace_root(
context.context.event.unified_msg_origin
)
current_workspace_root.mkdir(parents=True, exist_ok=True)
result = await sb.python.exec(
code,
timeout=effective_timeout,
silent=silent,
cwd=str(current_workspace_root),
)
return await handle_result(result, context.context.event)
except Exception as e:
Expand Down
50 changes: 50 additions & 0 deletions tests/unit/test_python_tools.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
import platform
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest

from astrbot.core.agent.run_context import ContextWrapper
from astrbot.core.tools.computer_tools.python import LocalPythonTool, PythonTool


Expand All @@ -18,3 +23,48 @@ def test_local_python_tool_description_contains_os():
assert current_os in tool.description
assert "Python environment" in tool.description
assert "system-compatible" in tool.description


@pytest.mark.asyncio
async def test_local_python_tool_uses_session_workspace(tmp_path, monkeypatch):
"""Local Python execution should use the same workspace as local shell."""
tool = LocalPythonTool()
python_exec = AsyncMock(
return_value={"data": {"output": {"text": "ok", "images": []}, "error": ""}}
)
monkeypatch.setattr(
"astrbot.core.tools.computer_tools.python.get_local_booter",
lambda: SimpleNamespace(python=SimpleNamespace(exec=python_exec)),
)
monkeypatch.setattr(
"astrbot.core.tools.computer_tools.python.workspace_root",
lambda umo: tmp_path / umo.replace(":", "_"),
)

event = SimpleNamespace(
unified_msg_origin="onebot:GroupMessage:12345",
role="admin",
get_platform_name=lambda: "onebot",
)
context = ContextWrapper(
context=SimpleNamespace(
event=event,
context=SimpleNamespace(
get_config=lambda **_kwargs: {
"provider_settings": {"computer_use_require_admin": True}
}
),
),
tool_call_timeout=60,
)

await tool.call(context, code="print('ok')", timeout=30)

workspace = tmp_path / "onebot_GroupMessage_12345"
assert workspace.is_dir()
python_exec.assert_awaited_once_with(
"print('ok')",
timeout=30,
silent=False,
cwd=str(workspace.resolve(strict=False)),
)
Comment on lines +35 to +70

@gemini-code-assist gemini-code-assist Bot Jun 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

On environments where the temporary directory path contains symlinks (such as macOS, where /var is a symlink to /private/var), the unresolved path returned by the mocked workspace_root will not match the resolved path used in the assertion (workspace.resolve(strict=False)). This can cause flaky test failures.

To make the test robust and platform-independent, we can resolve the workspace path once and use it consistently for both the mock and the assertion.

    workspace = (tmp_path / "onebot_GroupMessage_12345").resolve(strict=False)
    monkeypatch.setattr(
        "astrbot.core.tools.computer_tools.python.get_local_booter",
        lambda: SimpleNamespace(python=SimpleNamespace(exec=python_exec)),
    )
    monkeypatch.setattr(
        "astrbot.core.tools.computer_tools.python.workspace_root",
        lambda umo: workspace,
    )

    event = SimpleNamespace(
        unified_msg_origin="onebot:GroupMessage:12345",
        role="admin",
        get_platform_name=lambda: "onebot",
    )
    context = ContextWrapper(
        context=SimpleNamespace(
            event=event,
            context=SimpleNamespace(
                get_config=lambda **_kwargs: {
                    "provider_settings": {"computer_use_require_admin": True}
                }
            ),
        ),
        tool_call_timeout=60,
    )

    await tool.call(context, code="print('ok')", timeout=30)

    assert workspace.is_dir()
    python_exec.assert_awaited_once_with(
        "print('ok')",
        timeout=30,
        silent=False,
        cwd=str(workspace),
    )

Loading