feat: task distribution + smart routing (Phase 3 of Agent Teams)

bin/ask

@@ -45,9 +45,12 @@ sys.path.insert(0, str(lib_dir))
 from compat import read_stdin_text, setup_windows_encoding
 setup_windows_encoding()
 
+from aliases import load_aliases, resolve_alias
 from cli_output import EXIT_ERROR, EXIT_OK
 from providers import parse_qualified_provider
 from session_utils import find_project_session_file
+from task_router import auto_route
+from team_config import load_team_config, resolve_team_agent
 
 
 # Provider to daemon command mapping
@@ -481,43 +484,48 @@ def make_task_id() -> str:
 
 
 def _usage() -> None:
+    aliases = load_aliases()
+    alias_list = ", ".join(f"{k}→{v}" for k, v in sorted(aliases.items()))
     print("Usage: ask <provider> [options] <message>", file=sys.stderr)
     print("", file=sys.stderr)
     print("Providers:", file=sys.stderr)
     print("  gemini, codex, opencode, droid, claude, copilot, codebuddy, qwen", file=sys.stderr)
     print("", file=sys.stderr)
+    print("Aliases:", file=sys.stderr)
+    print(f"  {alias_list}", file=sys.stderr)
+    print("", file=sys.stderr)
+    team = load_team_config(Path.cwd())
+    if team:
+        agents = ", ".join(f"{a.name}→{a.provider}" for a in team.agents)
+        print(f"Team '{team.name}' ({team.strategy}):", file=sys.stderr)
+        print(f"  {agents}", file=sys.stderr)
+        print("", file=sys.stderr)
     print("Options:", file=sys.stderr)
     print("  -h, --help              Show this help message", file=sys.stderr)
     print("  -t, --timeout SECONDS   Request timeout (default: 3600)", file=sys.stderr)
     print("  --notify                Sync send, no wait for reply (for notifications)", file=sys.stderr)
     print("  --foreground            Run in foreground (no nohup/background)", file=sys.stderr)
     print("  --background            Force background mode", file=sys.stderr)
     print("  --no-wrap               Don't wrap with CCB protocol markers", file=sys.stderr)
+    print("  --auto                  Auto-select provider based on message content", file=sys.stderr)
 
 
 def main(argv: list[str]) -> int:
     if len(argv) <= 1:
         _usage()
         return EXIT_ERROR
 
-    # First argument must be the provider
+    # First argument must be the provider (or --auto / --help)
     raw_provider = argv[1].lower()
 
     if raw_provider in ("-h", "--help"):
         _usage()
         return EXIT_OK
 
-    base_provider, instance = parse_qualified_provider(raw_provider)
-
-    if base_provider not in PROVIDER_DAEMONS:
-        print(f"[ERROR] Unknown provider: {base_provider}", file=sys.stderr)
-        print(f"[ERROR] Available: {', '.join(PROVIDER_DAEMONS.keys())}", file=sys.stderr)
-        return EXIT_ERROR
-
-    daemon_cmd = PROVIDER_DAEMONS[base_provider]
-    provider = raw_provider  # keep full qualified key for daemon routing
+    auto_mode = raw_provider == "--auto"
+    cwd = Path.cwd()
 
-    # Parse remaining arguments
+    # Parse remaining arguments (shared by both auto and normal modes)
     timeout: float = 3600.0
     notify_mode = False
     no_wrap = False
@@ -560,6 +568,34 @@ def main(argv: list[str]) -> int:
         print("[ERROR] Message cannot be empty", file=sys.stderr)
         return EXIT_ERROR
 
+    # --auto mode: select provider based on message content
+    if auto_mode:
+        team = load_team_config(cwd)
+        route = auto_route(message, team)
+        raw_provider = route.provider
+        print(f"[AUTO] → {raw_provider}" + (f" ({route.reason})" if route.reason else ""), file=sys.stderr)
+    else:
+        # Resolution order: team agents > aliases > direct provider names
+        team = load_team_config(cwd)
+        team_agent = resolve_team_agent(raw_provider, team)
+        if team_agent:
+            raw_provider = team_agent.provider
+        else:
+            aliases = load_aliases(cwd)
+            base_part, _, instance_part = raw_provider.partition(":")
+            base_part = resolve_alias(base_part, aliases)
+            raw_provider = f"{base_part}:{instance_part}" if instance_part else base_part
+
+    base_provider, instance = parse_qualified_provider(raw_provider)
+
+    if base_provider not in PROVIDER_DAEMONS:
+        print(f"[ERROR] Unknown provider: {base_provider}", file=sys.stderr)
+        print(f"[ERROR] Available: {', '.join(PROVIDER_DAEMONS.keys())}", file=sys.stderr)
+        return EXIT_ERROR
+
+    daemon_cmd = PROVIDER_DAEMONS[base_provider]
+    provider = raw_provider  # keep full qualified key for daemon routing
+
     # Notify mode: sync send, no wait for reply (used for hook notifications)
     if notify_mode:
         _require_caller()

lib/aliases.py

@@ -0,0 +1,62 @@
+"""Agent name aliases for CCB.
+
+Resolves short aliases (a, b, c, ...) to provider names.
+
+Configuration layers (higher overrides lower):
+1. Hardcoded defaults (DEFAULT_ALIASES)
+2. ~/.ccb/aliases.json (global)
+3. .ccb/aliases.json (project-level, relative to work_dir)
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import Dict, Optional
+
+DEFAULT_ALIASES: Dict[str, str] = {
+    "a": "codex",
+    "b": "gemini",
+    "c": "claude",
+    "d": "opencode",
+    "e": "droid",
+    "f": "copilot",
+    "g": "codebuddy",
+    "h": "qwen",
+}
+
+
+def _load_json(path: Path) -> Dict[str, str]:
+    """Load aliases from a JSON file, returning {} on any error."""
+    try:
+        if not path.is_file():
+            return {}
+        data = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(data, dict):
+            return {}
+        # Only keep str->str entries
+        return {str(k): str(v) for k, v in data.items()}
+    except (json.JSONDecodeError, OSError, ValueError):
+        print(f"[WARN] Failed to parse alias config: {path}", file=sys.stderr)
+        return {}
+
+
+def load_aliases(work_dir: Optional[Path] = None) -> Dict[str, str]:
+    """Merge alias configs: defaults < ~/.ccb/aliases.json < .ccb/aliases.json."""
+    merged = dict(DEFAULT_ALIASES)
+
+    global_path = Path.home() / ".ccb" / "aliases.json"
+    merged.update(_load_json(global_path))
+
+    if work_dir is not None:
+        project_path = work_dir / ".ccb" / "aliases.json"
+        merged.update(_load_json(project_path))
+
+    return merged
+
+
+def resolve_alias(name: str, aliases: Dict[str, str]) -> str:
+    """Resolve an alias to a provider name. Non-aliases pass through unchanged."""
+    key = (name or "").strip().lower()
+    return aliases.get(key, key)

lib/task_router.py

@@ -0,0 +1,210 @@
+"""Smart task routing for CCB Agent Teams.
+
+Routes tasks to the best provider based on message content analysis.
+Supports keyword matching (Chinese + English) and team skill-based matching.
+
+Used by `ask --auto "message"` to auto-select the best provider.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Sequence
+
+from team_config import TeamAgent, TeamConfig
+
+
+@dataclass
+class RouteResult:
+    """Result of routing a task to a provider."""
+    provider: str
+    model: str = ""
+    reason: str = ""
+    score: float = 0.0
+
+
+# ---------------------------------------------------------------------------
+# Keyword → provider routing rules
+# ---------------------------------------------------------------------------
+
+@dataclass
+class RoutingRule:
+    """A keyword-based routing rule."""
+    provider: str
+    model: str
+    keywords: List[str]
+    weight: float = 1.0
+
+
+# Default routing rules (reference: HiveMind ProviderRouter + CLAUDE.md mapping)
+DEFAULT_ROUTING_RULES: List[RoutingRule] = [
+    RoutingRule(
+        provider="gemini", model="3f",
+        keywords=["frontend", "前端", "react", "vue", "css", "html", "ui", "design", "设计", "样式", "组件", "tailwind", "nextjs"],
+        weight=1.5,
+    ),
+    RoutingRule(
+        provider="codex", model="o3",
+        keywords=["algorithm", "算法", "math", "数学", "proof", "证明", "reasoning", "推理", "逻辑", "logic", "complexity", "复杂度"],
+        weight=1.5,
+    ),
+    RoutingRule(
+        provider="codex", model="o3",
+        keywords=["review", "审查", "审核", "audit", "security", "安全", "code review", "代码审查"],
+        weight=1.5,
+    ),
+    RoutingRule(
+        provider="qwen", model="",
+        keywords=["python", "编程", "代码生成", "code", "coding", "implement", "实现", "sql", "database", "数据库", "数据分析"],
+        weight=1.0,
+    ),
+    RoutingRule(
+        provider="kimi", model="thinking",
+        keywords=["中文", "chinese", "翻译", "translate", "translation", "文案", "写作", "writing", "长文", "文档", "document", "总结", "summary", "分析"],
+        weight=1.0,
+    ),
+    RoutingRule(
+        provider="kimi", model="",
+        keywords=["explain", "解释", "概念", "concept", "快速", "quick", "shell", "bash", "运维", "devops"],
+        weight=0.8,
+    ),
+    RoutingRule(
+        provider="claude", model="",
+        keywords=["architecture", "架构", "设计模式", "design pattern", "重构", "refactor", "planning", "规划"],
+        weight=1.0,
+    ),
+]
+
+# Default fallback when no keywords match
+DEFAULT_FALLBACK = RouteResult(provider="kimi", model="", reason="default fallback", score=0.0)
+
+
+def _score_message(message: str, keywords: List[str]) -> float:
+    """Score a message against a list of keywords. Returns number of matches."""
+    text = message.lower()
+    score = 0.0
+    for kw in keywords:
+        if kw.lower() in text:
+            score += 1.0
+    return score
+
+
+def route_by_keywords(
+    message: str,
+    rules: Optional[List[RoutingRule]] = None,
+    fallback: Optional[RouteResult] = None,
+) -> RouteResult:
+    """Route a message to a provider based on keyword matching.
+
+    Returns the RouteResult with the highest weighted score.
+    """
+    if rules is None:
+        rules = DEFAULT_ROUTING_RULES
+    if fallback is None:
+        fallback = DEFAULT_FALLBACK
+
+    if not message or not message.strip():
+        return fallback
+
+    best: Optional[RouteResult] = None
+    best_score = 0.0
+
+    for rule in rules:
+        raw_score = _score_message(message, rule.keywords)
+        if raw_score <= 0:
+            continue
+        weighted = raw_score * rule.weight
+        if weighted > best_score:
+            best_score = weighted
+            matched = [kw for kw in rule.keywords if kw.lower() in message.lower()]
+            best = RouteResult(
+                provider=rule.provider,
+                model=rule.model,
+                reason=f"keywords: {', '.join(matched[:3])}",
+                score=weighted,
+            )
+
+    return best if best else fallback
+
+
+# ---------------------------------------------------------------------------
+# Team skill-based routing
+# ---------------------------------------------------------------------------
+
+def _score_agent_skills(agent: TeamAgent, message: str) -> float:
+    """Score a team agent against a message based on skills + role keywords."""
+    if not agent.skills and not agent.role:
+        return 0.0
+    text = message.lower()
+    score = 0.0
+    for skill in agent.skills:
+        if skill in text:
+            score += 1.5  # skills are more specific, higher weight
+    if agent.role and agent.role in text:
+        score += 1.0
+    return score
+
+
+def route_by_team(
+    message: str,
+    team: TeamConfig,
+) -> Optional[RouteResult]:
+    """Route a message to the best team agent based on skills and role matching.
+
+    Returns None if no agent has a positive match score.
+    """
+    if not message or not message.strip() or not team.agents:
+        return None
+
+    best_agent: Optional[TeamAgent] = None
+    best_score = 0.0
+
+    for agent in team.agents:
+        score = _score_agent_skills(agent, message)
+        if score > best_score:
+            best_score = score
+            best_agent = agent
+
+    if best_agent is None:
+        return None
+
+    matched = []
+    text = message.lower()
+    for s in best_agent.skills:
+        if s in text:
+            matched.append(s)
+    if best_agent.role and best_agent.role in text:
+        matched.append(f"role:{best_agent.role}")
+
+    return RouteResult(
+        provider=best_agent.provider,
+        model=best_agent.model,
+        reason=f"team:{best_agent.name} ({', '.join(matched[:3])})",
+        score=best_score,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Unified auto-route
+# ---------------------------------------------------------------------------
+
+def auto_route(
+    message: str,
+    team: Optional[TeamConfig] = None,
+) -> RouteResult:
+    """Auto-route a message to the best provider.
+
+    Resolution order:
+    1. Team skill-based matching (if team config exists)
+    2. Keyword-based matching (default rules)
+    3. Default fallback
+    """
+    # Try team-based routing first
+    if team is not None:
+        result = route_by_team(message, team)
+        if result:
+            return result
+
+    # Fall back to keyword-based routing
+    return route_by_keywords(message)