symfonic.evals.fixture_ingestion

fixture_ingestion

Public ingestion plans for versioned evaluation fixtures.

The two supported routes deliberately share a section identity but not an implementation. The default scaffold asks its public remember tool to commit each section in its own turn. An adopter that composes document knowledge can instead expose the same checksum-pinned sections through the public :func:symfonic.capabilities.knowledge.knowledge_sources door.

Neither route treats a fluent acknowledgement as proof. The remember journey requires a successful tool result for every section and finishes by checking payload-free record evidence for exact coverage and duplicate publication. The document adapter pins every public document revision to the fixture checksum and rejects duplicate or unknown requested ids before prompt assembly.

FixtureDocument dataclass

FixtureDocument(document_id: str, title: str, text: str, revision: str = '', media_type: str = 'text/plain')

A document value consumed structurally by the public knowledge bridge.

FixtureDocumentStore dataclass

FixtureDocumentStore(fixture: BookFixture = load_book_fixture())

A checksum-versioned public DocumentStore over one book fixture.

document_ids property

document_ids: tuple[str, ...]

Fixture section ids in manifest order.

fetch

fetch(document_id: str) -> FixtureDocument | None

Satisfy the public knowledge DocumentStore protocol.

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

def fetch(self, document_id: str) -> FixtureDocument | None:
    """Satisfy the public knowledge ``DocumentStore`` protocol."""
    return self._documents.get(document_id)

select_ids

select_ids(section_ids: Sequence[str] | None = None) -> tuple[str, ...]

Validate an exact, duplicate-free subset for knowledge_sources.

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

def select_ids(self, section_ids: Sequence[str] | None = None) -> tuple[str, ...]:
    """Validate an exact, duplicate-free subset for ``knowledge_sources``."""
    selected = self.document_ids if section_ids is None else tuple(section_ids)
    if len(selected) != len(set(selected)):
        raise ValueError("fixture knowledge section ids must not be duplicated")
    unknown = tuple(
        section_id for section_id in selected if section_id not in self._documents
    )
    if unknown:
        raise ValueError(f"unknown fixture knowledge section ids: {unknown!r}")
    return selected

FixtureIngestionComplete dataclass

FixtureIngestionComplete(fixture: BookFixture = load_book_fixture(), attribute: str = 'fixture_ingestion_records', name: str = 'fixture_ingestion_complete')

Require one canonical, checksum-matching record per fixture section.

Targets expose only identity evidence under fixture_ingestion_records: section_id, source_sha256, record_id and layer. Memory content never enters the report.

FixtureIngestionMode

Bases: StrEnum

The public application seam used to ingest the fixture.

FixtureIngestionReport dataclass

FixtureIngestionReport(fixture_id: str, routes: tuple[IngestionApplicability, ...])

Applicability for both fixture ingestion routes, never a silent skip.

not_applicable property

not_applicable: tuple[str, ...]

Stable mode names that were not offered by this deployment.

as_dict

as_dict() -> dict[str, object]

Return a JSON-safe, content-free report.

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

def as_dict(self) -> dict[str, object]:
    """Return a JSON-safe, content-free report."""
    return {
        "schema_version": 1,
        "fixture_id": self.fixture_id,
        "routes": [
            {
                "mode": row.mode.value,
                "applicable": row.applicable,
                "missing": list(row.missing),
                "reason": row.reason,
            }
            for row in self.routes
        ],
        "not_applicable": list(self.not_applicable),
    }

IngestionApplicability dataclass

IngestionApplicability(mode: FixtureIngestionMode, applicable: bool, missing: tuple[str, ...] = ())

One explicit applicable/not-applicable ingestion verdict.

reason property

reason: str

A payload-free explanation suitable for an evaluation report.

fixture_ingestion_applicability

fixture_ingestion_applicability(*, fixture: BookFixture | None = None, registered_tools: Iterable[str] = (), document_store: FixtureDocumentStore | None = None) -> FixtureIngestionReport

Report which ingestion routes the adopter actually supplied.

knowledge is not a compiled capability name in Symfonic; documents are sources composed through prompting. Applicability therefore follows the concrete public adapter, not a fictitious capability flag. The default route follows the registered tool palette for the same reason.

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

def fixture_ingestion_applicability(
    *,
    fixture: BookFixture | None = None,
    registered_tools: Iterable[str] = (),
    document_store: FixtureDocumentStore | None = None,
) -> FixtureIngestionReport:
    """Report which ingestion routes the adopter actually supplied.

    ``knowledge`` is not a compiled capability name in Symfonic; documents are
    sources composed through prompting.  Applicability therefore follows the
    concrete public adapter, not a fictitious capability flag.  The default
    route follows the registered tool palette for the same reason.
    """
    book = fixture or load_book_fixture()
    if document_store is not None and document_store.fixture.id != book.id:
        raise ValueError("the document adapter belongs to a different fixture")
    tools = _names(registered_tools)
    remember_missing = () if "remember" in tools else ("registered tool remember",)
    knowledge_missing = () if document_store is not None else ("fixture document source",)
    return FixtureIngestionReport(
        fixture_id=book.id,
        routes=(
            IngestionApplicability(
                FixtureIngestionMode.REMEMBER_TOOL,
                applicable=not remember_missing,
                missing=remember_missing,
            ),
            IngestionApplicability(
                FixtureIngestionMode.KNOWLEDGE_SOURCE,
                applicable=not knowledge_missing,
                missing=knowledge_missing,
            ),
        ),
    )

remember_fixture_scenario

remember_fixture_scenario(fixture: BookFixture | None = None, *, conversation: str = 'glass-harbor-ingestion', scope: str = 'owner') -> Scenario

Build SC-05: one successful remember turn per checksum-pinned section.

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

def remember_fixture_scenario(
    fixture: BookFixture | None = None,
    *,
    conversation: str = "glass-harbor-ingestion",
    scope: str = "owner",
) -> Scenario:
    """Build SC-05: one successful remember turn per checksum-pinned section."""
    book = fixture or load_book_fixture()
    steps = tuple(
        EvalStep(
            _remember_prompt(book, section),
            (ToolSucceeded("remember", times=1),),
            conversation=conversation,
            scope=scope,
        )
        for section in book.sections
    )
    audit = EvalStep(
        f"How many distinct source sections of {book.title} did I ask you to retain?",
        (ToolSucceeded("remember", times=0), FixtureIngestionComplete(book)),
        conversation=conversation,
        scope=scope,
    )
    return Scenario(
        "glass-harbor-book-ingestion",
        (*steps, audit),
        policy=TrialPolicy(timeout_seconds=180),
        tags=frozenset({"live", "memory", "book", "scaffold"}),
    )