每用户并发 Agent 上限：MAX_AGENTS_PER_USER 默认 10

需求：限制每个用户在 swarm 中并发的 Agent 数为 10。按「并发 WS 连接数/用户」口径实现，env 可调（默认 10）。

orchestrator/main.py：ConnectionManager 新增 per-user 记账（agent_user / user_agents + user_agent_count / can_bind_user / bind_user / unbind）；max_agents_per_user() 读 MAX_AGENTS_PER_USER（默认 10）。WS register 携带 user_id 且该用户已达上限时回 registration_rejected(reason,limit) 并 close(1008)，不注册；同 agent_id 重连放行；断开 unbind 释放名额。未带 user_id 的 Agent 为 unbound、不受限。

agent/main.py：新增 user_id 构造参数 + HEICODE_USER_ID 环境回退，并在 register 载荷中带上（仅在设置时）。

测试 scripts/test-max-agents-per-user.py：单元 + WS 集成（MAX_AGENTS_PER_USER=3）。接入 CI。docs/integration/security-boundary.md §6 记录该配额。

影响范围：仅 agent_swarm；不改 Manager↔Swarm 契约、计费、审批链、密钥处理。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.github/workflows/ci.yml

@@ -55,6 +55,10 @@ jobs:
         env: { REDIS_FAKE: "1" }
         run: python scripts/test-workflow-e2e.py
 
+      - name: Per-user agent cap (MAX_AGENTS_PER_USER)
+        env: { REDIS_FAKE: "1" }
+        run: python scripts/test-max-agents-per-user.py
+
       - name: Manager event-contract test
         env: { REDIS_FAKE: "1" }
         run: python scripts/test-contract-events.py

agent/main.py

@@ -84,11 +84,15 @@ class Agent:
         capabilities: Optional[list[str]] = None,
         workspace_dir: str = "/workspace",
         git_repo_url: Optional[str] = None,
+        user_id: Optional[str] = None,
     ):
         # Initialize connection state, execution limits, and workspace helpers for this agent runtime.
         self.agent_id = agent_id or f"agent-{uuid.uuid4().hex[:8]}"
         self.orchestrator_url = orchestrator_url.rstrip("/")
         self.capabilities = capabilities or ["general"]
+        # The owning user; the orchestrator caps concurrent agents per user (MAX_AGENTS_PER_USER).
+        # The agent platform sets this; falls back to env. None => unbound (not subject to the cap).
+        self.user_id = user_id or os.getenv("HEICODE_USER_ID")
         self.workspace_dir = Path(workspace_dir)
         self.git_repo_url = git_repo_url
 
@@ -147,14 +151,17 @@ class Agent:
     async def register(self):
         # Announce this agent and its current capacity to the orchestrator after connecting.
         try:
-            await self.safe_send({
+            register_msg = {
                 "type": "register",
                 "agent_id": self.agent_id,
                 "capabilities": self.capabilities,
                 # Backward-compatible extra fields; older orchestrators ignore them.
                 "available_slots": self.available_slots(),
                 "active_task_ids": list(self.active_tasks.keys()),
-            })
+            }
+            if self.user_id:
+                register_msg["user_id"] = self.user_id  # subjects this agent to the per-user cap
+            await self.safe_send(register_msg)
             logger.info(f"Registered agent {self.agent_id} with capabilities: {self.capabilities}")
             return True
         except Exception as e:

docs/integration/security-boundary.md

@@ -54,6 +54,7 @@
 
 - **Swarm 模型**：Agent 主动出站连编排器 WebSocket（`/ws/{agent_id}`），**不**对公网暴露每 Agent 子域名。AM 单 Agent 模型里的「客户端↔agent 直连 + `AGENT_ACCESS_TOKEN` 本地校验」**不适用于** swarm（无直连回路）。
 - 服务间鉴权：HM→Swarm 用 `AGENT_RUNTIME_SERVICE_TOKEN`（Bearer）；回调 HMAC 签名。
+- **每用户并发 Agent 配额**：一个 `user_id` 同时连接的 Agent 数上限为 `MAX_AGENTS_PER_USER`（env，默认 10）。注册（WS `register` 消息携带 `user_id`）超额即被拒绝（回 `registration_rejected` 并关闭，code 1008），断开后释放名额。归因主轴仍为 `user.id`/`channelId`。未带 `user_id` 的 Agent 为 unbound，不计入该配额。实现：`ConnectionManager.can_bind_user/bind_user/unbind` + 注册处强制；测试 `scripts/test-max-agents-per-user.py`。
 - 🟡 待接入：多租户运行时隔离（命名空间/网络/配额）由 Agent 平台（AKS Workload Identity）承载，非本仓编排器；归因主轴为 `user.id`/`channelId`（见 `usage-billing-schema.md`），不引入 tenant 概念。
 
 ## 7. 外部 API 与传输

orchestrator/main.py

@@ -67,6 +67,10 @@ class ConnectionManager:
 
     def __init__(self):
         self.active_connections: Dict[str, WebSocket] = {}
+        # Per-user concurrent-agent accounting (cap enforced at register; see max_agents_per_user).
+        # Agents that register without a user_id are "unbound" and not subject to the per-user cap.
+        self.agent_user: Dict[str, str] = {}                 # agent_id -> user_id (bound agents)
+        self.user_agents: Dict[str, set] = defaultdict(set)  # user_id  -> {agent_id}
 
     async def connect(self, agent_id: str, websocket: WebSocket):
         """Accept and store WebSocket connection."""
@@ -80,6 +84,36 @@ class ConnectionManager:
             del self.active_connections[agent_id]
             logger.info(f"Agent {agent_id} disconnected")
 
+    def user_agent_count(self, user_id: str) -> int:
+        """How many distinct agents this user currently has connected."""
+        return len(self.user_agents.get(user_id, ()))
+
+    def can_bind_user(self, agent_id: str, user_id: str, limit: int) -> bool:
+        """Whether `agent_id` may register under `user_id` without exceeding the per-user cap.
+
+        A reconnect by an already-counted agent_id is always allowed (it adds no new agent).
+        """
+        agents = self.user_agents.get(user_id, set())
+        if agent_id in agents:
+            return True
+        return len(agents) < limit
+
+    def bind_user(self, agent_id: str, user_id: str):
+        """Record that `agent_id` belongs to `user_id` (call only after a passing can_bind_user)."""
+        self.agent_user[agent_id] = user_id
+        self.user_agents[user_id].add(agent_id)
+
+    def unbind(self, agent_id: str):
+        """Drop an agent's user binding on disconnect (no-op for unbound agents)."""
+        user_id = self.agent_user.pop(agent_id, None)
+        if user_id is None:
+            return
+        agents = self.user_agents.get(user_id)
+        if agents is not None:
+            agents.discard(agent_id)
+            if not agents:
+                self.user_agents.pop(user_id, None)
+
     async def send_message(self, agent_id: str, message: dict) -> bool:
         """Send message to specific agent. Returns True if it was delivered."""
         if agent_id in self.active_connections:
@@ -109,6 +143,14 @@ manager = ConnectionManager()
 AGENT_SLOTS: Dict[str, int] = {}
 
 
+def max_agents_per_user() -> int:
+    """Max concurrent agents one user may have connected (per-user swarm cap). Env-tunable; default 10."""
+    try:
+        return max(1, int(os.getenv("MAX_AGENTS_PER_USER", "10") or 10))
+    except ValueError:
+        return 10
+
+
 def record_agent_slots(agent_id: str, message: dict):
     """Update the cached free-slot count for an agent from a protocol message."""
     slots = message.get("available_slots")
@@ -2204,7 +2246,30 @@ async def websocket_endpoint(websocket: WebSocket, agent_id: str):
 
         if data.get("type") == "register":
             capabilities = data.get("capabilities", [])
+            user_id = data.get("user_id")
+
+            # Per-user swarm cap: a user may have at most MAX_AGENTS_PER_USER (default 10) agents
+            # connected at once. Agents that omit user_id are unbound and not capped. A reconnect by
+            # an already-counted agent_id is allowed (can_bind_user handles it).
+            if user_id:
+                limit = max_agents_per_user()
+                if not manager.can_bind_user(agent_id, user_id, limit):
+                    logger.warning(
+                        "Rejecting agent %s: user %s already at max agents (%d connected)",
+                        agent_id, user_id, manager.user_agent_count(user_id),
+                    )
+                    await websocket.send_json({
+                        "type": "registration_rejected",
+                        "agent_id": agent_id,
+                        "reason": "max_agents_per_user_exceeded",
+                        "limit": limit,
+                    })
+                    await websocket.close(code=1008, reason="Max agents per user exceeded")
+                    return
+
             await agent_registry.register_agent(agent_id, capabilities)
+            if user_id:
+                manager.bind_user(agent_id, user_id)
             record_agent_slots(agent_id, data)
 
             await websocket.send_json({
@@ -2746,6 +2811,7 @@ async def websocket_endpoint(websocket: WebSocket, agent_id: str):
     finally:
         # Cleanup
         manager.disconnect(agent_id)
+        manager.unbind(agent_id)  # free the user's per-user agent slot
         AGENT_SLOTS.pop(agent_id, None)
         await agent_registry.deregister_agent(agent_id)
 

scripts/test-max-agents-per-user.py

@@ -0,0 +1,158 @@
+"""Per-user agent cap tests: at most MAX_AGENTS_PER_USER concurrent agents per user.
+
+Covers:
+  * unit — ConnectionManager per-user accounting (can_bind_user / bind_user / unbind / count) and
+    the env-tunable limit (default 10);
+  * integration — the REAL orchestrator over WebSocket: registrations beyond the cap are rejected
+    (registration_rejected + socket close), a different user is unaffected, and freeing a slot
+    (disconnect) lets a new agent in. Unbound agents (no user_id) are never capped.
+
+Uses MAX_AGENTS_PER_USER=3 so the integration test stays small/fast.
+
+Run from agent_swarm_v6 (install deps first):
+    pip install -r orchestrator/requirements.txt -r agent/requirements.txt
+    REDIS_FAKE=1 python scripts/test-max-agents-per-user.py
+"""
+import asyncio
+import json
+import logging
+import os
+import sys
+import threading
+from pathlib import Path
+
+os.environ["REDIS_FAKE"] = "1"
+os.environ["MAX_AGENTS_PER_USER"] = "3"
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+import httpx
+import uvicorn
+import websockets
+from orchestrator import main as orch
+
+PORT = 8137
+BASE = f"http://127.0.0.1:{PORT}"
+WS = f"ws://127.0.0.1:{PORT}"
+
+logging.getLogger("orchestrator.swarm_runtime").setLevel(logging.ERROR)
+
+failures = []
+
+
+def check(name, cond):
+    print(("PASS" if cond else "FAIL"), "-", name)
+    if not cond:
+        failures.append(name)
+
+
+def test_unit():
+    cm = orch.ConnectionManager()
+    limit = 3
+    for i in range(limit):
+        check(f"unit: agent a{i} allowed under cap", cm.can_bind_user(f"a{i}", "U", limit))
+        cm.bind_user(f"a{i}", "U")
+    check("unit: user count == 3", cm.user_agent_count("U") == 3)
+    check("unit: 4th distinct agent rejected at cap", not cm.can_bind_user("a3", "U", limit))
+    check("unit: reconnect by an already-counted agent allowed at cap", cm.can_bind_user("a0", "U", limit))
+    check("unit: a different user is independent", cm.can_bind_user("b0", "V", limit))
+    cm.unbind("a0")
+    check("unit: count drops after unbind", cm.user_agent_count("U") == 2)
+    check("unit: freed slot lets a new agent in", cm.can_bind_user("a3", "U", limit))
+    cm.unbind("nonexistent")  # no-op, must not raise
+    check("unit: env limit default respected (=3 here)", orch.max_agents_per_user() == 3)
+
+
+async def _register(agent_id, user_id=None):
+    """Open a WS, send register (optionally with user_id), return (ws, response_dict)."""
+    ws = await websockets.connect(f"{WS}/ws/{agent_id}")
+    msg = {"type": "register", "agent_id": agent_id, "capabilities": ["general"]}
+    if user_id is not None:
+        msg["user_id"] = user_id
+    await ws.send(json.dumps(msg))
+    resp = json.loads(await ws.recv())
+    return ws, resp
+
+
+async def wait_health(client):
+    for _ in range(150):
+        try:
+            if (await client.get(f"{BASE}/health")).status_code == 200:
+                return True
+        except Exception:
+            pass
+        await asyncio.sleep(0.1)
+    return False
+
+
+async def test_integration():
+    config = uvicorn.Config(orch.app, host="127.0.0.1", port=PORT, log_level="warning")
+    server = uvicorn.Server(config)
+    server.install_signal_handlers = lambda: None
+    thread = threading.Thread(target=server.run, daemon=True)
+    thread.start()
+    open_ws = []
+    try:
+        async with httpx.AsyncClient(timeout=10) as client:
+            if not await wait_health(client):
+                check("orchestrator started", False)
+                return
+
+            # 3 agents for one user all register OK
+            for i in range(3):
+                ws, resp = await _register(f"cap-{i}", user_id="user-cap")
+                check(f"integration: agent cap-{i} registered", resp.get("type") == "registered")
+                open_ws.append(ws)
+
+            # 4th for the same user is rejected
+            ws4, resp4 = await _register("cap-3", user_id="user-cap")
+            check("integration: 4th agent rejected (over cap)", resp4.get("type") == "registration_rejected")
+            check("integration: rejection states reason + limit",
+                  resp4.get("reason") == "max_agents_per_user_exceeded" and resp4.get("limit") == 3)
+            await ws4.close()
+
+            # a different user is unaffected
+            wsB, respB = await _register("other-0", user_id="user-other")
+            check("integration: different user unaffected", respB.get("type") == "registered")
+            open_ws.append(wsB)
+
+            # freeing a slot (disconnect) lets a new agent for the capped user in
+            await open_ws[0].close()
+            open_ws.pop(0)
+            await asyncio.sleep(0.4)  # let the server's finally-block run unbind
+            ws5, resp5 = await _register("cap-4", user_id="user-cap")
+            check("integration: freed slot admits a new agent", resp5.get("type") == "registered")
+            open_ws.append(ws5)
+
+            # unbound agents (no user_id) are never capped — many can register
+            for i in range(5):
+                ws, resp = await _register(f"unbound-{i}")  # no user_id
+                check(f"integration: unbound agent unbound-{i} not capped", resp.get("type") == "registered")
+                open_ws.append(ws)
+    finally:
+        for ws in open_ws:
+            try:
+                await ws.close()
+            except Exception:
+                pass
+        server.should_exit = True
+        thread.join(timeout=5)
+
+
+async def main():
+    test_unit()
+    await test_integration()
+    print()
+    if failures:
+        print(f"{len(failures)} per-user-cap check(s) FAILED: {failures}")
+        return 1
+    print("all per-user agent-cap checks passed")
+    return 0
+
+
+if __name__ == "__main__":
+    _code = asyncio.run(main())
+    sys.stdout.flush()
+    sys.stderr.flush()
+    os._exit(_code or 0)