# Chat 两路会话并发 改造方案

- 站点：`https://xybcloud.online/chat/`
- 落地：`/var/www/cutebot/cutebot.py` + `/var/www/html/chat/index.php`
- 备份：`/var/www/html/data/back/cutebot-bak-20260930091239.py`
- 时间：2026-09-30

## 1. 现状（实测）

| 项 | 值 |
|---|---|
| 后端进程 | `python3 /var/www/cutebot/cutebot.py`（pid 1409281），systemd `cutebot.service` |
| Web 框架 | aiohttp 单进程事件循环，监听 `127.0.0.1:8765` |
| Apache 反代 | `ProxyPass /api/ → http://127.0.0.1:8765/api/`（`000-default-ssl.conf:33-34`） |
| LLM 后端 | Anthropic 原生 `/v1/messages`（`x-api-key` + `anthropic-version: 2023-06-01`），可配 `ANTHROPIC_BASE_URL` 走反代 |
| LLM 调用 | **(a)** Claude Code CLI 子进程 `claude -p --output-format stream-json --resume <sid>`；(b) aiohttp 直连 `/v1/messages` |
| 模型组配置 | 从 `~/.bashrc` 解析 `export MODEL_VENDOR/ANTHROPIC_API_KEY/ANTHROPIC_BASE_URL/ANTHROPIC_MODEL` |

## 2. 根因

`SessionLockManager`（cutebot.py:294）本身就是按 `session_id` 加锁的设计 —— 但 `handle_send`（cutebot.py:1191）把 `auth['session_key']`（**用户级**）当作 `session_id` 传进去了。所以：

- 同一用户（同一登录 token）不管开几个标签、不管 session_id 是否不同 → **共用一把 asyncio.Lock**，全串行
- 前端 `cancel`（index.php:1859, 2008）只带 Authorization → `handle_cancel`（cutebot.py:1646）按 `task:{session_key}` 查 → **停 A 会把 B 也停掉**

aiohttp 事件循环本身完全支持并发（`create_subprocess_exec` 和 `aiohttp.ClientSession` 都是 awaitable），瓶颈只在锁粒度。

## 3. 改造方案

### 3.1 目标
- 同一登录用户最多同时 **2 路** 会话生成，不同 `session_id` 互不阻塞
- 单路"停止生成"只停本路
- 超 2 路时新请求进入排队（前端收到 `type:queue` 事件而非 429）
- 配置化：`CHAT_MAX_CONCURRENT_PER_USER` 环境变量，默认 2

### 3.2 cutebot.py 改动（~50 行）

**位置 A — 配置（line 65 附近）**
```python
USER_CONCURRENCY_LIMIT = int(os.environ.get('CHAT_MAX_CONCURRENT_PER_USER', '2'))
```

**位置 B — 全局信号量表（line 1029 后）**
```python
_user_semaphores: dict[str, asyncio.Semaphore] = {}  # user_key → Semaphore(N)

def _get_user_sem(user_key: str) -> asyncio.Semaphore:
    sem = _user_semaphores.get(user_key)
    if sem is None:
        sem = asyncio.Semaphore(USER_CONCURRENCY_LIMIT)
        _user_semaphores[user_key] = sem
    return sem
```

**位置 C — `handle_send`（line 1173-1381）关键改造**

在原 `async with lock:`（line 1263）外层套 user semaphore，并把锁/busy/queue/task_key 的 key 从 `session_key` 换成真正的 `session_id`：

```python
# 早解析 session_id（覆盖新对话/无 session_id 场景）
internal_session_id = (
    client_session_id
    or SessionManager.get_session_id(session_key)
    or f"pending:{secrets.token_hex(8)}"
)

# user-level semaphore：硬限 2 路并发
user_sem = _get_user_sem(session_key)
acquired = False
try:
    if not user_sem.locked() or user_sem._value > 0:
        # 还有空位，直接抢
        acquired = True
        await user_sem.acquire()
    else:
        # 满了，先告诉前端要排队，再异步等
        try:
            await response.write(
                f"data: {json.dumps({
                    'type': 'queue',
                    'text': f'⏳ 并发已满（{USER_CONCURRENCY_LIMIT} 路），请稍候...',
                    'limit': USER_CONCURRENCY_LIMIT,
                }, ensure_ascii=False)}\n\n".encode('utf-8'))
        except Exception: pass
        await user_sem.acquire()
        acquired = True
        if not client_disconnected:
            try:
                await response.write(
                    f"data: {json.dumps({'type':'start','text':'▶️ 拿到并发名额，开始处理...'}, ensure_ascii=False)}\n\n".encode('utf-8'))
            except Exception: pass

    # session-level lock（保护同 session_id 历史顺序追加）
    lock = SessionLockManager.get_lock(internal_session_id)
    async with lock:
        SessionLockManager.set_busy(internal_session_id, True)
        task_key = f"task:session:{internal_session_id}"
        try:
            # ... 原有 handle_send 主体 ...
            # storage_key 也用 internal_session_id（与 line 1281 一致）
            storage_key = internal_session_id
        finally:
            SessionLockManager.set_busy(internal_session_id, False)
finally:
    if acquired:
        user_sem.release()
```

**位置 D — `handle_cancel`（line 1646）接受 session_id**

```python
async def handle_cancel(request):
    auth = _verify_web_token(request)
    if not auth: ...
    session_key = auth['session_key']

    # 优先取请求体里的 session_id；否则回退到"该用户当前活跃 session"
    try:
        data = await request.json()
    except Exception:
        data = {}
    target_sid = (data or {}).get('session_id') or SessionManager.get_session_id(session_key)

    if not target_sid:
        return aiohttp_web.json_response({'ok': True, 'message': '没有进行中的任务'}, headers=_cors_headers())

    task_key = f"task:session:{target_sid}"
    # ... 原有查找/杀进程逻辑 ...
```

**位置 E — `handle_status`（line 1910）返回 user-level + 各 session 状态**

```python
async def handle_status(request):
    auth = _verify_web_token(request)
    ...
    session_key = auth['session_key']
    sem = _user_semaphores.get(session_key)
    user_inflight = (USER_CONCURRENCY_LIMIT - sem._value) if sem else 0

    # 该用户所有非空 session 的 busy/queue
    sessions = {}
    for sid in SessionLockManager._busy:
        # 只暴露当前用户拥有的 sid（这里 _busy 全局共享，靠 sessions.json 反查归属）
        ...
    return aiohttp_web.json_response({
        'ok': True,
        'user_limit': USER_CONCURRENCY_LIMIT,
        'user_inflight': user_inflight,
        'sessions': sessions,
    }, headers=_cors_headers())
```

注：`_busy` 是全用户共享 dict，跨用户隔离用 `sessions.json` 维护 `session_key → [sid1, sid2, ...]`。本期前端只读 `user_inflight`，`sessions` 字段先返回空 dict，预留接口。

### 3.3 前端 index.php 改动（最小 2 行）

`stopBackendTask` 和 stop 调用（line 1858, 2008）：附 `session_id: currentConvSid || ''`：

```js
fetch(API_BASE + '/api/cancel', {
    method: 'POST',
    headers: {'Authorization': 'Bearer ' + authToken, 'Content-Type': 'application/json'},
    body: JSON.stringify({ session_id: currentConvSid || '' })
})
```

`currentConvSid` 已经是按标签隔离的全局变量（line 1288），多标签天然并行，不需要再改 streaming 状态。

## 4. 验证清单

- [ ] python3 -m py_compile /var/www/cutebot/cutebot.py 无语法错误
- [ ] 单标签正常生成（回归）
- [ ] 两个标签同 session_id 双开：消息按发送顺序串行追加，不会交错
- [ ] 两个标签不同 session_id：A 跑 60s，B 也跑 60s，两者**同时输出**（关键验收点）
- [ ] 第 3 个标签：收到 `type:queue` 事件；前两个任一结束后，第 3 路立即拿到 `type:start` 开始生成
- [ ] 标签 A 停止生成：A 立即停，B 不受影响
- [ ] `tail -f /var/www/cutebot/botpy.log | grep -E "ERROR|WARN"` 无新增 traceback
- [ ] 重启 service 后 `systemctl status cutebot` active

## 5. 风险与缓解

| 风险 | 缓解 |
|---|---|
| Claude Code CLI 子进程并发 N 个（每并发一个 node 进程） | 上限 2 可控；如 CPU 紧张，可改"全程走 HTTP 直连"砍掉 CLI |
| 同 session_id 双开仍按 session_lock 串行（这是设计意图，保护历史顺序） | 测试用例 2 验证 |
| session_id 解析时机（前端 cancel 时 tab 已关） | `handle_cancel` 找不到 task_key → 返回 `{ok:true, message:'没有进行中的任务'}` |
| `bot.log` 238MB 膨胀 | 本期不动；后续接 logrotate |

## 6. 部署流程

1. **备份**：`cutebot.py → /var/www/html/data/back/cutebot-bak-*.py` ✅ 已完成
2. **副本实验**：先在 `/var/www/cutebot2/cutebot.py` 上改，跑在 8766 端口
3. **本地验证**：py_compile + 单标签回归 + 双标签并行
4. **切流确认**（需用户确认）：把改动同步到 `/var/www/cutebot/cutebot.py` 并 `systemctl restart cutebot`
5. **线上验证**：通过 Apache 反代路径 `https://xybcloud.online/chat/` 跑 Puppeteer 压测