Standard Nodes and Effects

Generic Utilities

spawn_node

spawn_node(node_type: type[N], **args: Any) -> NodeBuilder[Any, Any]

A convenience helper to write effect triggering functions.

Attributes:

Name	Type	Description
node_type		The type of the node to spawn (e.g., Branch).
args		Arguments to populate the node fields, passed to Node.spawn.

Source code in src/delphyne/stdlib/nodes.py

def spawn_node[N: dp.Node](
    node_type: type[N], **args: Any
) -> dp.NodeBuilder[Any, Any]:
    """
    A convenience helper to write effect triggering functions.

    Attributes:
        node_type: The type of the node to spawn (e.g., `Branch`).
        args: Arguments to populate the node fields, passed to
            [`Node.spawn`][delphyne.core.trees.Node.spawn].
    """
    return dp.NodeBuilder(lambda spawner: node_type.spawn(spawner, **args))

FromPolicy

FromPolicy: Callable[[Any], T]

Type for an inner-policy-dependent data field.

We use Any instead of introducing an inner policy type parameter P, since Node is not parametric either. Thus, this alias is mostly meant for readability and expressing intent.

NodeMeta

Abstract base class for node metadata.

Nodes can feature fields with arbitrary metadata accessible to policies (e.g., meta field of Branch). Typing those fields with NodeMeta instead of object or Any allows for better type safety. In particular, it prevents errors that arise from accidentally passing uninstantiated parametric inner policy fields.

Source code in src/delphyne/stdlib/nodes.py

class NodeMeta:
    """
    Abstract base class for node metadata.

    Nodes can feature fields with arbitrary metadata accessible to
    policies (e.g., `meta` field of `Branch`). Typing those fields with
    `NodeMeta` instead of `object` or `Any` allows for better type
    safety. In particular, it prevents errors that arise from
    accidentally passing uninstantiated parametric inner policy fields.
    """

    pass

Branch

Branch dataclass

Bases: Node

The standard Branch effect.

Can be triggered using the branch function, which allows branching over elements of an opaque space.

Source code in src/delphyne/stdlib/nodes.py

@dataclass(frozen=True)
class Branch(dp.Node):
    """
    The standard `Branch` effect.

    Can be triggered using the `branch` function, which allows branching
    over elements of an opaque space.
    """

    cands: OpaqueSpace[Any, Any]
    meta: FromPolicy[NodeMeta] | None

    @override
    def navigate(self) -> dp.Navigation:
        return (yield self.cands)

    @override
    def primary_space(self) -> dp.Space[Any]:
        return self.cands

branch

branch(
    cands: Opaque[P, T],
    meta: Callable[[P], NodeMeta] | None = None,
    inner_policy_type: type[P] | None = None,
) -> Strategy[Branch, P, T]

Branch over the elements of an opaque space.

Parameters:

Name	Type	Description	Default
cands	Opaque[P, T]	An opaque space, which can be defined from either a query or a strategy via the using method.	required
meta	Callable[[P], NodeMeta] | None	An optional mapping from the ambient inner policy to arbitrary metadata accessible to search policies.	None
inner_policy_type	type[P] | None	Ambient inner policy type. This information is not used at runtime but it can be provided to help type inference when necessary.	None

Source code in src/delphyne/stdlib/nodes.py

def branch[P, T](
    cands: Opaque[P, T],
    meta: Callable[[P], NodeMeta] | None = None,
    inner_policy_type: type[P] | None = None,
) -> dp.Strategy[Branch, P, T]:
    """
    Branch over the elements of an opaque space.

    Arguments:
        cands: An opaque space, which can be defined from either a query
            or a strategy via the `using` method.
        meta: An optional mapping from the ambient inner policy to
            arbitrary metadata accessible to search policies.
        inner_policy_type: Ambient inner policy type. This information
            is not used at runtime but it can be provided to help type
            inference when necessary.
    """
    ret = yield spawn_node(Branch, cands=cands, meta=meta)
    return cast(T, ret)

Fail

Fail dataclass

Bases: Node

The standard Fail effect.

Can be triggered using the fail and ensure functions.

Source code in src/delphyne/stdlib/nodes.py

@dataclass(frozen=True)
class Fail(dp.Node):
    """
    The standard `Fail` effect.

    Can be triggered using the `fail` and `ensure` functions.
    """

    error: dp.Error

    @override
    def leaf_node(self) -> bool:
        return True

    @override
    def navigate(self) -> dp.Navigation:
        assert False
        yield

    @override
    def summary_message(self) -> str:
        return str(self.error)

fail

fail(
    label: str | None = None, *, message: str | None = None, error: Error | None = None
) -> Strategy[Fail, object, NoReturn]

Fail immediately with an error.

The error can be specified using the error keyword argument. As a shortcut, the label and message arguments can also be used to directly specify the corresponding fields of the Error type. Those arguments can only be used if error is not provided.

Warning

Like all effect triggering functions, this function must be invoked as:

yield from fail(...)

Forgetting yield from may not result in a type error but will result in a no-op at runtime.

Source code in src/delphyne/stdlib/nodes.py

def fail(
    label: str | None = None,
    *,
    message: str | None = None,
    error: dp.Error | None = None,
) -> dp.Strategy[Fail, object, NoReturn]:
    """
    Fail immediately with an error.

    The error can be specified using the `error` keyword argument. As a
    shortcut, the `label` and `message` arguments can also be used to
    directly specify the corresponding fields of the `Error` type. Those
    arguments can only be used if `error` is not provided.

    !!! warning
        Like all effect triggering functions, this function must be
        invoked as:

            yield from fail(...)

        Forgetting `yield from` may not result in a type error but will
        result in a no-op at runtime.
    """
    yield spawn_node(Fail, error=_build_error(message, label, error))
    assert False

ensure

ensure(
    prop: bool,
    label: str | None = None,
    *,
    message: str | None = None,
    error: Error | None = None,
) -> Strategy[Fail, object, None]

Ensure that a property holds, otherwise fail with an error.

See fail regarding the label, message and error arguments.

Warning

Like all effect triggering functions, this function must be invoked as:

yield from ensure(...)

Forgetting yield from may not result in a type error but will result in a no-op at runtime.

Source code in src/delphyne/stdlib/nodes.py

def ensure(
    prop: bool,
    label: str | None = None,
    *,
    message: str | None = None,
    error: dp.Error | None = None,
) -> dp.Strategy[Fail, object, None]:
    """
    Ensure that a property holds, otherwise fail with an error.

    See `fail` regarding the `label`, `message` and `error` arguments.

    !!! warning
        Like all effect triggering functions, this function must be
        invoked as:

            yield from ensure(...)

        Forgetting `yield from` may not result in a type error but will
        result in a no-op at runtime.
    """
    if not prop:
        yield from fail(label=label, message=message, error=error)

Message

Message dataclass

Bases: Node

The standard Message effect.

Message nodes are tree nodes carrying a message. They have a unique child. They can be eliminated using the elim_messages tree transformer, which automatically logs their content.

This effect is useful for debugging strategies. Using print statements in strategies is discouraged since strategy computations are replayed every time a child of the associated tree is computed, causing the same message to be repeatedly printed.

Source code in src/delphyne/stdlib/nodes.py

@dataclass
class Message(dp.Node):
    """
    The standard `Message` effect.

    Message nodes are tree nodes carrying a message. They have a unique
    child. They can be eliminated using the `elim_messages` tree
    transformer, which automatically logs their content.

    This effect is useful for debugging strategies. Using `print`
    statements in strategies is discouraged since strategy computations
    are replayed every time a child of the associated tree is computed,
    causing the same message to be repeatedly printed.
    """

    msg: str
    data: object | None

    @override
    def navigate(self) -> dp.Navigation:
        return None
        yield

    @override
    def summary_message(self) -> str:
        return self.msg

message

message(msg: str, data: object | None = None) -> Strategy[Message, object, None]

Log a debugging message.

Parameters:

Name	Type	Description	Default
msg	str	The message to log.	required
data	object | None	Optional data to attach to the message.	None

Warning

Like all effect triggering functions, this function must be invoked as:

yield from message(...)

Forgetting yield from may not result in a type error but will result in a no-op at runtime.

Source code in src/delphyne/stdlib/nodes.py

def message(
    msg: str, data: object | None = None
) -> dp.Strategy[Message, object, None]:
    """
    Log a debugging message.

    Arguments:
        msg: The message to log.
        data: Optional data to attach to the message.

    !!! warning
        Like all effect triggering functions, this function must be
        invoked as:

            yield from message(...)

        Forgetting `yield from` may not result in a type error but will
        result in a no-op at runtime.
    """
    yield spawn_node(Message, msg=msg, data=data)

elim_messages

elim_messages(
    env: PolicyEnv, policy: Any, show_in_log: bool = True
) -> PureTreeTransformerFn[Message, Never]

Eliminate the Message effect.

Parameters:

Name	Type	Description	Default
show_in_log	bool	Whether to log the associated content whenever a message node is encountered.	True

Source code in src/delphyne/stdlib/nodes.py

@pol.contextual_tree_transformer
def elim_messages(
    env: PolicyEnv,
    policy: Any,
    show_in_log: bool = True,
) -> pol.PureTreeTransformerFn[Message, Never]:
    """
    Eliminate the `Message` effect.

    Arguments:
        show_in_log: Whether to log the associated content whenever a
            message node is encountered.
    """

    def transform[N: dp.Node, P, T](
        tree: dp.Tree[Message | N, P, T],
    ) -> dp.Tree[N, P, T]:
        if isinstance(tree.node, Message):
            if show_in_log:
                metadata = {"attached": tree.node.data}
                env.tracer.log(tree.node.msg, metadata=metadata)
            return transform(tree.child(None))
        return tree.transform(tree.node, transform)

    return transform

Factor and Value

Factor dataclass

Bases: Node

The standard Factor effect.

A Factor node allows computing a confidence score in the [0, 1] interval. This confidence can be computed by a query or a dedicated strategy but only one element will be generated from the resulting space. Instead of having an oracle compute a numerical value directly, it computes an evaluation object that is then transformed into a number using a policy-provided function. This allows greater flexibility on the policy side. If no such function is given, the whole node is ignored.

Source code in src/delphyne/stdlib/nodes.py

@dataclass(frozen=True)
class Factor(dp.Node):
    """
    The standard `Factor` effect.

    A `Factor` node allows computing a confidence score in the [0, 1]
    interval. This confidence can be computed by a query or a dedicated
    strategy but only one element will be generated from the resulting
    space. Instead of having an oracle compute a numerical value
    directly, it computes an evaluation object that is then transformed
    into a number using a policy-provided function. This allows greater
    flexibility on the policy side. If no such function is given, the
    whole node is ignored.
    """

    eval: OpaqueSpace[Any, Any]
    factor: FromPolicy[Callable[[Any], float] | None]

    @override
    def navigate(self) -> dp.Navigation:
        return None
        yield

    @override
    def primary_space(self):
        return self.eval

factor

factor(
    eval: Opaque[P, E],
    factor: Callable[[P], Callable[[E], float] | None],
    inner_policy_type: type[P] | None = None,
) -> Strategy[Factor, P, None]

Apply a multiplicative penalty to the current search branch.

Parameters:

Name	Type	Description	Default
eval	Opaque[P, E]	An opaque space, typically induced by a strategy or a query whose purpose is to evaluate the current search state, returning an evaluation object of type E. Policies typically extract a single element from this space.	required
factor	Callable[[P], Callable[[E], float] | None]	An inner-policy-dependent function that maps an evaluation object of type E to an actual numerical penalty, as a multiplicative factor in the [0, 1] interval. If no function is provided, the whole node is ignored. Separating eval and factor allows greater flexibility for policies to tune how multidimensional state evaluations are reduced into a single numbers.	required

Warning

Like all effect triggering functions, this function must be invoked as:

yield from factor(...)

Forgetting yield from may not result in a type error but will result in a no-op at runtime.

Source code in src/delphyne/stdlib/nodes.py

def factor[E, P](
    eval: Opaque[P, E],
    factor: Callable[[P], Callable[[E], float] | None],
    inner_policy_type: type[P] | None = None,
) -> dp.Strategy[Factor, P, None]:
    """
    Apply a multiplicative penalty to the current search branch.

    Arguments:
        eval: An opaque space, typically induced by a strategy or a
            query whose purpose is to evaluate the current search state,
            returning an evaluation object of type `E`. Policies
            typically extract a single element from this space.
        factor: An inner-policy-dependent function that maps an
            evaluation object of type `E` to an actual numerical
            penalty, as a multiplicative factor in the [0, 1] interval.
            If no function is provided, the whole node is ignored.
            Separating `eval` and `factor` allows greater flexibility
            for policies to tune how multidimensional state evaluations
            are reduced into a single numbers.

    !!! warning
        Like all effect triggering functions, this function must be
        invoked as:

            yield from factor(...)

        Forgetting `yield from` may not result in a type error but will
        result in a no-op at runtime.
    """
    yield spawn_node(Factor, eval=eval, factor=factor)

Value dataclass

Bases: Node

The standard Value effect.

Similar to Factor, except that the resulting number is used to set the whole value of the branch instead of just multiplying it.

Source code in src/delphyne/stdlib/nodes.py

@dataclass(frozen=True)
class Value(dp.Node):
    """
    The standard `Value` effect.

    Similar to `Factor`, except that the resulting number is used to set
    the whole value of the branch instead of just multiplying it.
    """

    eval: OpaqueSpace[Any, Any]
    value: FromPolicy[Callable[[Any], float] | None]

    @override
    def navigate(self) -> dp.Navigation:
        return None
        yield

    @override
    def primary_space(self):
        return self.eval

value

value(
    eval: Opaque[P, E],
    value: Callable[[P], Callable[[E], float] | None],
    inner_policy_type: type[P] | None = None,
) -> Strategy[Value, P, None]

Set the value of the current search branch.

See the similar factor function for more details.

Warning

Like all effect triggering functions, this function must be invoked as:

yield from message(...)

Forgetting yield from may not result in a type error but will result in a no-op at runtime.

Source code in src/delphyne/stdlib/nodes.py

def value[E, P](
    eval: Opaque[P, E],
    value: Callable[[P], Callable[[E], float] | None],
    inner_policy_type: type[P] | None = None,
) -> dp.Strategy[Value, P, None]:
    """
    Set the value of the current search branch.

    See the similar `factor` function for more details.

    !!! warning
        Like all effect triggering functions, this function must be
        invoked as:

            yield from message(...)

        Forgetting `yield from` may not result in a type error but will
        result in a no-op at runtime.
    """
    yield spawn_node(Value, eval=eval, value=value)

Join

Join dataclass

Bases: Node

The standard Join effect.

This effect can be triggered using the join function. A Join node features a sequence of embedded trees.

Source code in src/delphyne/stdlib/nodes.py

@dataclass(frozen=True)
class Join(dp.Node):
    """
    The standard `Join` effect.

    This effect can be triggered using the `join` function. A `Join`
    node features a sequence of embedded trees.
    """

    subs: Sequence[dp.EmbeddedTree[Any, Any, Any]]
    meta: FromPolicy[NodeMeta] | None

    @override
    def navigate(self) -> dp.Navigation:
        ret: list[Any] = []
        for sub in self.subs:
            ret.append((yield sub))
        return tuple(ret)

join

join(
    subs: Sequence[StrategyComp[N, P, T]],
    meta: Callable[[P], NodeMeta] | None = None,
) -> Strategy[N, P, Sequence[T]]

Evaluate a sequence of independent strategy computations, possibly in parallel.

Parameters:

Name	Type	Description	Default
subs	Sequence[StrategyComp[N, P, T]]	A sequence of strategy computations to evaluate.	required
meta	Callable[[P], NodeMeta] | None	An optional mapping from the ambient inner policy to arbitrary metadata accessible to search policies.	None

Returns:

Type	Description
Strategy[N, P, Sequence[T]]	A sequence featuring all computation results.

Source code in src/delphyne/stdlib/nodes.py

def join[N: dp.Node, P, T](
    subs: Sequence[dp.StrategyComp[N, P, T]],
    meta: Callable[[P], NodeMeta] | None = None,
) -> dp.Strategy[N, P, Sequence[T]]:
    """
    Evaluate a sequence of independent strategy computations, possibly
    in parallel.

    Arguments:
        subs: A sequence of strategy computations to evaluate.
        meta: An optional mapping from the ambient inner policy to
            arbitrary metadata accessible to search policies.

    Returns:
        A sequence featuring all computation results.
    """
    ret = yield spawn_node(Join, subs=subs, meta=meta)
    return cast(Sequence[T], ret)

Compute

Compute dataclass

Bases: ComputationNode

The standard Compute effect.

For efficiency and replicability reasons, strategies must not directly perform expensive and possibly non-replicable computations. For example, a strategy should not directly call an SMT solver since:

• The call may be expensive, and stratgy computations are replayed from scratch every time a child is computed in the corresponding tree (see documentation for reify).
• SMT solvers using wall-time timeouts may return different results when called repeatedly on the same input.

The Compute effect allows performing an expensive and possibly non-deterministic computation by issuing a special __Computation__ query that specifies the computation to be performed. Such a query is not answered by an LLM, but by actually performing the described computation. Special support is available in the demonstration interpreter in the form of implicit answers, allowing __Computation__ queries to be automatically answered when running tests. Generated answers can be hardcoded in demonstrations after the fact via proper editor support.

Source code in src/delphyne/stdlib/computations.py

@dataclass
class Compute(dp.ComputationNode):
    """
    The standard `Compute` effect.

    For efficiency and replicability reasons, strategies must not directly
    perform expensive and possibly non-replicable computations. For example,
    a strategy should not directly call an SMT solver since:

    - The call may be expensive, and stratgy computations are replayed from
    scratch every time a child is computed in the corresponding tree (see
    documentation for `reify`).
    - SMT solvers using wall-time timeouts may return different results when
    called repeatedly on the same input.

    The `Compute` effect allows performing an expensive and possibly
    non-deterministic computation by issuing a special `__Computation__`
    query that specifies the computation to be performed. Such a query is
    not answered by an LLM, but by _actually_ performing the described
    computation. Special support is available in the demonstration
    interpreter in the form of *implicit answers*, allowing
    `__Computation__` queries to be automatically answered when running
    tests. Generated answers can be hardcoded in demonstrations **after**
    the fact via proper editor support.
    """

    query: dp.TransparentQuery[Any]
    _comp: Callable[[], Any]
    _ret_type: ty.TypeAnnot[Any]

    def navigate(self) -> dp.Navigation:
        return (yield self.query)

    def run_computation(self) -> str:
        ret = self._comp()
        serialized = dump_yaml(self._ret_type, ret)
        return serialized

    def run_computation_with_cache(self, cache: dp.LLMCache | None) -> str:
        """
        Run the computation using a fake oracle so that the LLM caching
        mechanism can be reused.
        """
        chat = create_prompt(
            self.query.attached.query,
            examples=[],
            params={},
            mode=None,
            env=None,
        )
        req = dp.LLMRequest(chat=chat, num_completions=1, options={})
        model = ComputationOracle(self.run_computation)
        resp = model.send_request(req, cache)
        assert len(resp.outputs) == 1
        answer = resp.outputs[0].content
        assert isinstance(answer, str)
        return answer

compute

compute(
    f: Callable[P, T], *args: P.args, **kwargs: P.kwargs
) -> Strategy[Compute, object, T]

Triggering function for the Compute effect.

Parameters:

Name	Type	Description	Default
f	Callable[P, T]	Function performing an expensive computation. It must feature type annotations and its arguments must be JSON-serializable. It does not need to be deterministic.	required
*args	P.args	Positional arguments to pass to f.	()
**kwargs	P.kwargs	Keyword arguments to pass to f.	{}

Source code in src/delphyne/stdlib/computations.py

def compute[**P, T](
    f: Callable[P, T], *args: P.args, **kwargs: P.kwargs
) -> dp.Strategy[Compute, object, T]:
    """
    Triggering function for the `Compute` effect.

    Arguments:
        f: Function performing an expensive computation. It must feature
            type annotations and its arguments must be
            JSON-serializable. It does not need to be deterministic.
        *args: Positional arguments to pass to `f`.
        **kwargs: Keyword arguments to pass to `f`.
    """
    comp = partial(f, *args, **kwargs)
    ret_type = insp.function_return_type(f)
    assert not isinstance(ret_type, ty.NoTypeInfo)
    fun_args = insp.function_args_dict(f, args, kwargs)
    fun = insp.function_name(f)
    assert fun is not None
    query = dp.TransparentQuery.build(__Computation__(fun, fun_args))
    unparsed = yield spawn_node(
        Compute, query=query, _comp=comp, _ret_type=ret_type
    )
    ret = ty.pydantic_load(ret_type, unparsed)
    return cast(T, ret)

elim_compute

elim_compute(
    env: PolicyEnv, policy: Any, force_bypass_cache: bool = False
) -> PureTreeTransformerFn[Compute, Never]

Eliminate the Compute effect by performing the computation when computing tree children (making the child function possibly nondeterministic).

Parameters:

Name	Type	Description	Default
force_bypass_cache	bool	if set to True, do not cache computation results, even if a cache is available in the global policy environment.	False

Source code in src/delphyne/stdlib/computations.py

@dp.contextual_tree_transformer
def elim_compute(
    env: PolicyEnv,
    policy: Any,
    force_bypass_cache: bool = False,
) -> dp.PureTreeTransformerFn[Compute, Never]:
    """
    Eliminate the `Compute` effect by performing the computation when
    computing tree children (making the `child` function possibly
    nondeterministic).

    Arguments:
        force_bypass_cache: if set to `True`, do not cache computation
            results, even if a cache is available in the global policy
            environment.
    """

    def transform[N: dp.Node, P, T](
        tree: dp.Tree[Compute | N, P, T],
    ) -> dp.Tree[N, P, T]:
        if isinstance(tree.node, Compute):
            cache = None
            if not force_bypass_cache:
                cache = env.requests_cache
            answer = tree.node.run_computation_with_cache(cache)
            tracked = tree.node.query.attached.parse_answer(
                dp.Answer(None, answer)
            )
            assert not isinstance(tracked, dp.ParseError)
            return transform(tree.child(tracked))

        return tree.transform(tree.node, transform)

    return transform

__Computation__ dataclass

Bases: AbstractQuery[object]

A special query that represents a computation.

Returns a parsed JSON result.

Attributes:

Name	Type	Description
fun	str	Name of the function to call.
args	dict[str, Any]	Arguments to pass to the function, as a dictionary.

Source code in src/delphyne/stdlib/computations.py

@dataclass
class __Computation__(dp.AbstractQuery[object]):
    """
    A special query that represents a computation.

    Returns a parsed JSON result.

    Attributes:
        fun: Name of the function to call.
        args: Arguments to pass to the function, as a dictionary.
    """

    fun: str
    args: dict[str, Any]

    @override
    def generate_prompt(
        self,
        *,
        kind: str,
        mode: dp.AnswerMode,
        params: dict[str, object],
        extra_args: dict[str, object] | None = None,
        env: dp.AbstractTemplatesManager | None = None,
    ):
        return dump_yaml(Any, self.__dict__)

    @override
    def query_modes(self):
        return [None]

    @override
    def answer_type(self):
        return object

    @override
    def parse_answer(self, answer: dp.Answer) -> object | dp.ParseError:
        try:
            assert isinstance(answer.content, str)
            return yaml.safe_load(answer.content)
        except yaml.parser.ParserError as e:
            return dp.ParseError(description=str(e))

Flag

Flag dataclass

Bases: Node

The standard Flag effect.

Flags allow providing several implementations for a strategy component, and have policies select which variant to use (or perform search at runtime for selecting variants).

For each flag, a subclass of FlagQuery must be defined, which admits a finite set of answers (one per allowed flag value), along with a default answer. Type parameter F denotes the type of the flag query that can be issued. To express a signature wih several flag queries, use Flag[A] | Flag[B] instead of Flag[A | B], so that both kinds of flags can be eliminated separately.

Behavior in demonstrations

Because flag queries override AbstractQuery.default_answer, default flag values are automatically selected by the demonstration interpreter. This behaviour can be overriden by adding answers for flag queries in the queries section, or by using value-based hints (i.e., #flag_value, which is possible since flag queries implement AbstractQuery.finite_answer_set).

Source code in src/delphyne/stdlib/flags.py

@dataclass(frozen=True)
class Flag[F: FlagQuery[Any]](dp.Node):
    """
    The standard `Flag` effect.

    Flags allow providing several implementations for a strategy
    component, and have policies select which variant to use (or perform
    search at runtime for selecting variants).

    For each flag, a subclass of `FlagQuery` must be defined, which
    admits a finite set of answers (one per allowed flag value), along
    with a default answer. Type parameter `F` denotes the type of the
    flag query that can be issued. To express a signature wih several
    flag queries, use `Flag[A] | Flag[B]` instead of `Flag[A | B]`, so
    that both kinds of flags can be eliminated separately.

    ??? info "Behavior in demonstrations"
        Because flag queries override `AbstractQuery.default_answer`,
        default flag values are automatically selected by the
        demonstration interpreter. This behaviour can be overriden by
        adding answers for flag queries in the `queries` section, or by
        using value-based hints (i.e., `#flag_value`, which is possible
        since flag queries implement `AbstractQuery.finite_answer_set`).
    """

    flag: dp.TransparentQuery[Any]

    def summary_message(self) -> str:
        query = self.flag.attached.query
        assert isinstance(query, FlagQuery)
        name = query.query_name()
        return f"{name}: {', '.join(query.flag_values())}"

    def navigate(self) -> dp.Navigation:
        return (yield self.flag)

get_flag

get_flag(flag: type[FlagQuery[T]]) -> Strategy[Flag[Any], object, T]

Triggering function for the Flag effect.

Takes a flag query type as an argument and return a flag value.

Info

A more precise type cannot be given for this function since Python does not support higher-kinded types.

Source code in src/delphyne/stdlib/flags.py

def get_flag[T: str](
    flag: type[FlagQuery[T]],
) -> dp.Strategy[Flag[Any], object, T]:
    """
    Triggering function for the `Flag` effect.

    Takes a flag query type as an argument and return a flag value.

    !!! info
        A more precise type cannot be given for this function since
        Python does not support higher-kinded types.
    """
    query = dp.TransparentQuery.build(flag())
    ret = yield spawn_node(Flag, flag=query)
    return cast(T, ret)

elim_flag

elim_flag(
    env: PolicyEnv, policy: Any, flag: type[F], val: str
) -> PureTreeTransformerFn[Flag[F], Never]

Source code in src/delphyne/stdlib/flags.py

@dp.contextual_tree_transformer
def elim_flag[F: FlagQuery[Any]](
    env: PolicyEnv,
    policy: Any,
    flag: type[F],
    val: str,
) -> dp.PureTreeTransformerFn[Flag[F], Never]:
    assert val in flag.flag_values()

    def transform[N: dp.Node, P, T](
        tree: dp.Tree[Flag[F] | N, P, T],
    ) -> dp.Tree[N, P, T]:
        if isinstance(tree.node, Flag):
            tree = cast(dp.Tree[Any, P, T], tree)
            node = cast(Flag[Any], tree.node)
            if isinstance(query := node.flag.attached.query, flag):
                node = cast(Flag[F], node)
                answer = dp.Answer(None, val)
                assert answer in query.finite_answer_set()
                tracked = node.flag.attached.parse_answer(answer)
                assert not isinstance(tracked, dp.ParseError)
                return transform(tree.child(tracked))
        tree = cast(dp.Tree[Any, P, T], tree)
        node = cast(Any, tree.node)
        return tree.transform(node, transform)

    return transform

FlagQuery

Bases: Query[T]

Base class for flag queries.

Type parameter T must be of the form Literal[s1, ..., sn] where si are string literals. The first value is considered the default.

Source code in src/delphyne/stdlib/flags.py

class FlagQuery[T: str](Query[T]):
    """
    Base class for flag queries.

    Type parameter `T` must be of the form `Literal[s1, ..., sn]` where
    `si` are string literals. The first value is considered the default.
    """

    @classmethod
    def flag_values(cls) -> Sequence[str]:
        ans = insp.first_parameter_of_base_class(cls)
        assert (args := insp.literal_type_args(ans)) is not None
        assert len(args) > 0
        assert all(isinstance(a, str) for a in args)
        return cast(Sequence[str], args)

    def finite_answer_set(self) -> Sequence[dp.Answer]:
        vals = self.flag_values()
        return [dp.Answer(None, v) for v in vals]

    def default_answer(self) -> dp.Answer:
        return self.finite_answer_set()[0]

    def parse_answer(self, answer: dp.Answer) -> T | dp.ParseError:
        if not isinstance(answer.content, str):
            return dp.ParseError(description="Flag answer must be a string.")
        ans = self.flag_values()
        if answer.content not in ans:
            msg = f"Invalid flag value: '{answer.content}'."
            return dp.ParseError(description=msg)
        return cast(T, answer.content)