Skip to content

Structured Error Reports

Page Maps

graph LR
  family["Python Programming"]
  program["Python Functional Programming"]
  section["Streaming Resilience Failure Handling"]
  page["Structured Error Reports"]
  capstone["Capstone evidence"]

  family --> program --> section --> page
  page -.applies in.-> capstone
flowchart LR
  orient["Orient on the page map"] --> read["Read the main claim and examples"]
  read --> inspect["Inspect the related code, proof, or capstone surface"]
  inspect --> verify["Run or review the verification path"]
  verify --> apply["Apply the idea back to the module and capstone"]

This page is the payoff for the whole module. After making failures explicit, bounded, retryable, and resource-safe, the next move is turning that work into diagnostics that a team can actually read and act on.

Start With the Reporting Gap

Without a structured report, the earlier resilience work still tends to collapse back into noisy logs and scattered anecdotes. Reporting is the final functional reduction over the whole failure story.

  • If the only output is raw log lines, counts and patterns are hard to trust.
  • If sample errors are unbounded, the report becomes another memory risk.
  • If grouping rules are implicit, you cannot explain why the report is actionable rather than just verbose.

Core question:
How do you turn every error in a streaming pipeline into structured, serialisable, grouped reports — complete with counts, ordered samples, and retry metadata — while keeping the pipeline pure and memory-bounded?

This lesson introduces error reports as a bounded summary fold over the failure stream:

  • group failures by the dimensions the team actually uses to debug runs
  • keep bounded ordered samples so reports stay useful without exploding in size
  • produce serializable data structures that can be stored, diffed, and surfaced outside the running process

The motivating question is straightforward and important: once failures are captured well, how do we turn them into something actionable instead of noisy?

The naïve solution is scattered logging:

for r in embedded:
    if isinstance(r, Err):
        logger.error("Embedding failed: %s", r.error)

This loses the structure the whole module has been building.

The production solution folds the Result stream into an immutable report that preserves the counts, samples, and provenance worth carrying forward.

Use this when you need complete, structured failure diagnostics from batch RAG runs.

Outcome:
1. You will aggregate every error into a rich, bounded, serialisable report with one fold.
2. You will group by code/stage/path_prefix and extract retry metadata automatically.
3. You will ship perfect JSON error reports that survive 1 % or 50 % failure rates.

This section formalises exactly what you should review in reporting code: completeness, bounded memory, ordered samples, and purity of the resulting report value.

Concrete Motivating Example

Same 100 000 chunk tree from previous cores:

  • 95 000 embed successfully.
  • 4 800 transient failures → retried and succeeded.
  • 200 genuine failures (invalid content).

Desired final report (JSON):

{
  "total_errs": 200,
  "total_items": 100000,
  "error_rate": 0.002,
  "avg_attempts": 2.41,
  "by_code": {
    "INVALID_CONTENT": {"count": 200, "samples": [...]}
  },
  "by_stage": {
    "embed": {"count": 200, "samples": [...]}
  },
  "by_path_prefix": {
    "2.4": {"count": 180, "samples": [...]}
  }
}

One fold, bounded memory, perfect provenance.

1. Laws & Invariants (machine-checked)

Law Formal Statement Enforcement
Completeness Every Err in a finite stream contributes exactly once to each relevant group’s count; samples are a prefix of encounter order up to max_samples per group. test_report_completeness, test_report_sample_ordering.
Purity Same input stream → identical report (deterministic, no side effects). Reproducibility test.
Bounded Memory Memory ≤ O(#groups × max_samples). test_report_bounded_memory.
Ordering Samples within each group appear in original encounter order. test_report_sample_ordering.

These laws guarantee your reports are complete, reproducible, and memory-safe.

2. Decision Table – Which Report Do You Actually Need?

Need Bounded Memory? Need Samples? Recommended Fold
Just counts by code Yes No fold_error_counts
Full grouped report with samples Yes Yes fold_error_report(max_samples=20)
Everything (including successes) No Yes collect_both (Core 6)

Cap samples aggressively in production — 10–50 per group is plenty for debugging.

3. Public API Surface (end-of-Module-04 refactor note)

Refactor note: error reports live in funcpipe_rag.policies.reports (capstone/src/funcpipe_rag/policies/reports.py) and are re-exported from funcpipe_rag.api.core.

from funcpipe_rag.api.core import ErrGroup, ErrReport, fold_error_counts, fold_error_report, report_to_jsonable

4. Reference Implementations

4.1 Error Normalisation (for grouping)

def _normalize_err(e: Any) -> tuple[str, str, tuple[int, ...]]:
    code = getattr(e, "code", "UNKNOWN")
    stage = getattr(e, "stage", "UNKNOWN")
    path = getattr(e, "path", ())
    # Treat breaker errors as a distinct code/stage for reporting
    if isinstance(e, BreakInfo):
        code = e.code if e.code.startswith("BREAK/") else f"BREAK/{code}"
        stage = "BREAK"
    return code, stage, path

4.2 Group Builder (internal)

@dataclass(slots=True)
class _GroupBuilder(Generic[E]):
    count: int = 0
    samples: list[E] = field(default_factory=list)
    cap: int = 10

    def add(self, e: E) -> None:
        self.count += 1
        if len(self.samples) < self.cap:
            self.samples.append(e)

    def freeze(self) -> ErrGroup[E]:
        return ErrGroup(self.count, tuple(self.samples))

4.3 Full Structured Report

def fold_error_report(
    stream: Iterable[Result[Any, E]],
    *,
    max_samples: int = 10,
    path_depth: int = 3,
) -> ErrReport[E]:
    total_errs = total_items = 0
    by_code: dict[str, _GroupBuilder[E]] = {}
    by_stage: dict[str, _GroupBuilder[E]] = {}
    by_path: dict[tuple[int, ...], _GroupBuilder[E]] = {}
    sum_attempts = sum_delay = 0.0
    cnt_attempts = cnt_delay = 0

    for r in stream:
        total_items += 1
        if isinstance(r, Err):
            total_errs += 1
            e = r.error
            code, stage, path = _normalize_err(e)
            prefix = path[:path_depth]

            by_code.setdefault(code, _GroupBuilder(max_samples)).add(e)
            by_stage.setdefault(stage, _GroupBuilder(max_samples)).add(e)
            by_path.setdefault(prefix, _GroupBuilder(max_samples)).add(e)

            ctx = getattr(e, "ctx", None)
            if isinstance(ctx, Mapping):
                a = ctx.get("attempt")
                d = ctx.get("next_delay_ms")
                if isinstance(a, (int, float)):
                    sum_attempts += a
                    cnt_attempts += 1
                if isinstance(d, (int, float)):
                    sum_delay += d
                    cnt_delay += 1

    return ErrReport(
        total_errs=total_errs,
        total_items=total_items,
        by_code=MappingProxyType({k: v.freeze() for k, v in by_code.items()}),
        by_stage=MappingProxyType({k: v.freeze() for k, v in by_stage.items()}),
        by_path_prefix=MappingProxyType({k: v.freeze() for k, v in by_path.items()}),
        ctx_summary=MappingProxyType({
            "avg_attempts": sum_attempts / cnt_attempts if cnt_attempts else 0.0,
            "avg_next_delay_ms": sum_delay / cnt_delay if cnt_delay else 0.0,
            "error_rate": total_errs / total_items if total_items else 0.0,
        }),
    )

4.4 JSON Serialisation

def report_to_jsonable(report: ErrReport[E]) -> dict[str, Any]:
    def group_to_dict(g: ErrGroup[E]) -> dict[str, Any]:
        return {
            "count": g.count,
            "samples": [asdict(s) if is_dataclass(s) else {"value": s} for s in g.samples],
        }

    return {
        "total_errs": report.total_errs,
        "total_items": report.total_items,
        "error_rate": report.ctx_summary["error_rate"],
        "avg_attempts": report.ctx_summary["avg_attempts"],
        "avg_next_delay_ms": report.ctx_summary["avg_next_delay_ms"],
        "by_code": {k: group_to_dict(v) for k, v in report.by_code.items()},
        "by_stage": {k: group_to_dict(v) for k, v in report.by_stage.items()},
        "by_path_prefix": { ".".join(map(str, k)): group_to_dict(v) for k, v in report.by_path_prefix.items()},
    }

4.5 Idiomatic RAG Usage

# The stream is single-pass; split with tee for reporting + indexing
from itertools import tee

report_stream, index_stream = tee(embedded)

report = fold_error_report(report_stream, max_samples=20)

if report.total_errs > 0:
    logger.error("RAG embedding failures:\n%s", 
                 json.dumps(report_to_jsonable(report), indent=2))
    send_to_monitoring(report_to_jsonable(report))
else:
    logger.info("Embedding succeeded for all %d chunks", report.total_items)

index_chunks(filter_ok(index_stream))

5. Property-Based Proofs (capstone/tests/test_reports.py)

@given(items=st.lists(st.integers()))
def test_report_completeness(items):
    def f(x: int) -> Result[int, ErrInfo]:
        return Err(make_errinfo(f"C{x}", f"msg{x}", "stage", (x,))) if x % 2 else Ok(x)
    report = fold_error_report(map_result_iter(f, items))
    assert report.total_errs == sum(1 for x in items if x % 2)
    assert report.total_items == len(items)

@given(items=st.lists(st.integers(), unique=True))
def test_report_sample_ordering(items):
    def f(x: int) -> Result[int, ErrInfo]:
        # mark odd values as errors
        return Err(make_errinfo("ERR", f"msg{x}", "s", (x,))) if x % 2 != 0 else Ok(x)

    report = fold_error_report(map_result_iter(f, items))
    group = report.by_code.get("ERR")
    # Vacuous truth: if there are no ERR samples, ordering is trivially satisfied
    if group is None or not group.samples:
        return

    samples = group.samples
    sample_xs = [int(s.msg[3:]) for s in samples]  # "msg{X}" -> X
    positions = [
        next(i for i, x in enumerate(items) if x == sx and x % 2 != 0)
        for sx in sample_xs
    ]
    # Encounter order: positions must be strictly increasing
    assert positions == sorted(positions)

@given(items=st.lists(st.integers()))
def test_report_bounded_memory(items):
    report = fold_error_report(map_result_iter(lambda x: Err("E"), items), max_samples=10)
    assert all(len(g.samples) <= 10 for g in report.by_code.values())

6. Big-O & Allocation Guarantees

Variant Time Space Notes
fold_error_report O(N) O(#groups × max_samples) Bounded by max_samples

7. Anti-Patterns & Immediate Fixes

Anti-Pattern Symptom Fix
Ad-hoc logging Scattered diagnostics Use fold_error_report
Unbounded sample collection Memory blowup Cap with max_samples
No grouping Hard analysis Group by code/stage/path_prefix
Ignoring retry metadata Lost insights Aggregate from ErrInfo.ctx

Read the report as a consumed-stream summary

The report describes values that reached it. It does not know about source items a breaker never requested, transient failures that later succeeded, or exceptions translated somewhere else.

The cumulative learning route uses four chunks:

flaky, bad-a, bad-b, unreached

A fair retry queue and max_errs=1 breaker produce:

Final stream value Report interpretation
Err(EMBED/INVALID for bad-a) Record failure
Ok(FLAKY) Transient failure recovered; final value is success
Err(EMBED/INVALID for bad-b) Threshold-crossing record failure
Err(BREAK/ERR_COUNT) Terminal policy event

unreached was queued by the bounded retry engine but never attempted after the breaker closed the stream.

Run the whole route:

course=programs/python-programming/python-functional-programming
state="$course/capstone/module-reference-states/module-04"
venv=artifacts/venv/python-programming/python-functional-programming/capstone

PYTHONPATH="$state/src" \
  "$venv/bin/pytest" -q \
  "$state/tests/learning/test_module_04_resilient_streaming.py" \
  -k retry_breaker_and_report

Predict the report before reading the assertions:

total_items = 4
total_errs = 3
EMBED/INVALID count = 2
BREAK/ERR_COUNT count = 1
error_rate = 3 / 4

The report counts a terminal breaker value as an error because it is an Err in the consumed protocol. A product metric may choose to separate “record failures” from “policy events,” but it must do so by stable code or stage rather than subtracting one unexplained value.

Why the nested serialization proof matters

Retry annotates ErrInfo.ctx with an immutable mapping. BreakInfo retains the last error. The earlier serializer passed the whole breaker to dataclasses.asdict(), which attempted to deepcopy that immutable mapping and raised TypeError.

Module 04 now serializes a breaker field by field and recursively normalizes its last_error. The focused unit test test_break_report_serializes_nested_immutable_context prevents later states from losing that provenance. This is a useful review lesson: a dataclass being serializable at one level does not imply that every nested value supports deep copy.

Bound the right dimensions

max_samples=1 keeps one example per group while counts continue to grow. That bounds samples per group, not the number of groups. If an error code or path prefix is derived from high-cardinality input, the group dictionaries can still grow with the run. Stable, low-cardinality codes and deliberate path_depth are therefore part of the memory contract.

The JSON conversion proves the saved payload uses JSON-compatible local values. It does not prove schema compatibility across course versions, safe transport, or that a monitoring backend accepts the payload.

8. Pre-Core Quiz

  1. Which failures does the report count? → Only Err values in the stream it consumes.
  2. Why is a breaker event counted separately? → It is a terminal policy value with a stable BREAK/... code.
  3. What does max_samples bound? → Retained examples per group, not group cardinality.
  4. Why avoid input-derived error codes? → They create unstable semantics and potentially unbounded group counts.
  5. What did field-by-field breaker serialization preserve? → The nested last error and its immutable retry context without deep-copying it.

9. Post-Core Exercise

Extend the cumulative route with a second permanent error code.

  • Predict total_items, total_errs, and each code count.
  • Keep max_samples=1 and prove counts exceed retained samples.
  • Serialize the report and inspect the nested breaker last_error.
  • Add a path deeper than path_depth and explain which provenance remains in the group key versus the sample.
  • State which failures the report cannot see because they recovered or were never attempted.

Move forward when you can explain both the evidence present in the report and the evidence outside its observation boundary.

You have completed Module 04.

You can now review a deterministic local RAG stream in terms of stack use, bounded demand, expected failure values, aggregation policy, early termination, resource ownership, retry scheduling, and retained evidence. Those contracts are foundations for production decisions; they are not proof of production scale, external-service behavior, or zero failure.

Use the module exercises and refactoring guide to review the entire route before continuing to Module 05's algebraic data modelling.