fix(canary): team-CS 分流改为有状态均衡随机，保证 staging 总量份额 / stateful balanced shunting guarantees staging share

src/xskill/canary.py

@@ -24,6 +24,7 @@
 import hashlib
 import json
 import logging
+import random
 import shutil
 import threading
 from dataclasses import dataclass
@@ -294,6 +295,105 @@ def pick_side_scoped(traj_id: str, skill_name: str, probability: float,
     return pick_side(traj_id, skill_name, probability)
 
 
+# ═══════════════════════════════════════════════════════════════════
+# 有状态分流：CanaryRouter —— 保证总量的均衡随机分配
+# ═══════════════════════════════════════════════════════════════════
+# pick_side 是无状态哈希：每次调用独立做一次伯努利试验。当分流 key 的
+# 基数很小（典型：team-CS 只有 3 个 worker），3 次 0.5 概率的独立试验完全
+# 可能全落到 main，staging 流量为 0，灰度永远拿不到样本。
+#
+# CanaryRouter 解决这个：按 (skill) 维护"已分配的 client→side"账本，新
+# client 进来时选让"当前 main/staging 比例最接近 probability"的那一侧（平
+# 局随机），从而把 staging 的份额"锁"进总量，而不是听天由命。同一 client
+# 在同一个 staging 版本内 side 钉死（轨迹一致性）；staging sha 或
+# probability 变了则重置账本、重新均衡。
+#
+# 只用在 team-CS manifest 路径（client 基数小）。高基数路径（traj_id /
+# window_id）继续用 pick_side——大数定律下无状态哈希天然均衡。
+
+
+class CanaryRouter:
+    """有状态灰度分流器：按 client 随机分配并保证 staging 总量份额。
+
+    与无状态 :func:`pick_side` 的区别：pick_side 每次调用独立哈希，client
+    很少时 staging 可能被饿死到 0。本路由器按 skill 记账，新 client 落到
+    "让运行比例最贴近 probability"的那一侧，staging 的份额被写进总量、不会
+    因为运气差而缺席。
+
+    - 同一 (client, skill, staging_sha, probability) → 同一 side（钉死，保证
+      轨迹一致性）。
+    - staging sha 变化（新候选 / 重建）或 probability 变化 → 该 skill 账本
+      清空、所有 client 重新均衡。
+    - 线程安全（team server 并发 sync）。
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        # skill_name -> {"staging_sha", "probability", "sides": {client_id: side}}
+        self._skills: dict[str, dict] = {}
+
+    def assign(self, *, client_id: str, skill_name: str,
+               probability: float, staging_sha: str) -> str:
+        """返回 ``client_id`` 对 ``skill_name`` 应走的 side。
+
+        首次见到该 (skill, staging_sha, probability) 组合时按均衡随机落点并
+        记账；之后同 client 直接回查。staging_sha / probability 变了则重置
+        该 skill 的账本后重新分配。
+        """
+        with self._lock:
+            st = self._skills.get(skill_name)
+            if (st is None
+                    or st["staging_sha"] != staging_sha
+                    or st["probability"] != probability):
+                st = {
+                    "staging_sha": staging_sha,
+                    "probability": probability,
+                    "sides": {},
+                }
+                self._skills[skill_name] = st
+            sides = st["sides"]
+            cached = sides.get(client_id)
+            if cached is not None:
+                return cached
+            n_main = sum(1 for v in sides.values() if v == "main")
+            n_staging = sum(1 for v in sides.values() if v == "staging")
+            side = self._balanced_side(n_main, n_staging, probability)
+            sides[client_id] = side
+            return side
+
+    @staticmethod
+    def _balanced_side(n_main: int, n_staging: int,
+                       probability: float) -> str:
+        """选让"加入本 client 后 main/staging 比例最接近 probability"的 side。
+
+        - probability≤0 → main；≥1 → staging（与 pick_side 边界一致）。
+        - 首个 client（无历史）→ 纯随机按 probability 取，给后续均衡留种子。
+        - 否则比较"加 staging"与"加 main"后的比例误差，取误差小的；误差相等
+          （典型 probability=0.5 的奇数位）随机破平局。
+        """
+        if probability <= 0:
+            return "main"
+        if probability >= 1:
+            return "staging"
+        total = n_main + n_staging
+        if total == 0:
+            return "staging" if random.random() < probability else "main"
+        ratio_if_staging = (n_staging + 1) / (total + 1)
+        ratio_if_main = n_staging / (total + 1)
+        err_staging = abs(ratio_if_staging - probability)
+        err_main = abs(ratio_if_main - probability)
+        if err_staging < err_main:
+            return "staging"
+        if err_main < err_staging:
+            return "main"
+        return "staging" if random.random() < 0.5 else "main"
+
+    def reset(self) -> None:
+        """清空全部账本（测试 / 生命周期重置用）。"""
+        with self._lock:
+            self._skills.clear()
+
+
 def read_skill_on_branch(skill_dir: Path, branch: str) -> str | None:
     """读取指定分支上的 SKILL.md 文本。不切分支，用 git show。"""
     code, out, _ = run_git(["show", f"{branch}:SKILL.md"], cwd=str(skill_dir))

src/xskill/team/server/client_registry.py

@@ -4,7 +4,7 @@
 ux_score 明细。这个文件是第一样。
 
 client_id 是 server 生成的 uuid——它同时是 ① canary 分桶 key（喂
-pick_side）② 上传轨迹的落盘分桶（clients/<client_id>/sessions/）③
+CanaryRouter.assign）② 上传轨迹的落盘分桶（clients/<client_id>/sessions/）③
 手改分支命名（user-staging/<client_id>）。
 """
 from __future__ import annotations

src/xskill/team/server/skill_manifest.py

@@ -1,30 +1,38 @@
 """skill_manifest.py — 给一个 client 现算它该持有的 ≤100 个 skill slot（SP1）
 
-server 端**不存"账本表"**。manifest = ``pick_side`` 纯函数 + skill git
-状态（has_staging / main_sha / staging_sha）的实时投影，每次 sync 现算。
+server 端的 skill 槽位投影不存表：ranked/recommended 排序 + skill git 状态
+（has_staging / main_sha / staging_sha）每次 sync 现算。**唯一例外是灰度 side
+决策**——它由 ``CanaryRouter`` 有状态记账（见下），因为无状态哈希在 client 基数
+很小时会把 staging 饿死到 0。
 
 slot 结构 = 80 ranked + 20 recommended：
 - ranked      —— 按 ux_score（main 侧近 30 天均分）滑窗取高分。
 - recommended —— SP3 = 用户画像质心推荐位：基于该 client 用过的 skill 的质心，
-                 从候选里取 cosine 最近邻（``profile_reco.py``）。无画像
-                 （冷启动）或非 team server 调用 → 退回 ux 排序往下取。
+                  从候选里取 cosine 最近邻（``profile_reco.py``）。无画像
+                  （冷启动）或非 team server 调用 → 退回 ux 排序往下取。
 
-灰度归因：某 skill 有 staging 分支 → side = pick_side(client_id, name, p)，
-确定性伪随机，同 client 同 skill 在整轮灰度内 side 钉死。无 staging → main。
+灰度归因：某 skill 有 staging 分支 → side = CanaryRouter.assign(client_id,
+name, p)，按 client 随机分配并保证 staging 总量份额（同 client 同 staging
+版本内 side 钉死）。无 staging → main。
 """
 from __future__ import annotations
 
 import logging
 import time
 from pathlib import Path
 
-from xskill.canary import has_staging, main_sha, pick_side, staging_sha
+from xskill.canary import CanaryRouter, has_staging, main_sha, staging_sha
 from xskill.skill.skill import Skill
 from xskill.skill.repo import SkillRepo
 from xskill.team.shared.protocol import SkillSlot, SyncResponse
 
 _logger = logging.getLogger("xskill.team.manifest")
 
+# team-CS manifest 路径的有状态灰度分流器：按 client 随机分配并保证 staging
+# 总量份额（见 CanaryRouter）。module 级单例——team server 单进程，跨 sync 持续
+# 记账；server 重启后账本清空、client 在下次 sync 重新均衡（可接受）。
+_ROUTER = CanaryRouter()
+
 
 def _rank_key(skill: Skill) -> tuple[float, int]:
     """排序键：(main 侧近 30 天 ux 均分, use_count)，都缺则 (0.0, 0)。"""
@@ -35,8 +43,12 @@ def _rank_key(skill: Skill) -> tuple[float, int]:
 def _resolve_slot(skill: Skill, client_id: str, probability: float, bucket: str) -> SkillSlot:
     """对一个 skill 现算它对该 client 的 side + sha。"""
     if has_staging(skill.path):
-        side = pick_side(client_id, skill.name, probability)
-        sha = staging_sha(skill.path) if side == "staging" else main_sha(skill.path)
+        s_sha = staging_sha(skill.path)
+        side = _ROUTER.assign(
+            client_id=client_id, skill_name=skill.name,
+            probability=probability, staging_sha=s_sha,
+        )
+        sha = s_sha if side == "staging" else main_sha(skill.path)
     else:
         side = "main"
         sha = main_sha(skill.path)

tests/test_canary.py

@@ -331,3 +331,84 @@ def test_jam_threshold_default_is_50():
 
 def test_jam_threshold_read_from_dict():
     assert canary.CanaryConfig.from_dict({"jam_threshold": 30}).jam_threshold == 30
+
+
+# ──────────────────────────────────────────────────────
+# CanaryRouter —— 有状态均衡随机分流
+# ──────────────────────────────────────────────────────
+
+def test_canary_router_locks_client_side():
+    """同 client 同 (staging_sha, probability) → 永远同一 side。"""
+    r = canary.CanaryRouter()
+    sides = {r.assign(client_id="c1", skill_name="s", probability=0.5,
+                      staging_sha="sha-X") for _ in range(10)}
+    assert len(sides) == 1            # 10 次调用 side 钉死
+
+
+def test_canary_router_probability_zero_all_main():
+    r = canary.CanaryRouter()
+    for i in range(10):
+        assert r.assign(client_id=f"c{i}", skill_name="s", probability=0.0,
+                        staging_sha="sha") == "main"
+
+
+def test_canary_router_probability_one_all_staging():
+    r = canary.CanaryRouter()
+    for i in range(10):
+        assert r.assign(client_id=f"c{i}", skill_name="s", probability=1.0,
+                        staging_sha="sha") == "staging"
+
+
+def test_canary_router_guarantees_staging_share_p05():
+    """p=0.5 + 3 client：staging 必拿到 1~2 个，永远不会被饿死到 0。
+
+    这是修复的核心：无状态 pick_side 在 3 client 下有 1/8 概率全 main。
+    """
+    for _ in range(200):              # 200 次独立试验（首个 client 随机）
+        r = canary.CanaryRouter()
+        sides = [r.assign(client_id=f"c{i}", skill_name="s", probability=0.5,
+                          staging_sha="sha") for i in range(3)]
+        n_staging = sides.count("staging")
+        assert 1 <= n_staging <= 2    # 份额被锁进总量，不会是 0 或 3
+
+
+def test_canary_router_guarantees_staging_share_p02():
+    """p=0.2 + 5 client：恰好 1 个 staging（≈ 20% 份额，确定性收敛）。"""
+    for _ in range(200):
+        r = canary.CanaryRouter()
+        sides = [r.assign(client_id=f"c{i}", skill_name="s", probability=0.2,
+                          staging_sha="sha") for i in range(5)]
+        assert sides.count("staging") == 1
+
+
+def test_canary_router_ratio_tracks_probability_many_clients():
+    """100 client 下运行比例贴近 probability（误差 ≤ 1 个）。"""
+    import random as _r
+    for p in (0.2, 0.3, 0.5, 0.7):
+        _r.seed(0)
+        r = canary.CanaryRouter()
+        sides = [r.assign(client_id=f"c{i}", skill_name="s", probability=p,
+                          staging_sha="sha") for i in range(100)]
+        ratio = sides.count("staging") / 100
+        assert abs(ratio - p) <= 0.02          # 误差 ≤ 2 个 client
+
+
+def test_canary_router_resets_on_staging_sha_change():
+    """staging sha 变了 → 账本重置，client 可被重新分配到对面。"""
+    r = canary.CanaryRouter()
+    first = r.assign(client_id="c1", skill_name="s", probability=1.0,
+                     staging_sha="sha-OLD")
+    assert first == "staging"
+    # 新 staging 版本 → 重置 → c1 重新分配；p=0 强制 main 验证确实重算了
+    second = r.assign(client_id="c1", skill_name="s", probability=0.0,
+                      staging_sha="sha-NEW")
+    assert second == "main"
+
+
+def test_canary_router_resets_on_probability_change():
+    """probability 变了 → 重算；p=1.0 即使该 client 之前被分到 main 也变 staging。"""
+    r = canary.CanaryRouter()
+    r.assign(client_id="c1", skill_name="s", probability=0.0, staging_sha="sha")
+    forced = r.assign(client_id="c1", skill_name="s", probability=1.0,
+                      staging_sha="sha")
+    assert forced == "staging"

tests/test_team_skill_manifest.py

@@ -8,7 +8,7 @@
 
 from xskill.pipeline.atom import AtomTask, AtomTaskStore
 from xskill.team.server.profile_reco import ClientProfileRecommender
-from xskill.team.server.skill_manifest import build_manifest
+from xskill.team.server.skill_manifest import build_manifest, _ROUTER
 
 
 def _git(args, cwd):
@@ -110,6 +110,26 @@ def test_manifest_staging_side_is_deterministic_per_client(tmp_path):
     assert forced.side == "staging"
 
 
+def test_manifest_guarantees_staging_traffic_with_few_clients(tmp_path):
+    """3 个 worker + 唯一 staging skill：至少 1 个 worker 拿到 staging。
+
+    这是 board6 灰度饿死 bug 的回归测试——旧的无状态 pick_side 会把 3 个
+    client 全哈希到 main，staging 流量为 0。CanaryRouter 把 staging 份额锁进
+    总量，保证 ≥1。
+    """
+    _ROUTER.reset()
+    skill_dir = tmp_path / "skill"
+    skill_dir.mkdir()
+    _make_skill(skill_dir, "only-skill", with_staging=True)
+    sides = [
+        build_manifest(client_id=f"worker-{i}", skill_dir=skill_dir,
+                       probability=0.5, ranked_slots=80, total_slots=100).slots[0].side
+        for i in range(3)
+    ]
+    assert sides.count("staging") >= 1
+    _ROUTER.reset()
+
+
 # ═══════════════════════════════════════════════════════════════════
 # SP3 — 画像推荐（profile_reco）
 # ═══════════════════════════════════════════════════════════════════