symfonic.evals.runner

runner

Async evaluation runner with per-trial target isolation.

EvalTarget

Bases: Protocol

Application boundary driven by the runner.

resume async

resume(*, conversation: str, scope: str, answer: Mapping[str, Any]) -> Observation

Redeem one outstanding pause through the deployment's own seam.

Separate from :meth:observe because it is a separate operation: it takes no prompt, it continues a turn that already ran, and the evidence it publishes is about the redemption rather than about an answer the model produced. A target that cannot do it says so by publishing no resume in operations, which makes an approval pack not-applicable instead of failing it on delivery.

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

async def resume(
    self, *, conversation: str, scope: str, answer: Mapping[str, Any]
) -> Observation:
    """Redeem one outstanding pause through the deployment's own seam.

    Separate from :meth:`observe` because it is a separate operation: it
    takes no prompt, it continues a turn that already ran, and the
    evidence it publishes is about the redemption rather than about an
    answer the model produced. A target that cannot do it says so by
    publishing no ``resume`` in ``operations``, which makes an approval
    pack not-applicable instead of failing it on delivery.
    """
    ...

InfrastructureUnavailable

Bases: RuntimeError

The scenario could not ask its question because a dependency is absent.

run_scenario async

run_scenario(scenario: Scenario, factory: TargetFactory) -> ScenarioResult

Run a scenario sequentially so external side effects stay attributable.

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

async def run_scenario(scenario: Scenario, factory: TargetFactory) -> ScenarioResult:
    """Run a scenario sequentially so external side effects stay attributable."""
    trials = tuple(
        [await _trial(scenario, factory, index) for index in range(scenario.policy.trials)]
    )
    passed = sum(row.status is EvalStatus.PASSED for row in trials)
    if any(row.status is EvalStatus.ERROR for row in trials):
        status = EvalStatus.ERROR
    elif passed >= scenario.policy.required_passes:
        status = EvalStatus.PASSED
    elif all(row.status is EvalStatus.UNAVAILABLE for row in trials):
        status = EvalStatus.UNAVAILABLE
    else:
        status = EvalStatus.FAILED
    return ScenarioResult(
        name=scenario.name,
        status=status,
        trials=trials,
        required_passes=scenario.policy.required_passes,
        tags=scenario.tags,
    )

run_suite async

run_suite(scenarios: Iterable[Scenario], factory: TargetFactory, *, tags: frozenset[str] = frozenset()) -> EvalReport

Run selected scenarios and return a deterministic aggregate report.

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

async def run_suite(
    scenarios: Iterable[Scenario],
    factory: TargetFactory,
    *,
    tags: frozenset[str] = frozenset(),
) -> EvalReport:
    """Run selected scenarios and return a deterministic aggregate report."""
    selected = [row for row in scenarios if not tags or tags <= row.tags]
    if not selected:
        requested = ", ".join(sorted(tags)) or "<none>"
        raise ValueError(f"no evaluation scenarios matched required tags: {requested}")
    results = tuple([await run_scenario(row, factory) for row in selected])
    try:
        from importlib.metadata import version

        framework_version = version("symfonic-core")
    except Exception:  # pragma: no cover - editable/import-only environments
        framework_version = "unknown"
    return EvalReport(results, framework_version=framework_version)