Execution

Share model-request capacity and account for inference resources across concurrent branches.

AdmissionPool

Limit active and queued requests, giving episodes and branches turns in the queue.

from focus.execution import AdmissionPool

pool = AdmissionPool(capacity=8, max_pending=64)

This pool runs up to eight operations at once and holds up to 64 waiting operations before rejecting further submissions.

Definition

class AdmissionPool

AdmissionPool.init

def __init__(self, capacity: int=16, max_pending: int=256, resource_limits: dict[str, int] | None=None) -> None

AdmissionPool.run

async def run[T](self, operation: Callable[[], Awaitable[T]], *, owner: Owner=DEFAULT_OWNER, resource: str='default', queue_timeout: float | None=None, execution_timeout: float | None=None) -> T

AdmissionPool.cancel

async def cancel(self, episode: str) -> None

AdmissionPool.aclose

async def aclose(self) -> None

CapacityError

Report that a request could not be admitted within the configured execution-capacity bounds.

from focus.execution import CapacityError

Definition

class CapacityError(RuntimeError)

ExecutionServices

Share request capacity, critic predictions, and usage accounting across agents. Agents derived for compaction use the same services, so compaction work is charged to the enclosing episode.

from focus.execution import AdmissionPool, ExecutionServices

services = ExecutionServices(pool=AdmissionPool(capacity=8))

Share up to eight simultaneous generation or critic requests within this process. Pass Execution(services=services) to participating agents. Calls go directly to each agent’s policy or critic object. An outer owner can share these services across collectors with own=False; close the services after all collections finish. Model clients remain the caller’s responsibility.

Definition

@dataclass
class ExecutionServices

Fields

pool: AdmissionPool = field(default_factory=AdmissionPool)
usage: dict[str, Usage] = field(default_factory=dict)
budgets: dict[tuple[str, str], ResourceBudget] = field(default_factory=dict)

ExecutionServices.attach_budget

def attach_budget(self, owner: Owner, budget: ResourceBudget) -> None

Register the branch’s resource allowance if that branch has no budget attached yet.

ExecutionServices.remaining

def remaining(self, owner: Owner) -> ResourceBudget

Return the branch’s remaining allowance. A branch without an attached budget has the default unrestricted allowance.

ExecutionServices.charge

def charge(self, owner: Owner, **costs: int) -> None

Deduct the supplied usage from the branch’s allowance.

ExecutionServices.astart

async def astart(self) -> None

Enter the service lifecycle. Local resources start when first used, so the default hook performs no work.

ExecutionServices.generate

async def generate(self, policy: Policy[Any], prompt: PreparedPrompt, params: SamplingParams | None, *, owner: Owner=DEFAULT_OWNER, timeout: float | None=None, queue_timeout: float | None=None) -> GenerateResult

Submit a generation request and account for its model call, prompt tokens, output tokens, and any failure under the supplied owner.

ExecutionServices.predict

async def predict(self, critic: Critic, contexts: list[str], *, owner: Owner=DEFAULT_OWNER) -> Sequence[float]

Submit a batch of critic contexts and account for the request under the supplied owner.

ExecutionServices.aenter

async def __aenter__(self) -> ExecutionServices

ExecutionServices.aexit

async def __aexit__(self, *exc: object) -> None

ExecutionServices.release_episode

def release_episode(self, episode: str) -> Usage | None

Remove an episode’s accounting after all its work has finished. Saved checkpoints retain their resource allowances.

ExecutionServices.aclose

async def aclose(self) -> None

Cancel active and queued requests in the pool. Policy and critic clients stay with their caller.

Owner

Identify the episode, branch, and invocation responsible for work in an admission pool.

from focus.execution import Owner

Definition

@dataclass(frozen=True)
class Owner

Fields

episode: str = 'default'
branch: str = 'root'
invocation: str = 'root'

Usage

Accumulate physical inference usage, including model calls and token counts.

from focus.execution import Usage

Definition

@dataclass
class Usage

Fields

calls: int = 0
prompt_tokens: int = 0
output_tokens: int = 0
failures: int = 0
tool_calls: int = 0
value_batches: int = 0

ResourceBudget

Track the model calls, prompt tokens, and output tokens a branch can still use. Each checkpoint carries its remaining allowance.

from focus.execution.budget import ResourceBudget

budget = ResourceBudget(model_calls=20, output_tokens=8192)

This branch can use 20 model calls and 8,192 generated tokens, with no prompt-token limit. Charging usage returns a new budget with the remaining allowance.

Definition

@dataclass(frozen=True)
class ResourceBudget

Fields

model_calls: int | None = None
prompt_tokens: int | None = None
output_tokens: int | None = None

ResourceBudget.charge

def charge(self, **costs: int) -> ResourceBudget