Skip to content

symfonic.services.budget.service

service

The offline budget service — one object that owns every budgeting decision.

The point of a facade here is not convenience, it is consistency. Counting, allocation, truncation, and overflow are individually pure functions, and a caller who assembled them by hand could easily measure a prompt with one counter and truncate it with another, or truncate against a ceiling the allocation plan never granted. The service resolves the counter once, derives the plan once, and hands both to every operation, so those mistakes are not expressible.

Resolution happens at construction, not at first use. An EXACT policy with nothing registered is a deployment error, and a deployment error should land where the plan is compiled — not halfway through the first request of the day.

BudgetService dataclass 

BudgetService(policy: BudgetPolicy, registry: ExactCounterRegistry | None = None)

Counting, budgets, truncation, and overflow for one invocation.

as_token_estimator 

as_token_estimator() -> TokenEstimatorAdapter

This service's counter, under the prompting capability's port name.

The seam the prompt compiler declared and left open: it budgets through a bound TokenEstimator and refuses to resolve one itself, so exact counting reaches prompt assembly without the capability layer importing a runtime service — an edge its row of the dependency matrix forbids.

Source code in src/symfonic/services/budget/service.py 
113
114
115
116
117
118
119
120
121
def as_token_estimator(self) -> TokenEstimatorAdapter:
    """This service's counter, under the prompting capability's port name.

    The seam the prompt compiler declared and left open: it budgets through
    a bound ``TokenEstimator`` and refuses to resolve one itself, so exact
    counting reaches prompt assembly without the capability layer importing
    a runtime service — an edge its row of the dependency matrix forbids.
    """
    return TokenEstimatorAdapter(self.counting.counter)

count 

count(text: str) -> int

Tokens in text under the counter this service resolved.

Source code in src/symfonic/services/budget/service.py 
105
106
107
def count(self, text: str) -> int:
    """Tokens in ``text`` under the counter this service resolved."""
    return self.counting.counter.count(text)

fit 

fit(items: Sequence[BudgetItem], kind: BudgetKind, *, action: OverflowAction | None = None, truncation: TruncationPolicy | None = None) -> OverflowResult

Fit items into kind's allowance under kind's overflow action.

Items of another kind are refused rather than budgeted: charging tool manifests against the memory line produces a plan whose arithmetic is right and whose meaning is wrong, and nothing downstream would notice.

Source code in src/symfonic/services/budget/service.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def fit(
    self,
    items: Sequence[BudgetItem],
    kind: BudgetKind,
    *,
    action: OverflowAction | None = None,
    truncation: TruncationPolicy | None = None,
) -> OverflowResult:
    """Fit ``items`` into ``kind``'s allowance under ``kind``'s overflow action.

    Items of another kind are refused rather than budgeted: charging tool
    manifests against the memory line produces a plan whose arithmetic is
    right and whose meaning is wrong, and nothing downstream would notice.
    """
    ceiling = self._ceiling(None, kind, "fit")
    foreign = sorted({item.kind.value for item in items if item.kind is not kind})
    if foreign:
        raise BudgetPolicyError(
            f"fit({kind.value}) received items of kind {', '.join(foreign)}: an "
            "item must be charged against the line it belongs to."
        )
    line = self.policy.line_for(kind)
    resolved = action or (line.overflow if line else OverflowAction.TRUNCATE)
    return apply_overflow(
        items,
        ceiling,
        resolved,
        self.counting.counter,
        truncation or self.policy.truncation,
    )

limit_for 

limit_for(kind: BudgetKind) -> int | None

The derived ceiling for kind, or None when it is unbudgeted.

Source code in src/symfonic/services/budget/service.py 
109
110
111
def limit_for(self, kind: BudgetKind) -> int | None:
    """The derived ceiling for ``kind``, or ``None`` when it is unbudgeted."""
    return self.plan.limit_for(kind)

offline classmethod 

offline(*, context_window: int, output_reserve: int = 0, prompt: float | None = None, tools: float | None = None, memory: float | None = None, chars_per_token: int | None = None) -> BudgetService

The zero-configuration constructor: a window, optional shares, no network.

Deliberately does not accept a registry or a counting mode. An adopter who wants exact counting is making a decision with operational consequences, and that decision belongs in an explicit :class:~.contracts.BudgetPolicy rather than in a convenience helper.

Source code in src/symfonic/services/budget/service.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@classmethod
def offline(
    cls,
    *,
    context_window: int,
    output_reserve: int = 0,
    prompt: float | None = None,
    tools: float | None = None,
    memory: float | None = None,
    chars_per_token: int | None = None,
) -> BudgetService:
    """The zero-configuration constructor: a window, optional shares, no network.

    Deliberately does not accept a registry or a counting mode. An adopter
    who wants exact counting is making a decision with operational
    consequences, and that decision belongs in an explicit
    :class:`~.contracts.BudgetPolicy` rather than in a convenience helper.
    """
    shares = ((BudgetKind.PROMPT, prompt), (BudgetKind.TOOLS, tools),
              (BudgetKind.MEMORY, memory))
    lines = tuple(
        BudgetLine(kind, share) for kind, share in shares if share is not None
    )
    policy = BudgetPolicy(
        context_window=context_window,
        output_reserve=output_reserve,
        lines=lines,
        chars_per_token=chars_per_token or DEFAULT_CHARS_PER_TOKEN,
    )
    return cls(policy=policy)

truncate 

truncate(text: str, *, max_tokens: int | None = None, kind: BudgetKind | None = None, truncation: TruncationPolicy | None = None) -> TruncationResult

Cut text to an explicit ceiling, or to kind's derived one.

Source code in src/symfonic/services/budget/service.py 
123
124
125
126
127
128
129
130
131
132
133
def truncate(
    self,
    text: str,
    *,
    max_tokens: int | None = None,
    kind: BudgetKind | None = None,
    truncation: TruncationPolicy | None = None,
) -> TruncationResult:
    """Cut ``text`` to an explicit ceiling, or to ``kind``'s derived one."""
    ceiling = self._ceiling(max_tokens, kind, "truncate")
    return self._truncate_to(text, ceiling, truncation)