symfonic.evals.effects

effects

Side-effect evidence fixtures for replay and retry evaluations.

EffectReceipt dataclass

EffectReceipt(applied: bool, duplicate: bool, result: Any = None)

Whether this attempt applied the effect or found it already applied.

SideEffectCount dataclass

SideEffectCount(ledger: SideEffectLedger, attempts: int, effects: int, duplicates: int, name: str = 'side_effect_count')

Assert attempted, applied and duplicate counts without exposing keys.

SideEffectLedger

SideEffectLedger()

In-process fixture proving a tool honors a business idempotency key.

This is evaluation infrastructure, not a production durability mechanism. A deployed tool should use its transactional database or durable command ledger with the same apply_once contract. The fixture deliberately counts attempts separately from applied effects, so a replay cannot pass merely because the second request disappeared before reaching the tool.

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

def __init__(self) -> None:
    self._applied_keys: set[str] = set()
    self._attempts = 0
    self._duplicates = 0
    self._effects = 0
    self._lock = asyncio.Lock()

apply_once async

apply_once(key: str, operation: Callable[[], Any | Awaitable[Any]]) -> EffectReceipt

Apply operation at most once for key within this fixture.

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

async def apply_once(
    self,
    key: str,
    operation: Callable[[], Any | Awaitable[Any]],
) -> EffectReceipt:
    """Apply ``operation`` at most once for ``key`` within this fixture."""
    if not key:
        raise ValueError("a side effect requires a non-empty idempotency key")
    if not callable(operation):
        raise TypeError("operation must be callable")
    async with self._lock:
        self._attempts += 1
        if key in self._applied_keys:
            self._duplicates += 1
            return EffectReceipt(applied=False, duplicate=True)
        result = operation()
        if inspect.isawaitable(result):
            result = await result
        self._applied_keys.add(key)
        self._effects += 1
        return EffectReceipt(applied=True, duplicate=False, result=result)