feat: add per-stage error isolation in pipeline.py run_pipeline() #36

New issue

Closed

opened 2026-05-25 18:58:34 -07:00 by pyr0ball · 0 comments

pyr0ball commented

2026-05-25 18:58:34 -07:00

Owner

Context

If any pipeline stage raises an unhandled exception, asyncio.to_thread() propagates it up and the SSE stream closes silently — the user sees a broken stream with no error event.

Each stage has good internal fallbacks, but a top-level guard in run_pipeline() would protect against code bugs:

try:
    timeline = await asyncio.to_thread(TimelineReconstructor().reconstruct, entries)
except Exception as exc:
    logger.exception("Stage 1 (timeline) failed — aborting pipeline")
    yield {"type": "error", "message": "Pipeline error in stage 1"}
    yield {"type": "done"}
    return

Repeat for each stage. The {"type": "error"} event type is already handled by the Vue frontend.

Found by

Post-implementation code review of feat/29-multi-agent-diagnose.

## Context If any pipeline stage raises an unhandled exception, `asyncio.to_thread()` propagates it up and the SSE stream closes silently — the user sees a broken stream with no error event. Each stage has good internal fallbacks, but a top-level guard in `run_pipeline()` would protect against code bugs: ```python try: timeline = await asyncio.to_thread(TimelineReconstructor().reconstruct, entries) except Exception as exc: logger.exception("Stage 1 (timeline) failed — aborting pipeline") yield {"type": "error", "message": "Pipeline error in stage 1"} yield {"type": "done"} return ``` Repeat for each stage. The `{"type": "error"}` event type is already handled by the Vue frontend. ## Found by Post-implementation code review of feat/29-multi-agent-diagnose.