symfonic.capabilities.caching.composition

composition

Provider-wire cache composition for the native Agent path.

Anthropic is presently the only supported wire dialect. A history marker is placed on a completed message in the outbound transcript and a manifest marker is placed on the last converted Anthropic tool definition. Both operations are pure projections: no conversation, scope, tool set, or provider response is retained between turns.

That lack of retained state is deliberate. The caller supplies history on every Agent.run/Agent.stream call, so a policy instance shared by two tenant-scoped agents cannot replay one tenant's transcript into another's request. The provider still owns any remote prompt-cache implementation; this module never presents a local cache as tenant storage.

HistoryCacheComposer

HistoryCacheComposer(policy: HistoryCachePolicy)

Annotate the latest completed text turn without retaining transcript state.

Source code in src/symfonic/capabilities/caching/composition.py

def __init__(self, policy: HistoryCachePolicy) -> None:
    self._policy = policy

compose

compose(messages: Sequence[Any]) -> list[Any]

Return a fresh wire-message list with one safe history marker.

Only closed user messages and assistant messages without pending tools can carry the marker. Tool-result exchanges are deliberately skipped: the facade's typed transcript does not retain the provider's original tool-use blocks, so claiming a cache point inside that reconstructed exchange would be speculative. A preceding ordinary user turn is still a valid conservative boundary.

Source code in src/symfonic/capabilities/caching/composition.py

def compose(self, messages: Sequence[Any]) -> list[Any]:
    """Return a fresh wire-message list with one safe history marker.

    Only closed user messages and assistant messages without pending tools
    can carry the marker.  Tool-result exchanges are deliberately skipped:
    the facade's typed transcript does not retain the provider's original
    tool-use blocks, so claiming a cache point inside that reconstructed
    exchange would be speculative.  A preceding ordinary user turn is
    still a valid conservative boundary.
    """
    marker = self._policy.marker()
    result = list(messages)
    if marker is None:
        return result
    boundary = self._boundary(result)
    if boundary is None:
        return result
    annotated = _with_cache_marker(result[boundary], marker)
    if annotated is not None:
        result[boundary] = annotated
    return result

HistoryCachePolicy dataclass

HistoryCachePolicy(enabled: bool = True, ttl: CacheTtl = CacheTtl.FIVE_MINUTES)

The marker policy for completed native conversation history.

enabled=False is a hard off switch: no message is reshaped and no cache-control field is emitted. The default five-minute marker is the canonical Anthropic wire spelling; the optional one-hour tier is explicit.

ManifestCachePolicy dataclass

ManifestCachePolicy(enabled: bool = True, position: ManifestCachePosition = ManifestCachePosition.CACHED, ttl: CacheTtl = CacheTtl.FIVE_MINUTES)

The marker policy for the provider's bound native tool schemas.

ManifestCachePosition

Bases: StrEnum

Whether the native bound-tool manifest receives an Anthropic marker.

CACHED stamps the final Anthropic tool definition, so the preceding definitions participate in that request's cache prefix. VOLATILE binds exactly the same native tools without a marker. It intentionally does not claim the legacy JIT text-manifest split: native tools are schemas on the provider request, not text inserted into a legacy prompt template.

NativeCacheComposition dataclass

NativeCacheComposition(history: HistoryCachePolicy = HistoryCachePolicy(), manifest: ManifestCachePolicy = ManifestCachePolicy())

Opt-in native history and tool-manifest request-cache composition.

Pass one instance in Agent(capabilities=[...]). The composition is valid only for adapters that actually send the Anthropic Messages API cache_control dialect. Unsupported providers are refused while a marker is requested; this avoids accepting a configuration that silently does nothing on a different wire protocol.

contribute

contribute(request: CapabilityRequest) -> CapabilityContribution

Join the capability fold without adding a kernel stage.

The policy is executed at the provider adapter seams, after the kernel has assembled the precise message and tool values. A stage would be too early for both serializer-specific operations.

Source code in src/symfonic/capabilities/caching/composition.py

def contribute(self, request: CapabilityRequest) -> CapabilityContribution:  # noqa: ARG002
    """Join the capability fold without adding a kernel stage.

    The policy is executed at the provider adapter seams, after the kernel
    has assembled the precise message and tool values.  A stage would be
    too early for both serializer-specific operations.
    """
    return CapabilityContribution(capability="native-cache-composition")

require_supported_provider

require_supported_provider(provider: Any, family: str) -> None

Refuse cache annotation where the transport cannot honour it.

Source code in src/symfonic/capabilities/caching/composition.py

def require_supported_provider(self, provider: Any, family: str) -> None:
    """Refuse cache annotation where the transport cannot honour it."""
    if not self.requests_marker:
        return
    # Bedrock can serve Claude but uses Converse ``cachePoint`` rather than
    # the Anthropic Messages ``cache_control`` field.  Its family alone is
    # therefore insufficient evidence for this request dialect.
    unsupported_bedrock = any(
        klass.__name__ == "AWSBedrockProvider" for klass in type(provider).__mro__
    )
    if family != "anthropic" or unsupported_bedrock:
        raise ConfigurationError(
            "NativeCacheComposition supports only Anthropic Messages API "
            "adapters that accept cache_control; this provider's transport "
            "does not. Disable both cache policies or compose a provider "
            "specific cache adapter. AWS Bedrock Converse uses cachePoint "
            "and is explicitly unsupported by this composition."
        )

ToolManifestBinder

ToolManifestBinder(policy: ManifestCachePolicy)

Bind native tools and, when requested, mark the Anthropic tool manifest.

Source code in src/symfonic/capabilities/caching/composition.py

def __init__(self, policy: ManifestCachePolicy) -> None:
    self._policy = policy

bind

bind(model: Any, tools: Sequence[Any], *, tool_choice: str | None = None) -> Any

Bind the exact tool sequence handed to this plan or palette.

Conversion occurs before bind_tools because LangChain Anthropic preserves cache_control only on an already-Anthropic schema. A changed palette is converted anew, so it cannot reuse a prior manifest's mutable dictionary or inherit its marker by accident.

Source code in src/symfonic/capabilities/caching/composition.py

def bind(
    self, model: Any, tools: Sequence[Any], *, tool_choice: str | None = None
) -> Any:
    """Bind the exact tool sequence handed to this plan or palette.

    Conversion occurs before ``bind_tools`` because LangChain Anthropic
    preserves ``cache_control`` only on an already-Anthropic schema.  A
    changed palette is converted anew, so it cannot reuse a prior
    manifest's mutable dictionary or inherit its marker by accident.
    """
    from symfonic.capabilities.tools.execution import SchemaBinder

    marker = self._policy.marker()
    bound_tools: list[Any] = list(tools)
    if marker is not None and bound_tools:
        try:
            from langchain_anthropic.chat_models import convert_to_anthropic_tool
        except ImportError as missing:
            raise ConfigurationError(
                "native manifest caching needs the Anthropic extra: "
                "pip install 'symfonic-core[anthropic]'"
            ) from missing

        rendered = [dict(convert_to_anthropic_tool(tool)) for tool in bound_tools]
        rendered[-1]["cache_control"] = marker
        bound_tools = rendered
    return SchemaBinder().bind(model, bound_tools, tool_choice=tool_choice)