Sub-stories & Log Capture
Nest story() calls to trace a request end-to-end across function and service boundaries, and route existing logging calls into the same event stream instead of a second, disconnected log.
Sub-stories: end-to-end call tracing
story() opened while another story() is already active (in the same sync/async context) automatically becomes a sub-story — no new API. It:
- Inherits
renderers,diagnostics_config, andfailure_analyzerfrom the parent unless you pass your own explicitly. - Sets
parent_story_idto the parent'sstory_id. - Sets
root_story_idto the top-most ancestor'sstory_id(so a 3-level nesting all shares oneroot_story_id). - Still succeeds/fails and times itself independently — a failed sub-story does not automatically fail its parent.
async def execute_query(sql: str):
async with story(f"DB: {sql}") as db_story: # auto-linked to whatever story is active
async with stage("Execute Query"):
await conn.execute(sql)
async def create_order():
async with story("POST /orders") as api_story:
async with stage("Persist Order"):
await execute_query("INSERT INTO orders ...")
# execute_query's story is a sub-story of api_story:
# parent_story_id == api_story.story_id
# root_story_id == api_story.story_id
ConsoleRenderer renders the resulting call tree directly, tagging every line with a [short_id] (the first 6 characters of that event's story_id), coloring a whole story family (a root story plus any sub-stories) the same deterministic color, and indenting by nesting depth so the call tree renders directly in the log:
[ad8cc2] ▶ Story started: POST /orders
[ad8cc2] ▶ Stage started: Persist Order
[d17c63] ▶ Story started: DB: INSERT orders
[d17c63] ▶ Stage started: Execute Query
[d17c63] ✔ Stage completed: Execute Query (0.021s)
[d17c63] ▶ Story ended: SUCCESS (0.034s)
[ad8cc2] ✔ Stage completed: Persist Order (0.034s)
[ad8cc2] ▶ Story ended: SUCCESS (0.052s)
The indent level for each story is tracked per story_id internally (base indent for a sub-story = parent's indent + parent's currently-open stage depth + 1) and cleaned up when that story's StoryCompleted is handled, so long-running processes don't leak memory per story.
This holds up under concurrency for free: asyncio.Task copies ContextVar state at creation time, and each OS thread starts with a fresh top-level context, so many concurrent callers sharing one helper (like execute_query above) never cross-link into each other's story tree. OtelRenderer uses parent_story_id to make sub-stories real child spans of their parent span.
Run: uv run python examples/substory_db_call.py
NarrativeLogHandler: capture existing logging calls
NarrativeLogHandler folds logging.warning()/.error() into the same event pipeline as story()/stage() — one stream instead of two, tagged with the story/stage it happened in. Each captured record becomes a LogRecorded event.
import logging
from runtime_narrative import NarrativeLogHandler
logging.getLogger().addHandler(NarrativeLogHandler(level=logging.WARNING))
Outside an active story, records fall through to an optional fallback handler so nothing outside instrumentation is silently dropped:
NarrativeLogHandler(level=logging.WARNING, fallback=logging.StreamHandler())
Same colors, same indentation as the story tree
ConsoleRenderer tags every rendered line — including LogRecorded — with the same [short_id], story-family color, and nesting-depth indentation used for stories and stages, so captured log lines slot directly into the call tree shown above rather than appearing as a separate stream.
Run: uv run python examples/logging_bridge.py
Structured fields, custom styles, and per-story routing
extra={...} passed to a captured logging call becomes LogRecorded.fields:
logger.warning("cache miss", extra={"order_id": "ORD-42"})
With the structlog extra installed, ConsoleRenderer renders LogRecorded using structlog's own default console style — colored level, timestamp, message, then fields as key=value:
[d9e653] 2026-07-01T16:28:34 [warning ] cache miss order_id=ORD-42 stage=Fetch
Without structlog, a plain equivalent (WARNING [stage] logger: message key=value ...) is used — no hard dependency either way. Two ConsoleRenderer constructor options let you customize further without writing a renderer from scratch:
| Option | Effect |
|---|---|
log_renderer=<callable> | Any (logger, name, event_dict) -> str callable, e.g. a custom structlog processor/renderer. |
level_icons={"warning": "⚠ ", "error": "✗ "} | Prefix string per lowercase level name, prepended to the message. Empty by default. |
For routing different story families to different styles or destinations, FilteredRenderer(predicate, renderer) wraps any renderer and only forwards events where the predicate returns True — every event carries story_name, so this covers cases like "all GET ... request stories render 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 "})),
]
Run: uv run python examples/structured_log_routing.py
For a combined look at colorful failure output plus emoji level_icons across every log level, run: uv run python examples/colorful_errors_and_emojis.py
Next step
Continue to SQLite Persistence & CLI to store the full story/stage/event history and query it after the fact.