References and Traces

References Summary

delphyne.core.refs

References to nodes, values, spaces and space elements.

References are serializable, immutable values that can be used to identify nodes, spaces and values in a tree (possibly deeply nested). References are useful for tooling and for representing serializable traces (Trace). Also, references are attached to success nodes and query answers (Tracked) so as to allow caching and enforce the locality invariant (see Tree).

Local references identify a node, space or space element relative to a given tree node. Global references are expressed relative to a single, global origin.

In addition, three kinds of references can be distinguished:

• Full references: the default kind of references produced by reify. Query answers are stored as strings and elements of spaces induced by strategies are denoted by sequences of value references.
• Id-based references: shorter references, where query answers and success values are identified by unique identifiers. This concise format is used for exporting traces (see Trace).
• Hint-based references: query answers and success values are identified by sequences of hints. This format is used in the demonstration language (e.g. argument of test instruction go compare(['', 'foo bar'])) and when visualizing traces resulting from demonstrations.

Query Answers

Answer dataclass

An answer to a query.

It can serve as a space element reference if the space in question is a query and the proposed answer correctly parses.

Attributes:

Name	Type	Description
mode	AnswerMode	The answer mode (see AnswerMode).
content	str | Structured	The answer content, which can be a raw string or a structured answer (see Structured).
tool_calls	tuple[ToolCall, ...]	An optional sequence of tool calls.
justification	str | None	Additional explanations for the answers, which are not passed to the parser but can be appended at the end of the answer in examples. In particular, this is useful when defining queries for which the oracle is not asked to produce a justification for its answer, but justifications can still be provided in examples for the sake of few-shot prompting.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class Answer:
    """
    An answer to a query.

    It can serve as a _space element reference_ if the space in question
    is a query and the proposed answer correctly parses.

    Attributes:
        mode: The answer mode (see `AnswerMode`).
        content: The answer content, which can be a raw string or a
            structured answer (see `Structured`).
        tool_calls: An optional sequence of tool calls.
        justification: Additional explanations for the answers, which
            are not passed to the parser but can be appended at the end
            of the answer in examples. In particular, this is useful
            when defining queries for which the oracle is not asked to
            produce a justification for its answer, but justifications
            can still be provided in examples for the sake of few-shot
            prompting.
    """

    mode: AnswerMode
    content: str | Structured
    tool_calls: tuple[ToolCall, ...] = ()
    justification: str | None = None

AnswerMode

AnswerMode: str | None

A name for an answer mode, which can be a string or None (the latter is typically used for naming default modes).

Queries are allowed to define multiple answer modes, each mode being possibly associated with different settings and with a different parser. An Answer value features the mode that must be used to parse it.

Structured dataclass

Wrapper for structured LLM answers.

Many LLM APIs allow producing JSON answers (sometimes following a given schema) instead of plain text.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class Structured:
    """
    Wrapper for structured LLM answers.

    Many LLM APIs allow producing JSON answers (sometimes following a
    given schema) instead of plain text.
    """

    structured: Any  # JSON object

    def __hash__(self) -> int:
        # See `ToolCall.__hash__` for explanations.
        import json

        return hash(json.dumps(self.__dict__))

ToolCall dataclass

A tool call, usually produced by an LLM oracle.

Tool calls can be attached to LLM answers (see Answer).

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class ToolCall:
    """
    A tool call, usually produced by an LLM oracle.

    Tool calls can be attached to LLM answers (see `Answer`).
    """

    name: str
    args: Mapping[str, Any]

    def __hash__(self) -> int:
        # Tool calls need to be hashable since they are part of answers
        # and references. However, they can feature arbitrary JSON
        # objects. This, we compute a hash for the serialized value.
        import json

        return hash(json.dumps(self.__dict__))

References

NodeRef

NodeRef: NodePath | NodeId

A node reference is either a path or a node identifier.

Only one of these forms may be allowed depending on the context (e.g. in the id-based references used for exporting traces, only node identifiers are used, while in the full references attached to trees by reify, only paths are used).

NodePath

NodePath: tuple[ValueRef, ...]

Encodes a sequence of actions leading to a node with respect to a given root.

NodeId dataclass

Global identifier of a node within a trace.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class NodeId:
    """
    Global identifier of a node within a trace.
    """

    id: int

ValueRef

ValueRef: Assembly[AtomicValueRef]

A reference to a local value, which is obtained by combining elements of (possibly multiple) local spaces.

Assembly

Assembly: T | None | tuple[Assembly[T], ...]

An S-expression whose atoms have type T.

AtomicValueRef

AtomicValueRef: IndexedRef | SpaceElementRef

An atomic value reference is a space element reference that is indexed zero or a finite number of times: space_elt_ref[i1][i2]...[in].

IndexedRef dataclass

Indexing an atomic value reference.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class IndexedRef:
    """
    Indexing an atomic value reference.
    """

    ref: AtomicValueRef
    index: int

SpaceElementRef dataclass

A reference to an element of a local space.

When the space field is None, the primary field is considered instead (if it exists).

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class SpaceElementRef:
    """
    A reference to an element of a local space.

    When the `space` field is `None`, the primary field is considered
    instead (if it exists).
    """

    space: SpaceRef | None
    element: AnswerRef | NodeRef | HintsRef

SpaceRef dataclass

A reference to a specific local space.

The arg argument should be () for nonparametric spaces and a n-uple for spaces parametric in n arguments. This differs from Orakell where all parametric spaces have one argument.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class SpaceRef:
    """
    A reference to a specific local space.

    The `arg` argument should be `()` for nonparametric spaces and a
    n-uple for spaces parametric in n arguments. This differs from
    Orakell where all parametric spaces have one argument.
    """

    name: SpaceName
    args: tuple[ValueRef, ...]

SpaceName dataclass

A name identifying a parametric space.

This name can feature integer indices. For example, subs[0] denotes the first subgoal of a Join node.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class SpaceName:
    """
    A name identifying a parametric space.

    This name can feature integer indices. For example, `subs[0]`
    denotes the first subgoal of a `Join` node.
    """

    name: str
    indices: tuple[int, ...]

    def __getitem__(self, index: int) -> "SpaceName":
        return SpaceName(self.name, (*self.indices, index))

AnswerRef

AnswerRef: Answer | AnswerId

A reference to a query answer.

AnswerId dataclass

The identifier to an Answer object stored within a trace.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class AnswerId:
    """
    The identifier to an `Answer` object stored within a trace.
    """

    id: int

HintsRef dataclass

References a local space element via a sequence of hints.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class HintsRef:
    """
    References a local space element via a sequence of hints.
    """

    hints: tuple[Hint, ...]

Hint dataclass

A hint for selecting a query answer.

A hint can be associated to a qualifier, which is the name of an imported demonstration defining the hint.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class Hint:
    """A hint for selecting a query answer.

    A hint can be associated to a qualifier, which is the name of an
    imported demonstration defining the hint.
    """

    qualifier: str | None
    hint: HintValue

HintValue

HintValue: str

A string that hints at a query answer.

GlobalNodePath

GlobalNodePath: tuple[tuple[SpaceRef, NodePath], ...]

Path to a node from the global origin, as a sequence of (space to enter, path to follow) instruction pairs.

GlobalSpacePath

GlobalSpacePath: tuple[GlobalNodePath, SpaceRef]

A path to a global node

NodeOrigin

NodeOrigin: ChildOf | NestedTreeOf

Origin of a tree.

A tree is either the child of another tree or the root of a nested tree. Traces can be exported as mappings from node identifiers to node origin information featuring id-based references (see Trace).

ChildOf dataclass

The tree of interest is the child of another one.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class ChildOf:
    """
    The tree of interest is the child of another one.
    """

    node: NodeId
    action: ValueRef

NestedTreeOf dataclass

The tree of interest is the root of a tree that induces a given space.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class NestedTreeOf:
    """
    The tree of interest is the root of a tree that induces a given
    space.
    """

    node: NodeId
    space: SpaceRef

Tracked Values

Tracked dataclass

Bases: Generic[T]

A tracked value, which pairs a value with a reference.

Attributes:

Name	Type	Description
value	T	The value being tracked.
ref	AtomicValueRef	A local reference to the value, relative to the node reference by the node field.
node	GlobalNodePath	A global reference to the node to which the space that the value originates from is attached. In particular, this field is useful to check the locality invariant at runtime (e.g., when passing a tracked value to Tree.child).
type_annot	TypeAnnot[T] | NoTypeInfo	An optional type annotation for the value field. This is mostly used for improving the rendering of values when exporting trace information for external tools.

Tracked sequences (or pairs) can be indexed using __getitem__, resulting in tracked values with IndexedRef references. Since __getitem__ is defined, tracked values are also iterable.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class Tracked(Generic[T]):
    """
    A tracked value, which pairs a value with a reference.

    Attributes:
        value: The value being tracked.
        ref: A local reference to the value, relative to the node
            reference by the `node` field.
        node: A global reference to the node to which the space that the
            value originates from is attached. In particular, this field
            is useful to check the locality invariant at runtime (e.g.,
            when passing a tracked value to `Tree.child`).
        type_annot: An optional type annotation for the `value` field.
            This is mostly used for improving the rendering of values
            when exporting trace information for external tools.

    Tracked sequences (or pairs) can be indexed using `__getitem__`,
    resulting in tracked values with `IndexedRef` references. Since
    `__getitem__` is defined, tracked values are also iterable.
    """

    value: T
    ref: AtomicValueRef
    node: GlobalNodePath
    type_annot: TypeAnnot[T] | NoTypeInfo

    @overload
    def __getitem__[A, B](
        self: "Tracked[tuple[A, B]]", index: Literal[0]
    ) -> "Tracked[A]": ...

    @overload
    def __getitem__[A, B](
        self: "Tracked[tuple[A, B]]", index: Literal[1]
    ) -> "Tracked[B]": ...

    @overload
    def __getitem__[U](
        self: "Tracked[Sequence[U]]", index: int
    ) -> "Tracked[U]": ...

    def __getitem__[U](
        self: "Tracked[Sequence[U] | tuple[Any, ...]]", index: int
    ) -> "Tracked[U | Any]":
        return Tracked(
            self.value[index],
            IndexedRef(self.ref, index),
            self.node,
            insp.element_type_of_sequence_type(self.type_annot, index),
        )

Value

Value: ExtAssembly[Tracked[Any]]

An assembly of local, tracked values.

Values can serve as actions or space parameters.

ExtAssembly

ExtAssembly: T | None | Sequence[ExtAssembly[T]]

Generalizing Assembly to allow arbitrary sequences (and not just tuples). The distinction is important because ValueRef needs to be hashable and so cannot contain lists, while Value can contain lists.

check_local_value

check_local_value(val: Value, node: GlobalNodePath)

Raise a LocalityError exception if a given value is not a local value relative to a given node.

Source code in src/delphyne/core/refs.py

def check_local_value(val: Value, node: GlobalNodePath):
    """
    Raise a `LocalityError` exception if a given value is not a local
    value relative to a given node.
    """
    match val:
        case None:
            pass
        case Sequence():
            for v in val:
                check_local_value(v, node)
        case Tracked():
            if val.node != node:
                raise LocalityError(
                    expected_node_ref=node,
                    node_ref=val.node,
                    local_ref=val.ref,
                )
        case _:
            assert False

LocalityError dataclass

Bases: Exception

Exception raised when the locality invariant is violated.

See Tree and check_local_value.

Source code in src/delphyne/core/refs.py

@dataclass(frozen=True)
class LocalityError(Exception):
    """
    Exception raised when the locality invariant is violated.

    See `Tree` and `check_local_value`.
    """

    expected_node_ref: GlobalNodePath
    node_ref: GlobalNodePath
    local_ref: AtomicValueRef

Traces

Trace

A collection of reachable nodes and spaces, which is encoded in a concise way by introducing unique identifiers for answers and nodes.

Traces are mutable. Methods are provided to convert full references into id-based references, creating fresh identifiers for new nodes and queries on the fly. Backward conversion methods are also provided for converting id-based references back into full references (assuming id-based references are valid, without which these methods fail with assertion errors).

Attributes:

Name	Type	Description
nodes	dict[NodeId, NodeOrigin]	a mapping from node identifiers to their origin.
node_ids	dict[NodeOrigin, NodeId]	reverse map of nodes.
answers	dict[AnswerId, tuple[QueryOrigin, Answer]]	a mapping from answer identifiers to actual answers, along with origin information on the associated query.
answer_ids	dict[QueryOrigin, dict[Answer, AnswerId]]	reverse map of answers.

Source code in src/delphyne/core/traces.py

class Trace:
    """
    A collection of reachable nodes and spaces, which is encoded in a
    concise way by introducing unique identifiers for answers and nodes.

    Traces are mutable. Methods are provided to convert full references
    into id-based references, creating fresh identifiers for new nodes
    and queries on the fly. Backward conversion methods are also
    provided for converting id-based references back into full
    references (assuming id-based references are valid, without which
    these methods fail with assertion errors).

    Attributes:
        nodes: a mapping from node identifiers to their origin.
        node_ids: reverse map of `nodes`.
        answers: a mapping from answer identifiers to actual answers,
            along with origin information on the associated query.
        answer_ids: reverse map of `answers`.
    """

    GLOBAL_ORIGIN_ID = refs.NodeId(0)

    def __init__(self):
        """
        Create an empty trace.
        """
        self.nodes: dict[refs.NodeId, refs.NodeOrigin] = {}
        self.node_ids: dict[refs.NodeOrigin, refs.NodeId] = {}
        self.answers: dict[refs.AnswerId, tuple[QueryOrigin, refs.Answer]] = {}
        self.answer_ids: dict[
            QueryOrigin, dict[refs.Answer, refs.AnswerId]
        ] = {}
        self._last_node_id: int = 0
        self._last_answer_id: int = 0

    def fresh_or_cached_node_id(self, origin: refs.NodeOrigin) -> refs.NodeId:
        """
        Obtain the identifier of a node described by its origin.
        Create a new identifier on the fly if it does not exist yet.
        """
        if origin in self.node_ids:
            return self.node_ids[origin]
        else:
            self._last_node_id += 1
            id = refs.NodeId(self._last_node_id)
            self.nodes[id] = origin
            self.node_ids[origin] = id
            return id

    def fresh_or_cached_answer_id(
        self, answer: refs.Answer, origin: QueryOrigin
    ) -> refs.AnswerId:
        """
        Obtain the identifier of an answer, given its content and the
        origin of the query that it corresponds to. Create a new, fresh
        identifier on the fly if it does not exist yet.
        """
        if origin not in self.answer_ids:
            self.answer_ids[origin] = {}
        if answer in self.answer_ids[origin]:
            return self.answer_ids[origin][answer]
        else:
            self._last_answer_id += 1
            id = refs.AnswerId(self._last_answer_id)
            self.answers[id] = (origin, answer)
            self.answer_ids[origin][answer] = id
            return id

    def register_query(self, origin: QueryOrigin) -> None:
        """
        Ensure that a query appears in the trace, even if not answers
        are associated with it yet.

        This is particularly useful for the demonstration interpreter.
        Indeed, when a test gets stuck on an unanswered query, it is
        desirable for this query to be part of the returned trace so
        that the user can visualize it.
        """
        if origin not in self.answer_ids:
            self.answer_ids[origin] = {}

    def export(self) -> ExportableTrace:
        """
        Export a trace into a lightweight, serializable format.
        """
        nodes = {
            id.id: pprint.node_origin(origin)
            for id, origin in self.nodes.items()
        }
        queries: list[ExportableQueryInfo] = []
        for q, a in self.answer_ids.items():
            ref = pprint.space_ref(q.ref)
            answers = {id.id: value for value, id in a.items()}
            queries.append(ExportableQueryInfo(q.node.id, ref, answers))
        return ExportableTrace(nodes, queries)

    def check_consistency(self) -> None:
        """
        Perform a sanity check on the trace.

        Each node identifier is expanded into a full reference and then
        converted back to an identifier, which must be equal to the
        original one.
        """
        for id in self.nodes:
            expanded = self.expand_node_id(id)
            assert id == self.convert_global_node_path(expanded)

    ### Convert full references into id-based references

    def convert_location(self, location: Location) -> ShortLocation:
        """
        Convert a full location into an id-based one.
        """
        id = self.convert_global_node_path(location.node)
        space = None
        if location.space is not None:
            space = self._convert_space_ref(id, location.space)
        return ShortLocation(id, space)

    def convert_query_origin(self, ref: refs.GlobalSpacePath) -> QueryOrigin:
        """
        Convert a full, global space reference denoting a quey origin
        into an id-based reference.
        """
        id = self.convert_global_node_path(ref[0])
        space = self._convert_space_ref(id, ref[1])
        origin = QueryOrigin(id, space)
        self.register_query(origin)
        return origin

    def convert_answer_ref(
        self, ref: tuple[refs.GlobalSpacePath, refs.Answer]
    ) -> refs.AnswerId:
        """
        Convert a full answer reference into an answer id.
        """
        node_path, space = ref[0]
        id = self.convert_global_node_path(node_path)
        space = self._convert_space_ref(id, space)
        origin = QueryOrigin(id, space)
        return self.fresh_or_cached_answer_id(ref[1], origin)

    def convert_global_node_path(
        self, path: refs.GlobalNodePath
    ) -> refs.NodeId:
        """
        Convert a full, global node reference into an id-based one.
        """
        id = Trace.GLOBAL_ORIGIN_ID
        for space, node_path in path:
            space_ref = self._convert_space_ref(id, space)
            id = self.fresh_or_cached_node_id(refs.NestedTreeOf(id, space_ref))
            id = self._convert_node_path(id, node_path)
        return id

    def convert_global_space_path(
        self, path: refs.GlobalSpacePath
    ) -> refs.SpaceRef:
        """
        Convert a full global space reference into an id-based one.
        """
        node_path, space_ref = path
        id = self.convert_global_node_path(node_path)
        return self._convert_space_ref(id, space_ref)

    def _convert_node_path(
        self, id: refs.NodeId, path: refs.NodePath
    ) -> refs.NodeId:
        """
        Convert a full local node path into an identifier, relative to a
        given node.
        """
        for a in path:
            action_ref = self._convert_value_ref(id, a)
            id = self.fresh_or_cached_node_id(refs.ChildOf(id, action_ref))
        return id

    def _convert_space_ref(
        self, id: refs.NodeId, ref: refs.SpaceRef
    ) -> refs.SpaceRef:
        """
        Convert a full local space reference into an id-based one, relative
        to a given node.
        """
        args = tuple(self._convert_value_ref(id, a) for a in ref.args)
        return refs.SpaceRef(ref.name, args)

    def _convert_atomic_value_ref(
        self, id: refs.NodeId, ref: refs.AtomicValueRef
    ) -> refs.AtomicValueRef:
        """
        Convert a full local atomic value reference into an id-based one,
        relative to a given node.
        """
        if isinstance(ref, refs.IndexedRef):
            return refs.IndexedRef(
                self._convert_atomic_value_ref(id, ref.ref), ref.index
            )
        else:
            return self._convert_space_element_ref(id, ref)

    def _convert_value_ref(
        self, id: refs.NodeId, ref: refs.ValueRef
    ) -> refs.ValueRef:
        """
        Convert a full local value reference into an id-based one,
        relative to a given node.
        """
        if ref is None:
            return None
        elif isinstance(ref, tuple):
            return tuple(self._convert_value_ref(id, a) for a in ref)
        else:
            return self._convert_atomic_value_ref(id, ref)

    def _convert_space_element_ref(
        self, id: refs.NodeId, ref: refs.SpaceElementRef
    ) -> refs.SpaceElementRef:
        """
        Convert a full local space element reference into an id-based one,
        relative to a given node.
        """
        space = None
        if ref.space is not None:
            space = self._convert_space_ref(id, ref.space)
        match ref.element:
            case refs.Answer():
                assert space is not None
                origin = QueryOrigin(id, space)
                element = self.fresh_or_cached_answer_id(ref.element, origin)
            case refs.AnswerId() | refs.NodeId():
                element = ref.element
            case refs.HintsRef():
                assert False
            case tuple():
                assert space is not None
                nested_root_orig = refs.NestedTreeOf(id, space)
                nested_root = self.fresh_or_cached_node_id(nested_root_orig)
                element = self._convert_node_path(nested_root, ref.element)
        return refs.SpaceElementRef(space, element)

    ### Reverse direction: expanding id-based references into full ones.

    def expand_space_ref(
        self, id: refs.NodeId, ref: refs.SpaceRef
    ) -> refs.SpaceRef:
        """
        Convert a local id-based space reference into a full one,
        relative to a given node.
        """
        args = tuple(self.expand_value_ref(id, a) for a in ref.args)
        return refs.SpaceRef(ref.name, args)

    def expand_value_ref(
        self, id: refs.NodeId, ref: refs.ValueRef
    ) -> refs.ValueRef:
        """
        Convert a local id-based value reference into a full one,
        relative to a given node.
        """
        if ref is None:
            return None
        elif isinstance(ref, tuple):
            return tuple(self.expand_value_ref(id, a) for a in ref)
        else:
            return self._expand_atomic_value_ref(id, ref)

    def expand_node_id(self, id: refs.NodeId) -> refs.GlobalNodePath:
        """
        Convert a node identifier into a full, global node reference.
        """
        rev_path: list[tuple[refs.SpaceRef, refs.NodePath]] = []
        while id != Trace.GLOBAL_ORIGIN_ID:
            id, space, path = self._recover_path(id)
            rev_path.append((space, path))
        return tuple(reversed(rev_path))

    def _expand_atomic_value_ref(
        self, id: refs.NodeId, ref: refs.AtomicValueRef
    ) -> refs.AtomicValueRef:
        """
        Convert a local id-based atomic value reference into a full one,
        relative to a given node.
        """
        if isinstance(ref, refs.IndexedRef):
            return refs.IndexedRef(
                self._expand_atomic_value_ref(id, ref.ref), ref.index
            )
        else:
            return self._expand_space_element_ref(id, ref)

    def _expand_space_element_ref(
        self, id: refs.NodeId, ref: refs.SpaceElementRef
    ) -> refs.SpaceElementRef:
        """
        Convert a local id-based space element reference into a full
        one, relative to a given node.
        """
        assert isinstance(ref, refs.SpaceElementRef)
        assert ref.space is not None
        space = self.expand_space_ref(id, ref.space)
        match ref.element:
            case refs.AnswerId():
                _orig, ans = self.answers[ref.element]
                element = ans
            case refs.NodeId():
                orig, _, element = self._recover_path(ref.element)
                assert orig == id
            case _:
                assert False
        return refs.SpaceElementRef(space, element)

    def _recover_path(
        self, dst: refs.NodeId
    ) -> tuple[refs.NodeId, refs.SpaceRef, refs.NodePath]:
        """
        Find the node from which the tree containing `dst` originates.

        Return the node in which the full surrounding tree is nested,
        the associated space reference, and a path to `dst` from the
        root of the surrounding tree.
        """
        rev_path: list[refs.ValueRef] = []
        while True:
            dst_origin = self.nodes[dst]
            match dst_origin:
                case refs.ChildOf(before, action):
                    rev_path.append(self.expand_value_ref(before, action))
                    dst = before
                case refs.NestedTreeOf(orig, space):
                    space = self.expand_space_ref(orig, space)
                    return orig, space, tuple(reversed(rev_path))

__init__

__init__()

Create an empty trace.

Source code in src/delphyne/core/traces.py

def __init__(self):
    """
    Create an empty trace.
    """
    self.nodes: dict[refs.NodeId, refs.NodeOrigin] = {}
    self.node_ids: dict[refs.NodeOrigin, refs.NodeId] = {}
    self.answers: dict[refs.AnswerId, tuple[QueryOrigin, refs.Answer]] = {}
    self.answer_ids: dict[
        QueryOrigin, dict[refs.Answer, refs.AnswerId]
    ] = {}
    self._last_node_id: int = 0
    self._last_answer_id: int = 0

fresh_or_cached_node_id

fresh_or_cached_node_id(origin: NodeOrigin) -> NodeId

Obtain the identifier of a node described by its origin. Create a new identifier on the fly if it does not exist yet.

Source code in src/delphyne/core/traces.py

def fresh_or_cached_node_id(self, origin: refs.NodeOrigin) -> refs.NodeId:
    """
    Obtain the identifier of a node described by its origin.
    Create a new identifier on the fly if it does not exist yet.
    """
    if origin in self.node_ids:
        return self.node_ids[origin]
    else:
        self._last_node_id += 1
        id = refs.NodeId(self._last_node_id)
        self.nodes[id] = origin
        self.node_ids[origin] = id
        return id

fresh_or_cached_answer_id

fresh_or_cached_answer_id(answer: Answer, origin: QueryOrigin) -> AnswerId

Obtain the identifier of an answer, given its content and the origin of the query that it corresponds to. Create a new, fresh identifier on the fly if it does not exist yet.

Source code in src/delphyne/core/traces.py

def fresh_or_cached_answer_id(
    self, answer: refs.Answer, origin: QueryOrigin
) -> refs.AnswerId:
    """
    Obtain the identifier of an answer, given its content and the
    origin of the query that it corresponds to. Create a new, fresh
    identifier on the fly if it does not exist yet.
    """
    if origin not in self.answer_ids:
        self.answer_ids[origin] = {}
    if answer in self.answer_ids[origin]:
        return self.answer_ids[origin][answer]
    else:
        self._last_answer_id += 1
        id = refs.AnswerId(self._last_answer_id)
        self.answers[id] = (origin, answer)
        self.answer_ids[origin][answer] = id
        return id

register_query

register_query(origin: QueryOrigin) -> None

Ensure that a query appears in the trace, even if not answers are associated with it yet.

This is particularly useful for the demonstration interpreter. Indeed, when a test gets stuck on an unanswered query, it is desirable for this query to be part of the returned trace so that the user can visualize it.

Source code in src/delphyne/core/traces.py

def register_query(self, origin: QueryOrigin) -> None:
    """
    Ensure that a query appears in the trace, even if not answers
    are associated with it yet.

    This is particularly useful for the demonstration interpreter.
    Indeed, when a test gets stuck on an unanswered query, it is
    desirable for this query to be part of the returned trace so
    that the user can visualize it.
    """
    if origin not in self.answer_ids:
        self.answer_ids[origin] = {}

export

export() -> ExportableTrace

Export a trace into a lightweight, serializable format.

Source code in src/delphyne/core/traces.py

def export(self) -> ExportableTrace:
    """
    Export a trace into a lightweight, serializable format.
    """
    nodes = {
        id.id: pprint.node_origin(origin)
        for id, origin in self.nodes.items()
    }
    queries: list[ExportableQueryInfo] = []
    for q, a in self.answer_ids.items():
        ref = pprint.space_ref(q.ref)
        answers = {id.id: value for value, id in a.items()}
        queries.append(ExportableQueryInfo(q.node.id, ref, answers))
    return ExportableTrace(nodes, queries)

check_consistency

check_consistency() -> None

Perform a sanity check on the trace.

Each node identifier is expanded into a full reference and then converted back to an identifier, which must be equal to the original one.

Source code in src/delphyne/core/traces.py

def check_consistency(self) -> None:
    """
    Perform a sanity check on the trace.

    Each node identifier is expanded into a full reference and then
    converted back to an identifier, which must be equal to the
    original one.
    """
    for id in self.nodes:
        expanded = self.expand_node_id(id)
        assert id == self.convert_global_node_path(expanded)

convert_location

convert_location(location: Location) -> ShortLocation

Convert a full location into an id-based one.

Source code in src/delphyne/core/traces.py

def convert_location(self, location: Location) -> ShortLocation:
    """
    Convert a full location into an id-based one.
    """
    id = self.convert_global_node_path(location.node)
    space = None
    if location.space is not None:
        space = self._convert_space_ref(id, location.space)
    return ShortLocation(id, space)

convert_query_origin

convert_query_origin(ref: GlobalSpacePath) -> QueryOrigin

Convert a full, global space reference denoting a quey origin into an id-based reference.

Source code in src/delphyne/core/traces.py

def convert_query_origin(self, ref: refs.GlobalSpacePath) -> QueryOrigin:
    """
    Convert a full, global space reference denoting a quey origin
    into an id-based reference.
    """
    id = self.convert_global_node_path(ref[0])
    space = self._convert_space_ref(id, ref[1])
    origin = QueryOrigin(id, space)
    self.register_query(origin)
    return origin

convert_answer_ref

convert_answer_ref(ref: tuple[GlobalSpacePath, Answer]) -> AnswerId

Convert a full answer reference into an answer id.

Source code in src/delphyne/core/traces.py

def convert_answer_ref(
    self, ref: tuple[refs.GlobalSpacePath, refs.Answer]
) -> refs.AnswerId:
    """
    Convert a full answer reference into an answer id.
    """
    node_path, space = ref[0]
    id = self.convert_global_node_path(node_path)
    space = self._convert_space_ref(id, space)
    origin = QueryOrigin(id, space)
    return self.fresh_or_cached_answer_id(ref[1], origin)

convert_global_node_path

convert_global_node_path(path: GlobalNodePath) -> NodeId

Convert a full, global node reference into an id-based one.

Source code in src/delphyne/core/traces.py

def convert_global_node_path(
    self, path: refs.GlobalNodePath
) -> refs.NodeId:
    """
    Convert a full, global node reference into an id-based one.
    """
    id = Trace.GLOBAL_ORIGIN_ID
    for space, node_path in path:
        space_ref = self._convert_space_ref(id, space)
        id = self.fresh_or_cached_node_id(refs.NestedTreeOf(id, space_ref))
        id = self._convert_node_path(id, node_path)
    return id

convert_global_space_path

convert_global_space_path(path: GlobalSpacePath) -> SpaceRef

Convert a full global space reference into an id-based one.

Source code in src/delphyne/core/traces.py

def convert_global_space_path(
    self, path: refs.GlobalSpacePath
) -> refs.SpaceRef:
    """
    Convert a full global space reference into an id-based one.
    """
    node_path, space_ref = path
    id = self.convert_global_node_path(node_path)
    return self._convert_space_ref(id, space_ref)

expand_space_ref

expand_space_ref(id: NodeId, ref: SpaceRef) -> SpaceRef

Convert a local id-based space reference into a full one, relative to a given node.

Source code in src/delphyne/core/traces.py

def expand_space_ref(
    self, id: refs.NodeId, ref: refs.SpaceRef
) -> refs.SpaceRef:
    """
    Convert a local id-based space reference into a full one,
    relative to a given node.
    """
    args = tuple(self.expand_value_ref(id, a) for a in ref.args)
    return refs.SpaceRef(ref.name, args)

expand_value_ref

expand_value_ref(id: NodeId, ref: ValueRef) -> ValueRef

Convert a local id-based value reference into a full one, relative to a given node.

Source code in src/delphyne/core/traces.py

def expand_value_ref(
    self, id: refs.NodeId, ref: refs.ValueRef
) -> refs.ValueRef:
    """
    Convert a local id-based value reference into a full one,
    relative to a given node.
    """
    if ref is None:
        return None
    elif isinstance(ref, tuple):
        return tuple(self.expand_value_ref(id, a) for a in ref)
    else:
        return self._expand_atomic_value_ref(id, ref)

expand_node_id

expand_node_id(id: NodeId) -> GlobalNodePath

Convert a node identifier into a full, global node reference.

Source code in src/delphyne/core/traces.py

def expand_node_id(self, id: refs.NodeId) -> refs.GlobalNodePath:
    """
    Convert a node identifier into a full, global node reference.
    """
    rev_path: list[tuple[refs.SpaceRef, refs.NodePath]] = []
    while id != Trace.GLOBAL_ORIGIN_ID:
        id, space, path = self._recover_path(id)
        rev_path.append((space, path))
    return tuple(reversed(rev_path))

QueryOrigin dataclass

A global, id-based reference to the space induced by a query.

Source code in src/delphyne/core/traces.py

@dataclass(frozen=True)
class QueryOrigin:
    """
    A global, id-based reference to the space induced by a query.
    """

    node: refs.NodeId
    ref: refs.SpaceRef

ExportableTrace dataclass

A lightweight trace format that can be easily exported to JSON/YAML.

Attributes:

Name	Type	Description
nodes	dict[int, NodeOriginStr]	a mapping from node ids to serialized origin information.
queries	list[ExportableQueryInfo]	a list of encountered queries with associated answers.

Source code in src/delphyne/core/traces.py

@dataclass
class ExportableTrace:
    """
    A lightweight trace format that can be easily exported to JSON/YAML.

    Attributes:
        nodes: a mapping from node ids to serialized origin information.
        queries: a list of encountered queries with associated answers.
    """

    nodes: dict[int, NodeOriginStr]
    queries: list[ExportableQueryInfo]

ExportableQueryInfo dataclass

Information about a query encountered in an exportable trace.

Attributes:

Name	Type	Description
node	int	Identifier of the node that the query is attached to.
space	str	Local, id-based reference of the space that the query belongs to. Serialized using pprint.space_ref.
answers	dict[int, Answer]	Mapping from answer identifiers to actual answers. Answer identifiers are unique across a whole exportable trace (and not only across an ExportableQueryInfo value).

Source code in src/delphyne/core/traces.py

@dataclass
class ExportableQueryInfo:
    """
    Information about a query encountered in an exportable trace.

    Attributes:
        node: Identifier of the node that the query is attached to.
        space: Local, id-based reference of the space that the query
            belongs to. Serialized using `pprint.space_ref`.
        answers: Mapping from answer identifiers to actual answers.
            Answer identifiers are unique across a whole exportable
            trace (and not only across an `ExportableQueryInfo` value).
    """

    node: int
    space: str
    answers: dict[int, refs.Answer]

NodeOriginStr

NodeOriginStr: str

A concise, serialized representation for NodeOrigin.

Location dataclass

A full, global reference to either a node or a space.

This is useful in particular for attaching location information to logging messages.

Source code in src/delphyne/core/traces.py

@dataclass(frozen=True)
class Location:
    """
    A **full**, global reference to either a node or a space.

    This is useful in particular for attaching location information to
    logging messages.
    """

    node: refs.GlobalNodePath
    space: refs.SpaceRef | None

ShortLocation dataclass

An id-based, global reference to either a node or a space.

This is the id-based counterpart of Location. Policies typically log messages with Location values attached (since trees feature full references), which are then converted into ShortLocation in the final exportable log.

Source code in src/delphyne/core/traces.py

@dataclass(frozen=True)
class ShortLocation:
    """
    An **id-based**, global reference to either a node or a space.

    This is the id-based counterpart of `Location`. Policies typically
    log messages with `Location` values attached (since trees feature
    full references), which are then converted into `ShortLocation` in
    the final exportable log.
    """

    node: refs.NodeId
    space: refs.SpaceRef | None

Tracers

Tracer

A mutable trace along with a mutable list of log messages.

Both components are protected by a lock to ensure thread-safety (some policies spawn multiple concurrent threads).

Attributes:

Name	Type	Description
trace		A mutable trace.
messages	list[LogMessage]	A mutable list of log messages.
lock		A reentrant lock protecting access to the trace and log. The lock is publicly exposed so that threads can log several successive messages without other threads interleaving new messages in between (TODO: there are cleaner ways to achieve this).

Source code in src/delphyne/core/traces.py

class Tracer:
    """
    A mutable trace along with a mutable list of log messages.

    Both components are protected by a lock to ensure thread-safety
    (some policies spawn multiple concurrent threads).

    Attributes:
        trace: A mutable trace.
        messages: A mutable list of log messages.
        lock: A reentrant lock protecting access to the trace and log.
            The lock is publicly exposed so that threads can log several
            successive messages without other threads interleaving new
            messages in between (TODO: there are cleaner ways to achieve
            this).
    """

    def __init__(self):
        self.trace = Trace()
        self.messages: list[LogMessage] = []

        # Different threads may be logging information or appending to
        # the trace in parallel.
        self.lock = threading.RLock()

    def trace_node(self, node: refs.GlobalNodePath) -> None:
        """
        Ensure that a node at a given reference is present in the trace.

        See `tracer_hook` for registering a hook that automatically
        calls this method on all encountered nodes.
        """
        with self.lock:
            self.trace.convert_location(Location(node, None))

    def trace_query(self, ref: refs.GlobalSpacePath) -> None:
        """
        Ensure that a query at a given reference is present in the
        trace, even if no answer is provided for it.
        """
        with self.lock:
            self.trace.convert_query_origin(ref)

    def trace_answer(
        self, space: refs.GlobalSpacePath, answer: refs.Answer
    ) -> None:
        """
        Ensure that a given query answer is present in the trace, even
        it is is not used to reach a node.
        """
        with self.lock:
            self.trace.convert_answer_ref((space, answer))

    def log(
        self,
        message: str,
        metadata: dict[str, Any] | None = None,
        location: Location | None = None,
    ):
        """
        Log a message, with optional metadata and location information.
        The metadata must be a dictionary of JSON values.
        """
        with self.lock:
            short_location = None
            if location is not None:
                short_location = self.trace.convert_location(location)
            self.messages.append(LogMessage(message, metadata, short_location))

    def export_log(self) -> Iterable[ExportableLogMessage]:
        """
        Export the log into an easily serializable format.
        """
        with self.lock:
            for m in self.messages:
                node = None
                space = None
                if (loc := m.location) is not None:
                    node = loc.node.id
                    if loc.space is not None:
                        space = pprint.space_ref(loc.space)
                yield ExportableLogMessage(m.message, node, space, m.metadata)

    def export_trace(self) -> ExportableTrace:
        """
        Export the trace into an easily serializable format.
        """
        with self.lock:
            return self.trace.export()

trace_node

trace_node(node: GlobalNodePath) -> None

Ensure that a node at a given reference is present in the trace.

See tracer_hook for registering a hook that automatically calls this method on all encountered nodes.

Source code in src/delphyne/core/traces.py

def trace_node(self, node: refs.GlobalNodePath) -> None:
    """
    Ensure that a node at a given reference is present in the trace.

    See `tracer_hook` for registering a hook that automatically
    calls this method on all encountered nodes.
    """
    with self.lock:
        self.trace.convert_location(Location(node, None))

trace_query

trace_query(ref: GlobalSpacePath) -> None

Ensure that a query at a given reference is present in the trace, even if no answer is provided for it.

Source code in src/delphyne/core/traces.py

def trace_query(self, ref: refs.GlobalSpacePath) -> None:
    """
    Ensure that a query at a given reference is present in the
    trace, even if no answer is provided for it.
    """
    with self.lock:
        self.trace.convert_query_origin(ref)

trace_answer

trace_answer(space: GlobalSpacePath, answer: Answer) -> None

Ensure that a given query answer is present in the trace, even it is is not used to reach a node.

Source code in src/delphyne/core/traces.py

def trace_answer(
    self, space: refs.GlobalSpacePath, answer: refs.Answer
) -> None:
    """
    Ensure that a given query answer is present in the trace, even
    it is is not used to reach a node.
    """
    with self.lock:
        self.trace.convert_answer_ref((space, answer))

log

log(
    message: str,
    metadata: dict[str, Any] | None = None,
    location: Location | None = None,
)

Log a message, with optional metadata and location information. The metadata must be a dictionary of JSON values.

Source code in src/delphyne/core/traces.py

def log(
    self,
    message: str,
    metadata: dict[str, Any] | None = None,
    location: Location | None = None,
):
    """
    Log a message, with optional metadata and location information.
    The metadata must be a dictionary of JSON values.
    """
    with self.lock:
        short_location = None
        if location is not None:
            short_location = self.trace.convert_location(location)
        self.messages.append(LogMessage(message, metadata, short_location))

export_log

export_log() -> Iterable[ExportableLogMessage]

Export the log into an easily serializable format.

Source code in src/delphyne/core/traces.py

def export_log(self) -> Iterable[ExportableLogMessage]:
    """
    Export the log into an easily serializable format.
    """
    with self.lock:
        for m in self.messages:
            node = None
            space = None
            if (loc := m.location) is not None:
                node = loc.node.id
                if loc.space is not None:
                    space = pprint.space_ref(loc.space)
            yield ExportableLogMessage(m.message, node, space, m.metadata)

export_trace

export_trace() -> ExportableTrace

Export the trace into an easily serializable format.

Source code in src/delphyne/core/traces.py

def export_trace(self) -> ExportableTrace:
    """
    Export the trace into an easily serializable format.
    """
    with self.lock:
        return self.trace.export()

LogMessage dataclass

A log message.

Attributes:

Name	Type	Description
message	str	The message to log.
metadata	dict[str, Any] | None	Optional metadata associated with the message, as a dictionary mapping string keys to JSON values.
location	ShortLocation | None	An optional location in the strategy tree where the message was logged, if applicable.

Source code in src/delphyne/core/traces.py

@dataclass(frozen=True)
class LogMessage:
    """
    A log message.

    Attributes:
        message: The message to log.
        metadata: Optional metadata associated with the message, as a
            dictionary mapping string keys to JSON values.
        location: An optional location in the strategy tree where the
            message was logged, if applicable.
    """

    message: str
    metadata: dict[str, Any] | None = None
    location: ShortLocation | None = None

ExportableLogMessage dataclass

An exportable log message, as a dataclass whose fields are JSON values (as opposed to LogMessage) and is thus easier to export.

Source code in src/delphyne/core/traces.py

@dataclass(frozen=True)
class ExportableLogMessage:
    """
    An exportable log message, as a dataclass whose fields are JSON
    values (as opposed to `LogMessage`) and is thus easier to export.
    """

    message: str
    node: int | None
    space: str | None
    metadata: dict[str, Any] | None = None

tracer_hook

tracer_hook(tracer: Tracer) -> Callable[[Tree[Any, Any, Any]], None]

Standard hook to be passed to TreeMonitor to automatically log visited nodes into a trace.

Source code in src/delphyne/core/traces.py

def tracer_hook(tracer: Tracer) -> Callable[[Tree[Any, Any, Any]], None]:
    """
    Standard hook to be passed to `TreeMonitor` to automatically log
    visited nodes into a trace.
    """
    return lambda tree: tracer.trace_node(tree.ref)