publish_event api accessible from anywhere in the framework

[Amir-R25]

Summary

This PR adds the following:

• observability module:
  • ensure_started: Starts the singleton Observer for this process and returns it
  • publish_event: An async fire-and-forget publish to the process-wide Observer, must be awaited on the same async context as ensure_started
  • configure_writers: register writers. Must be run before ensure_started. You can use register on the object itself however.
  • shutdown: drains per-writer queues and stops the Observer
• observability_bridge:
• for now a single helper method called make_session_event(event_type, payload) that builds a session-scoped Event from the InternalContext

Closes #1241

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

A little demo script to play with:

import asyncio
import logging
from pathlib import Path

import railtracks as rt
from railtracks.observability import (
    Event,
    JsonlWriter,
    configure_writers,
    ensure_started,
    publish_event,
    shutdown,
)
from railtracks.observability_bridge import make_session_event
from rich import print

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


class MemoryWriter:
    def __init__(self):
        self.events: list[Event] = []

    async def start(self):
        pass

    async def write(self, event: Event):
        self.events.append(event)

    async def shutdown(self):
        pass


class BrokenWriter:
    async def start(self):
        pass

    async def write(self, event: Event):
        raise RuntimeError(f"boom on {event.event_type}")

    async def shutdown(self):
        pass


async def main():
    jsonl_dir = Path("./demo_data")
    memory = MemoryWriter()

    configure_writers([JsonlWriter(jsonl_dir), memory, BrokenWriter()])
    await ensure_started()
    try:
        with rt.Session():
            await publish_event(
                make_session_event("session.start", {"hello": "world"})
            )
            await publish_event(
                make_session_event(
                    "node.create", {"node_id": "n1", "parent_node_id": None}
                )
            )
            await publish_event(
                make_session_event(
                    "node.complete", {"node_id": "n1", "duration_ms": 42}
                )
            )
    finally:
        await shutdown()  # drains queues into writers

    print("\n== MemoryWriter received ==")
    for e in memory.events:
        print(f"  {e.event_type:18s} scope_id={e.scope_id[:8]}...  payload={e.payload}")

    jsonl_path = jsonl_dir / "session.jsonl"
    print(f"\n== {jsonl_path} ==")
    for line in jsonl_path.read_text().splitlines():
        print(f"  {line}")


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