aeat.core.observability._sink module

Logging handler that bridges logging to JSONL run events.

The handler subscribes to the standard logging machinery so any caller using aeat.core.logging.get_logger() automatically picks up the JSONL sink while a aeat.core.observability.run_context() is active. Records that do not carry a run_event extra are skipped — bare log lines never leak into events.jsonl.

The run_id / step_id attributes are stamped onto every logging.LogRecord by the factory installed in aeat.core.logging.

Each sink instance is bound to a single run_id and filters any event whose run_id does not match. This prevents cross-run contamination when several aeat.core.observability.run_context() blocks execute concurrently (e.g. tasks in an asyncio event loop) and therefore have competing sinks attached to the root logger at the same time.

class JsonlRunSink(target, *, run_id)

Bases: Handler

Append-only JSONL sink for aeat.core.observability.RunEvent records.

The handler opens the target path lazily on first emit so a aeat.core.observability.run_context() enter is cheap when no events ever fire. Each emit flushes the file handle; close() additionally calls os.fsync() so a process kill mid-run still leaves a durable JSONL trailer on disk.

Concurrency: the sink is bound to a single run_id and rejects events carrying a different run_id. File-handle mutations are guarded by an internal threading.Lock so multiple worker threads may emit concurrently without interleaving bytes on disk.

Parameters:

• target (Path)

• run_id (str)

property run_id: str

The run identifier this sink is bound to.

emit(record)

Write the JSON-encoded RunEvent carried by record.

Drops the record when there is no run_event extra or when the event belongs to a different run — see the module docstring for the concurrency rationale.

JSON serialisation runs outside the file-handle lock so concurrent threads can encode in parallel; only the write and flush are serialised. The whole emit path (including the encode) is wrapped in a single try — a serialisation failure (e.g. the pydantic model grew a non-JSON-safe field in a future refactor) must not crash the logging system, and must instead fall through to handleError like any other handler failure.

Return type:

None

Parameters:

record (LogRecord)

close()

Flush, os.fsync(), and close the underlying file handle.

Always invokes the base logging.Handler.close() so the handler is removed from the logging registry even when the flush or fsync raises.

Return type:

None