reducer replaces merge_strategy.

src/wf_core/models/schemas.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Literal
+from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -20,7 +20,7 @@ class StateField(BaseModel):
     """Declared state path plus its runtime merge behavior."""
 
     type: str
-    merge_strategy: Literal["replace", "append", "merge_object"] = "replace"
+    reducer: str = "wf.std.replace"
     trace: bool = True
     default: Any = None
 

src/wf_core/runtime/ops/merges.py

@@ -1,57 +1,62 @@
 from __future__ import annotations
 
+from collections.abc import Callable, Mapping
 from typing import Any
 
 from wf_core.errors import WorkflowExecutionError
 
+Reducer = Callable[[Any, Any], Any]
 
-def apply_builtin_merge(
+
+def replace_reducer(_current_value: Any, incoming_value: Any) -> Any:
+    """Replace the current state value with the incoming value."""
+    return incoming_value
+
+
+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
+    if not isinstance(current_value, list):
+        raise TypeError("cannot append into non-list state value")
+    return (
+        [*current_value, *incoming_value]
+        if isinstance(incoming_value, list)
+        else [*current_value, incoming_value]
+    )
+
+
+def merge_object_reducer(current_value: Any, incoming_value: Any) -> Any:
+    """Shallow-merge object values at one exact state path."""
+    if current_value is None:
+        if not isinstance(incoming_value, dict):
+            raise TypeError("cannot merge non-object value")
+        return dict(incoming_value)
+    if not isinstance(current_value, dict) or not isinstance(incoming_value, dict):
+        raise TypeError("merge_object requires dict values")
+    return 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,
+}
+
+
+def apply_reducer(
     *,
-    strategy: str,
+    reducer_name: str,
     current_value: Any,
     incoming_value: Any,
     destination_path: str,
+    reducers: Mapping[str, Reducer] = DEFAULT_REDUCERS,
 ) -> Any:
-    """Apply one built-in merge rule.
-
-    This is the future seam for source-owned reducer libraries. The current core
-    still supports only built-in rules and keeps them pure over current and
-    incoming values.
-    """
-    if strategy == "replace":
-        return incoming_value
-
-    if strategy == "append":
-        if current_value is None:
-            return (
-                [incoming_value]
-                if not isinstance(incoming_value, list)
-                else incoming_value
-            )
-        if not isinstance(current_value, list):
-            raise WorkflowExecutionError(
-                f"cannot append into non-list state path {destination_path!r}"
-            )
-        return (
-            [
-                *current_value,
-                *incoming_value,
-            ]
-            if isinstance(incoming_value, list)
-            else [*current_value, incoming_value]
-        )
-
-    if strategy == "merge_object":
-        if current_value is None:
-            if not isinstance(incoming_value, dict):
-                raise WorkflowExecutionError(
-                    f"cannot merge non-object value into {destination_path!r}"
-                )
-            return dict(incoming_value)
-        if not isinstance(current_value, dict) or not isinstance(incoming_value, dict):
-            raise WorkflowExecutionError(
-                f"merge_object requires dict values at {destination_path!r}"
-            )
-        return current_value | incoming_value
-
-    raise WorkflowExecutionError(f"unknown merge strategy {strategy!r}")
+    """Apply one named pure reducer to a state write."""
+    reducer = reducers.get(reducer_name)
+    if reducer is None:
+        raise WorkflowExecutionError(f"unknown reducer {reducer_name!r}")
+    try:
+        return reducer(current_value, incoming_value)
+    except TypeError as exc:
+        raise WorkflowExecutionError(f"{exc} at {destination_path!r}") from exc

src/wf_core/runtime/ops/state.py

@@ -12,7 +12,7 @@ from wf_core.paths import (
     set_nested_value,
     split_graph_path,
 )
-from wf_core.runtime.ops.merges import apply_builtin_merge
+from wf_core.runtime.ops.merges import apply_reducer
 
 
 def apply_output_map(
@@ -73,11 +73,11 @@ def write_state_value(
 
     declared_path = ".".join(parts)
     declared_field = workflow.state_schema.fields.get(declared_path)
-    merge_strategy = declared_field.merge_strategy if declared_field else "replace"
+    reducer_name = declared_field.reducer if declared_field else "wf.std.replace"
     key_path = parts
     current_value = get_nested_value(state, key_path)
-    merged_value = apply_builtin_merge(
-        strategy=merge_strategy,
+    merged_value = apply_reducer(
+        reducer_name=reducer_name,
         current_value=current_value,
         incoming_value=value,
         destination_path=destination_path,