feat(pipeline): state as optional overlay on PipelineEngine (#245 layer 3)

[miguelgfierro]

Third layer of the unification (#245). `PipelineEngine` now accepts optional `state_schema=` and `state=` arguments. Stacked on layer 2 (#248) which is already merged into `issue-147-pipeline-evolution`.

This layer reclaims parallelism for state-aware pipelines — the answer to the "state-mode is sequential" tradeoff discussed in #245. The existing topological scheduler now runs disjoint state-writing nodes concurrently, and the reducer machinery handles concurrent writes.

New surface

```python
class FactoryState(BaseModel):
spec: str
test_results: Annotated[list[TestRun], extend] = []
critique_history: Annotated[list[str], append] = []

engine = PipelineEngine(dag, state_schema=FactoryState)
result = await engine.run(state=FactoryState(spec="build a CRM"))
print(result.final_state.test_results)
```

Semantics

Node return	Without `state_schema`	With `state_schema`
`None`	No port output	No-op
`dict`	Port output (legacy)	State update via reducers
Other value	Port output (legacy)	Port output (legacy)

Both modes coexist on the same node — a single node can choose its return shape to either update state or emit a port value.

Parallelism

Concurrent writes are reconciled by the reducer declared on that field:

Reducer	Behavior under parallel writes
`replace` (default)	Last-write-wins; race-prone for replace.
`append`	Commutative; both writes land.
`extend`	Commutative; both writes land.
`merge_dict`	Commutative; both keys present.

In-flight tasks see the state snapshot they were scheduled with; updates apply sequentially in done-task order, so commutative reducers are race-free regardless of completion order.

Engine changes

• `PipelineEngine(init)` accepts `state_schema: type[BaseModel] | None`. Caches `discover_reducers(schema)` at construction.
• `PipelineEngine.run()` accepts `state: BaseModel | None`. When `state_schema` is set: validates `state`, auto-instantiates from defaults if `None`, exposes on `ctx.state`.
• After each successful node: if return is dict and `state_schema` is set, `apply_update` reduces it into `context.state`.
• `_save_checkpoint` persists `context.state` under `"shared_state"`.
• `_load_for_resume` restores `context.state` from `"shared_state"`.
• `PipelineResult` gets `final_state` field.
• `PipelineContext` gets `state: Any = None` attribute.

Refactor (no behavior change)

• `discover_reducers` and `apply_update` lifted from `state_pipeline.py` to `reducers.py` so both `StatePipeline` and `PipelineEngine` can share them without circular imports.
• `StatePipeline` now imports them from `reducers.py`. Behavior unchanged — verified by 37 existing state-pipeline tests.

Software-factory parallel-build example

```python
class FactoryState(BaseModel):
frontend_files: Annotated[dict[str, str], merge_dict] = {}
backend_files: Annotated[dict[str, str], merge_dict] = {}

dag = DAG("factory")
dag.add_node(DAGNode(node_id="architect", step=Architect()))
dag.add_node(DAGNode(node_id="frontend", step=FrontendDev()))
dag.add_node(DAGNode(node_id="backend", step=BackendDev()))
dag.add_edge(DAGEdge(source="architect", target="frontend")) # parallel
dag.add_edge(DAGEdge(source="architect", target="backend")) # parallel

engine = PipelineEngine(dag, state_schema=FactoryState)
```

frontend and backend run concurrently after architect; their state writes target disjoint fields (`frontend_files`, `backend_files`) and are merged with `merge_dict`. This is parallelism that the legacy `StatePipeline` could not give you.

Tests

12 new tests in `tests/unit/pipeline/test_pipeline_engine_state_overlay.py`:

• Baseline (no `state_schema` = unchanged)
• Auto-instantiation from defaults
• Explicit `state` arg
• Dict return as state update (replace, append, extend, merge_dict)
• Non-dict return is still a port output (state untouched)
• Edge condition reading `ctx.state`
• Parallel nodes writing same field via `append` (commutative-reducer parallelism)
• Resume restores shared state from checkpoint
• Audit log works under state overlay

Full suite: 1573 passed.

What's NOT in this PR

• Build-time validation of unsafe parallel writes (two nodes that could run at the same level writing the same `replace`-reduced field). Defer to a follow-up — needs a `.writes=` annotation on `DAGNode` or static analysis of return types.
• Migrating `StatePipeline` to use `PipelineEngine(state_schema=...)` internally. Layer 7.
• Removing `StatePipeline`. Layer 8.

Refs

• feat(pipeline): unify port-based and state-based pipelines into one DAG + engine #245 — unification design (layer 3 of 8)
• feat(pipeline): PipelineEngine checkpoint, audit, and resume (#245 layer 1) #246 / feat(pipeline): unified EventHandler protocol (#245 layer 1B) #247 — layer 1 (checkpoint/audit/resume + unified events)
• feat(pipeline): branching as DAGEdge.condition (#245 layer 2) #248 — layer 2 (DAGEdge.condition branching)
• feat(pipeline): state-based PipelineBuilder — full #147 stack (phase 1+2+3a+3b+3c) #232 — umbrella PR