AetherCrew: a virtual company of 11 AI employees

Model fallback

What this code does: Picks the first available model from an employee priority list.

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: Turns the running cost into a budget state. At 100% or more, work stops.

    @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"