Deterministic, streaming input for the SC-20 bounded-volume gate.
The generator deliberately yields scope-homogeneous batches. A caller can
seed ten thousand rows without first building a ten-thousand-element list,
and every :class:~symfonic.capabilities.memory.records.WriteRequest remains
valid because all records in it name the same scope.
This is test data, not a retention policy. The five protected identifiers
are ground truth for the journey; a deployment proves their survival from
store identities before and after maintenance, never from model prose.
VolumeBatch
dataclass
VolumeBatch(scope_path: str, records: tuple[VolumeRecord, ...])
One bounded write batch whose records all share scope.
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}"
|