Observability Feature 1: Event Model, Observer pub/sub class, and Writer protocol

[Amir-R25]

Summary

railtracks.observability submodule:

• Feature 1 of the observability epic Rethinking observability into an events stream #1239 . This work adds an independent streaming-event pipeline:
• Event model, Observer with per-writer queues
• Writer protocol, and a JsonlWriter that appends per-scope .jsonl files
• Closes Required pieces: Event data model, Observer Submodule, and jsonl Writer #1240.

Type of change

• Bug fix
• Feature
• Breaking change
• Docs
• Refactor / chore / build / tests

Checklist

• Lint & format pass (ruff check . && ruff format .)
• Tests added/updated and pass locally (pytest tests)
• Docs updated if user-facing behavior changed
• Breaking changes include migration notes

Notes

This work is purely additive, no existing code paths change so it's safe to merge to main directly.

Demo

1. Fan-out from a single publish() to multiple writers, with per-scope
   routing landing events in .../{scope}.jsonl files.
2. Drop-oldest policy under overload and a WARNING logs per drop, so a
   stalled writer becomes visible instead of silently growing memory.

import asyncio
import logging
import shutil
from pathlib import Path

from railtracks.observability import (
    SCOPE_EVALUATION,
    SCOPE_RETRIEVAL,
    SCOPE_SESSION,
    Event,
    JsonlWriter,
    Observer,
    QueuePolicy,
    Stamp,
)

logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s")

DATA_DIR = Path("/tmp/rt-obs-demo")


class PrintWriter:
    """A tiny custom writer — 4 lines is all the Writer protocol asks for."""

    async def start(self) -> None: ...

    async def write(self, event: Event) -> None:
        print(f"  [stdout]  {event.scope_type:<10} {event.event_type:<22} scope_id={event.scope_id}")

    async def shutdown(self) -> None: ...


class SlowWriter:
    """Sleeps on every write so the publisher outpaces the queue."""

    async def start(self) -> None: ...

    async def write(self, event: Event) -> None:
        await asyncio.sleep(0.1)

    async def shutdown(self) -> None: ...


def make_event(
    scope: str,
    event_type: str,
    scope_id: str,
    parent_scope_id: str | None = None,
    **payload,
) -> Event:
    return Event(
        event_type=event_type,
        stamp=Stamp.now(),
        scope_type=scope,
        scope_id=scope_id,
        parent_scope_id=parent_scope_id,
        payload=payload,
    )


async def demo_fan_out_and_routing() -> None:
    print("\n=== 1. Fan-out to two writers + per-scope routing ===\n")
    shutil.rmtree(DATA_DIR, ignore_errors=True)

    async with Observer() as obs:
        await obs.register(JsonlWriter(DATA_DIR), "jsonl")
        await obs.register(PrintWriter(), "stdout")

        s = "session-42"
        await obs.publish(make_event(SCOPE_SESSION, "session.start", s, hello="world"))
        await obs.publish(make_event(SCOPE_RETRIEVAL, "retrieval.query", "ret-1", s, q="what is x"))
        await obs.publish(make_event(SCOPE_RETRIEVAL, "retrieval.query", "ret-2", s, q="what is y"))
        await obs.publish(make_event(SCOPE_EVALUATION, "evaluator.metric_result", "eval-1", s, metric="accuracy", score=0.87))
        await obs.publish(make_event(SCOPE_SESSION, "session.end", s))

    print(f"\n  Files written under {DATA_DIR}:")
    for path in sorted(DATA_DIR.glob("*.jsonl")):
        print(f"    {path.name}: {len(path.read_text().splitlines())} events")


async def demo_drop_oldest_policy() -> None:
    print("\n=== 2. Drop-oldest policy: publisher outpaces a slow writer ===\n")
    async with Observer() as obs:
        await obs.register(
            SlowWriter(),
            "slow",
            maxsize=3,
            policy=QueuePolicy.DROP_OLDEST,
        )
        for i in range(10):
            await obs.publish(make_event(SCOPE_SESSION, "session.tick", f"tick-{i}"))
        print("\n  (shutdown will drain the surviving events before returning...)")


async def main() -> None:
    await demo_fan_out_and_routing()
    await demo_drop_oldest_policy()


if __name__ == "__main__":
    asyncio.run(main())

[soulFood5632]

Some comments. But looks good to me

[soulFood5632]

LGTM