feat(pipeline): cycles, Send fan-out, Mermaid/JSON export (#147 phase 2)

[miguelgfierro]

Stacked on #232. Closes the remaining items from #147 left out of phase 1.

What's new

Cycles (ReAct / retry-with-critique). In state mode, the underlying DAG is now built with allow_cycles=True, so a node can route back to itself via the existing .branch(...) mechanism. A recursion_limit kwarg on PipelineBuilder (default 25) acts as the safety net — a runaway loop surfaces as result.success=False with a clear message, not an infinite hang.

async def step(state):
    return {"counter": state.counter + 1}

def route(state):
    return "done" if state.counter >= 3 else "step"

PipelineBuilder("loop", state=LoopState, recursion_limit=25)
    .add_node(step).add_node(done).branch(step, route).build()

Send(target, payload) runtime fan-out. Routers can return list[Send] to dispatch concurrent work over the same worker (or different workers). Each Send gets its own state copy with the payload merged in; the worker's return reduces back into shared state. Replaces the legacy FanOutStep pattern.

def dispatch(state):
    return [Send(\"worker\", {\"item\": x}) for x in state.items]

PipelineBuilder(\"mapreduce\", state=FanOutState)
    .add_node(planner).add_node(worker).add_node(collect)
    .add_edge(worker, collect)
    .branch(planner, dispatch)
    .build()

When all workers share a common successor (typical map-reduce shape), execution continues at that successor automatically. The aggregator runs once with all fan-out results already merged into state.

Mermaid + JSON export. DAG.to_mermaid() / DAG.to_json() for any DAG. StatePipeline.to_mermaid() adds branch-edge labels from the registered mapping (`a -->|label| b`).

Soft-deprecation of BranchStep / FanOutStep. Construction emits a DeprecationWarning pointing at .branch(...) and Send(...). Existing pipelines keep working — removal is a separate decision, gated on internal callers migrating.

API additions

Exported from fireflyframework_agentic.pipeline:

• Send
• RecursionLimitError

PipelineBuilder(...) gains recursion_limit: int = 25.

Implementation notes

• The executor remains sequential at the source level. Fan-out runs the per-Send tasks via asyncio.gather, so concurrency is intra-fan-out only. That matches the typical agentic use case (one branch at a time, fan-out for map-reduce) without committing to full parallel-DAG semantics.
• Per-Send state isolation uses the same apply_update reducer machinery that ordinary node updates use — no special-casing.
• After fan-out, _common_successor looks for a single shared successor across all worker targets and continues from there. If workers have divergent successors, the run ends (an aggregator pattern with a shared successor is the supported shape; richer fan-out topologies land in a future phase).
• Resume across a fan-out boundary is explicitly rejected for now with a clear error — the checkpoint format would need to track in-flight Send tasks, which is phase-3 territory.

Tests

tests/unit/pipeline/test_state_pipeline_phase2.py (9 tests, all green):

• Cycle with exit router runs the loop 3× then terminates.
• recursion_limit=5 triggers a clean failure on an infinite-loop router.
• 3-way fan-out runs all workers concurrently and the reducer aggregates results before the collector node runs.
• Send to an unknown target surfaces as a failed result (not an exception).
• DAG.to_mermaid() renders nodes and edges.
• DAG.to_json() produces a round-trippable JSON document.
• StatePipeline.to_mermaid() labels branch edges with the mapping labels.
• BranchStep(...) emits DeprecationWarning pointing to .branch(...).
• FanOutStep(...) emits DeprecationWarning pointing to Send.

Suite-wide:

• pytest tests/unit/pipeline/ — 97 passed (88 phase 1 + 9 phase 2)
• pytest tests/unit/ — 1405 passed (no regressions)
• ruff check + ruff format --check clean
• pyright clean on touched modules

Stacking

Targets issue-147-pipeline-evolution (the phase-1 branch). After #232 merges to main, this PR's base will retarget to main automatically.

Out of scope (phase 3 / follow-up)

• Resume across a fan-out boundary (in-flight Send checkpointing).
• Streaming partial state (pipeline.stream()).
• Typed-edge Literal[...] validation on node return annotations.
• Redis / Postgres checkpointer implementations.
• Removal (not just soft-deprecation) of BranchStep / FanOutStep.