"""Resource envelope accounting: one envelope per run, shared by executor, gate cycles and reviewer sessions.""" from __future__ import annotations from dataclasses import dataclass, field @dataclass class Envelope: max_budget_usd: float wall_seconds: int max_turns_per_session: int spent_usd: float = 0.0 spent_wall_s: float = 0.0 sessions: list = field(default_factory=list) # [{"session_id","role","cost","wall_s","turns"}] min_continuation_usd: float = 0.25 def remaining_usd(self) -> float: return max(0.0, self.max_budget_usd - self.spent_usd) def remaining_wall(self) -> int: return int(max(0, self.wall_seconds - self.spent_wall_s)) def can_continue(self, min_wall_s: int = 120) -> bool: return self.remaining_usd() >= self.min_continuation_usd and self.remaining_wall() >= min_wall_s def charge(self, session_id: str, role: str, cost: float, wall_s: float, turns: int | None): cost = float(cost or 0.0) self.spent_usd += cost; self.spent_wall_s += float(wall_s or 0.0) self.sessions.append({"session_id": session_id, "role": role, "cost_usd": cost, "wall_s": wall_s, "turns": turns}) def summary(self) -> dict: return {"max_budget_usd": self.max_budget_usd, "wall_seconds": self.wall_seconds, "max_turns_per_session": self.max_turns_per_session, "spent_usd": round(self.spent_usd, 6), "spent_wall_s": round(self.spent_wall_s, 1), "executor_usd": round(sum(s["cost_usd"] for s in self.sessions if s["role"] == "executor"), 6), "reviewer_usd": round(sum(s["cost_usd"] for s in self.sessions if s["role"] == "reviewer"), 6), "n_sessions": len(self.sessions), "sessions": self.sessions}