feat(pipeline): state-based PipelineBuilder — full #147 stack (phase 1+2+3a+3b+3c)

[miguelgfierro]

Closes #147 and #245. Follow-ups: #237 / #238 / #239 / #257.

This PR extends the existing PipelineEngine (the port-based DAG that was already in the repo) into a single engine that also handles agentic workflows. Nothing was thrown away: every previous capability is preserved, and a set of new capabilities is built on top of the same DAG, the same scheduler, and the same PipelineEngine class.

---

What we already had

The repo shipped a port-based DAG runtime:

• PipelineEngine executed a DAG made of DAGNode + DAGEdge objects.
• Nodes were StepExecutor objects (CallableStep, BranchStep, FanInStep, FanOutStep) connected by typed ports — each node's outputs flowed into the named inputs of downstream nodes.
• The scheduler was topological: nodes at the same level ran in parallel via asyncio.gather.
• This worked great for ETL/batch-shaped pipelines where data flows cleanly A → B → C, and where parallelism is the main concern.

What it could not express was anything that needed shared mutable state, cycles, runtime fan-out over a dynamic work list, human-in-the-loop pauses, or durable resume across crashes. Those are exactly the shapes an agentic / IDP / "software factory" workflow needs.

---

What we built on top of the DAG

Every new feature is an additive extension of the same engine. The port-based mode keeps working unchanged; activating the new features is opt-in.

1. A shared state object that travels with the DAG

PipelineBuilder("name", state=SomeModel) attaches a Pydantic state schema to the pipeline. Each node receives the current state, returns a dict update, and the engine merges the update into the live state object via per-field reducers:

class BuildState(BaseModel):
    artifacts: Annotated[list[str], extend] = []   # append-merge
    summary:   Annotated[str, replace] = ""        # last-write-wins
    decisions: Annotated[dict[str, Any], merge_dict] = {}

Reducers ship in pipeline/reducers.py: replace, append, extend, merge_dict. Custom ones are just functions.

State sits alongside ports — when state mode is on, the engine still schedules topologically and still passes ports through. State is an overlay, not a replacement.

2. Branching as a first-class builder call

The original way to branch was BranchStep(...) plus per-edge condition lambdas. That still works, but it was verbose. The new shape is one call on the builder:

pipeline = (
    PipelineBuilder("review", state=BuildState)
        .add_node(qa)
        .add_node(release)
        .add_node(codegen)
        .add_edge("builder", "qa")
        .branch("qa", lambda s: "release" if s.qa_status == "pass" else "codegen")
        .build()
)

Internally .branch() compiles the router function into DAGEdge.condition predicates, so the scheduler stays unchanged — it just skips edges whose predicate is false.

3. Cycles in the DAG

The original topological scheduler refused cycles. We added a cycle-aware scheduler that runs alongside the topological one and engages only when the DAG actually has back-edges. A recursion_limit guards against runaway loops (default 25, configurable per build).

This is what lets us express ReAct loops, retry-with-critique, QA feedback loops, planner ↔ executor ping-pong, etc.

4. Send — runtime fan-out with per-worker state copies

A node may return Send(target, payload) or list[Send]. The engine then dispatches N workers concurrently — each gets its own state copy with the Send's payload merged in. When the workers finish, their state updates fold back into the shared state through the same reducers.

This is dynamic fan-out (the count is decided at runtime by the planner). The original FanOutStep was static — it just returned a list and let downstream topology do whatever it did. Send makes "spawn N parallel sub-agents for a dynamic work list" a first-class operation.

5. Pause — human-in-the-loop gates

A node may return Pause(reason="..."). The engine halts cleanly, writes a checkpoint with paused=True, and returns. Calling invoke(run_id=..., approve_pause=True) later resumes from the successor of the paused node (the pause node itself is not re-executed). Without approve_pause=True a paused run raises PipelineError — silent skips are not allowed.

6. Checkpointing + resume

A Checkpointer Protocol with three production backends:

• FileCheckpointer — JSONL on disk.
• RedisCheckpointer — SET+EX per checkpoint, ZADD/ZRANGE run-id index, configurable TTL.
• PostgresCheckpointer — single firefly_checkpoints table, idempotent DDL, INSERT … ON CONFLICT DO UPDATE.

Checkpoints are written automatically after each successful node. invoke(run_id="...") resumes from the last good checkpoint; invoke(state, start_at="node") jumps mid-pipeline for debugging or replay.

7. Audit log — separate from checkpoint

A Checkpointer answers "where do I resume from?". An AuditLog answers "what happened?". They are distinct concerns, so they live in distinct protocols and distinct files (pipeline/audit.py).

Each AuditEntry records inputs_snapshot, outputs_snapshot, latency_ms, started_at, completed_at, status (success / error / paused), error_message, pause_reason, and a visit counter. Four backends:

• FileAuditLog (JSONL, queryable)
• PostgresAuditLog (queryable)
• LoggingAuditLog (stdlib logging, plays nicely with Splunk / Loki / Datadog / OTel-LoggingHandler)
• OtelAuditLog (OTel logs API with trace_id / span_id correlation)

Audit-write failures are non-fatal — logged and swallowed. Compliance / debugging / replay don't get to break business logic.

8. A unified EventHandler protocol + OTel spans

One EventHandler Protocol now covers both port-based and state-based pipelines: on_pipeline_start / on_node_start / on_node_complete / on_node_error / on_node_pause / on_pipeline_complete. Every callback carries run_id; on_node_start carries a visit counter so cycles and Send fan-out are distinguishable in observability tools.

Per-pipeline and per-node OTel spans are emitted automatically. Partial handlers (only some methods defined) work via hasattr-checked dispatch. Handler exceptions are swallowed.

9. Mermaid + JSON export

DAG.to_mermaid() and DAG.to_json() render the graph including branch conditions ("if?" labels), which is helpful when looking at a non-trivial agentic pipeline you didn't write.

---

What new capabilities are added

Each shape below names the concrete extension on the engine that enables it, so it's clear which piece of the PR unlocks which capability.

• Software factory / IDP — architect → codegen → builder → qa, with a QA feedback cycle back to codegen capped at 3 iterations. Lives in examples/software_factory/.
  Enabled by: state overlay (so qa can read/write qa_status), .branch() (qa_router picks the next node), the cycle-aware scheduler (qa → codegen is a back-edge), and recursion_limit (caps the loop). The original DAG had none of these — it would have rejected the back-edge at build time.

• ReAct loops — planner ↔ executor with bounded recursion.
  Enabled by: the cycle-aware scheduler + recursion_limit. The original topological scheduler rejected cycles outright.

• Retry-with-critique — generator → critic → (if "redo") generator.
  Enabled by: .branch() driving a back-edge into the cycle-aware scheduler. Same primitive as ReAct; different shape.

• Map-reduce over a dynamic work list — planner returns list[Send(...)], the engine dispatches N workers concurrently each with its own state copy + payload, and reducers fold the results back into shared state.
  Enabled by: the Send sentinel + per-worker state copies + reducers. The original FanOutStep returned a static list and let downstream topology handle parallelism; it could not vary the worker count at runtime or hand each worker an isolated state slice.

• Human-in-the-loop approval gates — build → Pause("await approval") → deploy. The pause is checkpointed; the run resumes days later with invoke(run_id=..., approve_pause=True); every node visit (including the pause itself, with status="paused") lands in the audit log.
  Enabled by: the Pause sentinel, the Checkpointer Protocol carrying paused=True / pause_reason, and the AuditLog Protocol recording the gate visit. None of the three existed in the original DAG.

• Crash-survivable long-running pipelines — the engine auto-writes a checkpoint after every successful node; invoke(run_id=...) picks up from the last good checkpoint without re-running completed nodes.
  Enabled by: the Checkpointer Protocol + FileCheckpointer / RedisCheckpointer / PostgresCheckpointer backends + resume logic in the engine. Survival across crashes is bounded by the chosen backend's durability — Postgres is fully transactional; Redis depends on its persistence config (AOF/RDB); the file backend survives the process but not the disk.

• Compliance and post-mortem investigation — every node visit is recorded with input snapshot, output snapshot, latency, status, error or pause reason, visit counter, and (with OtelAuditLog) trace/span IDs. Queryable backends (FileAuditLog, PostgresAuditLog) let you walk the trail node-by-node after the fact.
  Enabled by: the AuditLog Protocol + its four backends. Note: the audit log captures what happened; it is not a one-call replay engine. To re-run a single node, combine the captured input snapshot with start_at= below.

• Mid-pipeline debug entry — invoke(state, start_at="failing_node") reproduces one node in isolation against a hand-supplied state (typically reconstructed from the audit log).
  Enabled by: the start_at= kwarg added in unification layer 6 (feat(pipeline): start_at kwarg for mid-pipeline entry (#245 layer 6) #252). The original DAG ran from the source nodes only.

---

How this PR landed

This integration PR is the merge of five phases plus an eight-layer unification, all originally cut against issue-147-pipeline-evolution:

Phases (built the new capabilities, initially as a parallel StatePipeline class):

Phase	PR	What it added
1	(original #232)	state-attached builder, Checkpointer Protocol, FileCheckpointer, branching as .branch(), reducers
2	#233	cycles, Send fan-out, Mermaid / JSON export
3a	#234	RedisCheckpointer + PostgresCheckpointer
3b	#235	EventHandler protocol + OTel spans
3c	#236	Pause (HITL) + four-backend AuditLog

Unification (#245, eight layers — folded StatePipeline into the existing PipelineEngine):

After phase 3c shipped, we had two engines side by side: the original PipelineEngine (parallel, port-based) and the new StatePipeline (sequential, state-based). They duplicated checkpointing, audit, events, and OTel. Issue #245 collapsed them in eight stacked PRs:

Layer	PR	What it did
1	#246	Move checkpoint, audit, and resume into PipelineEngine.
1B	#247	Single EventHandler protocol shared by both modes.
2	#248	Branching expressed as DAGEdge.condition.
3	#249	State as an optional overlay on PipelineEngine.
4	#250	Cycle-aware scheduler in PipelineEngine; topological sort raises on cycles instead of silently truncating.
5	#251	Pause and Send recognized by the unified engine.
6	#252	start_at= kwarg for mid-pipeline entry.
7	#253	StatePipeline deprecated (engine became feature-equivalent).
8	#255	Delete StatePipeline entirely (~750 LOC). PipelineBuilder.build() always returns PipelineEngine.

Companion fixes inside the branch: #254 (stricter ruff rules surfaced by the PR gate) and #256 (the software_factory example was still importing the deleted StatePipeline).

Earlier cleanup on the same branch: #240 (collapse invoke() to a single return path), #241 (share psycopg scaffolding between checkpoint.py and audit.py), #242 (drop dead variables), #243 (assert result narrowing), #244 (extract software_factory example, drop Postgres/Redis from default deps).

End result: one engine, one scheduler, one set of events, one checkpoint format. State-mode pipelines get topological parallelism for free when the graph permits and only fall back to the cyclic scheduler when there are actual cycles.

---

How to run the example

The canonical example is examples/software_factory/ — an IDP-style pipeline (architect → codegen → builder → qa → release) with a QA feedback loop, recursion_limit=3, a progress event handler, and FileCheckpointer. Plain-Python "agents", no API key required.

source ~/.venvs/firefly/bin/activate
python -m examples.software_factory

It exercises: state-based nodes, .branch(), cycles (qa → codegen on fail), event-handler progress output, and checkpoint/resume.

---

Extending the IDP example (examples/idp_pipeline.py)

The IDP example (ingest → split → classify → extract → validate → assemble → explain) is wired today as a plain linear .chain() of CallableStep nodes with per-node timeout_seconds and one retry_max=1. Per-document work is done with asyncio.gather() inside the steps. With this PR landed, almost all of that can be expressed as DAG structure instead. Natural extensions, highest-value first:

• Conditional edges to skip stages. Use DAGEdge(condition=...) so a low-confidence classify result routes to a review path, and unsupported sub-doc types skip extract/validate straight to assemble. Skipped nodes surface as on_node_skip events instead of vanishing inside Python branching.
• Send-based fan-out + reducers instead of in-step asyncio.gather. Rewrite extract to emit [Send("extract_one", {doc: d}) for d in subdocs] against a typed state schema (Annotated[list[Result], extend]). Each sub-document becomes a first-class node with its own trace span, retry, and timeout — real map/reduce instead of an opaque single node.
• Pause for human-in-the-loop. When grounding ratio / validation drops below threshold, return Pause(reason="needs human review"), checkpoint, and resume with approve_pause=True — replacing the implicit "validation failed → Reflexion" path with an explicit gate.
• Checkpoint / resume. Add a FileCheckpointer (or Postgres) so a crashed run resumes at the last completed node, or start_at="extract" to skip the expensive 300s stage on re-run.
• Typed state schema. Replace the loose raw_text / sub_documents / classifications facts stuffed into MemoryManager/metadata with a PipelineBuilder(state=IDPState) Pydantic model + reducers — type-checked data flow, and a prerequisite for the Send fan-out above.
• Per-node FailureStrategy. Push the current per-subdoc try/except into the DAG: failure_strategy=SKIP_DOWNSTREAM (or PROPAGATE) so a failed optional stage (e.g. explain) doesn't abort the run.

Net effect: the example moves from "proves a pipeline runs" to "demonstrates branching, map/reduce, HITL, and resumability" — the features that justify the DAG over a for loop.

---

Extending flydocs

flydocs's PipelineOrchestrator builds a fresh PipelineEngine per request, but still uses builder.chain(*chain) with optional stages conditionally appended in Python and per-task concurrency hand-rolled via asyncio.gather() inside each stage. The engine in this PR unlocks several things that previously required custom code or weren't possible:

• Optional stages as conditional edges. Instead of building a different chain list per request, make the DAG static and let DAGEdge(condition=...) decide which of discover / classify / judge / authenticity fire from request config — skipped nodes then emit on_node_skip events instead of being silently absent.
• Native per-task fan-out with Send + reducers. Today per-segment concurrency lives inside a stage, so traces only have per-stage granularity. Converting extract / judge / authenticity fan-out to Send worker nodes yields per-task spans, per-task retry, and per-task timeout — observability that wasn't achievable without restructuring.
• First-class HITL via Pause / resume. A "needs human confirmation" gate (low-confidence judge, failed authenticity) becomes a native pipeline pause + checkpoint instead of a bespoke worker + EDA event.
• Stage-level checkpoint / resume. flydocs's reaper currently recovers at job granularity (re-run the whole extraction). PostgresCheckpointer + start_at= lets a revived job resume at the last completed stage — e.g. skip the 600s extract that already finished. Directly complementary to the existing reliability hardening.
• judge_escalation as a real cycle. Replace the hardcoded "if judge fail-rate > threshold, re-run extract+judge with a stronger model" stage with allow_cycles=True + a router and recursion_limit: extract → judge → router → (escalate back to extract | exit when pass-rate met).
• Consolidate the rule-engine DAG (speculative). RuleEngine runs its own graphlib.TopologicalSorter to evaluate business rules level-by-level — a second orchestration layer that could move onto the engine's execution_levels() + concurrent scheduler, removing the duplication. Worth weighing against the simplicity of the current graphlib approach.

---

Verification

Local:

• pytest tests/unit/ → 1594 passed. Zero regressions in port-based DAG tests; the 37 state-pipeline tests were migrated onto PipelineEngine during layer 8.
• pytest tests/unit/pipeline/ → all green (port-based + state-based + checkpoint backends + observability + HITL + audit + unification layers).
• ruff check + ruff format --check clean.
• pyright clean on touched modules.

CI on this PR: all 8 checks green (Lint, Pre-commit, Test, Type Check, CodeQL).

---

What's NOT in this PR (tracked follow-ups)

• [#147 phase 3d] StatePipeline streaming: pipeline.stream() partial-state generator #237 — Streaming partial state via pipeline.stream(). Generator-based progress API for chat UIs / SSE / websockets. ~500 LOC executor refactor.
• [#147 phase 3d] Typed-edge validation: catch branch-mapping mistakes via Literal[...] return annotations #238 — Typed-edge validation via Literal[...] return annotations. Catches branch-mapping typos at .build() instead of first invocation. ~100 LOC.
• [#147 phase 3d] Migrate internal callers off BranchStep / FanOutStep to state-mode API #239 — Migrate internal callers off BranchStep / FanOutStep. Cleanup sweep across Signature repos. Those classes stay (explicit user direction) — they keep working, the migration is just to shake out usability issues in the new APIs.
• discussion: FanInStep / Send naming asymmetry post-#245 #257 — FanInStep / Send naming asymmetry. Discussion-only issue exploring whether to rename FanInStep → MergeStep when [#147 phase 3d] Migrate internal callers off BranchStep / FanOutStep to state-mode API #239 lands.

Also out of scope:

• Removal of BranchStep / FanOutStep. They stay (deprecation warning); explicit user direction.
• Real-service Redis / Postgres integration tests. Tests in this PR use unittest.mock only.
• Cross-fan-out resume. A run that pauses mid-fan-out can't be resumed; a clear error message says so.