@reducer is here
This commit is contained in:
@@ -0,0 +1,57 @@
|
|||||||
|
# Reducer Authoring Implementation Plan
|
||||||
|
|
||||||
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||||
|
|
||||||
|
**Goal:** Add Python authoring ergonomics for reducer definitions, including optional Pydantic config models.
|
||||||
|
|
||||||
|
**Architecture:** Keep `wf_core` reducer execution as the runtime layer. Add `wf_authoring.reducers` as the ergonomic layer that builds `ReducerDefinition` objects from Python functions, similar to how node authoring builds `NodeSpec`s. Config models are optional; when present, they generate `ReducerSpec.config_schema` and receive parsed config objects at call time.
|
||||||
|
|
||||||
|
**Tech Stack:** Python, Pydantic `BaseModel`, pytest, existing reducer runtime.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## File Structure
|
||||||
|
|
||||||
|
- Create `src/wf_authoring/reducers/`
|
||||||
|
- `callables.py`: reducer callable protocols
|
||||||
|
- `decorator.py`: `@reducer(...)`
|
||||||
|
- `catalog.py`: `ReducerCatalog`
|
||||||
|
- `__init__.py`: reducer authoring exports
|
||||||
|
- Modify `src/wf_authoring/__init__.py`
|
||||||
|
- export reducer authoring API
|
||||||
|
- Add `tests/authoring/test_reducers.py`
|
||||||
|
- plain reducer authoring
|
||||||
|
- configured reducer authoring with BaseModel config
|
||||||
|
- catalog specs/definitions
|
||||||
|
|
||||||
|
## Tasks
|
||||||
|
|
||||||
|
### Task 1: Pin Authoring API
|
||||||
|
|
||||||
|
- [ ] Add tests proving:
|
||||||
|
- `@reducer(name="wf.std.add")` wraps a two-arg callable
|
||||||
|
- `@reducer(name="wf.std.modulo_add", config_model=ModuloConfig)` wraps a callable receiving parsed config
|
||||||
|
- config model JSON Schema becomes `ReducerSpec.config_schema`
|
||||||
|
- `ReducerCatalog.from_reducers(...)` exposes definitions and specs
|
||||||
|
- [ ] Run focused tests and confirm failure before implementation.
|
||||||
|
|
||||||
|
### Task 2: Implement Reducer Authoring
|
||||||
|
|
||||||
|
- [ ] Add typed callable protocols for plain/config reducers.
|
||||||
|
- [ ] Add a small wrapper object that owns a `ReducerDefinition`.
|
||||||
|
- [ ] Add `@reducer(...)` overloads for bare and configured reducers.
|
||||||
|
- [ ] Add `ReducerCatalog`.
|
||||||
|
- [ ] Export from `wf_authoring`.
|
||||||
|
|
||||||
|
### Task 3: Verify
|
||||||
|
|
||||||
|
- [ ] Run `uv run --with pytest pytest tests/authoring/test_reducers.py -q`
|
||||||
|
- [ ] Run `uv run --with pytest pytest tests/authoring -q`
|
||||||
|
- [ ] Run full suite and basedpyright.
|
||||||
|
|
||||||
|
## Non-Goals
|
||||||
|
|
||||||
|
- MCP tools for authoring reducers
|
||||||
|
- LLM-authored reducer code
|
||||||
|
- external reducer packages
|
||||||
|
- replacing current built-in registration in this pass
|
||||||
@@ -59,12 +59,14 @@ from .nodes import (
|
|||||||
node,
|
node,
|
||||||
outcome,
|
outcome,
|
||||||
)
|
)
|
||||||
|
from .reducers import AuthoredReducer, ReducerCatalog, reducer
|
||||||
from .schemas import StateFieldMetadata, state_field
|
from .schemas import StateFieldMetadata, state_field
|
||||||
from .subgraph import subgraph_node
|
from .subgraph import subgraph_node
|
||||||
|
|
||||||
__all__ = [
|
__all__ = [
|
||||||
"NodeCatalog",
|
"NodeCatalog",
|
||||||
"NodeCatalogEntry",
|
"NodeCatalogEntry",
|
||||||
|
"AuthoredReducer",
|
||||||
"BoolOutput",
|
"BoolOutput",
|
||||||
"CoalesceInput",
|
"CoalesceInput",
|
||||||
"ConstantInput",
|
"ConstantInput",
|
||||||
@@ -76,6 +78,7 @@ __all__ = [
|
|||||||
"PickKeyInput",
|
"PickKeyInput",
|
||||||
"PickPathInput",
|
"PickPathInput",
|
||||||
"ProjectFieldsInput",
|
"ProjectFieldsInput",
|
||||||
|
"ReducerCatalog",
|
||||||
"RenameFieldsInput",
|
"RenameFieldsInput",
|
||||||
"RuntimeErrorInput",
|
"RuntimeErrorInput",
|
||||||
"NodeReturn",
|
"NodeReturn",
|
||||||
@@ -117,6 +120,7 @@ __all__ = [
|
|||||||
"runtime_error",
|
"runtime_error",
|
||||||
"node",
|
"node",
|
||||||
"outcome",
|
"outcome",
|
||||||
|
"reducer",
|
||||||
"state",
|
"state",
|
||||||
"state_field",
|
"state_field",
|
||||||
"state_path",
|
"state_path",
|
||||||
|
|||||||
@@ -0,0 +1,8 @@
|
|||||||
|
from .catalog import ReducerCatalog
|
||||||
|
from .decorator import AuthoredReducer, reducer
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"AuthoredReducer",
|
||||||
|
"ReducerCatalog",
|
||||||
|
"reducer",
|
||||||
|
]
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any, Protocol, TypeVar
|
||||||
|
|
||||||
|
from pydantic import BaseModel
|
||||||
|
|
||||||
|
ConfigT = TypeVar("ConfigT", bound=BaseModel)
|
||||||
|
|
||||||
|
|
||||||
|
class PlainReducerCallable(Protocol):
|
||||||
|
def __call__(self, current: Any, incoming: Any, /) -> Any: ...
|
||||||
|
|
||||||
|
|
||||||
|
class ConfigReducerCallable(Protocol[ConfigT]):
|
||||||
|
def __call__(self, current: Any, incoming: Any, config: ConfigT, /) -> Any: ...
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
from wf_core import ReducerSpec
|
||||||
|
from wf_core.runtime.ops.merges import ReducerDefinition
|
||||||
|
|
||||||
|
from .decorator import AuthoredReducer
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class ReducerCatalog:
|
||||||
|
"""Collection of authored reducers ready for runtime and inventory use."""
|
||||||
|
|
||||||
|
definitions: dict[str, ReducerDefinition]
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def from_reducers(cls, *reducers: AuthoredReducer) -> "ReducerCatalog":
|
||||||
|
return cls(
|
||||||
|
definitions={
|
||||||
|
reducer.definition.spec.name: reducer.definition
|
||||||
|
for reducer in reducers
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def specs(self) -> dict[str, ReducerSpec]:
|
||||||
|
return {
|
||||||
|
name: definition.spec for name, definition in self.definitions.items()
|
||||||
|
}
|
||||||
@@ -0,0 +1,110 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable, Mapping
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from typing import Any, TypeVar, cast, overload
|
||||||
|
|
||||||
|
from pydantic import BaseModel
|
||||||
|
|
||||||
|
from wf_core import ReducerSpec
|
||||||
|
from wf_core.runtime.ops.merges import ReducerDefinition
|
||||||
|
|
||||||
|
from .callables import ConfigReducerCallable, ConfigT, PlainReducerCallable
|
||||||
|
|
||||||
|
PlainFnT = TypeVar("PlainFnT", bound=PlainReducerCallable)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class AuthoredReducer:
|
||||||
|
"""Authoring wrapper for one reducer implementation."""
|
||||||
|
|
||||||
|
definition: ReducerDefinition
|
||||||
|
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def reducer(
|
||||||
|
fn: PlainFnT,
|
||||||
|
/,
|
||||||
|
*,
|
||||||
|
name: str | None = None,
|
||||||
|
description: str | None = None,
|
||||||
|
) -> AuthoredReducer: ...
|
||||||
|
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def reducer(
|
||||||
|
*,
|
||||||
|
name: str,
|
||||||
|
description: str | None = None,
|
||||||
|
) -> Callable[[Callable[..., Any]], AuthoredReducer]: ...
|
||||||
|
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def reducer(
|
||||||
|
*,
|
||||||
|
name: str,
|
||||||
|
config_model: type[ConfigT],
|
||||||
|
description: str | None = None,
|
||||||
|
) -> Callable[[Callable[..., Any]], AuthoredReducer]: ...
|
||||||
|
|
||||||
|
|
||||||
|
def reducer(
|
||||||
|
fn: Callable[..., Any] | None = None,
|
||||||
|
/,
|
||||||
|
*,
|
||||||
|
name: str | None = None,
|
||||||
|
config_model: type[BaseModel] | None = None,
|
||||||
|
description: str | None = None,
|
||||||
|
) -> AuthoredReducer | Callable[[Callable[..., Any]], AuthoredReducer]:
|
||||||
|
"""Wrap a Python reducer function as a runtime reducer definition."""
|
||||||
|
|
||||||
|
def decorate(raw: Callable[..., Any]) -> AuthoredReducer:
|
||||||
|
reducer_name = name or raw.__name__
|
||||||
|
reducer_description = description or raw.__doc__
|
||||||
|
if config_model is None:
|
||||||
|
return AuthoredReducer(
|
||||||
|
ReducerDefinition(
|
||||||
|
spec=ReducerSpec(
|
||||||
|
name=reducer_name,
|
||||||
|
description=_clean_doc(reducer_description),
|
||||||
|
),
|
||||||
|
fn=raw,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
model_type = config_model
|
||||||
|
|
||||||
|
def runtime_fn(
|
||||||
|
current: Any,
|
||||||
|
incoming: Any,
|
||||||
|
config: Mapping[str, Any],
|
||||||
|
) -> Any:
|
||||||
|
parsed = model_type.model_validate(config)
|
||||||
|
return cast(ConfigReducerCallable[BaseModel], cast(object, raw))(
|
||||||
|
current,
|
||||||
|
incoming,
|
||||||
|
parsed,
|
||||||
|
)
|
||||||
|
|
||||||
|
return AuthoredReducer(
|
||||||
|
ReducerDefinition(
|
||||||
|
spec=ReducerSpec(
|
||||||
|
name=reducer_name,
|
||||||
|
description=_clean_doc(reducer_description),
|
||||||
|
config_schema=config_model.model_json_schema(),
|
||||||
|
),
|
||||||
|
fn=runtime_fn,
|
||||||
|
accepts_config=True,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
if fn is not None:
|
||||||
|
return decorate(fn)
|
||||||
|
return decorate
|
||||||
|
|
||||||
|
|
||||||
|
def _clean_doc(doc: str | None) -> str | None:
|
||||||
|
if doc is None:
|
||||||
|
return None
|
||||||
|
cleaned = doc.strip()
|
||||||
|
return cleaned or None
|
||||||
@@ -0,0 +1,63 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from pydantic import BaseModel, Field
|
||||||
|
|
||||||
|
from wf_authoring import ReducerCatalog, reducer
|
||||||
|
from wf_core import ReducerRef
|
||||||
|
|
||||||
|
|
||||||
|
class ModuloConfig(BaseModel):
|
||||||
|
modulus: int = Field(gt=0)
|
||||||
|
|
||||||
|
|
||||||
|
@reducer(name="wf.std.add")
|
||||||
|
def add(current: int | None, incoming: int) -> int:
|
||||||
|
"""Add incoming values into integer state."""
|
||||||
|
return (current or 0) + incoming
|
||||||
|
|
||||||
|
|
||||||
|
@reducer(name="wf.std.modulo_add", config_model=ModuloConfig)
|
||||||
|
def modulo_add(
|
||||||
|
current: int | None,
|
||||||
|
incoming: int,
|
||||||
|
config: ModuloConfig,
|
||||||
|
) -> int:
|
||||||
|
"""Add incoming values modulo a configured positive integer."""
|
||||||
|
return ((current or 0) + incoming) % config.modulus
|
||||||
|
|
||||||
|
|
||||||
|
def test_reducer_decorator_wraps_plain_callable() -> None:
|
||||||
|
assert add.definition.spec.name == "wf.std.add"
|
||||||
|
assert add.definition.spec.description == "Add incoming values into integer state."
|
||||||
|
|
||||||
|
result = add.definition.apply(
|
||||||
|
reducer=ReducerRef(name="wf.std.add"),
|
||||||
|
current_value=2,
|
||||||
|
incoming_value=3,
|
||||||
|
destination_path="state.total",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result == 5
|
||||||
|
|
||||||
|
|
||||||
|
def test_reducer_decorator_wraps_configured_basemodel_callable() -> None:
|
||||||
|
schema = modulo_add.definition.spec.config_schema
|
||||||
|
|
||||||
|
assert schema["properties"]["modulus"]["exclusiveMinimum"] == 0
|
||||||
|
|
||||||
|
result = modulo_add.definition.apply(
|
||||||
|
reducer=ReducerRef(name="wf.std.modulo_add", config={"modulus": 10}),
|
||||||
|
current_value=8,
|
||||||
|
incoming_value=5,
|
||||||
|
destination_path="state.total",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_reducer_catalog_exposes_definitions_and_specs() -> None:
|
||||||
|
catalog = ReducerCatalog.from_reducers(add, modulo_add)
|
||||||
|
|
||||||
|
assert set(catalog.definitions) == {"wf.std.add", "wf.std.modulo_add"}
|
||||||
|
assert catalog.specs["wf.std.add"].name == "wf.std.add"
|
||||||
|
assert catalog.specs["wf.std.modulo_add"].name == "wf.std.modulo_add"
|
||||||
Reference in New Issue
Block a user