mixle.telemetry package

Local-first telemetry for model, placement, routing, and reasoning decisions.

Telemetry records typed decision events with features, choices, outcomes, tags, and timestamps. These records support dashboards, auditing, and learned orchestration policies.

Events are append-only, JSONL-compatible, and designed to avoid raw user content. A process-global recorder makes record(...) convenient, while tests and isolated workflows can use their own Telemetry instance.

dashboard(telemetry)[source]

Fold the telemetry stream into a receipt summary (see module docstring).

Parameters:

telemetry (Any)

Return type:

dict[str, Any]

render_dashboard(telemetry)[source]

The dashboard as plain markdown – printable in a terminal, embeddable in a report.

Parameters:

telemetry (Any)

Return type:

str

class Event(kind, features=<factory>, choice=None, outcome=<factory>, tags=<factory>, ts=0.0)[source]

Bases: object

One typed, timestamped decision record. Features/choice describe the decision; outcome scores it.

Parameters:
kind: str
features: dict[str, Any]
choice: Any = None
outcome: dict[str, Any]
tags: dict[str, str]
ts: float = 0.0
as_row()[source]
Return type:

dict[str, Any]

class Telemetry(path=None, *, flush_every=1)[source]

Bases: object

A local-first event recorder: buffer in memory, append to a JSONL log, read back for training.

record(kind, features=..., choice=..., outcome=..., tags=...) appends one Event. events(kind=...) yields the buffer (optionally filtered). training_rows(kind) yields the (features, choice, outcome) triples the learned-orchestration models consume.

Parameters:

  • path (str | None)

  • flush_every (int)

record(kind, *, features=None, choice=None, outcome=None, tags=None, when=None)[source]

Record one decision event; returns it (mutate .outcome later to close the loop).

Parameters:
Return type:

Event

events(*, kind=None)[source]
Parameters:

kind (str | None)

Return type:

Iterator[Event]

training_rows(kind)[source]

The (features, choice, outcome) triples for a decision kind.

Parameters:

kind (str)

Return type:

list[tuple[dict[str, Any], Any, dict[str, Any]]]

flush()[source]
Return type:

None

record(kind, **kw)[source]

Record an event on the process-global recorder (see Telemetry.record()).

Parameters:
Return type:

Event

get_default_recorder()[source]

The process-global recorder (a no-path in-memory buffer until one is configured).

Return type:

Telemetry

set_default_recorder(recorder)[source]

Install (or clear) the process-global recorder – e.g. point it at the user’s telemetry log.

Parameters:

recorder (Telemetry | None)

Return type:

None

Submodules