symfonic.core.prompt.blocks.injection_parts

injection_parts

The block-injection vocabulary: placement, layer split, revision identity.

Split out of :mod:symfonic.core.prompt.blocks.injection (398 lines against the 300-line budget). injection is one class -- PromptBlockInjector, which resolves, renders and memoises. Everything here is what the three prompt paths agree on around it: where the fragment is spliced, which layer renders, what a turn's output looks like, and what identity its bytes have.

Keeping these separate matters beyond the line count: splice_blocks_part and BLOCKS_PART_INDEX are called by paths that never construct an injector, and the byte-identity tests drive them directly.

injection re-exports every name below, so existing imports are unchanged.

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.

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_parts.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_parts.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_parts.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]