validate output of node fit state

docs/core_state_mapping_and_merge.md

@@ -147,7 +147,9 @@ A successful step should produce a logical state patch before state is mutated:
 2. ensure every required output path exists
 3. ensure destination state paths do not overlap
 4. prepare the complete write set
-5. commit the write set according to state merge rules
+5. stage the write set on a copy
+6. validate affected declared state schemas
+7. commit the write set according to state merge rules
 
 This preserves the existing “no partial state commit before success” rule and
 creates a reusable boundary for:
@@ -157,6 +159,12 @@ creates a reusable boundary for:
 - future parallel foreach result combination
 - future subgraph completion
 
+State validation happens against the staged state before the original state is
+mutated. Runtime validates the exact destination schema when declared, declared
+ancestor schemas that could reject the write, and declared descendant schemas
+that exist after a parent replacement. This lets strict object schemas reject
+bad partial writes without committing half a patch.
+
 ## State Declarations and Merge Rules
 
 State merge behavior is attached to declared exact state paths. The canonical

src/wf_core/runtime/ops/state.py

@@ -18,6 +18,9 @@ from wf_core.paths import (
     split_graph_path,
 )
 from wf_core.runtime.ops.merges import ReducerDefinition, apply_reducer
+from wf_core.runtime.ops.schemas import validate_payload_against_schema
+
+_MISSING = object()
 
 
 def apply_output_map(
@@ -87,6 +90,7 @@ def apply_output_bindings(
     staged_state = deepcopy(state)
     for _destination_path, (key_path, merged_value) in prepared_patch.items():
         safe_set_nested_value(staged_state, key_path, merged_value)
+    validate_staged_state_patch(staged_state, prepared_patch, state_fields)
     state.clear()
     state.update(staged_state)
     return {str(path): value for path, value in resolved_patch.items()}
@@ -130,7 +134,15 @@ def write_state_value(
         value,
         reducers=reducers,
     )
-    safe_set_nested_value(state, key_path, merged_value)
+    staged_state = deepcopy(state)
+    safe_set_nested_value(staged_state, key_path, merged_value)
+    validate_staged_state_patch(
+        staged_state,
+        {StatePath.parse(destination_path): (key_path, merged_value)},
+        workflow.state_schema.field_map(),
+    )
+    state.clear()
+    state.update(staged_state)
 
 
 def prepare_state_value(
@@ -179,6 +191,64 @@ def project_output(workflow: Workflow, state: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def validate_staged_state_patch(
+    staged_state: dict[str, Any],
+    prepared_patch: Mapping[StatePath, tuple[list[str], Any]],
+    state_fields: Mapping[str, StateFieldDecl],
+) -> None:
+    """Validate affected declared state schemas before committing a patch.
+
+    Runtime writes are path-based, while JSON Schema is tree-based. A child
+    write can violate a declared parent schema, and a parent replacement can
+    violate declared child schemas. This helper validates every declared schema
+    that is on either side of a staged write path, without mutating the original
+    state first.
+    """
+    for field in _affected_state_fields(prepared_patch, state_fields):
+        value = _get_existing_nested_value(staged_state, list(field.path.parts))
+        if value is _MISSING:
+            continue
+        validate_payload_against_schema(
+            field.validation_schema,
+            value,
+            f"state write state.{'.'.join(field.path.parts)}",
+        )
+
+
+def _affected_state_fields(
+    prepared_patch: Mapping[StatePath, tuple[list[str], Any]],
+    state_fields: Mapping[str, StateFieldDecl],
+) -> list[StateFieldDecl]:
+    affected: dict[str, StateFieldDecl] = {}
+    for destination_path in prepared_patch:
+        destination_parts = destination_path.parts
+        for key, field in state_fields.items():
+            field_parts = field.path.parts
+            if _is_prefix(field_parts, destination_parts) or _is_prefix(
+                destination_parts,
+                field_parts,
+            ):
+                affected[key] = field
+    return sorted(
+        affected.values(),
+        key=lambda field: len(field.path.parts),
+        reverse=True,
+    )
+
+
+def _is_prefix(prefix: tuple[str, ...], value: tuple[str, ...]) -> bool:
+    return len(prefix) <= len(value) and value[: len(prefix)] == prefix
+
+
+def _get_existing_nested_value(state: Mapping[str, Any], path_parts: list[str]) -> Any:
+    current: Any = state
+    for part in path_parts:
+        if not isinstance(current, Mapping) or part not in current:
+            return _MISSING
+        current = current[part]
+    return current
+
+
 def safe_set_nested_value(
     state: dict[str, Any], path_parts: list[str], value: Any
 ) -> None:

tests/core/test_atomic_state_patches.py

@@ -109,6 +109,49 @@ def test_output_bindings_commit_to_staged_state_before_mutating_original() -> No
     assert state["blocked"] == "not-an-object"
 
 
+def test_output_bindings_validate_exact_state_schema_before_mutation() -> None:
+    workflow = _workflow(fields={"person.name": StateField(type="string")})
+    state = {"person": {"name": "old"}}
+
+    with pytest.raises(WorkflowExecutionError, match="state write state.person.name"):
+        apply_output_bindings(
+            workflow,
+            [_binding("person.name", "state.person.name")],
+            {"person": {"name": 7}},
+            state,
+        )
+
+    assert state["person"]["name"] == "old"
+
+
+def test_output_bindings_validate_declared_parent_schema_before_mutation() -> None:
+    workflow = _workflow_from_state_schema(
+        StateSchema.model_validate(
+            {
+                "type": "object",
+                "properties": {
+                    "person": {
+                        "type": "object",
+                        "properties": {"name": {"type": "string"}},
+                        "additionalProperties": False,
+                    }
+                },
+            }
+        )
+    )
+    state = {"person": {"name": "old"}}
+
+    with pytest.raises(WorkflowExecutionError, match="state write state.person"):
+        apply_output_bindings(
+            workflow,
+            [_binding("person.extra", "state.person.extra")],
+            {"person": {"extra": "bad"}},
+            state,
+        )
+
+    assert state["person"] == {"name": "old"}
+
+
 def test_full_workflow_execution_writes_canonical_output_bindings() -> None:
     workflow = _workflow_with_node()
     run = create_run_state(workflow, {})
@@ -135,17 +178,23 @@ def _binding(source: str, target: str) -> OutputBinding:
 def _workflow(
     fields: dict[str, StateField] | None = None,
 ) -> Workflow:
-    return Workflow(
-        name="patch",
-        input_schema=SchemaRef(type="object", properties={}),
-        state_schema=StateSchema.from_field_map(
+    return _workflow_from_state_schema(
+        StateSchema.from_field_map(
             fields
             or {
                 "person": StateField(type="object"),
                 "person.name": StateField(type="string"),
                 "person.extra": StateField(type="string"),
             }
-        ),
+        )
+    )
+
+
+def _workflow_from_state_schema(state_schema: StateSchema) -> Workflow:
+    return Workflow(
+        name="patch",
+        input_schema=SchemaRef(type="object", properties={}),
+        state_schema=state_schema,
         output_schema=SchemaRef(type="object", properties={}),
         start="n",
         nodes=[],