Renderers

A renderer is any object with a handle(self, event: object) method. There is no base class, no registration step. Sync renderers work with both with story(...) and async with story(...); async renderers (whose handle is a coroutine) are only awaited when using async with story(...). Renderer exceptions are caught, printed to stderr, and never crash the story.

from runtime_narrative import story, stage

async with story("My Pipeline", renderers=[RendererA(), RendererB()]):
    async with stage("Step 1"):
        ...

10.1 ConsoleRenderer

Prints colored, human-readable output to stdout by default. Uses typer.secho for color when the console extra is installed; falls back to plain print otherwise. Automatically substitutes ASCII glyphs (>, [ok], [FAIL]) for Unicode ones on non-UTF-8 terminals or output streams. Handles all seven events (including LogRecorded).

ConsoleRenderer is the default when no renderers= argument is passed to story().

Every line is prefixed with a YYYY-MM-DD HH:MM:SS.mmm timestamp (from the event's own .timestamp) followed by a [short_id] tag (first 6 characters of that event's story_id), colored per story family (a root story and its sub-stories share one deterministic color), and indented by nesting depth (one level per open stage / sub-story). See Sub-stories and Log Capture for the full nested output.

When known, the module that opened a story or stage (see story()/stage()'s module parameter) is shown in parentheses: always on the Story started line, and on a Stage started line only when it differs from the previous stage's module in that story — so a run of same-module stages doesn't repeat the tag, while a genuine cross-module transition stays visible:

2026-07-03 12:03:41.208 [abcdef] ▶ Story started: Process Upload (app.routes.upload)
2026-07-03 12:03:41.209 [abcdef]   ▶ Stage started: Validate Input (app.validators)
2026-07-03 12:03:41.210 [abcdef]   ✔ Stage completed: Validate Input (0.001s)
2026-07-03 12:03:41.211 [abcdef]   ▶ Stage started: Insert Record (app.db)

Validate Input and any further same-module stage right after it would show no (module) tag; the next transition to a different module shows it again.

from runtime_narrative import story, stage, ConsoleRenderer  # top-level import

with story("Import Pipeline", renderers=[ConsoleRenderer()]):
    with stage("Load CSV"):
        data = ["alice", "bob"]
    with stage("Validate"):
        assert data

ConsoleRenderer(*, log_renderer=None, level_icons=None, output=None):

• log_renderer — a callable (logger, name, event_dict) -> str used to format LogRecorded lines. Defaults to structlog.dev.ConsoleRenderer() (its own default style) when the structlog extra is installed; a plain built-in format otherwise. Pass any structlog renderer/processor, or your own callable, for a fully custom style.
• level_icons — dict[str, str] mapping a lowercase level name to a string prepended to the message (e.g. an emoji or short tag). Empty by default. Any character not representable on the terminal's encoding (common on Windows cp1252 consoles) is substituted rather than crashing the renderer, with or without the console extra installed.
• output — any file-like object with a write method. Defaults to sys.stdout. Pass an open file to send the same human-readable output there instead of the terminal:

with open("app.log", "a", encoding="utf-8") as log_file:
    with story("Import Pipeline", renderers=[ConsoleRenderer(output=log_file)]):
        with stage("Load CSV"):
            data = ["alice", "bob"]

typer/click auto-detect that a plain file isn't a terminal and strip ANSI color codes, so a file written this way is clean text. Unicode-vs-ASCII glyph selection is also re-evaluated against the given output's encoding rather than the terminal's. Output is flushed after every line, so the file reflects state up to the moment of a crash rather than sitting in an unflushed buffer. For structured, machine-readable file logging (NDJSON), use JsonRenderer instead — pass both renderers together to get a human-readable file and a JSON file at once.

Color and the story/stage glyphs (▶/✔/❌, falling back to >/[ok]/[FAIL] on non-UTF-8 terminals) are always on — there's no configuration needed to get colorful success/failure output, and no constructor option to change those specific glyphs today. level_icons is the supported way to add your own emoji/markers, scoped to LogRecorded lines. Run: uv run python examples/colorful_errors_and_emojis.py

10.1a FilteredRenderer

Wraps another renderer, forwarding an event only when predicate(event) is True. Since every event type carries story_name, this is the general mechanism for giving different story families their own style or destination — e.g. all "GET ..." request stories rendered differently from everything else:

from runtime_narrative import ConsoleRenderer, FilteredRenderer

renderers = [
    FilteredRenderer(lambda e: e.story_name.startswith("GET "), ConsoleRenderer()),
    FilteredRenderer(lambda e: not e.story_name.startswith("GET "), ConsoleRenderer(level_icons={"error": "X "})),
]

It mirrors the wrapped renderer's sync/async handle, so it composes with JsonRenderer, SqliteStoryRenderer, OtelRenderer, or any custom renderer — not just ConsoleRenderer.

10.1b CoalescingRenderer

Wraps another renderer, collapsing a run of identical back-to-back stages — e.g. a status-polling loop that runs with stage("Check Pipeline Status") every couple of seconds inside one long-running story — into a single summary line instead of flooding the log with one StageStarted/StageCompleted pair per iteration:

from runtime_narrative import story, stage, ConsoleRenderer, CoalescingRenderer

with story("Process Upload", renderers=[CoalescingRenderer(ConsoleRenderer())]):
    with stage("Upload File"):
        ...
    while not done:
        with stage("Check Pipeline Status"):
            done = check_status()
        time.sleep(2)

The wrapped renderer still sees the first threshold occurrences (default 2) of a repeated (story_id, stage_name) pair in full, so it's visible that polling started. From the next occurrence on, StageStarted/StageCompleted for that stage are suppressed and accumulated instead of forwarded. As soon as a different event arrives for that story — a different stage, a failure, or the story completing — the run is flushed as one LogRecorded summary line reporting the total call count and total time spent, e.g.:

'Check Pipeline Status' repeated 41 more times (43 total) over 84.200s (avg 1.958s/call)

Only wrap the human-facing renderer(s) with this — typically ConsoleRenderer. Renderers meant for full-fidelity machine consumption (JsonRenderer, SqliteStoryRenderer, OtelRenderer, and friends) should keep receiving every real event and should not be wrapped. It mirrors the wrapped renderer's sync/async handle, the same way FilteredRenderer does.

10.2 JsonRenderer

Emits one JSON object per line to stdout (or a file). Designed for log aggregators and NDJSON pipelines. FailureOccurred output includes the full extended diagnostics payload: stack_frames, source_snippet, locals_by_frame, traceback_text, and all other fields.

from runtime_narrative import story, stage
from runtime_narrative.renderer.json_renderer import JsonRenderer

# Write to stdout (default)
with story("ETL Run", renderers=[JsonRenderer()]):
    with stage("Extract"):
        pass

# Write to a file with indentation
with open("events.json", "w") as f:
    renderer = JsonRenderer(output=f, indent=2)
    with story("ETL Run", renderers=[renderer]):
        with stage("Extract"):
            pass

10.3 RotatingJsonRenderer

A subclass of JsonRenderer that rotates the output file when it reaches a size limit, following logging.handlers.RotatingFileHandler naming conventions.

from runtime_narrative import story, stage
from runtime_narrative.renderer.json_renderer import RotatingJsonRenderer

renderer = RotatingJsonRenderer(
    "narrative_events.log",
    max_bytes=5 * 1024 * 1024,  # rotate at 5 MB
    backup_count=3,              # keep .log, .log.1, .log.2, .log.3
)

with story("Nightly Batch", renderers=[renderer]):
    with stage("Fetch Orders"):
        pass

10.4 HtmlReportRenderer

Writes a self-contained HTML report file on StoryCompleted. Includes a story header with outcome badge, stage timeline bar chart (proportional durations), and (on failure) a failure detail section with the traceback and any LLM analysis. No external CSS or JS dependencies.

from runtime_narrative import story, stage
from runtime_narrative.renderer.html_renderer import HtmlReportRenderer

renderer = HtmlReportRenderer("report.html", open_browser=True)

with story("Data Pipeline", renderers=[renderer]):
    with stage("Load"):
        pass
    with stage("Transform"):
        pass
    with stage("Export"):
        pass
# After the with block, report.html is written and opened in the default browser

HtmlReportRenderer collects all events in memory and writes on StoryCompleted. If the process is killed before the story completes, no file is written.

10.5 SqliteStoryRenderer

Persists all story events to a SQLite database. No extra dependencies — uses stdlib sqlite3. The connection is opened lazily on the first handle() call; tables are created with CREATE TABLE IF NOT EXISTS.

See SQLite Persistence & CLI for the schema and CLI details.

from runtime_narrative import story, stage
from runtime_narrative.renderer.persistence_renderer import SqliteStoryRenderer

renderer = SqliteStoryRenderer("myapp.db")

with story("User Import", renderers=[renderer]):
    with stage("Fetch Users"):
        pass
    with stage("Upsert"):
        pass

10.6 OtelRenderer

Maps story events to OpenTelemetry spans. Each story becomes a root span; each stage becomes a child span. On FailureOccurred, the story span's status is set to ERROR with rich attributes. On LLMAnalysisReady, a span event is added. Requires the [otel] extra.

from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

from runtime_narrative import story, stage
from runtime_narrative.renderer.otel_renderer import OtelRenderer

exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))

renderer = OtelRenderer(
    tracer_provider=provider,
    tracer_name="myapp",
    min_duration_ms=5.0,              # skip spans faster than 5 ms
    exclude_stages={"Health Check"},  # never create spans for these stages
)

with story("Order Processing", renderers=[renderer]):
    with stage("Validate"):
        pass
    with stage("Charge Card"):
        pass

spans = exporter.get_finished_spans()
for span in spans:
    print(span.name, span.status.status_code.name)

10.7 OtelLogRenderer

Emits all six lifecycle events as OpenTelemetry log records. Automatically correlates records with the ambient OTel context (trace ID, span ID, trace flags). Requires the [otel] extra.

from runtime_narrative import story, stage
from runtime_narrative.renderer.otel_log_renderer import OtelLogRenderer

renderer = OtelLogRenderer(logger_name="myapp.narrative")

async with story("Checkout Flow", renderers=[renderer]):
    async with stage("Validate Cart"):
        pass
    async with stage("Submit Payment"):
        pass

10.8 OtelMetricsRenderer

Emits four OpenTelemetry metric instruments. Requires the [otel] extra.

from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader

from runtime_narrative import story, stage
from runtime_narrative.renderer.otel_metrics_renderer import OtelMetricsRenderer

reader = InMemoryMetricReader()
provider = MeterProvider(metric_readers=[reader])

renderer = OtelMetricsRenderer(meter_provider=provider, meter_name="myapp")

with story("Nightly Job", renderers=[renderer]):
    with stage("Extract"):
        pass
    with stage("Load"):
        pass

metrics = reader.get_metrics_data()

10.9 PrometheusRenderer

Emits four Prometheus metrics via prometheus_client. Requires the [prometheus] extra.

from prometheus_client import CollectorRegistry, generate_latest
from runtime_narrative import story, stage
from runtime_narrative.renderer.prometheus_renderer import PrometheusRenderer

registry = CollectorRegistry()
renderer = PrometheusRenderer(registry=registry)

with story("API Request", renderers=[renderer]):
    with stage("Auth"):
        pass
    with stage("Fetch Data"):
        pass

print(generate_latest(registry).decode())

Pass a custom registry to isolate metrics in tests or multi-tenant setups. If registry=None, the default global registry is used.

10.10 AlertRoutingRenderer

An async renderer that fires on FailureOccurred events only. Calls all configured destinations concurrently via asyncio.gather. Destination failures are caught and logged to stderr — they never propagate to the story.

import asyncio
from runtime_narrative import story, stage
from runtime_narrative.renderer.alert_renderer import (
    AlertRoutingRenderer,
    HttpWebhookDestination,
    SlackWebhookDestination,
)

alert_renderer = AlertRoutingRenderer(
    destinations=[
        HttpWebhookDestination(
            "https://alerts.internal/webhook",
            headers={"Authorization": "Bearer mysecret"},
            timeout=5.0,
        ),
        SlackWebhookDestination(
            "https://hooks.slack.com/services/T000/B000/xxxx",
            timeout=8.0,
        ),
    ],
    only_stories={"Order Processing", "Nightly Billing"},  # filter by story name
    only_error_types={"TimeoutError", "ConnectionError"},   # filter by error class name
)

async def main():
    try:
        async with story("Payment Flow", renderers=[alert_renderer]):
            async with stage("Charge"):
                raise ConnectionError("gateway unreachable")
    except ConnectionError:
        pass

asyncio.run(main())

HttpWebhookDestination POSTs a JSON payload with fields: story_id, story_name, stage_name, error_type, error_message, filename, lineno, function, llm_analysis, timestamp.

SlackWebhookDestination (subclass of HttpWebhookDestination) posts a Block Kit payload: a header block ("Story failed: {name}"), a section block with stage/error/location, and an optional analysis section when llm_analysis is present.