Custom Failure Analyzers

A failure analyzer is any object with an analyze_failure method. There is no base class.

The protocol

python

from runtime_narrative.failure import FailureSummary

class MyAnalyzer:
    def analyze_failure(
        self,
        *,
        story_name: str,
        stage_name: str,
        failure: FailureSummary,
        stage_timeline: str,
        progress_percent: int,
    ) -> str | None:
        ...

    # Optional async variant — if absent, analyze_failure is called via asyncio.to_thread
    async def analyze_failure_async(self, *, ...) -> str | None:
        ...

FailureSummary is a dataclass from runtime_narrative.failure with fields: error_type, error_message, filename, lineno, function, source_line, exception_chain, exact_cause, traceback_text.

Return a string to attach as llm_analysis on FailureOccurred, or return None to indicate no analysis.

Minimal custom analyzer

python

from runtime_narrative import story, stage

class RuleBasedAnalyzer:
    def analyze_failure(
        self, *, story_name, stage_name, failure, stage_timeline, progress_percent
    ) -> str | None:
        if failure.error_type == "FileNotFoundError":
            return (
                f"File not found at {failure.filename}:{failure.lineno}. "
                "Check that the input path exists and the service has read access."
            )
        if failure.error_type == "TimeoutError":
            return "Upstream timed out. Check network connectivity and retry budget."
        return None

with story("Pipeline", failure_analyzer=RuleBasedAnalyzer()):
    with stage("Read Config"):
        raise FileNotFoundError("config.json")

Adding async support

If analyze_failure_async is present, story.__aexit__ calls it directly. If only analyze_failure is present, the story wraps it in asyncio.to_thread to avoid blocking the event loop.

python

import asyncio
import aiohttp

class HttpAnalyzer:
    def __init__(self, url: str):
        self._url = url

    def analyze_failure(self, *, story_name, stage_name, failure, stage_timeline, progress_percent):
        # Sync fallback — used in sync story() context
        import urllib.request, json
        payload = json.dumps({"traceback": failure.traceback_text}).encode()
        try:
            with urllib.request.urlopen(self._url, data=payload, timeout=10) as resp:
                return json.loads(resp.read())["analysis"]
        except Exception:
            return None

    async def analyze_failure_async(self, *, story_name, stage_name, failure, stage_timeline, progress_percent):
        # Preferred in async context — called directly without to_thread overhead
        async with aiohttp.ClientSession() as session:
            try:
                resp = await session.post(
                    self._url,
                    json={"traceback": failure.traceback_text},
                    timeout=aiohttp.ClientTimeout(total=10),
                )
                data = await resp.json()
                return data.get("analysis")
            except Exception:
                return None

Pass the instance as failure_analyzer= to story(), middleware, or the @runtime_narrative_story decorator — the same way as any of the built-in analyzers.

Next step

Read Background Analysis to run an analyzer without blocking the failure path on a slow LLM call.