Skip to content

symfonic.capabilities.memory.telemetry

telemetry

Privacy-preserving telemetry for the native episodic lifecycle.

FrameworkConfig.episodic_telemetry_sink belonged to the retired engine path. A kernel-native deployment composes memory with memory_capabilities and publishes records from memory.lifecycle. This module is the public collaborator for observing that publication.

The event is deliberately an operational receipt, not a memory export. It contains scope and correlation identifiers plus committed record identifiers; it never contains record text, prompts, responses, or metadata. Consumers which need memory content must use an authorised memory-read/admin path rather than treating telemetry as a bypass around the scope contract.

Emission is best effort. A sink exception is logged without event payload and does not change the lifecycle result: a successful publication remains successful. No sink is the default and creates no event or awaitable work.

EpisodicLifecycleEvent dataclass 

EpisodicLifecycleEvent(occurred_at: str, scope_path: str, committed_record_ids: tuple[str, ...], run_id: str, root_run_id: str, parent_run_id: str | None = None)

A committed episodic lifecycle receipt safe to send to a sink.

run_id, root_run_id and parent_run_id identify the invocation which ran the lifecycle stage. They intentionally describe the lifecycle invocation, not authorship of every committed record: a flush may publish pending work from an earlier interrupted invocation. scope_path is the exact scope supplied to flush; callers must scope any downstream access independently.

EpisodicTelemetrySink

Bases: Protocol

Receives successful native episodic lifecycle publications.

emit async 

emit(event: EpisodicLifecycleEvent) -> None

Observe one successful lifecycle publication.

Source code in src/symfonic/capabilities/memory/telemetry.py 
57
58
59
async def emit(self, event: EpisodicLifecycleEvent) -> None:
    """Observe one successful lifecycle publication."""
    ...

lifecycle_event 

lifecycle_event(*, scope: MemoryScope, committed_record_ids: tuple[str, ...], request: object) -> EpisodicLifecycleEvent

Build the safe lifecycle receipt from the public stage vocabulary.

Source code in src/symfonic/capabilities/memory/telemetry.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def lifecycle_event(
    *,
    scope: MemoryScope,
    committed_record_ids: tuple[str, ...],
    request: object,
) -> EpisodicLifecycleEvent:
    """Build the safe lifecycle receipt from the public stage vocabulary."""
    run_id = str(getattr(request, "run_id", "") or "")
    root_run_id = str(getattr(request, "root_run_id", "") or run_id)
    parent_run_id = str(getattr(request, "parent_run_id", "") or "") or None
    return EpisodicLifecycleEvent(
        occurred_at=datetime.now(UTC).isoformat(),
        scope_path=scope.path,
        committed_record_ids=committed_record_ids,
        run_id=run_id,
        root_run_id=root_run_id,
        parent_run_id=parent_run_id,
    )