From 7b541a22563eae85326fb489d5ed5b63e6563f51 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Wed, 2 Sep 2026 16:21:33 +0800 Subject: [PATCH] feat(sandbox): raise per-user execution capacity --- config/agent.yaml | 2 +- core/executor.py | 1 + core/executor_docker.py | 8 +++- core/loop.py | 1 + core/sandbox/capacity.py | 60 ++++++++++++++++++++++--- core/sinks.py | 1 + tests/frontend_chat_queue.test.mjs | 9 ++++ tests/test_executor_docker.py | 27 ++++++++++- tests/test_sandbox_capacity_packages.py | 32 ++++++++++++- web/static/js/chat.js | 22 +++++++++ 10 files changed, 152 insertions(+), 11 deletions(-) diff --git a/config/agent.yaml b/config/agent.yaml index 8295f46..0c82f80 100644 --- a/config/agent.yaml +++ b/config/agent.yaml @@ -78,7 +78,7 @@ sandbox: idle_ttl_seconds: 600 # 普通容器无 exec 活动 10 分钟回收 max_active_execs: 6 # 宿主硬上限；前后台统一计数 max_background_execs: 4 - max_active_execs_per_user: 2 + max_active_execs_per_user: 3 min_mem_available: 1g # 低于阈值暂停新放行，不杀已运行任务 # 容器 DNS server 显式配置(docker run --dns,容器 /etc/resolv.conf 直接写, # 绕过 docker daemon 上游 DNS 探测路径;腾讯云轻量 / 部分云上 daemon 探测 diff --git a/core/executor.py b/core/executor.py index d9895f1..699d5f0 100644 --- a/core/executor.py +++ b/core/executor.py @@ -35,6 +35,7 @@ class ExecCtx: task_id: UUID working_dir: Path cancel_check: Optional[Callable[[], bool]] = None + event_emit: Optional[Callable[[dict], None]] = None @dataclass diff --git a/core/executor_docker.py b/core/executor_docker.py index 04efcb0..732fbf2 100644 --- a/core/executor_docker.py +++ b/core/executor_docker.py @@ -163,7 +163,13 @@ class DockerExecutor(Executor): def _call_heavy_foreground(self, name: str, args: Dict[str, Any], ctx: ExecCtx) -> ToolResult: """前台重型工具先排共享槽；排队不进入命令 timeout。""" - with self.pool.capacity.foreground(str(self.user_id), ctx.cancel_check) as admitted: + def notify(payload: dict) -> None: + if ctx.event_emit is not None: + ctx.event_emit({"type": "execution_queue", **payload}) + + with self.pool.capacity.foreground( + str(self.user_id), ctx.cancel_check, notify + ) as admitted: if not admitted: return ToolResult(content="[Error] command cancelled by user while queued", exit_code=130) self.pool.exec_started(self.user_id) diff --git a/core/loop.py b/core/loop.py index 88ef9cc..7a21a4a 100644 --- a/core/loop.py +++ b/core/loop.py @@ -772,6 +772,7 @@ class AgentLoop: task_id=self.session.task_id, working_dir=self.working_dir, cancel_check=self.cancel_check, + event_emit=self._emit, ) tool_started_at = time.time() tool_result = self.executor.call_tool(name, args, ctx) diff --git a/core/sandbox/capacity.py b/core/sandbox/capacity.py index d742788..0a6e494 100644 --- a/core/sandbox/capacity.py +++ b/core/sandbox/capacity.py @@ -10,7 +10,7 @@ import json import os import time import uuid -from contextlib import contextmanager +from contextlib import contextmanager, suppress from pathlib import Path from typing import Callable, Dict, Iterator, Optional @@ -18,7 +18,7 @@ from core.file_store import atomic_write_text, interprocess_file_lock DEFAULT_MAX_ACTIVE_EXECS = 6 DEFAULT_MAX_BACKGROUND_EXECS = 4 -DEFAULT_MAX_ACTIVE_EXECS_PER_USER = 2 +DEFAULT_MAX_ACTIVE_EXECS_PER_USER = 3 DEFAULT_MIN_MEM_AVAILABLE_BYTES = 1024 ** 3 @@ -123,13 +123,21 @@ class ExecCapacity: self._write(state) return lease_id - def acquire_foreground(self, user_id: str, cancel_check: Optional[Callable[[], bool]] = None) -> Optional[str]: + def acquire_foreground( + self, + user_id: str, + cancel_check: Optional[Callable[[], bool]] = None, + wait_notify: Optional[Callable[[dict], None]] = None, + ) -> Optional[str]: ticket = uuid.uuid4().hex + wait_notified = False with interprocess_file_lock(self.lock_path, timeout_seconds=None): state = self._read(); self._prune(state) state["foreground_queue"].append({"ticket": ticket, "user_id": str(user_id), "created_ts": time.time(), "owner_pid": os.getpid()}) self._write(state) while True: + admitted = False + notification = None with interprocess_file_lock(self.lock_path, timeout_seconds=None): state = self._read(); self._prune(state) queue = state["foreground_queue"] @@ -138,8 +146,41 @@ class ExecCapacity: queue[:] = [q for q in queue if q.get("ticket") != ticket] state["leases"][ticket] = {"lease_id": ticket, "user_id": str(user_id), "kind": "foreground", "owner_pid": os.getpid(), "started_ts": time.time()} self._write(state) - return ticket - self._write(state) + admitted = True + else: + self._write(state) + if not admitted and not wait_notified and wait_notify is not None: + leases = list(state["leases"].values()) + user_running = sum( + 1 for lease in leases + if lease.get("user_id") == str(user_id) + ) + available = mem_available_bytes() + if user_running >= self.max_per_user: + reason = "per_user_limit" + elif len(leases) >= self.max_active: + reason = "global_limit" + elif available is not None and available < self.min_mem_available: + reason = "memory_pressure" + else: + reason = "queue_order" + notification = { + "state": "waiting", + "reason": reason, + "user_running": user_running, + "user_limit": self.max_per_user, + "global_running": len(leases), + "global_limit": self.max_active, + } + wait_notified = True + if admitted: + if wait_notified and wait_notify is not None: + with suppress(Exception): + wait_notify({"state": "admitted"}) + return ticket + if notification is not None and wait_notify is not None: + with suppress(Exception): + wait_notify(notification) if cancel_check is not None and cancel_check(): self.cancel_waiter(ticket) return None @@ -189,8 +230,13 @@ class ExecCapacity: self._write(state) @contextmanager - def foreground(self, user_id: str, cancel_check: Optional[Callable[[], bool]] = None) -> Iterator[bool]: - lease = self.acquire_foreground(user_id, cancel_check) + def foreground( + self, + user_id: str, + cancel_check: Optional[Callable[[], bool]] = None, + wait_notify: Optional[Callable[[dict], None]] = None, + ) -> Iterator[bool]: + lease = self.acquire_foreground(user_id, cancel_check, wait_notify) try: yield lease is not None finally: diff --git a/core/sinks.py b/core/sinks.py index 8c9273e..df6c791 100644 --- a/core/sinks.py +++ b/core/sinks.py @@ -7,6 +7,7 @@ Loop 不直接 print,改 emit({type, ...})。Sink 决定怎么呈现。 llm_end {type, prompt_tokens, completion_tokens, elapsed} text {type, content} —— assistant 文字段(整段,非流式) tool_call {type, name, args, args_preview} + execution_queue {type, state, reason, ...} —— 前台重型工具等待/放行 tool_result {type, name, result, preview, truncated} done {type} —— 一次 run 全部结束 diff --git a/tests/frontend_chat_queue.test.mjs b/tests/frontend_chat_queue.test.mjs index 8f89b3e..74266f7 100644 --- a/tests/frontend_chat_queue.test.mjs +++ b/tests/frontend_chat_queue.test.mjs @@ -67,6 +67,15 @@ test("composer uses one Codex-style action button for stop and queued send", () assert.match(chat, /removeQueuedMessage\([\s\S]*streamSse\(r\.events_url, run\)/); }); +test("heavy execution queue exposes the blocking reason", () => { + const chat = readFileSync(new URL("../web/static/js/chat.js", import.meta.url), "utf8"); + assert.match(chat, /t === "execution_queue"/); + assert.match(chat, /等待执行容量（当前用户/); + assert.match(chat, /等待执行容量（整机/); + assert.match(chat, /等待执行容量（宿主内存压力）/); + assert.match(chat, /dataset\.runningLabel/); +}); + test("automatic title polling starts before a long-running response finishes", () => { const chat = readFileSync(new URL("../web/static/js/chat.js", import.meta.url), "utf8"); const sendMessage = chat.indexOf("async function sendMessage("); diff --git a/tests/test_executor_docker.py b/tests/test_executor_docker.py index f0f61cb..595625f 100644 --- a/tests/test_executor_docker.py +++ b/tests/test_executor_docker.py @@ -38,9 +38,13 @@ class FakePool: self.mark_active_calls = [] self.active = 0 self.capacity = self + self.capacity_notices = [] @contextmanager - def foreground(self, user_id, cancel_check=None): + def foreground(self, user_id, cancel_check=None, wait_notify=None): + if wait_notify is not None: + for notice in self.capacity_notices: + wait_notify(notice) yield True def exec_started(self, user_id): @@ -174,6 +178,27 @@ class TestShellExec(unittest.TestCase): self.assertEqual(pool.ensure_calls, [executor.user_id]) self.assertEqual(pool.mark_active_calls, [executor.user_id]) + def test_shell_forwards_capacity_queue_events(self): + executor, pool, _ = make_executor() + pool.capacity_notices = [ + {"state": "waiting", "reason": "per_user_limit"}, + {"state": "admitted"}, + ] + events = [] + ctx = make_ctx(executor) + ctx.event_emit = events.append + proc = MagicMock() + proc.communicate.return_value = ("ok\n", "") + proc.returncode = 0 + + with patch("core.executor_docker.subprocess.Popen", return_value=proc): + executor.call_tool("shell", {"command": "echo ok"}, ctx) + + self.assertEqual(events, [ + {"type": "execution_queue", "state": "waiting", "reason": "per_user_limit"}, + {"type": "execution_queue", "state": "admitted"}, + ]) + def test_shell_bad_args(self): executor, _, _ = make_executor() ctx = make_ctx(executor) diff --git a/tests/test_sandbox_capacity_packages.py b/tests/test_sandbox_capacity_packages.py index cc65fec..f0f5320 100644 --- a/tests/test_sandbox_capacity_packages.py +++ b/tests/test_sandbox_capacity_packages.py @@ -34,7 +34,7 @@ class CapacityTests(unittest.TestCase): with tempfile.TemporaryDirectory() as td: cap = ExecCapacity(Path(td), {"max_active_execs": 99, "max_background_execs": 99, "max_active_execs_per_user": 99}) - self.assertEqual((cap.max_active, cap.max_background, cap.max_per_user), (6, 4, 2)) + self.assertEqual((cap.max_active, cap.max_background, cap.max_per_user), (6, 4, 3)) def test_global_per_user_and_background_limits(self): with tempfile.TemporaryDirectory() as td, patch("core.sandbox.capacity.mem_available_bytes", return_value=10**12): @@ -60,6 +60,36 @@ class CapacityTests(unittest.TestCase): self.assertEqual(two.snapshot()["foreground_queued"], 0) one.release(lease) + def test_foreground_queue_reports_limit_and_admission(self): + with tempfile.TemporaryDirectory() as td, patch("core.sandbox.capacity.mem_available_bytes", return_value=10**12): + cap = ExecCapacity(Path(td), {"max_active_execs": 2, "max_active_execs_per_user": 1}) + held = cap.try_acquire("u1", "background", lease_id="held") + notices = [] + result = [] + t = threading.Thread( + target=lambda: result.append(cap.acquire_foreground("u1", wait_notify=notices.append)) + ) + t.start(); time.sleep(.15) + self.assertEqual(notices, [{ + "state": "waiting", "reason": "per_user_limit", + "user_running": 1, "user_limit": 1, + "global_running": 1, "global_limit": 2, + }]) + cap.release(held); t.join(2) + self.assertEqual(len(result), 1) + self.assertIsNotNone(result[0]) + self.assertEqual(notices[-1], {"state": "admitted"}) + cap.release(result[0]) + + def test_immediate_foreground_admission_does_not_report_queue(self): + with tempfile.TemporaryDirectory() as td, patch("core.sandbox.capacity.mem_available_bytes", return_value=10**12): + cap = ExecCapacity(Path(td), {"max_active_execs": 1}) + notices = [] + lease = cap.acquire_foreground("u1", wait_notify=notices.append) + self.assertIsNotNone(lease) + self.assertEqual(notices, []) + cap.release(lease) + def test_memory_pressure_pauses_new_admission(self): with tempfile.TemporaryDirectory() as td, patch("core.sandbox.capacity.mem_available_bytes", return_value=100): cap = ExecCapacity(Path(td), {"min_mem_available": "1g"}) diff --git a/web/static/js/chat.js b/web/static/js/chat.js index 6120f05..bacc762 100644 --- a/web/static/js/chat.js +++ b/web/static/js/chat.js @@ -3483,6 +3483,27 @@ function handleSseEvent(ev, asstCard, ctx) { if (!ctx.runId && snapshot.run_id) ctx.runId = snapshot.run_id; ctx.progressSteps = cloneProgressSteps(snapshot.steps); setTaskProgress(ctx.taskId, ctx.progressSteps); + } else if (t === "execution_queue") { + const det = ctx.runningToolEl; + if (!det) return; + const labelEl = det.querySelector(".tool-label"); + if (!labelEl) return; + const queue = ev.data || {}; + if (queue.state === "waiting") { + det.classList.add("waiting"); + if (queue.reason === "per_user_limit") { + labelEl.textContent = `等待执行容量（当前用户 ${Number(queue.user_running) || 0}/${Number(queue.user_limit) || 0}）`; + } else if (queue.reason === "global_limit") { + labelEl.textContent = `等待执行容量（整机 ${Number(queue.global_running) || 0}/${Number(queue.global_limit) || 0}）`; + } else if (queue.reason === "memory_pressure") { + labelEl.textContent = "等待执行容量（宿主内存压力）"; + } else { + labelEl.textContent = "等待执行容量（前方任务优先）"; + } + } else if (queue.state === "admitted") { + det.classList.remove("waiting"); + labelEl.textContent = det.dataset.runningLabel || labelEl.textContent; + } } else if (t === "tool_call") { const fn = (ev.data && ev.data.name) || "?"; const args = (ev.data && ev.data.args) || ""; @@ -3506,6 +3527,7 @@ function handleSseEvent(ev, asstCard, ctx) { const label = toolActivityLabel(fn, args); const det = document.createElement("details"); det.className = "tool-call activity running"; + det.dataset.runningLabel = label; det.innerHTML = `${escapeHtml(label)}
${escapeHtml(argsStr)}
`; asstCard.appendChild(det); // 运行态:spinner + 跳秒,tool_result 到达时定格(区分"在执行"和"卡死")