Skip to content

symfonic.capabilities.memory.rendered_selection

rendered_selection

Exact selected rows that survived memory's own block assembly.

This is hydration evidence only. Prompt compilation and the model wire may still drop or rewrite the block, so these values authorize no acknowledgement.

RenderedSelectedLine dataclass

RenderedSelectedLine(publication: PublishedRecallRef, line: str, ordinal: int, linked_edge: LinkedRenderedWitness | None = None)

One uniquely attributable line in the composed hydration block.

admitted_mask

admitted_mask(lines: list[str], *, char_ceiling: int, max_bytes: int | None = None) -> tuple[bool, ...]

Which whole lines fit, in order: the single admission rule for the block.

Admission is greedy over a prefix, so the mask of the first n lines does not depend on anything after them.

Source code in src/symfonic/capabilities/memory/rendered_selection.py
def admitted_mask(
    lines: list[str], *, char_ceiling: int, max_bytes: int | None = None,
) -> tuple[bool, ...]:
    """Which whole lines fit, in order: the single admission rule for the block.

    Admission is greedy over a prefix, so the mask of the first ``n`` lines does
    not depend on anything after them.
    """
    ceiling = char_ceiling if max_bytes is None else max_bytes
    mask: list[bool] = []
    used, started = 0, False
    for line in lines:
        cost = len(line if max_bytes is None else line.encode("utf-8"))
        cost += 1 if started else 0
        fits = used + cost <= ceiling
        if fits:
            used, started = used + cost, True
        mask.append(fits)
    return tuple(mask)

bind_linked_lines

bind_linked_lines(selected, linked)

Keep an edge witness only for its unique admitted whole target line.

Source code in src/symfonic/capabilities/memory/rendered_selection.py
def bind_linked_lines(selected, linked):
    """Keep an edge witness only for its unique admitted whole target line."""
    keys = [(item.memory.selected_publication, item.line()) for item in linked]
    counts = Counter(keys)
    witnesses = {
        (item.memory.selected_publication, item.line()): LinkedRenderedWitness.from_edge(
            item.linked_edge)
        for item in linked if item.linked_edge is not None
        and counts[(item.memory.selected_publication, item.line())] == 1
    }
    return tuple(
        replace(item, linked_edge=witnesses.get((item.publication, item.line)))
        for item in selected
    )

bound_block

bound_block(candidates: list[tuple[str, str, PublishedRecallRef | None]], *, char_ceiling: int, max_bytes: int | None = None) -> tuple[str, tuple[tuple[str, str], ...], tuple[RenderedSelectedLine, ...]]

Keep whole lines and their verified identities in exactly the same pass.

Source code in src/symfonic/capabilities/memory/rendered_selection.py
def bound_block(
    candidates: list[tuple[str, str, PublishedRecallRef | None]],
    *,
    char_ceiling: int,
    max_bytes: int | None = None,
) -> tuple[str, tuple[tuple[str, str], ...], tuple[RenderedSelectedLine, ...]]:
    """Keep whole lines and their verified identities in exactly the same pass."""
    ceiling = char_ceiling if max_bytes is None else max_bytes
    unit = "char" if max_bytes is None else "byte"
    occurrences = Counter(line for _, line, _ in candidates)
    single_line = all(line.splitlines() == [line] for _, line, _ in candidates)
    mask = admitted_mask([line for _, line, _ in candidates],
                         char_ceiling=char_ceiling, max_bytes=max_bytes)
    admitted: list[str] = []
    dropped: list[tuple[str, str]] = []
    selected: list[RenderedSelectedLine] = []
    for (identifier, line, publication), fits in zip(candidates, mask, strict=True):
        if not fits:
            dropped.append((
                identifier,
                f"would take the block past the {ceiling}-{unit} ceiling; lines are "
                "dropped whole, because a memory cut mid-sentence still reads as a "
                "complete statement",
            ))
            continue
        if (single_line and publication is not None and publication.record_id == identifier
                and occurrences[line] == 1):
            selected.append(RenderedSelectedLine(publication, line, len(admitted)))
        admitted.append(line)
    return "\n".join(admitted), tuple(dropped), tuple(selected)