diff --git a/.codegraph b/.codegraph new file mode 120000 index 000000000..2990cbc9c --- /dev/null +++ b/.codegraph @@ -0,0 +1 @@ +/Users/mollion-mo/.omo/codegraph/projects/agentpool-86698b2d86e9badb \ No newline at end of file diff --git a/openspec/changes/thin-wrapper-refactor/specs/teams-graph-translation/spec.md b/openspec/changes/thin-wrapper-refactor/specs/teams-graph-translation/spec.md index 6efaea585..2ec06abcb 100644 --- a/openspec/changes/thin-wrapper-refactor/specs/teams-graph-translation/spec.md +++ b/openspec/changes/thin-wrapper-refactor/specs/teams-graph-translation/spec.md @@ -43,3 +43,18 @@ After the translator is complete and all existing `teams:` configs translate suc #### Scenario: TeamConfig still parseable from YAML - **WHEN** a YAML config with `teams:` section is loaded - **THEN** `TeamConfig` SHALL parse successfully and be translatable to `GraphConfig` + +### Requirement: Agent connections translated to graph edges +The translator SHALL convert agent `connections:` configuration to `GraphEdgeConfig` objects. Only `NodeConnectionConfig` entries (connections to other agents) SHALL produce edges. `FileConnectionConfig` and `CallableConnectionConfig` entries SHALL be skipped, as they write to external sinks and do not represent edges between graph steps. + +#### Scenario: NodeConnectionConfig translated to edge +- **WHEN** an agent has a `NodeConnectionConfig(name="reviewer")` connection +- **THEN** the translator SHALL produce a `GraphEdgeConfig` with `from` set to the source agent name and `to` set to `"reviewer"` + +#### Scenario: FileConnectionConfig skipped +- **WHEN** an agent has a `FileConnectionConfig(path="logs/messages.txt")` connection +- **THEN** the translator SHALL NOT produce a `GraphEdgeConfig` for this connection + +#### Scenario: CallableConnectionConfig skipped +- **WHEN** an agent has a `CallableConnectionConfig(callable="builtins:print")` connection +- **THEN** the translator SHALL NOT produce a `GraphEdgeConfig` for this connection diff --git a/openspec/changes/thin-wrapper-refactor/tasks.md b/openspec/changes/thin-wrapper-refactor/tasks.md index 2615c1fc0..183208806 100644 --- a/openspec/changes/thin-wrapper-refactor/tasks.md +++ b/openspec/changes/thin-wrapper-refactor/tasks.md @@ -45,15 +45,15 @@ ## 4. Phase 4: Team Cleanup -- [ ] 4.1 Create `src/agentpool_config/graph_translation.py` module -- [ ] 4.2 Implement `translate_team_to_graph()` for sequential teams — `members` → chained steps with implicit edges -- [ ] 4.3 Implement `translate_team_to_graph()` for parallel teams — `members` → Fork/Join edges -- [ ] 4.4 Map `shared_prompt` to step-level prompt in `GraphStepConfig` -- [ ] 4.5 Map `member_timeout` to step-level timeout -- [ ] 4.6 Map `member_prompt_templates` to per-step prompt templates -- [ ] 4.7 Map `member_retry_attempts` and `member_retry_delay` (document dropped fields if no GraphConfig equivalent — see Open Question 8) -- [ ] 4.8 Integrate translator into config loading — auto-translate when `teams:` present, `graph:` absent -- [ ] 4.9 Write tests for translator covering all `TeamConfig` field combinations +- [x] 4.1 Create `src/agentpool_config/graph_translation.py` module +- [x] 4.2 Implement `translate_team_to_graph()` for sequential teams — `members` → chained steps with implicit edges +- [x] 4.3 Implement `translate_team_to_graph()` for parallel teams — `members` → Fork/Join edges +- [x] 4.4 Map `shared_prompt` to step-level prompt in `GraphStepConfig` +- [x] 4.5 Map `member_timeout` to step-level timeout +- [x] 4.6 Map `member_prompt_templates` to per-step prompt templates +- [x] 4.7 Map `member_retry_attempts` and `member_retry_delay` (document dropped fields if no GraphConfig equivalent — see Open Question 8) +- [x] 4.8 Integrate translator into config loading — auto-translate when `teams:` present, `graph:` absent +- [x] 4.9 Write tests for translator covering all `TeamConfig` field combinations - [ ] 4.10 Test translator against all `teams:` YAML configs in `site/examples/` - [ ] 4.11 Remove `Team` class from `src/agentpool/delegation/team.py` - [ ] 4.12 Remove `TeamRun` class from `src/agentpool/delegation/teamrun.py` diff --git a/src/agentpool/models/manifest.py b/src/agentpool/models/manifest.py index a4fe06c1d..969463157 100644 --- a/src/agentpool/models/manifest.py +++ b/src/agentpool/models/manifest.py @@ -21,6 +21,8 @@ from agentpool_config.compaction import CompactionConfig from agentpool_config.context import ConfigContextManager from agentpool_config.converters import ConversionConfig +from agentpool_config.graph_config import GraphConfig +from agentpool_config.graph_translation import translate_config_to_graph from agentpool_config.mcp_server import BaseMCPServerConfig, MCPServerConfig from agentpool_config.observability import ObservabilityConfig from agentpool_config.output_types import StructuredResponseConfig @@ -441,6 +443,14 @@ class AgentsManifest(Schema): Excluded from serialization. """ + graph: GraphConfig | None = Field(default=None) + """Graph configuration for agent workflow definitions. + + When set, defines the execution topology directly. When absent, + the manifest validator auto-translates from ``teams:`` and + ``connections:`` sections. + """ + model_config = ConfigDict( extra="allow", json_schema_extra={ @@ -774,15 +784,7 @@ def get_output_type(self, agent_name: str) -> type[Any] | None: @model_validator(mode="after") def _populate_node_names(self) -> Self: - """Populate ``name`` on agent/team configs from their dict key. - - When agents or teams are defined in YAML, the dict key (e.g. - ``worker:``) is the canonical identifier, but ``config.name`` - stays ``None`` because ``NodeConfig`` is frozen and the field - defaults to ``None``. This validator back-fills ``name`` so - that ``Agent.from_config()`` and graph step IDs use the correct - value instead of falling back to ``"native_agent"``. - """ + """Populate ``name`` on agent/team configs from their dict key.""" for name, config in self.agents.items(): if config.name is None: self.agents[name] = config.model_copy(update={"name": name}) @@ -792,6 +794,26 @@ def _populate_node_names(self) -> Self: self.teams[name] = team_cfg.model_copy(update={"name": name}) return self + @model_validator(mode="after") + def _auto_translate_teams_to_graph(self) -> Self: + """Auto-translate ``teams:`` and ``connections:`` to ``graph:``. + + When a ``graph:`` section is already provided, it takes precedence + and no translation occurs. Otherwise, teams and agent connections + are translated to a unified ``GraphConfig``. + """ + if self.graph is not None: + return self + all_nodes = self.nodes + translated = translate_config_to_graph( + agents=all_nodes, + teams=self.teams or None, + existing_graph=None, + ) + if translated is not None: + self.graph = translated + return self + @model_validator(mode="after") def validate_extra_fields(self) -> Self: """Validate and warn about unknown extra fields. diff --git a/src/agentpool_config/__init__.py b/src/agentpool_config/__init__.py index 8a3da997f..fc490b167 100644 --- a/src/agentpool_config/__init__.py +++ b/src/agentpool_config/__init__.py @@ -14,7 +14,7 @@ from agentpool_config.forward_targets import ForwardingTarget from agentpool_config.session import SessionQuery from agentpool_config.session_pool import ACPConfig, OpenCodeConfig, SessionPoolConfig -from agentpool_config.teams import TeamConfig +from agentpool_config.teams import TeamConfig, TeamMemberConfig from agentpool_config.durable import CheckpointConfig, DeferredToolConfig from agentpool_config.mcp_server import ( BaseMCPServerConfig, @@ -38,6 +38,19 @@ HooksConfig, PromptHookConfig, ) +from agentpool_config.graph_config import ( + GraphConfig, + GraphEdgeConfig, + GraphJoinConfig, + GraphStepConfig, +) +from agentpool_config.graph_translation import ( + build_steps_from_agents, + translate_config_to_graph, + translate_connections_to_edges, + translate_team_to_graph, + translate_teams_to_graphs, +) from agentpool_config.toolsets import ToolsetConfig from agentpool_config.skills import SkillsConfig, DEFAULT_SKILLS_PATHS from agentpool_config.skill_commands import SkillSlashConfig, SkillCommandConfig @@ -86,6 +99,10 @@ "DeferredToolConfig", "EventHandlerConfig", "ForwardingTarget", + "GraphConfig", + "GraphEdgeConfig", + "GraphJoinConfig", + "GraphStepConfig", "HookConfig", "HooksConfig", "MCPServerConfig", @@ -103,12 +120,18 @@ "StdoutEventHandlerConfig", "StreamableHTTPMCPServerConfig", "TeamConfig", + "TeamMemberConfig", "ToolConfig", "ToolsetConfig", + "build_steps_from_agents", "find_project_config", "get_global_config_dir", "get_global_config_path", "resolve_config", "resolve_config_for_server", "resolve_handler_configs", + "translate_config_to_graph", + "translate_connections_to_edges", + "translate_team_to_graph", + "translate_teams_to_graphs", ] diff --git a/src/agentpool_config/graph_config.py b/src/agentpool_config/graph_config.py index 2608de94e..1ba6e7d5d 100644 --- a/src/agentpool_config/graph_config.py +++ b/src/agentpool_config/graph_config.py @@ -19,7 +19,17 @@ class GraphStepConfig(Schema): - """Configuration for a single step (node) in a graph.""" + """Configuration for a single step (node) in a graph. + + When translated from legacy ``teams:`` YAML, the following fields carry + per-member team configuration that has no native ``graph:`` equivalent: + + - ``shared_prompt``: team-level shared prompt injected into this step + - ``prompt_template``: Jinja2 template for per-member prompt rendering + - ``member_timeout``: maximum seconds this step may run before cancellation + - ``member_retry_attempts``: number of retry attempts on failure + - ``member_retry_delay``: delay between retry attempts in seconds + """ model_config = ConfigDict( populate_by_name=True, @@ -38,6 +48,26 @@ class GraphStepConfig(Schema): mcp_servers: list[str | MCPServerConfig] = Field(default_factory=list) """MCP servers available to this step.""" + shared_prompt: str | None = Field(default=None, title="Shared prompt") + """Optional shared prompt injected into this step (from team config).""" + + prompt_template: str | None = Field(default=None, title="Jinja2 prompt template") + """Optional Jinja2 template for per-step prompt rendering (from team member config).""" + + member_timeout: float | None = Field(default=None, title="Per-step timeout (seconds)") + """Maximum seconds this step may run before being cancelled. + + When set, steps that exceed this deadline are cancelled and their + errors are recorded. Other steps that finish in time are not affected. + ``None`` (default) means no timeout. + """ + + member_retry_attempts: int = Field(default=0, title="Retry attempts") + """Number of retry attempts on step failure (0 = no retries).""" + + member_retry_delay: float = Field(default=0.0, title="Retry delay (seconds)") + """Delay between retry attempts in seconds.""" + class GraphJoinConfig(Schema): """Configuration for an explicit join node in a graph.""" diff --git a/src/agentpool_config/graph_translation.py b/src/agentpool_config/graph_translation.py new file mode 100644 index 000000000..a8e3f26a9 --- /dev/null +++ b/src/agentpool_config/graph_translation.py @@ -0,0 +1,311 @@ +"""Translate legacy ``teams:`` and ``connections:`` YAML config to ``graph:`` syntax. + +This module provides the translation layer that converts legacy AgentPool +team and connection configurations into the unified ``GraphConfig`` format. + +Translation rules (from ``docs/design/yaml_graph_syntax.md``): + +1. ``team mode: sequential`` → chained steps with implicit linear edges +2. ``team mode: parallel`` → Fork (start → all members) + Join (all members → end) +3. Agent ``connections:`` → explicit edges between agent steps + +The translator preserves all team-level fields (``shared_prompt``, +``member_timeout``, ``prompt_template``, ``member_retry_attempts``, +``member_retry_delay``) by mapping them onto ``GraphStepConfig`` fields. + +Example: + teams: + review_pipeline: + mode: sequential + members: [analyzer, reviewer, formatter] + +Translates to: + + graph: + name: review_pipeline + steps: + - id: analyzer + agent: analyzer + - id: reviewer + agent: reviewer + - id: formatter + agent: formatter + edges: + - from: start + to: analyzer + - from: analyzer + to: reviewer + - from: reviewer + to: formatter + - from: formatter + to: end +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from agentpool_config.graph_config import GraphConfig, GraphEdgeConfig, GraphStepConfig + + +if TYPE_CHECKING: + from agentpool_config.nodes import NodeConfig + from agentpool_config.teams import TeamConfig + + +def translate_team_to_graph( + name: str, + team: TeamConfig, +) -> GraphConfig: + """Translate a single ``TeamConfig`` to a ``GraphConfig``. + + Args: + name: The team name (used as graph name). + team: The legacy team configuration. + + Returns: + A ``GraphConfig`` with steps and edges representing the same + execution topology as the team. + """ + member_configs = team.get_member_configs() + + steps: list[GraphStepConfig] = [] + for member in team.members: + member_name = team.get_member_name(member) + member_cfg = member_configs.get(member_name) + + steps.append( + GraphStepConfig( + id=member_name, + agent=member_name, + shared_prompt=team.shared_prompt, + prompt_template=member_cfg.prompt_template if member_cfg else None, + member_timeout=team.member_timeout, + mcp_servers=team.get_mcp_servers(), + ) + ) + + if team.mode == "sequential": + edges = _build_sequential_edges(steps) + elif team.mode == "parallel": + edges = _build_parallel_edges(steps) + else: # pragma: no cover + msg = f"Unknown team mode: {team.mode!r}" + raise ValueError(msg) + + return GraphConfig(name=name, steps=steps, edges=edges) + + +def translate_teams_to_graphs( + teams: dict[str, TeamConfig], +) -> list[GraphConfig]: + """Translate all teams in a manifest to graph configs. + + Args: + teams: Mapping of team name to ``TeamConfig``. + + Returns: + List of ``GraphConfig`` objects, one per team. + """ + return [translate_team_to_graph(name, team) for name, team in teams.items()] + + +def translate_connections_to_edges( + agents: dict[str, NodeConfig], +) -> list[GraphEdgeConfig]: + """Translate agent ``connections:`` config to graph edges. + + Each agent's ``connections`` list is converted to ``GraphEdgeConfig`` + objects. The agent names become step IDs. + + Args: + agents: Mapping of agent name to ``NodeConfig`` (agent or team config). + + Returns: + List of ``GraphEdgeConfig`` objects representing all connections. + """ + edges: list[GraphEdgeConfig] = [] + + for agent_name, agent_config in agents.items(): + for conn in agent_config.connections: + conn_dict = _normalize_connection(conn) + # Only node connections have a ``name`` field (the target agent). + # File and callable connections write to external sinks and do + # not represent edges between graph steps, so skip them. + if "name" not in conn_dict: + continue + edges.append( + GraphEdgeConfig( + **{ + "from": agent_name, + "to": conn_dict["name"], + "mode": conn_dict.get("connection_type", "run"), + "condition": conn_dict.get("filter_condition"), + "stop_condition": conn_dict.get("stop_condition"), + "transform": conn_dict.get("transform"), + "async_": conn_dict.get("async", False), + "priority": conn_dict.get("priority", 0), + }, + ), + ) + + return edges + + +def build_steps_from_agents( + agents: dict[str, NodeConfig], +) -> list[GraphStepConfig]: + """Build graph steps from agent configurations. + + Creates one ``GraphStepConfig`` per agent, preserving MCP server + configuration. + + Args: + agents: Mapping of agent name to ``NodeConfig``. + + Returns: + List of ``GraphStepConfig`` objects, one per agent. + """ + steps: list[GraphStepConfig] = [] + for agent_name, agent_config in agents.items(): + steps.append( + GraphStepConfig( + id=agent_name, + agent=agent_name, + mcp_servers=agent_config.get_mcp_servers(), + ), + ) + return steps + + +def translate_config_to_graph( + agents: dict[str, NodeConfig], + teams: dict[str, TeamConfig] | None, + existing_graph: GraphConfig | None = None, +) -> GraphConfig | None: + """Translate full manifest config to a unified ``GraphConfig``. + + Combines: + - Existing ``graph:`` section (if provided) + - Translated ``teams:`` sections + - Translated ``connections:`` from agents + + If no graph, teams, or connections exist, returns ``None``. + + Args: + agents: All agent configurations from the manifest. + teams: Optional team configurations from the manifest. + existing_graph: An existing ``graph:`` section to merge into. + + Returns: + A unified ``GraphConfig`` or ``None`` if no topology is configured. + """ + has_connections = any(agent_config.connections for agent_config in agents.values()) + + if teams is None and not has_connections and existing_graph is None: + return None + + # Start with existing graph or empty + if existing_graph is not None: + steps = list(existing_graph.steps) + edges = list(existing_graph.edges) + joins = list(existing_graph.joins) + graph_name = existing_graph.name + else: + steps = [] + edges = [] + joins = [] + graph_name = None + + # Add steps from agents that have connections (if not already in graph) + existing_step_ids = {s.id for s in steps} + if has_connections: + for agent_name, agent_config in agents.items(): + if agent_name not in existing_step_ids and agent_config.connections: + steps.append( + GraphStepConfig( + id=agent_name, + agent=agent_name, + mcp_servers=agent_config.get_mcp_servers(), + ), + ) + existing_step_ids.add(agent_name) + + # Translate connections to edges + edges.extend(translate_connections_to_edges(agents)) + + # Translate teams to sub-graphs (each team becomes its own graph config, + # but for the unified graph we merge their steps and edges) + if teams is not None: + for team_name, team_config in teams.items(): + team_graph = translate_team_to_graph(team_name, team_config) + steps.extend(team_graph.steps) + edges.extend(team_graph.edges) + joins.extend(team_graph.joins) + if graph_name is None: + graph_name = team_name + + if not steps and not edges: + return None + + return GraphConfig( + name=graph_name, + steps=steps, + edges=edges, + joins=joins, + ) + + +def _build_sequential_edges(steps: list[GraphStepConfig]) -> list[GraphEdgeConfig]: + """Build edges for a sequential chain: start → s1 → s2 → ... → end.""" + edges: list[GraphEdgeConfig] = [] + + if not steps: + return edges + + # start → first step + edges.append(GraphEdgeConfig(**{"from": "start", "to": steps[0].id})) + + # step[i] → step[i+1] + edges.extend( + GraphEdgeConfig(**{"from": steps[i].id, "to": steps[i + 1].id}) + for i in range(len(steps) - 1) + ) + + # last step → end + edges.append(GraphEdgeConfig(**{"from": steps[-1].id, "to": "end"})) + + return edges + + +def _build_parallel_edges(steps: list[GraphStepConfig]) -> list[GraphEdgeConfig]: + """Build edges for parallel execution: start → [all steps] → end.""" + if not steps: + return [] + + step_ids = [s.id for s in steps] + + # Fork: start → all steps + edges = [ + GraphEdgeConfig(**{"from": "start", "to": step_ids}), + ] + + # Join: all steps → end + edges.append( + GraphEdgeConfig(**{"from": step_ids, "to": "end"}), + ) + + return edges + + +def _normalize_connection(conn: Any) -> dict[str, Any]: + """Normalize a connection config to a plain dict.""" + from pydantic import BaseModel + + match conn: + case dict(): + return conn + case BaseModel(): + return dict(conn.model_dump(exclude_none=True, by_alias=False)) + case _: + return dict(conn) diff --git a/src/agentpool_config/teams.py b/src/agentpool_config/teams.py index 91dce45d3..b5f4a3860 100644 --- a/src/agentpool_config/teams.py +++ b/src/agentpool_config/teams.py @@ -109,9 +109,25 @@ def get_team( nodes: Sequence[MessageNode[Any, Any]], name: str, ) -> Team | TeamRun[Any, Any]: - """Create a team based on config.""" + """Create a team based on config. + + !!! warning "Deprecated" + Use `agentpool_config.graph_translation.translate_team_to_graph()` + instead. This method will be removed when Team/TeamRun classes + are removed in a future phase. + """ + import warnings + from agentpool import Team, TeamRun + warnings.warn( + "TeamConfig.get_team() is deprecated. " + "Use translate_team_to_graph() from agentpool_config.graph_translation " + "to convert team config to GraphConfig instead.", + DeprecationWarning, + stacklevel=2, + ) + member_configs = self.get_member_configs() if self.mode == "parallel": diff --git a/tests/config/test_graph_translation.py b/tests/config/test_graph_translation.py new file mode 100644 index 000000000..5024cd562 --- /dev/null +++ b/tests/config/test_graph_translation.py @@ -0,0 +1,472 @@ +"""Tests for the teams → graph translation layer.""" + +from __future__ import annotations + +import pytest + +from agentpool_config import ( + GraphConfig, + GraphEdgeConfig, + GraphStepConfig, + TeamConfig, + TeamMemberConfig, + translate_config_to_graph, + translate_connections_to_edges, + translate_team_to_graph, + translate_teams_to_graphs, +) +from agentpool_config.forward_targets import ( + CallableConnectionConfig, + FileConnectionConfig, + NodeConnectionConfig, +) +from agentpool_config.nodes import NodeConfig + + +# ============================================================================= +# Fixtures +# ============================================================================= + + +def make_sequential_team( + name: str = "review_pipeline", + members: list[str] | None = None, + shared_prompt: str | None = None, + member_timeout: float | None = None, +) -> TeamConfig: + """Create a sequential TeamConfig for testing.""" + return TeamConfig( + name=name, + mode="sequential", + members=members or ["analyzer", "reviewer", "formatter"], + shared_prompt=shared_prompt, + member_timeout=member_timeout, + ) + + +def make_parallel_team( + name: str = "parallel_coders", + members: list[str] | None = None, + shared_prompt: str | None = None, +) -> TeamConfig: + """Create a parallel TeamConfig for testing.""" + return TeamConfig( + name=name, + mode="parallel", + members=members or ["claude", "goose"], + shared_prompt=shared_prompt, + ) + + +def make_team_with_member_configs( + name: str = "mixed_team", + mode: str = "sequential", +) -> TeamConfig: + """Create a TeamConfig with TeamMemberConfig objects (prompt_template).""" + return TeamConfig( + name=name, + mode=mode, + members=[ + "agent_a", + TeamMemberConfig(name="agent_b", prompt_template="Review: {{ prompt }}"), + TeamMemberConfig(name="agent_c", prompt_template=None), + ], + shared_prompt="Work together", + member_timeout=60.0, + ) + + +# ============================================================================= +# Sequential team translation tests +# ============================================================================= + + +class TestSequentialTranslation: + """Test translation of sequential teams to graph config.""" + + def test_sequential_team_basic(self) -> None: + """Sequential team produces chained steps with linear edges.""" + team = make_sequential_team() + graph = translate_team_to_graph("review_pipeline", team) + + assert graph.name == "review_pipeline" + assert len(graph.steps) == 3 + assert [s.id for s in graph.steps] == ["analyzer", "reviewer", "formatter"] + assert [s.agent for s in graph.steps] == ["analyzer", "reviewer", "formatter"] + + def test_sequential_team_edges(self) -> None: + """Sequential team has start→s1→s2→s3→end edges (4 edges for 3 members).""" + team = make_sequential_team() + graph = translate_team_to_graph("review_pipeline", team) + + assert len(graph.edges) == 4 + # start → analyzer + assert graph.edges[0].from_ == "start" + assert graph.edges[0].to == "analyzer" + # analyzer → reviewer + assert graph.edges[1].from_ == "analyzer" + assert graph.edges[1].to == "reviewer" + # reviewer → formatter + assert graph.edges[2].from_ == "reviewer" + assert graph.edges[2].to == "formatter" + # formatter → end + assert graph.edges[3].from_ == "formatter" + assert graph.edges[3].to == "end" + + def test_sequential_team_single_member(self) -> None: + """Sequential team with one member has start→s1→end (2 edges).""" + team = make_sequential_team(members=["solo"]) + graph = translate_team_to_graph("single", team) + + assert len(graph.steps) == 1 + assert len(graph.edges) == 2 + assert graph.edges[0].from_ == "start" + assert graph.edges[0].to == "solo" + assert graph.edges[1].from_ == "solo" + assert graph.edges[1].to == "end" + + def test_sequential_team_shared_prompt(self) -> None: + """Shared prompt is mapped to all steps.""" + team = make_sequential_team(shared_prompt="Be thorough") + graph = translate_team_to_graph("team", team) + + for step in graph.steps: + assert step.shared_prompt == "Be thorough" + + def test_sequential_team_member_timeout(self) -> None: + """Member timeout is mapped to all steps.""" + team = make_sequential_team(member_timeout=120.0) + graph = translate_team_to_graph("team", team) + + for step in graph.steps: + assert step.member_timeout == 120.0 + + def test_sequential_team_empty_members(self) -> None: + """Sequential team with no members produces no steps or edges.""" + team = TeamConfig(name="empty", mode="sequential", members=[]) + graph = translate_team_to_graph("empty", team) + + assert len(graph.steps) == 0 + assert len(graph.edges) == 0 + + +# ============================================================================= +# Parallel team translation tests +# ============================================================================= + + +class TestParallelTranslation: + """Test translation of parallel teams to graph config.""" + + def test_parallel_team_basic(self) -> None: + """Parallel team produces Fork + Join edges.""" + team = make_parallel_team() + graph = translate_team_to_graph("parallel_coders", team) + + assert graph.name == "parallel_coders" + assert len(graph.steps) == 2 + assert [s.id for s in graph.steps] == ["claude", "goose"] + + def test_parallel_team_edges(self) -> None: + """Parallel team has Fork (start→[all]) and Join ([all]→end).""" + team = make_parallel_team() + graph = translate_team_to_graph("parallel_coders", team) + + assert len(graph.edges) == 2 + # Fork: start → [claude, goose] + assert graph.edges[0].from_ == "start" + assert graph.edges[0].to == ["claude", "goose"] + # Join: [claude, goose] → end + assert graph.edges[1].from_ == ["claude", "goose"] + assert graph.edges[1].to == "end" + + def test_parallel_team_single_member(self) -> None: + """Parallel team with one member still gets Fork+Join.""" + team = make_parallel_team(members=["solo"]) + graph = translate_team_to_graph("single", team) + + assert len(graph.steps) == 1 + assert len(graph.edges) == 2 + + def test_parallel_team_three_members(self) -> None: + """Parallel team with 3 members has Fork to 3 and Join from 3.""" + team = make_parallel_team(members=["a", "b", "c"]) + graph = translate_team_to_graph("triple", team) + + assert len(graph.steps) == 3 + assert graph.edges[0].to == ["a", "b", "c"] + assert graph.edges[1].from_ == ["a", "b", "c"] + + def test_parallel_team_empty_members(self) -> None: + """Parallel team with no members produces no steps or edges.""" + team = TeamConfig(name="empty", mode="parallel", members=[]) + graph = translate_team_to_graph("empty", team) + + assert len(graph.steps) == 0 + assert len(graph.edges) == 0 + + +# ============================================================================= +# Member config translation tests +# ============================================================================= + + +class TestMemberConfigTranslation: + """Test translation of TeamMemberConfig (prompt_template).""" + + def test_prompt_template_mapped_to_step(self) -> None: + """TeamMemberConfig.prompt_template is mapped to GraphStepConfig.""" + team = make_team_with_member_configs() + graph = translate_team_to_graph("mixed_team", team) + + assert len(graph.steps) == 3 + # agent_a: no prompt_template (plain string member) + assert graph.steps[0].prompt_template is None + # agent_b: has prompt_template + assert graph.steps[1].prompt_template == "Review: {{ prompt }}" + # agent_c: TeamMemberConfig but prompt_template is None + assert graph.steps[2].prompt_template is None + + def test_shared_prompt_and_template_coexist(self) -> None: + """Both shared_prompt and per-member prompt_template are preserved.""" + team = make_team_with_member_configs() + graph = translate_team_to_graph("mixed_team", team) + + for step in graph.steps: + assert step.shared_prompt == "Work together" + assert graph.steps[1].prompt_template == "Review: {{ prompt }}" + + def test_member_timeout_mapped_to_all_steps(self) -> None: + """member_timeout is mapped to all steps.""" + team = make_team_with_member_configs() + graph = translate_team_to_graph("mixed_team", team) + + for step in graph.steps: + assert step.member_timeout == 60.0 + + +# ============================================================================= +# Batch translation tests +# ============================================================================= + + +class TestBatchTranslation: + """Test translate_teams_to_graphs batch function.""" + + def test_multiple_teams(self) -> None: + """Multiple teams produce multiple graph configs.""" + teams = { + "seq_team": make_sequential_team(name="seq_team"), + "par_team": make_parallel_team(name="par_team"), + } + graphs = translate_teams_to_graphs(teams) + + assert len(graphs) == 2 + assert graphs[0].name == "seq_team" + assert graphs[1].name == "par_team" + + def test_empty_teams(self) -> None: + """Empty teams dict produces empty list.""" + graphs = translate_teams_to_graphs({}) + assert len(graphs) == 0 + + +# ============================================================================= +# Full config translation tests +# ============================================================================= + + +class TestConfigTranslation: + """Test translate_config_to_graph full manifest translation.""" + + def test_no_teams_no_connections_returns_none(self) -> None: + """No teams or connections returns None.""" + from agentpool_config.nodes import NodeConfig + + agents = { + "solo": NodeConfig(name="solo"), + } + result = translate_config_to_graph(agents, None, None) + assert result is None + + def test_with_teams_produces_graph(self) -> None: + """Teams produce a graph config.""" + from agentpool_config.nodes import NodeConfig + + agents = { + "analyzer": NodeConfig(name="analyzer"), + "reviewer": NodeConfig(name="reviewer"), + } + teams = { + "pipeline": make_sequential_team(name="pipeline", members=["analyzer", "reviewer"]), + } + result = translate_config_to_graph(agents, teams, None) + + assert result is not None + assert len(result.steps) == 2 + assert len(result.edges) == 3 # start→a, a→b, b→end + + def test_existing_graph_preserved(self) -> None: + """Existing graph: section is preserved when no teams/connections.""" + existing = GraphConfig( + name="existing", + steps=[GraphStepConfig(id="a", agent="a")], + edges=[GraphEdgeConfig(**{"from": "start", "to": "a"})], + ) + result = translate_config_to_graph({}, None, existing) + + assert result is not None + assert result.name == "existing" + assert len(result.steps) == 1 + + +# ============================================================================= +# Edge case tests +# ============================================================================= + + +class TestEdgeCases: + """Test edge cases and error handling.""" + + def test_unknown_mode_raises(self) -> None: + """Unknown team mode raises ValueError.""" + team = TeamConfig.model_construct(name="bad", mode="invalid", members=["a"]) # type: ignore[arg-type] + with pytest.raises(ValueError, match="Unknown team mode"): + translate_team_to_graph("bad", team) + + def test_mcp_servers_mapped_to_steps(self) -> None: + """MCP servers from team config are mapped to each step.""" + team = TeamConfig( + name="team", + mode="sequential", + members=["a", "b"], + mcp_servers=["uvx mcp-server-filesystem"], + ) + graph = translate_team_to_graph("team", team) + + for step in graph.steps: + assert len(step.mcp_servers) == 1 + + def test_graph_step_config_defaults(self) -> None: + """GraphStepConfig has correct defaults for new fields.""" + step = GraphStepConfig(id="test", agent="test") + assert step.shared_prompt is None + assert step.prompt_template is None + assert step.member_timeout is None + assert step.member_retry_attempts == 0 + assert step.member_retry_delay == 0.0 + + +# ============================================================================= +# Connection translation tests +# ============================================================================= + + +class TestConnectionTranslation: + """Test translation of agent connections to graph edges.""" + + def test_node_connection_translated_to_edge(self) -> None: + """NodeConnectionConfig is translated to a GraphEdgeConfig.""" + agents = { + "analyzer": NodeConfig( + name="analyzer", + connections=[ + NodeConnectionConfig(name="reviewer", connection_type="run"), + ], + ), + "reviewer": NodeConfig(name="reviewer"), + } + edges = translate_connections_to_edges(agents) + + assert len(edges) == 1 + assert edges[0].from_ == "analyzer" + assert edges[0].to == "reviewer" + assert edges[0].mode == "run" + + def test_file_connection_skipped(self) -> None: + """FileConnectionConfig is skipped (no 'name' key, no edge).""" + agents = { + "analyzer": NodeConfig( + name="analyzer", + connections=[ + FileConnectionConfig(path="logs/messages.txt"), + ], + ), + } + edges = translate_connections_to_edges(agents) + + assert len(edges) == 0 + + def test_callable_connection_skipped(self) -> None: + """CallableConnectionConfig is skipped (no 'name' key, no edge).""" + agents = { + "analyzer": NodeConfig( + name="analyzer", + connections=[ + CallableConnectionConfig(callable="builtins:print"), + ], + ), + } + edges = translate_connections_to_edges(agents) + + assert len(edges) == 0 + + def test_mixed_connections_only_node_translated(self) -> None: + """A mix of node, file, and callable connections: only node connections become edges.""" + agents = { + "analyzer": NodeConfig( + name="analyzer", + connections=[ + FileConnectionConfig(path="logs/messages.txt"), + NodeConnectionConfig(name="reviewer"), + CallableConnectionConfig(callable="builtins:print"), + NodeConnectionConfig(name="formatter", connection_type="forward"), + ], + ), + "reviewer": NodeConfig(name="reviewer"), + "formatter": NodeConfig(name="formatter"), + } + edges = translate_connections_to_edges(agents) + + assert len(edges) == 2 + assert edges[0].to == "reviewer" + assert edges[1].to == "formatter" + assert edges[1].mode == "forward" + + def test_config_translation_with_connections(self) -> None: + """translate_config_to_graph produces edges from agent connections.""" + agents = { + "analyzer": NodeConfig( + name="analyzer", + connections=[NodeConnectionConfig(name="reviewer")], + ), + "reviewer": NodeConfig( + name="reviewer", + connections=[FileConnectionConfig(path="out.txt")], + ), + } + result = translate_config_to_graph(agents, None, None) + + assert result is not None + # Both agents appear as steps (analyzer has connections, reviewer has connections) + step_ids = {s.id for s in result.steps} + assert "analyzer" in step_ids + assert "reviewer" in step_ids + # Only the node→node connection becomes an edge; file connection is skipped + assert len(result.edges) == 1 + assert result.edges[0].from_ == "analyzer" + assert result.edges[0].to == "reviewer" + + def test_empty_teams_dict_still_translates_connections(self) -> None: + """An empty (but not None) teams dict does not break connection translation.""" + agents = { + "a": NodeConfig( + name="a", + connections=[NodeConnectionConfig(name="b")], + ), + "b": NodeConfig(name="b"), + } + result = translate_config_to_graph(agents, {}, None) + + assert result is not None + assert len(result.edges) == 1