From d62b36293dd3abfdd6ecba2359b606b1df689538 Mon Sep 17 00:00:00 2001 From: supreme0597 Date: Wed, 25 Mar 2026 00:00:23 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=AE=8C=E5=96=84sandbox=20backend?= =?UTF-8?q?=EF=BC=8C=E5=9F=BA=E4=BA=8Ek8s=E5=AE=8C=E6=88=90=E6=89=80?= =?UTF-8?q?=E6=9C=89=E5=B7=A5=E5=85=B7=E7=9A=84=E5=9C=BA=E6=99=AF=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=EF=BC=9B=E8=A7=A3=E5=86=B3=E5=88=9B=E5=BB=BApod?= =?UTF-8?q?=EF=BC=8C=E6=B2=99=E7=AE=B1=E6=8E=A5=E5=8F=A3=E8=B4=A6=E5=8F=B7?= =?UTF-8?q?=E4=B8=BAgem=EF=BC=8C=E5=92=8C=E5=AE=B9=E5=99=A8=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=E8=B4=A6=E5=8F=B7root=E5=AF=BC=E8=87=B4=E6=9D=83?= =?UTF-8?q?=E9=99=90=E4=B8=8D=E4=B8=80=E8=87=B4=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docker/sandbox_provisioner/app.py | 452 ++++++------------------ src/agents/common/backends/composite.py | 64 +++- src/sandbox/backend.py | 123 +++---- 3 files changed, 218 insertions(+), 421 deletions(-) diff --git a/docker/sandbox_provisioner/app.py b/docker/sandbox_provisioner/app.py index 406c7b97..217e70de 100644 --- a/docker/sandbox_provisioner/app.py +++ b/docker/sandbox_provisioner/app.py @@ -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 diff --git a/src/agents/common/backends/composite.py b/src/agents/common/backends/composite.py index 6c3ea0b1..370dc033 100644 --- a/src/agents/common/backends/composite.py +++ b/src/agents/common/backends/composite.py @@ -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), diff --git a/src/sandbox/backend.py b/src/sandbox/backend.py index b1d881ce..cfe02dee 100644 --- a/src/sandbox/backend.py +++ b/src/sandbox/backend.py @@ -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)