Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions bin/ask
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,11 @@ 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 team_config import load_team_config, resolve_team_agent


# Provider to daemon command mapping
Expand Down Expand Up @@ -481,11 +483,22 @@ 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)
Expand All @@ -507,6 +520,18 @@ def main(argv: list[str]) -> int:
_usage()
return EXIT_OK

# Resolution order: team agents > aliases > direct provider names
cwd = Path.cwd()
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:
Expand Down
62 changes: 62 additions & 0 deletions lib/aliases.py
Original file line number Diff line number Diff line change
@@ -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)
129 changes: 129 additions & 0 deletions lib/team_config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
"""Team configuration for CCB Agent Teams.

Loads team config from JSON files and resolves team agent names to providers.

Configuration layers (higher overrides lower):
1. ~/.ccb/team.json (global)
2. .ccb/team.json (project-level)

A team config defines named agents with provider, model, role, and skills.
Team agent names take priority over aliases when resolving provider names.
"""

from __future__ import annotations

import json
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional


@dataclass
class TeamAgent:
"""A named agent within a team."""
name: str
provider: str
model: str = ""
role: str = ""
skills: List[str] = field(default_factory=list)


@dataclass
class TeamConfig:
"""Team configuration with named agents and allocation strategy."""
name: str
agents: List[TeamAgent] = field(default_factory=list)
strategy: str = "skill_based" # round_robin | load_balance | skill_based
description: str = ""

def agent_map(self) -> Dict[str, TeamAgent]:
"""Build name → TeamAgent lookup (case-insensitive)."""
return {a.name.lower(): a for a in self.agents}


VALID_STRATEGIES = {"round_robin", "load_balance", "skill_based"}


def _parse_agent(raw: dict) -> Optional[TeamAgent]:
"""Parse a single agent entry from JSON. Returns None on invalid data."""
if not isinstance(raw, dict):
return None
name = str(raw.get("name", "")).strip()
provider = str(raw.get("provider", "")).strip().lower()
if not name or not provider:
return None
return TeamAgent(
name=name.lower(),
provider=provider,
model=str(raw.get("model", "")).strip(),
role=str(raw.get("role", "")).strip().lower(),
skills=[str(s).strip().lower() for s in raw.get("skills", []) if str(s).strip()],
)


def _load_team_json(path: Path) -> Optional[TeamConfig]:
"""Load a team config from a JSON file. Returns None on any error."""
try:
if not path.is_file():
return None
data = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(data, dict):
return None
except (json.JSONDecodeError, OSError, ValueError):
print(f"[WARN] Failed to parse team config: {path}", file=sys.stderr)
return None

name = str(data.get("name", "")).strip()
if not name:
name = "default"

strategy = str(data.get("strategy", "skill_based")).strip().lower()
if strategy not in VALID_STRATEGIES:
strategy = "skill_based"

agents: List[TeamAgent] = []
for raw_agent in data.get("agents", []):
agent = _parse_agent(raw_agent)
if agent:
agents.append(agent)

if not agents:
return None

return TeamConfig(
name=name,
agents=agents,
strategy=strategy,
description=str(data.get("description", "")).strip(),
)


def load_team_config(work_dir: Optional[Path] = None) -> Optional[TeamConfig]:
"""Load team config: project .ccb/team.json overrides global ~/.ccb/team.json.

Returns None if no valid team config is found.
"""
global_path = Path.home() / ".ccb" / "team.json"
global_config = _load_team_json(global_path)

project_config: Optional[TeamConfig] = None
if work_dir is not None:
project_path = work_dir / ".ccb" / "team.json"
project_config = _load_team_json(project_path)

# Project-level takes full priority (not merged)
return project_config or global_config


def resolve_team_agent(
name: str,
team: Optional[TeamConfig],
) -> Optional[TeamAgent]:
"""Resolve a name to a TeamAgent. Returns None if not a team agent."""
if team is None:
return None
key = (name or "").strip().lower()
if not key:
return None
return team.agent_map().get(key)
Loading
Loading