AetherCrew: a virtual company of 11 AI employees Design and implementation

Model fallback

What this code does: Takes an employee’s model priority list and a set of unavailable models, checks them in order from the front, and picks the first model that can be used. Skipped models are recorded in fallbacks, so it is clear which choice the work fell back to, and an empty list or a fully blocked list raises an error instead of passing silently. It checks only as many entries as the list holds, so fallback can never loop forever.

class FallbackManager:
    """Selects next available agent with bounded fallback traversal."""

    def select(
        self,
        priority_list: list[str],
        unavailable_agents: set[str] | None = None,
    ) -> FallbackResult:
        if not priority_list:
            raise ValueError("priority_list must not be empty")
        blocked = unavailable_agents or set()
        fallbacks: list[str] = []
        for idx, model in enumerate(priority_list):
            if model in blocked:
                fallbacks.append(model)
                continue
            return FallbackResult(selected_agent=model, selected_index=idx, fallbacks=fallbacks)
        raise RuntimeError("All agents are unavailable; fallback exhausted")

Budget states

What this code does: Adds up every recorded cost, computes usage against the budget, and returns one of four states: OK, WARN_80, WARN_95, or HARD_STOP. A budget set to zero or below counts as 100% used and becomes HARD_STOP at once, so a budget typo can never turn into unlimited spending. The orchestrator blocks new work while the state is HARD_STOP.

    @property
    def total_cost(self) -> float:
        return round(sum(e.cost_usd for e in self.entries), 6)

    def percent_used(self) -> float:
        if self.budget_policy.total_budget <= 0:
            return 100.0
        return (self.total_cost / self.budget_policy.total_budget) * 100

    def budget_state(self) -> str:
        pct = self.percent_used()
        if pct >= self.budget_policy.hard_stop_at_pct:
            return "HARD_STOP"
        if pct >= self.budget_policy.warn_at_pct[1]:
            return "WARN_95"
        if pct >= self.budget_policy.warn_at_pct[0]:
            return "WARN_80"
        return "OK"