Skip to content

crp.observability

Auto-generated reference for the crp.observability subpackage.

observability

crp.observability

Observability - events, metrics, audit log, quality reporting, telemetry.

AuditLog

Immutable append-only event log with query support.

Usage::

audit = AuditLog()
emitter.on_all(audit.record)   # auto-capture every event
...
timeline = audit.query(session_id="abc")
fact_events = audit.query(event_type="fact.created")

count property

Total number of recorded events.

record(event)

Append an event (thread-safe, never raises).

query(*, event_type=None, session_id=None, since=None, until=None)

Return events matching all supplied filters.

Parameters:

Name Type Description Default
event_type str | None

Exact event type string (e.g. "dispatch.completed").

 None
session_id str | None

Match events whose data["session_id"] equals this.

 None
since float | None

Only events with timestamp >= since.

 None
until float | None

Only events with timestamp <= until.

 None

reconstruct_session(session_id)

Build a summary dict for session_id from the event stream.

Returns a dict with keys
  • session_id
  • events - full ordered list of events (as dicts).
  • dispatch_count - number of dispatch.completed events.
  • facts_created - count of fact.created events.
  • facts_superseded - count of fact.superseded events.
  • duration_s - seconds between first and last event.

clear()

Remove all stored events (useful in testing).

CRPEvent dataclass

Single protocol event.

to_dict()

Return the event as a plain dict.

Returns:

Type Description
dict[str, Any]

Dict with event_type, timestamp, and data fields.

EventEmitter

Simple synchronous event bus.

Usage::

emitter = EventEmitter()
emitter.on("dispatch.started", my_callback)
emitter.emit("dispatch.started", {"task": "..."})

is_running property

Return whether this object is running.

listener_count property

Return the current listener count.

start()

Mark the emitter as active (startup step 6).

stop()

Stop accepting and delivering events.

on(event_type, listener)

Subscribe listener to event_type.

Deduplicates listeners and enforces a per-event cap (§audit4 REL-M2).

on_all(listener)

Subscribe listener to every event type.

off(event_type, listener)

Remove a specific listener. Returns True if removed.

emit(event_type, data=None)

Emit an event to all matching listeners.

If the emitter has not been started, events are silently dropped. Listener exceptions are logged but never propagate.

ExportFormat

Bases: Enum

Supported output formats.

HealthMonitor

Liveness and readiness probes for the CRP runtime.

Register named checks with add_check(name, callable). Each check returns True (healthy) or False (unhealthy).

Usage::

hm = HealthMonitor()
hm.add_check("emitter", lambda: emitter.is_running)
hm.add_check("warm_store", lambda: warm_store is not None)
status = hm.probe()
assert status.alive

add_check(name, check_fn)

Register a readiness check.

remove_check(name)

Remove a check. Returns True if it existed.

set_alive(alive)

Manually mark the runtime as alive or dead (e.g. on shutdown).

probe()

Run all registered checks and return combined status.

Liveness: True unless set_alive(False) was called. Readiness: True only if all registered checks pass.

HealthStatus dataclass

Result of a health probe.

to_dict()

Return the health status as a plain dict.

Returns:

Type Description
dict[str, Any]

Dict with alive, ready, and details fields.

MetricsExporter

Collects CRP metrics (counters, gauges, histograms) and exports them.

Usage::

mx = MetricsExporter()
mx.incr("dispatch.count")
mx.gauge("overhead.ratio", 0.12)
mx.observe("dispatch.latency_ms", 42.3)
print(mx.export(ExportFormat.JSON))

incr(name, delta=1)

Increment a counter by delta (default 1).

gauge(name, value)

Set a gauge to the current value.

observe(name, value)

Record an observation in a histogram bucket.

get_counter(name)

Return the current value of a counter metric.

Parameters:

Name Type Description Default
name str

Metric name.

 required

Returns:

Type Description
int

Counter value, or 0 if the counter does not exist.

get_gauge(name)

Return the current value of a gauge metric.

Parameters:

Name Type Description Default
name str

Metric name.

 required

Returns:

Type Description
float | None

Gauge value, or None if the gauge does not exist.

get_histogram(name)

Return all observations recorded for a histogram metric.

Parameters:

Name Type Description Default
name str

Metric name.

 required

Returns:

Type Description
list[float]

List of observed values, or an empty list if none exist.

export(fmt=ExportFormat.JSON)

Export all metrics in the given format.

Supported
  • JSON - human-friendly nested dict.
  • PROMETHEUS - text/plain Prometheus exposition format.
  • OTLP_JSON - OpenTelemetry-compatible JSON envelope.

reset()

Clear all collected metrics.

QualityReport dataclass

Result of a quality assessment.

to_dict()

Return the quality report as a plain dict.

Returns:

Type Description
dict[str, Any]

Dict with tier, overhead_pct, fact_miss_pct,
dict[str, Any]

degraded, and details fields.

QualityReporter

Classify and track quality tiers across a session.

Usage::

qr = QualityReporter()
report = qr.assess(overhead_pct=4.2, fact_miss_pct=1.0)
assert report.tier == QualityTier.S

# Later, if overhead rises:
report2 = qr.assess(overhead_pct=18.0, fact_miss_pct=30.0)
assert report2.tier == QualityTier.C
assert report2.degraded  # dropped from S → C

current_tier property

Most recently assessed tier, or None if never assessed.

history property

Return the history.

assess(overhead_pct, fact_miss_pct, details=None)

Compute the current quality tier and check for degradation.

Parameters:

Name Type Description Default
overhead_pct float

Current overhead as a percentage (0-100).

 required
fact_miss_pct float

Percentage of available facts not included in the envelope (0-100).

 required
details dict[str, Any] | None

Optional opaque dict attached to the report.

 None

QualityTier

Bases: Enum

CRP quality tiers (best → worst).

TelemetryWriter

Append-only JSONL writer for per-window telemetry.

Usage::

tw = TelemetryWriter("telemetry.jsonl")
tw.write(WindowTelemetry(session_id="abc", window_id="w0", ...))
tw.close()

Also works as a context manager::

with TelemetryWriter("telemetry.jsonl") as tw:
    tw.write(record)

records_written property

Return the records written.

write(record)

Serialize record and append one JSON line. Flushes immediately.

close()

Flush and close the underlying file if this writer owns it.

WindowTelemetry dataclass

Telemetry record for a single completed window.

Fields mirror the most useful per-window stats. Extra fields can be added via extra.

to_dict()

Return the telemetry record as a plain dict.

Returns:

Type Description
dict[str, Any]

Dict representation with extra fields flattened into the top level.

observability.audit

crp.observability.audit

Audit log - reconstruct sessions from recorded events (§8.9).

AuditLog stores every emitted CRPEvent and provides query helpers to
  • Replay the full timeline for a session or time range.
  • Filter by event type.
  • Reconstruct how many facts were created / superseded / archived.
Design goals
  • In-memory by default (no external DB).
  • Thread-safe append + query.
  • Easy to hook into EventEmitter via emitter.on_all(audit.record).

AuditLog

Immutable append-only event log with query support.

Usage::

audit = AuditLog()
emitter.on_all(audit.record)   # auto-capture every event
...
timeline = audit.query(session_id="abc")
fact_events = audit.query(event_type="fact.created")

count property

Total number of recorded events.

record(event)

Append an event (thread-safe, never raises).

query(*, event_type=None, session_id=None, since=None, until=None)

Return events matching all supplied filters.

Parameters:

Name Type Description Default
event_type str | None

Exact event type string (e.g. "dispatch.completed").

 None
session_id str | None

Match events whose data["session_id"] equals this.

 None
since float | None

Only events with timestamp >= since.

 None
until float | None

Only events with timestamp <= until.

 None

reconstruct_session(session_id)

Build a summary dict for session_id from the event stream.

Returns a dict with keys
  • session_id
  • events - full ordered list of events (as dicts).
  • dispatch_count - number of dispatch.completed events.
  • facts_created - count of fact.created events.
  • facts_superseded - count of fact.superseded events.
  • duration_s - seconds between first and last event.

clear()

Remove all stored events (useful in testing).

observability.events

crp.observability.events

Event emitter for CRP observability - structured event bus (§09 §9.5).

EventEmitter is a lightweight publish/subscribe bus that lets CLI, diagnostics, and future metrics sinks observe protocol activity without coupling to internals.

CRPEvent dataclass

Single protocol event.

to_dict()

Return the event as a plain dict.

Returns:

Type Description
dict[str, Any]

Dict with event_type, timestamp, and data fields.

EventEmitter

Simple synchronous event bus.

Usage::

emitter = EventEmitter()
emitter.on("dispatch.started", my_callback)
emitter.emit("dispatch.started", {"task": "..."})

is_running property

Return whether this object is running.

listener_count property

Return the current listener count.

start()

Mark the emitter as active (startup step 6).

stop()

Stop accepting and delivering events.

on(event_type, listener)

Subscribe listener to event_type.

Deduplicates listeners and enforces a per-event cap (§audit4 REL-M2).

on_all(listener)

Subscribe listener to every event type.

off(event_type, listener)

Remove a specific listener. Returns True if removed.

emit(event_type, data=None)

Emit an event to all matching listeners.

If the emitter has not been started, events are silently dropped. Listener exceptions are logged but never propagate.

observability.metrics

crp.observability.metrics

Metrics export and health monitoring (§8.9, §05).

MetricsExporter - collects counters/gauges and exports as Prometheus, OTLP-compatible JSON, or plain JSON. HealthMonitor - liveness + readiness probes for the CRP runtime.

Design goals
  • Zero external dependencies (no prometheus_client, no opentelemetry).
  • Thread-safe counters via stdlib threading.Lock.
  • Easy to read: one class per concern, plain dicts for state.

ExportFormat

Bases: Enum

Supported output formats.

MetricsExporter

Collects CRP metrics (counters, gauges, histograms) and exports them.

Usage::

mx = MetricsExporter()
mx.incr("dispatch.count")
mx.gauge("overhead.ratio", 0.12)
mx.observe("dispatch.latency_ms", 42.3)
print(mx.export(ExportFormat.JSON))

incr(name, delta=1)

Increment a counter by delta (default 1).

gauge(name, value)

Set a gauge to the current value.

observe(name, value)

Record an observation in a histogram bucket.

get_counter(name)

Return the current value of a counter metric.

Parameters:

Name Type Description Default
name str

Metric name.

 required

Returns:

Type Description
int

Counter value, or 0 if the counter does not exist.

get_gauge(name)

Return the current value of a gauge metric.

Parameters:

Name Type Description Default
name str

Metric name.

 required

Returns:

Type Description
float | None

Gauge value, or None if the gauge does not exist.

get_histogram(name)

Return all observations recorded for a histogram metric.

Parameters:

Name Type Description Default
name str

Metric name.

 required

Returns:

Type Description
list[float]

List of observed values, or an empty list if none exist.

export(fmt=ExportFormat.JSON)

Export all metrics in the given format.

Supported
  • JSON - human-friendly nested dict.
  • PROMETHEUS - text/plain Prometheus exposition format.
  • OTLP_JSON - OpenTelemetry-compatible JSON envelope.

reset()

Clear all collected metrics.

HealthStatus dataclass

Result of a health probe.

to_dict()

Return the health status as a plain dict.

Returns:

Type Description
dict[str, Any]

Dict with alive, ready, and details fields.

HealthMonitor

Liveness and readiness probes for the CRP runtime.

Register named checks with add_check(name, callable). Each check returns True (healthy) or False (unhealthy).

Usage::

hm = HealthMonitor()
hm.add_check("emitter", lambda: emitter.is_running)
hm.add_check("warm_store", lambda: warm_store is not None)
status = hm.probe()
assert status.alive

add_check(name, check_fn)

Register a readiness check.

remove_check(name)

Remove a check. Returns True if it existed.

set_alive(alive)

Manually mark the runtime as alive or dead (e.g. on shutdown).

probe()

Run all registered checks and return combined status.

Liveness: True unless set_alive(False) was called. Readiness: True only if all registered checks pass.

observability.quality

crp.observability.quality

Quality tier classification and reporting (§05, §10).

Each CRP session runs at a quality tier that tells the user how well the context pipeline is performing:

S  - all signals green, overhead < 5 %
A  - minor gaps, overhead < 10 %
B  - acceptable, overhead < 15 %
C  - degraded, hierarchical processing needed
D  - minimal / fallback mode

QualityReporter takes a handful of easily-computed metrics and maps them to one of these tiers. It can also detect degradation (tier dropping below a previous high-water mark).

QualityTier

Bases: Enum

CRP quality tiers (best → worst).

QualityReport dataclass

Result of a quality assessment.

to_dict()

Return the quality report as a plain dict.

Returns:

Type Description
dict[str, Any]

Dict with tier, overhead_pct, fact_miss_pct,
dict[str, Any]

degraded, and details fields.

QualityReporter

Classify and track quality tiers across a session.

Usage::

qr = QualityReporter()
report = qr.assess(overhead_pct=4.2, fact_miss_pct=1.0)
assert report.tier == QualityTier.S

# Later, if overhead rises:
report2 = qr.assess(overhead_pct=18.0, fact_miss_pct=30.0)
assert report2.tier == QualityTier.C
assert report2.degraded  # dropped from S → C

current_tier property

Most recently assessed tier, or None if never assessed.

history property

Return the history.

assess(overhead_pct, fact_miss_pct, details=None)

Compute the current quality tier and check for degradation.

Parameters:

Name Type Description Default
overhead_pct float

Current overhead as a percentage (0-100).

 required
fact_miss_pct float

Percentage of available facts not included in the envelope (0-100).

 required
details dict[str, Any] | None

Optional opaque dict attached to the report.

 None

observability.structured_logging

crp.observability.structured_logging

Structured logging with correlation IDs for CRP pipeline (§audit H9).

Provides: - Per-request correlation IDs propagated across dispatch pipeline stages - JSON-structured log formatter for machine-parseable logs - Context-local storage for request metadata

StructuredFormatter

Bases: Formatter

JSON log formatter with automatic correlation ID injection.

format(record)

Format record as a JSON log line with correlation IDs.

Parameters:

Name Type Description Default
record LogRecord

Standard library log record.

 required

Returns:

Type Description
str

JSON-encoded log entry.

new_correlation_id()

Generate and set a new correlation ID for the current context.

get_correlation_id()

Get the current correlation ID (empty string if none set).

set_session_context(session_id)

Set the session ID for the current logging context.

configure_structured_logging(level=logging.INFO)

Configure the root 'crp' logger with structured JSON output.

Call once at application startup for machine-parseable logs.

observability.telemetry

crp.observability.telemetry

Per-window telemetry writer - telemetry.jsonl output (§8.9).

TelemetryWriter appends one JSON object per line for every completed window. The JSONL file can be consumed by downstream analytics, dashboards, or the AuditLog.

Design goals
  • One line per window → easy to grep, jq, or stream-parse.
  • Flush after every write so no data is lost on crash.
  • Works with file paths or any writable file-like object.

WindowTelemetry dataclass

Telemetry record for a single completed window.

Fields mirror the most useful per-window stats. Extra fields can be added via extra.

to_dict()

Return the telemetry record as a plain dict.

Returns:

Type Description
dict[str, Any]

Dict representation with extra fields flattened into the top level.

TelemetryWriter

Append-only JSONL writer for per-window telemetry.

Usage::

tw = TelemetryWriter("telemetry.jsonl")
tw.write(WindowTelemetry(session_id="abc", window_id="w0", ...))
tw.close()

Also works as a context manager::

with TelemetryWriter("telemetry.jsonl") as tw:
    tw.write(record)

records_written property

Return the records written.

write(record)

Serialize record and append one JSON line. Flushes immediately.

close()

Flush and close the underlying file if this writer owns it.