fix: validate interrupt request payloads

src/wf_core/runtime/ops/interrupts.py

@@ -20,6 +20,7 @@ from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace
 from wf_core.runtime.ops.index import WorkflowIndex
 from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.overlays import state_view_for_frame
+from wf_core.runtime.ops.schemas import validate_payload_against_schema
 from wf_core.runtime.ops.state import build_output_patch
 
 
@@ -53,6 +54,11 @@ def build_interrupt_request(
             set_local_value(payload, binding.target, value)
         except LocalPathError as exc:
             raise WorkflowExecutionError(str(exc)) from exc
+    validate_payload_against_schema(
+        node.request_schema,
+        payload,
+        f"interrupt request for {node.id}",
+    )
     return InterruptRequest(
         id=f"interrupt:{public_node_id or node.id}",
         frame_id=public_frame_id or frame_id,

tests/core/test_execution_results.py

@@ -20,6 +20,7 @@ from wf_core import (
     resume_workflow_async,
     resume_workflow_result_async,
 )
+from wf_core.models.steps import InputValueBinding
 
 
 async def explode(_payload: dict[str, Any], _context: RuntimeContext) -> dict[str, Any]:
@@ -120,3 +121,38 @@ def _interrupt_then_fail_workflow() -> Workflow:
 
 def _schema() -> SchemaRef:
     return SchemaRef(type="object", properties={})
+
+
+async def test_interrupt_request_payload_validates_against_schema() -> None:
+    workflow = Workflow(
+        name="bad_interrupt_request",
+        input_schema=_schema(),
+        state_schema=StateSchema.from_field_map({}),
+        output_schema=_schema(),
+        outcomes=["ok"],
+        start="ask",
+        nodes=[
+            InterruptNode(
+                id="ask",
+                type="interrupt",
+                kind="approval",
+                request=[
+                    InputValueBinding(target="count", value="not a number")  # type: ignore[arg-type]
+                ],
+                request_schema={
+                    "type": "object",
+                    "properties": {"count": {"type": "number"}},
+                    "required": ["count"],
+                    "additionalProperties": False,
+                },
+            ),
+        ],
+        edges=[Edge.model_validate({"from": "ask", "outcome": "submitted", "to": END})],
+    )
+
+    run = await execute_workflow_result_async(workflow, {}, {})
+
+    assert run.status == RunStatus.FAILED
+    assert run.interrupt is None
+    assert run.error is not None
+    assert "interrupt request for ask" in run.error