fix: 修复长命令取消后沙盒清理阻塞及会话排队

backend/package/yuxi/agents/backends/sandbox/backend.py

@@ -545,31 +545,99 @@ def execute(self, command: str, *, timeout: int | None = None) -> ExecuteRespons
         payload size before being returned.
         """
         try:
-            kwargs: dict[str, Any] = {"command": command}
-            if timeout is not None:
-                kwargs["timeout"] = timeout
-                kwargs["hard_timeout"] = timeout
-                kwargs["request_options"] = {"timeout_in_seconds": timeout}
+            effective_timeout = timeout if timeout is not None else self._command_timeout_seconds
+            kwargs: dict[str, Any] = {
+                "command": command,
+                "timeout": effective_timeout,
+                "hard_timeout": effective_timeout,
+                "request_options": {"timeout_in_seconds": effective_timeout},
+            }
             result = self._get_client().shell.exec_command(**kwargs)
+            return self._execute_response(result.data)
+        except Exception as exc:  # noqa: BLE001
+            logger.error(f"Sandbox execute failed for thread {self._thread_id}: {exc}")
+            return ExecuteResponse(output=f"Error: {exc}", exit_code=1, truncated=False)
 
-            output = result.data.output or ""
-            exit_code = result.data.exit_code
-
-            truncated = False
-            encoded = output.encode("utf-8", errors="ignore")
-            if len(encoded) > self._max_output_bytes:
-                output = encoded[: self._max_output_bytes].decode("utf-8", errors="ignore")
-                truncated = True
-
-            return ExecuteResponse(
-                output=output,
-                exit_code=exit_code if isinstance(exit_code, int) else None,
-                truncated=truncated,
-            )
+    async def aexecute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse:
+        """以独立 GEM session 执行命令，取消时显式终止远端进程。"""
+        effective_timeout = timeout if timeout is not None else self._command_timeout_seconds
+        try:
+            connection = await asyncio.to_thread(self._get_connection)
+            async with httpx.AsyncClient(timeout=10, trust_env=False) as http_client:
+                client = self._build_async_client(connection.sandbox_url, http_client)
+                session_id = f"yuxi-{uuid.uuid4().hex}"
+                await client.shell.create_session(id=session_id, request_options={"timeout_in_seconds": 10})
+                launch = asyncio.create_task(
+                    client.shell.exec_command(
+                        id=session_id,
+                        command=command,
+                        async_mode=True,
+                        hard_timeout=effective_timeout,
+                        request_options={"timeout_in_seconds": 10},
+                    )
+                )
+                try:
+                    async with asyncio.timeout(effective_timeout + 10):
+                        # 启动确认先于终止，避免 kill 到达后命令才在远端开始。
+                        result = await asyncio.shield(launch)
+                        status = result.data.status
+                        while status in {"running", "no_change_timeout"}:
+                            waited = await client.shell.wait_for_process(
+                                id=session_id,
+                                seconds=1,
+                                max_wait_seconds=1,
+                                request_options={"timeout_in_seconds": 5},
+                            )
+                            status = waited.data.status
+                        final = await client.shell.view(id=session_id, request_options={"timeout_in_seconds": 5})
+                        return self._execute_response(final.data)
+                except BaseException:
+                    await self._terminate_command(client, session_id, launch)
+                    raise
         except Exception as exc:  # noqa: BLE001
             logger.error(f"Sandbox execute failed for thread {self._thread_id}: {exc}")
             return ExecuteResponse(output=f"Error: {exc}", exit_code=1, truncated=False)
 
+    async def _terminate_command(self, client: Any, session_id: str, launch: asyncio.Task) -> None:
+        """收敛启动与 kill，重复取消不能打断这个有界的远端收尾。"""
+
+        async def stop():
+            """启动失败仍尝试 kill，处理响应丢失但远端已执行的情形。"""
+            with suppress(Exception):
+                async with asyncio.timeout(10):
+                    await launch
+            async with asyncio.timeout(5):
+                await client.shell.kill_process(id=session_id, request_options={"timeout_in_seconds": 5})
+
+        cleanup = asyncio.create_task(stop())
+        while not cleanup.done():
+            try:
+                await asyncio.shield(cleanup)
+            except asyncio.CancelledError:
+                continue
+            except Exception:
+                break
+        try:
+            cleanup.result()
+        except Exception:
+            logger.exception(f"Sandbox command termination failed: thread={self._thread_id}, session={session_id}")
+
+    def _execute_response(self, data: Any) -> ExecuteResponse:
+        """归一化最终命令输出并限制返回字节数。"""
+        output = data.output or ""
+        encoded = output.encode("utf-8", errors="ignore")
+        truncated = len(encoded) > self._max_output_bytes
+        if truncated:
+            output = encoded[: self._max_output_bytes].decode("utf-8", errors="ignore")
+        exit_code = data.exit_code
+        if exit_code is None and getattr(data, "status", None) in {"hard_timeout", "terminated"}:
+            exit_code = 124 if data.status == "hard_timeout" else 130
+        return ExecuteResponse(
+            output=output,
+            exit_code=exit_code if isinstance(exit_code, int) else None,
+            truncated=truncated,
+        )
+
     def ls(self, path: str) -> LsResult:
         """List direct children of an allowed sandbox path with lightweight metadata."""
         try:

backend/package/yuxi/services/agent_run_service.py

@@ -1051,6 +1051,6 @@ async def get_active_run_by_thread(*, thread_id: str, current_uid: str, db: Asyn
         .limit(1)
     )
     run = result.scalar_one_or_none()
-    if run and run.status in ("pending", "running", "cancel_requested", "interrupted"):
+    if run and (run.runtime_cleanup_pending or run.status in ("pending", "running", "cancel_requested", "interrupted")):
         return {"run": run.to_dict()}
     return {"run": None}

backend/package/yuxi/services/run_worker.py

@@ -494,7 +494,7 @@ async def reconcile_pending_runtime_cleanups() -> list[str]:
             if not await _release_runtime_if_idle(run):
                 continue
         except Exception:
-            logger.error("Failed to reconcile execution-tree runtime cleanup: run=%s", run.id, exc_info=True)
+            logger.exception(f"Failed to reconcile execution-tree runtime cleanup: run={run.id}")
             continue
         if run.status in TERMINAL_RUN_STATUSES:
             await _append_end_event(run.id, run.status, thread_id=run.conversation_thread_id)

backend/test/http_helpers.py

@@ -0,0 +1,29 @@
+"""真实 TCP HTTP 测试的服务生命周期。"""
+
+import asyncio
+import socket
+from contextlib import asynccontextmanager
+
+import uvicorn
+
+
+@asynccontextmanager
+async def live_http_server(app):
+    """启动临时本地服务，结束后关闭监听与连接。"""
+    listener = socket.socket()
+    listener.bind(("127.0.0.1", 0))
+    port = listener.getsockname()[1]
+    server = uvicorn.Server(uvicorn.Config(app, lifespan="off", log_level="error"))
+    serving = asyncio.create_task(server.serve(sockets=[listener]))
+    try:
+        async with asyncio.timeout(5):
+            while not server.started:
+                if serving.done():
+                    await serving
+                    raise RuntimeError("Test HTTP server did not start")
+                await asyncio.sleep(0.01)
+        yield f"http://127.0.0.1:{port}"
+    finally:
+        server.should_exit = True
+        await asyncio.wait_for(serving, timeout=5)
+        listener.close()