symfonic.capabilities.prompting.cache

cache

The cache-region model: what a provider caches, and what it may never hold.

A region is one contiguous run of rendered contributions that share a cache annotation. Regions are derived, never declared: a contributor states whether its content is cacheable and at which TTL tier, and the compiler decides where the breakpoints fall. That split is what keeps two capabilities from each claiming a breakpoint and pushing the prompt past the provider's ceiling.

Two invariants are enforced here rather than reviewed:

1. A volatile contribution never enters a cached region. A per-turn token inside a cached prefix invalidates that prefix on every turn, which converts a cache into a slower, more expensive uncached prompt.
2. The emitted TTL ladder is non-increasing. Anthropic's messages endpoint rejects an ascending ladder outright (ttl values must be in descending order); a 5m breakpoint followed by a 1h breakpoint is a 400, not a degraded cache. Promotion, never demotion: lowering a declared tier would silently shorten a cache an operator paid to lengthen.

CacheDirective dataclass

CacheDirective(cacheable: bool = False, ttl: CacheTtl | None = None)

One contribution's (or region's) cache annotation.

ttl without cacheable is inert by construction rather than by convention: :meth:marker reads cacheable first, so a directive that names a tier it never earned cannot leak a marker onto the wire.

rank property

rank: int

TTL tier as a comparable rank; None and 5m share tier 0.

marker

marker() -> dict[str, str] | None

The wire-level cache_control value, or None when uncached.

The marker is reconstructed from ttl rather than passed through from anything a caller built, so the wire shape stays canonical however the directive was assembled.

Source code in src/symfonic/capabilities/prompting/cache.py

def marker(self) -> dict[str, str] | None:
    """The wire-level ``cache_control`` value, or ``None`` when uncached.

    The marker is *reconstructed* from ``ttl`` rather than passed through
    from anything a caller built, so the wire shape stays canonical however
    the directive was assembled.
    """
    if not self.cacheable:
        return None
    if self.ttl is CacheTtl.ONE_HOUR:
        return {"type": "ephemeral", "ttl": "1h"}
    return {"type": "ephemeral"}

CacheRegion dataclass

CacheRegion(index: int, layer: Layer, directive: CacheDirective, text: str, contribution_ids: tuple[str, ...], digest: str)

One cache-addressable span of the compiled prompt.

annotation

annotation() -> dict[str, object]

This region as a provider content block, marker included when cached.

Source code in src/symfonic/capabilities/prompting/cache.py

def annotation(self) -> dict[str, object]:
    """This region as a provider content block, marker included when cached."""
    block: dict[str, object] = {"type": "text", "text": self.text}
    marker = self.directive.marker()
    if marker is not None:
        block["cache_control"] = marker
    return block

CacheTtl

Bases: StrEnum

The two TTL tiers a cache breakpoint can advertise.

RegionRow

Bases: NamedTuple

One rendered contribution, as the region planner sees it.

A narrow tuple rather than the full rendered value: the planner must not be able to read a tier, a trust flag, or a source, because any of those would become a second place cache decisions get made.

normalise_ttl_ladder

normalise_ttl_ladder(regions: Sequence[CacheRegion]) -> tuple[CacheRegion, ...]

Promote earlier cached regions to the highest tier appearing to their right.

Right-to-left scan tracking the high-water tier. Uncached regions are skipped — they carry no marker and so do not participate in the provider's ladder — which is why a volatile region between two cached ones does not reset the rule.

Source code in src/symfonic/capabilities/prompting/cache.py

def normalise_ttl_ladder(regions: Sequence[CacheRegion]) -> tuple[CacheRegion, ...]:
    """Promote earlier cached regions to the highest tier appearing to their right.

    Right-to-left scan tracking the high-water tier. Uncached regions are
    skipped — they carry no marker and so do not participate in the provider's
    ladder — which is why a volatile region between two cached ones does not
    reset the rule.
    """
    result = list(regions)
    highest = -1
    for position in range(len(result) - 1, -1, -1):
        region = result[position]
        if not region.directive.cacheable:
            continue
        rank = region.directive.rank
        if rank < highest:
            promoted = replace(region.directive, ttl=_TTL_AT[highest])
            result[position] = replace(region, directive=promoted)
        else:
            highest = rank
    return tuple(result)

plan_regions

plan_regions(rows: Sequence[RegionRow]) -> tuple[CacheRegion, ...]

Group rows into regions, breaking wherever the annotation changes.

Rows arrive already ordered by the compiler. Empty rows are skipped rather than emitted: a region whose only content is an empty string still costs a breakpoint, and breakpoints are the scarce resource here.

Source code in src/symfonic/capabilities/prompting/cache.py

def plan_regions(rows: Sequence[RegionRow]) -> tuple[CacheRegion, ...]:
    """Group ``rows`` into regions, breaking wherever the annotation changes.

    Rows arrive already ordered by the compiler. Empty rows are skipped rather
    than emitted: a region whose only content is an empty string still costs a
    breakpoint, and breakpoints are the scarce resource here.
    """
    regions: list[CacheRegion] = []
    bucket: list[RegionRow] = []

    def flush() -> None:
        if not bucket:
            return
        text = REGION_SEPARATOR.join(row.text for row in bucket)
        regions.append(
            CacheRegion(
                index=len(regions),
                layer=bucket[0].layer,
                directive=bucket[0].directive,
                text=text,
                contribution_ids=tuple(row.contribution_id for row in bucket),
                digest=_digest(text),
            )
        )
        bucket.clear()

    for row in rows:
        if not row.text:
            continue
        if row.directive.cacheable and is_volatile(row.layer):
            raise CacheRegionError(
                f"contribution {row.contribution_id!r} is volatile ({row.layer.value}) and "
                "cacheable; per-turn content inside a cached region invalidates the whole "
                "prefix on every turn. Move it to a stable layer or leave it uncached."
            )
        if bucket and (bucket[0].layer is not row.layer or bucket[0].directive != row.directive):
            flush()
        bucket.append(row)
    flush()

    breakpoints = sum(1 for region in regions if region.directive.cacheable)
    if breakpoints > MAX_CACHE_BREAKPOINTS:
        raise CacheRegionError(
            f"the compiled prompt declares {breakpoints} cache breakpoints; providers accept "
            f"at most {MAX_CACHE_BREAKPOINTS}. Coalesce contributions onto shared TTL tiers."
        )
    return tuple(regions)