symfonic.services.budget.overflow

overflow

Overflow policy: what happens to content that does not fit.

Three answers, and which one is right depends on what the content is — which is why it is declared per line rather than chosen here. Dropping a memory section costs recall the model can live without; truncating a tool manifest produces a half-parsed schema; refusing to build is the only honest response when the operator's own instructions are what overflowed.

Whichever action applies, two properties hold. Order is total: items are processed by (order, item_id), and dropping walks that order reversed — whatever renders last is whatever goes first — so the same input always yields the same output and a byte-comparable corpus is possible. Pinned content is never dropped and never truncated, and pinned content that alone exceeds the ceiling fails under every action, including TRUNCATE. That is the same rule the prompt compiler holds for pinned contributions, deliberately: a context missing its boundaries is worse than a context that refused to build, and two layers that disagreed about it would produce a prompt whose pinned half survived one budget and not the other.

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.

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

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),
    )