kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: 完善sandbox backend，基于k8s完成所有工具的场景测试；解决创建pod，沙箱接口账号为gem，和容器启动账号root导致权限不一致的问题

Browse Source
This commit is contained in:
supreme0597 2026-03-25 00:00:23 +08:00
parent 7d0a2fb2f1
commit d62b36293d
3 changed files with 218 additions and 421 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

452
docker/sandbox_provisioner/app.py
View File

@ -62,7 +62,7 @@ class MemoryProvisionerBackend:
return template
def create(self, sandbox_id: str, thread_id: str) -> SandboxRecord:
del thread_id
_ = thread_id # unused in memory backend
with self._lock:
existing = self._records.get(sandbox_id)
if existing is not None:
@ -240,31 +240,16 @@ class LocalContainerProvisionerBackend:
raise RuntimeError(f"sandbox {sandbox_id} is not ready at {record.sandbox_url}")
return record
# 检测是否是 Windows 绝对路径 (如 D:/ 或 D:\)
threads_root_str = self._threads_host_path
is_windows_path = len(threads_root_str) >= 2 and threads_root_str[1] == ":"
threads_root = Path(self._threads_host_path).resolve()
thread_user_data = (threads_root / safe_thread_id / "user-data").resolve()
try:
thread_user_data.relative_to(threads_root)
except ValueError as exc:
raise ValueError("thread_id resolved outside threads host root") from exc
thread_user_data.mkdir(parents=True, exist_ok=True)
if is_windows_path:
# Windows 路径，直接使用，不调用 resolve()
threads_root = Path(threads_root_str)
thread_user_data = threads_root / safe_thread_id / "user-data"
# Windows 路径下无法在 Linux 容器内创建目录，跳过 mkdir
else:
threads_root = Path(threads_root_str).resolve()
thread_user_data = (threads_root / safe_thread_id / "user-data").resolve()
try:
thread_user_data.relative_to(threads_root)
except ValueError as exc:
raise ValueError("thread_id resolved outside threads host root") from exc
thread_user_data.mkdir(parents=True, exist_ok=True)
skills_path_str = self._skills_host_path
is_skills_windows = len(skills_path_str) >= 2 and skills_path_str[1] == ":"
if is_skills_windows:
skills_path = Path(skills_path_str)
else:
skills_path = Path(skills_path_str)
skills_path.mkdir(parents=True, exist_ok=True)
skills_path = Path(self._skills_host_path)
skills_path.mkdir(parents=True, exist_ok=True)
container_name = self._container_name(sandbox_id)
run_kwargs = {
@ -280,11 +265,6 @@ class LocalContainerProvisionerBackend:
str(thread_user_data): {"bind": "/home/gem/user-data", "mode": "rw"},
str(skills_path): {"bind": "/skills", "mode": "ro"},
},
"tmpfs": "/tmp:size=256m,mode=1777",
"environment": {
"HOME": "/home/gem/user-data",
"TMPDIR": "/tmp",
},
"ports": {f"{self._container_port}/tcp": None},
"security_opt": ["seccomp=unconfined"],
}
@ -345,12 +325,10 @@ class KubernetesProvisionerBackend:
"SANDBOX_IMAGE",
"enterprise-public-cn-beijing.cr.volces.com/vefaas-public/all-in-one-sandbox:latest",
)
self._skill_pvc = os.getenv("SKILLS_PVC", "yuxi-skills")
self._thread_pvc = os.getenv("THREAD_PVC", "yuxi-thread")
self._node_host = os.getenv("NODE_HOST", "host.docker.internal")
self._container_port = int(os.getenv("SANDBOX_CONTAINER_PORT", "8080"))
self._shared_userdata_pvc = os.getenv("K8S_SHARED_USERDATA_PVC", "shared-userdata-pvc")
self._shared_skills_pvc = os.getenv("K8S_SHARED_SKILLS_PVC", "shared-skills-pvc")
self._allow_shared_pvc_reuse = os.getenv("K8S_ALLOW_SHARED_PVC_REUSE", "false").strip().lower() == "true"
self._ready_timeout_seconds = int(os.getenv("SANDBOX_READY_TIMEOUT_SECONDS", "60"))
self._ready_poll_interval_seconds = max(1, int(os.getenv("SANDBOX_READY_POLL_INTERVAL_SECONDS", "2")))
kubeconfig_path = os.getenv("KUBECONFIG_PATH")
if kubeconfig_path:
@ -364,285 +342,82 @@ class KubernetesProvisionerBackend:
self._core_api = client.CoreV1Api()
self._client = client
@staticmethod
def _validate_thread_id(thread_id: str) -> str:
candidate = str(thread_id or "").strip()
if not candidate:
raise ValueError("thread_id is required")
if any(ch in candidate for ch in ("/", "\\", "\x00")):
raise ValueError("thread_id must be a single safe path segment")
if candidate in {".", ".."} or ".." in candidate:
raise ValueError("thread_id contains invalid path traversal sequence")
return candidate
@staticmethod
def _sanitize_k8s_name(value: str) -> str:
sanitized = "".join(ch if ch.isalnum() else "-" for ch in value.strip().lower())
while "--" in sanitized:
sanitized = sanitized.replace("--", "-")
sanitized = sanitized.strip("-")
return sanitized[:40] or "sandbox"
def _thread_subpath(self, thread_id: str) -> str:
safe_thread_id = self._validate_thread_id(thread_id)
return f"threads/{safe_thread_id}/user-data"
@staticmethod
def _pod_name(sandbox_id: str) -> str:
safe_id = KubernetesProvisionerBackend._sanitize_k8s_name(sandbox_id)
return f"sandbox-{safe_id}"
return f"sandbox-{sandbox_id}"
@staticmethod
def _service_name(sandbox_id: str) -> str:
safe_id = KubernetesProvisionerBackend._sanitize_k8s_name(sandbox_id)
return f"sandbox-{safe_id}"
def _sandbox_url(self, sandbox_id: str) -> str:
service_name = self._service_name(sandbox_id)
return f"http://{service_name}.{self._namespace}.svc.cluster.local:{self._container_port}"
def _read_pvc(self, pvc_name: str):
from kubernetes.client.rest import ApiException
try:
return self._core_api.read_namespaced_persistent_volume_claim(
name=pvc_name,
namespace=self._namespace,
)
except ApiException as exc:
if exc.status == 404:
return None
raise
def _read_pod(self, sandbox_id: str):
from kubernetes.client.rest import ApiException
pod_name = self._pod_name(sandbox_id)
try:
return self._core_api.read_namespaced_pod(name=pod_name, namespace=self._namespace)
except ApiException as exc:
if exc.status == 404:
return None
raise
def _read_service(self, sandbox_id: str):
from kubernetes.client.rest import ApiException
service_name = self._service_name(sandbox_id)
try:
return self._core_api.read_namespaced_service(name=service_name, namespace=self._namespace)
except ApiException as exc:
if exc.status == 404:
return None
raise
def _ensure_shared_pvcs_ready(self) -> None:
for pvc_name in {self._shared_userdata_pvc, self._shared_skills_pvc}:
pvc = self._read_pvc(pvc_name)
if pvc is None:
raise RuntimeError(f"required pvc not found: {pvc_name}")
phase = pvc.status.phase if pvc and pvc.status else None
if phase != "Bound":
raise RuntimeError(f"required pvc is not Bound: {pvc_name} phase={phase}")
def _validate_pvc_configuration(self) -> None:
same_pvc = self._shared_userdata_pvc == self._shared_skills_pvc
if same_pvc and not self._allow_shared_pvc_reuse:
raise RuntimeError(
"K8S_SHARED_USERDATA_PVC and K8S_SHARED_SKILLS_PVC point to the same PVC; "
"set K8S_ALLOW_SHARED_PVC_REUSE=true only if your storage supports this reliably"
)
def _assert_existing_sandbox_matches(self, sandbox_id: str, thread_id: str) -> None:
pod = self._read_pod(sandbox_id)
if pod is None:
return
annotations = (pod.metadata.annotations if pod.metadata else None) or {}
existing_thread_id = annotations.get("thread-id")
if existing_thread_id and existing_thread_id != thread_id:
raise RuntimeError(
f"sandbox {sandbox_id} already exists for thread_id={existing_thread_id}, not {thread_id}"
)
def _pod_ready(self, pod) -> bool:
if not pod or not pod.status or pod.status.phase != "Running":
return False
conditions = pod.status.conditions or []
ready_condition = next((c for c in conditions if c.type == "Ready"), None)
if not ready_condition or ready_condition.status != "True":
return False
container_statuses = pod.status.container_statuses or []
sandbox_container = next((c for c in container_statuses if c.name == "sandbox"), None)
return bool(sandbox_container and sandbox_container.ready is True)
def _extract_pod_failure(self, pod) -> str | None:
if not pod or not pod.status:
return None
if pod.status.phase == "Failed":
reason = pod.status.reason or "Unknown"
message = pod.status.message or ""
return f"pod failed: {reason} {message}".strip()
for cond in pod.status.conditions or []:
if cond.type == "PodScheduled" and cond.status == "False":
return f"pod scheduling failed: {(cond.reason or '').strip()} {(cond.message or '').strip()}".strip()
for container_status in pod.status.container_statuses or []:
state = container_status.state
if state and state.waiting:
reason = state.waiting.reason or "Waiting"
message = state.waiting.message or ""
if reason in {
"ImagePullBackOff",
"ErrImagePull",
"CrashLoopBackOff",
"CreateContainerConfigError",
"CreateContainerError",
}:
return f"container {container_status.name} waiting: {reason} {message}".strip()
if state and state.terminated:
reason = state.terminated.reason or "Terminated"
exit_code = state.terminated.exit_code
message = state.terminated.message or ""
return f"container {container_status.name} terminated: {reason} exit={exit_code} {message}".strip()
return None
return f"sandbox-{sandbox_id}"
def _build_pod_spec(self, sandbox_id: str, thread_id: str):
pod_name = self._pod_name(sandbox_id)
safe_thread_id = self._validate_thread_id(thread_id)
user_data_subpath = self._thread_subpath(safe_thread_id)
same_pvc = self._shared_userdata_pvc == self._shared_skills_pvc
if same_pvc:
volumes = [
self._client.V1Volume(
name="shared-storage",
persistent_volume_claim=self._client.V1PersistentVolumeClaimVolumeSource(
claim_name=self._shared_userdata_pvc,
),
)
]
volume_mounts = [
self._client.V1VolumeMount(
name="shared-storage",
mount_path="/home/gem/user-data",
sub_path=user_data_subpath,
),
self._client.V1VolumeMount(
name="shared-storage",
mount_path="/skills",
sub_path="skills",
read_only=True,
),
]
else:
volumes = [
self._client.V1Volume(
name="user-data",
persistent_volume_claim=self._client.V1PersistentVolumeClaimVolumeSource(
claim_name=self._shared_userdata_pvc,
),
),
self._client.V1Volume(
name="skills",
persistent_volume_claim=self._client.V1PersistentVolumeClaimVolumeSource(
claim_name=self._shared_skills_pvc,
read_only=True,
),
),
]
volume_mounts = [
self._client.V1VolumeMount(
name="user-data",
mount_path="/home/gem/user-data",
sub_path=user_data_subpath,
),
self._client.V1VolumeMount(
name="skills",
mount_path="/skills",
sub_path="skills",
read_only=True,
),
]
volume_mounts.append(self._client.V1VolumeMount(name="tmp", mount_path="/tmp"))
volumes.append(
self._client.V1Volume(
name="tmp",
empty_dir=self._client.V1EmptyDirVolumeSource(medium="Memory"),
)
)
init_script = f'set -eu; mkdir -p "/home/gem/user-data/threads/{safe_thread_id}/user-data"; test -d "/skills"'
init_volume_mounts = [
self._client.V1VolumeMount(name="tmp", mount_path="/tmp"),
]
if same_pvc:
init_volume_mounts.extend(
[
self._client.V1VolumeMount(name="shared-storage", mount_path="/home/gem/user-data"),
self._client.V1VolumeMount(name="shared-storage", mount_path="/skills", sub_path="skills"),
]
)
else:
init_volume_mounts.extend(
[
self._client.V1VolumeMount(name="user-data", mount_path="/home/gem/user-data"),
self._client.V1VolumeMount(name="skills", mount_path="/skills", sub_path="skills", read_only=True),
]
)
return self._client.V1Pod(
metadata=self._client.V1ObjectMeta(
name=pod_name,
labels={
"app": "yuxi-sandbox",
"sandbox-id": sandbox_id,
"managed-by": "yuxi-sandbox-provisioner",
},
annotations={"thread-id": safe_thread_id},
labels={"app": "yuxi-sandbox", "sandbox-id": sandbox_id},
annotations={"thread-id": thread_id},
),
spec=self._client.V1PodSpec(
restart_policy="Never",
security_context=self._client.V1PodSecurityContext(fs_group=1000),
security_context=self._client.V1PodSecurityContext(
fs_group=0,
run_as_user=0,
),
init_containers=[
self._client.V1Container(
name="prepare-user-data",
name="init-user-data",
image=self._sandbox_image,
command=["sh", "-lc", init_script],
volume_mounts=init_volume_mounts,
)
command=["sh", "-c"],
args=[
"chmod 777 /home/gem "
"&& mkdir -p /home/gem/user-data/workspace /home/gem/user-data/uploads /home/gem/user-data/outputs "
"&& chmod -R 777 /home/gem/user-data ",
],
volume_mounts=[
self._client.V1VolumeMount(name="home-dir", mount_path="/home/gem"),
self._client.V1VolumeMount(
name="shared-data",
mount_path="/home/gem/user-data",
sub_path=f"threads/{thread_id}/user-data",
),
],
),
],
containers=[
self._client.V1Container(
name="sandbox",
image=self._sandbox_image,
ports=[self._client.V1ContainerPort(container_port=self._container_port)],
env=[
self._client.V1EnvVar(name="HOME", value="/home/gem/user-data"),
self._client.V1EnvVar(name="TMPDIR", value="/tmp"),
volume_mounts=[
self._client.V1VolumeMount(name="home-dir", mount_path="/home/gem"),
self._client.V1VolumeMount(
name="shared-data",
mount_path="/home/gem/user-data",
sub_path=f"threads/{thread_id}/user-data",
),
self._client.V1VolumeMount(
name="shared-data",
mount_path="/skills",
sub_path="skills",
read_only=False,
),
],
volume_mounts=volume_mounts,
readiness_probe=self._client.V1Probe(
http_get=self._client.V1HTTPGetAction(
path="/v1/sandbox",
port=self._container_port,
),
period_seconds=2,
timeout_seconds=2,
failure_threshold=15,
),
startup_probe=self._client.V1Probe(
http_get=self._client.V1HTTPGetAction(
path="/v1/sandbox",
port=self._container_port,
),
period_seconds=2,
timeout_seconds=2,
failure_threshold=30,
),
)
],
volumes=volumes,
volumes=[
self._client.V1Volume(
name="shared-data",
persistent_volume_claim=self._client.V1PersistentVolumeClaimVolumeSource(
claim_name=self._thread_pvc,
read_only=False,
),
),
self._client.V1Volume(
name="home-dir",
empty_dir=self._client.V1EmptyDirVolumeSource(),
),
],
),
)
@ -651,14 +426,10 @@ class KubernetesProvisionerBackend:
return self._client.V1Service(
metadata=self._client.V1ObjectMeta(
name=service_name,
labels={
"app": "yuxi-sandbox",
"sandbox-id": sandbox_id,
"managed-by": "yuxi-sandbox-provisioner",
},
labels={"app": "yuxi-sandbox", "sandbox-id": sandbox_id},
),
spec=self._client.V1ServiceSpec(
type="ClusterIP",
type="NodePort",
selector={"sandbox-id": sandbox_id},
ports=[
self._client.V1ServicePort(
@ -671,55 +442,21 @@ class KubernetesProvisionerBackend:
),
)
def wait_ready(self, sandbox_id: str) -> SandboxRecord:
deadline = time.monotonic() + self._ready_timeout_seconds
sandbox_url = self._sandbox_url(sandbox_id)
last_failure = None
while time.monotonic() < deadline:
pod = self._read_pod(sandbox_id)
service = self._read_service(sandbox_id)
if pod is None or service is None:
time.sleep(self._ready_poll_interval_seconds)
continue
failure = self._extract_pod_failure(pod)
if failure:
raise RuntimeError(f"sandbox {sandbox_id} failed before ready: {failure}")
if self._pod_ready(pod):
if wait_for_sandbox_ready(sandbox_url, timeout_seconds=3):
return SandboxRecord(
sandbox_id=sandbox_id,
sandbox_url=sandbox_url,
status="Running",
)
last_failure = f"http health check not ready at {sandbox_url}"
time.sleep(self._ready_poll_interval_seconds)
raise RuntimeError(f"sandbox {sandbox_id} timed out waiting ready: {last_failure or 'unknown'}")
def create(self, sandbox_id: str, thread_id: str) -> SandboxRecord:
from kubernetes.client.rest import ApiException
with self._lock:
safe_thread_id = self._validate_thread_id(thread_id)
self._validate_pvc_configuration()
self._ensure_shared_pvcs_ready()
discovered = self.discover(sandbox_id)
if discovered is not None:
return discovered
existing_pod = self._read_pod(sandbox_id)
existing_service = self._read_service(sandbox_id)
if existing_pod is not None:
self._assert_existing_sandbox_matches(sandbox_id, safe_thread_id)
if existing_pod is not None and existing_service is not None:
return self.wait_ready(sandbox_id)
self._pod_name(sandbox_id)
self._service_name(sandbox_id)
try:
self._core_api.create_namespaced_pod(
namespace=self._namespace,
body=self._build_pod_spec(sandbox_id, safe_thread_id),
body=self._build_pod_spec(sandbox_id, thread_id),
)
except ApiException as exc:
if exc.status != 409:
@ -734,27 +471,43 @@ class KubernetesProvisionerBackend:
if exc.status != 409:
raise
return self.wait_ready(sandbox_id)
health_timeout = int(os.getenv("SANDBOX_HEALTH_TIMEOUT_SECONDS", "60"))
record = self.discover(sandbox_id)
if record is None:
raise RuntimeError(f"failed to discover sandbox after create: {sandbox_id}")
if not wait_for_sandbox_ready(record.sandbox_url, timeout_seconds=health_timeout):
try:
self.delete(sandbox_id)
except Exception:
pass
raise RuntimeError(f"sandbox {sandbox_id} is not ready at {record.sandbox_url}")
return record
def discover(self, sandbox_id: str) -> SandboxRecord | None:
pod = self._read_pod(sandbox_id)
service = self._read_service(sandbox_id)
if pod is None or service is None:
return None
from kubernetes.client.rest import ApiException
failure = self._extract_pod_failure(pod)
if failure:
status = f"Failed:{failure}"
elif self._pod_ready(pod):
status = "Running"
pod_name = self._pod_name(sandbox_id)
service_name = self._service_name(sandbox_id)
try:
pod = self._core_api.read_namespaced_pod(name=pod_name, namespace=self._namespace)
service = self._core_api.read_namespaced_service(name=service_name, namespace=self._namespace)
except ApiException as exc:
if exc.status == 404:
return None
raise
node_port = None
if service.spec and service.spec.ports:
node_port = service.spec.ports[0].node_port
if not node_port:
sandbox_url = ""
else:
phase = pod.status.phase if pod and pod.status else "Unknown"
status = f"NotReady:{phase}"
sandbox_url = f"http://{self._node_host}:{node_port}"
return SandboxRecord(
sandbox_id=sandbox_id,
sandbox_url=self._sandbox_url(sandbox_id),
status=status,
sandbox_url=sandbox_url,
status=(pod.status.phase if pod and pod.status else "Unknown"),
)
def list(self) -> list[SandboxRecord]:
@ -803,7 +556,7 @@ class SandboxIdleReaper:
self._stop_event = threading.Event()
self._thread: threading.Thread | None = None
self._exec_timeout_seconds = int(os.getenv("SANDBOX_EXEC_TIMEOUT_SECONDS", "180"))
configured_idle_timeout = int(os.getenv("SANDBOX_IDLE_TIMEOUT_SECONDS", "120"))
configured_idle_timeout = int(os.getenv("SANDBOX_IDLE_TIMEOUT_SECONDS", "600"))
if 0 < configured_idle_timeout <= self._exec_timeout_seconds:
logger.warning(
"SANDBOX_IDLE_TIMEOUT_SECONDS=%s is <= SANDBOX_EXEC_TIMEOUT_SECONDS=%s; "
@ -913,6 +666,7 @@ def health():
@app.post("/api/sandboxes", response_model=SandboxResponse)
def create_sandbox(payload: CreateSandboxRequest):
try:
# Backend.create() already handles container reuse (discovers existing container first)
record = backend_impl.create(payload.sandbox_id, payload.thread_id)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc

64
src/agents/common/backends/composite.py
View File

@ -1,13 +1,71 @@
from __future__ import annotations
from deepagents.backends import CompositeBackend
from deepagents.backends.composite import (
CompositeBackend,
_route_for_path,
_remap_file_info_path,
_strip_route_from_pattern,
)
from deepagents.backends.protocol import FileInfo
from src.agents.common.middlewares.skills_middleware import normalize_selected_skills
from src.sandbox import ProvisionerSandboxBackend
from .skills_backend import SelectedSkillsReadonlyBackend
class CustomCompositeBackend(CompositeBackend):
"""修复 glob_info 路由逻辑的 CompositeBackend。
修复内容：当 path 不匹配任何路由时应该只搜索 default 后端，
而不是错误地遍历所有路由后端搜索。
"""
def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
backend, backend_path, route_prefix = _route_for_path(
default=self.default,
sorted_routes=self.sorted_routes,
path=path,
)
if route_prefix is not None:
infos = backend.glob_info(pattern, backend_path)
return [_remap_file_info_path(fi, route_prefix) for fi in infos]
# 只在 path 为 None 或 "/" 时搜索所有后端，其他只搜索 default
if path is None or path == "/":
results: list[FileInfo] = []
results.extend(self.default.glob_info(pattern, path))
for route_prefix, backend in self.routes.items():
route_pattern = _strip_route_from_pattern(pattern, route_prefix)
infos = backend.glob_info(route_pattern, "/")
results.extend(_remap_file_info_path(fi, route_prefix) for fi in infos)
results.sort(key=lambda x: x.get("path", ""))
return results
return self.default.glob_info(pattern, path)
async def aglob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
backend, backend_path, route_prefix = _route_for_path(
default=self.default,
sorted_routes=self.sorted_routes,
path=path,
)
if route_prefix is not None:
infos = await backend.aglob_info(pattern, backend_path)
return [_remap_file_info_path(fi, route_prefix) for fi in infos]
if path is None or path == "/":
results: list[FileInfo] = []
results.extend(await self.default.aglob_info(pattern, path))
for route_prefix, backend in self.routes.items():
route_pattern = _strip_route_from_pattern(pattern, route_prefix)
infos = await backend.aglob_info(route_pattern, "/")
results.extend(_remap_file_info_path(fi, route_prefix) for fi in infos)
results.sort(key=lambda x: x.get("path", ""))
return results
return await self.default.aglob_info(pattern, path)
def _get_visible_skills_from_runtime(runtime) -> list[str]:
"""获取运行时可见的 skills 列表"""
context = getattr(runtime, "context", None)
@ -35,7 +93,7 @@ def _extract_thread_id(runtime) -> str:
def create_agent_composite_backend(runtime) -> CompositeBackend:
visible_skills = _get_visible_skills_from_runtime(runtime)
thread_id = _extract_thread_id(runtime)
return CompositeBackend(
return CustomCompositeBackend(
default=ProvisionerSandboxBackend(thread_id=thread_id),
routes={
"/skills/": SelectedSkillsReadonlyBackend(selected_slugs=visible_skills),

123
src/sandbox/backend.py
View File

@ -71,16 +71,23 @@ class ProvisionerSandboxBackend(BaseSandbox):
return self._client
def _read_binary(self, path: str) -> bytes:
def _read_binary(self, path: str, offset: int = 0, limit: int | None = None) -> bytes:
"""Read file content from the sandbox file API and normalize it to bytes.
The underlying API may return base64 text, raw bytes, or plain strings.
This helper is the single normalization point used by read(), edit(), and
download_files() so all read paths share the same transport semantics.
"""
result = self._get_client().file.read_file(file=path, encoding="base64")
start_line = max(0, int(offset)) if offset else None
end_line = (start_line + int(limit)) if limit and start_line is not None else None
content = result.content
result = self._get_client().file.read_file(
file=path,
start_line=start_line,
end_line=end_line,
)
content = result.data.content
if content is None:
return b""
if isinstance(content, bytes):
@ -106,10 +113,9 @@ class ProvisionerSandboxBackend(BaseSandbox):
"""
normalized_path = _normalize_path(file_path)
start = max(0, int(offset))
size = max(0, int(limit))
try:
content = self._read_binary(normalized_path)
content = self._read_binary(normalized_path, offset=offset, limit=limit)
except Exception: # noqa: BLE001
return f"Error: File '{file_path}' not found"
@ -117,13 +123,11 @@ class ProvisionerSandboxBackend(BaseSandbox):
return "System reminder: File exists but has empty contents"
text = content.decode("utf-8", errors="replace")
lines = text.splitlines()
selected_lines = lines[start : start + size]
if not selected_lines:
if not text:
return ""
return "\n".join(f"{start + idx + 1:6d}\t{line}" for idx, line in enumerate(selected_lines))
lines = text.splitlines()
return "\n".join(f"{start + idx + 1:6d}\t{line}" for idx, line in enumerate(lines))
def execute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse:
"""Execute a shell command in the sandbox.
@ -137,19 +141,8 @@ class ProvisionerSandboxBackend(BaseSandbox):
kwargs["timeout"] = timeout
result = self._get_client().shell.exec_command(**kwargs)
output = getattr(result, "output", None)
if output is None and isinstance(result, dict):
output = result.get("output")
if output is None:
output = str(result) if result is not None else ""
if not isinstance(output, str):
output = str(output)
exit_code = getattr(result, "exit_code", None)
if exit_code is None and isinstance(result, dict):
exit_code = result.get("exit_code")
if isinstance(exit_code, str) and exit_code.isdigit():
exit_code = int(exit_code)
output = result.data.output or ""
exit_code = result.data.exit_code
truncated = False
encoded = output.encode("utf-8", errors="ignore")
@ -174,7 +167,7 @@ class ProvisionerSandboxBackend(BaseSandbox):
except Exception: # noqa: BLE001
return []
entries = result.files or []
entries = result.data.files or []
infos: list[FileInfo] = []
for entry in entries:
info: FileInfo = {"path": entry.path, "is_dir": entry.is_directory}
@ -182,11 +175,16 @@ class ProvisionerSandboxBackend(BaseSandbox):
if isinstance(size, int):
info["size"] = size
modified_time = entry.modified_time
if isinstance(modified_time, str) and modified_time:
try:
info["modified_at"] = datetime.fromisoformat(modified_time).isoformat()
except ValueError:
info["modified_at"] = modified_time
if modified_time:
if isinstance(modified_time, str) and modified_time.isdigit():
info["modified_at"] = datetime.fromtimestamp(int(modified_time)).isoformat()
elif isinstance(modified_time, str):
try:
info["modified_at"] = datetime.fromisoformat(modified_time).isoformat()
except ValueError:
info["modified_at"] = modified_time
elif isinstance(modified_time, (int, float)):
info["modified_at"] = datetime.fromtimestamp(modified_time).isoformat()
infos.append(info)
return infos
@ -207,7 +205,9 @@ class ProvisionerSandboxBackend(BaseSandbox):
return WriteResult(error=f"Error: File '{file_path}' already exists")
try:
self._get_client().file.write_file(file=normalized_path, content=content)
result = self._get_client().file.write_file(file=normalized_path, content=content)
if not result.success:
return WriteResult(error=result.message or f"Failed to write file '{file_path}'")
except Exception as exc: # noqa: BLE001
return WriteResult(error=str(exc) or f"Failed to write file '{file_path}'")
@ -226,6 +226,8 @@ class ProvisionerSandboxBackend(BaseSandbox):
are not supported here and should be handled via download/upload flows.
"""
normalized_path = _normalize_path(file_path)
# Check if old_string exists
try:
text = self._read_binary(normalized_path).decode("utf-8", errors="replace")
except Exception: # noqa: BLE001
@ -242,12 +244,20 @@ class ProvisionerSandboxBackend(BaseSandbox):
)
)
updated = text.replace(old_string, new_string) if replace_all else text.replace(old_string, new_string, 1)
# Use str_replace_editor API
replace_mode = "ALL" if replace_all else "FIRST"
try:
self._get_client().file.write_file(file=normalized_path, content=updated)
result = self._get_client().file.str_replace_editor(
command="str_replace",
path=normalized_path,
old_str=old_string,
new_str=new_string,
replace_mode=replace_mode,
)
if not result.data.success:
return EditResult(error=result.data.message or f"Error editing file '{file_path}'")
except Exception as exc: # noqa: BLE001
return EditResult(error=f"Error editing file (exit code 1): {exc or 'Unknown error'}")
return EditResult(error=f"Error editing file: {exc}")
return EditResult(path=normalized_path, files_update=None, occurrences=count if replace_all else 1)
@ -263,56 +273,29 @@ class ProvisionerSandboxBackend(BaseSandbox):
optional include glob.
"""
search_path = _normalize_path(path or "/")
include = [glob] if glob else None
try:
result = self._get_client().file.grep_files(
path=search_path,
pattern=pattern,
include=include,
fixed_strings=True,
recursive=True,
)
return super().grep_raw(pattern=pattern, path=search_path, glob=glob)
except Exception as exc: # noqa: BLE001
return str(exc)
matches_out: list[GrepMatch] = []
for match in result.matches or []:
matches_out.append(
{
"path": match.file,
"line": match.line_number,
"text": match.line_content,
}
)
return matches_out
def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
"""Return files matching a glob pattern with optional metadata."""
normalized_path = _normalize_path(path)
try:
result = self._get_client().file.glob_files(
# return super().glob_info(pattern=pattern, path=path)
result = self._get_client().file.find_files(
path=normalized_path,
pattern=pattern,
include_metadata=True,
glob=pattern,
)
except Exception: # noqa: BLE001
return []
infos: list[FileInfo] = []
for entry in result.files or []:
info: FileInfo = {"path": entry.path}
if isinstance(entry.is_directory, bool):
info["is_dir"] = entry.is_directory
if isinstance(entry.size, int):
info["size"] = entry.size
if isinstance(entry.modified_time, str) and entry.modified_time:
try:
info["modified_at"] = datetime.fromisoformat(entry.modified_time).isoformat()
except ValueError:
info["modified_at"] = entry.modified_time
infos.append(info)
for file_path in result.data.files or []:
infos.append({"path": file_path})
return infos
def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]:
@ -325,11 +308,13 @@ class ProvisionerSandboxBackend(BaseSandbox):
for path, content in files:
try:
normalized_path = _normalize_path(path)
self._get_client().file.write_file(
result = self._get_client().file.write_file(
file=normalized_path,
content=base64.b64encode(content).decode("ascii"),
encoding="base64",
)
if not result.success:
raise Exception(result.message or "Upload failed")
responses.append(FileUploadResponse(path=normalized_path, error=None))
except PermissionError:
normalized_path = str(path)