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(argsStr)}`;
asstCard.appendChild(det);
// 运行态:spinner + 跳秒,tool_result 到达时定格(区分"在执行"和"卡死")