Skip to content

symfonic.services.budget

budget

The offline budget and truncation service (T3.2.2, REQ-S3.2).

Everything that decides how much the model reads: token estimation, optional provider-exact counting, the split of a context window across prompt, tools and memory, deterministic truncation, and what happens when content does not fit. Those five concerns were spread across four packages that each carried their own arithmetic — and disagreed about it — so a budget computed in one was not comparable with a budget computed in another.

Two properties hold across the whole package and are worth stating once.

Offline by construction. Nothing here imports a tokenizer library, reads a cache directory, or opens a socket; the default estimate is arithmetic over len(text). Exactness is available, but it arrives as an adopter-registered port (:mod:.registry) that must declare itself offline, so a cold container with no egress budgets exactly as a warm one does. A tokenizer download inside prompt assembly fails on the machine least able to debug it.

Deterministic. Allocation rounds down, admission orders items totally by (order, item_id), truncation binary-searches to a guaranteed fit and is idempotent. The same inputs produce the same bytes on every machine, which is what makes a golden corpus possible at all.

The three deliverables of this task map onto the modules:

  • :mod:.contracts — the budget policy contracts: what an adopter declares (windows, shares, modes, overflow actions) and the two one-method ports through which counting is bound.
  • :mod:.service — the offline budget service: one object that resolves the counter once, derives the plan once, and applies both consistently.
  • :mod:.estimation, :mod:.registry, :mod:.allocation, :mod:.truncation, :mod:.overflow — the pure functions the service composes.

The seam to the prompting capability is :meth:.BudgetService.as_token_estimator. The prompt compiler budgets through a bound TokenEstimator port and refuses to resolve one itself; this package produces one. Neither imports the other — capability and runtime-service may not, so the two protocols meet as identical shapes rather than as a shared base class.

BudgetAllocation dataclass 

BudgetAllocation(kind: BudgetKind, limit: int)

One kind's derived ceiling in tokens.

BudgetDiagnostic dataclass 

BudgetDiagnostic(kind: str, subject: str, detail: str)

One recorded budgeting decision.

kind is the stage that made it (counting, budget), subject the item or provider it concerns.

BudgetError

Bases: ConfigurationError

Root of the budgeting taxonomy. Never raised directly.

Parented on :class:ConfigurationError rather than on a fresh root for the same reason the prompt compiler's taxonomy is: budgeting runs inside plan compilation, and an adopter who already catches configuration failures around Agent(...) must not have to learn a second base class.

BudgetItem dataclass 

BudgetItem(item_id: str, kind: BudgetKind, text: str, order: int = 0, pinned: bool = False)

One admission candidate: identity, kind, content, position, pinned-ness.

An item declares what it is, never what it costs or where it belongs in the budget. Cost is measured here with the resolved counter so two items cannot be measured by two different rules, and position comes from order broken by item_id so the sequence is totally ordered even when two callers pick the same number.

BudgetKind

Bases: Enum

The kinds of content that share a context window.

Declared in ladder order: iteration order is the canonical order of an allocation plan, so two policies that mean the same split are the same plan regardless of the order their lines were written in.

BudgetLine dataclass 

BudgetLine(kind: BudgetKind, share: float, minimum: int = 0, overflow: OverflowAction = OverflowAction.TRUNCATE)

One kind's claim on the usable window.

share is a fraction of the usable window (the context window less the output reserve), and minimum is a floor in tokens for the case where a proportional share of a small window would round down to something useless.

BudgetOverflowError

Bases: BudgetError

Content exceeded its budget and the policy said to refuse.

Also raised, under every overflow action, when pinned content alone exceeds the budget: dropping or truncating a pinned block produces a context whose missing half is invisible at run time.

BudgetPlan dataclass 

BudgetPlan(context_window: int, output_reserve: int, usable: int, allocations: tuple[BudgetAllocation, ...] = (), unallocated: int = 0)

The split of one context window across the kinds that share it.

limit_for 

limit_for(kind: BudgetKind) -> int | None

The ceiling for kind, or None when it was never allocated.

None rather than 0: an undeclared kind is unbudgeted, and answering zero would silently truncate every block of a kind the operator simply never mentioned.

Source code in src/symfonic/services/budget/values.py 
85
86
87
88
89
90
91
92
93
94
95
def limit_for(self, kind: BudgetKind) -> int | None:
    """The ceiling for ``kind``, or ``None`` when it was never allocated.

    ``None`` rather than ``0``: an undeclared kind is unbudgeted, and
    answering zero would silently truncate every block of a kind the
    operator simply never mentioned.
    """
    for allocation in self.allocations:
        if allocation.kind is kind:
            return allocation.limit
    return None

BudgetPolicy dataclass 

BudgetPolicy(context_window: int, output_reserve: int = 0, lines: tuple[BudgetLine, ...] = (), mode: CountingMode = CountingMode.HEURISTIC, provider: str = '', chars_per_token: int = DEFAULT_CHARS_PER_TOKEN, truncation: TruncationPolicy = TruncationPolicy())

The whole budgeting intent for one invocation.

Frozen and self-validating: every impossible combination is refused at construction, so a policy object in hand is a policy that can be satisfied by arithmetic — the only remaining way to fail is content that does not fit, which is :class:~.errors.BudgetOverflowError and a different type.

usable property 

usable: int

Tokens available for input, after the output reserve is set aside.

line_for 

line_for(kind: BudgetKind) -> BudgetLine | None

The declared line for kind, or None when it has no claim.

Source code in src/symfonic/services/budget/contracts.py 
233
234
235
236
237
238
def line_for(self, kind: BudgetKind) -> BudgetLine | None:
    """The declared line for ``kind``, or ``None`` when it has no claim."""
    for line in self.lines:
        if line.kind is kind:
            return line
    return None

BudgetPolicyError

Bases: BudgetError

The declared budget cannot be satisfied by arithmetic, whatever the input.

Raised at construction or at allocation — never mid-request — because a policy whose shares sum past the window is wrong on an empty prompt too.

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)

CounterResolution dataclass 

CounterResolution(counter: TokenCounter, mode: CountingMode, provider: str = '', revision: str = '', degraded: bool = False, reason: str = '')

Which counter a policy actually got, and whether that was what it asked for.

degraded is never inferred by the caller from mode: an EXACT_PREFERRED policy that fell back and a HEURISTIC policy that got exactly what it wanted both report HEURISTIC, and only one of them is a degradation worth surfacing.

CountingMode

Bases: Enum

How exact the counting has to be.

EXACT fails when no counter is registered for the provider; that is the difference between it and EXACT_PREFERRED, and the whole reason both exist. An operator reconciling invoices needs the failure; an operator who merely prefers precision needs the fallback, plus a diagnostic saying it happened.

ExactCounterRegistry 

ExactCounterRegistry()

The adopter's provider-exact counters, by provider name.

Mutable by design and by construction order: registration happens once, at assembly time, before any policy is resolved against it. Lookup is read-only, so a resolved :class:~.values.CounterResolution cannot change under a running request.

Source code in src/symfonic/services/budget/registry.py 
55
56
def __init__(self) -> None:
    self._specs: dict[str, ExactCounterSpec] = {}

get 

get(provider: str) -> ExactCounterSpec | None

The registered spec for provider, or None.

Source code in src/symfonic/services/budget/registry.py 
80
81
82
def get(self, provider: str) -> ExactCounterSpec | None:
    """The registered spec for ``provider``, or ``None``."""
    return self._specs.get(_key(provider))

providers 

providers() -> tuple[str, ...]

Every registered provider, in registration order.

Source code in src/symfonic/services/budget/registry.py 
84
85
86
def providers(self) -> tuple[str, ...]:
    """Every registered provider, in registration order."""
    return tuple(spec.provider for spec in self._specs.values())

register 

register(spec: ExactCounterSpec, *, replace: bool = False) -> None

Accept a counter for spec.provider.

Refuses a spec that does not claim to be offline, and refuses to silently shadow an existing registration — a second registration for the same provider is either a bug or an intentional override, and only the caller knows which, so replace makes them say.

Source code in src/symfonic/services/budget/registry.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def register(self, spec: ExactCounterSpec, *, replace: bool = False) -> None:
    """Accept a counter for ``spec.provider``.

    Refuses a spec that does not claim to be offline, and refuses to
    silently shadow an existing registration — a second registration for
    the same provider is either a bug or an intentional override, and only
    the caller knows which, so ``replace`` makes them say.
    """
    if not spec.offline:
        raise ExactCounterUnavailableError(
            f"counter for {spec.provider!r} does not declare itself offline. "
            "Budgeting must not fetch a tokenizer: warm the vocabulary at "
            "build time and register a counter that reads it locally."
        )
    key = _key(spec.provider)
    if key in self._specs and not replace:
        raise ExactCounterUnavailableError(
            f"a counter for {spec.provider!r} is already registered; pass "
            "replace=True to override it deliberately."
        )
    self._specs[key] = spec

ExactCounterSpec dataclass 

ExactCounterSpec(provider: str, counter: TokenCounter, offline: bool = False, revision: str = '')

An adopter's provider-exact counter, offered for registration.

offline is an assertion the registrant makes and the registry enforces: a counter that would fetch an encoding on first use is refused at registration, on the machine that wrote the config, rather than at 3am on the container that has no egress.

ExactCounterUnavailableError

Bases: BudgetError

An exact token counter was required and none was registered — or the one offered admits it is not offline.

Refusing here is the point of the class. The alternative, silently estimating for an operator who asked for exact arithmetic, produces billing and truncation numbers that are wrong in a direction nobody can see.

FittedItem dataclass 

FittedItem(item_id: str, kind: BudgetKind, text: str, tokens: int, truncated: bool = False, pinned: bool = False)

One item after admission, with what survived of it.

HeuristicTokenCounter dataclass 

HeuristicTokenCounter(chars_per_token: int = DEFAULT_CHARS_PER_TOKEN)

Characters-per-token arithmetic — the offline default.

Named an estimate because that is what it is. A provider-exact counter is a legitimate substitution, but it arrives as a registered port (:mod:.registry) rather than as a download this module performs.

Implements both port spellings: count for this package and estimate for the prompting capability's :class:TokenEstimator, so one object can be handed to either without an adapter or an import edge.

count 

count(text: str) -> int

Tokens in text, rounded up; empty text costs nothing.

The floor of one for non-empty text matters more than it looks: without it a very large chars_per_token makes short strings free, and a budget filled with free strings has no ceiling at all.

Source code in src/symfonic/services/budget/estimation.py 
61
62
63
64
65
66
67
68
69
70
def count(self, text: str) -> int:
    """Tokens in ``text``, rounded up; empty text costs nothing.

    The floor of one for non-empty text matters more than it looks: without
    it a very large ``chars_per_token`` makes short strings free, and a
    budget filled with free strings has no ceiling at all.
    """
    if not text:
        return 0
    return max(1, -(-len(text) // self.chars_per_token))

estimate 

estimate(text: str) -> int

Alias of :meth:count under the prompting capability's port name.

Source code in src/symfonic/services/budget/estimation.py 
72
73
74
def estimate(self, text: str) -> int:
    """Alias of :meth:`count` under the prompting capability's port name."""
    return self.count(text)

Keep

Bases: Enum

Which end of an over-long string survives truncation.

OverflowAction

Bases: Enum

What to do when a kind's content exceeds its allowance.

OverflowResult dataclass 

OverflowResult(kept: tuple[FittedItem, ...] = (), dropped: tuple[str, ...] = (), limit: int = 0, total_tokens: int = 0, overflowed: bool = False, diagnostics: tuple[BudgetDiagnostic, ...] = ())

The outcome of fitting a sequence of items into one ceiling.

TokenCounter

Bases: Protocol

Counts the tokens in a string.

Implementations must be pure (same string, same number) and monotone non-decreasing in prefix length — truncation's binary search assumes the second, and its shrink loop survives an implementation that breaks it.

TokenEstimator

Bases: Protocol

The prompting capability's spelling of the same port.

symfonic.capabilities.prompting binds an estimator by structure, not by import: capability and runtime-service may not import each other, so the two protocols meet as identical shapes rather than as a shared base class.

TokenEstimatorAdapter dataclass 

TokenEstimatorAdapter(counter: TokenCounter)

Exposes any :class:TokenCounter under the prompting port's name.

The one line of glue that lets an adopter's provider-exact counter budget a compiled prompt: the prompting capability binds estimate, this package speaks count, and neither imports the other.

TruncationPolicy dataclass 

TruncationPolicy(keep: Keep = Keep.HEAD, marker: str = '...[truncated]')

How an over-long string is cut down.

The marker is mandatory. A silent elision is a defect rather than a style choice: neither the model nor the operator reading a transcript can tell truncated content from content that was always that short, and the two lead to opposite debugging conclusions.

TruncationResult dataclass 

TruncationResult(text: str, tokens: int, truncated: bool, dropped_tokens: int = 0)

A string cut to fit, with the arithmetic that explains it.

allocate 

allocate(policy: BudgetPolicy) -> BudgetPlan

Derive the per-kind ceilings policy implies.

A pure function of the policy: no clock, no environment, no counter. The plan for a given policy is the same on every machine and in every process, which is what makes it comparable across a deploy.

Source code in src/symfonic/services/budget/allocation.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def allocate(policy: BudgetPolicy) -> BudgetPlan:
    """Derive the per-kind ceilings ``policy`` implies.

    A pure function of the policy: no clock, no environment, no counter. The
    plan for a given policy is the same on every machine and in every process,
    which is what makes it comparable across a deploy.
    """
    usable = policy.usable
    minimums = sum(line.minimum for line in policy.lines)
    if minimums > usable:
        raise BudgetPolicyError(
            f"declared minimums total {minimums} tokens but only {usable} are usable "
            f"({policy.context_window} window less {policy.output_reserve} reserved "
            "for output). Lower a minimum or widen the window."
        )

    limits: dict[BudgetKind, int] = {}
    for line in policy.lines:
        limits[line.kind] = max(line.minimum, int(usable * line.share))

    allocated = sum(limits.values())
    if allocated > usable:
        forced = ", ".join(
            f"{kind.value}={limit}" for kind, limit in sorted(
                limits.items(), key=lambda item: item[0].value
            )
        )
        raise BudgetPolicyError(
            f"per-kind minimums raise the total allocation to {allocated} tokens "
            f"against {usable} usable ({forced}). Minimums are floors, not "
            "priorities: the budget cannot honour all of them at once."
        )

    allocations = tuple(
        BudgetAllocation(kind=kind, limit=limits[kind])
        for kind in BudgetKind
        if kind in limits
    )
    return BudgetPlan(
        context_window=policy.context_window,
        output_reserve=policy.output_reserve,
        usable=usable,
        allocations=allocations,
        unallocated=usable - allocated,
    )

apply_overflow 

apply_overflow(items: Sequence[BudgetItem], limit: int, action: OverflowAction, counter: TokenCounter, truncation: TruncationPolicy | None = None) -> OverflowResult

Fit items into limit tokens under action.

Returns the survivors in the order they arrived — only membership and content are decided here. Reordering content to make it fit would change what the model reads for a reason the operator never declared.

Source code in src/symfonic/services/budget/overflow.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def apply_overflow(
    items: Sequence[BudgetItem],
    limit: int,
    action: OverflowAction,
    counter: TokenCounter,
    truncation: TruncationPolicy | None = None,
) -> OverflowResult:
    """Fit ``items`` into ``limit`` tokens under ``action``.

    Returns the survivors in the order they arrived — only membership and
    content are decided here. Reordering content to make it fit would change
    what the model reads for a reason the operator never declared.
    """
    if limit < 0:
        raise BudgetPolicyError(f"limit must be >= 0, got {limit}")

    total = sum(counter.count(item.text) for item in items)
    if total <= limit:
        kept = tuple(
            _fitted(item, item.text, counter.count(item.text), truncated=False)
            for item in items
        )
        return OverflowResult(kept=kept, limit=limit, total_tokens=total)

    pinned_total = _guard_pinned(items, limit, counter)

    if action is OverflowAction.FAIL:
        raise BudgetOverflowError(
            f"content needs {total} tokens but the budget is {limit}, and the "
            "declared overflow policy is 'fail'. Nothing was truncated or dropped."
        )

    if action is OverflowAction.DROP:
        dropped, diagnostics = _drop(items, limit, counter, total)
        kept = tuple(
            _fitted(item, item.text, counter.count(item.text), truncated=False)
            for item in items
            if item.item_id not in dropped
        )
    else:
        fitted, diagnostics = _truncate_in_order(
            items, limit, counter, truncation, pinned_total
        )
        kept = tuple(fitted[item.item_id] for item in items if item.item_id in fitted)
        dropped = {item.item_id for item in items if item.item_id not in fitted}

    return OverflowResult(
        kept=kept,
        dropped=tuple(item.item_id for item in items if item.item_id in dropped),
        limit=limit,
        total_tokens=sum(item.tokens for item in kept),
        overflowed=True,
        diagnostics=tuple(diagnostics),
    )

count_parts 

count_parts(parts: Iterable[str], counter: TokenCounter, *, per_part_overhead: int = 0) -> int

Total cost of parts counted separately, plus a per-part envelope.

Counting parts separately rather than concatenating them is deliberate and is what makes this a ceiling: a tokenizer merges across a join, so the concatenated cost is never higher. A budget wants the ceiling.

per_part_overhead is the structural cost a part carries beyond its text — a tool-use block's type, id, and name field. Empty parts are skipped entirely: an absent block has no envelope either.

Source code in src/symfonic/services/budget/estimation.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def count_parts(
    parts: Iterable[str], counter: TokenCounter, *, per_part_overhead: int = 0
) -> int:
    """Total cost of ``parts`` counted separately, plus a per-part envelope.

    Counting parts separately rather than concatenating them is deliberate and
    is what makes this a *ceiling*: a tokenizer merges across a join, so the
    concatenated cost is never higher. A budget wants the ceiling.

    ``per_part_overhead`` is the structural cost a part carries beyond its text
    — a tool-use block's type, id, and name field. Empty parts are skipped
    entirely: an absent block has no envelope either.
    """
    if per_part_overhead < 0:
        raise BudgetPolicyError(
            f"per_part_overhead must be >= 0, got {per_part_overhead}"
        )
    return sum(counter.count(part) + per_part_overhead for part in parts if part)

resolve_counter 

resolve_counter(policy: BudgetPolicy, registry: ExactCounterRegistry | None) -> CounterResolution

Pick the counter policy asked for, or say why it could not be had.

Three modes, three different failure stances, and the differences are the reason the enum has three members rather than a boolean:

  • HEURISTIC never consults the registry at all — a policy that asked for the cheap deterministic counter gets it even where an exact one exists, because reproducibility was the thing it wanted.
  • EXACT raises when nothing is registered. Estimating for an operator who asked for exact arithmetic is wrong in a direction nobody can see.
  • EXACT_PREFERRED falls back and records the degradation, so the difference between "precise" and "close enough" survives into the diagnostics instead of being lost at the call site.
Source code in src/symfonic/services/budget/registry.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def resolve_counter(
    policy: BudgetPolicy, registry: ExactCounterRegistry | None
) -> CounterResolution:
    """Pick the counter ``policy`` asked for, or say why it could not be had.

    Three modes, three different failure stances, and the differences are the
    reason the enum has three members rather than a boolean:

    * ``HEURISTIC`` never consults the registry at all — a policy that asked
      for the cheap deterministic counter gets it even where an exact one
      exists, because reproducibility was the thing it wanted.
    * ``EXACT`` raises when nothing is registered. Estimating for an operator
      who asked for exact arithmetic is wrong in a direction nobody can see.
    * ``EXACT_PREFERRED`` falls back **and records the degradation**, so the
      difference between "precise" and "close enough" survives into the
      diagnostics instead of being lost at the call site.
    """
    fallback = HeuristicTokenCounter(chars_per_token=policy.chars_per_token)
    if policy.mode is CountingMode.HEURISTIC:
        return CounterResolution(counter=fallback, mode=CountingMode.HEURISTIC)

    spec = registry.get(policy.provider) if registry is not None else None
    if spec is not None:
        return CounterResolution(
            counter=spec.counter,
            mode=CountingMode.EXACT,
            provider=spec.provider,
            revision=spec.revision,
        )

    if policy.mode is CountingMode.EXACT:
        raise ExactCounterUnavailableError(
            f"no exact token counter is registered for provider {policy.provider!r}. "
            "Register one with ExactCounterRegistry.register(...), or declare "
            "mode=CountingMode.EXACT_PREFERRED to accept the offline estimate."
        )

    return CounterResolution(
        counter=fallback,
        mode=CountingMode.HEURISTIC,
        provider=policy.provider,
        degraded=True,
        reason=(
            f"no exact counter registered for {policy.provider!r}; budgeting with "
            f"the offline estimate at {policy.chars_per_token} chars/token."
        ),
    )

truncate 

truncate(text: str, max_tokens: int, counter: TokenCounter, policy: TruncationPolicy | None = None) -> TruncationResult

Cut text down until it costs at most max_tokens.

Returns the input untouched when it already fits — no marker, no copy, and truncated=False, so a caller can tell "was not cut" from "was cut to exactly the limit".

When the ceiling is too small to hold even the elision marker the result is the empty string rather than a bare marker. A marker on its own is not content: it would consume the entire allowance to communicate that the allowance was consumed.

Source code in src/symfonic/services/budget/truncation.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 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
def truncate(
    text: str,
    max_tokens: int,
    counter: TokenCounter,
    policy: TruncationPolicy | None = None,
) -> TruncationResult:
    """Cut ``text`` down until it costs at most ``max_tokens``.

    Returns the input untouched when it already fits — no marker, no copy, and
    ``truncated=False``, so a caller can tell "was not cut" from "was cut to
    exactly the limit".

    When the ceiling is too small to hold even the elision marker the result is
    the empty string rather than a bare marker. A marker on its own is not
    content: it would consume the entire allowance to communicate that the
    allowance was consumed.
    """
    if max_tokens < 0:
        raise BudgetPolicyError(f"max_tokens must be >= 0, got {max_tokens}")
    active = policy or _DEFAULT_POLICY

    original = counter.count(text)
    if original <= max_tokens:
        return TruncationResult(text=text, tokens=original, truncated=False)

    low, high, best = 0, len(text), -1
    while low <= high:
        mid = (low + high) // 2
        if counter.count(_compose(text, mid, active)) <= max_tokens:
            best, low = mid, mid + 1
        else:
            high = mid - 1

    # ``best`` is only ever assigned to a candidate that was *measured* under
    # the ceiling, which is what makes the fit guarantee independent of the
    # counter's behaviour: a badly-behaved counter can cost this search the
    # optimal answer, but it cannot buy an over-budget one.
    candidate = _compose(text, best, active) if best >= 0 else ""

    tokens = counter.count(candidate)
    return TruncationResult(
        text=candidate,
        tokens=tokens,
        truncated=True,
        dropped_tokens=original - tokens,
    )