Skip to content

crp.tel

Auto-generated reference for the crp.tel subpackage.

tel

crp.tel

Transparency Emission Layer (TEL) - stream CRP governance as AG-UI events (CRP-SPEC-056).

The crp.tel package maps CRP's internal governance evidence onto the AG-UI event vocabulary so any standard frontend can render a CRP agent, while adding namespaced governance events (crp.safety_scan, crp.quality, crp.provenance, ...) that carry the protocol's differentiators.

CRPEmitter

Adapter: CRP governance hooks -> AG-UI-compatible event stream.

Emitter

Ergonomic facade agents call. Knows nothing about transport.

SessionBus

Isolated event bus for one session.

Buffers the most recent buffer_size events so Last-Event-ID replay is possible, and fans out live events to both synchronous and asynchronous subscribers.

emit(ev)

Assign a sequence number, buffer, and fan out ev.

replay_after(seq)

Return all buffered events with seq greater than seq.

subscribe()

Block and yield live events until :meth:close is called.

The returned iterator registers its queue immediately, so a producer thread started right after this call will not race the registration. It also exposes :meth:_SyncSubscriber.poll for timeout-bounded reads (used for SSE heartbeats).

asubscribe()

Return an async iterator yielding live events until the bus is closed.

close()

Close the bus and unblock all subscribers.

Event dataclass

A single typed event on the transparency stream.

Each event carries a monotonic seq (ordering + replay cursor), a stable id, a wall-clock ts, and a payload whose schema is determined by type.

to_dict()

Render as a JSON-serialisable dictionary.

to_sse()

Render to the Server-Sent Events wire format.

The id: field enables Last-Event-ID replay; the event: field lets EventSource route by type.

EventType

Bases: str, Enum

AG-UI event vocabulary with CRP governance extensions carried in CUSTOM.

map_agent_event(event)

Convert an internal :class:AgentEvent into AG-UI events.

Returns a list because one internal transition may surface as multiple display events (e.g. tool selection + reasoning).

custom(name, value)

Namespaced governance event (e.g. crp.safety_scan).

build_report(buffer, tier='casual')

Build a human-readable transparency report from an event buffer.

The same event stream drives all tiers; tier is a rendering filter, not a different backend. casual returns narrative + governance badges; power adds tools and sources; developer adds the raw event log; auditor adds the full provenance chain.

stream_events_sync(bus, last_event_id=None, *, heartbeat_interval=None)

Yield events from bus, replaying missed events when last_event_id is given.

This synchronous generator is useful for tests, CLIs, and adapters that run in a worker thread. It stops when the bus is closed.

stream_frames_sync(bus, last_event_id=None, *, heartbeat_interval=15.0)

Yield SSE wire frames (strings) with replay and heartbeats.

Replays buffered events with seq > last_event_id first (the Last-Event-ID resume contract), then streams live events. When no event arrives within heartbeat_interval seconds a : ping comment frame is emitted so proxies and browsers keep the connection open. Stops when the bus is closed. Pass heartbeat_interval=None to disable heartbeats.

to_sse(ev)

Render an :class:Event to the SSE wire format.

tel.adapter

crp.tel.adapter

Map CRP runtime hooks to AG-UI + governance events (CRP-SPEC-056 §8.3.1).

CRPEmitter

Adapter: CRP governance hooks -> AG-UI-compatible event stream.

map_agent_event(event)

Convert an internal :class:AgentEvent into AG-UI events.

Returns a list because one internal transition may surface as multiple display events (e.g. tool selection + reasoning).

tel.emitter

crp.tel.emitter

Per-session event bus: sequence, replay buffer, and fan-out (CRP-SPEC-056 §8.3.5).

SessionBus

Isolated event bus for one session.

Buffers the most recent buffer_size events so Last-Event-ID replay is possible, and fans out live events to both synchronous and asynchronous subscribers.

emit(ev)

Assign a sequence number, buffer, and fan out ev.

replay_after(seq)

Return all buffered events with seq greater than seq.

subscribe()

Block and yield live events until :meth:close is called.

The returned iterator registers its queue immediately, so a producer thread started right after this call will not race the registration. It also exposes :meth:_SyncSubscriber.poll for timeout-bounded reads (used for SSE heartbeats).

asubscribe()

Return an async iterator yielding live events until the bus is closed.

close()

Close the bus and unblock all subscribers.

Emitter

Ergonomic facade agents call. Knows nothing about transport.

get_bus(session_id, buffer_size=2000)

Return the existing bus for session_id or create one.

tel.events

crp.tel.events

AG-UI-compatible event model + CRP governance extensions (CRP-SPEC-056 §8.3.1).

EventType

Bases: str, Enum

AG-UI event vocabulary with CRP governance extensions carried in CUSTOM.

Event dataclass

A single typed event on the transparency stream.

Each event carries a monotonic seq (ordering + replay cursor), a stable id, a wall-clock ts, and a payload whose schema is determined by type.

to_dict()

Render as a JSON-serialisable dictionary.

to_sse()

Render to the Server-Sent Events wire format.

The id: field enables Last-Event-ID replay; the event: field lets EventSource route by type.

custom(name, value)

Namespaced governance event (e.g. crp.safety_scan).

tel.faithful

crp.tel.faithful

Faithful-narration contract - check a claim against the event trace (CRP-SPEC-056 §8.3.2).

Narration claims are gated through the NLI entailment stack from crp.provenance.entailment_verifier (cross-encoder, lazy-loaded): a claim is only shown when the flattened trajectory/tool-result evidence entails it. When the NLI model is unavailable or inference fails, the module falls back to the deterministic word-subset heuristic so the contract still holds on every profile.

reset_nli_cache()

Reset the NLI availability flag (for testing).

entailment_check(claim, evidence)

Return True when the evidence entails/supports claim.

Prefers the NLI cross-encoder from crp.provenance (premise = flattened trace, hypothesis = claim). Falls back to the deterministic word-subset heuristic when the model is unavailable or inference fails. Unsupported claims must be withheld from the user stream.

filter_faithful(claims, evidence)

Return only claims supported by the evidence.

tel.narrative

crp.tel.narrative

Turn a CRP/AG-UI event stream into a human-readable chain-of-thought narrative.

The :class:NarrativeBuilder consumes the same events a frontend would receive and produces a structured story: what the agent intended, which operations it chose, which tools it called, what safety and verification checks ran, and how the provenance chain links each step. The result can be rendered as Markdown, HTML, or JSON for consoles, audit UIs, and CLI output.

NarrativeStep dataclass

One human-readable step in the agent's reasoning story.

A single link in the HMAC chain rendered for display.

NarrativeBuilder

Accumulate AG-UI + CRP events and build a readable chain-of-thought.

Example::

builder = NarrativeBuilder()
for event in agent.run_tel("What's the weather in Sydney?"):
    builder.ingest(event)
story = builder.to_markdown()

ingest(event)

Add one event to the narrative.

ingest_many(events)

Add multiple events in order.

to_dict()

Return the full narrative as a JSON-serialisable dict.

to_markdown()

Render the narrative as Markdown.

to_html()

Render the narrative as a safe HTML fragment for consoles.

tel.report

crp.tel.report

Localized transparency reports from an event buffer (CRP-SPEC-056 §8.3).

build_report(buffer, tier='casual')

Build a human-readable transparency report from an event buffer.

The same event stream drives all tiers; tier is a rendering filter, not a different backend. casual returns narrative + governance badges; power adds tools and sources; developer adds the raw event log; auditor adds the full provenance chain.

tel.sse

crp.tel.sse

Server-Sent Events wire utilities and stream helpers (CRP-SPEC-056 §8.3).

to_sse(ev)

Render an :class:Event to the SSE wire format.

parse_sse(raw)

Parse a raw SSE payload into event dictionaries.

This is a test helper and a reference parser; production clients typically use the browser's EventSource.

stream_events_sync(bus, last_event_id=None, *, heartbeat_interval=None)

Yield events from bus, replaying missed events when last_event_id is given.

This synchronous generator is useful for tests, CLIs, and adapters that run in a worker thread. It stops when the bus is closed.

stream_events(bus, last_event_id=None, *, heartbeat_interval=None) async

Async variant of :func:stream_events_sync.

stream_frames_sync(bus, last_event_id=None, *, heartbeat_interval=15.0)

Yield SSE wire frames (strings) with replay and heartbeats.

Replays buffered events with seq > last_event_id first (the Last-Event-ID resume contract), then streams live events. When no event arrives within heartbeat_interval seconds a : ping comment frame is emitted so proxies and browsers keep the connection open. Stops when the bus is closed. Pass heartbeat_interval=None to disable heartbeats.

stream_frames(bus, last_event_id=None, *, heartbeat_interval=15.0) async

Async variant of :func:stream_frames_sync.