Environment API reference

AgentEloConfig

Bases: BaseModel

Validated configuration for a new or resumed Elo environment.

Source code in src/ursa/environments/config.py

class AgentEloConfig(BaseModel):
    """Validated configuration for a new or resumed Elo environment."""

    model_config = ConfigDict(
        frozen=True, extra="forbid", validate_default=True
    )

    name: str
    group: str = "default"
    description: str | None = None
    inference_providers: dict[str, InferenceProviderConfig] = Field(
        default_factory=dict
    )
    members: list[EnvironmentMemberConfig] = Field(default_factory=list)
    workspace: str | None = None
    defaults: dict[str, Any] = Field(default_factory=dict)
    initial_rating: float = Field(default=1500.0, allow_inf_nan=False)
    k_factor: float = Field(default=32.0, gt=0, allow_inf_nan=False)
    deaths_per_round: int = Field(default=1, ge=0)
    seed: int | None = None
    generations: int = Field(default=1, ge=1)
    restart_from_json: str | None = None
    member_timeout_seconds: float | None = Field(
        default=None, gt=0, allow_inf_nan=False
    )
    judge_prompt: str | None = None

    @field_validator("workspace", "restart_from_json", mode="before")
    @classmethod
    def _path_to_string(cls, value: Any) -> Any:
        return str(value) if isinstance(value, Path) else value

    @model_validator(mode="before")
    @classmethod
    def _resolve_members(cls, value: Any) -> Any:
        if not isinstance(value, Mapping):
            return value
        raw = dict(value)
        group = raw.get("group")
        if group is not None and not isinstance(group, str):
            raise ValueError("Group name must be a string")
        raw["group"] = validate_group_name(group)
        provider_data = raw.get("inference_providers") or {}
        if not isinstance(provider_data, Mapping):
            raise ValueError("inference_providers must be a mapping")
        providers = _inference_providers(provider_data)
        raw["inference_providers"] = providers
        members = raw.get("members", [])
        if isinstance(members, (list, tuple)):
            resolved = []
            for member in members:
                if isinstance(member, Mapping):
                    member = dict(member)
                    model = member.get("model")
                elif isinstance(member, EnvironmentMemberConfig):
                    model = member.model
                else:
                    resolved.append(member)
                    continue
                if model is not None:
                    if not isinstance(model, ModelConfig):
                        model = ModelConfig.model_validate(model)
                    if not model._inference_provider_resolved:
                        model = model.resolve_inference_provider(providers)
                    enforce_group_base_url_policy(model.base_url, raw["group"])
                    if isinstance(member, dict):
                        member["model"] = model
                    else:
                        member = replace(member, model=model)
                resolved.append(member)
            raw["members"] = resolved
        return raw

    @model_validator(mode="after")
    def _validate_population(self) -> Self:
        seen: set[str] = set()
        for member in self.members:
            self.validate_member_config(member)
            if member.name in seen:
                raise ValueError(
                    f"Elo member name {member.name!r} is duplicated. "
                    "Each member must have a unique name."
                )
            seen.add(member.name)
        if self.restart_from_json is None:
            self.validate_population_size(len(self.members))
        return self

    @staticmethod
    def validate_population_size(population_size: int) -> None:
        if population_size < 2:
            raise ValueError(
                "AgentEloEnvironment requires at least two active members."
            )
        if population_size % 2:
            raise ValueError(
                "AgentEloEnvironment requires an even number of active members. "
                f"Received {population_size}."
            )

    @staticmethod
    def validate_member_config(member: EnvironmentMemberConfig) -> None:
        """Validate member settings for both new and restored populations."""
        if "workspace" in (member.config or {}):
            raise ValueError(
                f"Member {member.name!r} sets config.workspace. "
                "AgentEloEnvironment manages member workspaces automatically. "
                "Set the top-level workspace instead."
            )
        if "agent_name" in (member.config or {}):
            raise ValueError(
                f"Member {member.name!r} sets config.agent_name. "
                "AgentEloEnvironment generates persistent agent identities "
                "automatically. Remove config.agent_name and use the "
                "member's name field instead."
            )

    @model_serializer(mode="wrap")
    def _serialize_resolved_models(self, handler):
        # Resolved endpoints are self-contained. Retaining the provider reference
        # would make the saved model invalid when read from YAML or dashboard JSON.
        data = handler(self)
        for member in data.get("members", []):
            model = member.get("model")
            if isinstance(model, dict):
                model.pop("inference_provider", None)
        return data

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> Self:
        """Compatibility entry point for callers loading a mapping."""
        return cls.model_validate(data)

    @classmethod
    def from_source(
        cls,
        config: Self | Mapping[str, Any] | str | Path | None = None,
        **overrides: Any,
    ) -> Self:
        """Merge constructor overrides before validating the final configuration.

        None overrides retain the supplied configuration, as in the environment
        constructor. Existing model instances are preserved during merging.
        """
        if isinstance(config, (str, Path)):
            raw = load_yaml_mapping(config)
        elif isinstance(config, cls):
            raw = {name: getattr(config, name) for name in cls.model_fields}
        elif isinstance(config, Mapping):
            raw = dict(config)
        elif config is None:
            raw = {"name": "agent_elo"}
        else:
            raise TypeError(
                "Elo config must be a config object, mapping, or YAML path"
            )
        raw.update({
            key: value for key, value in overrides.items() if value is not None
        })
        return cls.model_validate(raw)

from_mapping(data) classmethod

Compatibility entry point for callers loading a mapping.

Source code in src/ursa/environments/config.py

@classmethod
def from_mapping(cls, data: Mapping[str, Any]) -> Self:
    """Compatibility entry point for callers loading a mapping."""
    return cls.model_validate(data)

from_source(config=None, **overrides) classmethod

Merge constructor overrides before validating the final configuration.

None overrides retain the supplied configuration, as in the environment constructor. Existing model instances are preserved during merging.

Source code in src/ursa/environments/config.py

@classmethod
def from_source(
    cls,
    config: Self | Mapping[str, Any] | str | Path | None = None,
    **overrides: Any,
) -> Self:
    """Merge constructor overrides before validating the final configuration.

    None overrides retain the supplied configuration, as in the environment
    constructor. Existing model instances are preserved during merging.
    """
    if isinstance(config, (str, Path)):
        raw = load_yaml_mapping(config)
    elif isinstance(config, cls):
        raw = {name: getattr(config, name) for name in cls.model_fields}
    elif isinstance(config, Mapping):
        raw = dict(config)
    elif config is None:
        raw = {"name": "agent_elo"}
    else:
        raise TypeError(
            "Elo config must be a config object, mapping, or YAML path"
        )
    raw.update({
        key: value for key, value in overrides.items() if value is not None
    })
    return cls.model_validate(raw)

validate_member_config(member) staticmethod

Validate member settings for both new and restored populations.

Source code in src/ursa/environments/config.py

@staticmethod
def validate_member_config(member: EnvironmentMemberConfig) -> None:
    """Validate member settings for both new and restored populations."""
    if "workspace" in (member.config or {}):
        raise ValueError(
            f"Member {member.name!r} sets config.workspace. "
            "AgentEloEnvironment manages member workspaces automatically. "
            "Set the top-level workspace instead."
        )
    if "agent_name" in (member.config or {}):
        raise ValueError(
            f"Member {member.name!r} sets config.agent_name. "
            "AgentEloEnvironment generates persistent agent identities "
            "automatically. Remove config.agent_name and use the "
            "member's name field instead."
        )

AgentEloEnvironment

Bases: BaseEnvironment

Evolutionary pairwise agent environment using Elo ratings.

Each round:

1. Every active member independently works on the task.
2. Members compete pairwise.
3. An LLM judge returns A, B, or DRAW.
4. Elo ratings are updated deterministically.
5. Up to deaths_per_round decisive losers are eliminated.
6. The highest-rated survivors reproduce to restore population size.

For persistent URSA agents, descendants inherit:

• a copy of the parent's workspace,
• a fork of the parent's LangGraph checkpoint database,
• a fork of the parent's LangGraph store,
• the parent's Elo rating,
• lineage metadata.

Parent and child subsequently evolve independently.

Source code in src/ursa/environments/agent_elo.py

class AgentEloEnvironment(BaseEnvironment):
    """Evolutionary pairwise agent environment using Elo ratings.

    Each round:

    1. Every active member independently works on the task.
    2. Members compete pairwise.
    3. An LLM judge returns A, B, or DRAW.
    4. Elo ratings are updated deterministically.
    5. Up to ``deaths_per_round`` decisive losers are eliminated.
    6. The highest-rated survivors reproduce to restore population size.

    For persistent URSA agents, descendants inherit:

    - a copy of the parent's workspace,
    - a fork of the parent's LangGraph checkpoint database,
    - a fork of the parent's LangGraph store,
    - the parent's Elo rating,
    - lineage metadata.

    Parent and child subsequently evolve independently.
    """

    def __init__(
        self,
        llm: BaseChatModel,
        *,
        config: (AgentEloConfig | Mapping[str, Any] | str | Path | None) = None,
        name: str | None = None,
        group: str | None = None,
        members: (
            list[EnvironmentMemberConfig | Mapping[str, Any]] | None
        ) = None,
        workspace: str | Path | None = None,
        initial_rating: float | None = None,
        k_factor: float | None = None,
        deaths_per_round: int | None = None,
        seed: int | None = None,
        generations: int | None = None,
        member_timeout_seconds: float | None = None,
        restart_from_json: str | Path | None = None,
        judge_prompt: str | None = None,
        persist_members: bool = True,
        **kwargs: Any,
    ):
        elo_config = AgentEloConfig.from_source(
            config=config,
            name=name,
            group=group,
            members=members,
            workspace=workspace,
            initial_rating=initial_rating,
            k_factor=k_factor,
            deaths_per_round=deaths_per_round,
            seed=seed,
            generations=generations,
            member_timeout_seconds=member_timeout_seconds,
            restart_from_json=restart_from_json,
            judge_prompt=judge_prompt,
        )

        super().__init__(
            llm,
            name=elo_config.name,
            group=elo_config.group,
            workspace=elo_config.workspace or workspace,
            persist_members=persist_members,
            **kwargs,
        )

        self.config = elo_config

        self.initial_rating = elo_config.initial_rating
        self.k_factor = elo_config.k_factor
        self.deaths_per_round = elo_config.deaths_per_round
        self.generations = elo_config.generations
        self.member_timeout_seconds = elo_config.member_timeout_seconds

        self.seed = elo_config.seed

        self._rng = random.Random(self.seed)

        self.generation_index = 0

        self._offspring_counts: dict[
            str,
            int,
        ] = {}

        if self.config.restart_from_json is not None:
            if not self.persist_members:
                raise ValueError(
                    "restart_from_json requires persist_members=True."
                )

            state = self._load_environment_state(self.config.restart_from_json)

            self._restore_environment_state(state)

        else:
            self.member_configs = list(self.config.members)

            self._validate_population_size(len(self.member_configs))

            self.members = {
                member.name: self.build_member(member)
                for member in self.member_configs
            }

            self.players = {
                member.name: EloPlayer(
                    name=member.name,
                    rating=self.initial_rating,
                )
                for member in self.member_configs
            }

        self.judge_prompt = self.config.judge_prompt or ""

        self.judge = AgentEloJudge(
            llm=self.llm,
            workspace=self.workspace,
            group=self.group,
            judge_prompt=self.judge_prompt,
        )

    # ------------------------------------------------------------------
    # Configuration
    # ------------------------------------------------------------------

    # Keep checking live population sizes after reproduction and on restart.
    _validate_population_size = staticmethod(AgentEloConfig.validate_population_size)

    @classmethod
    def from_yaml(
        cls,
        path: str | Path,
        *,
        llm: BaseChatModel,
        **kwargs: Any,
    ) -> "AgentEloEnvironment":
        return cls(
            llm=llm,
            config=path,
            **kwargs,
        )

    # ------------------------------------------------------------------
    # For lightweight environment level persistence
    # ------------------------------------------------------------------

    def _events(
        self,
        config: Mapping[str, Any] | None,
    ) -> EnvironmentEvents:
        return EnvironmentEvents(
            environment=self.name,
            config=config,
            environment_type="agent_elo",
            environment_id=self.name,
            path=[self.name],
        )

    def _source(
        self,
        name: str,
        *,
        kind: str = "agent",
    ) -> dict[str, Any]:
        return {
            "id": f"{self.name}.{name}",
            "name": name,
            "kind": kind,
            "path": [
                self.name,
                name,
            ],
        }

    def _member_runtime_config(
        self,
        base_config: Mapping[str, Any] | None,
        member: EnvironmentMemberConfig,
    ) -> dict[str, Any] | None:
        """Attach stable Elo-member identity to nested agent/tool events."""

        if base_config is None:
            return None

        member_id = f"{self.name}.{member.name}"

        merged = dict(base_config)

        base_metadata = merged.get("metadata")

        metadata = (
            dict(base_metadata)
            if isinstance(
                base_metadata,
                Mapping,
            )
            else {}
        )

        metadata.update({
            "environment_id": self.name,
            "environment_member": (member.name),
            "environment_member_id": (member_id),
            "environment_member_role": (member.role),
            "environment_member_path": [
                self.name,
                member.name,
            ],
            "agent": member.name,
            "agent_id": member_id,
        })

        merged["metadata"] = metadata

        base_tags = merged.get("tags")

        if isinstance(base_tags, str):
            tags = [base_tags]
        else:
            tags = list(base_tags) if base_tags else []

        for tag in (
            member.name,
            member_id,
            "environment_member",
            "elo_member",
        ):
            if tag not in tags:
                tags.append(tag)

        merged["tags"] = tags

        return merged

    def _topology_payload(
        self,
        *,
        generation: int | None = None,
    ) -> dict[str, Any]:
        active_names = {member.name for member in self.member_configs}

        nodes: list[dict[str, Any]] = []

        # Active population.
        for member in self.member_configs:
            player = self.players[member.name]

            nodes.append({
                "id": (f"{self.name}.{member.name}"),
                "name": member.name,
                "kind": "agent",
                "role": member.role,
                "agent_class": (member.agent),
                "path": [
                    self.name,
                    member.name,
                ],
                "rating": player.rating,
                "lineage_generation": (player.generation),
                "parent": player.parent,
                "active": True,
            })

        # A restart can contain a child whose parent is no
        # longer active. Include a lightweight historical
        # parent node so its lineage edge still has a source.
        historical_parents = {
            player.parent
            for player in self.players.values()
            if (player.parent and player.parent not in active_names)
        }

        for parent in sorted(historical_parents):
            nodes.append({
                "id": (f"{self.name}.{parent}"),
                "name": parent,
                "kind": "agent",
                "role": ("Historical Elo parent"),
                "path": [
                    self.name,
                    parent,
                ],
                "active": False,
            })

        edges = []

        for player in self.players.values():
            if player.parent:
                edges.append({
                    "source": (f"{self.name}.{player.parent}"),
                    "target": (f"{self.name}.{player.name}"),
                    "kind": "parent_of",
                })

        return {
            "kind": "agent_elo",
            "name": self.name,
            "description": (self.config.description),
            "generation": (
                self.generation_index if generation is None else generation
            ),
            "nodes": nodes,
            "edges": edges,
        }

    def _member_config(
        self,
        name: str,
    ) -> EnvironmentMemberConfig:
        """Return the active config for one Elo member."""

        for member in self.member_configs:
            if member.name == name:
                return member

        raise KeyError(f"Unknown Elo member: {name}")

    @staticmethod
    def _member_config_to_mapping(
        member: EnvironmentMemberConfig,
    ) -> dict[str, Any]:
        """Convert a member config into JSON-safe data."""

        model = None

        if member.model is not None:
            model = member.model.model_dump(
                mode="json",
                exclude_none=True,
                # Persist resolved settings without requiring the original
                # provider registry (or conflicting with its resolved URL).
                exclude={"inference_provider"},
            )

        return {
            "name": member.name,
            "role": member.role,
            "agent": member.agent,
            "model": model,
            "config": member.config,
            "prompt": member.prompt,
            "reviewer": member.reviewer,
        }

    @staticmethod
    def _rng_state_to_json(
        value: Any,
    ) -> Any:
        if isinstance(value, tuple):
            return [
                AgentEloEnvironment._rng_state_to_json(item) for item in value
            ]

        return value

    @staticmethod
    def _rng_state_from_json(
        value: Any,
    ) -> Any:
        if isinstance(value, list):
            return tuple(
                AgentEloEnvironment._rng_state_from_json(item) for item in value
            )

        return value

    def _environment_state_path(
        self,
    ) -> Path:
        return Path(self.workspace) / "environment_state.json"

    def _environment_state_payload(
        self,
    ) -> dict[str, Any]:
        """Return lightweight resumable environment state."""

        config_by_name = {member.name: member for member in self.member_configs}

        active_players = []

        # Use member_configs ordering so active population order
        # is preserved exactly across restart.
        for member in self.member_configs:
            player = self.players[member.name]

            active_players.append({
                "name": player.name,
                "rating": player.rating,
                "generation": player.generation,
                "parent": player.parent,
                "member_config": (
                    self._member_config_to_mapping(config_by_name[player.name])
                ),
            })

        return {
            "schema_version": 1,
            "environment_name": self.name,
            "group": self.group,
            "generation": self.generation_index,
            "seed": self.seed,
            "rng_state": self._rng_state_to_json(self._rng.getstate()),
            "active_players": active_players,
        }

    def _save_environment_state(
        self,
    ) -> Path:
        """Atomically save lightweight environment restart state."""

        target = self._environment_state_path()

        target.parent.mkdir(
            parents=True,
            exist_ok=True,
        )

        temporary = target.with_suffix(".json.tmp")

        payload = self._environment_state_payload()

        temporary.write_text(
            json.dumps(
                payload,
                indent=2,
            )
            + "\n",
            encoding="utf-8",
        )

        temporary.replace(target)

        return target

    @staticmethod
    def _load_environment_state(
        path: str | Path,
    ) -> dict[str, Any]:
        state_path = Path(path).expanduser()

        if not state_path.exists():
            raise FileNotFoundError(
                f"Elo environment restart file does not exist: {state_path}"
            )

        state = json.loads(state_path.read_text(encoding="utf-8"))

        if not isinstance(state, dict):
            raise ValueError(
                "Elo environment restart file must contain a JSON object."
            )

        if state.get("schema_version") != 1:
            raise ValueError(
                "Unsupported Elo environment "
                "state schema version: "
                f"{state.get('schema_version')!r}"
            )

        return state

    def _restore_environment_state(
        self,
        state: Mapping[str, Any],
    ) -> None:
        """Restore the active evolutionary population.

        This method restores only environment metadata.

        Individual URSA agents reopen their own persistence
        independently when build_member() is called.
        """

        saved_name = state.get("environment_name")

        saved_group = state.get("group")

        if saved_name != self.name:
            raise ValueError(
                "Restart environment name mismatch: "
                f"snapshot={saved_name!r}, "
                f"current={self.name!r}"
            )

        if saved_group != self.group:
            raise ValueError(
                "Restart group mismatch: "
                f"snapshot={saved_group!r}, "
                f"current={self.group!r}"
            )

        raw_players = state.get("active_players")

        if not isinstance(
            raw_players,
            list,
        ):
            raise ValueError("Restart state is missing 'active_players'.")

        self._validate_population_size(len(raw_players))

        member_configs: list[EnvironmentMemberConfig] = []

        players: dict[
            str,
            EloPlayer,
        ] = {}

        for raw_player in raw_players:
            if not isinstance(
                raw_player,
                Mapping,
            ):
                raise ValueError("Each active player entry must be a mapping.")

            raw_member_config = raw_player.get("member_config")

            if not isinstance(
                raw_member_config,
                Mapping,
            ):
                raise ValueError("Restart player is missing 'member_config'.")

            restored_member_config = dict(raw_member_config)
            saved_model = restored_member_config.get("model")
            if isinstance(saved_model, Mapping):
                # Older snapshots retained the provider name alongside the
                # resolved settings. They too must load without a registry.
                saved_model = dict(saved_model)
                saved_model.pop("inference_provider", None)
                restored_member_config["model"] = saved_model

            member_config = EnvironmentMemberConfig.from_mapping(
                restored_member_config,
                group=self.group,
            )
            AgentEloConfig.validate_member_config(member_config)

            name = str(raw_player["name"])

            if name in players:
                raise ValueError(
                    f"Elo member name {name!r} is duplicated in restart state. "
                    "Each member must have a unique name."
                )

            if member_config.name != name:
                raise ValueError(
                    "Restart player/member config "
                    f"name mismatch: {name!r} vs "
                    f"{member_config.name!r}"
                )

            # A restart should resume an existing persistent
            # URSA agent, never silently create a fresh one.
            den = self._member_den(name)

            if not den.exists():
                raise FileNotFoundError(
                    "Persistent URSA agent den "
                    "required for restart does not exist: "
                    f"{den}"
                )

            member_workspace = self._member_workspace(name)

            if not member_workspace.exists():
                raise FileNotFoundError(
                    "Agent workspace required for "
                    "restart does not exist: "
                    f"{member_workspace}"
                )

            member_configs.append(member_config)

            players[name] = EloPlayer(
                name=name,
                rating=float(raw_player["rating"]),
                generation=int(raw_player["generation"]),
                parent=raw_player.get("parent"),
            )

        self.member_configs = member_configs

        self.players = players

        # Constructing by the same names causes URSA to reopen
        # the agents' existing persistent state.
        self.members = {
            member.name: self.build_member(member)
            for member in self.member_configs
        }

        self.generation_index = int(
            state.get(
                "generation",
                0,
            )
        )

        # The restart snapshot is authoritative for the
        # evolutionary RNG metadata.
        self.seed = state.get("seed")

        rng_state = state.get("rng_state")

        if rng_state is not None:
            self._rng.setstate(self._rng_state_from_json(rng_state))
        else:
            # Defensive support for snapshots that contain a seed
            # but no serialized RNG state.
            self._rng = random.Random(self.seed)

    # ------------------------------------------------------------------
    # Persistent URSA state
    # ------------------------------------------------------------------

    def _member_den(
        self,
        member_name: str,
    ) -> Path:
        """Return the persistent URSA den for a member."""
        agent_name = self._member_agent_name(member_name)

        if agent_name is None:
            raise RuntimeError(
                "Agent persistence is disabled. "
                "Persistent inheritance requires "
                "persist_members=True."
            )

        return group_agents_dir(self.group) / agent_name

    @staticmethod
    def _backup_sqlite_database(
        source: Path,
        destination: Path,
    ) -> None:
        """Create an independent SQLite snapshot."""
        source = Path(source)
        destination = Path(destination)

        if not source.exists():
            raise FileNotFoundError(f"SQLite source does not exist: {source}")

        destination.parent.mkdir(
            parents=True,
            exist_ok=True,
        )

        if destination.exists():
            raise FileExistsError(
                f"SQLite destination already exists: {destination}"
            )

        # SQLite's connection context manager only manages transactions.
        # Close both handles so rollback can remove child databases on Windows.
        with closing(sqlite3.connect(source)) as source_conn:
            with closing(sqlite3.connect(destination)) as destination_conn:
                source_conn.backup(destination_conn)

    def _fork_parent_persistence(
        self,
        parent_name: str,
        child_name: str,
    ) -> None:
        """Fork parent URSA persistence into a new child den."""
        parent_den = self._member_den(parent_name)

        child_den = self._member_den(child_name)

        if not parent_den.exists():
            raise FileNotFoundError(f"Parent den does not exist: {parent_den}")

        if child_den.exists():
            raise FileExistsError(f"Child den already exists: {child_den}")

        child_den.mkdir(
            parents=True,
            exist_ok=False,
        )

        try:
            self._backup_sqlite_database(
                parent_den / "db" / "checkpointer.db",
                child_den / "db" / "checkpointer.db",
            )

            graph_store_source = parent_den / "graph_store.sqlite"

            if graph_store_source.exists():
                self._backup_sqlite_database(
                    graph_store_source,
                    child_den / "graph_store.sqlite",
                )

        except Exception:
            # Do not leave a half-forked lineage behind.
            shutil.rmtree(
                child_den,
                ignore_errors=True,
            )
            raise

    # ------------------------------------------------------------------
    # Elo
    # ------------------------------------------------------------------

    def _apply_match_result(
        self,
        result: MatchResult,
    ) -> None:
        player_a = self.players[result.player_a]

        player_b = self.players[result.player_b]

        new_a, new_b = result.updated_ratings(
            player_a.rating,
            player_b.rating,
            self.k_factor,
        )

        player_a.rating = new_a
        player_b.rating = new_b

    # ------------------------------------------------------------------
    # Member execution
    # ------------------------------------------------------------------

    def _make_pairs(
        self,
        names: list[str],
    ) -> list[tuple[str, str]]:
        """Randomly pair all active members.

        AgentEloEnvironment requires an even population, so every
        active member participates in exactly one match per generation.
        """

        self._validate_population_size(len(names))

        shuffled = list(names)

        self._rng.shuffle(shuffled)

        return [
            (
                shuffled[index],
                shuffled[index + 1],
            )
            for index in range(
                0,
                len(shuffled),
                2,
            )
        ]

    def _progress_report_relative_path(self) -> Path:
        return (
            Path("_elo_progress") / f"generation_{self.generation_index + 1}.md"
        )

    def _read_progress_report(self, member_name: str) -> str | None:
        path = (
            self._member_workspace(member_name)
            / self._progress_report_relative_path()
        )
        try:
            report = path.read_text(encoding="utf-8", errors="replace").strip()
        except OSError:
            return None
        return report or None

    def _member_prompt(
        self,
        member: EnvironmentMemberConfig,
        task: str,
        *,
        deadline: datetime | None = None,
    ) -> str:
        player = self.players[member.name]
        if self.generation_index == 0:
            approach = "Develop and validate a strong independent approach to the task."
        elif player.parent is not None:
            approach = (
                "Build on your parent's work while exploring a substantive "
                "improvement or alternative approach. Preserve useful results "
                "and validate your changes; novelty alone is not success."
            )
        else:
            approach = (
                "Build on your existing work. Favor refinement and stronger "
                "validation, making larger changes when justified."
            )
        execution_budget = (
            f"\nDeadline (UTC): {deadline.isoformat()}\n"
            "Your run will be terminated at this deadline. Save useful results "
            "as you work and aim to provide a final response before time runs out.\n"
            if deadline is not None
            else ""
        )
        extra = (
            f"\nAdditional guidance:\n{member.prompt}\n"
            if member.prompt
            else ""
        )
        report_path = self._progress_report_relative_path().as_posix()
        return (
            f"You are competitor '{member.name}'.\n"
            f"Role: {member.role}\n"
            f"Environment generation: {self.generation_index + 1}\n"
            f"Lineage generation: {player.generation}\n"
            f"Parent: {player.parent or 'None'}\n\n"
            f"Task:\n{task}\n\n"
            f"Approach:\n{approach}\n\n"
            "Follow the task's evaluation criteria. Distinguish verified results "
            "from plans or untested claims.\n"
            f"{execution_budget}\n"
            "If you can write files, maintain a concise progress report at:\n"
            f"{report_path}\n\n"
            "Update it after meaningful milestones with completed work, verified "
            "results, remaining limitations, and supporting file paths. "
            "Distinguish new progress from inherited work. Write updates to a "
            "temporary file, then rename it into place.\n\n"
            "Your final response should summarize results, evidence, limitations, "
            "and files produced. If you time out, the judge will use your progress "
            "report or briefly inspect your workspace. A timeout alone is not a loss.\n"
            f"{extra}"
        )

    def _format_member_result(
        self,
        member_name: str,
        result: Any,
    ) -> str:
        """Return the member agent's canonical user-facing result."""

        member = self.members[member_name]

        formatter = getattr(
            member,
            "format_result",
            None,
        )

        if callable(formatter):
            formatted = formatter(result)

            if formatted is not None:
                return str(formatted)

        return result_to_text(result)

    async def _run_member(
        self,
        member: EnvironmentMemberConfig,
        task: str,
        invoke_kwargs: Mapping[str, Any],
        *,
        deadline: datetime | None = None,
    ) -> MemberRunResult:
        prompt = self._member_prompt(
            member,
            task,
            deadline=deadline,
        )

        async def ainvoke() -> Any:
            return await self._invoke_member_async(
                self.members[member.name],
                prompt,
                **invoke_kwargs,
            )

        deadline_text = deadline.isoformat() if deadline is not None else None
        status = "completed"
        output = None
        error = None
        progress_report = None

        try:
            if deadline is None:
                result = await ainvoke()

            else:
                remaining_seconds = (
                    deadline - datetime.now(timezone.utc)
                ).total_seconds()

                if remaining_seconds <= 0:
                    raise TimeoutError

                # Cancellation stops the environment from awaiting this
                # member. A blocking subprocess already running in an
                # executor may continue until that subprocess exits or
                # reaches its own timeout.
                result = await asyncio.wait_for(
                    ainvoke(),
                    timeout=remaining_seconds,
                )

            output = self._format_member_result(member.name, result)
            if not output.strip():
                progress_report = self._read_progress_report(member.name)

        except TimeoutError:
            status = "timed_out"
            output = None
            progress_report = self._read_progress_report(member.name)

        except Exception as exc:
            status = "failed"
            output = None
            error = f"{type(exc).__name__}: {exc}"

        return MemberRunResult(
            name=member.name,
            status=status,
            output=output,
            deadline=deadline_text,
            error=error,
            progress_report=progress_report,
        )

    # ------------------------------------------------------------------
    # Judging
    # ------------------------------------------------------------------

    @staticmethod
    def _judge_submission(run: MemberRunResult) -> str:
        if run.output and run.output.strip():
            evidence = f"Final response:\n{run.output}"
        elif run.progress_report:
            evidence = f"Saved progress report:\n{run.progress_report}"
        else:
            evidence = (
                "No final response or progress report is available. "
                "Briefly inspect this candidate's workspace for evidence."
            )
        return f"Execution status: {run.status}\n\n{evidence}"

    async def _resolve_match(
        self,
        *,
        task: str,
        player_a: str,
        run_a: MemberRunResult,
        player_b: str,
        run_b: MemberRunResult,
    ) -> MatchResult:
        """Resolve a match using execution status before LLM judging."""

        # Timeouts remain eligible for judging using partial work.
        if (run_a.completed or run_a.timed_out) and (
            run_b.completed or run_b.timed_out
        ):
            return await self._judge_match(
                task=task,
                player_a=player_a,
                output_a=self._judge_submission(run_a),
                player_b=player_b,
                output_b=self._judge_submission(run_b),
            )

        # A completed and B did not.
        if run_a.completed:
            return MatchResult(
                player_a=player_a,
                player_b=player_b,
                score_a=1.0,
                reasoning=(
                    f"{player_a} completed the generation successfully; "
                    f"{player_b} did not complete successfully "
                    f"(status={run_b.status}). "
                    f"{player_a} therefore wins automatically."
                ),
            )

        # B completed and A did not.
        if run_b.completed:
            return MatchResult(
                player_a=player_a,
                player_b=player_b,
                score_a=0.0,
                reasoning=(
                    f"{player_b} completed the generation successfully; "
                    f"{player_a} did not complete successfully "
                    f"(status={run_a.status}). "
                    f"{player_b} therefore wins automatically."
                ),
            )

        if run_a.timed_out and run_b.failed:
            return MatchResult(
                player_a=player_a,
                player_b=player_b,
                score_a=1.0,
                reasoning=(
                    f"{player_a} reached the execution deadline; "
                    f"{player_b} failed during execution and forfeits the match. "
                    f"{player_a} therefore wins automatically."
                ),
            )

        if run_b.timed_out and run_a.failed:
            return MatchResult(
                player_a=player_a,
                player_b=player_b,
                score_a=0.0,
                reasoning=(
                    f"{player_b} reached the execution deadline; "
                    f"{player_a} failed during execution and forfeits the match. "
                    f"{player_b} therefore wins automatically."
                ),
            )

        # Both failed.
        return MatchResult(
            player_a=player_a,
            player_b=player_b,
            score_a=0.5,
            reasoning=(
                "Neither competitor completed the generation successfully. "
                f"{player_a} status={run_a.status}; "
                f"{player_b} status={run_b.status}. "
                "Match recorded as a draw."
            ),
        )

    async def _judge_match(
        self,
        task: str,
        player_a: str,
        output_a: str,
        player_b: str,
        output_b: str,
    ) -> MatchResult:
        config_a = self._member_config(player_a)

        config_b = self._member_config(player_b)

        decision = await self.judge.judge_match(
            task=task,
            player_a=player_a,
            agent_type_a=config_a.agent,
            output_a=output_a,
            player_b=player_b,
            agent_type_b=config_b.agent,
            output_b=output_b,
        )

        if decision.winner == "A":
            score_a = 1.0

        elif decision.winner == "B":
            score_a = 0.0

        else:
            score_a = 0.5

        return MatchResult(
            player_a=player_a,
            player_b=player_b,
            score_a=score_a,
            reasoning=decision.reasoning,
        )

    # ------------------------------------------------------------------
    # Selection
    # ------------------------------------------------------------------

    def _select_losers(
        self,
        match_results: list[MatchResult],
    ) -> list[str]:
        """Choose decisive losers for elimination.

        Draws produce no loser.

        If more decisive losers exist than the configured number
        of deaths, the lowest-rated losing agents are eliminated.
        """
        losers = [
            result.loser for result in match_results if result.loser is not None
        ]

        # Defensive deduplication.
        losers = list(dict.fromkeys(losers))

        self._rng.shuffle(losers)

        losers.sort(key=lambda name: (self.players[name].rating))

        return losers[: self.deaths_per_round]

    def _eliminate(
        self,
        losers: list[str],
    ) -> None:
        """Remove agents from the active population.

        Persistent dens and workspaces are intentionally left on disk
        so extinct lineages remain inspectable.
        """
        loser_set = set(losers)

        for name in losers:
            member = self.members.pop(
                name,
                None,
            )

            if member is not None:
                close = getattr(
                    member,
                    "close",
                    None,
                )

                if callable(close):
                    try:
                        close()
                    except Exception:
                        logging.getLogger(__name__).warning(
                            "Failed to close eliminated Elo member %s",
                            name,
                            exc_info=True,
                        )

            self.players.pop(
                name,
                None,
            )

        self.member_configs = [
            member
            for member in self.member_configs
            if member.name not in loser_set
        ]

    def _top_survivors(
        self,
        count: int,
        *,
        excluded: list[str] | None = None,
    ) -> list[EloPlayer]:
        """Return the highest-rated surviving agents.

        Equal-rated survivors are ordered randomly using
        the environment RNG.
        """
        excluded_names = set(excluded or [])
        candidates = [
            player
            for player in self.players.values()
            if player.name not in excluded_names
        ]

        self._rng.shuffle(candidates)

        candidates.sort(
            key=lambda player: player.rating,
            reverse=True,
        )

        return candidates[:count]

    # ------------------------------------------------------------------
    # Reproduction
    # ------------------------------------------------------------------

    def _child_name_available(
        self,
        child_name: str,
    ) -> bool:
        """Return whether a name is safe to use for a new birth.

        A generated child name must not collide with:

        - an active player,
        - an active member,
        - an active member config,
        - an existing environment workspace,
        - an existing persistent URSA den.

        Existing persistent dens are checked only when member
        persistence is enabled.
        """

        if child_name in self.players:
            return False

        if child_name in self.members:
            return False

        if any(member.name == child_name for member in self.member_configs):
            return False

        child_workspace = self._member_workspace(child_name)

        if child_workspace.exists():
            return False

        if self.persist_members:
            child_den = self._member_den(child_name)

            if child_den.exists():
                return False

        return True

    def _next_child_name(
        self,
        parent: EloPlayer,
    ) -> str:
        """Generate the next unused descendant identity.

        The search starts after the highest child count generated
        by this environment instance, but also checks existing
        workspaces and persistent agent dens so resumed runs cannot
        accidentally reuse an older descendant.
        """

        count = (
            self._offspring_counts.get(
                parent.name,
                0,
            )
            + 1
        )

        child_generation = parent.generation + 1

        while True:
            child_name = f"{parent.name}_g{child_generation}_{count}"

            if self._child_name_available(child_name):
                self._offspring_counts[parent.name] = count

                return child_name

            count += 1

    def _copy_parent_workspace(
        self,
        parent_name: str,
        child_name: str,
    ) -> None:
        """Fork the parent's working filesystem."""
        parent_workspace = self._member_workspace(parent_name)

        child_workspace = self._member_workspace(child_name)

        if child_workspace.exists():
            raise FileExistsError(
                f"Child workspace already exists: {child_workspace}"
            )

        if parent_workspace.exists():
            shutil.copytree(
                parent_workspace,
                child_workspace,
            )

        else:
            child_workspace.mkdir(
                parents=True,
                exist_ok=False,
            )

    def _cleanup_failed_child(
        self,
        child_name: str,
    ) -> None:
        """Best-effort rollback of a failed reproduction."""
        child_workspace = self._member_workspace(child_name)

        shutil.rmtree(
            child_workspace,
            ignore_errors=True,
        )

        if self.persist_members:
            try:
                child_den = self._member_den(child_name)
            except Exception:
                return

            shutil.rmtree(
                child_den,
                ignore_errors=True,
            )

    def _reproduce(
        self,
        parents: list[EloPlayer],
    ) -> list[str]:
        """Create one independent child for every selected parent.

        Children inherit:

        - Elo rating,
        - lineage metadata,
        - workspace,
        - URSA checkpoint state,
        - LangGraph persistent store.
        """
        children: list[str] = []
        prepared: list[tuple[EnvironmentMemberConfig, Any, EloPlayer]] = []
        offspring_counts_before = dict(self._offspring_counts)
        config_by_name = {member.name: member for member in self.member_configs}

        try:
            for parent in parents:
                parent_config = config_by_name[parent.name]
                child_name = self._next_child_name(parent)
                children.append(child_name)
                child_config = replace(parent_config, name=child_name)

                # Prepare every child before changing the active population.
                self._copy_parent_workspace(parent.name, child_name)
                if self.persist_members:
                    self._fork_parent_persistence(parent.name, child_name)
                child = self.build_member(child_config)
                prepared.append((
                    child_config,
                    child,
                    EloPlayer(
                        name=child_name,
                        rating=parent.rating,
                        generation=parent.generation + 1,
                        parent=parent.name,
                    ),
                ))
        except BaseException:
            # Close constructed children before deleting their persistence.
            # A cleanup error must not prevent rollback of the other children.
            for _, child, _ in prepared:
                try:
                    close = getattr(child, "close", None)
                    if callable(close):
                        close()
                except Exception:
                    pass
            for child_name in children:
                self._cleanup_failed_child(child_name)
            self._offspring_counts = offspring_counts_before
            raise

        for child_config, child, player in prepared:
            self.members[player.name] = child
            self.member_configs.append(child_config)
            self.players[player.name] = player

        return children

    # ------------------------------------------------------------------
    # Reporting
    # ------------------------------------------------------------------

    def standings(
        self,
    ) -> list[dict[str, Any]]:
        ordered = sorted(
            self.players.values(),
            key=lambda player: (player.rating),
            reverse=True,
        )

        return [
            {
                "rank": rank,
                "name": player.name,
                "rating": player.rating,
                "generation": player.generation,
                "parent": player.parent,
            }
            for rank, player in enumerate(
                ordered,
                start=1,
            )
        ]

    # ------------------------------------------------------------------
    # Environment invocation
    # ------------------------------------------------------------------

    def _invoke(
        self,
        inputs: Mapping[str, Any],
        **config: Any,
    ) -> dict[str, Any]:
        return self._run_ainvoke_from_sync(
            inputs,
            **config,
        )

    async def _run_generation(
        self,
        task: str,
        invoke_kwargs: Mapping[str, Any],
        *,
        runtime_config: Mapping[
            str,
            Any,
        ]
        | None = None,
    ) -> dict[str, Any]:
        """Run one complete evolutionary generation."""

        self._validate_population_size(len(self.member_configs))

        if not (
            len(self.member_configs) == len(self.members) == len(self.players)
        ):
            raise RuntimeError(
                "AgentEloEnvironment active population state is inconsistent: "
                f"member_configs={len(self.member_configs)}, "
                f"members={len(self.members)}, "
                f"players={len(self.players)}."
            )

        events = self._events(runtime_config)

        generation_start = perf_counter()

        generation_number = self.generation_index + 1

        initial_population_size = len(self.players)
        rng_state_before = self._rng.getstate()

        # Capture ratings before competition for reporting.
        ratings_before = {
            name: player.rating for name, player in self.players.items()
        }

        generation_deadline: datetime | None = None

        if self.member_timeout_seconds is not None:
            generation_deadline = datetime.now(timezone.utc) + timedelta(
                seconds=self.member_timeout_seconds
            )

        await events.aemit(
            (f"Elo generation {generation_number} started"),
            stage="generation",
            phase="start",
            event_type="generation_started",
            generation=generation_number,
            population_size=len(self.players),
            ratings={
                name: player.rating for name, player in self.players.items()
            },
            deadline=(
                generation_deadline.isoformat()
                if generation_deadline is not None
                else None
            ),
        )

        async def run_member(
            member: EnvironmentMemberConfig,
        ) -> MemberRunResult:
            source = self._source(member.name)

            start = perf_counter()

            await events.aemit(
                (f"Elo member {member.name} started"),
                stage="member",
                phase="start",
                event_type="member_started",
                generation=generation_number,
                source=source,
            )

            member_kwargs = dict(invoke_kwargs)

            member_runtime_config = self._member_runtime_config(
                runtime_config,
                member,
            )

            if member_runtime_config is not None:
                member_kwargs["config"] = member_runtime_config

            result = await self._run_member(
                member,
                task,
                member_kwargs,
                deadline=generation_deadline,
            )

            if result.completed:
                event_type = "member_completed"
                phase = "end"
                level = "info"

            elif result.timed_out:
                event_type = "member_timed_out"
                phase = "error"
                level = "warning"

            else:
                event_type = "member_failed"
                phase = "error"
                level = "error"

            await events.aemit(
                (f"Elo member {member.name} {result.status}"),
                stage="member",
                phase=phase,
                event_type=event_type,
                level=level,
                generation=generation_number,
                source=source,
                status=result.status,
                result=result.output,
                error=result.error,
                deadline=result.deadline,
                elapsed_seconds=(perf_counter() - start),
            )

            return result

        # ----------------------------------------------------------
        # Phase 1: independent research
        # ----------------------------------------------------------

        member_results = await asyncio.gather(*[
            run_member(member) for member in self.member_configs
        ])

        runs = {result.name: result for result in member_results}

        outputs = {
            name: (result.output if result.completed else None)
            for name, result in runs.items()
        }

        timed_out = [name for name, result in runs.items() if result.timed_out]

        failed = [name for name, result in runs.items() if result.failed]

        # ----------------------------------------------------------
        # Phase 2: randomized pairwise competition
        # ----------------------------------------------------------

        pairs = self._make_pairs([
            member.name for member in self.member_configs
        ])

        await events.aemit(
            (f"Elo generation {generation_number} pairings created"),
            stage="pairing",
            phase="declared",
            event_type="pairings_declared",
            generation=generation_number,
            pairs=[list(pair) for pair in pairs],
        )

        match_results: list[MatchResult] = []

        for player_a, player_b in pairs:
            rating_a_before = self.players[player_a].rating

            rating_b_before = self.players[player_b].rating

            await events.aemit(
                (f"Elo match {player_a} vs {player_b} started"),
                stage="match",
                phase="start",
                event_type="match_started",
                generation=generation_number,
                source=self._source(player_a),
                target=self._source(player_b),
            )

            result = await self._resolve_match(
                task=task,
                player_a=player_a,
                run_a=runs[player_a],
                player_b=player_b,
                run_b=runs[player_b],
            )

            self._apply_match_result(result)

            await events.aemit(
                (f"Elo match {player_a} vs {player_b} completed"),
                stage="match",
                phase="end",
                event_type="match_completed",
                generation=generation_number,
                source=self._source(player_a),
                target=self._source(player_b),
                winner=result.winner,
                loser=result.loser,
                score_a=result.score_a,
                reasoning=result.reasoning,
                rating_a_before=(rating_a_before),
                rating_a_after=(self.players[player_a].rating),
                rating_b_before=(rating_b_before),
                rating_b_after=(self.players[player_b].rating),
            )

            match_results.append(result)

        standings_after_matches = self.standings()

        # ----------------------------------------------------------
        # Phase 3: prepare replacement children, then eliminate losers
        # ----------------------------------------------------------

        eliminated = self._select_losers(match_results)
        parents = self._top_survivors(len(eliminated), excluded=eliminated)
        parent_names = [parent.name for parent in parents]

        try:
            children = self._reproduce(parents)
        except BaseException:
            # Keep failed generations retryable without applying Elo twice.
            for name, rating in ratings_before.items():
                self.players[name].rating = rating
            self._rng.setstate(rng_state_before)
            raise

        eliminated_ratings = {
            name: self.players[name].rating for name in eliminated
        }
        self._eliminate(eliminated)

        for name in eliminated:
            await events.aemit(
                (f"Elo member {name} eliminated"),
                stage="selection",
                phase="end",
                event_type="member_eliminated",
                generation=generation_number,
                source=self._source(name),
                rating=eliminated_ratings[name],
            )

        for child_name in children:
            child = self.players[child_name]

            await events.aemit(
                (f"Elo child {child_name} created"),
                stage="reproduction",
                phase="end",
                event_type="child_created",
                generation=generation_number,
                source=(self._source(child.parent) if child.parent else None),
                target=self._source(child_name),
                child=child_name,
                parent=child.parent,
                rating=child.rating,
                lineage_generation=(child.generation),
            )

        await events.aemit(
            (f"Elo generation {generation_number} topology updated"),
            stage="elo",
            phase="topology",
            event_type="topology_declared",
            generation=generation_number,
            topology=self._topology_payload(
                generation=generation_number,
            ),
        )

        final_population_size = len(self.players)

        if final_population_size != initial_population_size:
            raise RuntimeError(
                "Population size changed unexpectedly: "
                f"{initial_population_size} -> "
                f"{final_population_size}"
            )

        # ----------------------------------------------------------
        # Generation successfully completed.
        #
        # Increment BEFORE snapshotting so restart begins at the
        # following generation.
        # ----------------------------------------------------------

        self.generation_index += 1

        state_path = self._save_environment_state()

        await events.aemit(
            (f"Elo generation {generation_number} completed"),
            stage="generation",
            phase="end",
            event_type="generation_completed",
            generation=generation_number,
            eliminated=eliminated,
            children=children,
            standings=self.standings(),
            elapsed_seconds=(perf_counter() - generation_start),
        )

        return {
            "generation": generation_number,
            "task": task,
            "outputs": outputs,
            "pairs": [list(pair) for pair in pairs],
            "ratings_before": ratings_before,
            "matches": [
                {
                    "player_a": result.player_a,
                    "player_b": result.player_b,
                    "winner": result.winner,
                    "loser": result.loser,
                    "score_a": result.score_a,
                    "reasoning": result.reasoning,
                }
                for result in match_results
            ],
            "standings_after_matches": (standings_after_matches),
            "eliminated": eliminated,
            "reproducing_parents": (parent_names),
            "children": children,
            "standings": self.standings(),
            "population_size": (final_population_size),
            "environment_state": str(state_path),
            "member_runs": {
                name: {
                    "status": run.status,
                    "deadline": run.deadline,
                    "error": run.error,
                    "progress_report": run.progress_report,
                }
                for name, run in runs.items()
            },
            "timed_out": timed_out,
            "failed": failed,
            "generation_deadline": (
                generation_deadline.isoformat()
                if generation_deadline is not None
                else None
            ),
        }

    async def _ainvoke(
        self,
        inputs: Mapping[str, Any],
        **config: Any,
    ) -> dict[str, Any]:
        task = str(inputs.get("task") or inputs.get("prompt") or inputs)

        runtime_config = runnable_config_from_kwargs(config)

        events = self._events(runtime_config)

        run_start = perf_counter()

        invoke_kwargs = invocation_kwargs(config)

        generation_results = []

        starting_generation = self.generation_index

        await events.aemit(
            f"Agent Elo {self.name} started",
            stage="elo",
            phase="start",
            event_type="elo_started",
            task=task,
            starting_generation=(self.generation_index),
            requested_generations=(self.generations),
            topology=(self._topology_payload()),
        )

        await events.aemit(
            (f"Agent Elo {self.name} topology declared"),
            stage="elo",
            phase="topology",
            event_type="topology_declared",
            topology=(self._topology_payload()),
        )

        try:
            for _ in range(self.generations):
                result = await self._run_generation(
                    task,
                    invoke_kwargs,
                    runtime_config=(runtime_config),
                )

                generation_results.append(result)

        except BaseException as exc:
            await events.aemit(
                f"Agent Elo {self.name} failed",
                stage="elo",
                phase="error",
                event_type="elo_failed",
                level="error",
                task=task,
                error=str(exc),
                elapsed_seconds=(perf_counter() - run_start),
            )

            raise

        final_result = {
            "task": task,
            "starting_generation": (starting_generation),
            "completed_generations": (len(generation_results)),
            "ending_generation": (self.generation_index),
            "generations": (generation_results),
            "standings": self.standings(),
            "population_size": len(self.players),
            "environment_state": str(self._environment_state_path()),
        }

        await events.aemit(
            f"Agent Elo {self.name} completed",
            stage="elo",
            phase="end",
            event_type="elo_completed",
            task=task,
            completed_generations=(len(generation_results)),
            ending_generation=(self.generation_index),
            standings=self.standings(),
            result={
                "ending_generation": (self.generation_index),
                "standings": (self.standings()),
            },
            elapsed_seconds=(perf_counter() - run_start),
        )

        return final_result

AgentSymposiumConfig dataclass

YAML-loadable configuration for an Agent Symposium environment.

Source code in src/ursa/environments/config.py

@dataclass(frozen=True)
class AgentSymposiumConfig:
    """YAML-loadable configuration for an Agent Symposium environment."""

    name: str
    group: str = "default"
    description: str | None = None
    inference_providers: dict[str, InferenceProviderConfig] = field(
        default_factory=dict
    )
    organizer: EnvironmentMemberConfig = field(
        default_factory=lambda: EnvironmentMemberConfig(
            name="organizer", role="Symposium organizer", agent="ChatAgent"
        )
    )
    members: list[EnvironmentMemberConfig] = field(default_factory=list)
    workspace: str | None = None
    defaults: dict[str, Any] = field(default_factory=dict)
    revision_rounds: int = 1

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> "AgentSymposiumConfig":
        raw = dict(data)
        providers = _inference_providers(raw.get("inference_providers") or {})
        raw["inference_providers"] = providers
        group = str(raw.get("group") or "default")
        if "organizer" in raw and isinstance(raw["organizer"], Mapping):
            raw["organizer"] = EnvironmentMemberConfig.from_mapping(
                raw["organizer"], providers, group
            )
        if "members" in raw:
            raw["members"] = [
                EnvironmentMemberConfig.from_mapping(member, providers, group)
                for member in raw["members"]
            ]
        return cls(**raw)

AgentSymposiumEnvironment

Bases: BaseEnvironment

Competitive/collaborative multi-agent symposium environment.

A symposium organizer dispatches the same complex problem to several members or nested teams. Members first work independently. The environment then sends all writeups to each member for critical review, explicitly instructing them not to modify reviewed work and to only view/run code as needed for assessment. Finally, each member revises its own work using the feedback it received and insights gained from reviewing others. The organizer synthesizes the final symposium report.

Source code in src/ursa/environments/agent_symposium.py

class AgentSymposiumEnvironment(BaseEnvironment):
    """Competitive/collaborative multi-agent symposium environment.

    A symposium organizer dispatches the same complex problem to several members
    or nested teams. Members first work independently. The environment then sends
    all writeups to each member for critical review, explicitly instructing them
    not to modify reviewed work and to only view/run code as needed for assessment.
    Finally, each member revises its own work using the feedback it received and
    insights gained from reviewing others. The organizer synthesizes the final
    symposium report.
    """

    def __init__(
        self,
        llm: BaseChatModel,
        *,
        config: AgentSymposiumConfig
        | Mapping[str, Any]
        | str
        | Path
        | None = None,
        name: str | None = None,
        group: str | None = None,
        organizer: EnvironmentMemberConfig | Mapping[str, Any] | None = None,
        members: list[EnvironmentMemberConfig | Mapping[str, Any]]
        | None = None,
        workspace: str | Path | None = None,
        revision_rounds: int | None = None,
        persist_members: bool = True,
        **kwargs: Any,
    ):
        symposium_config = self._coerce_config(
            config=config,
            name=name,
            group=group,
            organizer=organizer,
            members=members,
            workspace=workspace,
            revision_rounds=revision_rounds,
        )
        super().__init__(
            llm,
            name=symposium_config.name,
            group=symposium_config.group,
            workspace=symposium_config.workspace or workspace,
            persist_members=persist_members,
            **kwargs,
        )
        self.config = symposium_config
        self.members = {
            member.name: self._build_symposium_participant(member)
            for member in self.config.members
        }
        self.member_configs = {
            member.name: member for member in self.config.members
        }
        self.organizer = self._build_symposium_organizer(self.config.organizer)

    def _events(self, config: Mapping[str, Any] | None) -> EnvironmentEvents:
        return EnvironmentEvents(
            environment=self.name,
            config=config,
            environment_type="agent_symposium",
            environment_id=self.name,
            path=[self.name],
        )

    def _source(self, name: str, *, kind: str = "agent") -> dict[str, Any]:
        return {
            "id": f"{self.name}.{name}",
            "name": name,
            "kind": kind,
            "path": [self.name, name],
        }

    def _environment_source(self) -> dict[str, Any]:
        return {
            "id": self.name,
            "name": self.name,
            "kind": "environment",
            "path": [self.name],
        }

    def _member_node(self, member: EnvironmentMemberConfig) -> dict[str, Any]:
        member_obj = self.members.get(member.name)
        kind = (
            "environment"
            if isinstance(member_obj, BaseEnvironment)
            else "agent"
        )
        return {
            "id": f"{self.name}.{member.name}",
            "name": member.name,
            "kind": kind,
            "role": member.role,
            "agent_class": member.agent,
            "path": [self.name, member.name],
        }

    def _topology_payload(self) -> dict[str, Any]:
        organizer = self.config.organizer
        organizer_kind = (
            "environment"
            if isinstance(self.organizer, BaseEnvironment)
            else "agent"
        )
        nodes = [
            {
                "id": f"{self.name}.{organizer.name}",
                "name": organizer.name,
                "kind": organizer_kind,
                "role": organizer.role,
                "agent_class": organizer.agent,
                "path": [self.name, organizer.name],
            },
            *[self._member_node(member) for member in self.config.members],
        ]
        edges = []
        for member in self.config.members:
            edges.append({
                "source": self.name,
                "target": f"{self.name}.{member.name}",
                "kind": "dispatches_to",
            })
            edges.append({
                "source": f"{self.name}.{member.name}",
                "target": f"{self.name}.{organizer.name}",
                "kind": "synthesized_by",
            })
        return {
            "kind": "agent_symposium",
            "name": self.name,
            "description": self.config.description,
            "revision_rounds": self.config.revision_rounds,
            "nodes": nodes,
            "edges": edges,
        }

    @classmethod
    def from_yaml(
        cls,
        path: str | Path,
        *,
        llm: BaseChatModel,
        **kwargs: Any,
    ) -> "AgentSymposiumEnvironment":
        return cls(llm=llm, config=load_symposium_config(path), **kwargs)

    def _coerce_config(
        self,
        *,
        config: AgentSymposiumConfig | Mapping[str, Any] | str | Path | None,
        name: str | None,
        group: str | None,
        organizer: EnvironmentMemberConfig | Mapping[str, Any] | None,
        members: list[EnvironmentMemberConfig | Mapping[str, Any]] | None,
        workspace: str | Path | None,
        revision_rounds: int | None,
    ) -> AgentSymposiumConfig:
        if isinstance(config, (str, Path)):
            base = load_symposium_config(config)
        elif isinstance(config, Mapping):
            base = AgentSymposiumConfig.from_mapping(config)
        elif isinstance(config, AgentSymposiumConfig):
            base = config
        else:
            organizer_cfg = self._coerce_member(
                organizer
                or {
                    "name": "organizer",
                    "role": "Symposium organizer / final synthesizer",
                    "agent": "ChatAgent",
                }
            )
            member_cfgs = [self._coerce_member(m) for m in (members or [])]
            base = AgentSymposiumConfig(
                name=name or "agent_symposium",
                group=group or "default",
                organizer=organizer_cfg,
                members=member_cfgs,
                workspace=str(workspace) if workspace else None,
                revision_rounds=revision_rounds or 1,
            )
        return AgentSymposiumConfig(
            name=name or base.name,
            group=group or base.group,
            description=base.description,
            organizer=base.organizer,
            members=base.members,
            workspace=str(workspace) if workspace else base.workspace,
            defaults=base.defaults,
            revision_rounds=revision_rounds
            if revision_rounds is not None
            else base.revision_rounds,
        )

    @staticmethod
    def _coerce_member(
        member: EnvironmentMemberConfig | Mapping[str, Any],
    ) -> EnvironmentMemberConfig:
        if isinstance(member, EnvironmentMemberConfig):
            return member
        return EnvironmentMemberConfig.from_mapping(member)

    def _build_symposium_participant(
        self, member: EnvironmentMemberConfig
    ) -> Any:
        """Instantiate a symposium participant in its own child workspace.

        Symposium member names are stable URSA agent names. This lets users
        reuse existing named agents directly from YAML instead of creating new
        ``<symposium>_<member>`` checkpoints.
        """
        return self._build_symposium_agent(
            member,
            workspace=self._member_workspace(member.name),
            agent_name=member.name,
        )

    def _build_symposium_organizer(
        self, member: EnvironmentMemberConfig
    ) -> Any:
        """Instantiate the organizer in the parent symposium workspace."""
        return self._build_symposium_agent(
            member,
            workspace=self.workspace,
            agent_name=member.name,
        )

    def _build_symposium_agent(
        self,
        member: EnvironmentMemberConfig,
        *,
        workspace: Path,
        agent_name: str,
    ) -> Any:
        cls = load_object(member.agent)
        llm = make_llm(self.llm, member.model)
        kwargs = dict(member.config or {})
        kwargs.setdefault("workspace", workspace)
        kwargs.setdefault("group", self.group)
        if isinstance(cls, type) and issubclass(cls, BaseEnvironment):
            kwargs.setdefault("name", member.name)
            kwargs.setdefault("persist_members", self.persist_members)
        elif self.persist_members:
            kwargs.setdefault("agent_name", agent_name)
        return cls(llm=llm, **kwargs)

    def _member_workspace_roster(self) -> str:
        return (
            "\n".join(
                f"- {member.name}: {member.name}/ "
                f"({self._member_workspace(member.name)})"
                for member in self.config.members
            )
            or "No symposium member workspaces configured."
        )

    def _member_roster(self) -> str:
        return (
            "\n".join(
                f"- {member.name}: {member.role} ({member.agent})"
                for member in self.config.members
            )
            or "No symposium members configured."
        )

    def _initial_prompt(
        self, member: EnvironmentMemberConfig, task: str
    ) -> str:
        extra = (
            f"\n\nMember-specific guidance:\n{member.prompt}"
            if member.prompt
            else ""
        )
        return (
            f"You are symposium participant '{member.name}' with role: {member.role}.\n"
            "Work independently on the complex problem below. Produce a detailed, "
            "self-contained writeup of your approach, methods, assumptions, files or "
            "commands used, findings, uncertainty, and reproducibility instructions.\n"
            "Your work will be reviewed by other symposium participants. Clean, "
            "compact organization, documentation, and reproducibility are important "
            "parts of how your contribution will be assessed.\n"
            f"{extra}\n\nProblem:\n{task}"
        )

    def _review_prompt(
        self,
        reviewer: EnvironmentMemberConfig,
        task: str,
        writeups: Mapping[str, str],
    ) -> str:
        writeup_block = "\n\n".join(
            f"## Writeup from {name}\n{writeup}"
            for name, writeup in writeups.items()
        )
        return (
            f"You are symposium reviewer '{reviewer.name}' with role: {reviewer.role}.\n"
            "Read all submitted writeups, including your own if present. Assess the "
            "quality of the findings, compare methods and conclusions, and write a "
            "fair but critical review for each writeup. Discuss strengths, weaknesses, "
            "reproducibility, evidence quality, missing checks, and concrete ways to "
            "improve the solution.\n\n"
            "For this review phase, your workspace is the parent symposium "
            "workspace. Participant artifacts are available in these child "
            "workspace directories, using the relative paths shown:\n"
            f"{self._member_workspace_roster()}\n\n"
            "Important review constraints:\n"
            "- Do not change, edit, overwrite, or reorganize any work you are reviewing.\n"
            "- If code or files are referenced, you may inspect or run them only to assess "
            "correctness/reproducibility.\n"
            "- Make clear which findings are well supported and which are speculative.\n"
            "- Your own work will also be judged on clarity, compact organization, "
            "documentation, and reproducibility.\n\n"
            f"Original problem:\n{task}\n\n"
            f"Submitted writeups:\n{writeup_block}"
        )

    def _revision_prompt(
        self,
        member: EnvironmentMemberConfig,
        task: str,
        own_writeup: str,
        all_writeups: Mapping[str, str],
        reviews: Mapping[str, str],
        round_index: int,
    ) -> str:
        other_writeups = "\n\n".join(
            f"## {name}\n{writeup}" for name, writeup in all_writeups.items()
        )
        review_block = "\n\n".join(
            f"## Review from {name}\n{review}"
            for name, review in reviews.items()
        )
        return (
            f"You are symposium participant '{member.name}' revising your own work "
            f"after review round {round_index}.\n"
            "Use the feedback you received and anything you learned from reviewing "
            "other submissions to improve your own solution. You may change only your "
            "own work/artifacts. Do not modify the work of other symposium members.\n"
            "Return a revised detailed writeup with clear improvements, evidence, "
            "limitations, and reproducibility instructions.\n\n"
            f"Original problem:\n{task}\n\n"
            f"Your previous writeup:\n{own_writeup}\n\n"
            f"All writeups you saw:\n{other_writeups}\n\n"
            f"Reviews from symposium members:\n{review_block}"
        )

    def _synthesis_prompt(
        self,
        task: str,
        writeups: Mapping[str, str],
        reviews: Mapping[str, str],
    ) -> str:
        description = (
            f"\nSymposium description: {self.config.description}\n"
            if self.config.description
            else ""
        )
        writeup_block = "\n\n".join(
            f"## Final writeup from {name}\n{writeup}"
            for name, writeup in writeups.items()
        )
        review_block = "\n\n".join(
            f"## Review by {name}\n{review}" for name, review in reviews.items()
        )
        return (
            "You are the organizer of an URSA Agent Symposium. Synthesize the final "
            "results for the user after independent work, peer review, and revision. "
            "Compare participant outputs, identify consensus and disagreement, judge "
            "evidence quality, and provide a final recommendation or solution.\n"
            f"{description}\n"
            "Symposium members:\n"
            f"{self._member_roster()}\n\n"
            "The organizer workspace is the parent symposium workspace. Final "
            "member artifacts are available in these child workspace directories:\n"
            f"{self._member_workspace_roster()}\n\n"
            f"Original problem:\n{task}\n\n"
            f"Final participant writeups:\n{writeup_block}\n\n"
            f"Peer reviews:\n{review_block}"
        )

    def _invoke(
        self, inputs: Mapping[str, Any], **config: Any
    ) -> dict[str, Any]:
        return self._run_ainvoke_from_sync(inputs, **config)

    async def _member_writeup(
        self,
        member_config: EnvironmentMemberConfig,
        prompt: str,
        invoke_kwargs: Mapping[str, Any],
        events: EnvironmentEvents,
        *,
        workspace: Path | None = None,
        event_prefix: str,
        phase_name: str,
        round_index: int | None = None,
    ) -> tuple[str, str]:
        member_name = member_config.name
        source = self._source(member_name)
        start = perf_counter()
        await events.aemit(
            f"{member_name} started {phase_name}",
            stage="symposium",
            phase=phase_name,
            event_type=f"{event_prefix}_started",
            source=source,
            target=self._environment_source(),
            member=member_name,
            round_index=round_index,
            prompt=prompt,
        )
        member = self.members[member_name]
        try:
            result = await self._invoke_member_with_workspace_async(
                member, prompt, workspace=workspace, **invoke_kwargs
            )
        except BaseException as exc:
            await events.aemit(
                f"{member_name} failed {phase_name}",
                stage="symposium",
                phase=phase_name,
                event_type=f"{event_prefix}_failed",
                level="error",
                source=source,
                target=self._environment_source(),
                member=member_name,
                round_index=round_index,
                error=str(exc),
                elapsed_seconds=perf_counter() - start,
            )
            raise
        text = result_to_text(result)
        await events.aemit(
            f"{member_name} completed {phase_name}",
            stage="symposium",
            phase=phase_name,
            event_type=f"{event_prefix}_completed",
            source=source,
            target=self._environment_source(),
            member=member_name,
            round_index=round_index,
            result=text,
            elapsed_seconds=perf_counter() - start,
        )
        return member_name, text

    async def _invoke_member_with_workspace_async(
        self,
        member: Any,
        prompt: str,
        *,
        workspace: Path | None = None,
        **kwargs: Any,
    ) -> Any:
        """Invoke a member, optionally using a temporary workspace.

        Participants normally work in their own child workspaces. During review,
        reviewers are temporarily run from the parent symposium workspace so
        file tools can inspect every participant's child workspace. The original
        workspace is restored after the invocation.
        """
        if workspace is None or not hasattr(member, "workspace"):
            return await self._invoke_member_async(member, prompt, **kwargs)

        original_workspace = getattr(member, "workspace")
        setattr(member, "workspace", Path(workspace))
        Path(workspace).mkdir(parents=True, exist_ok=True)
        try:
            return await self._invoke_member_async(member, prompt, **kwargs)
        finally:
            setattr(member, "workspace", original_workspace)

    async def _ainvoke(
        self, inputs: Mapping[str, Any], **config: Any
    ) -> dict[str, Any]:
        task = str(inputs.get("task") or inputs.get("prompt") or inputs)
        if not self.config.members:
            raise ValueError(
                "AgentSymposiumEnvironment requires at least one member."
            )

        runtime_config = runnable_config_from_kwargs(config)
        events = self._events(runtime_config)
        invoke_kwargs = invocation_kwargs(config)
        start = perf_counter()
        await events.aemit(
            f"Agent symposium {self.name} started",
            stage="symposium",
            phase="start",
            event_type="symposium_started",
            task=task,
            topology=self._topology_payload(),
        )
        await events.aemit(
            f"Agent symposium {self.name} topology declared",
            stage="symposium",
            phase="topology",
            event_type="topology_declared",
            topology=self._topology_payload(),
        )
        try:
            await events.aemit(
                "Initial symposium work started",
                stage="symposium",
                phase="initial_work",
                event_type="symposium_phase_started",
                task=task,
            )
            initial_pairs = await asyncio.gather(*[
                self._member_writeup(
                    member_config,
                    self._initial_prompt(member_config, task),
                    invoke_kwargs,
                    events,
                    event_prefix="initial_work",
                    phase_name="initial_work",
                )
                for member_config in self.config.members
            ])
            initial_writeups = dict(initial_pairs)
            await events.aemit(
                "Initial symposium work completed",
                stage="symposium",
                phase="initial_work",
                event_type="symposium_phase_completed",
                result=initial_writeups,
            )

            current_writeups = dict(initial_writeups)
            review_rounds: list[dict[str, str]] = []
            latest_reviews: dict[str, str] = {}
            reviewer_configs = [
                member for member in self.config.members if member.reviewer
            ]

            for round_index in range(
                1, max(1, self.config.revision_rounds) + 1
            ):
                await events.aemit(
                    f"Review round {round_index} started",
                    stage="symposium",
                    phase="review",
                    event_type="review_round_started",
                    round_index=round_index,
                    writeups=current_writeups,
                )
                review_pairs = await asyncio.gather(*[
                    self._member_writeup(
                        reviewer_config,
                        self._review_prompt(
                            reviewer_config, task, current_writeups
                        ),
                        invoke_kwargs,
                        events,
                        event_prefix="review",
                        phase_name="review",
                        round_index=round_index,
                        workspace=self.workspace,
                    )
                    for reviewer_config in reviewer_configs
                ])
                round_reviews = dict(review_pairs)
                latest_reviews = round_reviews
                review_rounds.append(round_reviews)
                await events.aemit(
                    f"Review round {round_index} completed",
                    stage="symposium",
                    phase="review",
                    event_type="review_round_completed",
                    round_index=round_index,
                    reviews=round_reviews,
                )

                await events.aemit(
                    f"Revision round {round_index} started",
                    stage="symposium",
                    phase="revision",
                    event_type="revision_round_started",
                    round_index=round_index,
                    reviews=round_reviews,
                )
                revision_pairs = await asyncio.gather(*[
                    self._member_writeup(
                        member_config,
                        self._revision_prompt(
                            member_config,
                            task,
                            current_writeups[member_config.name],
                            current_writeups,
                            round_reviews,
                            round_index,
                        ),
                        invoke_kwargs,
                        events,
                        event_prefix="revision",
                        phase_name="revision",
                        round_index=round_index,
                    )
                    for member_config in self.config.members
                ])
                current_writeups = dict(revision_pairs)
                await events.aemit(
                    f"Revision round {round_index} completed",
                    stage="symposium",
                    phase="revision",
                    event_type="revision_round_completed",
                    round_index=round_index,
                    writeups=current_writeups,
                )

            synthesis_prompt = self._synthesis_prompt(
                task, current_writeups, latest_reviews
            )
            synth_start = perf_counter()
            await events.aemit(
                "Symposium synthesis started",
                stage="symposium",
                phase="synthesis",
                event_type="synthesis_started",
                source=self._source(self.config.organizer.name),
                task=task,
                prompt=synthesis_prompt,
            )
            organizer_result = await self._invoke_member_async(
                self.organizer,
                synthesis_prompt,
                **invoke_kwargs,
            )
            final = result_to_text(organizer_result)
            await events.aemit(
                "Symposium synthesis completed",
                stage="symposium",
                phase="synthesis",
                event_type="synthesis_completed",
                source=self._source(self.config.organizer.name),
                result=final,
                elapsed_seconds=perf_counter() - synth_start,
            )
            result = {
                "task": task,
                "initial_writeups": initial_writeups,
                "review_rounds": review_rounds,
                "reviews": latest_reviews,
                "final_writeups": current_writeups,
                "organizer_result": organizer_result,
                "final": final,
            }
        except BaseException as exc:
            await events.aemit(
                f"Agent symposium {self.name} failed",
                stage="symposium",
                phase="error",
                event_type="symposium_failed",
                level="error",
                task=task,
                error=str(exc),
                elapsed_seconds=perf_counter() - start,
            )
            raise
        await events.aemit(
            f"Agent symposium {self.name} completed",
            stage="symposium",
            phase="end",
            event_type="symposium_completed",
            task=task,
            result=result["final"],
            elapsed_seconds=perf_counter() - start,
        )
        return result

AgentTeamConfig dataclass

YAML-loadable configuration for an Agent Team environment.

Source code in src/ursa/environments/config.py

@dataclass(frozen=True)
class AgentTeamConfig:
    """YAML-loadable configuration for an Agent Team environment."""

    name: str
    group: str = "default"
    description: str | None = None
    inference_providers: dict[str, InferenceProviderConfig] = field(
        default_factory=dict
    )
    pi: EnvironmentMemberConfig = field(
        default_factory=lambda: EnvironmentMemberConfig(
            name="pi", role="Principal investigator", agent="ExecutionAgent"
        )
    )
    members: list[EnvironmentMemberConfig] = field(default_factory=list)
    workspace: str | None = None
    defaults: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> "AgentTeamConfig":
        raw = dict(data)
        providers = _inference_providers(raw.get("inference_providers") or {})
        raw["inference_providers"] = providers
        group = str(raw.get("group") or "default")
        if "pi" in raw and isinstance(raw["pi"], Mapping):
            raw["pi"] = EnvironmentMemberConfig.from_mapping(
                raw["pi"], providers, group
            )
        if "members" in raw:
            raw["members"] = [
                EnvironmentMemberConfig.from_mapping(member, providers, group)
                for member in raw["members"]
            ]
        return cls(**raw)

AgentTeamEnvironment

Bases: BaseEnvironment

Hierarchical multi-agent team coordinated by a PI agent.

The PI is user-facing. Team members are exposed to the PI as tools, one tool per member, so the PI can plan, delegate, compare returned work, ask follow-up questions, and synthesize a final answer. The environment itself exposes a normal invoke method and can therefore be nested inside other environments such as an Agent Symposium.

Source code in src/ursa/environments/agent_team.py

class AgentTeamEnvironment(BaseEnvironment):
    """Hierarchical multi-agent team coordinated by a PI agent.

    The PI is user-facing. Team members are exposed to the PI as tools, one tool
    per member, so the PI can plan, delegate, compare returned work, ask follow-up
    questions, and synthesize a final answer. The environment itself exposes a
    normal ``invoke`` method and can therefore be nested inside other environments
    such as an Agent Symposium.
    """

    def __init__(
        self,
        llm: BaseChatModel,
        *,
        config: AgentTeamConfig | Mapping[str, Any] | str | Path | None = None,
        name: str | None = None,
        group: str | None = None,
        pi: EnvironmentMemberConfig | Mapping[str, Any] | None = None,
        members: list[EnvironmentMemberConfig | Mapping[str, Any]]
        | None = None,
        workspace: str | Path | None = None,
        persist_members: bool = True,
        trace_delegation: bool = True,
        trace_character_limit: int = 4000,
        **kwargs: Any,
    ):
        team_config = self._coerce_config(
            config=config,
            name=name,
            group=group,
            pi=pi,
            members=members,
            workspace=workspace,
        )
        super().__init__(
            llm,
            name=team_config.name,
            group=team_config.group,
            workspace=team_config.workspace or workspace,
            persist_members=persist_members,
            **kwargs,
        )
        self.config = team_config
        self.trace_delegation = trace_delegation
        self.trace_character_limit = trace_character_limit
        self.members = {
            member.name: self._build_team_member(member)
            for member in self.config.members
        }
        self.member_configs = {
            member.name: member for member in self.config.members
        }
        self.pi = self._build_pi()

    @classmethod
    def from_yaml(
        cls,
        path: str | Path,
        *,
        llm: BaseChatModel,
        **kwargs: Any,
    ) -> "AgentTeamEnvironment":
        return cls(llm=llm, config=load_team_config(path), **kwargs)

    def _coerce_config(
        self,
        *,
        config: AgentTeamConfig | Mapping[str, Any] | str | Path | None,
        name: str | None,
        group: str | None,
        pi: EnvironmentMemberConfig | Mapping[str, Any] | None,
        members: list[EnvironmentMemberConfig | Mapping[str, Any]] | None,
        workspace: str | Path | None,
    ) -> AgentTeamConfig:
        if isinstance(config, (str, Path)):
            base = load_team_config(config)
        elif isinstance(config, Mapping):
            base = AgentTeamConfig.from_mapping(config)
        elif isinstance(config, AgentTeamConfig):
            base = config
        else:
            pi_cfg = self._coerce_member(
                pi
                or {
                    "name": "pi",
                    "role": "Principal investigator / team lead",
                    "agent": "ExecutionAgent",
                }
            )
            member_cfgs = [self._coerce_member(m) for m in (members or [])]
            base = AgentTeamConfig(
                name=name or "agent_team",
                group=group or "default",
                pi=pi_cfg,
                members=member_cfgs,
                workspace=str(workspace) if workspace else None,
            )
        return AgentTeamConfig(
            name=name or base.name,
            group=group or base.group,
            description=base.description,
            pi=base.pi,
            members=base.members,
            workspace=str(workspace) if workspace else base.workspace,
            defaults=base.defaults,
        )

    @staticmethod
    def _coerce_member(
        member: EnvironmentMemberConfig | Mapping[str, Any],
    ) -> EnvironmentMemberConfig:
        if isinstance(member, EnvironmentMemberConfig):
            return member
        return EnvironmentMemberConfig.from_mapping(member)

    def _build_team_member(self, member: EnvironmentMemberConfig) -> Any:
        """Instantiate a team member in the shared team workspace.

        Team member names are stable URSA agent names. This lets users reuse
        existing named agents directly from YAML, e.g. ``team_nif_expert`` rather
        than creating a new ``<team>_<member>`` checkpoint. All team members see
        the same team workspace so they can collaborate through shared files.
        """
        return self._build_named_team_agent(member, agent_name=member.name)

    def _build_team_pi(self, member: EnvironmentMemberConfig) -> Any:
        """Instantiate the PI in the shared team workspace.

        The PI also uses its configured name directly. If that named PI does not
        yet exist, BaseAgent will create its checkpoint under the usual URSA agent
        cache for the configured group.
        """
        return self._build_named_team_agent(member, agent_name=member.name)

    def _build_named_team_agent(
        self,
        member: EnvironmentMemberConfig,
        *,
        agent_name: str,
    ) -> Any:
        cls = load_object(member.agent)
        llm = make_llm(self.llm, member.model)
        kwargs = dict(member.config or {})
        kwargs.setdefault("workspace", self.workspace)
        kwargs.setdefault("group", self.group)
        if isinstance(cls, type) and issubclass(cls, BaseEnvironment):
            kwargs.setdefault("name", member.name)
            kwargs.setdefault("persist_members", self.persist_members)
        elif self.persist_members:
            kwargs.setdefault("agent_name", agent_name)
        return cls(llm=llm, **kwargs)

    def _build_pi(self) -> Any:
        pi_config = self.config.pi
        cls_kwargs = dict(pi_config.config or {})
        delegation_tools = [
            self._make_delegation_tool(member_config)
            for member_config in self.config.members
        ]

        # ExecutionAgent explicitly supports `extra_tools`, making it the preferred
        # PI implementation. Other AgentWithTools-style agents can receive tools
        # after construction via `add_tool` if they expose that method.
        pass_extra_tools = (
            pi_config.agent.endswith("ExecutionAgent")
            or pi_config.agent == "ExecutionAgent"
        )
        if pass_extra_tools:
            cls_kwargs.setdefault("extra_tools", delegation_tools)

        pi_member = EnvironmentMemberConfig(
            name=pi_config.name,
            role=pi_config.role,
            agent=pi_config.agent,
            model=pi_config.model,
            config=cls_kwargs,
            prompt=pi_config.prompt,
        )
        pi_agent = self._build_team_pi(pi_member)
        if (
            not pass_extra_tools
            and delegation_tools
            and hasattr(pi_agent, "add_tool")
        ):
            pi_agent.add_tool(delegation_tools)
        elif not pass_extra_tools and delegation_tools:
            raise TypeError(
                f"Configured PI agent {pi_config.agent!r} cannot accept delegation tools. "
                "Use ExecutionAgent or another AgentWithTools-compatible agent."
            )
        return pi_agent

    def _events(self, config: Mapping[str, Any] | None) -> EnvironmentEvents:
        return EnvironmentEvents(
            environment=self.name,
            config=config,
            environment_type="agent_team",
            environment_id=self.name,
            path=[self.name],
        )

    def _source(self, name: str, *, kind: str = "agent") -> dict[str, Any]:
        return {
            "id": f"{self.name}.{name}",
            "name": name,
            "kind": kind,
            "path": [self.name, name],
        }

    def _member_runtime_config(
        self,
        base_config: Mapping[str, Any] | None,
        member: EnvironmentMemberConfig,
    ) -> dict[str, Any] | None:
        """Attach stable team-member identity to nested agent/tool events."""
        if base_config is None:
            return None
        member_id = f"{self.name}.{member.name}"
        merged = dict(base_config)
        base_metadata = merged.get("metadata")
        metadata = (
            dict(base_metadata) if isinstance(base_metadata, Mapping) else {}
        )
        metadata.update({
            "environment_id": self.name,
            "environment_member": member.name,
            "environment_member_id": member_id,
            "environment_member_role": member.role,
            "environment_member_path": [self.name, member.name],
            "agent": member.name,
            "agent_id": member_id,
        })
        merged["metadata"] = metadata

        base_tags = merged.get("tags")
        if isinstance(base_tags, str):
            tags = [base_tags]
        else:
            tags = list(base_tags) if base_tags else []
        for tag in (member.name, member_id, "environment_member"):
            if tag not in tags:
                tags.append(tag)
        merged["tags"] = tags
        return merged

    def _member_node(self, member: EnvironmentMemberConfig) -> dict[str, Any]:
        kind = (
            "environment"
            if isinstance(self.members.get(member.name), BaseEnvironment)
            else "agent"
        )
        return {
            "id": f"{self.name}.{member.name}",
            "name": member.name,
            "kind": kind,
            "role": member.role,
            "agent_class": member.agent,
            "path": [self.name, member.name],
        }

    def _topology_payload(self) -> dict[str, Any]:
        pi = self.config.pi
        nodes = [
            {
                "id": f"{self.name}.{pi.name}",
                "name": pi.name,
                "kind": "agent",
                "role": pi.role,
                "agent_class": pi.agent,
                "path": [self.name, pi.name],
            },
            *[self._member_node(member) for member in self.config.members],
        ]
        return {
            "kind": "agent_team",
            "name": self.name,
            "description": self.config.description,
            "nodes": nodes,
            "edges": [
                {
                    "source": f"{self.name}.{pi.name}",
                    "target": f"{self.name}.{member.name}",
                    "kind": "delegates_to",
                }
                for member in self.config.members
            ],
        }

    def _make_delegation_tool(
        self, member: EnvironmentMemberConfig
    ) -> StructuredTool:
        member_name = member.name
        role = member.role
        tool_name = f"delegate_to_{_slug_tool_name(member_name)}"

        def delegate(task: str, context: str = "") -> str:
            prompt = self._delegation_prompt(member, task=task, context=context)
            runtime_config = current_environment_config()
            events = self._events(runtime_config)
            source = self._source(self.config.pi.name)
            target = self._source(member_name)
            start = perf_counter()
            events.emit(
                f"Delegating to {member_name}",
                stage="delegation",
                phase="start",
                event_type="delegation_started",
                source=source,
                target=target,
                task=task,
                context=context,
                prompt=prompt,
            )
            self._trace_delegation(
                f"PI -> {member_name}",
                f"Task:\n{task}\n\nContext:\n{context or 'No additional context provided.'}",
            )
            try:
                member_config = self._member_runtime_config(
                    runtime_config, member
                )
                kwargs = {"config": member_config} if member_config else {}
                result = self.members[member_name].invoke(prompt, **kwargs)
            except BaseException as exc:
                events.emit(
                    f"Delegation to {member_name} failed",
                    stage="delegation",
                    phase="error",
                    event_type="delegation_failed",
                    level="error",
                    source=source,
                    target=target,
                    task=task,
                    context=context,
                    error=str(exc),
                    elapsed_seconds=perf_counter() - start,
                )
                raise
            text = result_to_text(result)
            events.emit(
                f"Delegation to {member_name} completed",
                stage="delegation",
                phase="end",
                event_type="delegation_completed",
                source=target,
                target=source,
                task=task,
                context=context,
                result=text,
                elapsed_seconds=perf_counter() - start,
            )
            self._trace_delegation(f"{member_name} -> PI", text)
            return text

        async def adelegate(task: str, context: str = "") -> str:
            prompt = self._delegation_prompt(member, task=task, context=context)
            runtime_config = current_environment_config()
            events = self._events(runtime_config)
            source = self._source(self.config.pi.name)
            target = self._source(member_name)
            start = perf_counter()
            await events.aemit(
                f"Delegating to {member_name}",
                stage="delegation",
                phase="start",
                event_type="delegation_started",
                source=source,
                target=target,
                task=task,
                context=context,
                prompt=prompt,
            )
            self._trace_delegation(
                f"PI -> {member_name}",
                f"Task:\n{task}\n\nContext:\n{context or 'No additional context provided.'}",
            )
            try:
                member_config = self._member_runtime_config(
                    runtime_config, member
                )
                kwargs = {"config": member_config} if member_config else {}
                result = await self._invoke_member_async(
                    self.members[member_name], prompt, **kwargs
                )
            except BaseException as exc:
                await events.aemit(
                    f"Delegation to {member_name} failed",
                    stage="delegation",
                    phase="error",
                    event_type="delegation_failed",
                    level="error",
                    source=source,
                    target=target,
                    task=task,
                    context=context,
                    error=str(exc),
                    elapsed_seconds=perf_counter() - start,
                )
                raise
            text = result_to_text(result)
            await events.aemit(
                f"Delegation to {member_name} completed",
                stage="delegation",
                phase="end",
                event_type="delegation_completed",
                source=target,
                target=source,
                task=task,
                context=context,
                result=text,
                elapsed_seconds=perf_counter() - start,
            )
            self._trace_delegation(f"{member_name} -> PI", text)
            return text

        return StructuredTool.from_function(
            func=delegate,
            coroutine=adelegate,
            name=tool_name,
            description=(
                f"Delegate work to team member '{member_name}' ({role}). "
                "Use this when that member's specialty is relevant. Provide a "
                "self-contained task and any context needed for independent work."
            ),
            args_schema=DelegateInput,
        )

    def _trace_delegation(self, label: str, message: str) -> None:
        """Log a small, explicit delegation trace."""
        if not self.trace_delegation:
            return
        text = message
        if (
            self.trace_character_limit > 0
            and len(text) > self.trace_character_limit
        ):
            text = text[: self.trace_character_limit] + "\n... [truncated]"
        logger.info(f"\n[AgentTeam:{self.name}] {label}\n{text}\n")

    def _delegation_prompt(
        self,
        member: EnvironmentMemberConfig,
        *,
        task: str,
        context: str,
    ) -> str:
        guidance = (
            f"\n\nMember-specific guidance:\n{member.prompt}"
            if member.prompt
            else ""
        )
        return (
            f"You are acting as team member '{member.name}' with role: {member.role}.\n"
            "You have been delegated a task by the team PI. Complete the delegated "
            "task thoroughly, using your available tools when appropriate, and return "
            "a clear writeup of methods, evidence, outputs, limitations, and any files "
            "created or commands needed to reproduce the work.\n"
            f"{guidance}\n\n"
            f"Overall context:\n{context or 'No additional context provided.'}\n\n"
            f"Delegated task:\n{task}"
        )

    def _team_roster(self) -> str:
        if not self.config.members:
            return "No team members are configured. Solve directly as PI."
        return "\n".join(
            f"- {member.name}: {member.role} ({member.agent})"
            for member in self.config.members
        )

    def _pi_prompt(self, task: str) -> str:
        description = (
            f"\nTeam description: {self.config.description}\n"
            if self.config.description
            else ""
        )
        pi_extra = (
            f"\nPI-specific guidance:\n{self.config.pi.prompt}\n"
            if self.config.pi.prompt
            else ""
        )
        return (
            "You are the PI/team leader of a hierarchical Agent Team. "
            "You are the user-facing coordinator responsible for satisfying the "
            "user's overall goal. Formulate an approach, decide which team members "
            "to assign work to, call member-delegation tools as needed, review their "
            "returns critically, request follow-up work if necessary, and synthesize "
            "a final answer for the user.\n"
            "Do not claim a member completed work unless you have delegated it and "
            "reviewed the result. You are personally responsible for final "
            "organization and presentation: integrate the delegated work into one "
            "clean, coherent, easily shareable answer with a clear structure, "
            "actionable conclusions, supporting evidence, limitations, and "
            "reproducibility details where relevant.\n"
            f"{description}{pi_extra}\n"
            "Available team members:\n"
            f"{self._team_roster()}\n\n"
            f"User task:\n{task}"
        )

    def _invoke(self, inputs: Mapping[str, Any], **config: Any) -> Any:
        return self._run_ainvoke_from_sync(inputs, **config)

    async def _ainvoke(self, inputs: Mapping[str, Any], **config: Any) -> Any:
        task = str(inputs.get("task") or inputs.get("prompt") or inputs)
        runtime_config = runnable_config_from_kwargs(config)
        events = self._events(runtime_config)
        token = bind_current_environment_config(runtime_config)
        start = perf_counter()
        await events.aemit(
            f"Agent team {self.name} started",
            stage="team",
            phase="start",
            event_type="team_started",
            task=task,
            topology=self._topology_payload(),
        )
        await events.aemit(
            f"Agent team {self.name} topology declared",
            stage="team",
            phase="topology",
            event_type="topology_declared",
            topology=self._topology_payload(),
        )
        try:
            pi_kwargs = invocation_kwargs(config)
            pi_runtime_config = self._member_runtime_config(
                runtime_config, self.config.pi
            )
            if pi_runtime_config is not None:
                pi_kwargs["config"] = pi_runtime_config
            result = await self._invoke_member_async(
                self.pi, self._pi_prompt(task), **pi_kwargs
            )
        except BaseException as exc:
            await events.aemit(
                f"Agent team {self.name} failed",
                stage="team",
                phase="error",
                event_type="team_failed",
                level="error",
                task=task,
                error=str(exc),
                elapsed_seconds=perf_counter() - start,
            )
            raise
        finally:
            reset_current_environment_config(token)
        await events.aemit(
            f"Agent team {self.name} completed",
            stage="team",
            phase="end",
            event_type="team_completed",
            task=task,
            result=result_to_text(result),
            elapsed_seconds=perf_counter() - start,
        )
        return result

BaseEnvironment

Bases: BaseWorkflow

Base class for multi-agent URSA environments.

Environments compose agents and/or other environments while exposing the same simple invoke surface used by workflows. They deliberately keep persistent configuration separate from agent graph checkpoints: environment definitions live under ~/.cache/ursa/<group>/environments/, while member agents use the shared ~/.cache/ursa/<group>/agents/<agent_name> persistence mechanism.

Source code in src/ursa/environments/base.py

class BaseEnvironment(BaseWorkflow):
    """Base class for multi-agent URSA environments.

    Environments compose agents and/or other environments while exposing the same
    simple ``invoke`` surface used by workflows. They deliberately keep persistent
    configuration separate from agent graph checkpoints: environment definitions live
    under ``~/.cache/ursa/<group>/environments/``, while member agents use the
    shared ``~/.cache/ursa/<group>/agents/<agent_name>`` persistence mechanism.
    """

    def __init__(
        self,
        llm: BaseChatModel,
        *,
        name: str,
        group: str = "default",
        workspace: str | Path | None = None,
        persist_members: bool = True,
        **kwargs: Any,
    ):
        super().__init__(**kwargs)
        self.llm = llm
        self.name = name
        self.group = validate_group_name(group)
        self.workspace = Path(
            workspace
            or group_environments_dir(self.group) / "workspaces" / name
        )
        self.workspace.mkdir(parents=True, exist_ok=True)
        self.persist_members = persist_members

    def _normalize_inputs(self, inputs: InputLike) -> Mapping[str, Any]:
        if isinstance(inputs, str):
            return {"task": inputs}
        if isinstance(inputs, Mapping):
            return inputs
        raise TypeError(f"Unsupported input type: {type(inputs)}")

    def _member_workspace(self, member_name: str) -> Path:
        return self.workspace / member_name

    def _member_agent_name(self, member_name: str) -> str | None:
        if not self.persist_members:
            return None
        return f"{self.name}_{member_name}"

    def build_member(self, member: EnvironmentMemberConfig) -> Any:
        """Instantiate a configured member agent or nested environment.

        BaseAgent subclasses use ``agent_name`` for checkpoint persistence, while
        nested environments use ``name`` and manage their own member persistence.
        The distinction keeps teams usable as symposium members without requiring
        environment constructors to accept BaseAgent-specific keywords.
        """
        cls = load_object(member.agent)
        llm = make_llm(self.llm, member.model)
        kwargs = dict(member.config or {})
        kwargs.setdefault("workspace", self._member_workspace(member.name))
        kwargs.setdefault("group", self.group)
        if isinstance(cls, type) and issubclass(cls, BaseEnvironment):
            kwargs.setdefault("name", member.name)
            kwargs.setdefault("persist_members", self.persist_members)
        elif self.persist_members:
            kwargs.setdefault(
                "agent_name", self._member_agent_name(member.name)
            )
        return cls(llm=llm, **kwargs)

    def _run_ainvoke_from_sync(
        self, inputs: Mapping[str, Any], **config: Any
    ) -> Any:
        """Run this environment's async implementation for sync callers.

        Environments are natively async internally so nested agents/tools can use
        async-only implementations. The public ``invoke`` surface remains useful
        for scripts by creating an event loop at the outer boundary. If a loop is
        already running, callers must use ``await environment.ainvoke(...)``.
        """
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(self._ainvoke(inputs, **config))

        raise RuntimeError(
            "This environment uses async execution internally, but `.invoke()` "
            "was called from an async context. Use "
            "`await environment.ainvoke(...)` instead."
        )

    async def _invoke_member_async(
        self, member: Any, prompt: str, **kwargs: Any
    ) -> Any:
        """Invoke a member through its async API when available.

        BaseAgent and BaseEnvironment instances expose ``ainvoke``. Lightweight
        test doubles or custom sync-only members may expose only ``invoke``; run
        those in a worker thread so async environment phases remain non-blocking.
        """
        ainvoke = getattr(member, "ainvoke", None)
        if callable(ainvoke):
            result = ainvoke(prompt, **kwargs)
            if inspect.isawaitable(result):
                return await result
            return result

        invoke = getattr(member, "invoke")
        return await asyncio.to_thread(invoke, prompt, **kwargs)

build_member(member)

Instantiate a configured member agent or nested environment.

BaseAgent subclasses use agent_name for checkpoint persistence, while nested environments use name and manage their own member persistence. The distinction keeps teams usable as symposium members without requiring environment constructors to accept BaseAgent-specific keywords.

Source code in src/ursa/environments/base.py

def build_member(self, member: EnvironmentMemberConfig) -> Any:
    """Instantiate a configured member agent or nested environment.

    BaseAgent subclasses use ``agent_name`` for checkpoint persistence, while
    nested environments use ``name`` and manage their own member persistence.
    The distinction keeps teams usable as symposium members without requiring
    environment constructors to accept BaseAgent-specific keywords.
    """
    cls = load_object(member.agent)
    llm = make_llm(self.llm, member.model)
    kwargs = dict(member.config or {})
    kwargs.setdefault("workspace", self._member_workspace(member.name))
    kwargs.setdefault("group", self.group)
    if isinstance(cls, type) and issubclass(cls, BaseEnvironment):
        kwargs.setdefault("name", member.name)
        kwargs.setdefault("persist_members", self.persist_members)
    elif self.persist_members:
        kwargs.setdefault(
            "agent_name", self._member_agent_name(member.name)
        )
    return cls(llm=llm, **kwargs)

EnvironmentEventRecorder

Bases: BaseCallbackHandler

Record URSA structured progress events to a replayable JSONL file.

Source code in src/ursa/environments/visualization.py

class EnvironmentEventRecorder(BaseCallbackHandler):
    """Record URSA structured progress events to a replayable JSONL file."""

    def __init__(
        self,
        *,
        run_id: str,
        group: str = "default",
        environment_name: str,
        environment_type: str,
        run_dir: Path | None = None,
        event_name: str = DEFAULT_EVENT_NAME,
        max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
    ) -> None:
        self.run_id = run_id
        self.group = validate_group_name(group)
        self.environment_name = environment_name
        self.environment_type = environment_type
        self.event_name = event_name
        self.max_payload_chars = max_payload_chars
        if run_dir is None:
            self.paths = get_environment_run_paths(self.group, run_id)
        else:
            self.paths = EnvironmentRunPaths(
                run_dir=run_dir,
                manifest_path=run_dir / "manifest.json",
                events_path=run_dir / "events.jsonl",
                artifacts_dir=run_dir / "artifacts",
                logs_dir=run_dir / "logs",
            )
        ensure_environment_run_dirs(self.paths)
        self._lock = threading.Lock()
        self._seq = 0

    @property
    def config(self) -> RunnableConfig:
        return {
            "callbacks": [self],
            "metadata": {
                "environment_run_id": self.run_id,
                "environment_name": self.environment_name,
                "environment_type": self.environment_type,
                "group": self.group,
            },
            "tags": ["environment_run", self.environment_name],
        }

    def write_manifest(
        self,
        *,
        status: str,
        task: Any | None = None,
        error: str | None = None,
    ) -> None:
        existing: dict[str, Any] = {}
        if self.paths.manifest_path.exists():
            try:
                existing = json.loads(
                    self.paths.manifest_path.read_text(encoding="utf-8")
                )
            except Exception:
                existing = {}
        now = utc_now_rfc3339()
        manifest = {
            **existing,
            "schema_version": ENVIRONMENT_RUN_SCHEMA_VERSION,
            "run_id": self.run_id,
            "group": self.group,
            "environment_name": self.environment_name,
            "environment_type": self.environment_type,
            "status": status,
            "updated_at": now,
            "events_path": "events.jsonl",
            "artifacts_path": "artifacts",
            "logs_path": "logs",
        }
        manifest.setdefault("created_at", now)
        if task is not None:
            manifest["task_preview"] = _preview(task)
        if error:
            manifest["error"] = error
        self.paths.manifest_path.write_text(
            json.dumps(manifest, indent=2, ensure_ascii=False, default=str),
            encoding="utf-8",
        )

    def on_custom_event(
        self,
        name: str,
        data: Any,
        *,
        run_id,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> None:
        if name != self.event_name or not isinstance(data, Mapping):
            return
        event = self.normalize_event(data, tags=tags, metadata=metadata)
        self.append_event(event)

    def normalize_event(
        self,
        data: Mapping[str, Any],
        *,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        raw_payload_tags = data.get("tags")
        safe_data = _make_json_safe(data, max_chars=self.max_payload_chars)
        if not isinstance(safe_data, dict):
            safe_data = {"value": safe_data}
        event_type = _infer_event_type(safe_data)
        source = _source_from_payload(safe_data)
        target = safe_data.get("target")
        if isinstance(target, str):
            target = {"id": target, "name": target}
        elif isinstance(target, Mapping):
            target = dict(target)
        else:
            target = None
        if (
            target is None
            and safe_data.get("tool")
            and source.get("kind") != "tool"
        ):
            target = _tool_target_from_payload(safe_data)
        environment_name = str(
            safe_data.get("environment") or self.environment_name
        )
        return {
            "schema_version": ENVIRONMENT_EVENT_SCHEMA_VERSION,
            "event_id": new_event_id(),
            "seq": 0,
            "ts": utc_now_rfc3339(),
            "monotonic_timestamp_ns": safe_data.get(
                "monotonic_timestamp_ns", monotonic_ns()
            ),
            "run_id": self.run_id,
            "environment_id": safe_data.get("environment_id")
            or environment_name,
            "environment_name": environment_name,
            "environment_type": safe_data.get("environment_type")
            or self.environment_type,
            "event_type": event_type,
            "stage": safe_data.get("stage"),
            "phase": safe_data.get("phase"),
            "level": safe_data.get("level", "info"),
            "source": source,
            "target": target,
            "message": safe_data.get("message") or event_type,
            "payload": safe_data,
            "tags": _normalize_tags(tags) + _normalize_tags(raw_payload_tags),
            "metadata": _make_json_safe(
                metadata or {}, max_chars=self.max_payload_chars
            ),
        }

    def append_event(self, event: Mapping[str, Any]) -> dict[str, Any]:
        with self._lock:
            self._seq += 1
            record = dict(event)
            record["seq"] = self._seq
            with self.paths.events_path.open("a", encoding="utf-8") as f:
                f.write(
                    json.dumps(record, ensure_ascii=False, default=str) + "\n"
                )
                f.flush()
            return record

EnvironmentMemberConfig dataclass

Configuration for one agent or nested environment member.

YAML fields

name: stable member name used in prompts/tool names/persistence role: human-readable role or specialty agent: Python class path or URSA agent class name, e.g. ExecutionAgent model: optional ModelConfig-compatible mapping for this member config: kwargs passed to the agent/environment constructor prompt: optional extra role/system guidance included in delegated tasks reviewer: whether this member participates in symposium review phases

Source code in src/ursa/environments/config.py

@dataclass(frozen=True)
class EnvironmentMemberConfig:
    """Configuration for one agent or nested environment member.

    YAML fields:
      name: stable member name used in prompts/tool names/persistence
      role: human-readable role or specialty
      agent: Python class path or URSA agent class name, e.g. ExecutionAgent
      model: optional ModelConfig-compatible mapping for this member
      config: kwargs passed to the agent/environment constructor
      prompt: optional extra role/system guidance included in delegated tasks
      reviewer: whether this member participates in symposium review phases
    """

    name: str
    role: str = "Team member"
    agent: str = "ExecutionAgent"
    model: ModelConfig | None = None
    config: dict[str, Any] = field(default_factory=dict)
    prompt: str | None = None
    reviewer: bool = True

    @classmethod
    def from_mapping(
        cls,
        data: Mapping[str, Any],
        inference_providers: Mapping[str, InferenceProviderConfig]
        | None = None,
        group: str = "default",
    ) -> "EnvironmentMemberConfig":
        raw = dict(data)
        model = raw.get("model")
        if isinstance(model, Mapping):
            model_config = ModelConfig.model_validate(model)
            model_config = model_config.resolve_inference_provider(
                dict(inference_providers or {})
            )
            enforce_group_base_url_policy(model_config.base_url, group)
            raw["model"] = model_config
        return cls(**raw)

arun_with_visualization(environment, inputs, *, config=None, run_id=None, max_payload_chars=DEFAULT_MAX_PAYLOAD_CHARS, **kwargs) async

Run an environment asynchronously while recording visualization events.

Source code in src/ursa/environments/visualization.py

async def arun_with_visualization(
    environment: Any,
    inputs: InputLike,
    *,
    config: RunnableConfig | None = None,
    run_id: str | None = None,
    max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
    **kwargs: Any,
) -> Any:
    """Run an environment asynchronously while recording visualization events."""
    recorder = EnvironmentEventRecorder(
        run_id=run_id or new_run_id(),
        group=getattr(environment, "group", "default"),
        environment_name=getattr(
            environment, "name", type(environment).__name__
        ),
        environment_type=type(environment).__name__,
        max_payload_chars=max_payload_chars,
    )
    recorder.write_manifest(status="running", task=inputs)
    run_config = visualization_config(recorder, config)
    try:
        result = await environment.ainvoke(inputs, config=run_config, **kwargs)
    except BaseException as exc:
        recorder.write_manifest(status="failed", task=inputs, error=str(exc))
        raise
    recorder.write_manifest(status="succeeded", task=inputs)
    return result

elo_cache_dir(group, name)

Return the persistent configuration directory for a named Elo environment.

Source code in src/ursa/environments/config.py

def elo_cache_dir(
    group: str,
    name: str,
) -> Path:
    """Return the persistent configuration directory for a named Elo environment."""
    return group_environments_dir(group) / "agent_elo" / name

environment_run_recorder(environment, *, task=None, config=None, run_id=None, max_payload_chars=DEFAULT_MAX_PAYLOAD_CHARS)

Create a recorder and runnable config for an environment run.

Source code in src/ursa/environments/visualization.py

@contextmanager
def environment_run_recorder(
    environment: Any,
    *,
    task: Any | None = None,
    config: RunnableConfig | None = None,
    run_id: str | None = None,
    max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
) -> Iterator[tuple[EnvironmentEventRecorder, RunnableConfig]]:
    """Create a recorder and runnable config for an environment run."""
    recorder = EnvironmentEventRecorder(
        run_id=run_id or new_run_id(),
        group=getattr(environment, "group", "default"),
        environment_name=getattr(
            environment, "name", type(environment).__name__
        ),
        environment_type=type(environment).__name__,
        max_payload_chars=max_payload_chars,
    )
    recorder.write_manifest(status="running", task=task)
    try:
        yield recorder, visualization_config(recorder, config)
    except BaseException as exc:
        recorder.write_manifest(status="failed", task=task, error=str(exc))
        raise
    else:
        recorder.write_manifest(status="succeeded", task=task)

environment_runs_dir(group=None)

Return the directory that stores recorded environment visualization runs.

Source code in src/ursa/environments/visualization.py

def environment_runs_dir(group: str | None = None) -> Path:
    """Return the directory that stores recorded environment visualization runs."""
    return group_root_dir(validate_group_name(group)) / "environment_runs"

record_environment_run(environment, *, task=None, config=None, run_id=None, max_payload_chars=DEFAULT_MAX_PAYLOAD_CHARS)

Alias for environment_run_recorder for readable user code.

Source code in src/ursa/environments/visualization.py

@contextmanager
def record_environment_run(
    environment: Any,
    *,
    task: Any | None = None,
    config: RunnableConfig | None = None,
    run_id: str | None = None,
    max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
) -> Iterator[tuple[EnvironmentEventRecorder, RunnableConfig]]:
    """Alias for ``environment_run_recorder`` for readable user code."""
    with environment_run_recorder(
        environment,
        task=task,
        config=config,
        run_id=run_id,
        max_payload_chars=max_payload_chars,
    ) as value:
        yield value

run_with_visualization(environment, inputs, *, config=None, run_id=None, max_payload_chars=DEFAULT_MAX_PAYLOAD_CHARS, **kwargs)

Run an environment synchronously while recording visualization events.

Source code in src/ursa/environments/visualization.py

def run_with_visualization(
    environment: Any,
    inputs: InputLike,
    *,
    config: RunnableConfig | None = None,
    run_id: str | None = None,
    max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
    **kwargs: Any,
) -> Any:
    """Run an environment synchronously while recording visualization events."""
    recorder = EnvironmentEventRecorder(
        run_id=run_id or new_run_id(),
        group=getattr(environment, "group", "default"),
        environment_name=getattr(
            environment, "name", type(environment).__name__
        ),
        environment_type=type(environment).__name__,
        max_payload_chars=max_payload_chars,
    )
    recorder.write_manifest(status="running", task=inputs)
    run_config = visualization_config(recorder, config)
    try:
        result = environment.invoke(inputs, config=run_config, **kwargs)
    except BaseException as exc:
        recorder.write_manifest(status="failed", task=inputs, error=str(exc))
        raise
    recorder.write_manifest(status="succeeded", task=inputs)
    return result

save_elo_config(config, path=None)

Persist an Elo environment configuration.

Source code in src/ursa/environments/config.py

def save_elo_config(
    config: AgentEloConfig,
    path: str | Path | None = None,
) -> Path:
    """Persist an Elo environment configuration."""
    target = (
        Path(path).expanduser()
        if path
        else elo_cache_dir(
            config.group,
            config.name,
        )
        / "elo.yaml"
    )

    target.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    target.write_text(
        yaml.safe_dump(
            _dataclass_to_plain(config),
            sort_keys=False,
        ),
        encoding="utf-8",
    )

    return target

save_symposium_config(config, path=None)

Persist a symposium configuration under ~/.cache/ursa//environments by default.

Source code in src/ursa/environments/config.py

def save_symposium_config(
    config: AgentSymposiumConfig, path: str | Path | None = None
) -> Path:
    """Persist a symposium configuration under ~/.cache/ursa/<group>/environments by default."""
    target = (
        Path(path).expanduser()
        if path
        else symposium_cache_dir(config.group, config.name) / "symposium.yaml"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(
        yaml.safe_dump(_dataclass_to_plain(config), sort_keys=False),
        encoding="utf-8",
    )
    return target

save_team_config(config, path=None)

Persist a team configuration under ~/.cache/ursa//environments by default.

Source code in src/ursa/environments/config.py

def save_team_config(
    config: AgentTeamConfig, path: str | Path | None = None
) -> Path:
    """Persist a team configuration under ~/.cache/ursa/<group>/environments by default."""
    target = (
        Path(path).expanduser()
        if path
        else team_cache_dir(config.group, config.name) / "team.yaml"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(
        yaml.safe_dump(_dataclass_to_plain(config), sort_keys=False),
        encoding="utf-8",
    )
    return target

symposium_cache_dir(group, name)

Return the persistent configuration directory for a named symposium.

Source code in src/ursa/environments/config.py

def symposium_cache_dir(group: str, name: str) -> Path:
    """Return the persistent configuration directory for a named symposium."""
    return group_environments_dir(group) / "agent_symposia" / name

team_cache_dir(group, name)

Return the persistent configuration directory for a named team.

Source code in src/ursa/environments/config.py

def team_cache_dir(group: str, name: str) -> Path:
    """Return the persistent configuration directory for a named team."""
    return group_environments_dir(group) / "agent_teams" / name

agent_elo

AgentEloEnvironment

Bases: BaseEnvironment

Evolutionary pairwise agent environment using Elo ratings.

Each round:

1. Every active member independently works on the task.
2. Members compete pairwise.
3. An LLM judge returns A, B, or DRAW.
4. Elo ratings are updated deterministically.
5. Up to deaths_per_round decisive losers are eliminated.
6. The highest-rated survivors reproduce to restore population size.

For persistent URSA agents, descendants inherit:

• a copy of the parent's workspace,
• a fork of the parent's LangGraph checkpoint database,
• a fork of the parent's LangGraph store,
• the parent's Elo rating,
• lineage metadata.

Parent and child subsequently evolve independently.

Source code in src/ursa/environments/agent_elo.py

class AgentEloEnvironment(BaseEnvironment):
    """Evolutionary pairwise agent environment using Elo ratings.

    Each round:

    1. Every active member independently works on the task.
    2. Members compete pairwise.
    3. An LLM judge returns A, B, or DRAW.
    4. Elo ratings are updated deterministically.
    5. Up to ``deaths_per_round`` decisive losers are eliminated.
    6. The highest-rated survivors reproduce to restore population size.

    For persistent URSA agents, descendants inherit:

    - a copy of the parent's workspace,
    - a fork of the parent's LangGraph checkpoint database,
    - a fork of the parent's LangGraph store,
    - the parent's Elo rating,
    - lineage metadata.

    Parent and child subsequently evolve independently.
    """

    def __init__(
        self,
        llm: BaseChatModel,
        *,
        config: (AgentEloConfig | Mapping[str, Any] | str | Path | None) = None,
        name: str | None = None,
        group: str | None = None,
        members: (
            list[EnvironmentMemberConfig | Mapping[str, Any]] | None
        ) = None,
        workspace: str | Path | None = None,
        initial_rating: float | None = None,
        k_factor: float | None = None,
        deaths_per_round: int | None = None,
        seed: int | None = None,
        generations: int | None = None,
        member_timeout_seconds: float | None = None,
        restart_from_json: str | Path | None = None,
        judge_prompt: str | None = None,
        persist_members: bool = True,
        **kwargs: Any,
    ):
        elo_config = AgentEloConfig.from_source(
            config=config,
            name=name,
            group=group,
            members=members,
            workspace=workspace,
            initial_rating=initial_rating,
            k_factor=k_factor,
            deaths_per_round=deaths_per_round,
            seed=seed,
            generations=generations,
            member_timeout_seconds=member_timeout_seconds,
            restart_from_json=restart_from_json,
            judge_prompt=judge_prompt,
        )

        super().__init__(
            llm,
            name=elo_config.name,
            group=elo_config.group,
            workspace=elo_config.workspace or workspace,
            persist_members=persist_members,
            **kwargs,
        )

        self.config = elo_config

        self.initial_rating = elo_config.initial_rating
        self.k_factor = elo_config.k_factor
        self.deaths_per_round = elo_config.deaths_per_round
        self.generations = elo_config.generations
        self.member_timeout_seconds = elo_config.member_timeout_seconds

        self.seed = elo_config.seed

        self._rng = random.Random(self.seed)

        self.generation_index = 0

        self._offspring_counts: dict[
            str,
            int,
        ] = {}

        if self.config.restart_from_json is not None:
            if not self.persist_members:
                raise ValueError(
                    "restart_from_json requires persist_members=True."
                )

            state = self._load_environment_state(self.config.restart_from_json)

            self._restore_environment_state(state)

        else:
            self.member_configs = list(self.config.members)

            self._validate_population_size(len(self.member_configs))

            self.members = {
                member.name: self.build_member(member)
                for member in self.member_configs
            }

            self.players = {
                member.name: EloPlayer(
                    name=member.name,
                    rating=self.initial_rating,
                )
                for member in self.member_configs
            }

        self.judge_prompt = self.config.judge_prompt or ""

        self.judge = AgentEloJudge(
            llm=self.llm,
            workspace=self.workspace,
            group=self.group,
            judge_prompt=self.judge_prompt,
        )

    # ------------------------------------------------------------------
    # Configuration
    # ------------------------------------------------------------------

    # Keep checking live population sizes after reproduction and on restart.
    _validate_population_size = staticmethod(AgentEloConfig.validate_population_size)

    @classmethod
    def from_yaml(
        cls,
        path: str | Path,
        *,
        llm: BaseChatModel,
        **kwargs: Any,
    ) -> "AgentEloEnvironment":
        return cls(
            llm=llm,
            config=path,
            **kwargs,
        )

    # ------------------------------------------------------------------
    # For lightweight environment level persistence
    # ------------------------------------------------------------------

    def _events(
        self,
        config: Mapping[str, Any] | None,
    ) -> EnvironmentEvents:
        return EnvironmentEvents(
            environment=self.name,
            config=config,
            environment_type="agent_elo",
            environment_id=self.name,
            path=[self.name],
        )

    def _source(
        self,
        name: str,
        *,
        kind: str = "agent",
    ) -> dict[str, Any]:
        return {
            "id": f"{self.name}.{name}",
            "name": name,
            "kind": kind,
            "path": [
                self.name,
                name,
            ],
        }

    def _member_runtime_config(
        self,
        base_config: Mapping[str, Any] | None,
        member: EnvironmentMemberConfig,
    ) -> dict[str, Any] | None:
        """Attach stable Elo-member identity to nested agent/tool events."""

        if base_config is None:
            return None

        member_id = f"{self.name}.{member.name}"

        merged = dict(base_config)

        base_metadata = merged.get("metadata")

        metadata = (
            dict(base_metadata)
            if isinstance(
                base_metadata,
                Mapping,
            )
            else {}
        )

        metadata.update({
            "environment_id": self.name,
            "environment_member": (member.name),
            "environment_member_id": (member_id),
            "environment_member_role": (member.role),
            "environment_member_path": [
                self.name,
                member.name,
            ],
            "agent": member.name,
            "agent_id": member_id,
        })

        merged["metadata"] = metadata

        base_tags = merged.get("tags")

        if isinstance(base_tags, str):
            tags = [base_tags]
        else:
            tags = list(base_tags) if base_tags else []

        for tag in (
            member.name,
            member_id,
            "environment_member",
            "elo_member",
        ):
            if tag not in tags:
                tags.append(tag)

        merged["tags"] = tags

        return merged

    def _topology_payload(
        self,
        *,
        generation: int | None = None,
    ) -> dict[str, Any]:
        active_names = {member.name for member in self.member_configs}

        nodes: list[dict[str, Any]] = []

        # Active population.
        for member in self.member_configs:
            player = self.players[member.name]

            nodes.append({
                "id": (f"{self.name}.{member.name}"),
                "name": member.name,
                "kind": "agent",
                "role": member.role,
                "agent_class": (member.agent),
                "path": [
                    self.name,
                    member.name,
                ],
                "rating": player.rating,
                "lineage_generation": (player.generation),
                "parent": player.parent,
                "active": True,
            })

        # A restart can contain a child whose parent is no
        # longer active. Include a lightweight historical
        # parent node so its lineage edge still has a source.
        historical_parents = {
            player.parent
            for player in self.players.values()
            if (player.parent and player.parent not in active_names)
        }

        for parent in sorted(historical_parents):
            nodes.append({
                "id": (f"{self.name}.{parent}"),
                "name": parent,
                "kind": "agent",
                "role": ("Historical Elo parent"),
                "path": [
                    self.name,
                    parent,
                ],
                "active": False,
            })

        edges = []

        for player in self.players.values():
            if player.parent:
                edges.append({
                    "source": (f"{self.name}.{player.parent}"),
                    "target": (f"{self.name}.{player.name}"),
                    "kind": "parent_of",
                })

        return {
            "kind": "agent_elo",
            "name": self.name,
            "description": (self.config.description),
            "generation": (
                self.generation_index if generation is None else generation
            ),
            "nodes": nodes,
            "edges": edges,
        }

    def _member_config(
        self,
        name: str,
    ) -> EnvironmentMemberConfig:
        """Return the active config for one Elo member."""

        for member in self.member_configs:
            if member.name == name:
                return member

        raise KeyError(f"Unknown Elo member: {name}")

    @staticmethod
    def _member_config_to_mapping(
        member: EnvironmentMemberConfig,
    ) -> dict[str, Any]:
        """Convert a member config into JSON-safe data."""

        model = None

        if member.model is not None:
            model = member.model.model_dump(
                mode="json",
                exclude_none=True,
                # Persist resolved settings without requiring the original
                # provider registry (or conflicting with its resolved URL).
                exclude={"inference_provider"},
            )

        return {
            "name": member.name,
            "role": member.role,
            "agent": member.agent,
            "model": model,
            "config": member.config,
            "prompt": member.prompt,
            "reviewer": member.reviewer,
        }

    @staticmethod
    def _rng_state_to_json(
        value: Any,
    ) -> Any:
        if isinstance(value, tuple):
            return [
                AgentEloEnvironment._rng_state_to_json(item) for item in value
            ]

        return value

    @staticmethod
    def _rng_state_from_json(
        value: Any,
    ) -> Any:
        if isinstance(value, list):
            return tuple(
                AgentEloEnvironment._rng_state_from_json(item) for item in value
            )

        return value

    def _environment_state_path(
        self,
    ) -> Path:
        return Path(self.workspace) / "environment_state.json"

    def _environment_state_payload(
        self,
    ) -> dict[str, Any]:
        """Return lightweight resumable environment state."""

        config_by_name = {member.name: member for member in self.member_configs}

        active_players = []

        # Use member_configs ordering so active population order
        # is preserved exactly across restart.
        for member in self.member_configs:
            player = self.players[member.name]

            active_players.append({
                "name": player.name,
                "rating": player.rating,
                "generation": player.generation,
                "parent": player.parent,
                "member_config": (
                    self._member_config_to_mapping(config_by_name[player.name])
                ),
            })

        return {
            "schema_version": 1,
            "environment_name": self.name,
            "group": self.group,
            "generation": self.generation_index,
            "seed": self.seed,
            "rng_state": self._rng_state_to_json(self._rng.getstate()),
            "active_players": active_players,
        }

    def _save_environment_state(
        self,
    ) -> Path:
        """Atomically save lightweight environment restart state."""

        target = self._environment_state_path()

        target.parent.mkdir(
            parents=True,
            exist_ok=True,
        )

        temporary = target.with_suffix(".json.tmp")

        payload = self._environment_state_payload()

        temporary.write_text(
            json.dumps(
                payload,
                indent=2,
            )
            + "\n",
            encoding="utf-8",
        )

        temporary.replace(target)

        return target

    @staticmethod
    def _load_environment_state(
        path: str | Path,
    ) -> dict[str, Any]:
        state_path = Path(path).expanduser()

        if not state_path.exists():
            raise FileNotFoundError(
                f"Elo environment restart file does not exist: {state_path}"
            )

        state = json.loads(state_path.read_text(encoding="utf-8"))

        if not isinstance(state, dict):
            raise ValueError(
                "Elo environment restart file must contain a JSON object."
            )

        if state.get("schema_version") != 1:
            raise ValueError(
                "Unsupported Elo environment "
                "state schema version: "
                f"{state.get('schema_version')!r}"
            )

        return state

    def _restore_environment_state(
        self,
        state: Mapping[str, Any],
    ) -> None:
        """Restore the active evolutionary population.

        This method restores only environment metadata.

        Individual URSA agents reopen their own persistence
        independently when build_member() is called.
        """

        saved_name = state.get("environment_name")

        saved_group = state.get("group")

        if saved_name != self.name:
            raise ValueError(
                "Restart environment name mismatch: "
                f"snapshot={saved_name!r}, "
                f"current={self.name!r}"
            )

        if saved_group != self.group:
            raise ValueError(
                "Restart group mismatch: "
                f"snapshot={saved_group!r}, "
                f"current={self.group!r}"
            )

        raw_players = state.get("active_players")

        if not isinstance(
            raw_players,
            list,
        ):
            raise ValueError("Restart state is missing 'active_players'.")

        self._validate_population_size(len(raw_players))

        member_configs: list[EnvironmentMemberConfig] = []

        players: dict[
            str,
            EloPlayer,
        ] = {}

        for raw_player in raw_players:
            if not isinstance(
                raw_player,
                Mapping,
            ):
                raise ValueError("Each active player entry must be a mapping.")

            raw_member_config = raw_player.get("member_config")

            if not isinstance(
                raw_member_config,
                Mapping,
            ):
                raise ValueError("Restart player is missing 'member_config'.")

            restored_member_config = dict(raw_member_config)
            saved_model = restored_member_config.get("model")
            if isinstance(saved_model, Mapping):
                # Older snapshots retained the provider name alongside the
                # resolved settings. They too must load without a registry.
                saved_model = dict(saved_model)
                saved_model.pop("inference_provider", None)
                restored_member_config["model"] = saved_model

            member_config = EnvironmentMemberConfig.from_mapping(
                restored_member_config,
                group=self.group,
            )
            AgentEloConfig.validate_member_config(member_config)

            name = str(raw_player["name"])

            if name in players:
                raise ValueError(
                    f"Elo member name {name!r} is duplicated in restart state. "
                    "Each member must have a unique name."
                )

            if member_config.name != name:
                raise ValueError(
                    "Restart player/member config "
                    f"name mismatch: {name!r} vs "
                    f"{member_config.name!r}"
                )

            # A restart should resume an existing persistent
            # URSA agent, never silently create a fresh one.
            den = self._member_den(name)

            if not den.exists():
                raise FileNotFoundError(
                    "Persistent URSA agent den "
                    "required for restart does not exist: "
                    f"{den}"
                )

            member_workspace = self._member_workspace(name)

            if not member_workspace.exists():
                raise FileNotFoundError(
                    "Agent workspace required for "
                    "restart does not exist: "
                    f"{member_workspace}"
                )

            member_configs.append(member_config)

            players[name] = EloPlayer(
                name=name,
                rating=float(raw_player["rating"]),
                generation=int(raw_player["generation"]),
                parent=raw_player.get("parent"),
            )

        self.member_configs = member_configs

        self.players = players

        # Constructing by the same names causes URSA to reopen
        # the agents' existing persistent state.
        self.members = {
            member.name: self.build_member(member)
            for member in self.member_configs
        }

        self.generation_index = int(
            state.get(
                "generation",
                0,
            )
        )

        # The restart snapshot is authoritative for the
        # evolutionary RNG metadata.
        self.seed = state.get("seed")

        rng_state = state.get("rng_state")

        if rng_state is not None:
            self._rng.setstate(self._rng_state_from_json(rng_state))
        else:
            # Defensive support for snapshots that contain a seed
            # but no serialized RNG state.
            self._rng = random.Random(self.seed)

    # ------------------------------------------------------------------
    # Persistent URSA state
    # ------------------------------------------------------------------

    def _member_den(
        self,
        member_name: str,
    ) -> Path:
        """Return the persistent URSA den for a member."""
        agent_name = self._member_agent_name(member_name)

        if agent_name is None:
            raise RuntimeError(
                "Agent persistence is disabled. "
                "Persistent inheritance requires "
                "persist_members=True."
            )

        return group_agents_dir(self.group) / agent_name

    @staticmethod
    def _backup_sqlite_database(
        source: Path,
        destination: Path,
    ) -> None:
        """Create an independent SQLite snapshot."""
        source = Path(source)
        destination = Path(destination)

        if not source.exists():
            raise FileNotFoundError(f"SQLite source does not exist: {source}")

        destination.parent.mkdir(
            parents=True,
            exist_ok=True,
        )

        if destination.exists():
            raise FileExistsError(
                f"SQLite destination already exists: {destination}"
            )

        # SQLite's connection context manager only manages transactions.
        # Close both handles so rollback can remove child databases on Windows.
        with closing(sqlite3.connect(source)) as source_conn:
            with closing(sqlite3.connect(destination)) as destination_conn:
                source_conn.backup(destination_conn)

    def _fork_parent_persistence(
        self,
        parent_name: str,
        child_name: str,
    ) -> None:
        """Fork parent URSA persistence into a new child den."""
        parent_den = self._member_den(parent_name)

        child_den = self._member_den(child_name)

        if not parent_den.exists():
            raise FileNotFoundError(f"Parent den does not exist: {parent_den}")

        if child_den.exists():
            raise FileExistsError(f"Child den already exists: {child_den}")

        child_den.mkdir(
            parents=True,
            exist_ok=False,
        )

        try:
            self._backup_sqlite_database(
                parent_den / "db" / "checkpointer.db",
                child_den / "db" / "checkpointer.db",
            )

            graph_store_source = parent_den / "graph_store.sqlite"

            if graph_store_source.exists():
                self._backup_sqlite_database(
                    graph_store_source,
                    child_den / "graph_store.sqlite",
                )

        except Exception:
            # Do not leave a half-forked lineage behind.
            shutil.rmtree(
                child_den,
                ignore_errors=True,
            )
            raise

    # ------------------------------------------------------------------
    # Elo
    # ------------------------------------------------------------------

    def _apply_match_result(
        self,
        result: MatchResult,
    ) -> None:
        player_a = self.players[result.player_a]

        player_b = self.players[result.player_b]

        new_a, new_b = result.updated_ratings(
            player_a.rating,
            player_b.rating,
            self.k_factor,
        )

        player_a.rating = new_a
        player_b.rating = new_b

    # ------------------------------------------------------------------
    # Member execution
    # ------------------------------------------------------------------

    def _make_pairs(
        self,
        names: list[str],
    ) -> list[tuple[str, str]]:
        """Randomly pair all active members.

        AgentEloEnvironment requires an even population, so every
        active member participates in exactly one match per generation.
        """

        self._validate_population_size(len(names))

        shuffled = list(names)

        self._rng.shuffle(shuffled)

        return [
            (
                shuffled[index],
                shuffled[index + 1],
            )
            for index in range(
                0,
                len(shuffled),
                2,
            )
        ]

    def _progress_report_relative_path(self) -> Path:
        return (
            Path("_elo_progress") / f"generation_{self.generation_index + 1}.md"
        )

    def _read_progress_report(self, member_name: str) -> str | None:
        path = (
            self._member_workspace(member_name)
            / self._progress_report_relative_path()
        )
        try:
            report = path.read_text(encoding="utf-8", errors="replace").strip()
        except OSError:
            return None
        return report or None

    def _member_prompt(
        self,
        member: EnvironmentMemberConfig,
        task: str,
        *,
        deadline: datetime | None = None,
    ) -> str:
        player = self.players[member.name]
        if self.generation_index == 0:
            approach = "Develop and validate a strong independent approach to the task."
        elif player.parent is not None:
            approach = (
                "Build on your parent's work while exploring a substantive "
                "improvement or alternative approach. Preserve useful results "
                "and validate your changes; novelty alone is not success."
            )
        else:
            approach = (
                "Build on your existing work. Favor refinement and stronger "
                "validation, making larger changes when justified."
            )
        execution_budget = (
            f"\nDeadline (UTC): {deadline.isoformat()}\n"
            "Your run will be terminated at this deadline. Save useful results "
            "as you work and aim to provide a final response before time runs out.\n"
            if deadline is not None
            else ""
        )
        extra = (
            f"\nAdditional guidance:\n{member.prompt}\n"
            if member.prompt
            else ""
        )
        report_path = self._progress_report_relative_path().as_posix()
        return (
            f"You are competitor '{member.name}'.\n"
            f"Role: {member.role}\n"
            f"Environment generation: {self.generation_index + 1}\n"
            f"Lineage generation: {player.generation}\n"
            f"Parent: {player.parent or 'None'}\n\n"
            f"Task:\n{task}\n\n"
            f"Approach:\n{approach}\n\n"
            "Follow the task's evaluation criteria. Distinguish verified results "
            "from plans or untested claims.\n"
            f"{execution_budget}\n"
            "If you can write files, maintain a concise progress report at:\n"
            f"{report_path}\n\n"
            "Update it after meaningful milestones with completed work, verified "
            "results, remaining limitations, and supporting file paths. "
            "Distinguish new progress from inherited work. Write updates to a "
            "temporary file, then rename it into place.\n\n"
            "Your final response should summarize results, evidence, limitations, "
            "and files produced. If you time out, the judge will use your progress "
            "report or briefly inspect your workspace. A timeout alone is not a loss.\n"
            f"{extra}"
        )

    def _format_member_result(
        self,
        member_name: str,
        result: Any,
    ) -> str:
        """Return the member agent's canonical user-facing result."""

        member = self.members[member_name]

        formatter = getattr(
            member,
            "format_result",
            None,
        )

        if callable(formatter):
            formatted = formatter(result)

            if formatted is not None:
                return str(formatted)

        return result_to_text(result)

    async def _run_member(
        self,
        member: EnvironmentMemberConfig,
        task: str,
        invoke_kwargs: Mapping[str, Any],
        *,
        deadline: datetime | None = None,
    ) -> MemberRunResult:
        prompt = self._member_prompt(
            member,
            task,
            deadline=deadline,
        )

        async def ainvoke() -> Any:
            return await self._invoke_member_async(
                self.members[member.name],
                prompt,
                **invoke_kwargs,
            )

        deadline_text = deadline.isoformat() if deadline is not None else None
        status = "completed"
        output = None
        error = None
        progress_report = None

        try:
            if deadline is None:
                result = await ainvoke()

            else:
                remaining_seconds = (
                    deadline - datetime.now(timezone.utc)
                ).total_seconds()

                if remaining_seconds <= 0:
                    raise TimeoutError

                # Cancellation stops the environment from awaiting this
                # member. A blocking subprocess already running in an
                # executor may continue until that subprocess exits or
                # reaches its own timeout.
                result = await asyncio.wait_for(
                    ainvoke(),
                    timeout=remaining_seconds,
                )

            output = self._format_member_result(member.name, result)
            if not output.strip():
                progress_report = self._read_progress_report(member.name)

        except TimeoutError:
            status = "timed_out"
            output = None
            progress_report = self._read_progress_report(member.name)

        except Exception as exc:
            status = "failed"
            output = None
            error = f"{type(exc).__name__}: {exc}"

        return MemberRunResult(
            name=member.name,
            status=status,
            output=output,
            deadline=deadline_text,
            error=error,
            progress_report=progress_report,
        )

    # ------------------------------------------------------------------
    # Judging
    # ------------------------------------------------------------------

    @staticmethod
    def _judge_submission(run: MemberRunResult) -> str:
        if run.output and run.output.strip():
            evidence = f"Final response:\n{run.output}"
        elif run.progress_report:
            evidence = f"Saved progress report:\n{run.progress_report}"
        else:
            evidence = (
                "No final response or progress report is available. "
                "Briefly inspect this candidate's workspace for evidence."
            )
        return f"Execution status: {run.status}\n\n{evidence}"

    async def _resolve_match(
        self,
        *,
        task: str,
        player_a: str,
        run_a: MemberRunResult,
        player_b: str,
        run_b: MemberRunResult,
    ) -> MatchResult:
        """Resolve a match using execution status before LLM judging."""

        # Timeouts remain eligible for judging using partial work.
        if (run_a.completed or run_a.timed_out) and (
            run_b.completed or run_b.timed_out
        ):
            return await self._judge_match(
                task=task,
                player_a=player_a,
                output_a=self._judge_submission(run_a),
                player_b=player_b,
                output_b=self._judge_submission(run_b),
            )

        # A completed and B did not.
        if run_a.completed:
            return MatchResult(
                player_a=player_a,
                player_b=player_b,
                score_a=1.0,
                reasoning=(
                    f"{player_a} completed the generation successfully; "
                    f"{player_b} did not complete successfully "
                    f"(status={run_b.status}). "
                    f"{player_a} therefore wins automatically."
                ),
            )

        # B completed and A did not.
        if run_b.completed:
            return MatchResult(
                player_a=player_a,
                player_b=player_b,
                score_a=0.0,
                reasoning=(
                    f"{player_b} completed the generation successfully; "
                    f"{player_a} did not complete successfully "
                    f"(status={run_a.status}). "
                    f"{player_b} therefore wins automatically."
                ),
            )

        if run_a.timed_out and run_b.failed:
            return MatchResult(
                player_a=player_a,
                player_b=player_b,
                score_a=1.0,
                reasoning=(
                    f"{player_a} reached the execution deadline; "
                    f"{player_b} failed during execution and forfeits the match. "
                    f"{player_a} therefore wins automatically."
                ),
            )

        if run_b.timed_out and run_a.failed:
            return MatchResult(
                player_a=player_a,
                player_b=player_b,
                score_a=0.0,
                reasoning=(
                    f"{player_b} reached the execution deadline; "
                    f"{player_a} failed during execution and forfeits the match. "
                    f"{player_b} therefore wins automatically."
                ),
            )

        # Both failed.
        return MatchResult(
            player_a=player_a,
            player_b=player_b,
            score_a=0.5,
            reasoning=(
                "Neither competitor completed the generation successfully. "
                f"{player_a} status={run_a.status}; "
                f"{player_b} status={run_b.status}. "
                "Match recorded as a draw."
            ),
        )

    async def _judge_match(
        self,
        task: str,
        player_a: str,
        output_a: str,
        player_b: str,
        output_b: str,
    ) -> MatchResult:
        config_a = self._member_config(player_a)

        config_b = self._member_config(player_b)

        decision = await self.judge.judge_match(
            task=task,
            player_a=player_a,
            agent_type_a=config_a.agent,
            output_a=output_a,
            player_b=player_b,
            agent_type_b=config_b.agent,
            output_b=output_b,
        )

        if decision.winner == "A":
            score_a = 1.0

        elif decision.winner == "B":
            score_a = 0.0

        else:
            score_a = 0.5

        return MatchResult(
            player_a=player_a,
            player_b=player_b,
            score_a=score_a,
            reasoning=decision.reasoning,
        )

    # ------------------------------------------------------------------
    # Selection
    # ------------------------------------------------------------------

    def _select_losers(
        self,
        match_results: list[MatchResult],
    ) -> list[str]:
        """Choose decisive losers for elimination.

        Draws produce no loser.

        If more decisive losers exist than the configured number
        of deaths, the lowest-rated losing agents are eliminated.
        """
        losers = [
            result.loser for result in match_results if result.loser is not None
        ]

        # Defensive deduplication.
        losers = list(dict.fromkeys(losers))

        self._rng.shuffle(losers)

        losers.sort(key=lambda name: (self.players[name].rating))

        return losers[: self.deaths_per_round]

    def _eliminate(
        self,
        losers: list[str],
    ) -> None:
        """Remove agents from the active population.

        Persistent dens and workspaces are intentionally left on disk
        so extinct lineages remain inspectable.
        """
        loser_set = set(losers)

        for name in losers:
            member = self.members.pop(
                name,
                None,
            )

            if member is not None:
                close = getattr(
                    member,
                    "close",
                    None,
                )

                if callable(close):
                    try:
                        close()
                    except Exception:
                        logging.getLogger(__name__).warning(
                            "Failed to close eliminated Elo member %s",
                            name,
                            exc_info=True,
                        )

            self.players.pop(
                name,
                None,
            )

        self.member_configs = [
            member
            for member in self.member_configs
            if member.name not in loser_set
        ]

    def _top_survivors(
        self,
        count: int,
        *,
        excluded: list[str] | None = None,
    ) -> list[EloPlayer]:
        """Return the highest-rated surviving agents.

        Equal-rated survivors are ordered randomly using
        the environment RNG.
        """
        excluded_names = set(excluded or [])
        candidates = [
            player
            for player in self.players.values()
            if player.name not in excluded_names
        ]

        self._rng.shuffle(candidates)

        candidates.sort(
            key=lambda player: player.rating,
            reverse=True,
        )

        return candidates[:count]

    # ------------------------------------------------------------------
    # Reproduction
    # ------------------------------------------------------------------

    def _child_name_available(
        self,
        child_name: str,
    ) -> bool:
        """Return whether a name is safe to use for a new birth.

        A generated child name must not collide with:

        - an active player,
        - an active member,
        - an active member config,
        - an existing environment workspace,
        - an existing persistent URSA den.

        Existing persistent dens are checked only when member
        persistence is enabled.
        """

        if child_name in self.players:
            return False

        if child_name in self.members:
            return False

        if any(member.name == child_name for member in self.member_configs):
            return False

        child_workspace = self._member_workspace(child_name)

        if child_workspace.exists():
            return False

        if self.persist_members:
            child_den = self._member_den(child_name)

            if child_den.exists():
                return False

        return True

    def _next_child_name(
        self,
        parent: EloPlayer,
    ) -> str:
        """Generate the next unused descendant identity.

        The search starts after the highest child count generated
        by this environment instance, but also checks existing
        workspaces and persistent agent dens so resumed runs cannot
        accidentally reuse an older descendant.
        """

        count = (
            self._offspring_counts.get(
                parent.name,
                0,
            )
            + 1
        )

        child_generation = parent.generation + 1

        while True:
            child_name = f"{parent.name}_g{child_generation}_{count}"

            if self._child_name_available(child_name):
                self._offspring_counts[parent.name] = count

                return child_name

            count += 1

    def _copy_parent_workspace(
        self,
        parent_name: str,
        child_name: str,
    ) -> None:
        """Fork the parent's working filesystem."""
        parent_workspace = self._member_workspace(parent_name)

        child_workspace = self._member_workspace(child_name)

        if child_workspace.exists():
            raise FileExistsError(
                f"Child workspace already exists: {child_workspace}"
            )

        if parent_workspace.exists():
            shutil.copytree(
                parent_workspace,
                child_workspace,
            )

        else:
            child_workspace.mkdir(
                parents=True,
                exist_ok=False,
            )

    def _cleanup_failed_child(
        self,
        child_name: str,
    ) -> None:
        """Best-effort rollback of a failed reproduction."""
        child_workspace = self._member_workspace(child_name)

        shutil.rmtree(
            child_workspace,
            ignore_errors=True,
        )

        if self.persist_members:
            try:
                child_den = self._member_den(child_name)
            except Exception:
                return

            shutil.rmtree(
                child_den,
                ignore_errors=True,
            )

    def _reproduce(
        self,
        parents: list[EloPlayer],
    ) -> list[str]:
        """Create one independent child for every selected parent.

        Children inherit:

        - Elo rating,
        - lineage metadata,
        - workspace,
        - URSA checkpoint state,
        - LangGraph persistent store.
        """
        children: list[str] = []
        prepared: list[tuple[EnvironmentMemberConfig, Any, EloPlayer]] = []
        offspring_counts_before = dict(self._offspring_counts)
        config_by_name = {member.name: member for member in self.member_configs}

        try:
            for parent in parents:
                parent_config = config_by_name[parent.name]
                child_name = self._next_child_name(parent)
                children.append(child_name)
                child_config = replace(parent_config, name=child_name)

                # Prepare every child before changing the active population.
                self._copy_parent_workspace(parent.name, child_name)
                if self.persist_members:
                    self._fork_parent_persistence(parent.name, child_name)
                child = self.build_member(child_config)
                prepared.append((
                    child_config,
                    child,
                    EloPlayer(
                        name=child_name,
                        rating=parent.rating,
                        generation=parent.generation + 1,
                        parent=parent.name,
                    ),
                ))
        except BaseException:
            # Close constructed children before deleting their persistence.
            # A cleanup error must not prevent rollback of the other children.
            for _, child, _ in prepared:
                try:
                    close = getattr(child, "close", None)
                    if callable(close):
                        close()
                except Exception:
                    pass
            for child_name in children:
                self._cleanup_failed_child(child_name)
            self._offspring_counts = offspring_counts_before
            raise

        for child_config, child, player in prepared:
            self.members[player.name] = child
            self.member_configs.append(child_config)
            self.players[player.name] = player

        return children

    # ------------------------------------------------------------------
    # Reporting
    # ------------------------------------------------------------------

    def standings(
        self,
    ) -> list[dict[str, Any]]:
        ordered = sorted(
            self.players.values(),
            key=lambda player: (player.rating),
            reverse=True,
        )

        return [
            {
                "rank": rank,
                "name": player.name,
                "rating": player.rating,
                "generation": player.generation,
                "parent": player.parent,
            }
            for rank, player in enumerate(
                ordered,
                start=1,
            )
        ]

    # ------------------------------------------------------------------
    # Environment invocation
    # ------------------------------------------------------------------

    def _invoke(
        self,
        inputs: Mapping[str, Any],
        **config: Any,
    ) -> dict[str, Any]:
        return self._run_ainvoke_from_sync(
            inputs,
            **config,
        )

    async def _run_generation(
        self,
        task: str,
        invoke_kwargs: Mapping[str, Any],
        *,
        runtime_config: Mapping[
            str,
            Any,
        ]
        | None = None,
    ) -> dict[str, Any]:
        """Run one complete evolutionary generation."""

        self._validate_population_size(len(self.member_configs))

        if not (
            len(self.member_configs) == len(self.members) == len(self.players)
        ):
            raise RuntimeError(
                "AgentEloEnvironment active population state is inconsistent: "
                f"member_configs={len(self.member_configs)}, "
                f"members={len(self.members)}, "
                f"players={len(self.players)}."
            )

        events = self._events(runtime_config)

        generation_start = perf_counter()

        generation_number = self.generation_index + 1

        initial_population_size = len(self.players)
        rng_state_before = self._rng.getstate()

        # Capture ratings before competition for reporting.
        ratings_before = {
            name: player.rating for name, player in self.players.items()
        }

        generation_deadline: datetime | None = None

        if self.member_timeout_seconds is not None:
            generation_deadline = datetime.now(timezone.utc) + timedelta(
                seconds=self.member_timeout_seconds
            )

        await events.aemit(
            (f"Elo generation {generation_number} started"),
            stage="generation",
            phase="start",
            event_type="generation_started",
            generation=generation_number,
            population_size=len(self.players),
            ratings={
                name: player.rating for name, player in self.players.items()
            },
            deadline=(
                generation_deadline.isoformat()
                if generation_deadline is not None
                else None
            ),
        )

        async def run_member(
            member: EnvironmentMemberConfig,
        ) -> MemberRunResult:
            source = self._source(member.name)

            start = perf_counter()

            await events.aemit(
                (f"Elo member {member.name} started"),
                stage="member",
                phase="start",
                event_type="member_started",
                generation=generation_number,
                source=source,
            )

            member_kwargs = dict(invoke_kwargs)

            member_runtime_config = self._member_runtime_config(
                runtime_config,
                member,
            )

            if member_runtime_config is not None:
                member_kwargs["config"] = member_runtime_config

            result = await self._run_member(
                member,
                task,
                member_kwargs,
                deadline=generation_deadline,
            )

            if result.completed:
                event_type = "member_completed"
                phase = "end"
                level = "info"

            elif result.timed_out:
                event_type = "member_timed_out"
                phase = "error"
                level = "warning"

            else:
                event_type = "member_failed"
                phase = "error"
                level = "error"

            await events.aemit(
                (f"Elo member {member.name} {result.status}"),
                stage="member",
                phase=phase,
                event_type=event_type,
                level=level,
                generation=generation_number,
                source=source,
                status=result.status,
                result=result.output,
                error=result.error,
                deadline=result.deadline,
                elapsed_seconds=(perf_counter() - start),
            )

            return result

        # ----------------------------------------------------------
        # Phase 1: independent research
        # ----------------------------------------------------------

        member_results = await asyncio.gather(*[
            run_member(member) for member in self.member_configs
        ])

        runs = {result.name: result for result in member_results}

        outputs = {
            name: (result.output if result.completed else None)
            for name, result in runs.items()
        }

        timed_out = [name for name, result in runs.items() if result.timed_out]

        failed = [name for name, result in runs.items() if result.failed]

        # ----------------------------------------------------------
        # Phase 2: randomized pairwise competition
        # ----------------------------------------------------------

        pairs = self._make_pairs([
            member.name for member in self.member_configs
        ])

        await events.aemit(
            (f"Elo generation {generation_number} pairings created"),
            stage="pairing",
            phase="declared",
            event_type="pairings_declared",
            generation=generation_number,
            pairs=[list(pair) for pair in pairs],
        )

        match_results: list[MatchResult] = []

        for player_a, player_b in pairs:
            rating_a_before = self.players[player_a].rating

            rating_b_before = self.players[player_b].rating

            await events.aemit(
                (f"Elo match {player_a} vs {player_b} started"),
                stage="match",
                phase="start",
                event_type="match_started",
                generation=generation_number,
                source=self._source(player_a),
                target=self._source(player_b),
            )

            result = await self._resolve_match(
                task=task,
                player_a=player_a,
                run_a=runs[player_a],
                player_b=player_b,
                run_b=runs[player_b],
            )

            self._apply_match_result(result)

            await events.aemit(
                (f"Elo match {player_a} vs {player_b} completed"),
                stage="match",
                phase="end",
                event_type="match_completed",
                generation=generation_number,
                source=self._source(player_a),
                target=self._source(player_b),
                winner=result.winner,
                loser=result.loser,
                score_a=result.score_a,
                reasoning=result.reasoning,
                rating_a_before=(rating_a_before),
                rating_a_after=(self.players[player_a].rating),
                rating_b_before=(rating_b_before),
                rating_b_after=(self.players[player_b].rating),
            )

            match_results.append(result)

        standings_after_matches = self.standings()

        # ----------------------------------------------------------
        # Phase 3: prepare replacement children, then eliminate losers
        # ----------------------------------------------------------

        eliminated = self._select_losers(match_results)
        parents = self._top_survivors(len(eliminated), excluded=eliminated)
        parent_names = [parent.name for parent in parents]

        try:
            children = self._reproduce(parents)
        except BaseException:
            # Keep failed generations retryable without applying Elo twice.
            for name, rating in ratings_before.items():
                self.players[name].rating = rating
            self._rng.setstate(rng_state_before)
            raise

        eliminated_ratings = {
            name: self.players[name].rating for name in eliminated
        }
        self._eliminate(eliminated)

        for name in eliminated:
            await events.aemit(
                (f"Elo member {name} eliminated"),
                stage="selection",
                phase="end",
                event_type="member_eliminated",
                generation=generation_number,
                source=self._source(name),
                rating=eliminated_ratings[name],
            )

        for child_name in children:
            child = self.players[child_name]

            await events.aemit(
                (f"Elo child {child_name} created"),
                stage="reproduction",
                phase="end",
                event_type="child_created",
                generation=generation_number,
                source=(self._source(child.parent) if child.parent else None),
                target=self._source(child_name),
                child=child_name,
                parent=child.parent,
                rating=child.rating,
                lineage_generation=(child.generation),
            )

        await events.aemit(
            (f"Elo generation {generation_number} topology updated"),
            stage="elo",
            phase="topology",
            event_type="topology_declared",
            generation=generation_number,
            topology=self._topology_payload(
                generation=generation_number,
            ),
        )

        final_population_size = len(self.players)

        if final_population_size != initial_population_size:
            raise RuntimeError(
                "Population size changed unexpectedly: "
                f"{initial_population_size} -> "
                f"{final_population_size}"
            )

        # ----------------------------------------------------------
        # Generation successfully completed.
        #
        # Increment BEFORE snapshotting so restart begins at the
        # following generation.
        # ----------------------------------------------------------

        self.generation_index += 1

        state_path = self._save_environment_state()

        await events.aemit(
            (f"Elo generation {generation_number} completed"),
            stage="generation",
            phase="end",
            event_type="generation_completed",
            generation=generation_number,
            eliminated=eliminated,
            children=children,
            standings=self.standings(),
            elapsed_seconds=(perf_counter() - generation_start),
        )

        return {
            "generation": generation_number,
            "task": task,
            "outputs": outputs,
            "pairs": [list(pair) for pair in pairs],
            "ratings_before": ratings_before,
            "matches": [
                {
                    "player_a": result.player_a,
                    "player_b": result.player_b,
                    "winner": result.winner,
                    "loser": result.loser,
                    "score_a": result.score_a,
                    "reasoning": result.reasoning,
                }
                for result in match_results
            ],
            "standings_after_matches": (standings_after_matches),
            "eliminated": eliminated,
            "reproducing_parents": (parent_names),
            "children": children,
            "standings": self.standings(),
            "population_size": (final_population_size),
            "environment_state": str(state_path),
            "member_runs": {
                name: {
                    "status": run.status,
                    "deadline": run.deadline,
                    "error": run.error,
                    "progress_report": run.progress_report,
                }
                for name, run in runs.items()
            },
            "timed_out": timed_out,
            "failed": failed,
            "generation_deadline": (
                generation_deadline.isoformat()
                if generation_deadline is not None
                else None
            ),
        }

    async def _ainvoke(
        self,
        inputs: Mapping[str, Any],
        **config: Any,
    ) -> dict[str, Any]:
        task = str(inputs.get("task") or inputs.get("prompt") or inputs)

        runtime_config = runnable_config_from_kwargs(config)

        events = self._events(runtime_config)

        run_start = perf_counter()

        invoke_kwargs = invocation_kwargs(config)

        generation_results = []

        starting_generation = self.generation_index

        await events.aemit(
            f"Agent Elo {self.name} started",
            stage="elo",
            phase="start",
            event_type="elo_started",
            task=task,
            starting_generation=(self.generation_index),
            requested_generations=(self.generations),
            topology=(self._topology_payload()),
        )

        await events.aemit(
            (f"Agent Elo {self.name} topology declared"),
            stage="elo",
            phase="topology",
            event_type="topology_declared",
            topology=(self._topology_payload()),
        )

        try:
            for _ in range(self.generations):
                result = await self._run_generation(
                    task,
                    invoke_kwargs,
                    runtime_config=(runtime_config),
                )

                generation_results.append(result)

        except BaseException as exc:
            await events.aemit(
                f"Agent Elo {self.name} failed",
                stage="elo",
                phase="error",
                event_type="elo_failed",
                level="error",
                task=task,
                error=str(exc),
                elapsed_seconds=(perf_counter() - run_start),
            )

            raise

        final_result = {
            "task": task,
            "starting_generation": (starting_generation),
            "completed_generations": (len(generation_results)),
            "ending_generation": (self.generation_index),
            "generations": (generation_results),
            "standings": self.standings(),
            "population_size": len(self.players),
            "environment_state": str(self._environment_state_path()),
        }

        await events.aemit(
            f"Agent Elo {self.name} completed",
            stage="elo",
            phase="end",
            event_type="elo_completed",
            task=task,
            completed_generations=(len(generation_results)),
            ending_generation=(self.generation_index),
            standings=self.standings(),
            result={
                "ending_generation": (self.generation_index),
                "standings": (self.standings()),
            },
            elapsed_seconds=(perf_counter() - run_start),
        )

        return final_result

EloPlayer dataclass

Evolutionary metadata associated with one active agent.

Source code in src/ursa/environments/agent_elo.py

@dataclass
class EloPlayer:
    """Evolutionary metadata associated with one active agent."""

    name: str
    rating: float = 1500.0
    generation: int = 0
    parent: str | None = None

MatchResult dataclass

Result of one pairwise competition.

Source code in src/ursa/environments/agent_elo.py

@dataclass
class MatchResult:
    """Result of one pairwise competition."""

    player_a: str
    player_b: str
    score_a: float
    reasoning: str

    @property
    def winner(self) -> str | None:
        if self.score_a == 1.0:
            return self.player_a

        if self.score_a == 0.0:
            return self.player_b

        return None

    @property
    def loser(self) -> str | None:
        if self.score_a == 1.0:
            return self.player_b

        if self.score_a == 0.0:
            return self.player_a

        return None

    @staticmethod
    def expected_score(
        rating_a: float,
        rating_b: float,
    ) -> float:
        """Return A's expected Elo score against B."""
        difference = (rating_b - rating_a) / 400.0
        # A nonpositive exponent avoids overflow for large rating differences.
        factor = 10.0 ** (-abs(difference))
        if difference >= 0:
            return factor / (1.0 + factor)
        return 1.0 / (1.0 + factor)

    def updated_ratings(
        self,
        rating_a: float,
        rating_b: float,
        k_factor: float,
    ) -> tuple[float, float]:
        """Return updated Elo ratings for a match.

        ``score_a``:
            1.0 -> A wins
            0.5 -> draw
            0.0 -> B wins
        """
        score_a = self.score_a
        if score_a not in {
            0.0,
            0.5,
            1.0,
        }:
            raise ValueError("score_a must be one of 0.0, 0.5, or 1.0")

        expected_a = self.expected_score(
            rating_a,
            rating_b,
        )

        expected_b = 1.0 - expected_a
        score_b = 1.0 - score_a

        new_a = rating_a + k_factor * (score_a - expected_a)

        new_b = rating_b + k_factor * (score_b - expected_b)

        return new_a, new_b

expected_score(rating_a, rating_b) staticmethod

Return A's expected Elo score against B.

Source code in src/ursa/environments/agent_elo.py

@staticmethod
def expected_score(
    rating_a: float,
    rating_b: float,
) -> float:
    """Return A's expected Elo score against B."""
    difference = (rating_b - rating_a) / 400.0
    # A nonpositive exponent avoids overflow for large rating differences.
    factor = 10.0 ** (-abs(difference))
    if difference >= 0:
        return factor / (1.0 + factor)
    return 1.0 / (1.0 + factor)

updated_ratings(rating_a, rating_b, k_factor)

Return updated Elo ratings for a match.

score_a: 1.0 -> A wins 0.5 -> draw 0.0 -> B wins

Source code in src/ursa/environments/agent_elo.py

def updated_ratings(
    self,
    rating_a: float,
    rating_b: float,
    k_factor: float,
) -> tuple[float, float]:
    """Return updated Elo ratings for a match.

    ``score_a``:
        1.0 -> A wins
        0.5 -> draw
        0.0 -> B wins
    """
    score_a = self.score_a
    if score_a not in {
        0.0,
        0.5,
        1.0,
    }:
        raise ValueError("score_a must be one of 0.0, 0.5, or 1.0")

    expected_a = self.expected_score(
        rating_a,
        rating_b,
    )

    expected_b = 1.0 - expected_a
    score_b = 1.0 - score_a

    new_a = rating_a + k_factor * (score_a - expected_a)

    new_b = rating_b + k_factor * (score_b - expected_b)

    return new_a, new_b

MemberRunResult dataclass

Result of one member's work during a generation.

Source code in src/ursa/environments/agent_elo.py

@dataclass
class MemberRunResult:
    """Result of one member's work during a generation."""

    name: str
    status: str
    output: str | None
    deadline: str | None = None
    error: str | None = None
    progress_report: str | None = None

    @property
    def completed(self) -> bool:
        return self.status == "completed"

    @property
    def timed_out(self) -> bool:
        return self.status == "timed_out"

    @property
    def failed(self) -> bool:
        return self.status == "failed"

agent_elo_judge

AgentEloJudge

Ephemeral ChatAgent-based judge for Elo matches.

Source code in src/ursa/environments/agent_elo_judge.py

class AgentEloJudge:
    """Ephemeral ChatAgent-based judge for Elo matches."""

    OUTPUT_INSTRUCTIONS = (
        "\n\n"
        "After evaluating the candidates, return exactly one JSON object "
        "with this schema:\n"
        "{\n"
        '  "winner": "A" | "B" | "DRAW",\n'
        '  "reasoning": "brief explanation"\n'
        "}\n"
        "Do not include markdown fences or text outside the JSON object."
    )

    def __init__(
        self,
        *,
        llm: BaseChatModel,
        workspace: Path,
        group: str,
        judge_prompt: str,
    ):
        self.llm = llm
        self.workspace = Path(workspace)
        self.group = group
        self.judge_prompt = judge_prompt

    def _judge_prompt(
        self,
        *,
        task: str,
        player_a: str,
        agent_type_a: str,
        output_a: str,
        player_b: str,
        agent_type_b: str,
        output_b: str,
    ) -> str:
        guidance = (
            f"\nAdditional judging guidance:\n{self.judge_prompt}\n"
            if self.judge_prompt
            else ""
        )
        return (
            "Compare the candidates using the task's evaluation criteria. "
            "If none are specified, judge how well each fulfills the task. "
            "A timeout alone is not a loss.\n\n"
            "Use the supplied final responses or progress reports first. "
            "If neither is available, briefly inspect the candidate's workspace. "
            "Otherwise, inspect files only when useful for verification. "
            "Prioritize relevant summaries and results; avoid extensive searches "
            "or lengthy calculations. If evidence remains unavailable, state that "
            "limitation. Do not modify candidate files.\n\n"
            "Your workspace is the parent of the candidate directories below.\n"
            f"{guidance}\n"
            f"Task:\n{task}\n\n"
            f"Candidate A: {player_a}\n"
            f"Agent type: {agent_type_a}\n"
            f"Workspace: {player_a}/\n"
            f"Submission:\n{output_a}\n\n"
            f"Candidate B: {player_b}\n"
            f"Agent type: {agent_type_b}\n"
            f"Workspace: {player_b}/\n"
            f"Submission:\n{output_b}\n"
        )

    @staticmethod
    def _parse_decision(
        text: str,
    ) -> JudgeDecision:
        try:
            judgment = json.loads(text)

        except json.JSONDecodeError as exc:
            raise ValueError(
                f"Elo ChatAgent judge returned invalid JSON:\n{text}"
            ) from exc

        winner = str(judgment.get("winner", "")).strip().upper()

        reasoning = str(judgment.get("reasoning", "")).strip()

        if winner not in {
            "A",
            "B",
            "DRAW",
        }:
            raise ValueError(
                "Elo judge must return winner as "
                "'A', 'B', or 'DRAW'. "
                f"Received: {winner!r}"
            )

        return JudgeDecision(
            winner=winner,
            reasoning=reasoning,
        )

    async def judge_match(
        self,
        *,
        task: str,
        player_a: str,
        agent_type_a: str,
        output_a: str,
        player_b: str,
        agent_type_b: str,
        output_b: str,
    ) -> JudgeDecision:
        """Judge one match with fault-tolerant fallback."""

        prompt = self._judge_prompt(
            task=task,
            player_a=player_a,
            agent_type_a=agent_type_a,
            output_a=output_a,
            player_b=player_b,
            agent_type_b=agent_type_b,
            output_b=output_b,
        )

        judge = None

        try:
            try:
                judge = EloJudgeAgent(
                    llm=self.llm,
                    workspace=self.workspace,
                    group=self.group,
                    use_web=False,
                )
                result = await judge.ainvoke(
                    prompt
                    + "\nInspect evidence as needed, then summarize your "
                    "assessment. A final structured judgment step follows.\n"
                )

                formatter = getattr(
                    judge,
                    "format_result",
                    None,
                )

                if callable(formatter):
                    text = str(formatter(result))
                else:
                    text = result_to_text(result)

                decision = self._parse_decision(text)

                return JudgeDecision(
                    winner=decision.winner,
                    reasoning=(decision.reasoning),
                    method="chat_agent",
                )

            except Exception as agent_exc:
                try:
                    fallback = await self._judge_with_llm(prompt)

                    return JudgeDecision(
                        winner=(fallback.winner),
                        reasoning=(
                            "Agentic judge failed "
                            f"({type(agent_exc).__name__}: "
                            f"{agent_exc}). "
                            "Decision obtained from "
                            "raw-LLM fallback. "
                            f"{fallback.reasoning}"
                        ),
                        method="llm_fallback",
                    )

                except Exception as fallback_exc:
                    return JudgeDecision(
                        winner="DRAW",
                        reasoning=(
                            "Judging failed in both "
                            "the ChatAgent judge and "
                            "raw-LLM fallback. "
                            "Match recorded as a draw. "
                            "ChatAgent error: "
                            f"{type(agent_exc).__name__}: "
                            f"{agent_exc}. "
                            "Fallback error: "
                            f"{type(fallback_exc).__name__}: "
                            f"{fallback_exc}."
                        ),
                        method="failed_draw",
                    )

        finally:
            close = getattr(
                judge,
                "close",
                None,
            )

            if callable(close):
                try:
                    close()
                except Exception:
                    logging.getLogger(__name__).warning(
                        "Failed to close Elo judge",
                        exc_info=True,
                    )

    async def _judge_with_llm(
        self,
        prompt: str,
    ) -> JudgeDecision:
        """Fallback to a single raw LLM judgment."""

        response = await self.llm.ainvoke([
            HumanMessage(content=prompt + self.OUTPUT_INSTRUCTIONS)
        ])

        text = result_to_text(response)

        decision = self._parse_decision(text)

        return JudgeDecision(
            winner=decision.winner,
            reasoning=decision.reasoning,
            method="llm_fallback",
        )

judge_match(*, task, player_a, agent_type_a, output_a, player_b, agent_type_b, output_b) async

Judge one match with fault-tolerant fallback.

Source code in src/ursa/environments/agent_elo_judge.py

async def judge_match(
    self,
    *,
    task: str,
    player_a: str,
    agent_type_a: str,
    output_a: str,
    player_b: str,
    agent_type_b: str,
    output_b: str,
) -> JudgeDecision:
    """Judge one match with fault-tolerant fallback."""

    prompt = self._judge_prompt(
        task=task,
        player_a=player_a,
        agent_type_a=agent_type_a,
        output_a=output_a,
        player_b=player_b,
        agent_type_b=agent_type_b,
        output_b=output_b,
    )

    judge = None

    try:
        try:
            judge = EloJudgeAgent(
                llm=self.llm,
                workspace=self.workspace,
                group=self.group,
                use_web=False,
            )
            result = await judge.ainvoke(
                prompt
                + "\nInspect evidence as needed, then summarize your "
                "assessment. A final structured judgment step follows.\n"
            )

            formatter = getattr(
                judge,
                "format_result",
                None,
            )

            if callable(formatter):
                text = str(formatter(result))
            else:
                text = result_to_text(result)

            decision = self._parse_decision(text)

            return JudgeDecision(
                winner=decision.winner,
                reasoning=(decision.reasoning),
                method="chat_agent",
            )

        except Exception as agent_exc:
            try:
                fallback = await self._judge_with_llm(prompt)

                return JudgeDecision(
                    winner=(fallback.winner),
                    reasoning=(
                        "Agentic judge failed "
                        f"({type(agent_exc).__name__}: "
                        f"{agent_exc}). "
                        "Decision obtained from "
                        "raw-LLM fallback. "
                        f"{fallback.reasoning}"
                    ),
                    method="llm_fallback",
                )

            except Exception as fallback_exc:
                return JudgeDecision(
                    winner="DRAW",
                    reasoning=(
                        "Judging failed in both "
                        "the ChatAgent judge and "
                        "raw-LLM fallback. "
                        "Match recorded as a draw. "
                        "ChatAgent error: "
                        f"{type(agent_exc).__name__}: "
                        f"{agent_exc}. "
                        "Fallback error: "
                        f"{type(fallback_exc).__name__}: "
                        f"{fallback_exc}."
                    ),
                    method="failed_draw",
                )

    finally:
        close = getattr(
            judge,
            "close",
            None,
        )

        if callable(close):
            try:
                close()
            except Exception:
                logging.getLogger(__name__).warning(
                    "Failed to close Elo judge",
                    exc_info=True,
                )

EloJudgeAgent

Bases: ChatAgent

Inspect evidence with tools, then produce a structured judgment.

Source code in src/ursa/environments/agent_elo_judge.py

class EloJudgeAgent(ChatAgent):
    """Inspect evidence with tools, then produce a structured judgment."""

    def __init__(self, llm: BaseChatModel, **kwargs):
        # Preserve the unbound model across graph rebuilds.
        self._judgment_llm = llm
        super().__init__(llm=llm, **kwargs)

    def _build_graph(self):
        self.llm = self._judgment_llm.bind_tools(self.tools.values())
        self.add_node(self._response_node, "respond")
        self.add_node(self.tool_node, "tool_node")
        self.add_node(self._final_judgment, "final_judgment")
        self.graph.set_entry_point("respond")
        self.graph.add_conditional_edges(
            "respond",
            self._wrap_cond(should_continue, "should_continue"),
            {"continue": "tool_node", "finish": "final_judgment"},
        )
        self.graph.add_edge("tool_node", "respond")
        self.graph.add_edge("final_judgment", END)

    def _final_judgment(self, state: ChatState, config: RunnableConfig) -> dict:
        # LangGraph runs synchronous nodes in an executor during ainvoke,
        # so the shared synchronous helper does not block the event loop.
        decision = invoke_structured(
            self._judgment_llm,
            Judgment,
            list(state["messages"])
            + [
                HumanMessage(
                    content=(
                        "Return your final judgment using the task's evaluation criteria "
                        "and the submissions and inspected evidence above. Choose A, B, "
                        "or DRAW and give a brief explanation."
                    )
                )
            ],
            config=config,
            context="Elo final judgment",
        )
        return {"messages": [AIMessage(content=decision.model_dump_json())]}

JudgeDecision dataclass

Structured result returned by the Elo judge.

Source code in src/ursa/environments/agent_elo_judge.py

@dataclass(frozen=True)
class JudgeDecision:
    """Structured result returned by the Elo judge."""

    winner: str
    reasoning: str
    method: str = "chat_agent"

Judgment

Bases: BaseModel

Validated final response, separate from internal decision metadata.

Source code in src/ursa/environments/agent_elo_judge.py

class Judgment(BaseModel):
    """Validated final response, separate from internal decision metadata."""

    model_config = ConfigDict(extra="forbid")
    winner: Literal["A", "B", "DRAW"]
    reasoning: str

agent_symposium

AgentSymposiumEnvironment

Bases: BaseEnvironment

Competitive/collaborative multi-agent symposium environment.

A symposium organizer dispatches the same complex problem to several members or nested teams. Members first work independently. The environment then sends all writeups to each member for critical review, explicitly instructing them not to modify reviewed work and to only view/run code as needed for assessment. Finally, each member revises its own work using the feedback it received and insights gained from reviewing others. The organizer synthesizes the final symposium report.

Source code in src/ursa/environments/agent_symposium.py

class AgentSymposiumEnvironment(BaseEnvironment):
    """Competitive/collaborative multi-agent symposium environment.

    A symposium organizer dispatches the same complex problem to several members
    or nested teams. Members first work independently. The environment then sends
    all writeups to each member for critical review, explicitly instructing them
    not to modify reviewed work and to only view/run code as needed for assessment.
    Finally, each member revises its own work using the feedback it received and
    insights gained from reviewing others. The organizer synthesizes the final
    symposium report.
    """

    def __init__(
        self,
        llm: BaseChatModel,
        *,
        config: AgentSymposiumConfig
        | Mapping[str, Any]
        | str
        | Path
        | None = None,
        name: str | None = None,
        group: str | None = None,
        organizer: EnvironmentMemberConfig | Mapping[str, Any] | None = None,
        members: list[EnvironmentMemberConfig | Mapping[str, Any]]
        | None = None,
        workspace: str | Path | None = None,
        revision_rounds: int | None = None,
        persist_members: bool = True,
        **kwargs: Any,
    ):
        symposium_config = self._coerce_config(
            config=config,
            name=name,
            group=group,
            organizer=organizer,
            members=members,
            workspace=workspace,
            revision_rounds=revision_rounds,
        )
        super().__init__(
            llm,
            name=symposium_config.name,
            group=symposium_config.group,
            workspace=symposium_config.workspace or workspace,
            persist_members=persist_members,
            **kwargs,
        )
        self.config = symposium_config
        self.members = {
            member.name: self._build_symposium_participant(member)
            for member in self.config.members
        }
        self.member_configs = {
            member.name: member for member in self.config.members
        }
        self.organizer = self._build_symposium_organizer(self.config.organizer)

    def _events(self, config: Mapping[str, Any] | None) -> EnvironmentEvents:
        return EnvironmentEvents(
            environment=self.name,
            config=config,
            environment_type="agent_symposium",
            environment_id=self.name,
            path=[self.name],
        )

    def _source(self, name: str, *, kind: str = "agent") -> dict[str, Any]:
        return {
            "id": f"{self.name}.{name}",
            "name": name,
            "kind": kind,
            "path": [self.name, name],
        }

    def _environment_source(self) -> dict[str, Any]:
        return {
            "id": self.name,
            "name": self.name,
            "kind": "environment",
            "path": [self.name],
        }

    def _member_node(self, member: EnvironmentMemberConfig) -> dict[str, Any]:
        member_obj = self.members.get(member.name)
        kind = (
            "environment"
            if isinstance(member_obj, BaseEnvironment)
            else "agent"
        )
        return {
            "id": f"{self.name}.{member.name}",
            "name": member.name,
            "kind": kind,
            "role": member.role,
            "agent_class": member.agent,
            "path": [self.name, member.name],
        }

    def _topology_payload(self) -> dict[str, Any]:
        organizer = self.config.organizer
        organizer_kind = (
            "environment"
            if isinstance(self.organizer, BaseEnvironment)
            else "agent"
        )
        nodes = [
            {
                "id": f"{self.name}.{organizer.name}",
                "name": organizer.name,
                "kind": organizer_kind,
                "role": organizer.role,
                "agent_class": organizer.agent,
                "path": [self.name, organizer.name],
            },
            *[self._member_node(member) for member in self.config.members],
        ]
        edges = []
        for member in self.config.members:
            edges.append({
                "source": self.name,
                "target": f"{self.name}.{member.name}",
                "kind": "dispatches_to",
            })
            edges.append({
                "source": f"{self.name}.{member.name}",
                "target": f"{self.name}.{organizer.name}",
                "kind": "synthesized_by",
            })
        return {
            "kind": "agent_symposium",
            "name": self.name,
            "description": self.config.description,
            "revision_rounds": self.config.revision_rounds,
            "nodes": nodes,
            "edges": edges,
        }

    @classmethod
    def from_yaml(
        cls,
        path: str | Path,
        *,
        llm: BaseChatModel,
        **kwargs: Any,
    ) -> "AgentSymposiumEnvironment":
        return cls(llm=llm, config=load_symposium_config(path), **kwargs)

    def _coerce_config(
        self,
        *,
        config: AgentSymposiumConfig | Mapping[str, Any] | str | Path | None,
        name: str | None,
        group: str | None,
        organizer: EnvironmentMemberConfig | Mapping[str, Any] | None,
        members: list[EnvironmentMemberConfig | Mapping[str, Any]] | None,
        workspace: str | Path | None,
        revision_rounds: int | None,
    ) -> AgentSymposiumConfig:
        if isinstance(config, (str, Path)):
            base = load_symposium_config(config)
        elif isinstance(config, Mapping):
            base = AgentSymposiumConfig.from_mapping(config)
        elif isinstance(config, AgentSymposiumConfig):
            base = config
        else:
            organizer_cfg = self._coerce_member(
                organizer
                or {
                    "name": "organizer",
                    "role": "Symposium organizer / final synthesizer",
                    "agent": "ChatAgent",
                }
            )
            member_cfgs = [self._coerce_member(m) for m in (members or [])]
            base = AgentSymposiumConfig(
                name=name or "agent_symposium",
                group=group or "default",
                organizer=organizer_cfg,
                members=member_cfgs,
                workspace=str(workspace) if workspace else None,
                revision_rounds=revision_rounds or 1,
            )
        return AgentSymposiumConfig(
            name=name or base.name,
            group=group or base.group,
            description=base.description,
            organizer=base.organizer,
            members=base.members,
            workspace=str(workspace) if workspace else base.workspace,
            defaults=base.defaults,
            revision_rounds=revision_rounds
            if revision_rounds is not None
            else base.revision_rounds,
        )

    @staticmethod
    def _coerce_member(
        member: EnvironmentMemberConfig | Mapping[str, Any],
    ) -> EnvironmentMemberConfig:
        if isinstance(member, EnvironmentMemberConfig):
            return member
        return EnvironmentMemberConfig.from_mapping(member)

    def _build_symposium_participant(
        self, member: EnvironmentMemberConfig
    ) -> Any:
        """Instantiate a symposium participant in its own child workspace.

        Symposium member names are stable URSA agent names. This lets users
        reuse existing named agents directly from YAML instead of creating new
        ``<symposium>_<member>`` checkpoints.
        """
        return self._build_symposium_agent(
            member,
            workspace=self._member_workspace(member.name),
            agent_name=member.name,
        )

    def _build_symposium_organizer(
        self, member: EnvironmentMemberConfig
    ) -> Any:
        """Instantiate the organizer in the parent symposium workspace."""
        return self._build_symposium_agent(
            member,
            workspace=self.workspace,
            agent_name=member.name,
        )

    def _build_symposium_agent(
        self,
        member: EnvironmentMemberConfig,
        *,
        workspace: Path,
        agent_name: str,
    ) -> Any:
        cls = load_object(member.agent)
        llm = make_llm(self.llm, member.model)
        kwargs = dict(member.config or {})
        kwargs.setdefault("workspace", workspace)
        kwargs.setdefault("group", self.group)
        if isinstance(cls, type) and issubclass(cls, BaseEnvironment):
            kwargs.setdefault("name", member.name)
            kwargs.setdefault("persist_members", self.persist_members)
        elif self.persist_members:
            kwargs.setdefault("agent_name", agent_name)
        return cls(llm=llm, **kwargs)

    def _member_workspace_roster(self) -> str:
        return (
            "\n".join(
                f"- {member.name}: {member.name}/ "
                f"({self._member_workspace(member.name)})"
                for member in self.config.members
            )
            or "No symposium member workspaces configured."
        )

    def _member_roster(self) -> str:
        return (
            "\n".join(
                f"- {member.name}: {member.role} ({member.agent})"
                for member in self.config.members
            )
            or "No symposium members configured."
        )

    def _initial_prompt(
        self, member: EnvironmentMemberConfig, task: str
    ) -> str:
        extra = (
            f"\n\nMember-specific guidance:\n{member.prompt}"
            if member.prompt
            else ""
        )
        return (
            f"You are symposium participant '{member.name}' with role: {member.role}.\n"
            "Work independently on the complex problem below. Produce a detailed, "
            "self-contained writeup of your approach, methods, assumptions, files or "
            "commands used, findings, uncertainty, and reproducibility instructions.\n"
            "Your work will be reviewed by other symposium participants. Clean, "
            "compact organization, documentation, and reproducibility are important "
            "parts of how your contribution will be assessed.\n"
            f"{extra}\n\nProblem:\n{task}"
        )

    def _review_prompt(
        self,
        reviewer: EnvironmentMemberConfig,
        task: str,
        writeups: Mapping[str, str],
    ) -> str:
        writeup_block = "\n\n".join(
            f"## Writeup from {name}\n{writeup}"
            for name, writeup in writeups.items()
        )
        return (
            f"You are symposium reviewer '{reviewer.name}' with role: {reviewer.role}.\n"
            "Read all submitted writeups, including your own if present. Assess the "
            "quality of the findings, compare methods and conclusions, and write a "
            "fair but critical review for each writeup. Discuss strengths, weaknesses, "
            "reproducibility, evidence quality, missing checks, and concrete ways to "
            "improve the solution.\n\n"
            "For this review phase, your workspace is the parent symposium "
            "workspace. Participant artifacts are available in these child "
            "workspace directories, using the relative paths shown:\n"
            f"{self._member_workspace_roster()}\n\n"
            "Important review constraints:\n"
            "- Do not change, edit, overwrite, or reorganize any work you are reviewing.\n"
            "- If code or files are referenced, you may inspect or run them only to assess "
            "correctness/reproducibility.\n"
            "- Make clear which findings are well supported and which are speculative.\n"
            "- Your own work will also be judged on clarity, compact organization, "
            "documentation, and reproducibility.\n\n"
            f"Original problem:\n{task}\n\n"
            f"Submitted writeups:\n{writeup_block}"
        )

    def _revision_prompt(
        self,
        member: EnvironmentMemberConfig,
        task: str,
        own_writeup: str,
        all_writeups: Mapping[str, str],
        reviews: Mapping[str, str],
        round_index: int,
    ) -> str:
        other_writeups = "\n\n".join(
            f"## {name}\n{writeup}" for name, writeup in all_writeups.items()
        )
        review_block = "\n\n".join(
            f"## Review from {name}\n{review}"
            for name, review in reviews.items()
        )
        return (
            f"You are symposium participant '{member.name}' revising your own work "
            f"after review round {round_index}.\n"
            "Use the feedback you received and anything you learned from reviewing "
            "other submissions to improve your own solution. You may change only your "
            "own work/artifacts. Do not modify the work of other symposium members.\n"
            "Return a revised detailed writeup with clear improvements, evidence, "
            "limitations, and reproducibility instructions.\n\n"
            f"Original problem:\n{task}\n\n"
            f"Your previous writeup:\n{own_writeup}\n\n"
            f"All writeups you saw:\n{other_writeups}\n\n"
            f"Reviews from symposium members:\n{review_block}"
        )

    def _synthesis_prompt(
        self,
        task: str,
        writeups: Mapping[str, str],
        reviews: Mapping[str, str],
    ) -> str:
        description = (
            f"\nSymposium description: {self.config.description}\n"
            if self.config.description
            else ""
        )
        writeup_block = "\n\n".join(
            f"## Final writeup from {name}\n{writeup}"
            for name, writeup in writeups.items()
        )
        review_block = "\n\n".join(
            f"## Review by {name}\n{review}" for name, review in reviews.items()
        )
        return (
            "You are the organizer of an URSA Agent Symposium. Synthesize the final "
            "results for the user after independent work, peer review, and revision. "
            "Compare participant outputs, identify consensus and disagreement, judge "
            "evidence quality, and provide a final recommendation or solution.\n"
            f"{description}\n"
            "Symposium members:\n"
            f"{self._member_roster()}\n\n"
            "The organizer workspace is the parent symposium workspace. Final "
            "member artifacts are available in these child workspace directories:\n"
            f"{self._member_workspace_roster()}\n\n"
            f"Original problem:\n{task}\n\n"
            f"Final participant writeups:\n{writeup_block}\n\n"
            f"Peer reviews:\n{review_block}"
        )

    def _invoke(
        self, inputs: Mapping[str, Any], **config: Any
    ) -> dict[str, Any]:
        return self._run_ainvoke_from_sync(inputs, **config)

    async def _member_writeup(
        self,
        member_config: EnvironmentMemberConfig,
        prompt: str,
        invoke_kwargs: Mapping[str, Any],
        events: EnvironmentEvents,
        *,
        workspace: Path | None = None,
        event_prefix: str,
        phase_name: str,
        round_index: int | None = None,
    ) -> tuple[str, str]:
        member_name = member_config.name
        source = self._source(member_name)
        start = perf_counter()
        await events.aemit(
            f"{member_name} started {phase_name}",
            stage="symposium",
            phase=phase_name,
            event_type=f"{event_prefix}_started",
            source=source,
            target=self._environment_source(),
            member=member_name,
            round_index=round_index,
            prompt=prompt,
        )
        member = self.members[member_name]
        try:
            result = await self._invoke_member_with_workspace_async(
                member, prompt, workspace=workspace, **invoke_kwargs
            )
        except BaseException as exc:
            await events.aemit(
                f"{member_name} failed {phase_name}",
                stage="symposium",
                phase=phase_name,
                event_type=f"{event_prefix}_failed",
                level="error",
                source=source,
                target=self._environment_source(),
                member=member_name,
                round_index=round_index,
                error=str(exc),
                elapsed_seconds=perf_counter() - start,
            )
            raise
        text = result_to_text(result)
        await events.aemit(
            f"{member_name} completed {phase_name}",
            stage="symposium",
            phase=phase_name,
            event_type=f"{event_prefix}_completed",
            source=source,
            target=self._environment_source(),
            member=member_name,
            round_index=round_index,
            result=text,
            elapsed_seconds=perf_counter() - start,
        )
        return member_name, text

    async def _invoke_member_with_workspace_async(
        self,
        member: Any,
        prompt: str,
        *,
        workspace: Path | None = None,
        **kwargs: Any,
    ) -> Any:
        """Invoke a member, optionally using a temporary workspace.

        Participants normally work in their own child workspaces. During review,
        reviewers are temporarily run from the parent symposium workspace so
        file tools can inspect every participant's child workspace. The original
        workspace is restored after the invocation.
        """
        if workspace is None or not hasattr(member, "workspace"):
            return await self._invoke_member_async(member, prompt, **kwargs)

        original_workspace = getattr(member, "workspace")
        setattr(member, "workspace", Path(workspace))
        Path(workspace).mkdir(parents=True, exist_ok=True)
        try:
            return await self._invoke_member_async(member, prompt, **kwargs)
        finally:
            setattr(member, "workspace", original_workspace)

    async def _ainvoke(
        self, inputs: Mapping[str, Any], **config: Any
    ) -> dict[str, Any]:
        task = str(inputs.get("task") or inputs.get("prompt") or inputs)
        if not self.config.members:
            raise ValueError(
                "AgentSymposiumEnvironment requires at least one member."
            )

        runtime_config = runnable_config_from_kwargs(config)
        events = self._events(runtime_config)
        invoke_kwargs = invocation_kwargs(config)
        start = perf_counter()
        await events.aemit(
            f"Agent symposium {self.name} started",
            stage="symposium",
            phase="start",
            event_type="symposium_started",
            task=task,
            topology=self._topology_payload(),
        )
        await events.aemit(
            f"Agent symposium {self.name} topology declared",
            stage="symposium",
            phase="topology",
            event_type="topology_declared",
            topology=self._topology_payload(),
        )
        try:
            await events.aemit(
                "Initial symposium work started",
                stage="symposium",
                phase="initial_work",
                event_type="symposium_phase_started",
                task=task,
            )
            initial_pairs = await asyncio.gather(*[
                self._member_writeup(
                    member_config,
                    self._initial_prompt(member_config, task),
                    invoke_kwargs,
                    events,
                    event_prefix="initial_work",
                    phase_name="initial_work",
                )
                for member_config in self.config.members
            ])
            initial_writeups = dict(initial_pairs)
            await events.aemit(
                "Initial symposium work completed",
                stage="symposium",
                phase="initial_work",
                event_type="symposium_phase_completed",
                result=initial_writeups,
            )

            current_writeups = dict(initial_writeups)
            review_rounds: list[dict[str, str]] = []
            latest_reviews: dict[str, str] = {}
            reviewer_configs = [
                member for member in self.config.members if member.reviewer
            ]

            for round_index in range(
                1, max(1, self.config.revision_rounds) + 1
            ):
                await events.aemit(
                    f"Review round {round_index} started",
                    stage="symposium",
                    phase="review",
                    event_type="review_round_started",
                    round_index=round_index,
                    writeups=current_writeups,
                )
                review_pairs = await asyncio.gather(*[
                    self._member_writeup(
                        reviewer_config,
                        self._review_prompt(
                            reviewer_config, task, current_writeups
                        ),
                        invoke_kwargs,
                        events,
                        event_prefix="review",
                        phase_name="review",
                        round_index=round_index,
                        workspace=self.workspace,
                    )
                    for reviewer_config in reviewer_configs
                ])
                round_reviews = dict(review_pairs)
                latest_reviews = round_reviews
                review_rounds.append(round_reviews)
                await events.aemit(
                    f"Review round {round_index} completed",
                    stage="symposium",
                    phase="review",
                    event_type="review_round_completed",
                    round_index=round_index,
                    reviews=round_reviews,
                )

                await events.aemit(
                    f"Revision round {round_index} started",
                    stage="symposium",
                    phase="revision",
                    event_type="revision_round_started",
                    round_index=round_index,
                    reviews=round_reviews,
                )
                revision_pairs = await asyncio.gather(*[
                    self._member_writeup(
                        member_config,
                        self._revision_prompt(
                            member_config,
                            task,
                            current_writeups[member_config.name],
                            current_writeups,
                            round_reviews,
                            round_index,
                        ),
                        invoke_kwargs,
                        events,
                        event_prefix="revision",
                        phase_name="revision",
                        round_index=round_index,
                    )
                    for member_config in self.config.members
                ])
                current_writeups = dict(revision_pairs)
                await events.aemit(
                    f"Revision round {round_index} completed",
                    stage="symposium",
                    phase="revision",
                    event_type="revision_round_completed",
                    round_index=round_index,
                    writeups=current_writeups,
                )

            synthesis_prompt = self._synthesis_prompt(
                task, current_writeups, latest_reviews
            )
            synth_start = perf_counter()
            await events.aemit(
                "Symposium synthesis started",
                stage="symposium",
                phase="synthesis",
                event_type="synthesis_started",
                source=self._source(self.config.organizer.name),
                task=task,
                prompt=synthesis_prompt,
            )
            organizer_result = await self._invoke_member_async(
                self.organizer,
                synthesis_prompt,
                **invoke_kwargs,
            )
            final = result_to_text(organizer_result)
            await events.aemit(
                "Symposium synthesis completed",
                stage="symposium",
                phase="synthesis",
                event_type="synthesis_completed",
                source=self._source(self.config.organizer.name),
                result=final,
                elapsed_seconds=perf_counter() - synth_start,
            )
            result = {
                "task": task,
                "initial_writeups": initial_writeups,
                "review_rounds": review_rounds,
                "reviews": latest_reviews,
                "final_writeups": current_writeups,
                "organizer_result": organizer_result,
                "final": final,
            }
        except BaseException as exc:
            await events.aemit(
                f"Agent symposium {self.name} failed",
                stage="symposium",
                phase="error",
                event_type="symposium_failed",
                level="error",
                task=task,
                error=str(exc),
                elapsed_seconds=perf_counter() - start,
            )
            raise
        await events.aemit(
            f"Agent symposium {self.name} completed",
            stage="symposium",
            phase="end",
            event_type="symposium_completed",
            task=task,
            result=result["final"],
            elapsed_seconds=perf_counter() - start,
        )
        return result

agent_team

AgentTeamEnvironment

Bases: BaseEnvironment

Hierarchical multi-agent team coordinated by a PI agent.

The PI is user-facing. Team members are exposed to the PI as tools, one tool per member, so the PI can plan, delegate, compare returned work, ask follow-up questions, and synthesize a final answer. The environment itself exposes a normal invoke method and can therefore be nested inside other environments such as an Agent Symposium.

Source code in src/ursa/environments/agent_team.py

class AgentTeamEnvironment(BaseEnvironment):
    """Hierarchical multi-agent team coordinated by a PI agent.

    The PI is user-facing. Team members are exposed to the PI as tools, one tool
    per member, so the PI can plan, delegate, compare returned work, ask follow-up
    questions, and synthesize a final answer. The environment itself exposes a
    normal ``invoke`` method and can therefore be nested inside other environments
    such as an Agent Symposium.
    """

    def __init__(
        self,
        llm: BaseChatModel,
        *,
        config: AgentTeamConfig | Mapping[str, Any] | str | Path | None = None,
        name: str | None = None,
        group: str | None = None,
        pi: EnvironmentMemberConfig | Mapping[str, Any] | None = None,
        members: list[EnvironmentMemberConfig | Mapping[str, Any]]
        | None = None,
        workspace: str | Path | None = None,
        persist_members: bool = True,
        trace_delegation: bool = True,
        trace_character_limit: int = 4000,
        **kwargs: Any,
    ):
        team_config = self._coerce_config(
            config=config,
            name=name,
            group=group,
            pi=pi,
            members=members,
            workspace=workspace,
        )
        super().__init__(
            llm,
            name=team_config.name,
            group=team_config.group,
            workspace=team_config.workspace or workspace,
            persist_members=persist_members,
            **kwargs,
        )
        self.config = team_config
        self.trace_delegation = trace_delegation
        self.trace_character_limit = trace_character_limit
        self.members = {
            member.name: self._build_team_member(member)
            for member in self.config.members
        }
        self.member_configs = {
            member.name: member for member in self.config.members
        }
        self.pi = self._build_pi()

    @classmethod
    def from_yaml(
        cls,
        path: str | Path,
        *,
        llm: BaseChatModel,
        **kwargs: Any,
    ) -> "AgentTeamEnvironment":
        return cls(llm=llm, config=load_team_config(path), **kwargs)

    def _coerce_config(
        self,
        *,
        config: AgentTeamConfig | Mapping[str, Any] | str | Path | None,
        name: str | None,
        group: str | None,
        pi: EnvironmentMemberConfig | Mapping[str, Any] | None,
        members: list[EnvironmentMemberConfig | Mapping[str, Any]] | None,
        workspace: str | Path | None,
    ) -> AgentTeamConfig:
        if isinstance(config, (str, Path)):
            base = load_team_config(config)
        elif isinstance(config, Mapping):
            base = AgentTeamConfig.from_mapping(config)
        elif isinstance(config, AgentTeamConfig):
            base = config
        else:
            pi_cfg = self._coerce_member(
                pi
                or {
                    "name": "pi",
                    "role": "Principal investigator / team lead",
                    "agent": "ExecutionAgent",
                }
            )
            member_cfgs = [self._coerce_member(m) for m in (members or [])]
            base = AgentTeamConfig(
                name=name or "agent_team",
                group=group or "default",
                pi=pi_cfg,
                members=member_cfgs,
                workspace=str(workspace) if workspace else None,
            )
        return AgentTeamConfig(
            name=name or base.name,
            group=group or base.group,
            description=base.description,
            pi=base.pi,
            members=base.members,
            workspace=str(workspace) if workspace else base.workspace,
            defaults=base.defaults,
        )

    @staticmethod
    def _coerce_member(
        member: EnvironmentMemberConfig | Mapping[str, Any],
    ) -> EnvironmentMemberConfig:
        if isinstance(member, EnvironmentMemberConfig):
            return member
        return EnvironmentMemberConfig.from_mapping(member)

    def _build_team_member(self, member: EnvironmentMemberConfig) -> Any:
        """Instantiate a team member in the shared team workspace.

        Team member names are stable URSA agent names. This lets users reuse
        existing named agents directly from YAML, e.g. ``team_nif_expert`` rather
        than creating a new ``<team>_<member>`` checkpoint. All team members see
        the same team workspace so they can collaborate through shared files.
        """
        return self._build_named_team_agent(member, agent_name=member.name)

    def _build_team_pi(self, member: EnvironmentMemberConfig) -> Any:
        """Instantiate the PI in the shared team workspace.

        The PI also uses its configured name directly. If that named PI does not
        yet exist, BaseAgent will create its checkpoint under the usual URSA agent
        cache for the configured group.
        """
        return self._build_named_team_agent(member, agent_name=member.name)

    def _build_named_team_agent(
        self,
        member: EnvironmentMemberConfig,
        *,
        agent_name: str,
    ) -> Any:
        cls = load_object(member.agent)
        llm = make_llm(self.llm, member.model)
        kwargs = dict(member.config or {})
        kwargs.setdefault("workspace", self.workspace)
        kwargs.setdefault("group", self.group)
        if isinstance(cls, type) and issubclass(cls, BaseEnvironment):
            kwargs.setdefault("name", member.name)
            kwargs.setdefault("persist_members", self.persist_members)
        elif self.persist_members:
            kwargs.setdefault("agent_name", agent_name)
        return cls(llm=llm, **kwargs)

    def _build_pi(self) -> Any:
        pi_config = self.config.pi
        cls_kwargs = dict(pi_config.config or {})
        delegation_tools = [
            self._make_delegation_tool(member_config)
            for member_config in self.config.members
        ]

        # ExecutionAgent explicitly supports `extra_tools`, making it the preferred
        # PI implementation. Other AgentWithTools-style agents can receive tools
        # after construction via `add_tool` if they expose that method.
        pass_extra_tools = (
            pi_config.agent.endswith("ExecutionAgent")
            or pi_config.agent == "ExecutionAgent"
        )
        if pass_extra_tools:
            cls_kwargs.setdefault("extra_tools", delegation_tools)

        pi_member = EnvironmentMemberConfig(
            name=pi_config.name,
            role=pi_config.role,
            agent=pi_config.agent,
            model=pi_config.model,
            config=cls_kwargs,
            prompt=pi_config.prompt,
        )
        pi_agent = self._build_team_pi(pi_member)
        if (
            not pass_extra_tools
            and delegation_tools
            and hasattr(pi_agent, "add_tool")
        ):
            pi_agent.add_tool(delegation_tools)
        elif not pass_extra_tools and delegation_tools:
            raise TypeError(
                f"Configured PI agent {pi_config.agent!r} cannot accept delegation tools. "
                "Use ExecutionAgent or another AgentWithTools-compatible agent."
            )
        return pi_agent

    def _events(self, config: Mapping[str, Any] | None) -> EnvironmentEvents:
        return EnvironmentEvents(
            environment=self.name,
            config=config,
            environment_type="agent_team",
            environment_id=self.name,
            path=[self.name],
        )

    def _source(self, name: str, *, kind: str = "agent") -> dict[str, Any]:
        return {
            "id": f"{self.name}.{name}",
            "name": name,
            "kind": kind,
            "path": [self.name, name],
        }

    def _member_runtime_config(
        self,
        base_config: Mapping[str, Any] | None,
        member: EnvironmentMemberConfig,
    ) -> dict[str, Any] | None:
        """Attach stable team-member identity to nested agent/tool events."""
        if base_config is None:
            return None
        member_id = f"{self.name}.{member.name}"
        merged = dict(base_config)
        base_metadata = merged.get("metadata")
        metadata = (
            dict(base_metadata) if isinstance(base_metadata, Mapping) else {}
        )
        metadata.update({
            "environment_id": self.name,
            "environment_member": member.name,
            "environment_member_id": member_id,
            "environment_member_role": member.role,
            "environment_member_path": [self.name, member.name],
            "agent": member.name,
            "agent_id": member_id,
        })
        merged["metadata"] = metadata

        base_tags = merged.get("tags")
        if isinstance(base_tags, str):
            tags = [base_tags]
        else:
            tags = list(base_tags) if base_tags else []
        for tag in (member.name, member_id, "environment_member"):
            if tag not in tags:
                tags.append(tag)
        merged["tags"] = tags
        return merged

    def _member_node(self, member: EnvironmentMemberConfig) -> dict[str, Any]:
        kind = (
            "environment"
            if isinstance(self.members.get(member.name), BaseEnvironment)
            else "agent"
        )
        return {
            "id": f"{self.name}.{member.name}",
            "name": member.name,
            "kind": kind,
            "role": member.role,
            "agent_class": member.agent,
            "path": [self.name, member.name],
        }

    def _topology_payload(self) -> dict[str, Any]:
        pi = self.config.pi
        nodes = [
            {
                "id": f"{self.name}.{pi.name}",
                "name": pi.name,
                "kind": "agent",
                "role": pi.role,
                "agent_class": pi.agent,
                "path": [self.name, pi.name],
            },
            *[self._member_node(member) for member in self.config.members],
        ]
        return {
            "kind": "agent_team",
            "name": self.name,
            "description": self.config.description,
            "nodes": nodes,
            "edges": [
                {
                    "source": f"{self.name}.{pi.name}",
                    "target": f"{self.name}.{member.name}",
                    "kind": "delegates_to",
                }
                for member in self.config.members
            ],
        }

    def _make_delegation_tool(
        self, member: EnvironmentMemberConfig
    ) -> StructuredTool:
        member_name = member.name
        role = member.role
        tool_name = f"delegate_to_{_slug_tool_name(member_name)}"

        def delegate(task: str, context: str = "") -> str:
            prompt = self._delegation_prompt(member, task=task, context=context)
            runtime_config = current_environment_config()
            events = self._events(runtime_config)
            source = self._source(self.config.pi.name)
            target = self._source(member_name)
            start = perf_counter()
            events.emit(
                f"Delegating to {member_name}",
                stage="delegation",
                phase="start",
                event_type="delegation_started",
                source=source,
                target=target,
                task=task,
                context=context,
                prompt=prompt,
            )
            self._trace_delegation(
                f"PI -> {member_name}",
                f"Task:\n{task}\n\nContext:\n{context or 'No additional context provided.'}",
            )
            try:
                member_config = self._member_runtime_config(
                    runtime_config, member
                )
                kwargs = {"config": member_config} if member_config else {}
                result = self.members[member_name].invoke(prompt, **kwargs)
            except BaseException as exc:
                events.emit(
                    f"Delegation to {member_name} failed",
                    stage="delegation",
                    phase="error",
                    event_type="delegation_failed",
                    level="error",
                    source=source,
                    target=target,
                    task=task,
                    context=context,
                    error=str(exc),
                    elapsed_seconds=perf_counter() - start,
                )
                raise
            text = result_to_text(result)
            events.emit(
                f"Delegation to {member_name} completed",
                stage="delegation",
                phase="end",
                event_type="delegation_completed",
                source=target,
                target=source,
                task=task,
                context=context,
                result=text,
                elapsed_seconds=perf_counter() - start,
            )
            self._trace_delegation(f"{member_name} -> PI", text)
            return text

        async def adelegate(task: str, context: str = "") -> str:
            prompt = self._delegation_prompt(member, task=task, context=context)
            runtime_config = current_environment_config()
            events = self._events(runtime_config)
            source = self._source(self.config.pi.name)
            target = self._source(member_name)
            start = perf_counter()
            await events.aemit(
                f"Delegating to {member_name}",
                stage="delegation",
                phase="start",
                event_type="delegation_started",
                source=source,
                target=target,
                task=task,
                context=context,
                prompt=prompt,
            )
            self._trace_delegation(
                f"PI -> {member_name}",
                f"Task:\n{task}\n\nContext:\n{context or 'No additional context provided.'}",
            )
            try:
                member_config = self._member_runtime_config(
                    runtime_config, member
                )
                kwargs = {"config": member_config} if member_config else {}
                result = await self._invoke_member_async(
                    self.members[member_name], prompt, **kwargs
                )
            except BaseException as exc:
                await events.aemit(
                    f"Delegation to {member_name} failed",
                    stage="delegation",
                    phase="error",
                    event_type="delegation_failed",
                    level="error",
                    source=source,
                    target=target,
                    task=task,
                    context=context,
                    error=str(exc),
                    elapsed_seconds=perf_counter() - start,
                )
                raise
            text = result_to_text(result)
            await events.aemit(
                f"Delegation to {member_name} completed",
                stage="delegation",
                phase="end",
                event_type="delegation_completed",
                source=target,
                target=source,
                task=task,
                context=context,
                result=text,
                elapsed_seconds=perf_counter() - start,
            )
            self._trace_delegation(f"{member_name} -> PI", text)
            return text

        return StructuredTool.from_function(
            func=delegate,
            coroutine=adelegate,
            name=tool_name,
            description=(
                f"Delegate work to team member '{member_name}' ({role}). "
                "Use this when that member's specialty is relevant. Provide a "
                "self-contained task and any context needed for independent work."
            ),
            args_schema=DelegateInput,
        )

    def _trace_delegation(self, label: str, message: str) -> None:
        """Log a small, explicit delegation trace."""
        if not self.trace_delegation:
            return
        text = message
        if (
            self.trace_character_limit > 0
            and len(text) > self.trace_character_limit
        ):
            text = text[: self.trace_character_limit] + "\n... [truncated]"
        logger.info(f"\n[AgentTeam:{self.name}] {label}\n{text}\n")

    def _delegation_prompt(
        self,
        member: EnvironmentMemberConfig,
        *,
        task: str,
        context: str,
    ) -> str:
        guidance = (
            f"\n\nMember-specific guidance:\n{member.prompt}"
            if member.prompt
            else ""
        )
        return (
            f"You are acting as team member '{member.name}' with role: {member.role}.\n"
            "You have been delegated a task by the team PI. Complete the delegated "
            "task thoroughly, using your available tools when appropriate, and return "
            "a clear writeup of methods, evidence, outputs, limitations, and any files "
            "created or commands needed to reproduce the work.\n"
            f"{guidance}\n\n"
            f"Overall context:\n{context or 'No additional context provided.'}\n\n"
            f"Delegated task:\n{task}"
        )

    def _team_roster(self) -> str:
        if not self.config.members:
            return "No team members are configured. Solve directly as PI."
        return "\n".join(
            f"- {member.name}: {member.role} ({member.agent})"
            for member in self.config.members
        )

    def _pi_prompt(self, task: str) -> str:
        description = (
            f"\nTeam description: {self.config.description}\n"
            if self.config.description
            else ""
        )
        pi_extra = (
            f"\nPI-specific guidance:\n{self.config.pi.prompt}\n"
            if self.config.pi.prompt
            else ""
        )
        return (
            "You are the PI/team leader of a hierarchical Agent Team. "
            "You are the user-facing coordinator responsible for satisfying the "
            "user's overall goal. Formulate an approach, decide which team members "
            "to assign work to, call member-delegation tools as needed, review their "
            "returns critically, request follow-up work if necessary, and synthesize "
            "a final answer for the user.\n"
            "Do not claim a member completed work unless you have delegated it and "
            "reviewed the result. You are personally responsible for final "
            "organization and presentation: integrate the delegated work into one "
            "clean, coherent, easily shareable answer with a clear structure, "
            "actionable conclusions, supporting evidence, limitations, and "
            "reproducibility details where relevant.\n"
            f"{description}{pi_extra}\n"
            "Available team members:\n"
            f"{self._team_roster()}\n\n"
            f"User task:\n{task}"
        )

    def _invoke(self, inputs: Mapping[str, Any], **config: Any) -> Any:
        return self._run_ainvoke_from_sync(inputs, **config)

    async def _ainvoke(self, inputs: Mapping[str, Any], **config: Any) -> Any:
        task = str(inputs.get("task") or inputs.get("prompt") or inputs)
        runtime_config = runnable_config_from_kwargs(config)
        events = self._events(runtime_config)
        token = bind_current_environment_config(runtime_config)
        start = perf_counter()
        await events.aemit(
            f"Agent team {self.name} started",
            stage="team",
            phase="start",
            event_type="team_started",
            task=task,
            topology=self._topology_payload(),
        )
        await events.aemit(
            f"Agent team {self.name} topology declared",
            stage="team",
            phase="topology",
            event_type="topology_declared",
            topology=self._topology_payload(),
        )
        try:
            pi_kwargs = invocation_kwargs(config)
            pi_runtime_config = self._member_runtime_config(
                runtime_config, self.config.pi
            )
            if pi_runtime_config is not None:
                pi_kwargs["config"] = pi_runtime_config
            result = await self._invoke_member_async(
                self.pi, self._pi_prompt(task), **pi_kwargs
            )
        except BaseException as exc:
            await events.aemit(
                f"Agent team {self.name} failed",
                stage="team",
                phase="error",
                event_type="team_failed",
                level="error",
                task=task,
                error=str(exc),
                elapsed_seconds=perf_counter() - start,
            )
            raise
        finally:
            reset_current_environment_config(token)
        await events.aemit(
            f"Agent team {self.name} completed",
            stage="team",
            phase="end",
            event_type="team_completed",
            task=task,
            result=result_to_text(result),
            elapsed_seconds=perf_counter() - start,
        )
        return result

DelegateInput

Bases: BaseModel

Input schema for PI-to-member delegation tools.

Source code in src/ursa/environments/agent_team.py

class DelegateInput(BaseModel):
    """Input schema for PI-to-member delegation tools."""

    task: str = Field(
        ...,
        description="Specific task or question to delegate to this team member.",
    )
    context: str = Field(
        "",
        description="Relevant global context, constraints, prior results, or success criteria.",
    )

base

BaseEnvironment

Bases: BaseWorkflow

Base class for multi-agent URSA environments.

Environments compose agents and/or other environments while exposing the same simple invoke surface used by workflows. They deliberately keep persistent configuration separate from agent graph checkpoints: environment definitions live under ~/.cache/ursa/<group>/environments/, while member agents use the shared ~/.cache/ursa/<group>/agents/<agent_name> persistence mechanism.

Source code in src/ursa/environments/base.py

class BaseEnvironment(BaseWorkflow):
    """Base class for multi-agent URSA environments.

    Environments compose agents and/or other environments while exposing the same
    simple ``invoke`` surface used by workflows. They deliberately keep persistent
    configuration separate from agent graph checkpoints: environment definitions live
    under ``~/.cache/ursa/<group>/environments/``, while member agents use the
    shared ``~/.cache/ursa/<group>/agents/<agent_name>`` persistence mechanism.
    """

    def __init__(
        self,
        llm: BaseChatModel,
        *,
        name: str,
        group: str = "default",
        workspace: str | Path | None = None,
        persist_members: bool = True,
        **kwargs: Any,
    ):
        super().__init__(**kwargs)
        self.llm = llm
        self.name = name
        self.group = validate_group_name(group)
        self.workspace = Path(
            workspace
            or group_environments_dir(self.group) / "workspaces" / name
        )
        self.workspace.mkdir(parents=True, exist_ok=True)
        self.persist_members = persist_members

    def _normalize_inputs(self, inputs: InputLike) -> Mapping[str, Any]:
        if isinstance(inputs, str):
            return {"task": inputs}
        if isinstance(inputs, Mapping):
            return inputs
        raise TypeError(f"Unsupported input type: {type(inputs)}")

    def _member_workspace(self, member_name: str) -> Path:
        return self.workspace / member_name

    def _member_agent_name(self, member_name: str) -> str | None:
        if not self.persist_members:
            return None
        return f"{self.name}_{member_name}"

    def build_member(self, member: EnvironmentMemberConfig) -> Any:
        """Instantiate a configured member agent or nested environment.

        BaseAgent subclasses use ``agent_name`` for checkpoint persistence, while
        nested environments use ``name`` and manage their own member persistence.
        The distinction keeps teams usable as symposium members without requiring
        environment constructors to accept BaseAgent-specific keywords.
        """
        cls = load_object(member.agent)
        llm = make_llm(self.llm, member.model)
        kwargs = dict(member.config or {})
        kwargs.setdefault("workspace", self._member_workspace(member.name))
        kwargs.setdefault("group", self.group)
        if isinstance(cls, type) and issubclass(cls, BaseEnvironment):
            kwargs.setdefault("name", member.name)
            kwargs.setdefault("persist_members", self.persist_members)
        elif self.persist_members:
            kwargs.setdefault(
                "agent_name", self._member_agent_name(member.name)
            )
        return cls(llm=llm, **kwargs)

    def _run_ainvoke_from_sync(
        self, inputs: Mapping[str, Any], **config: Any
    ) -> Any:
        """Run this environment's async implementation for sync callers.

        Environments are natively async internally so nested agents/tools can use
        async-only implementations. The public ``invoke`` surface remains useful
        for scripts by creating an event loop at the outer boundary. If a loop is
        already running, callers must use ``await environment.ainvoke(...)``.
        """
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return asyncio.run(self._ainvoke(inputs, **config))

        raise RuntimeError(
            "This environment uses async execution internally, but `.invoke()` "
            "was called from an async context. Use "
            "`await environment.ainvoke(...)` instead."
        )

    async def _invoke_member_async(
        self, member: Any, prompt: str, **kwargs: Any
    ) -> Any:
        """Invoke a member through its async API when available.

        BaseAgent and BaseEnvironment instances expose ``ainvoke``. Lightweight
        test doubles or custom sync-only members may expose only ``invoke``; run
        those in a worker thread so async environment phases remain non-blocking.
        """
        ainvoke = getattr(member, "ainvoke", None)
        if callable(ainvoke):
            result = ainvoke(prompt, **kwargs)
            if inspect.isawaitable(result):
                return await result
            return result

        invoke = getattr(member, "invoke")
        return await asyncio.to_thread(invoke, prompt, **kwargs)

build_member(member)

Instantiate a configured member agent or nested environment.

BaseAgent subclasses use agent_name for checkpoint persistence, while nested environments use name and manage their own member persistence. The distinction keeps teams usable as symposium members without requiring environment constructors to accept BaseAgent-specific keywords.

Source code in src/ursa/environments/base.py

def build_member(self, member: EnvironmentMemberConfig) -> Any:
    """Instantiate a configured member agent or nested environment.

    BaseAgent subclasses use ``agent_name`` for checkpoint persistence, while
    nested environments use ``name`` and manage their own member persistence.
    The distinction keeps teams usable as symposium members without requiring
    environment constructors to accept BaseAgent-specific keywords.
    """
    cls = load_object(member.agent)
    llm = make_llm(self.llm, member.model)
    kwargs = dict(member.config or {})
    kwargs.setdefault("workspace", self._member_workspace(member.name))
    kwargs.setdefault("group", self.group)
    if isinstance(cls, type) and issubclass(cls, BaseEnvironment):
        kwargs.setdefault("name", member.name)
        kwargs.setdefault("persist_members", self.persist_members)
    elif self.persist_members:
        kwargs.setdefault(
            "agent_name", self._member_agent_name(member.name)
        )
    return cls(llm=llm, **kwargs)

bind_current_environment_config(config)

Bind runnable config for delegation tools created outside the run scope.

Source code in src/ursa/environments/base.py

def bind_current_environment_config(
    config: RunnableConfig | None,
) -> Token[RunnableConfig | None]:
    """Bind runnable config for delegation tools created outside the run scope."""
    return _CURRENT_ENVIRONMENT_CONFIG.set(config)

current_environment_config()

Return the runnable config bound to the current environment invocation.

Source code in src/ursa/environments/base.py

def current_environment_config() -> RunnableConfig | None:
    """Return the runnable config bound to the current environment invocation."""
    return _CURRENT_ENVIRONMENT_CONFIG.get()

invocation_kwargs(config)

Drop empty control kwargs before forwarding nested invocations.

Source code in src/ursa/environments/base.py

def invocation_kwargs(config: Mapping[str, Any]) -> dict[str, Any]:
    """Drop empty control kwargs before forwarding nested invocations."""
    return {key: value for key, value in config.items() if value is not None}

result_to_text(result)

Best-effort extraction of human-readable text from an agent result.

Source code in src/ursa/environments/base.py

def result_to_text(result: Any) -> str:
    """Best-effort extraction of human-readable text from an agent result."""
    if result is None:
        return ""
    if isinstance(result, str):
        return result
    if isinstance(result, Mapping):
        if "prompt" in result and isinstance(result["prompt"], str):
            return result["prompt"]
        if "final" in result and isinstance(result["final"], str):
            return result["final"]
        if "result" in result and isinstance(result["result"], str):
            return result["result"]
        messages = result.get("messages")
        if isinstance(messages, list) and messages:
            last = messages[-1]
            text = getattr(last, "text", None)
            if text:
                return str(text)
            content = getattr(last, "content", None)
            if content:
                return str(content)
    text = getattr(result, "text", None)
    if text:
        return str(text)
    content = getattr(result, "content", None)
    if content:
        return str(content)
    return str(result)

runnable_config_from_kwargs(config)

Extract the LangChain runnable config from environment kwargs.

Source code in src/ursa/environments/base.py

def runnable_config_from_kwargs(
    config: Mapping[str, Any],
) -> RunnableConfig | None:
    """Extract the LangChain runnable config from environment kwargs."""
    value = config.get("config")
    if isinstance(value, Mapping):
        return dict(value)
    return None

config

AgentEloConfig

Bases: BaseModel

Validated configuration for a new or resumed Elo environment.

Source code in src/ursa/environments/config.py

class AgentEloConfig(BaseModel):
    """Validated configuration for a new or resumed Elo environment."""

    model_config = ConfigDict(
        frozen=True, extra="forbid", validate_default=True
    )

    name: str
    group: str = "default"
    description: str | None = None
    inference_providers: dict[str, InferenceProviderConfig] = Field(
        default_factory=dict
    )
    members: list[EnvironmentMemberConfig] = Field(default_factory=list)
    workspace: str | None = None
    defaults: dict[str, Any] = Field(default_factory=dict)
    initial_rating: float = Field(default=1500.0, allow_inf_nan=False)
    k_factor: float = Field(default=32.0, gt=0, allow_inf_nan=False)
    deaths_per_round: int = Field(default=1, ge=0)
    seed: int | None = None
    generations: int = Field(default=1, ge=1)
    restart_from_json: str | None = None
    member_timeout_seconds: float | None = Field(
        default=None, gt=0, allow_inf_nan=False
    )
    judge_prompt: str | None = None

    @field_validator("workspace", "restart_from_json", mode="before")
    @classmethod
    def _path_to_string(cls, value: Any) -> Any:
        return str(value) if isinstance(value, Path) else value

    @model_validator(mode="before")
    @classmethod
    def _resolve_members(cls, value: Any) -> Any:
        if not isinstance(value, Mapping):
            return value
        raw = dict(value)
        group = raw.get("group")
        if group is not None and not isinstance(group, str):
            raise ValueError("Group name must be a string")
        raw["group"] = validate_group_name(group)
        provider_data = raw.get("inference_providers") or {}
        if not isinstance(provider_data, Mapping):
            raise ValueError("inference_providers must be a mapping")
        providers = _inference_providers(provider_data)
        raw["inference_providers"] = providers
        members = raw.get("members", [])
        if isinstance(members, (list, tuple)):
            resolved = []
            for member in members:
                if isinstance(member, Mapping):
                    member = dict(member)
                    model = member.get("model")
                elif isinstance(member, EnvironmentMemberConfig):
                    model = member.model
                else:
                    resolved.append(member)
                    continue
                if model is not None:
                    if not isinstance(model, ModelConfig):
                        model = ModelConfig.model_validate(model)
                    if not model._inference_provider_resolved:
                        model = model.resolve_inference_provider(providers)
                    enforce_group_base_url_policy(model.base_url, raw["group"])
                    if isinstance(member, dict):
                        member["model"] = model
                    else:
                        member = replace(member, model=model)
                resolved.append(member)
            raw["members"] = resolved
        return raw

    @model_validator(mode="after")
    def _validate_population(self) -> Self:
        seen: set[str] = set()
        for member in self.members:
            self.validate_member_config(member)
            if member.name in seen:
                raise ValueError(
                    f"Elo member name {member.name!r} is duplicated. "
                    "Each member must have a unique name."
                )
            seen.add(member.name)
        if self.restart_from_json is None:
            self.validate_population_size(len(self.members))
        return self

    @staticmethod
    def validate_population_size(population_size: int) -> None:
        if population_size < 2:
            raise ValueError(
                "AgentEloEnvironment requires at least two active members."
            )
        if population_size % 2:
            raise ValueError(
                "AgentEloEnvironment requires an even number of active members. "
                f"Received {population_size}."
            )

    @staticmethod
    def validate_member_config(member: EnvironmentMemberConfig) -> None:
        """Validate member settings for both new and restored populations."""
        if "workspace" in (member.config or {}):
            raise ValueError(
                f"Member {member.name!r} sets config.workspace. "
                "AgentEloEnvironment manages member workspaces automatically. "
                "Set the top-level workspace instead."
            )
        if "agent_name" in (member.config or {}):
            raise ValueError(
                f"Member {member.name!r} sets config.agent_name. "
                "AgentEloEnvironment generates persistent agent identities "
                "automatically. Remove config.agent_name and use the "
                "member's name field instead."
            )

    @model_serializer(mode="wrap")
    def _serialize_resolved_models(self, handler):
        # Resolved endpoints are self-contained. Retaining the provider reference
        # would make the saved model invalid when read from YAML or dashboard JSON.
        data = handler(self)
        for member in data.get("members", []):
            model = member.get("model")
            if isinstance(model, dict):
                model.pop("inference_provider", None)
        return data

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> Self:
        """Compatibility entry point for callers loading a mapping."""
        return cls.model_validate(data)

    @classmethod
    def from_source(
        cls,
        config: Self | Mapping[str, Any] | str | Path | None = None,
        **overrides: Any,
    ) -> Self:
        """Merge constructor overrides before validating the final configuration.

        None overrides retain the supplied configuration, as in the environment
        constructor. Existing model instances are preserved during merging.
        """
        if isinstance(config, (str, Path)):
            raw = load_yaml_mapping(config)
        elif isinstance(config, cls):
            raw = {name: getattr(config, name) for name in cls.model_fields}
        elif isinstance(config, Mapping):
            raw = dict(config)
        elif config is None:
            raw = {"name": "agent_elo"}
        else:
            raise TypeError(
                "Elo config must be a config object, mapping, or YAML path"
            )
        raw.update({
            key: value for key, value in overrides.items() if value is not None
        })
        return cls.model_validate(raw)

from_mapping(data) classmethod

Compatibility entry point for callers loading a mapping.

Source code in src/ursa/environments/config.py

@classmethod
def from_mapping(cls, data: Mapping[str, Any]) -> Self:
    """Compatibility entry point for callers loading a mapping."""
    return cls.model_validate(data)

from_source(config=None, **overrides) classmethod

Merge constructor overrides before validating the final configuration.

None overrides retain the supplied configuration, as in the environment constructor. Existing model instances are preserved during merging.

Source code in src/ursa/environments/config.py

@classmethod
def from_source(
    cls,
    config: Self | Mapping[str, Any] | str | Path | None = None,
    **overrides: Any,
) -> Self:
    """Merge constructor overrides before validating the final configuration.

    None overrides retain the supplied configuration, as in the environment
    constructor. Existing model instances are preserved during merging.
    """
    if isinstance(config, (str, Path)):
        raw = load_yaml_mapping(config)
    elif isinstance(config, cls):
        raw = {name: getattr(config, name) for name in cls.model_fields}
    elif isinstance(config, Mapping):
        raw = dict(config)
    elif config is None:
        raw = {"name": "agent_elo"}
    else:
        raise TypeError(
            "Elo config must be a config object, mapping, or YAML path"
        )
    raw.update({
        key: value for key, value in overrides.items() if value is not None
    })
    return cls.model_validate(raw)

validate_member_config(member) staticmethod

Validate member settings for both new and restored populations.

Source code in src/ursa/environments/config.py

@staticmethod
def validate_member_config(member: EnvironmentMemberConfig) -> None:
    """Validate member settings for both new and restored populations."""
    if "workspace" in (member.config or {}):
        raise ValueError(
            f"Member {member.name!r} sets config.workspace. "
            "AgentEloEnvironment manages member workspaces automatically. "
            "Set the top-level workspace instead."
        )
    if "agent_name" in (member.config or {}):
        raise ValueError(
            f"Member {member.name!r} sets config.agent_name. "
            "AgentEloEnvironment generates persistent agent identities "
            "automatically. Remove config.agent_name and use the "
            "member's name field instead."
        )

AgentSymposiumConfig dataclass

YAML-loadable configuration for an Agent Symposium environment.

Source code in src/ursa/environments/config.py

@dataclass(frozen=True)
class AgentSymposiumConfig:
    """YAML-loadable configuration for an Agent Symposium environment."""

    name: str
    group: str = "default"
    description: str | None = None
    inference_providers: dict[str, InferenceProviderConfig] = field(
        default_factory=dict
    )
    organizer: EnvironmentMemberConfig = field(
        default_factory=lambda: EnvironmentMemberConfig(
            name="organizer", role="Symposium organizer", agent="ChatAgent"
        )
    )
    members: list[EnvironmentMemberConfig] = field(default_factory=list)
    workspace: str | None = None
    defaults: dict[str, Any] = field(default_factory=dict)
    revision_rounds: int = 1

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> "AgentSymposiumConfig":
        raw = dict(data)
        providers = _inference_providers(raw.get("inference_providers") or {})
        raw["inference_providers"] = providers
        group = str(raw.get("group") or "default")
        if "organizer" in raw and isinstance(raw["organizer"], Mapping):
            raw["organizer"] = EnvironmentMemberConfig.from_mapping(
                raw["organizer"], providers, group
            )
        if "members" in raw:
            raw["members"] = [
                EnvironmentMemberConfig.from_mapping(member, providers, group)
                for member in raw["members"]
            ]
        return cls(**raw)

AgentTeamConfig dataclass

YAML-loadable configuration for an Agent Team environment.

Source code in src/ursa/environments/config.py

@dataclass(frozen=True)
class AgentTeamConfig:
    """YAML-loadable configuration for an Agent Team environment."""

    name: str
    group: str = "default"
    description: str | None = None
    inference_providers: dict[str, InferenceProviderConfig] = field(
        default_factory=dict
    )
    pi: EnvironmentMemberConfig = field(
        default_factory=lambda: EnvironmentMemberConfig(
            name="pi", role="Principal investigator", agent="ExecutionAgent"
        )
    )
    members: list[EnvironmentMemberConfig] = field(default_factory=list)
    workspace: str | None = None
    defaults: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_mapping(cls, data: Mapping[str, Any]) -> "AgentTeamConfig":
        raw = dict(data)
        providers = _inference_providers(raw.get("inference_providers") or {})
        raw["inference_providers"] = providers
        group = str(raw.get("group") or "default")
        if "pi" in raw and isinstance(raw["pi"], Mapping):
            raw["pi"] = EnvironmentMemberConfig.from_mapping(
                raw["pi"], providers, group
            )
        if "members" in raw:
            raw["members"] = [
                EnvironmentMemberConfig.from_mapping(member, providers, group)
                for member in raw["members"]
            ]
        return cls(**raw)

EnvironmentMemberConfig dataclass

Configuration for one agent or nested environment member.

YAML fields

name: stable member name used in prompts/tool names/persistence role: human-readable role or specialty agent: Python class path or URSA agent class name, e.g. ExecutionAgent model: optional ModelConfig-compatible mapping for this member config: kwargs passed to the agent/environment constructor prompt: optional extra role/system guidance included in delegated tasks reviewer: whether this member participates in symposium review phases

Source code in src/ursa/environments/config.py

@dataclass(frozen=True)
class EnvironmentMemberConfig:
    """Configuration for one agent or nested environment member.

    YAML fields:
      name: stable member name used in prompts/tool names/persistence
      role: human-readable role or specialty
      agent: Python class path or URSA agent class name, e.g. ExecutionAgent
      model: optional ModelConfig-compatible mapping for this member
      config: kwargs passed to the agent/environment constructor
      prompt: optional extra role/system guidance included in delegated tasks
      reviewer: whether this member participates in symposium review phases
    """

    name: str
    role: str = "Team member"
    agent: str = "ExecutionAgent"
    model: ModelConfig | None = None
    config: dict[str, Any] = field(default_factory=dict)
    prompt: str | None = None
    reviewer: bool = True

    @classmethod
    def from_mapping(
        cls,
        data: Mapping[str, Any],
        inference_providers: Mapping[str, InferenceProviderConfig]
        | None = None,
        group: str = "default",
    ) -> "EnvironmentMemberConfig":
        raw = dict(data)
        model = raw.get("model")
        if isinstance(model, Mapping):
            model_config = ModelConfig.model_validate(model)
            model_config = model_config.resolve_inference_provider(
                dict(inference_providers or {})
            )
            enforce_group_base_url_policy(model_config.base_url, group)
            raw["model"] = model_config
        return cls(**raw)

elo_cache_dir(group, name)

Return the persistent configuration directory for a named Elo environment.

Source code in src/ursa/environments/config.py

def elo_cache_dir(
    group: str,
    name: str,
) -> Path:
    """Return the persistent configuration directory for a named Elo environment."""
    return group_environments_dir(group) / "agent_elo" / name

load_object(path_or_name)

Load a class by URSA short name or full module path.

Short names are resolved first against ursa.agents and then against ursa.environments. This lets YAML use concise names such as ExecutionAgent or AgentTeamEnvironment while still allowing fully qualified custom class paths.

Source code in src/ursa/environments/config.py

def load_object(path_or_name: str) -> Any:
    """Load a class by URSA short name or full module path.

    Short names are resolved first against ``ursa.agents`` and then against
    ``ursa.environments``. This lets YAML use concise names such as
    ``ExecutionAgent`` or ``AgentTeamEnvironment`` while still allowing fully
    qualified custom class paths.
    """
    if "." not in path_or_name:
        for module_name in ("ursa.agents", "ursa.environments"):
            module = importlib.import_module(module_name)
            try:
                return getattr(module, path_or_name)
            except AttributeError:
                continue
        raise AttributeError(
            f"Could not resolve {path_or_name!r} in ursa.agents or "
            "ursa.environments. Use a full Python import path for custom classes."
        )
    module_name, attr_name = path_or_name.rsplit(".", 1)
    module = importlib.import_module(module_name)
    return getattr(module, attr_name)

load_yaml_mapping(path)

Load a YAML mapping with URSA-style environment interpolation.

Source code in src/ursa/environments/config.py

def load_yaml_mapping(path: str | Path) -> dict[str, Any]:
    """Load a YAML mapping with URSA-style environment interpolation."""
    p = Path(path).expanduser()
    with p.open("r", encoding="utf-8") as handle:
        data = yaml.safe_load(handle) or {}
    if not isinstance(data, Mapping):
        raise ValueError(f"YAML file {p} must contain a top-level mapping.")
    return deep_interp_env(dict(data))

make_llm(default_llm, model_config)

Return a member-specific model if configured, otherwise the default.

Source code in src/ursa/environments/config.py

def make_llm(
    default_llm: BaseChatModel,
    model_config: ModelConfig | Mapping[str, Any] | None,
) -> BaseChatModel:
    """Return a member-specific model if configured, otherwise the default."""
    if model_config is None:
        return default_llm
    if isinstance(model_config, Mapping):
        model_config = ModelConfig.model_validate(model_config)
    return init_chat_model(**model_config.kwargs)

save_elo_config(config, path=None)

Persist an Elo environment configuration.

Source code in src/ursa/environments/config.py

def save_elo_config(
    config: AgentEloConfig,
    path: str | Path | None = None,
) -> Path:
    """Persist an Elo environment configuration."""
    target = (
        Path(path).expanduser()
        if path
        else elo_cache_dir(
            config.group,
            config.name,
        )
        / "elo.yaml"
    )

    target.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    target.write_text(
        yaml.safe_dump(
            _dataclass_to_plain(config),
            sort_keys=False,
        ),
        encoding="utf-8",
    )

    return target

save_symposium_config(config, path=None)

Persist a symposium configuration under ~/.cache/ursa//environments by default.

Source code in src/ursa/environments/config.py

def save_symposium_config(
    config: AgentSymposiumConfig, path: str | Path | None = None
) -> Path:
    """Persist a symposium configuration under ~/.cache/ursa/<group>/environments by default."""
    target = (
        Path(path).expanduser()
        if path
        else symposium_cache_dir(config.group, config.name) / "symposium.yaml"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(
        yaml.safe_dump(_dataclass_to_plain(config), sort_keys=False),
        encoding="utf-8",
    )
    return target

save_team_config(config, path=None)

Persist a team configuration under ~/.cache/ursa//environments by default.

Source code in src/ursa/environments/config.py

def save_team_config(
    config: AgentTeamConfig, path: str | Path | None = None
) -> Path:
    """Persist a team configuration under ~/.cache/ursa/<group>/environments by default."""
    target = (
        Path(path).expanduser()
        if path
        else team_cache_dir(config.group, config.name) / "team.yaml"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(
        yaml.safe_dump(_dataclass_to_plain(config), sort_keys=False),
        encoding="utf-8",
    )
    return target

symposium_cache_dir(group, name)

Return the persistent configuration directory for a named symposium.

Source code in src/ursa/environments/config.py

def symposium_cache_dir(group: str, name: str) -> Path:
    """Return the persistent configuration directory for a named symposium."""
    return group_environments_dir(group) / "agent_symposia" / name

team_cache_dir(group, name)

Return the persistent configuration directory for a named team.

Source code in src/ursa/environments/config.py

def team_cache_dir(group: str, name: str) -> Path:
    """Return the persistent configuration directory for a named team."""
    return group_environments_dir(group) / "agent_teams" / name

visualization

EnvironmentEventRecorder

Bases: BaseCallbackHandler

Record URSA structured progress events to a replayable JSONL file.

Source code in src/ursa/environments/visualization.py

class EnvironmentEventRecorder(BaseCallbackHandler):
    """Record URSA structured progress events to a replayable JSONL file."""

    def __init__(
        self,
        *,
        run_id: str,
        group: str = "default",
        environment_name: str,
        environment_type: str,
        run_dir: Path | None = None,
        event_name: str = DEFAULT_EVENT_NAME,
        max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
    ) -> None:
        self.run_id = run_id
        self.group = validate_group_name(group)
        self.environment_name = environment_name
        self.environment_type = environment_type
        self.event_name = event_name
        self.max_payload_chars = max_payload_chars
        if run_dir is None:
            self.paths = get_environment_run_paths(self.group, run_id)
        else:
            self.paths = EnvironmentRunPaths(
                run_dir=run_dir,
                manifest_path=run_dir / "manifest.json",
                events_path=run_dir / "events.jsonl",
                artifacts_dir=run_dir / "artifacts",
                logs_dir=run_dir / "logs",
            )
        ensure_environment_run_dirs(self.paths)
        self._lock = threading.Lock()
        self._seq = 0

    @property
    def config(self) -> RunnableConfig:
        return {
            "callbacks": [self],
            "metadata": {
                "environment_run_id": self.run_id,
                "environment_name": self.environment_name,
                "environment_type": self.environment_type,
                "group": self.group,
            },
            "tags": ["environment_run", self.environment_name],
        }

    def write_manifest(
        self,
        *,
        status: str,
        task: Any | None = None,
        error: str | None = None,
    ) -> None:
        existing: dict[str, Any] = {}
        if self.paths.manifest_path.exists():
            try:
                existing = json.loads(
                    self.paths.manifest_path.read_text(encoding="utf-8")
                )
            except Exception:
                existing = {}
        now = utc_now_rfc3339()
        manifest = {
            **existing,
            "schema_version": ENVIRONMENT_RUN_SCHEMA_VERSION,
            "run_id": self.run_id,
            "group": self.group,
            "environment_name": self.environment_name,
            "environment_type": self.environment_type,
            "status": status,
            "updated_at": now,
            "events_path": "events.jsonl",
            "artifacts_path": "artifacts",
            "logs_path": "logs",
        }
        manifest.setdefault("created_at", now)
        if task is not None:
            manifest["task_preview"] = _preview(task)
        if error:
            manifest["error"] = error
        self.paths.manifest_path.write_text(
            json.dumps(manifest, indent=2, ensure_ascii=False, default=str),
            encoding="utf-8",
        )

    def on_custom_event(
        self,
        name: str,
        data: Any,
        *,
        run_id,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> None:
        if name != self.event_name or not isinstance(data, Mapping):
            return
        event = self.normalize_event(data, tags=tags, metadata=metadata)
        self.append_event(event)

    def normalize_event(
        self,
        data: Mapping[str, Any],
        *,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        raw_payload_tags = data.get("tags")
        safe_data = _make_json_safe(data, max_chars=self.max_payload_chars)
        if not isinstance(safe_data, dict):
            safe_data = {"value": safe_data}
        event_type = _infer_event_type(safe_data)
        source = _source_from_payload(safe_data)
        target = safe_data.get("target")
        if isinstance(target, str):
            target = {"id": target, "name": target}
        elif isinstance(target, Mapping):
            target = dict(target)
        else:
            target = None
        if (
            target is None
            and safe_data.get("tool")
            and source.get("kind") != "tool"
        ):
            target = _tool_target_from_payload(safe_data)
        environment_name = str(
            safe_data.get("environment") or self.environment_name
        )
        return {
            "schema_version": ENVIRONMENT_EVENT_SCHEMA_VERSION,
            "event_id": new_event_id(),
            "seq": 0,
            "ts": utc_now_rfc3339(),
            "monotonic_timestamp_ns": safe_data.get(
                "monotonic_timestamp_ns", monotonic_ns()
            ),
            "run_id": self.run_id,
            "environment_id": safe_data.get("environment_id")
            or environment_name,
            "environment_name": environment_name,
            "environment_type": safe_data.get("environment_type")
            or self.environment_type,
            "event_type": event_type,
            "stage": safe_data.get("stage"),
            "phase": safe_data.get("phase"),
            "level": safe_data.get("level", "info"),
            "source": source,
            "target": target,
            "message": safe_data.get("message") or event_type,
            "payload": safe_data,
            "tags": _normalize_tags(tags) + _normalize_tags(raw_payload_tags),
            "metadata": _make_json_safe(
                metadata or {}, max_chars=self.max_payload_chars
            ),
        }

    def append_event(self, event: Mapping[str, Any]) -> dict[str, Any]:
        with self._lock:
            self._seq += 1
            record = dict(event)
            record["seq"] = self._seq
            with self.paths.events_path.open("a", encoding="utf-8") as f:
                f.write(
                    json.dumps(record, ensure_ascii=False, default=str) + "\n"
                )
                f.flush()
            return record

arun_with_visualization(environment, inputs, *, config=None, run_id=None, max_payload_chars=DEFAULT_MAX_PAYLOAD_CHARS, **kwargs) async

Run an environment asynchronously while recording visualization events.

Source code in src/ursa/environments/visualization.py

async def arun_with_visualization(
    environment: Any,
    inputs: InputLike,
    *,
    config: RunnableConfig | None = None,
    run_id: str | None = None,
    max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
    **kwargs: Any,
) -> Any:
    """Run an environment asynchronously while recording visualization events."""
    recorder = EnvironmentEventRecorder(
        run_id=run_id or new_run_id(),
        group=getattr(environment, "group", "default"),
        environment_name=getattr(
            environment, "name", type(environment).__name__
        ),
        environment_type=type(environment).__name__,
        max_payload_chars=max_payload_chars,
    )
    recorder.write_manifest(status="running", task=inputs)
    run_config = visualization_config(recorder, config)
    try:
        result = await environment.ainvoke(inputs, config=run_config, **kwargs)
    except BaseException as exc:
        recorder.write_manifest(status="failed", task=inputs, error=str(exc))
        raise
    recorder.write_manifest(status="succeeded", task=inputs)
    return result

environment_run_recorder(environment, *, task=None, config=None, run_id=None, max_payload_chars=DEFAULT_MAX_PAYLOAD_CHARS)

Create a recorder and runnable config for an environment run.

Source code in src/ursa/environments/visualization.py

@contextmanager
def environment_run_recorder(
    environment: Any,
    *,
    task: Any | None = None,
    config: RunnableConfig | None = None,
    run_id: str | None = None,
    max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
) -> Iterator[tuple[EnvironmentEventRecorder, RunnableConfig]]:
    """Create a recorder and runnable config for an environment run."""
    recorder = EnvironmentEventRecorder(
        run_id=run_id or new_run_id(),
        group=getattr(environment, "group", "default"),
        environment_name=getattr(
            environment, "name", type(environment).__name__
        ),
        environment_type=type(environment).__name__,
        max_payload_chars=max_payload_chars,
    )
    recorder.write_manifest(status="running", task=task)
    try:
        yield recorder, visualization_config(recorder, config)
    except BaseException as exc:
        recorder.write_manifest(status="failed", task=task, error=str(exc))
        raise
    else:
        recorder.write_manifest(status="succeeded", task=task)

environment_runs_dir(group=None)

Return the directory that stores recorded environment visualization runs.

Source code in src/ursa/environments/visualization.py

def environment_runs_dir(group: str | None = None) -> Path:
    """Return the directory that stores recorded environment visualization runs."""
    return group_root_dir(validate_group_name(group)) / "environment_runs"

record_environment_run(environment, *, task=None, config=None, run_id=None, max_payload_chars=DEFAULT_MAX_PAYLOAD_CHARS)

Alias for environment_run_recorder for readable user code.

Source code in src/ursa/environments/visualization.py

@contextmanager
def record_environment_run(
    environment: Any,
    *,
    task: Any | None = None,
    config: RunnableConfig | None = None,
    run_id: str | None = None,
    max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
) -> Iterator[tuple[EnvironmentEventRecorder, RunnableConfig]]:
    """Alias for ``environment_run_recorder`` for readable user code."""
    with environment_run_recorder(
        environment,
        task=task,
        config=config,
        run_id=run_id,
        max_payload_chars=max_payload_chars,
    ) as value:
        yield value

run_with_visualization(environment, inputs, *, config=None, run_id=None, max_payload_chars=DEFAULT_MAX_PAYLOAD_CHARS, **kwargs)

Run an environment synchronously while recording visualization events.

Source code in src/ursa/environments/visualization.py

def run_with_visualization(
    environment: Any,
    inputs: InputLike,
    *,
    config: RunnableConfig | None = None,
    run_id: str | None = None,
    max_payload_chars: int = DEFAULT_MAX_PAYLOAD_CHARS,
    **kwargs: Any,
) -> Any:
    """Run an environment synchronously while recording visualization events."""
    recorder = EnvironmentEventRecorder(
        run_id=run_id or new_run_id(),
        group=getattr(environment, "group", "default"),
        environment_name=getattr(
            environment, "name", type(environment).__name__
        ),
        environment_type=type(environment).__name__,
        max_payload_chars=max_payload_chars,
    )
    recorder.write_manifest(status="running", task=inputs)
    run_config = visualization_config(recorder, config)
    try:
        result = environment.invoke(inputs, config=run_config, **kwargs)
    except BaseException as exc:
        recorder.write_manifest(status="failed", task=inputs, error=str(exc))
        raise
    recorder.write_manifest(status="succeeded", task=inputs)
    return result

visualization_config(recorder, config=None)

Merge an optional runnable config with the recorder callback config.

Source code in src/ursa/environments/visualization.py

def visualization_config(
    recorder: EnvironmentEventRecorder,
    config: RunnableConfig | None = None,
) -> RunnableConfig:
    """Merge an optional runnable config with the recorder callback config."""
    if config is None:
        return recorder.config
    return merge_configs(recorder.config, config)