feat: add interrupt schema builder helpers

src/wf_authoring/builder/core.py

@@ -603,6 +603,8 @@ class WorkflowBuilder:
         request: Sequence[InputBindingArg] | None = None,
         resume: Sequence[OutputBindingArg] | None = None,
         outcomes: list[str] | None = None,
+        request_schema: Mapping[str, Any] | None = None,
+        resume_schema: Mapping[str, Any] | None = None,
     ) -> InterruptNode: ...
 
     @overload
@@ -615,6 +617,8 @@ class WorkflowBuilder:
         request_map: MapArg | None = None,
         out_map: MapArg | None = None,
         outcomes: list[str] | None = None,
+        request_schema: Mapping[str, Any] | None = None,
+        resume_schema: Mapping[str, Any] | None = None,
     ) -> InterruptNode: ...
 
     def interrupt(
@@ -627,6 +631,8 @@ class WorkflowBuilder:
         request_map: MapArg | None = None,
         out_map: MapArg | None = None,
         outcomes: list[str] | None = None,
+        request_schema: Mapping[str, Any] | None = None,
+        resume_schema: Mapping[str, Any] | None = None,
     ) -> InterruptNode:
         if request is not None and request_map is not None:
             raise TypeError("cannot mix canonical request with deprecated request_map")
@@ -652,14 +658,19 @@ class WorkflowBuilder:
             if resume is not None
             else _canonical_output_bindings(normalize_output_mapping(out_map))
         )
-        node = InterruptNode(
+        payload: dict[str, Any] = {
-            id=id or self._next_step_id(f"interrupt_{slug_id(kind)}"),
+            "id": id or self._next_step_id(f"interrupt_{slug_id(kind)}"),
-            type="interrupt",
+            "type": "interrupt",
-            kind=kind,
+            "kind": kind,
-            request=request_bindings,
+            "request": request_bindings,
-            resume=resume_bindings,
+            "resume": resume_bindings,
-            outcomes=outcomes or ["submitted"],
+            "outcomes": outcomes or ["submitted"],
-        )
+        }
+        if request_schema is not None:
+            payload["request_schema"] = dict(request_schema)
+        if resume_schema is not None:
+            payload["resume_schema"] = dict(resume_schema)
+        node = InterruptNode.model_validate(payload)
         self.nodes.append(node)
         return node
 

tests/authoring/test_builder.py

@@ -299,6 +299,33 @@ def test_builder_interrupt_accepts_canonical_request_and_resume_bindings() -> No
     assert interrupt.resume[0].target == StatePath.of("text")
 
 
+def test_builder_interrupt_accepts_request_and_resume_schemas() -> None:
+    builder = WorkflowBuilder(
+        name="interrupt_contract",
+        input_schema={"type": "object", "properties": {}},
+        state_schema={"fields": {}},
+        output_schema={"type": "object", "properties": {}},
+    )
+
+    interrupt = builder.interrupt(
+        kind="approval",
+        request_schema={
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+            "required": ["message"],
+        },
+        resume_schema={
+            "type": "object",
+            "properties": {"approved": {"type": "boolean"}},
+            "required": ["approved"],
+        },
+    )
+
+    assert interrupt.request_schema["required"] == ["message"]
+    assert interrupt.resume_schema["required"] == ["approved"]
+    assert interrupt.has_explicit_contract is True
+
+
 def test_builder_connect_can_use_node_specs_and_returns_resolved_refs() -> None:
     builder = WorkflowBuilder(
         name="connect_specs_demo",