symfonic.evals.volume

volume

Public SC-20 surface, grouped to keep the package facade bounded.

BoundedMemoryVolumeSafe dataclass

BoundedMemoryVolumeSafe(attribute: str = 'bounded_memory_volume', budgets: VolumeBudgets = VolumeBudgets(), name: str = 'bounded_memory_volume_safe')

Require bounded 10k behavior from identity/counter evidence only.

VolumeBatch dataclass

VolumeBatch(scope_path: str, records: tuple[VolumeRecord, ...])

One bounded write batch whose records all share scope.

VolumeBudgets dataclass

VolumeBudgets(minimum_records: int = TOTAL_RECORDS, exact_relations: int = 2500, minimum_relevant_distractors: int = RELEVANT_DISTRACTORS, retrieval_p95_ms: int = 1500, max_candidates_per_query: int = 128, max_operation_rows: int = 500, max_peak_rss_delta_bytes: int = 96 * 1024 * 1024, max_seed_batch: int = 200, max_admin_page: int = 200, max_browser_projection: int = 400, max_quick_examined: int = 500, max_deep_examined: int = 500, max_post_created_records: int = 10, max_post_deleted_records: int = 500, max_post_created_relations: int = 2500, max_post_deleted_relations: int = 500)

Hard ceilings asserted by the ordinary 10k release gate.

VolumeQuestion dataclass

VolumeQuestion(question_id: str, cue: str, expected_record_id: str)

A fixed cue and the durable identity it must retrieve.

expected_record_ids property

expected_record_ids: tuple[str, ...]

Complete deterministic top-five identity order for this cue.

VolumeRecord dataclass

VolumeRecord(record_id: str, layer: str, text: str, scope_path: str, salience: float, origin: str, edited_by: str = '', metadata: Mapping[str, object] = dict())

Eval-owned seed row; targets adapt it to their memory implementation.

iter_isolation_sentinels

iter_isolation_sentinels(tag: str) -> Iterator[VolumeBatch]

Yield non-counted controls in a sibling scope and a foreign tenant.

Source code in src/symfonic/evals/volume_corpus.py

def iter_isolation_sentinels(tag: str) -> Iterator[VolumeBatch]:
    """Yield non-counted controls in a sibling scope and a foreign tenant."""
    tenant = f"sc20-{tag}"
    scopes = (
        f"{tenant}/owner/sibling",
        f"sc20-other-{tag}/other/foreign",
    )
    for record_id, scope in zip(ISOLATION_SENTINEL_IDS, scopes, strict=True):
        yield VolumeBatch(
            scope,
            (
                VolumeRecord(
                    record_id=record_id,
                    layer="semantic",
                    text="Owner profile name is isolation sentinel",
                    scope_path=scope,
                    salience=1.0,
                    origin="sc20-isolation",
                ),
            ),
        )

iter_volume_batches

iter_volume_batches(tag: str, *, total: int = TOTAL_RECORDS, batch_size: int = _BATCH_SIZE) -> Iterator[VolumeBatch]

Yield exactly total rows in one tested scope, in bounded batches.

Isolation controls are deliberately separate. Mixing them into this population made 10_000 true while only 105 rows were visible and vectorised by the retrieval under test.

Source code in src/symfonic/evals/volume_corpus.py

def iter_volume_batches(
    tag: str,
    *,
    total: int = TOTAL_RECORDS,
    batch_size: int = _BATCH_SIZE,
) -> Iterator[VolumeBatch]:
    """Yield exactly ``total`` rows in one tested scope, in bounded batches.

    Isolation controls are deliberately separate.  Mixing them into this
    population made ``10_000`` true while only 105 rows were visible and
    vectorised by the retrieval under test.
    """
    charset = "abcdefghijklmnopqrstuvwxyz0123456789-"
    if not tag or any(character not in charset for character in tag):
        raise ValueError("volume corpus tag must use lowercase letters, digits, or hyphens")
    if total < TOTAL_RECORDS:
        raise ValueError(f"SC-20 requires at least {TOTAL_RECORDS} records")
    if not 1 <= batch_size <= _BATCH_SIZE:
        raise ValueError(f"SC-20 seed batches must be in 1..{_BATCH_SIZE}")

    tenant = f"sc20-{tag}"
    owner = f"{tenant}/owner/target"
    protected = _protected(owner)
    distractors = tuple(_distractor(owner, index) for index in range(RELEVANT_DISTRACTORS))
    yield from _chunk(owner, (*protected, *distractors), batch_size)

    produced = len(protected) + len(distractors)
    while produced < total:
        count = min(batch_size, total - produced)
        records = tuple(_ordinary(owner, produced + index) for index in range(count))
        yield VolumeBatch(owner, records)
        produced += count

volume_page_ids

volume_page_ids(offset: int, limit: int) -> tuple[str, ...]

Expected newest-first admin ids without materialising the corpus.

Source code in src/symfonic/evals/volume_corpus.py

def volume_page_ids(offset: int, limit: int) -> tuple[str, ...]:
    """Expected newest-first admin ids without materialising the corpus."""
    if offset < 0 or limit < 0:
        raise ValueError("volume page offset and limit must be non-negative")
    stop = min(offset + limit, TOTAL_RECORDS)
    return tuple(
        volume_record_id_at(TOTAL_RECORDS - 1 - position)
        for position in range(offset, stop)
    )

volume_record_id_at

volume_record_id_at(ordinal: int) -> str

Return the deterministic id at one target-corpus insertion ordinal.

Source code in src/symfonic/evals/volume_corpus.py

def volume_record_id_at(ordinal: int) -> str:
    """Return the deterministic id at one target-corpus insertion ordinal."""
    if not 0 <= ordinal < TOTAL_RECORDS:
        raise ValueError("volume ordinal is outside the fixed corpus")
    if ordinal < len(PROTECTED_FACT_IDS):
        return PROTECTED_FACT_IDS[ordinal]
    if ordinal < len(PROTECTED_FACT_IDS) + RELEVANT_DISTRACTORS:
        return f"relevant-distractor-{ordinal - len(PROTECTED_FACT_IDS):03d}"
    return f"volume-{ordinal:05d}"