# Observability Accepted-event sinks, the metrics protocol, and shipped backends. For the conceptual model see the [Operating in production](https://docs.activegraph.ai/guides/operating-in-production/index.md) guide. ## Accepted-event sinks Bases: `Protocol` Outbound observer of accepted runtime events. The dispatcher invokes all four methods on the attachment's dedicated worker. Implementations must not rely on return values influencing the runtime: every return is ignored, and every exception is isolated and exposed through sink status/metrics instead of crossing into execution. ## `open()` Acquire adapter resources before the first delivery. ## `on_event(event, context)` Observe one isolated event copy and its delivery metadata. ## `flush()` Make all previously handled deliveries externally visible. ## `close()` Release adapter resources; implementations should be idempotent. Immutable metadata accompanying one event offered to a sink. `sequence` is the event's one-based position in the graph's currently materialized accepted log. `mode` is `"live"` for the R1 dispatch path; `"replay_export"` is reserved for a future explicit historical export API and is never produced by normal load, fork, or strict replay. ## `to_dict()` Return the stable JSON-ready delivery metadata shape. Configuration for one isolated EventSink attachment. A capacity of 1024 and `DROP_NEWEST` are the locked defaults. Supply `name` when attaching two instances of the same sink class to one graph; attachment names are unique status keys and metric label values. Bases: `str`, `Enum` Non-blocking policy used when one sink's private queue is full. `DROP_NEWEST` preserves the queued prefix, `DROP_OLDEST` keeps the newest observation by evicting the oldest waiting item, and `FAIL_SINK` disables the attachment after counting the overflow. None of the policies waits for capacity on the runtime hot path. Control and status handle for one EventSink attachment. Handles are returned by `Graph.add_sink` / `Runtime.add_sink`. Delivery injection is runtime-internal; callers use `status`, `flush`, and `close`. Flush and close wait only at this explicit control-plane boundary and return `False` on timeout or adapter failure. ## `name` Return the immutable attachment name used by status and metrics. ## `run_id` Return the immutable run id associated with this attachment. ## `queue_capacity` Return the immutable maximum number of waiting deliveries. ## `overflow_policy` Return the immutable policy selected when the queue is full. ## `status()` Return an immutable exact local status snapshot. ## `flush(timeout=5.0)` Flush deliveries accepted before this call within `timeout`. The adapter's `flush` method executes on its worker after every earlier accepted delivery has either completed or been observably dropped. A hanging adapter cannot make this method wait forever unless the caller explicitly passes `timeout=None`. ## `close(timeout=5.0)` Drain and close this attachment within `timeout`. A graph-owned handle first detaches through its owning Graph, so direct handle closure cannot strand a dead attachment in live fan-out. Closing is idempotent. On timeout the daemon worker is left alive and status remains `closing` (or `failed`); Python cannot safely cancel arbitrary adapter code already executing in another thread. Point-in-time health and loss accounting for one sink attachment. Counts remain available with `NoOpMetrics` and are therefore the exact local observability surface. `queue_depth` excludes an item currently executing inside `on_event`. `dropped_by_reason` is sorted so status snapshots and JSON renderings are deterministic. Bases: `str`, `Enum` Lifecycle state reported for one sink worker. Opening and running attachments accept offers; closing attachments are detached. Closed and failed are terminal worker outcomes; graph status retains terminal failures for diagnosis. Write one canonical context/event envelope per UTF-8 JSONL line. Files are opened in append mode and never rotated or truncated. The adapter is thread-safe and reference-counts worker `open` / `close` calls so one instance can be shared by concurrent run attachments. Concurrent writers to one path must share that instance; distinct adapter instances and processes do not coordinate a path-level lock. Store serde normalization is reused for Decimal, date/datetime, set, and Unicode payload values before keys are sorted into canonical compact JSON. ## `open()` Open the target in append mode, sharing one file across workers. ## `on_event(event, context)` Append one stable JSON envelope followed by exactly one newline. ## `flush()` Flush buffered JSONL bytes when the adapter is open. ## `close()` Release one attachment reference and close after the last one. Thread-safe in-memory EventSink for downstream tests. One instance may be shared across concurrent runtimes. `deliveries` returns an immutable snapshot; `wait_for` provides deterministic test synchronization without sleeps. ## `deliveries` Return a stable snapshot of every captured delivery. ## `events` Return only the captured events, in delivery order. ## `open()` Register one worker attachment. ## `on_event(event, context)` Capture a defensive event copy and wake waiting tests. ## `flush()` No-op: in-memory deliveries are visible immediately. ## `close()` Release one worker attachment; repeated closes are harmless. ## `wait_for(count, timeout=5.0)` Wait until at least `count` deliveries exist, returning success. ## `clear()` Remove captured deliveries without changing attachment lifecycle. One event/context pair captured by :class:`RecordingSink`. Both values are isolated snapshots, suitable for deterministic assertions after the live runtime and its sink workers have continued or closed. ## Adapter conformance Future sink adapters subclass this suite, implement its two read/factory hooks, and set `__test__ = True` for pytest collection. Bases: `ABC` Mix into a pytest class to validate an EventSink adapter. Subclasses implement `make_sink` and `read_deliveries` and set `__test__ = True` so pytest collects the inherited cases. A fresh sink is requested per test; `cleanup` may remove files or release adapter-specific resources. The inherited tests drive adapters only through the public Runtime/Graph attachment boundary. ## `make_sink()` Return a fresh candidate adapter for one conformance test. ## `read_deliveries(sink)` Read adapter output as event/context pairs after a flush. ## `cleanup()` Release adapter-specific test resources. Default: no-op. ## Metrics protocol Bases: `Protocol` Three methods, all best-effort, all non-throwing. Implementations MUST tolerate unknown metric names. Unknown tag keys are also accepted; cardinality discipline is the caller's job. Methods may be called concurrently by independent runtime and sink workers. ## Backends Default Metrics implementation. Does nothing. Three method bodies, each a single `return`. The runtime is fully functional with NoOpMetrics. Profile-checked for zero allocation pressure under steady load. Drop-in Metrics implementation backed by prometheus_client. Instruments are lazy. Tag keys for an instrument are fixed by the first observation; subsequent observations with a different key set raise (prometheus_client behavior). This matches the standard metric list's fixed tag schemas. Drop-in Metrics implementation backed by OpenTelemetry metrics. Pass a custom `meter` in tests or applications that already own a MeterProvider. Otherwise the global `opentelemetry.metrics` meter provider is used.