def parse_team_topology(data: Mapping[str, Any], *, base_dir: Path) -> TeamTopology:
"""Validate a mapping (typically loaded from ``naqsha.toml``) into a ``TeamTopology``."""
_ = base_dir # reserved for relative-path resolution alongside caller workspace root
allowed_top = {"workspace", "memory", "reflection", "agents"}
extra = set(data) - allowed_top
if extra:
raise ProfileValidationError(f"Unknown top-level keys: {sorted(extra)}.")
ws = data.get("workspace")
if not isinstance(ws, Mapping):
raise ProfileValidationError("'workspace' section is required.")
name = _as_str(ws.get("name", "team"), "workspace.name")
orchestrator = _as_str(ws.get("orchestrator"), "workspace.orchestrator")
dval = ws.get("description", "")
if dval is None or dval == "":
description = ""
elif isinstance(dval, str):
description = dval.strip()
else:
raise ProfileValidationError("'workspace.description' must be a string.")
trace_dir_raw = ws.get("trace_dir", ".naqsha/traces")
trace_dir = Path(_as_str(trace_dir_raw, "workspace.trace_dir"))
auto_approve = _as_bool(ws.get("auto_approve", False), "workspace.auto_approve")
approval_required_tiers = _parse_risk_tiers(ws.get("approval_required_tiers"))
sanitizer_max_chars = _as_int(
ws.get("sanitizer_max_chars", 4000), "workspace.sanitizer_max_chars", minimum=1
)
ws_extra = set(ws) - {
"name",
"orchestrator",
"description",
"trace_dir",
"auto_approve",
"approval_required_tiers",
"sanitizer_max_chars",
}
if ws_extra:
raise ProfileValidationError(f"Unknown workspace keys: {sorted(ws_extra)}.")
mem_blob = data.get("memory") or {}
if not isinstance(mem_blob, Mapping):
raise ProfileValidationError("'memory' must be a table.")
mem_type = _as_str(mem_blob.get("type", "sqlite"), "memory.type").lower()
db_path = Path(_as_str(mem_blob.get("db_path", ".naqsha/memory.db"), "memory.db_path"))
embeddings = _as_bool(mem_blob.get("embeddings", False), "memory.embeddings")
mem_extra = set(mem_blob) - {"type", "db_path", "embeddings"}
if mem_extra:
raise ProfileValidationError(f"Unknown memory keys: {sorted(mem_extra)}.")
memory = TeamMemoryConfig(type=mem_type, db_path=db_path, embeddings=embeddings)
refl_blob = data.get("reflection") or {}
if not isinstance(refl_blob, Mapping):
raise ProfileValidationError("'reflection' must be a table.")
reflection = ReflectionSection(
enabled=_as_bool(refl_blob.get("enabled", False), "reflection.enabled"),
auto_merge=_as_bool(refl_blob.get("auto_merge", False), "reflection.auto_merge"),
reliability_gate=_as_bool(
refl_blob.get("reliability_gate", True), "reflection.reliability_gate"
),
)
refl_extra = set(refl_blob) - {"enabled", "auto_merge", "reliability_gate"}
if refl_extra:
raise ProfileValidationError(f"Unknown reflection keys: {sorted(refl_extra)}.")
agents_blob = data.get("agents")
if not isinstance(agents_blob, Mapping) or not agents_blob:
raise ProfileValidationError("'agents' must be a non-empty table.")
if orchestrator not in agents_blob:
raise ProfileValidationError(
f"workspace.orchestrator {orchestrator!r} is missing from [agents]."
)
starter = starter_tool_names()
agent_keys = [_as_str(str(k), "agents") for k in agents_blob]
delegates_for_team = frozenset(
_safe_delegate_tool_name(wid) for wid in agent_keys if wid != orchestrator
)
agents: dict[str, AgentRoleConfig] = {}
for aid_raw, acfg in agents_blob.items():
agent_id = _as_str(str(aid_raw), "agents.<name>")
if not isinstance(acfg, Mapping):
raise ProfileValidationError(f"agents.{agent_id} must be a table.")
role = _as_str(acfg.get("role", "agent"), f"agents.{agent_id}.role")
model_adapter = (
_as_str(acfg.get("model_adapter", "fake"), f"agents.{agent_id}.model_adapter")
.strip()
.lower()
.replace("-", "_")
)
if model_adapter not in {"fake", "openai_compat", "anthropic", "gemini", "ollama"}:
raise ProfileValidationError(
f"agents.{agent_id}.model_adapter must be one of "
f"fake, openai_compat, anthropic, gemini, ollama; got {model_adapter!r}."
)
openai_blob = acfg.get("openai_compat")
anthropic_blob = acfg.get("anthropic")
gemini_blob = acfg.get("gemini")
ollama_blob = acfg.get("ollama")
if openai_blob is not None and model_adapter != "openai_compat":
raise ProfileValidationError(
f"agents.{agent_id}.openai_compat is only valid when "
f"model_adapter is openai_compat."
)
if anthropic_blob is not None and model_adapter != "anthropic":
raise ProfileValidationError(
f"agents.{agent_id}.anthropic is only valid when model_adapter is anthropic."
)
if gemini_blob is not None and model_adapter != "gemini":
raise ProfileValidationError(
f"agents.{agent_id}.gemini is only valid when model_adapter is gemini."
)
if ollama_blob is not None and model_adapter != "ollama":
raise ProfileValidationError(
f"agents.{agent_id}.ollama is only valid when model_adapter is ollama."
)
if openai_blob is not None and not isinstance(openai_blob, Mapping):
raise ProfileValidationError(f"agents.{agent_id}.openai_compat must be a table.")
if anthropic_blob is not None and not isinstance(anthropic_blob, Mapping):
raise ProfileValidationError(f"agents.{agent_id}.anthropic must be a table.")
if gemini_blob is not None and not isinstance(gemini_blob, Mapping):
raise ProfileValidationError(f"agents.{agent_id}.gemini must be a table.")
if ollama_blob is not None and not isinstance(ollama_blob, Mapping):
raise ProfileValidationError(f"agents.{agent_id}.ollama must be a table.")
tools_raw = acfg.get("tools")
if not isinstance(tools_raw, list) or not tools_raw:
raise ProfileValidationError(f"agents.{agent_id}.tools must be a non-empty array.")
tool_set: set[str] = set()
for t in tools_raw:
tname = _as_str(t, f"agents.{agent_id}.tools[]")
tool_set.add(tname)
if agent_id != orchestrator:
illegal_del = {t for t in tool_set if t.startswith("delegate_to_")}
if illegal_del:
raise ProfileValidationError(
f"agents.{agent_id} cannot declare delegation tools: {sorted(illegal_del)}."
)
elif delegates_for_team:
tool_set |= set(delegates_for_team)
diff = frozenset(tool_set) - starter - MEMORY_DECORATED_TOOL_NAMES
unknown_tools = diff - delegates_for_team
if unknown_tools:
raise ProfileValidationError(
f"agents.{agent_id}.tools contains unknown names: {sorted(unknown_tools)}."
)
default_b = BudgetLimits()
max_steps = _as_int(
acfg.get("max_steps", default_b.max_steps), f"agents.{agent_id}.max_steps"
)
max_tokens = acfg.get("max_model_tokens")
if max_tokens is None:
mt: int | None = default_b.max_model_tokens
elif isinstance(max_tokens, bool) or not isinstance(max_tokens, int):
raise ProfileValidationError(f"agents.{agent_id}.max_model_tokens must be int or null.")
elif max_tokens < 1:
raise ProfileValidationError(f"agents.{agent_id}.max_model_tokens must be >= 1.")
else:
mt = max_tokens
budgets = BudgetLimits(
max_steps=max_steps,
max_tool_calls=default_b.max_tool_calls,
wall_clock_seconds=default_b.wall_clock_seconds,
per_tool_seconds=default_b.per_tool_seconds,
max_model_tokens=mt,
)
budgets = _parse_budgets_for_agent(acfg.get("budgets"), defaults=budgets)
max_retries = _as_int(
acfg.get("max_retries", 3), f"agents.{agent_id}.max_retries", minimum=0
)
ins_raw = acfg.get("instructions", "")
if ins_raw is None or ins_raw == "":
agent_instructions_txt = ""
elif isinstance(ins_raw, str):
agent_instructions_txt = ins_raw
else:
raise ProfileValidationError(f"agents.{agent_id}.instructions must be a string.")
art = acfg.get("approval_required_tiers")
tiers: frozenset[RiskTier] | None
if art is None:
tiers = None
else:
tiers = _parse_risk_tiers(art)
fake_messages: tuple[dict[str, Any], ...] | None = None
fm = acfg.get("fake_model")
json_messages = acfg.get("fake_model_json")
if fm is not None and json_messages is not None:
raise ProfileValidationError(
f"agents.{agent_id}: use only one of fake_model or fake_model_json."
)
if fm is not None:
if not isinstance(fm, Mapping):
raise ProfileValidationError(f"agents.{agent_id}.fake_model must be a table.")
if model_adapter != "fake":
raise ProfileValidationError(
f"agents.{agent_id}.fake_model is only valid when model_adapter is fake."
)
fm_extra = set(fm) - {"messages"}
if fm_extra:
raise ProfileValidationError(
f"Unknown agents.{agent_id}.fake_model keys: {sorted(fm_extra)}."
)
fake_messages = _validate_fake_messages(fm.get("messages"))
elif json_messages is not None:
if model_adapter != "fake":
raise ProfileValidationError(
f"agents.{agent_id}.fake_model_json is only valid when model_adapter is fake."
)
if not isinstance(json_messages, str):
raise ProfileValidationError(
f"agents.{agent_id}.fake_model_json must be a string of JSON."
)
parsed = json.loads(json_messages)
if not isinstance(parsed, list):
raise ProfileValidationError("fake_model_json must decode to a JSON array.")
fake_messages = _validate_fake_messages(parsed)
acfg_extra = set(acfg) - {
"role",
"model_adapter",
"tools",
"max_steps",
"max_model_tokens",
"max_retries",
"budgets",
"approval_required_tiers",
"fake_model",
"fake_model_json",
"instructions",
"openai_compat",
"anthropic",
"gemini",
"ollama",
}
if acfg_extra:
raise ProfileValidationError(
f"Unknown keys for agents.{agent_id}: {sorted(acfg_extra)}."
)
agents[agent_id] = AgentRoleConfig(
agent_id=agent_id,
role=role,
model_adapter=model_adapter,
tools=frozenset(tool_set),
budgets=budgets,
max_retries=max_retries,
approval_required_tiers=tiers,
fake_model_messages=fake_messages,
instructions=agent_instructions_txt,
openai_compat=dict(openai_blob) if openai_blob is not None else None,
anthropic=dict(anthropic_blob) if anthropic_blob is not None else None,
gemini=dict(gemini_blob) if gemini_blob is not None else None,
ollama=dict(ollama_blob) if ollama_blob is not None else None,
)
return TeamTopology(
workspace=WorkspaceSection(
name=name,
orchestrator=orchestrator,
description=description,
trace_dir=trace_dir,
auto_approve=auto_approve,
approval_required_tiers=approval_required_tiers,
sanitizer_max_chars=sanitizer_max_chars,
),
agents=agents,
memory=memory,
reflection=reflection,
)