Skip to content

symfonic.services.budget.ports

ports

The declared port of the budget service (LAY-ADR §2, port).

Same cell, same rule as :mod:symfonic.services.models.ports and :mod:symfonic.services.shadow.ports: facade-compiler → runtime-service is port, so the composition root in :mod:symfonic.agent.cutover may compose this service only through a declared narrow interface, never through :mod:~symfonic.services.budget.service or the pure functions behind it.

The three names below are exactly what a composition root needs to bind the seam this package's docstring declared and left open — "the seam to the prompting capability is :meth:~symfonic.services.budget.service.BudgetService.as_token_estimator ": one counter to price every prompt region, and the derived per-kind ceiling when the deployment declared a window and a share to split it by.

:class:~symfonic.services.budget.estimation.HeuristicTokenCounter is here for the deployment that declared no window. It answers estimate as well as count, so the composition root can bind the package's one arithmetic rule without inventing a context_window the adopter never asked for — the alternative being the fourth private estimator this package exists to retire.

Nothing here is defined locally; every name is re-exported from its owning module.

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.

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)

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)