▶ runtime-narrative

story()

A dual sync/async context manager that models one execution as a story. It creates a StoryRuntime, tracks nested stages, and on failure builds an enriched, structured failure report before re-raising the exception.

Constructor signature

python
story(
    name: str,
    *,
    renderers: Sequence[object] | None = None,
    failure_analyzer = None,
    background_analysis: bool = False,
    diagnostics_config: FailureDiagnosticsConfig | None = None,
    runtime_environment: str | None = None,
    failure_diagnostics: str | None = None,
    allow_rich_in_production: bool | None = None,
    app_roots: Sequence[str] | None = None,
    redact_extra: Sequence[str] | None = None,
    total_stages: int | None = None,
    dry_run: bool = False,
    module: str | None = None,
    suppress_traceback: bool = False,
)

__enter__ / __aenter__ returns a StoryRuntime instance.

Parameter reference

ParameterTypeDefaultDescription
namestrrequiredHuman-readable story name. Appears in all events and renderer output.
renderersSequence[object] | None[ConsoleRenderer()]One or more renderer objects. Pass an empty list [] to silence all output.
failure_analyzerobject | NoneNoneAnalyzer called on exception to produce LLM analysis text.
background_analysisboolFalseEmit FailureOccurred immediately with llm_analysis=None, then emit LLMAnalysisReady asynchronously. Requires async with.
diagnostics_configFailureDiagnosticsConfig | NoneNoneFull config object. When provided, all individual diagnostic kwargs are ignored.
runtime_environmentstr | Nonereads env"development" or "production". Production caps traceback length and forces lean mode.
failure_diagnosticsstr | Nonereads env"lean" (default) or "rich" (captures local variables on failure).
allow_rich_in_productionbool | NoneNoneWhen True, rich mode is permitted even in production.
app_rootsSequence[str] | NoneNone (uses cwd)Directories considered "app code" for primary frame selection. Frames inside these paths are preferred over stdlib/site-packages.
redact_extraSequence[str] | NoneNoneAdditional key substrings to redact from local variables in rich mode.
total_stagesint | NoneNoneDeclared total stages. Enables accurate progress percentages. Can also be set later via runtime.set_total_stages(n).
dry_runboolFalseStage exceptions are suppressed; all stages complete; story always succeeds.
modulestr | Noneauto-detectedThe module that opened the story, shown by ConsoleRenderer. Auto-detected from the caller's frame via sys._getframe(1) (cheap — no stack walking) if not given. Pass explicitly when story() is opened from inside a wrapper (a decorator, middleware, etc.) so it doesn't report the wrapper's own module instead of the real call site — @runtime_narrative_story does this automatically with func.__module__.
suppress_tracebackboolFalseWhen True, story() swallows the exception after emitting FailureOccurred/StoryCompleted instead of re-raising it — Python's own uncaught-exception traceback is not printed on top of the narration. False by default: the exception still propagates, matching plain with/try semantics.

Leave suppress_traceback at its default of False unless you own the surrounding code

By design, story() narrates a failure (FailureOccurred: type, message, location, stack summary, locals in rich mode) and then still re-raises the exception — same as a plain with/try block. Unless something else catches it, Python's own uncaught-exception handler prints its raw traceback right after the narration. Passing suppress_traceback=True swallows the exception after emitting FailureOccurred/StoryCompleted so nothing propagates past the block.

This defaults to False and should stay that way for any story() you don't fully control the surrounding code for. RuntimeNarrativeMiddleware, Celery's NarrativeTask, and NarrativeTaskGroup each open their own story() internally and rely on the exception propagating out of it — that's how a failed request becomes an HTTP 500, a failed task gets retried, and a failed sub-task shows up in NarrativeTaskGroupError. Turning this on for those would silently turn failures into "successes." Only set it on top-level story() calls you own, where you've already decided nothing outside the block needs to see the exception — e.g. a script's main(), or a batch job that logs and moves on to the next item.

Suppressing the raw traceback

python
with story("Linked List Checks", suppress_traceback=True):
    run_checks()  # narrated on failure; nothing propagates past this block

Sync usage

python
from runtime_narrative import story, stage

with story("Deploy Release", total_stages=4) as runtime:
    print(f"Story ID: {runtime.story_id}")
    with stage("Run Tests"):
        pass
    with stage("Build Image"):
        pass
    with stage("Push to Registry"):
        pass
    with stage("Update Manifests"):
        pass

Async usage

python
import asyncio
from runtime_narrative import story, stage

async def deploy():
    async with story("Deploy Release", total_stages=4) as runtime:
        async with stage("Run Tests"):
            await asyncio.sleep(0.5)
        async with stage("Build Image"):
            await asyncio.sleep(1.0)
        print(f"Progress after 2 stages: {runtime.progress_percent}%")
        async with stage("Push to Registry"):
            await asyncio.sleep(0.2)
        async with stage("Update Manifests"):
            await asyncio.sleep(0.1)

asyncio.run(deploy())

StoryRuntime API

story.__enter__ / story.__aenter__ returns a StoryRuntime:

MemberTypeDescription
namestrStory name
story_idstrUUID for this execution
started_atdatetimeEntry timestamp
stageslist[StageRecord]All stages opened so far
dry_runboolWhether dry_run is active
completed_stagesint (property)Count of StageCompleted events
total_stagesint (property)Declared or inferred total
progress_percentint (property)completed_stages / total_stages * 100
set_total_stages(n)methodUpdate the declared total after entry
build_stage_timeline()methodReturns a one-line stage summary string
emit(event)methodSync dispatch to all renderers
emit_async(event)methodAsync dispatch, awaiting async renderers
record_failure(exc, *, stage_name=None)async methodEmit FailureOccurred for exc without affecting exception propagation (see below)

record_failure() — explicit failure recording

Use await runtime.record_failure(exc) in saga or rollback handlers where you need to record a failure without relying on __aexit__ owning the exception lifecycle:

python
import asyncio
from runtime_narrative import story, stage

async def saga():
    async with story("Payment Saga", renderers=[...]) as runtime:
        try:
            async with stage("Charge Card"):
                raise ConnectionError("gateway timeout")
        except ConnectionError as exc:
            # Record the failure — FailureOccurred is emitted with full diagnostics
            await runtime.record_failure(exc, stage_name="Charge Card")
            # You control what happens next — compensate, re-raise, or swallow
            await rollback()
            raise

asyncio.run(saga())

record_failure runs the same diagnostics pipeline as the normal __aexit__ path (lean/rich mode, redaction, traceback capping) and emits FailureOccurred via emit_async. It never suppresses or re-raises the exception.

Run: uv run python examples/saga_record_failure.py

Tracking progress

python
from runtime_narrative import story, stage

with story("Data Pipeline") as runtime:
    runtime.set_total_stages(5)  # can set after entry instead of in constructor

    with stage("Extract"):
        pass
    print(f"{runtime.progress_percent}%")   # 20%

    with stage("Transform"):
        pass
    print(f"{runtime.progress_percent}%")   # 40%

    with stage("Load"):
        pass
    print(f"Completed: {runtime.completed_stages} / {runtime.total_stages}")

Multiple renderers

python
from runtime_narrative import story
from runtime_narrative.renderer.console import ConsoleRenderer
from runtime_narrative.renderer.json_renderer import JsonRenderer

with open("events.jsonl", "a") as log_file:
    with story(
        "Import Customers",
        renderers=[
            ConsoleRenderer(),
            JsonRenderer(output=log_file),
        ],
    ):
        ...

Silencing output

Pass an empty list to suppress all output — useful in tests or library code where the caller decides rendering:

python
with story("Silent Operation", renderers=[]):
    ...

Next step

Continue to stage() for nested stage context managers, or jump to Failure Diagnostics for the full lean/rich diagnostics pipeline referenced above.