symfonic.evals.error_recovery

error_recovery

Synthetic paired native/legacy tool-error recovery evaluation, not cutover.

Each route has fresh tools, ledger and provider. No provider is discovered or approved here. Fixed-script tests verify instrumentation, never real recovery.

RecoveryEvidence dataclass

RecoveryEvidence(route: str, answer: str, read_attempts: int, record_attempts: int, applied_effects: int, duplicate_effects: int, recorded_values: tuple[int, ...], model_calls: int, stop_reason: str | None, error_observation: str | None, legacy_body_entries: int, read_model_rounds: tuple[int, ...], record_model_rounds: tuple[int, ...])

passed property

passed: bool

Outcome oracle cannot pass on a scripted final answer alone.

run_pair async

run_pair(provider_factory: Callable[[], Any], *, native_factory: Callable[..., Any], legacy_factory: Callable[..., Any]) -> tuple[RecoveryEvidence, ...]

Provider factory must supply fresh state and separately approved config.

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

async def run_pair(
    provider_factory: Callable[[], Any],
    *,
    native_factory: Callable[..., Any],
    legacy_factory: Callable[..., Any],
) -> tuple[RecoveryEvidence, ...]:
    """Provider factory must supply fresh state and separately approved config."""
    native = provider_factory()
    legacy = provider_factory()
    if native is legacy:
        raise ValueError("paired routes require distinct provider instances")
    return (
        await run_trial(native, "native", native_factory),
        await run_trial(legacy, "legacy", legacy_factory),
    )

run_trial async

run_trial(provider: Any, route: str, agent_factory: Callable[..., Any]) -> RecoveryEvidence

Run one <=4-call, <=256-token-per-call trial inside a 60s timeout.

This exercises run() only. Legacy uses its actual natural fallback due to missing lazy prerequisites; counting wraps the live body without replacing its result. Raw synthetic tool errors only are retained, not whole prompts. Legacy has no exposed stop reason; None means UNKNOWN, not successful stop.

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

async def run_trial(
    provider: Any, route: str, agent_factory: Callable[..., Any]
) -> RecoveryEvidence:
    """Run one <=4-call, <=256-token-per-call trial inside a 60s timeout.

    This exercises run() only. Legacy uses its actual natural fallback due to
    missing lazy prerequisites; counting wraps the live body without replacing
    its result. Raw synthetic tool errors only are retained, not whole prompts.
    Legacy has no exposed stop reason; None means UNKNOWN, not successful stop.
    """
    if route not in {"native", "legacy"}:
        raise ValueError("route must be native or legacy")
    reads = 0
    records = 0
    entries = 0
    values: list[int] = []
    read_rounds: list[int] = []
    record_rounds: list[int] = []
    ledger = SideEffectLedger()

    @tool
    def read_value() -> int:
        """Read the synthetic value; retry once if the read fails."""
        nonlocal reads
        reads += 1
        read_rounds.append(budget.guard.calls)
        if reads == 1:
            raise ValueError("synthetic retry required")
        return 42

    @tool
    async def record_value(value: int) -> str:
        """Record the successfully read synthetic value exactly once."""
        nonlocal records
        records += 1
        record_rounds.append(budget.guard.calls)
        if value != 42 or reads < 2:
            raise ValueError("record requires a successful synthetic read of 42")
        receipt = await ledger.apply_once("synthetic-result", lambda: values.append(value))
        return "recorded" if receipt.applied else "duplicate record rejected"

    budget = RecoveryBudgetProvider(provider)
    recorder = ProviderEvidenceRecorder(budget)
    answer = ""
    stop = None
    host = None
    try:
        async with asyncio.timeout(60):
            if route == "native":
                host = agent_factory(recorder, [read_value, record_value])
                result = await host.run(PROMPT)
                answer, stop = result.text, result.stop_reason
            else:
                host = agent_factory(recorder, [read_value, record_value])
                original = host._legacy_run_impl

                async def counted(*args, **kwargs):
                    nonlocal entries
                    entries += 1
                    return await original(*args, **kwargs)

                host._legacy_run_impl = counted
                answer = (await host.run(PROMPT)).final_response or ""
                await host.flush_background_tasks()
    except BudgetConfigurationRefused:
        stop = "budget_configuration_refused"
    except CallBudgetExceeded:
        stop = "provider_call_limit"
    except TimeoutError:
        stop = "timeout"
    except Exception:
        # No provider exception content: it may contain credentials/endpoints.
        stop = "trial_error"
    if budget.guard.blocked:
        stop = "provider_call_limit"
    error = next(
        (
            str(message.content)
            for batch in recorder.calls[1:2]
            for message in batch
            if getattr(message, "type", None) == "tool"
            and "synthetic retry required" in str(message.content)
        ),
        None,
    )
    return RecoveryEvidence(
        route,
        answer,
        reads,
        records,
        ledger.effects,
        ledger.duplicates,
        tuple(values),
        budget.guard.calls,
        stop,
        error,
        entries,
        tuple(read_rounds),
        tuple(record_rounds),
    )