feat(sandbox): add Windows filesystem support

docs/i18n/zh-Hans/docusaurus-plugin-content-docs/version-1.11.x/References/Python SDK References/file_system.md

@@ -71,6 +71,10 @@ await sandbox.fs.chmod(
 
 ## upload_dir - 上传目录
 
+`upload_dir` 使用的是**替换语义（replace）**，而不是合并语义。如果
+`target_dir` 已存在，上传内容解压前会删除整个目标目录及其中的所有内容。
+`source_dir` 中不存在的原有文件不会被保留。
+
 ```python
 import os
 from pathlib import Path

docs/versioned_docs/version-1.11.x/References/Python SDK References/file_system.md

@@ -71,6 +71,11 @@ await sandbox.fs.chmod(
 
 ## upload_dir - Upload Directory
 
+`upload_dir` uses **replace semantics**, not merge semantics. If `target_dir`
+already exists, the entire target directory and all of its contents are removed
+before the uploaded directory is extracted. Existing files that are not present
+in `source_dir` are not preserved.
+
 ```python
 import os
 from pathlib import Path

docs/versions.json

@@ -1,4 +1,5 @@
 [
+  "1.11.x",
   "1.10.x",
   "1.9.x",
   "1.8.x",

pyproject.toml

@@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
 authors = [{ name = "chatos@alibaba" }]
 requires-python = "<4.0,>=3.10"
 name = "rl-rock"
-version = "1.10.2"
+version = "1.11.0"
 description = "ROCK-Reinforcement Open Construction Kit"
 readme = "README.md"
 dependencies = [

rock/rocklet/windows.py

@@ -55,6 +55,7 @@ class PowerShellSession(Session):
     _BEGIN_MARKER = "ROCKLET_PS_BEGIN_29234"
     _END_MARKER = "ROCKLET_PS_END_29234"
     _EXIT_MARKER = "ROCKLET_PS_EXIT_29234:"
+    _PROMPT_MARKER = "ROCKLET_PS_PROMPT_29234"
 
     def __init__(self, request: CreateBashSessionRequest):
         self.request = request
@@ -127,10 +128,16 @@ async def start(self) -> CreateBashSessionResponse:
         # PowerShell caches its [Console].Out TextWriter at process init — late
         # OutputEncoding mutations don't affect already-cached writers used by
         # built-in formatters (Get-ChildItem, Format-Table, etc.).
+        #
+        # ConsoleHost normally writes both its prompt and every stdin command to
+        # stdout. Read redirected stdin ourselves to suppress the command echo,
+        # and use a private prompt marker that _run_command removes from output.
         ps_setup = (
             "chcp 65001 > $null; "
             "[Console]::OutputEncoding = [System.Text.Encoding]::UTF8; "
-            "$OutputEncoding = [System.Text.Encoding]::UTF8"
+            "$OutputEncoding = [System.Text.Encoding]::UTF8; "
+            f"function global:prompt {{ '{self._PROMPT_MARKER}' }}; "
+            "function global:PSConsoleHostReadLine { [Console]::In.ReadLine() }"
         )
         self._process = subprocess.Popen(
             [ps_cmd, "-NoLogo", "-NoProfile", "-NoExit", "-Command", ps_setup],
@@ -227,6 +234,7 @@ def _run_command(self, action: BashAction) -> BashObservation:
                         raise SessionNotInitializedError("PowerShell process terminated unexpectedly")
                     continue
 
+                line = line.replace(self._PROMPT_MARKER, "")
                 if self._BEGIN_MARKER in line:
                     started = True
                     continue

rock/sdk/sandbox/client.py

@@ -49,7 +49,7 @@
 from rock.sdk.sandbox.agent.rock_agent import RockAgent
 from rock.sdk.sandbox.config import SandboxConfig, SandboxGroupConfig
 from rock.sdk.sandbox.deploy import Deploy
-from rock.sdk.sandbox.file_system import FileSystem, LinuxFileSystem
+from rock.sdk.sandbox.file_system import FileSystem, LinuxFileSystem, WindowsFileSystem
 from rock.sdk.sandbox.model_service.base import ModelService
 from rock.sdk.sandbox.network import Network
 from rock.sdk.sandbox.oss_client import OssClient
@@ -95,7 +95,7 @@ def __init__(self, config: SandboxConfig):
         self.remote_user = LinuxRemoteUser(self)
         self.process = Process(self)
         self.network = Network(self)
-        self.fs = LinuxFileSystem(self)
+        self.fs = WindowsFileSystem(self) if self.config.image_os.casefold() == "windows" else LinuxFileSystem(self)
         self.runtime_envs = {}
         self.deploy = Deploy(self)
         self.agent = RockAgent(self)

rock/sdk/sandbox/file_system.py

@@ -1,20 +1,28 @@
+import base64
 import shlex
+import shutil
 import tarfile
 import tempfile
 import time
 from abc import ABC, abstractmethod
-from pathlib import Path
+from pathlib import Path, PureWindowsPath
 
-from rock.actions import CreateBashSessionRequest, Observation
+from rock.actions import BashAction, CreateBashSessionRequest, Observation
 from rock.actions.sandbox.base import AbstractSandbox
-from rock.actions.sandbox.request import ChmodRequest, ChownRequest, Command
+from rock.actions.sandbox.request import ChmodRequest, ChownRequest, Command, UploadMode
 from rock.actions.sandbox.response import ChmodResponse, ChownResponse, CommandResponse, DownloadFileResponse
 from rock.logger import init_logger
+from rock.sdk.common.exceptions import BadRequestRockError
 from rock.sdk.sandbox.constants import ENSURE_OSSUTIL_SCRIPT
 
 logger = init_logger(__name__)
 
 
+def _powershell_literal(value: str) -> str:
+    """Quote a value for use as a literal in a PowerShell command."""
+    return "'" + value.replace("'", "''") + "'"
+
+
 class FileSystem(ABC):
     sandbox: AbstractSandbox = None
 
@@ -36,6 +44,12 @@ async def upload_dir(
         target_dir: str,
         extract_timeout: int = 600,
     ) -> Observation:
+        """Upload a directory using replace semantics.
+
+        If ``target_dir`` already exists, it and all of its contents are
+        removed before ``source_dir`` is extracted. Existing contents are not
+        merged with the uploaded directory.
+        """
         pass
 
     @abstractmethod
@@ -95,7 +109,7 @@ async def upload_dir(
         - Check 'tar' exists; if not, return Observation with exit_code != 0
         - Pack source_dir fully into a tar.gz locally
         - Upload to sandbox /tmp
-        - Extract into target_dir
+        - Replace target_dir and extract into the newly created directory
         - Always cleanup local tar.gz
 
         Returns:
@@ -216,3 +230,189 @@ async def ensure_ossutil(self) -> bool:
             logger.warning(f"ossutil verify failed: {verify.stderr}")
             return False
         return True
+
+
+class WindowsFileSystem(FileSystem):
+    """File system operations for Windows sandboxes."""
+
+    async def chown(self, request: ChownRequest) -> ChownResponse:
+        paths = request.paths
+        if paths is None or len(paths) == 0:
+            raise BadRequestRockError("paths is empty")
+
+        responses: list[str] = []
+        for path in paths:
+            command = ["icacls", path, "/setowner", request.remote_user]
+            if request.recursive:
+                command.append("/T")
+            logger.info(f"chown command: {command}")
+
+            response: CommandResponse = await self.sandbox.execute(Command(command=command))
+            responses.append(str(response))
+            if response.exit_code != 0:
+                return ChownResponse(success=False, message="\n".join(responses))
+        return ChownResponse(success=True, message="\n".join(responses))
+
+    async def chmod(self, request: ChmodRequest) -> ChmodResponse:
+        paths = request.paths
+        if paths is None or len(paths) == 0:
+            raise BadRequestRockError("paths is empty")
+
+        try:
+            mode = int(request.mode, 8)
+            if mode < 0 or mode > 0o7777:
+                raise ValueError
+        except ValueError:
+            return ChmodResponse(success=False, message=f"Windows chmod requires an octal mode: {request.mode}")
+
+        attribute = "-R" if mode & 0o222 else "+R"
+        responses: list[str] = []
+        for path in paths:
+            if request.recursive:
+                target = _powershell_literal(path)
+                command = (
+                    "$ErrorActionPreference = 'Stop'; "
+                    f"$target = {target}; "
+                    f"& attrib.exe {attribute} $target; "
+                    'if ($LASTEXITCODE -ne 0) { throw "attrib failed with exit code $LASTEXITCODE" }; '
+                    "if (Test-Path -LiteralPath $target -PathType Container) { "
+                    f"& attrib.exe {attribute} (Join-Path -Path $target -ChildPath '*') /S /D; "
+                    'if ($LASTEXITCODE -ne 0) { throw "attrib failed with exit code $LASTEXITCODE" } '
+                    "}"
+                )
+                logger.info(f"chmod command: {command}")
+                try:
+                    response = await self.sandbox.arun(cmd=command)
+                except Exception as e:
+                    responses.append(str(e))
+                    return ChmodResponse(success=False, message="\n".join(responses))
+            else:
+                command = ["attrib", attribute, path]
+                logger.info(f"chmod command: {command}")
+                response = await self.sandbox.execute(Command(command=command))
+
+            responses.append(str(response))
+            if response.exit_code != 0:
+                return ChmodResponse(success=False, message="\n".join(responses))
+        return ChmodResponse(success=True, message="\n".join(responses))
+
+    async def upload_dir(
+        self,
+        source_dir: str | Path,
+        target_dir: str,
+        extract_timeout: int = 600,
+    ) -> Observation:
+        """Upload a directory as ZIP and replace the target using PowerShell.
+
+        If ``target_dir`` already exists, it and all of its contents are
+        removed before extraction. Existing contents are not merged.
+        """
+        local_zip_path: Path | None = None
+        remote_zip_path: str | None = None
+        session: str | None = None
+
+        try:
+            src = Path(source_dir).expanduser().resolve()
+            if not src.exists():
+                return Observation(exit_code=1, failure_reason=f"source_dir not found: {src}")
+            if not src.is_dir():
+                return Observation(exit_code=1, failure_reason=f"source_dir must be a directory: {src}")
+            if not isinstance(target_dir, str) or not PureWindowsPath(target_dir).is_absolute():
+                return Observation(
+                    exit_code=1,
+                    failure_reason=f"target_dir must be absolute Windows path: {target_dir}",
+                )
+
+            ts = str(time.time_ns())
+            local_zip_path = Path(tempfile.gettempdir()) / f"rock_upload_{ts}.zip"
+            session = f"powershell-{ts}"
+
+            await self.sandbox.create_session(CreateBashSessionRequest(session=session))
+
+            check = await self.sandbox.arun(
+                cmd=(
+                    "if (-not (Get-Command Expand-Archive -ErrorAction SilentlyContinue)) "
+                    "{ throw 'Expand-Archive is not available' }; "
+                    "[System.IO.Path]::GetTempPath()"
+                ),
+                session=session,
+            )
+            if check.exit_code != 0:
+                return Observation(exit_code=1, failure_reason="sandbox has no Expand-Archive command")
+            remote_temp_dir = check.output.strip()
+            if not PureWindowsPath(remote_temp_dir).is_absolute():
+                return Observation(exit_code=1, failure_reason=f"invalid sandbox temporary path: {remote_temp_dir}")
+            remote_zip_path = str(PureWindowsPath(remote_temp_dir) / f"rock_upload_{ts}.zip")
+
+            shutil.make_archive(str(local_zip_path.with_suffix("")), "zip", root_dir=src)
+            upload_response = await self.sandbox.upload_by_path(
+                file_path=str(local_zip_path),
+                target_path=remote_zip_path,
+                upload_mode=UploadMode.DIRECT,
+            )
+            if not upload_response.success:
+                return Observation(exit_code=1, failure_reason=f"zip upload failed: {upload_response.message}")
+
+            archive = _powershell_literal(remote_zip_path)
+            target = _powershell_literal(target_dir)
+            extract = await self.sandbox.run_in_session(
+                BashAction(
+                    command=(
+                        "$ErrorActionPreference = 'Stop'; "
+                        f"if (Test-Path -LiteralPath {target}) "
+                        f"{{ Remove-Item -LiteralPath {target} -Recurse -Force }}; "
+                        f"New-Item -ItemType Directory -Path {target} -Force | Out-Null; "
+                        f"Expand-Archive -LiteralPath {archive} -DestinationPath {target} -Force"
+                    ),
+                    session=session,
+                    timeout=extract_timeout,
+                ),
+            )
+            if extract.exit_code != 0:
+                return Observation(exit_code=1, failure_reason=f"zip extract failed: {extract.output}")
+
+            return Observation(exit_code=0, output=f"uploaded {src} -> {target_dir} via zip")
+        except Exception as e:
+            return Observation(exit_code=1, failure_reason=f"upload_dir unexpected error: {e}")
+        finally:
+            if remote_zip_path and session:
+                try:
+                    await self.sandbox.arun(
+                        cmd=(
+                            f"Remove-Item -LiteralPath {_powershell_literal(remote_zip_path)} "
+                            "-Force -ErrorAction SilentlyContinue"
+                        ),
+                        session=session,
+                    )
+                except Exception:
+                    pass
+            if local_zip_path:
+                try:
+                    local_zip_path.unlink(missing_ok=True)
+                except Exception:
+                    pass
+
+    async def download_file(
+        self,
+        remote_path: str,
+        local_path: str | Path,
+    ) -> DownloadFileResponse:
+        """Download a Windows sandbox file through binary-safe Base64 output."""
+        try:
+            remote = _powershell_literal(remote_path)
+            response = await self.sandbox.arun(
+                cmd=f"[Convert]::ToBase64String([System.IO.File]::ReadAllBytes({remote}))"
+            )
+            if response.exit_code != 0:
+                return DownloadFileResponse(
+                    success=False,
+                    message=f"Failed to read remote file: {response.output}",
+                )
+
+            content = base64.b64decode("".join(response.output.split()), validate=True)
+            local = Path(local_path).expanduser().resolve()
+            local.parent.mkdir(parents=True, exist_ok=True)
+            local.write_bytes(content)
+            return DownloadFileResponse(success=True, message=f"Successfully downloaded {remote_path} to {local}")
+        except Exception as e:
+            return DownloadFileResponse(success=False, message=f"Failed to download {remote_path}: {e}")

tests/unit/rocklet/test_windows.py

@@ -0,0 +1,45 @@
+import io
+from unittest.mock import MagicMock
+
+from rock.admin.proto.request import SandboxBashAction, SandboxCreateBashSessionRequest
+from rock.rocklet.windows import PowerShellSession
+
+
+async def test_powershell_session_configures_non_echoing_input(monkeypatch):
+    process = MagicMock()
+    process.stdin = io.StringIO()
+    process.stdout = io.StringIO()
+    popen = MagicMock(return_value=process)
+    monkeypatch.setattr("rock.rocklet.windows.subprocess.Popen", popen)
+    monkeypatch.setattr("rock.rocklet.windows.time.sleep", lambda _: None)
+    monkeypatch.setattr(PowerShellSession, "_find_powershell", staticmethod(lambda: "powershell"))
+    monkeypatch.setattr(PowerShellSession, "_drain_queue", lambda self, timeout=0.1: "")
+
+    session = PowerShellSession(SandboxCreateBashSessionRequest(session="test", sandbox_id="sandbox"))
+    await session.start()
+
+    command = popen.call_args.args[0]
+    setup = command[command.index("-Command") + 1]
+    assert "function global:PSConsoleHostReadLine { [Console]::In.ReadLine() }" in setup
+    assert PowerShellSession._PROMPT_MARKER in setup
+
+
+def test_powershell_session_removes_internal_prompt_from_output(monkeypatch):
+    session = PowerShellSession(SandboxCreateBashSessionRequest(session="test", sandbox_id="sandbox"))
+    process = MagicMock()
+    process.stdin = io.StringIO()
+    session._process = process
+    monkeypatch.setattr(session, "_drain_queue", lambda timeout=0.1: "")
+
+    prompt = "ROCKLET_PS_PROMPT_29234"
+    session._output_queue.put(f"{prompt}{session._BEGIN_MARKER}\n")
+    session._output_queue.put(f"{prompt}ROCK_REAL_OUTPUT\n")
+    session._output_queue.put(f"{prompt}{session._EXIT_MARKER}0\n")
+    session._output_queue.put(f"{prompt}{session._END_MARKER}\n")
+
+    result = session._run_command(
+        SandboxBashAction(command="Write-Output 'ROCK_REAL_OUTPUT'", session="test", sandbox_id="sandbox")
+    )
+
+    assert result.output == "ROCK_REAL_OUTPUT"
+    assert result.exit_code == 0