py/facett_demo/saga.py
"""SAGA — record what the wrapped components actually did, and read it back.
A facett surface that only *renders* is unauditable: nothing outside the browser
knows which component mounted, with what data, or whether it refused. This module
makes this app (and the robot that drives the shipped wheel) a **thin saga
producer**: it captures :class:`SagaEvent`-shaped moments and ships them to a
running ``nornir-server``'s ``nornir.v1.Saga`` service, which lands them in the
served workspace's Mímir ``saga_events`` table. The same tape reads back through
``ReadSession`` — so an assertion can be made against *what was recorded*, not
only against what was on screen at the time.
thin producer (this file) nornir-server
SagaRecorder.record(...) ──RecordEvents──▶ WarehouseTapeStore → `saga_events`
.flush() ◀──ReadSession─── (seq-ordered, one session)
Nothing here reimplements the tape format: the wire messages are the generated
``nornir.v1`` stubs (see :mod:`facett_demo.proto`), and the 16 fields are
the Rust ``nornir_saga::SagaEvent`` fields verbatim.
Recording is **best-effort by default** for the interactive app — a demo must not
die because the tape sink is down — but the failure is never silent: a disabled or
failed recorder reports it through :attr:`SagaRecorder.last_error` and
:meth:`SagaRecorder.status`. Tests construct the recorder with ``strict=True`` so
an unreachable server is a hard failure instead of a green run over an empty tape.
"""
from __future__ import annotations
import json
import os
import time
import uuid
from pathlib import Path
from typing import Any, Iterable
from facett_demo.buildinfo import build_info
__all__ = [
"SagaUnavailable",
"SagaRecorder",
"DEFAULT_TARGET",
"DEFAULT_WORKSPACE",
"resolve_token",
]
#: Where the nornir gRPC server listens (``NORNIR_SERVER`` overrides).
DEFAULT_TARGET = os.environ.get("NORNIR_SERVER", "127.0.0.1:7878")
#: Which served workspace's warehouse receives the tape. The server refuses an
#: unset header with NOT_FOUND, so this is required, not decorative.
DEFAULT_WORKSPACE = os.environ.get("NORNIR_WORKSPACE", "nordisk")
#: The event kinds the Rust side parses; anything else folds to ``log`` there.
KINDS = ("nav", "click", "log", "error", "state")
class SagaUnavailable(RuntimeError):
"""The saga sink could not be reached, or refused the tape."""
def resolve_token() -> str:
"""Bearer secret, most-authoritative first: ``NORNIR_TOKEN`` then ``~/.nornir/token``.
Returns ``""`` when neither exists — an open (no-policy) server accepts that,
and a closed one answers UNAUTHENTICATED, which is the honest outcome.
"""
tok = os.environ.get("NORNIR_TOKEN", "").strip()
if tok:
return tok
p = Path(os.environ.get("NORNIR_HOME", str(Path.home() / ".nornir"))) / "token"
try:
return p.read_text().strip()
except OSError:
return ""
class SagaRecorder:
"""Buffer saga events and ship them to a nornir server's ``Saga`` service.
``seq`` is assigned by this recorder, monotonically from 1, so a read-back is
ordered exactly as the moments happened. ``app`` / ``app_version`` /
``git_sha`` are stamped from :func:`~facett_demo.buildinfo.build_info`
on every event — a tape always says which build produced it.
"""
def __init__(
self,
*,
session_id: str | None = None,
app: str = "facett-demo",
user: str | None = None,
target: str = DEFAULT_TARGET,
workspace: str = DEFAULT_WORKSPACE,
strict: bool = False,
) -> None:
self.session_id = session_id or f"{app}-{uuid.uuid4().hex[:12]}"
self.app = app
self.user = user or os.environ.get("USER", "unknown")
self.target = target
self.workspace = workspace
self.strict = strict
self.last_error: str | None = None
self._seq = 0
self._buf: list[Any] = []
self._sent = 0
bi = build_info()
self._app_version = bi["version"]
self._git_sha = bi["git_hash"]
self._stub = None
self._channel = None
self._pb = None
# ── wire plumbing ──────────────────────────────────────────────────────────
def _connect(self):
"""Lazily open the channel. Raises :class:`SagaUnavailable` in strict mode."""
if self._stub is not None:
return self._stub
try:
import grpc # noqa: PLC0415 — optional at import time, required at use
from facett_demo.proto import saga_pb2 as pb
from facett_demo.proto import saga_pb2_grpc as pbg
self._pb = pb
self._channel = grpc.insecure_channel(self.target)
self._stub = pbg.SagaStub(self._channel)
except Exception as e: # import error, bad target — both are "no sink"
self._fail(f"cannot open saga channel to {self.target}: {e}")
return None
return self._stub
def _md(self) -> list[tuple[str, str]]:
md = [("nornir-workspace", self.workspace)]
tok = resolve_token()
if tok:
md.append(("authorization", f"Bearer {tok}"))
return md
def _fail(self, msg: str) -> None:
self.last_error = msg
if self.strict:
raise SagaUnavailable(msg)
# ── capture ────────────────────────────────────────────────────────────────
def record(
self,
kind: str,
*,
component: str = "",
action: str = "",
detail: str = "",
route: str = "",
state: Any = None,
level: str = "",
span_id: str = "",
parent: str = "",
) -> None:
"""Buffer one moment. ``state`` is JSON-encoded into ``state_json``.
``state`` may already be a JSON *string* (the component's own
``state_json``, which is the whole point) — it is then passed through
verbatim rather than double-encoded.
"""
if kind not in KINDS:
raise ValueError(f"unknown saga kind {kind!r}; expected one of {KINDS}")
if isinstance(state, str) or state is None:
state_json = state or ""
else:
state_json = json.dumps(state, separators=(",", ":"))
self._seq += 1
self._buf.append(
{
"seq": self._seq,
"ts_micros": int(time.time() * 1_000_000),
"session_id": self.session_id,
"user": self.user,
"kind": kind,
"route": route,
"component": component,
"action": action,
"detail": detail,
"state_json": state_json,
"level": level,
"span_id": span_id,
"parent": parent,
"app": self.app,
"app_version": self._app_version,
"git_sha": self._git_sha,
}
)
# Convenience wrappers naming the kinds the panes/robot actually emit.
def mounted(self, component: str, state_json: str, *, action: str = "mount", detail: str = "") -> None:
"""A component mounted and published its own state."""
self.record("state", component=component, action=action, detail=detail, state=state_json)
def refused(self, component: str, why: str) -> None:
"""A component honestly refused to render (fixture-only, no data, unknown)."""
self.record("error", component=component, action="refuse", detail=why, level="warn")
# ── ship + read back ───────────────────────────────────────────────────────
@property
def pending(self) -> int:
"""Events captured but not yet accepted by the server."""
return len(self._buf)
def flush(self) -> int:
"""Ship the buffer via ``RecordEvents``; returns the count the server accepted.
The buffer is cleared only on an accepted batch, so a transient outage
does not lose the tape — the next flush re-sends it.
"""
if not self._buf:
return 0
stub = self._connect()
if stub is None:
return 0
pb = self._pb
events = [pb.SagaEventPb(**e) for e in self._buf]
try:
resp = stub.RecordEvents(
pb.RecordSagaEventsRequest(session_id=self.session_id, events=events),
metadata=self._md(),
timeout=30,
)
except Exception as e:
self._fail(f"RecordEvents failed: {_rpc_msg(e)}")
return 0
accepted = int(resp.accepted)
if accepted != len(events):
self._fail(f"partial accept: sent {len(events)}, accepted {accepted}")
self._buf.clear()
self._sent += accepted
return accepted
def read_session(self, session_id: str | None = None) -> list[dict[str, Any]]:
"""Read one session's events back via ``ReadSession``, seq-ordered.
This is the half that makes a recording *provable*: the caller can assert
on the landed tape rather than on its own in-memory buffer.
"""
stub = self._connect()
if stub is None:
return []
pb = self._pb
try:
resp = stub.ReadSession(
pb.SagaSessionQuery(session_id=session_id or self.session_id),
metadata=self._md(),
timeout=30,
)
except Exception as e:
self._fail(f"ReadSession failed: {_rpc_msg(e)}")
return []
return [_event_to_dict(e) for e in resp.events]
def list_sessions(self) -> list[dict[str, Any]]:
"""The server's painful-first Sessions roll-up (a JSON array)."""
stub = self._connect()
if stub is None:
return []
pb = self._pb
try:
resp = stub.ListSessions(pb.Empty(), metadata=self._md(), timeout=30)
except Exception as e:
self._fail(f"ListSessions failed: {_rpc_msg(e)}")
return []
try:
rows = json.loads(resp.json or "[]")
except json.JSONDecodeError as e:
self._fail(f"ListSessions returned non-JSON: {e}")
return []
return rows if isinstance(rows, list) else []
def status(self) -> dict[str, Any]:
"""Honest one-glance state: where it ships, how much landed, what broke."""
return {
"session_id": self.session_id,
"target": self.target,
"workspace": self.workspace,
"captured": self._seq,
"accepted": self._sent,
"pending": self.pending,
"last_error": self.last_error,
}
def close(self) -> None:
"""Flush what is left and drop the channel."""
try:
self.flush()
finally:
if self._channel is not None:
self._channel.close()
self._channel = None
self._stub = None
def __enter__(self) -> "SagaRecorder":
return self
def __exit__(self, *exc: object) -> None:
self.close()
_FIELDS = (
"seq", "ts_micros", "session_id", "user", "kind", "route", "component", "action",
"detail", "state_json", "level", "span_id", "parent", "app", "app_version", "git_sha",
)
def _event_to_dict(e: Any) -> dict[str, Any]:
return {f: getattr(e, f) for f in _FIELDS}
def _rpc_msg(e: Exception) -> str:
"""A grpc error's code+details, or the exception itself."""
code = getattr(e, "code", None)
details = getattr(e, "details", None)
if callable(code) and callable(details):
return f"{code()}: {details()}"
return str(e)
def events_by_component(events: Iterable[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
"""Group a read-back tape by component — the shape assertions want."""
out: dict[str, list[dict[str, Any]]] = {}
for ev in events:
out.setdefault(ev.get("component", ""), []).append(ev)
return out