Skip to content

symfonic.evals.execution_packs

execution_packs

Optional packs for capabilities that act during a turn.

Two families here; the procedural pack is next door in :mod:symfonic.evals.procedural_pack, at this module's size budget.

  • approval and resume need human and a target that publishes a resume operation, because the capability pauses and something else redeems;
  • extension tools need extensions and the exact tool attributed to it.

Every pack takes the deployment's own prompts, tool names and fragments. None of them ship content: a reusable pack that guessed a question would be asserting against a corpus the adopter does not have, and would pass or fail for reasons that are not the framework's.

The capability name is the entry condition and never the verdict. human travels even when a deployment wired no pause binding, so the packs below prove the behaviour on a turn -- a run that actually paused, a redemption that actually rebuilt the paused turn, a replayed token that was actually refused.

approval_resume_pack

approval_resume_pack(evidence: CapabilityEvidence, *, pause_prompt: str, answer: Mapping[str, Any], interaction_tool: str, resumable: bool = True, resume_attribute: str = 'resume_outcome', resume_outcome: str = 'resumed', continued_attribute: str = 'resume_continued', checkpoint_attribute: str = 'resume_checkpoint', replay_error: str = 'PauseTokenReplayedError', conversation: str = 'approval', policy: TrialPolicy = _DEFAULT_POLICY) -> PackResolution

Pause for a person, resume through the public seam, refuse the replay.

Three steps, and the middle one is the reason this pack exists.

The pause is read off the run's own terminal event, not off the answer: a turn that described a pause and finished anyway is exactly the failure this pack exists to catch. pause_resumable is the capability's measured report that the stopped turn reached a checkpointer, so a deployment that publishes a pause nobody can answer fails here rather than at the moment a person tries.

The resume is an :attr:~symfonic.evals.model.EvalOperation.RESUME step carrying the person's answer, so it goes through the deployment's own redemption operation and nothing else. Two earlier shapes are both excluded by construction: another prompt would only show that the model can be told about an answer, and spending the token directly would skip every check between authenticating it and consuming it. The step therefore requires three separate facts, because a deployment can satisfy any two of them while failing a person:

  • the outcome came back at all;
  • the token was bound to a checkpoint;
  • the paused turn was rebuilt from it. A deployment that validates the answer, spends the token, and has no recorded turn state to continue reports a perfectly successful redemption and leaves the run stopped forever.

The third step redeems the same retained token again and requires the named refusal, so a token that redeemed twice fails even though both redemptions returned an outcome.

Parameters:

Name Type Description Default
answer Mapping[str, Any]

what the person replies. It is validated by the deployment's registered response schema and against the recorded question, so an answer that does not fit fails the resume rather than being quietly accepted.

required
interaction_tool str

the registered interaction name. No default -- the built-in ask_user is one registration among a deployment's own, and assuming it would evaluate a different interaction than the one under test.

required
resumable bool

what the deployment claims its pause is. Stated rather than defaulted-away, because False is a legitimate wiring and an evaluation that always demanded True could not express it. A pause declared unresumable cannot then be required to continue, so the continuation assertion follows this flag.

True
Source code in src/symfonic/evals/execution_packs.py
def approval_resume_pack(
    evidence: CapabilityEvidence,
    *,
    pause_prompt: str,
    answer: Mapping[str, Any],
    interaction_tool: str,
    resumable: bool = True,
    resume_attribute: str = "resume_outcome",
    resume_outcome: str = "resumed",
    continued_attribute: str = "resume_continued",
    checkpoint_attribute: str = "resume_checkpoint",
    replay_error: str = "PauseTokenReplayedError",
    conversation: str = "approval",
    policy: TrialPolicy = _DEFAULT_POLICY,
) -> PackResolution:
    """Pause for a person, resume through the public seam, refuse the replay.

    Three steps, and the middle one is the reason this pack exists.

    The pause is read off the run's own terminal event, not off the answer: a
    turn that described a pause and finished anyway is exactly the failure this
    pack exists to catch. ``pause_resumable`` is the capability's *measured*
    report that the stopped turn reached a checkpointer, so a deployment that
    publishes a pause nobody can answer fails here rather than at the moment a
    person tries.

    The resume is an :attr:`~symfonic.evals.model.EvalOperation.RESUME` step
    carrying the person's answer, so it goes through the deployment's own
    redemption operation and nothing else. Two earlier shapes are both
    excluded by construction: another *prompt* would only show that the model
    can be told about an answer, and spending the token directly would skip
    every check between authenticating it and consuming it. The step therefore
    requires three separate facts, because a deployment can satisfy any two of
    them while failing a person:

    * the outcome came back at all;
    * the token was bound to a checkpoint;
    * the paused turn was *rebuilt* from it. A deployment that validates the
      answer, spends the token, and has no recorded turn state to continue
      reports a perfectly successful redemption and leaves the run stopped
      forever.

    The third step redeems the same retained token again and requires the
    named refusal, so a token that redeemed twice fails even though both
    redemptions returned an outcome.

    Args:
        answer: what the person replies. It is validated by the deployment's
            registered response schema and against the recorded question, so
            an answer that does not fit fails the resume rather than being
            quietly accepted.
        interaction_tool: the registered interaction name. No default -- the
            built-in ``ask_user`` is one registration among a deployment's
            own, and assuming it would evaluate a different interaction than
            the one under test.
        resumable: what the deployment claims its pause is. Stated rather than
            defaulted-away, because ``False`` is a legitimate wiring and an
            evaluation that always demanded ``True`` could not express it. A
            pause declared unresumable cannot then be required to continue, so
            the continuation assertion follows this flag.
    """
    if not interaction_tool:
        raise ValueError("an approval pack requires the registered interaction name")
    if not answer:
        raise ValueError(
            "an approval pack requires the answer a person gives; redeeming a "
            "token with nothing to validate proves only that it was spent"
        )

    def build() -> tuple[Scenario, ...]:
        redeemed: list[Any] = [
            AttributeEquals(resume_attribute, resume_outcome),
            AttributeEquals("resume_name", interaction_tool),
            AttributeEquals(checkpoint_attribute, True),
            AttributeEquals(continued_attribute, resumable),
            ToolCalled(interaction_tool, times=0),
        ]
        return (
            Scenario(
                "approval-pause-resume-and-replay-refusal",
                (
                    EvalStep(
                        pause_prompt,
                        (
                            AttributeEquals("paused", True),
                            AttributeEquals("pause_name", interaction_tool),
                            AttributeEquals("pause_resumable", resumable),
                        ),
                        conversation=conversation,
                    ),
                    EvalStep(
                        assertions=tuple(redeemed),
                        conversation=conversation,
                        operation=EvalOperation.RESUME,
                        answer=answer,
                    ),
                    EvalStep(
                        conversation=conversation,
                        operation=EvalOperation.RESUME,
                        answer=answer,
                        expected_error=replay_error,
                    ),
                ),
                policy=policy,
                tags=frozenset({"approval", "pack"}),
            ),
        )

    return resolve_pack(
        "approval-resume",
        evidence,
        build,
        capabilities=("human",),
        operations=("resume",),
    )

extension_tool_pack

extension_tool_pack(evidence: CapabilityEvidence, *, prompt: str, tool_name: str, arguments: dict[str, object] | None = None, expected_fragments: Sequence[str] = (), conversation: str = 'extensions', policy: TrialPolicy = _DEFAULT_POLICY) -> PackResolution

Run one tool attributed to the compiled extension contribution.

The umbrella capability is insufficient: an empty bundle, or one that contributes a different tool, is explicitly not applicable. When present, an extension reaches a turn as an executable tool, so the evidence is a call and an error-free result. Requiring only the call would pass for a declaration bound with nothing behind it -- the failure the extensions door was rebuilt to make impossible -- and requiring only the answer would pass for a model that described the tool it never called.

Source code in src/symfonic/evals/execution_packs.py
def extension_tool_pack(
    evidence: CapabilityEvidence,
    *,
    prompt: str,
    tool_name: str,
    arguments: dict[str, object] | None = None,
    expected_fragments: Sequence[str] = (),
    conversation: str = "extensions",
    policy: TrialPolicy = _DEFAULT_POLICY,
) -> PackResolution:
    """Run one tool attributed to the compiled extension contribution.

    The umbrella capability is insufficient: an empty bundle, or one that
    contributes a different tool, is explicitly not applicable. When present,
    an extension reaches a turn as an executable tool, so the evidence is a
    call *and* an error-free result. Requiring only the call would pass for a
    declaration bound with nothing behind it -- the failure the extensions door
    was rebuilt to make impossible -- and requiring only the answer would pass
    for a model that described the tool it never called.
    """
    if not tool_name:
        raise ValueError("an extension pack requires the contributed tool name")

    def build() -> tuple[Scenario, ...]:
        assertions: list[object] = [
            ToolCalled(tool_name, times=1, arguments=arguments),
            ToolSucceeded(tool_name, times=1),
            ToolCallIdsUnique(minimum=1),
        ]
        if expected_fragments:
            assertions.append(ResponseContains(*expected_fragments))
        return (
            Scenario(
                "extension-tool-execution",
                (
                    EvalStep(
                        prompt,
                        tuple(assertions),  # type: ignore[arg-type]
                        conversation=conversation,
                    ),
                ),
                policy=policy,
                tags=frozenset({"extensions", "pack"}),
            ),
        )

    return resolve_pack(
        "extension-tools",
        evidence,
        build,
        capabilities=("extensions",),
        capability_tools=(("extensions", tool_name),),
    )