primus.runtime_budget module

exception primus.runtime_budget.RuntimeBudgetLimitExceeded

Bases: RuntimeError

Raised when worker execution exceeds its assigned runtime budget.

class primus.runtime_budget.RuntimeBudgetMeter(spec: RuntimeBudgetSpec, *, clock: Any | None = None)

Bases: object

Worker-side runtime budget meter for propagated child budgets.

__init__(spec: RuntimeBudgetSpec, *, clock: Any | None = None) → None

check_wallclock(operation: str) → None

elapsed_seconds() → float

enforce_wallclock(operation: str) → Iterator[None]

classmethod from_env(env_var: str = 'PRIMUS_CHILD_BUDGET') → RuntimeBudgetMeter | None

maybe_enforce_wallclock(operation: str)

record_tool_call(operation: str, count: int = 1) → None

record_usd(operation: str, usd: Any) → None

class primus.runtime_budget.RuntimeBudgetSpec(usd: 'float', wallclock_seconds: 'float', depth: 'int', tool_calls: 'int')

Bases: object

__init__(usd: float, wallclock_seconds: float, depth: int, tool_calls: int) → None

depth: int

classmethod from_dict(value: dict[str, Any]) → RuntimeBudgetSpec

to_dict() → dict[str, Any]

tool_calls: int

usd: float

wallclock_seconds: float

primus.runtime_budget.runtime_budget_spec_from_env(env_var: str = 'PRIMUS_CHILD_BUDGET') → RuntimeBudgetSpec | None