Skip to content

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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@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 _hashable_repr(self) -> str:
        # See comment in ToolCall._hashable_repr
        import json

        return json.dumps(self.__dict__, sort_keys=True)

    def __hash__(self) -> int:
        return hash(self._hashable_repr())

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Structured):
            return NotImplemented
        return self._hashable_repr() == other._hashable_repr()

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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@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 _hashable_repr(self) -> str:
        # Tool calls need to be hashable since they are part of answers
        # and references. However, they can feature arbitrary JSON
        # objects.
        import json

        return json.dumps(self.__dict__, sort_keys=True)

    def __hash__(self) -> int:
        return hash(self._hashable_repr())

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, ToolCall):
            return NotImplemented
        return self._hashable_repr() == other._hashable_repr()

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
202
203
204
205
206
207
208
@dataclass(frozen=True)
class NodeId:
    """
    Global identifier of a node within a trace.
    """

    id: int

ValueRef

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
164
165
166
167
168
169
170
171
@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
290
291
292
293
294
295
296
297
298
299
300
@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
222
223
224
225
226
227
228
229
230
231
232
233
@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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@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
248
249
250
251
252
253
254
@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
281
282
283
284
285
286
287
@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
269
270
271
272
273
274
275
276
277
278
@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
330
331
332
333
334
335
336
337
@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
340
341
342
343
344
345
346
347
348
@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
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
@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
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
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
420
421
422
423
424
425
426
427
428
429
430
@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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
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
130
131
132
133
134
135
136
137
138
139
140
141
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
143
144
145
146
147
148
149
150
151
152
153
154
155
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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
176
177
178
179
180
181
182
183
184
185
186
187
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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
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
204
205
206
207
208
209
210
211
212
213
214
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
218
219
220
221
222
223
224
225
226
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
228
229
230
231
232
233
234
235
236
237
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
239
240
241
242
243
244
245
246
247
248
249
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
251
252
253
254
255
256
257
258
259
260
261
262
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
264
265
266
267
268
269
270
271
272
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
352
353
354
355
356
357
358
359
360
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
362
363
364
365
366
367
368
369
370
371
372
373
374
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
376
377
378
379
380
381
382
383
384
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
 98
 99
100
101
102
103
104
105
@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
79
80
81
82
83
84
85
86
87
88
89
90
@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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@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
21
22
23
24
25
26
27
28
29
30
31
@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
34
35
36
37
38
39
40
41
42
43
44
45
46
@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.

Source code in src/delphyne/core/traces.py
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
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 good message order beyong
    # exposing the lock.

    def __init__(self, log_level: LogLevel = "info"):
        """
        Parameters:
            log_level: The minimum severity level of messages to log.
        """
        self.trace = Trace()
        self.messages: list[LogMessage] = []
        self.log_level: LogLevel = log_level

        # 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,
        level: LogLevel,
        message: str,
        metadata: object | None = None,
        location: Location | None = None,
    ):
        """
        Log a message, with optional metadata and location information.
        The metadata must be exportable to JSON using Pydantic.
        """
        if not log_level_greater_or_equal(level, self.log_level):
            return
        time = datetime.now()
        with self.lock:
            short_location = None
            if location is not None:
                short_location = self.trace.convert_location(location)
            self.messages.append(
                LogMessage(
                    message=message,
                    level=level,
                    time=time,
                    metadata=metadata,
                    location=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(
                    message=m.message,
                    level=m.level,
                    time=m.time,
                    node=node,
                    space=space,
                    metadata=pydantic_dump(object, m.metadata),
                )

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

__init__

__init__(log_level: LogLevel = 'info')

Parameters:

Name Type Description Default
log_level LogLevel

The minimum severity level of messages to log.

'info'
Source code in src/delphyne/core/traces.py
556
557
558
559
560
561
562
563
564
565
566
567
def __init__(self, log_level: LogLevel = "info"):
    """
    Parameters:
        log_level: The minimum severity level of messages to log.
    """
    self.trace = Trace()
    self.messages: list[LogMessage] = []
    self.log_level: LogLevel = log_level

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

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
569
570
571
572
573
574
575
576
577
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
579
580
581
582
583
584
585
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
587
588
589
590
591
592
593
594
595
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(
    level: LogLevel,
    message: str,
    metadata: object | None = None,
    location: Location | None = None,
)

Log a message, with optional metadata and location information. The metadata must be exportable to JSON using Pydantic.

Source code in src/delphyne/core/traces.py
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
def log(
    self,
    level: LogLevel,
    message: str,
    metadata: object | None = None,
    location: Location | None = None,
):
    """
    Log a message, with optional metadata and location information.
    The metadata must be exportable to JSON using Pydantic.
    """
    if not log_level_greater_or_equal(level, self.log_level):
        return
    time = datetime.now()
    with self.lock:
        short_location = None
        if location is not None:
            short_location = self.trace.convert_location(location)
        self.messages.append(
            LogMessage(
                message=message,
                level=level,
                time=time,
                metadata=metadata,
                location=short_location,
            )
        )

export_log

export_log() -> Iterable[ExportableLogMessage]

Export the log into an easily serializable format.

Source code in src/delphyne/core/traces.py
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
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(
                message=m.message,
                level=m.level,
                time=m.time,
                node=node,
                space=space,
                metadata=pydantic_dump(object, m.metadata),
            )

export_trace

export_trace() -> ExportableTrace

Export the trace into an easily serializable format.

Source code in src/delphyne/core/traces.py
646
647
648
649
650
651
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.

time datetime

Time at which the message was produced

metadata object | None

Optional metadata associated with the message, as an object that can be serialized to JSON using Pydantic.

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
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
@dataclass(frozen=True, kw_only=True)
class LogMessage:
    """
    A log message.

    Attributes:
        message: The message to log.
        time: Time at which the message was produced
        metadata: Optional metadata associated with the message, as an
            object that can be serialized to JSON using Pydantic.
        location: An optional location in the strategy tree where the
            message was logged, if applicable.
    """

    message: str
    level: LogLevel
    time: datetime
    metadata: object | 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
522
523
524
525
526
527
528
529
530
531
532
533
534
@dataclass(frozen=True, kw_only=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
    level: LogLevel
    time: datetime
    node: int | None = None
    space: str | None = None
    metadata: object | None = None  # JSON value

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
654
655
656
657
658
659
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)