symfonic.core.prompt.blocks.injection

injection

Placing rendered blocks inside the cached prefix, once per revision.

This module is the seam between "what the blocks say" (:mod:~symfonic.core.prompt.blocks.render) and "where those bytes go" (agent/engine.py). It exists because the placement is the entire economic case for the feature, and placement is easy to get wrong in a way that still looks correct in a diff.

Why not MEMORY_CONTEXT

The obvious hook -- "inject the standing context where the other memory-derived text goes" -- lands in MEMORY_CONTEXT, which is an L2 placeholder: it sits after the single cache_control breakpoint at the end of L1, and is therefore uncached by construction. A ~3.4 KB always-on spine placed there is re-billed at full input price on every turn of every session, which is precisely the cost this lane was built to remove. So blocks are spliced into l1_parts -- inside the L0 + L1 cached prefix -- and this module never produces an L2 fragment for anything.

Zero new breakpoints

The blocks are text inside the existing L1 section, not a section of their own. That is deliberate: the prefix breakpoint budget is shared with the rolling messages ladder (core/prompt/messages_cache.py::_count_prefix_cache_markers), and spending a marker on a region that already lives inside a cached prefix would steal a slot from the ladder and buy nothing. Splicing a string into a list that is later "\n\n---\n\n"-joined cannot add a breakpoint, so the invariant holds structurally rather than by test.

Placement within L1: before the plugin contributions, so a block edit invalidates the smallest possible suffix of L1 and the most stable content stays nearest L0.

Blocks that declare layer="L2"

The taxonomy lets a spec declare layer="L2" for genuinely volatile content (FOCUS), whose per-turn churn would thrash the cached prefix. This module resolves and renders L1 blocks only; an L2 block is skipped and reported once, by name, per injector. Rendering it into the volatile region is a separate lane with its own placement question, and quietly promoting it into L1 would do the one thing the layer field exists to prevent -- put per-turn text inside the cached prefix.

Render once per revision, not once per turn

Resolution is cheap and deterministic, but rendering is not free and its output must be byte-stable: the cache hits only if the L1 body is identical, byte for byte, to the one that created the entry. So the rendered text is memoised per scope under a key built from every block's durable revision_id -- the same identifier the sources publish -- and the render runs again only when that key changes. A turn whose blocks are unchanged returns the same string object the last turn returned, so "byte-identical" is not a property this module hopes for, it is one it cannot violate.

The key also carries the UTC date, because a learned fact renders its age ("recorded 2026-05-02, 93 days ago"). That phrase changes at midnight and nowhere else -- the renderer computes age at date granularity precisely so the cached prefix turns over at most once a day rather than on every second -- and a memo that ignored the date would serve yesterday's age string forever.

The memo is bounded and keyed on scope.scope_path: an unbounded dict keyed by scope is a per-tenant leak in a multi-tenant process, and keying on tenant_id would serve one brand's IDENTITY into another brand's prompt.

BLOCKS_PART_INDEX module-attribute

BLOCKS_PART_INDEX = 1

Where the blocks fragment is spliced into l1_parts.

Index 1 is "after the bundled L1 template body, before the plugin contributions". The value is a named constant because all three prompt paths must agree on it: the same fragment landing at a different offset on the legacy path would make the legacy/stratigraphic comparison tests pass while shipping two different prompts.

CACHED_LAYER module-attribute

CACHED_LAYER = 'L1'

The only layer this module renders into -- the cached L0 + L1 prefix.

DEFAULT_MAX_SCOPES module-attribute

DEFAULT_MAX_SCOPES = 256

Scopes retained in the render memo before the oldest is evicted.

NO_BLOCKS module-attribute

NO_BLOCKS = BlockParts()

What every path sees when the feature is off. Shared, immutable.

BlockParts dataclass

BlockParts(
    l1: str | None = None,
    revision_key: tuple[str, ...] = (),
)

One turn's rendered blocks, and the revisions that produced them.

l1 is None rather than "" when nothing renders: the prompt paths filter on truthiness, and None is the value that makes "no blocks" and "the feature is off" the same code path.

empty property

empty: bool

True when this turn contributes no bytes to any region.

PromptBlockInjector

PromptBlockInjector(
    resolver: PromptBlockResolver,
    *,
    policy: RenderPolicy = DEFAULT_POLICY,
    clock: Callable[[], datetime] | None = None,
    max_scopes: int = DEFAULT_MAX_SCOPES,
)

Resolves, renders and memoises the standing-context fragment.

One instance per agent. It owns the memo, so two agents in one process cannot serve each other's rendered blocks, and a test cannot inherit a cache entry it never created.

clock is injectable for the same reason the renderer takes now: the rendered age phrase is the only per-turn-varying input, and a test that cannot freeze it cannot assert byte identity.

Source code in src/symfonic/core/prompt/blocks/injection.py

def __init__(
    self,
    resolver: PromptBlockResolver,
    *,
    policy: RenderPolicy = DEFAULT_POLICY,
    clock: Callable[[], datetime] | None = None,
    max_scopes: int = DEFAULT_MAX_SCOPES,
) -> None:
    self._resolver = resolver
    self._policy = policy
    self._clock = clock or (lambda: datetime.now(UTC))
    self._max_scopes = max(1, int(max_scopes))
    self._memo: OrderedDict[str, tuple[tuple[str, ...], BlockParts]] = OrderedDict()
    self._skip_warned: set[str] = set()
    self.render_count = 0
    """Renders performed since construction.

    Public because "did that revision bump recompile L1 once, or
    once per turn?" is the question the whole memo exists to answer,
    and a counter is the only way to ask it that does not depend on
    string identity surviving an unrelated refactor.
    """
    # Warn as soon as a declared spec cannot ever render, rather than
    # waiting for the first `build()` call to discover it: a spec
    # is enough information to know its layer will never be `L1`,
    # and surfacing that at construction (typically at agent
    # start-up) makes a misconfigured deployment discoverable in
    # its own logs instead of only on the first live turn that
    # happens to hit this injector.
    self._report_skipped(
        spec.name for spec in resolver.specs if spec.layer != CACHED_LAYER
    )

enabled property

enabled: bool

True when at least one block is declared.

policy property

policy: RenderPolicy

The render policy applied to learned content.

Exposed alongside :attr:resolver so a snapshot render and a per-turn render cannot drift into two different fact caps.

render_count instance-attribute

render_count = 0

Renders performed since construction.

Public because "did that revision bump recompile L1 once, or once per turn?" is the question the whole memo exists to answer, and a counter is the only way to ask it that does not depend on string identity surviving an unrelated refactor.

resolver property

resolver: PromptBlockResolver

The resolver this injector reads through.

Public so the delegated-child lane (:mod:~symfonic.core.prompt.blocks.snapshot) can take its one-per-run capture through the same resolver the parent uses, rather than constructing a second one that would hold its own last_known_good cache and could serve a different revision.

build async

build(
    scope: TenantScope,
    *,
    specs: Sequence[PromptBlockSpec] | None = None,
) -> BlockParts

Resolve and render this scope's blocks, reusing unchanged bytes.

Raises whatever the resolver raises: a fail_closed block that cannot be read must stop the turn, and swallowing that here would produce an agent running without its own boundaries -- which is the failure the policy exists to prevent.

specs narrows the pass to a subset of the declared specs and is forwarded verbatim to :meth:~symfonic.core.prompt.blocks.resolver.PromptBlockResolver.resolve, which rejects anything that is not already declared here. Its one caller is the scope-less lane (:mod:~symfonic.core.prompt.blocks.scopeless). The memo needs no extra key component for it: that lane resolves under its own reserved scope_path, so a narrowed render and a full render can never land in the same memo slot.

Source code in src/symfonic/core/prompt/blocks/injection.py

async def build(
    self,
    scope: TenantScope,
    *,
    specs: Sequence[PromptBlockSpec] | None = None,
) -> BlockParts:
    """Resolve and render this scope's blocks, reusing unchanged bytes.

    Raises whatever the resolver raises: a ``fail_closed`` block that
    cannot be read must stop the turn, and swallowing that here
    would produce an agent running without its own boundaries --
    which is the failure the policy exists to prevent.

    ``specs`` narrows the pass to a subset of the declared specs and
    is forwarded verbatim to
    :meth:`~symfonic.core.prompt.blocks.resolver.PromptBlockResolver.resolve`,
    which rejects anything that is not already declared here. Its one
    caller is the scope-less lane
    (:mod:`~symfonic.core.prompt.blocks.scopeless`). The memo needs no
    extra key component for it: that lane resolves under its own
    reserved ``scope_path``, so a narrowed render and a full render
    can never land in the same memo slot.
    """
    if not self.enabled:
        return NO_BLOCKS
    if specs is not None and not specs:
        return NO_BLOCKS

    # Passed only when it is actually a narrowing. ``resolve`` is
    # reachable through a duck-typed resolver (the snapshot lane
    # takes one, and tests substitute spies), so an unconditional
    # ``specs=None`` keyword would make every existing
    # ``resolve(self, scope)`` implementation a TypeError on the
    # ordinary scoped path -- a new parameter for one caller
    # breaking the callers that never asked for it.
    blocks = (
        await self._resolver.resolve(scope)
        if specs is None
        else await self._resolver.resolve(scope, specs=specs)
    )
    now = self._clock()

    # Split BEFORE keying: an L2 block (e.g. the canonical FOCUS,
    # deliberately volatile) contributes zero bytes to `l1`, so its
    # revision must not vote on the identity of bytes it never
    # produces. Keying on the full resolve result here would make
    # L2 churn recompile L1 every turn even though the rendered
    # bytes never change -- exactly the acceptance criterion this
    # memo exists to satisfy, inverted.
    renderable, skipped = cached_layer_blocks(blocks)
    key = revision_key(renderable, now)

    cached = self._memo.get(scope.scope_path)
    if cached is not None and cached[0] == key:
        self._memo.move_to_end(scope.scope_path)
        return cached[1]

    self._report_skipped(skipped)
    parts = BlockParts(
        l1=render_blocks(renderable, policy=self._policy, now=now) or None,
        revision_key=key,
    )
    self.render_count += 1
    self._remember(scope.scope_path, key, parts)
    return parts

now

now() -> datetime

Read the injectable clock.

The snapshot renders with a frozen stamp, and freezing it against this clock is what lets a test drive both lanes from one fake clock instead of comparing a frozen render to a live one.

Source code in src/symfonic/core/prompt/blocks/injection.py

def now(self) -> datetime:
    """Read the injectable clock.

    The snapshot renders with a *frozen* stamp, and freezing it
    against this clock is what lets a test drive both lanes from one
    fake clock instead of comparing a frozen render to a live one.
    """
    return self._clock()

cached_layer_blocks

cached_layer_blocks(
    blocks: Iterable[ResolvedBlock],
) -> tuple[tuple[ResolvedBlock, ...], tuple[str, ...]]

Split blocks into the L1-layer ones and the names of the rest.

Returns (renderable, skipped_names). skipped_names is reported by the caller rather than logged here so the message can be emitted once per block per process instead of once per turn.

Source code in src/symfonic/core/prompt/blocks/injection.py

def cached_layer_blocks(
    blocks: Iterable[ResolvedBlock],
) -> tuple[tuple[ResolvedBlock, ...], tuple[str, ...]]:
    """Split ``blocks`` into the L1-layer ones and the names of the rest.

    Returns ``(renderable, skipped_names)``. ``skipped_names`` is
    reported by the caller rather than logged here so the message can be
    emitted once per block per process instead of once per turn.
    """
    renderable: list[ResolvedBlock] = []
    skipped: list[str] = []
    for block in blocks:
        if getattr(block.spec, "layer", CACHED_LAYER) == CACHED_LAYER:
            renderable.append(block)
        else:
            skipped.append(block.name)
    return tuple(renderable), tuple(skipped)

revision_key

revision_key(
    blocks: Iterable[ResolvedBlock], now: datetime
) -> tuple[str, ...]

The identity of the bytes blocks would render to.

Every component is durable -- a content hash, a mem:<count>:<max updated_at> stamp, a UTC date -- so a write made by another process (the onboarding router, the consolidation worker) changes the key on this process's next resolve. An in-process counter would not, and this agent would go on serving a block the user already corrected elsewhere.

now is normalised through :func:_as_utc -- not now.astimezone(UTC) -- so a naive clock rolls the date component over at the same instant the renderer rolls the age phrase over; see :func:_as_utc.

Source code in src/symfonic/core/prompt/blocks/injection.py

def revision_key(
    blocks: Iterable[ResolvedBlock], now: datetime
) -> tuple[str, ...]:
    """The identity of the bytes ``blocks`` would render to.

    Every component is durable -- a content hash, a
    ``mem:<count>:<max updated_at>`` stamp, a UTC date -- so a write
    made by another process (the onboarding router, the consolidation
    worker) changes the key on this process's next resolve. An
    in-process counter would not, and this agent would go on serving a
    block the user already corrected elsewhere.

    ``now`` is normalised through :func:`_as_utc` -- not
    ``now.astimezone(UTC)`` -- so a naive clock rolls the date component
    over at the same instant the renderer rolls the age phrase over; see
    :func:`_as_utc`.
    """
    return (
        f"date:{_as_utc(now).date().isoformat()}",
        *(f"{block.name}@{block.revision_id}" for block in blocks),
    )

splice_blocks_part

splice_blocks_part(
    parts: Sequence[str | None],
    blocks_part: str | None,
    *,
    index: int = BLOCKS_PART_INDEX,
) -> list[str]

Return the non-empty parts with blocks_part spliced in.

The one function all three prompt paths call, so "where do blocks go" has a single answer rather than three that drift.

blocks_part=None -- the default state of a deployment that declares no blocks -- returns exactly [p for p in parts if p], which is the pre-existing comprehension at every call site. That is what makes the byte-identity guarantee structural: with the feature off there is no added element, no added separator, and no added branch that could reorder anything.

Source code in src/symfonic/core/prompt/blocks/injection.py

def splice_blocks_part(
    parts: Sequence[str | None],
    blocks_part: str | None,
    *,
    index: int = BLOCKS_PART_INDEX,
) -> list[str]:
    """Return the non-empty ``parts`` with ``blocks_part`` spliced in.

    The one function all three prompt paths call, so "where do blocks
    go" has a single answer rather than three that drift.

    ``blocks_part=None`` -- the default state of a deployment that
    declares no blocks -- returns exactly ``[p for p in parts if p]``,
    which is the pre-existing comprehension at every call site. That is
    what makes the byte-identity guarantee structural: with the feature
    off there is no added element, no added separator, and no added
    branch that could reorder anything.
    """
    spliced = [*parts[:index], blocks_part, *parts[index:]]
    return [part for part in spliced if part]