reducer as wf_mcp capability

docs/core_state_mapping_and_merge.md

@@ -178,7 +178,9 @@ Existing built-in reducers remain distinct:
 
 - `wf.std.replace`
 - `wf.std.append`
+- `wf.std.max`
 - `wf.std.merge_object`
+- `wf.std.set_union`
 
 `wf.std.merge_object` means shallow object merge at the exact destination path, similar
 to `dict.update` or `operator.or_`. It is not a recursive deep merge.
@@ -211,7 +213,6 @@ and belongs in nodes or graph structure.
 Examples a future reducer library could support:
 
 - `max`
-- `set_union`
 - `modulo_add` with configuration such as modulus `10`
 
 ## Implementation Phases

docs/wf_mcp_capability_sources.md

@@ -61,7 +61,8 @@ Expected capabilities:
   `wf.std.truthy`, `wf.std.first_item`, `wf.std.first_item_maybe`,
   `wf.std.first_item_or_none`, `wf.std.last_item`, `wf.std.last_item_or_none`,
   `wf.std.length`, `wf.std.is_empty`.
-- `reducers`: `wf.std.replace`, `wf.std.append`, `wf.std.merge_object`.
+- `reducers`: `wf.std.replace`, `wf.std.append`, `wf.std.max`,
+  `wf.std.merge_object`, `wf.std.set_union`.
 - `prompts`: workflow authoring guide, error-handling guide, mapping guide.
 - `resources`: reference docs for stdlib node behavior.
 

docs/workflow_artifacts.md

@@ -498,6 +498,8 @@ Saved workflow execution eventually needs first-class runtime support for:
 - resume into child run state
 - child final outcome mapping to parent node outcome
 - dependency checks before execution
+- reducer capabilities referenced by declared workflow state fields are saved as
+  direct artifact dependencies just like node specs or tools
 
 The first implementation should prefer artifact validation and dependency
 diagnostics before attempting persistent nested resume.

src/wf_artifacts/factory.py

@@ -31,7 +31,10 @@ def create_workflow_artifact_from_plan(
         output_schema=_required_object_field(plan, "output_schema"),
         outcomes=outcomes,
         plan=plan,
-        required_capabilities=dict(required_capabilities or {}),
+        required_capabilities={
+            **_required_reducers_from_plan(plan),
+            **dict(required_capabilities or {}),
+        },
         created_from_catalog_version=created_from_catalog_version,
     )
 
@@ -68,3 +71,28 @@ def _validate_workflow_plan(plan: JsonObject) -> None:
             raise ValueError(
                 f"invalid workflow plan: edge destination {edge.to!r} does not exist"
             )
+
+
+def _required_reducers_from_plan(plan: JsonObject) -> dict[str, RequiredCapability]:
+    """Infer reducer dependencies from declared state fields in one plan."""
+    state_schema = plan.get("state_schema")
+    if not isinstance(state_schema, dict):
+        return {}
+    fields = state_schema.get("fields")
+    if not isinstance(fields, dict):
+        return {}
+
+    requirements: dict[str, RequiredCapability] = {}
+    for field in fields.values():
+        if not isinstance(field, dict):
+            continue
+        reducer = field.get("reducer", "wf.std.replace")
+        if not isinstance(reducer, str) or "." not in reducer:
+            continue
+        logical_source, _, capability_name = reducer.rpartition(".")
+        requirements[reducer] = RequiredCapability(
+            logical_source=logical_source,
+            capability_name=capability_name,
+            kind="reducer",
+        )
+    return requirements

src/wf_artifacts/models.py

@@ -29,7 +29,7 @@ class RequiredCapability(BaseModel):
 
     logical_source: str
     capability_name: str
-    kind: Literal["tool", "resource", "prompt", "node_spec", "workflow"]
+    kind: Literal["tool", "resource", "prompt", "node_spec", "reducer", "workflow"]
     input_schema_hash: str | None = None
     input_schema_snapshot: JsonObject | None = None
     output_schema_hash: str | None = None
@@ -42,7 +42,7 @@ class AvailableCapability(BaseModel):
     """Current contract for one capability exposed by a bound source."""
 
     name: str
-    kind: Literal["tool", "resource", "prompt", "node_spec", "workflow"]
+    kind: Literal["tool", "resource", "prompt", "node_spec", "reducer", "workflow"]
     input_schema_hash: str | None = None
     output_schema_hash: str | None = None
 

src/wf_authoring/schemas.py

@@ -83,7 +83,9 @@ def _iter_model_metadata(
             yield from _iter_model_metadata(annotation, prefix=path)
 
 
-def _flatten_state_properties(schema: SchemaRef) -> Iterator[tuple[str, dict[str, Any]]]:
+def _flatten_state_properties(
+    schema: SchemaRef,
+) -> Iterator[tuple[str, dict[str, Any]]]:
     raw_schema = schema.model_dump(exclude_none=True)
     yield from _iter_state_properties(raw_schema.get("properties", {}), raw_schema)
 
@@ -143,7 +145,11 @@ def _state_field_default(
     field_name: str,
     property_schema: object,
 ) -> object:
-    if "." not in field_name and isinstance(value, type) and issubclass(value, BaseModel):
+    if (
+        "." not in field_name
+        and isinstance(value, type)
+        and issubclass(value, BaseModel)
+    ):
         field_info = value.model_fields[field_name]
         if not field_info.is_required():
             return field_info.get_default(call_default_factory=True)

src/wf_core/runtime/ops/merges.py

@@ -16,7 +16,9 @@ def replace_reducer(_current_value: Any, incoming_value: Any) -> Any:
 def append_reducer(current_value: Any, incoming_value: Any) -> Any:
     """Append one value or many values into a list-valued state path."""
     if current_value is None:
-        return [incoming_value] if not isinstance(incoming_value, list) else incoming_value
+        return (
+            [incoming_value] if not isinstance(incoming_value, list) else incoming_value
+        )
     if not isinstance(current_value, list):
         raise TypeError("cannot append into non-list state value")
     return (
@@ -37,10 +39,38 @@ def merge_object_reducer(current_value: Any, incoming_value: Any) -> Any:
     return current_value | incoming_value
 
 
+def set_union_reducer(current_value: Any, incoming_value: Any) -> Any:
+    """Merge list values while preserving stable first-seen order."""
+    if current_value is None:
+        current_items: list[Any] = []
+    elif isinstance(current_value, list):
+        current_items = current_value
+    else:
+        raise TypeError("set_union requires list values")
+
+    if not isinstance(incoming_value, list):
+        raise TypeError("set_union requires list values")
+
+    merged: list[Any] = []
+    for item in [*current_items, *incoming_value]:
+        if item not in merged:
+            merged.append(item)
+    return merged
+
+
+def max_reducer(current_value: Any, incoming_value: Any) -> Any:
+    """Keep the larger of the current and incoming values."""
+    return (
+        incoming_value if current_value is None else max(current_value, incoming_value)
+    )
+
+
 DEFAULT_REDUCERS: Mapping[str, Reducer] = {
     "wf.std.replace": replace_reducer,
     "wf.std.append": append_reducer,
     "wf.std.merge_object": merge_object_reducer,
+    "wf.std.set_union": set_union_reducer,
+    "wf.std.max": max_reducer,
 }
 
 

src/wf_mcp/broker/service/builtins.py

@@ -125,6 +125,14 @@ def builtin_reducers() -> dict[str, ReducerSpec]:
             name="wf.std.merge_object",
             description="Shallow-merge object values at one exact state path.",
         ),
+        ReducerSpec(
+            name="wf.std.max",
+            description="Keep the larger of the current and incoming values.",
+        ),
+        ReducerSpec(
+            name="wf.std.set_union",
+            description="Merge list values while preserving stable first-seen order.",
+        ),
     )
     return {spec.name: spec for spec in specs}
 

src/wf_mcp/workflow_surface/handlers.py

@@ -292,6 +292,15 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
             )
             for spec in source.capabilities.node_specs.values()
         }
+        capabilities.update(
+            {
+                reducer.name.rsplit(".", maxsplit=1)[-1]: AvailableCapability(
+                    name=reducer.name.rsplit(".", maxsplit=1)[-1],
+                    kind="reducer",
+                )
+                for reducer in source.capabilities.reducers.values()
+            }
+        )
         sources.append(
             AvailableSource(
                 id=source.id,

tests/artifacts/test_factory.py

@@ -33,6 +33,26 @@ def test_create_workflow_artifact_from_plan_derives_boundary_schemas() -> None:
     assert artifact.created_from_catalog_version == "catalog-1"
 
 
+def test_create_workflow_artifact_from_plan_adds_reducer_dependencies() -> None:
+    plan = _plan()
+    plan["state_schema"] = {
+        "fields": {"best_score": {"type": "integer", "reducer": "wf.std.max"}}
+    }
+
+    artifact = create_workflow_artifact_from_plan(
+        artifact_id="score",
+        version=1,
+        title="Score",
+        plan=plan,
+        outcomes=("done",),
+    )
+
+    reducer = artifact.required_capabilities["wf.std.max"]
+    assert reducer.logical_source == "wf.std"
+    assert reducer.capability_name == "max"
+    assert reducer.kind == "reducer"
+
+
 def test_create_workflow_artifact_from_plan_accepts_wrapper_kind() -> None:
     artifact = create_workflow_artifact_from_plan(
         artifact_id="normalize_status",

tests/artifacts/test_validation.py

@@ -176,3 +176,28 @@ def test_validate_deployment_allows_changed_schema_when_policy_allows() -> None:
     )
 
     assert diagnostics == []
+
+
+def test_validate_deployment_accepts_reducer_capability() -> None:
+    reducer = RequiredCapability(
+        logical_source="wf.std",
+        capability_name="set_union",
+        kind="reducer",
+    )
+    diagnostics = validate_deployment_dependencies(
+        artifact=artifact_with(reducer),
+        deployment=deployment(bindings={"wf.std": "wf.std"}),
+        sources=[
+            AvailableSource(
+                id="wf.std",
+                capabilities={
+                    "set_union": AvailableCapability(
+                        name="set_union",
+                        kind="reducer",
+                    )
+                },
+            )
+        ],
+    )
+
+    assert diagnostics == []

tests/core/test_nested_state_paths.py

@@ -40,7 +40,9 @@ def test_state_field_defaults_to_replace_reducer() -> None:
 
 
 def test_unknown_state_reducer_fails_clearly() -> None:
-    workflow = _workflow(fields={"person.tags": StateField(type="array", reducer="x.nope")})
+    workflow = _workflow(
+        fields={"person.tags": StateField(type="array", reducer="x.nope")}
+    )
     state = {"person": {"tags": ["seed"]}}
 
     try:
@@ -51,6 +53,28 @@ def test_unknown_state_reducer_fails_clearly() -> None:
         raise AssertionError("expected unknown reducer to fail")
 
 
+def test_set_union_reducer_preserves_first_seen_order() -> None:
+    workflow = _workflow(
+        fields={"person.tags": StateField(type="array", reducer="wf.std.set_union")}
+    )
+    state = {"person": {"tags": ["alpha", "beta"]}}
+
+    write_state_value(workflow, state, "state.person.tags", ["beta", "gamma"])
+
+    assert state["person"]["tags"] == ["alpha", "beta", "gamma"]
+
+
+def test_max_reducer_keeps_larger_value() -> None:
+    workflow = _workflow(
+        fields={"best_score": StateField(type="integer", reducer="wf.std.max")}
+    )
+    state = {"best_score": 7}
+
+    write_state_value(workflow, state, "state.best_score", 9)
+
+    assert state["best_score"] == 9
+
+
 def _workflow(*, fields: dict[str, StateField]) -> Workflow:
     return Workflow(
         name="nested_state_paths",

tests/rewrite/models.py

@@ -125,9 +125,7 @@ class PartialRates(SophisticatedRates, total=False):
 
 
 class Rates(BaseModel):
-    rates: Annotated[
+    rates: Annotated[PartialRates, state_field(reducer="wf.std.merge_object")]  # or_!
-        PartialRates, state_field(reducer="wf.std.merge_object")
-    ]  # or_!
 
 
 class CurrentPools(BaseModel):

tests/wf_mcp/test_service.py

@@ -113,11 +113,13 @@ def test_service_lists_all_capability_sources_with_owned_capability_names() -> N
     assert "wf.std.runtime_error" in std_source["capabilities"]["node_specs"]
     assert std_source["capabilities"]["reducers"] == [
         "wf.std.append",
+        "wf.std.max",
         "wf.std.merge_object",
         "wf.std.replace",
+        "wf.std.set_union",
     ]
     assert std_source["capabilities"]["tools"] == []
-    assert std_source["reducer_count"] == 3
+    assert std_source["reducer_count"] == 5
 
     mcp_source = sources_by_id["wf.mcp"]
     assert mcp_source["capabilities"]["node_specs"] == ["wf.mcp.call_tool"]
@@ -159,7 +161,9 @@ def test_wf_std_source_contains_builtin_reducers() -> None:
     assert set(reducers) == {
         "wf.std.replace",
         "wf.std.append",
+        "wf.std.max",
         "wf.std.merge_object",
+        "wf.std.set_union",
     }