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
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
| Parameter | Type | Default | Description |
|---|---|---|---|
name | str | required | Human-readable story name. Appears in all events and renderer output. |
renderers | Sequence[object] | None | [ConsoleRenderer()] | One or more renderer objects. Pass an empty list [] to silence all output. |
failure_analyzer | object | None | None | Analyzer called on exception to produce LLM analysis text. |
background_analysis | bool | False | Emit FailureOccurred immediately with llm_analysis=None, then emit LLMAnalysisReady asynchronously. Requires async with. |
diagnostics_config | FailureDiagnosticsConfig | None | None | Full config object. When provided, all individual diagnostic kwargs are ignored. |
runtime_environment | str | None | reads env | "development" or "production". Production caps traceback length and forces lean mode. |
failure_diagnostics | str | None | reads env | "lean" (default) or "rich" (captures local variables on failure). |
allow_rich_in_production | bool | None | None | When True, rich mode is permitted even in production. |
app_roots | Sequence[str] | None | None (uses cwd) | Directories considered "app code" for primary frame selection. Frames inside these paths are preferred over stdlib/site-packages. |
redact_extra | Sequence[str] | None | None | Additional key substrings to redact from local variables in rich mode. |
total_stages | int | None | None | Declared total stages. Enables accurate progress percentages. Can also be set later via runtime.set_total_stages(n). |
dry_run | bool | False | Stage exceptions are suppressed; all stages complete; story always succeeds. |
module | str | None | auto-detected | The 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_traceback | bool | False | When 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
with story("Linked List Checks", suppress_traceback=True):
run_checks() # narrated on failure; nothing propagates past this block
Sync usage
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
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:
| Member | Type | Description |
|---|---|---|
name | str | Story name |
story_id | str | UUID for this execution |
started_at | datetime | Entry timestamp |
stages | list[StageRecord] | All stages opened so far |
dry_run | bool | Whether dry_run is active |
completed_stages | int (property) | Count of StageCompleted events |
total_stages | int (property) | Declared or inferred total |
progress_percent | int (property) | completed_stages / total_stages * 100 |
set_total_stages(n) | method | Update the declared total after entry |
build_stage_timeline() | method | Returns a one-line stage summary string |
emit(event) | method | Sync dispatch to all renderers |
emit_async(event) | method | Async dispatch, awaiting async renderers |
record_failure(exc, *, stage_name=None) | async method | Emit 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:
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
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
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:
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.