runtime error

so future outcome error can route to bro

src/wf_authoring/__init__.py

@@ -23,6 +23,7 @@ from .ops import (
     ItemOutput,
     MaybeItemOutput,
     PickKeyInput,
+    RuntimeErrorInput,
     SequenceInput,
     TruthyInput,
     ValueOutput,
@@ -37,6 +38,7 @@ from .ops import (
     last_item_or_none,
     length,
     pick_key,
+    runtime_error,
     truthy,
 )
 from .nodes import (
@@ -64,6 +66,7 @@ __all__ = [
     "ItemOutput",
     "MaybeItemOutput",
     "PickKeyInput",
+    "RuntimeErrorInput",
     "NodeReturn",
     "NodeSpec",
     "Nothing",
@@ -97,6 +100,7 @@ __all__ = [
     "last_item_or_none",
     "length",
     "pick_key",
+    "runtime_error",
     "node",
     "outcome",
     "state",

src/wf_authoring/ops/__init__.py

@@ -16,12 +16,14 @@ from .values import (
     CoalesceInput,
     ConstantInput,
     PickKeyInput,
+    RuntimeErrorInput,
     TruthyInput,
     ValueOutput,
     coalesce,
     constant,
     default_if_none,
     pick_key,
+    runtime_error,
     truthy,
 )
 
@@ -33,6 +35,7 @@ __all__ = [
     "ItemOutput",
     "MaybeItemOutput",
     "PickKeyInput",
+    "RuntimeErrorInput",
     "SequenceInput",
     "TruthyInput",
     "ValueOutput",
@@ -47,5 +50,6 @@ __all__ = [
     "last_item_or_none",
     "length",
     "pick_key",
+    "runtime_error",
     "truthy",
 ]

src/wf_authoring/ops/values.py

@@ -1,10 +1,10 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, NoReturn
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
-from wf_authoring.nodes import NodeReturn, node
+from wf_authoring.nodes import NodeReturn, Nothing, node
 
 
 class CoalesceInput(BaseModel):
@@ -39,6 +39,13 @@ class TruthyInput(BaseModel):
     value: Any
 
 
+class RuntimeErrorInput(BaseModel):
+    """Input model for intentionally failing a workflow branch."""
+
+    message: str
+    details: dict[str, Any] = Field(default_factory=dict)
+
+
 @node(
     name="authoring.coalesce",
     input_model=CoalesceInput,
@@ -92,3 +99,16 @@ def truthy(input: TruthyInput) -> NodeReturn[ValueOutput]:
     value = bool(input.value)
     outcome = "truthy" if value else "falsey"
     return NodeReturn(outcome=outcome, output=ValueOutput(value=value))
+
+
+@node(
+    name="authoring.runtime_error",
+    input_model=RuntimeErrorInput,
+    output_model=Nothing,
+    description="Raise RuntimeError with the provided message and details.",
+)
+def runtime_error(input: RuntimeErrorInput) -> NoReturn:
+    """Raise RuntimeError with the provided message and details."""
+    if input.details:
+        raise RuntimeError(f"{input.message}: {input.details!r}")
+    raise RuntimeError(input.message)

tests/authoring/test_ops.py

@@ -18,6 +18,7 @@ from wf_authoring import (
     last_item_or_none,
     length,
     pick_key,
+    runtime_error,
     state_path,
     truthy,
 )
@@ -258,3 +259,13 @@ def test_truthy_routes_truthy_and_falsey_outcomes() -> None:
 
     assert truthy_result == {"outcome": "truthy", "output": {"value": True}}
     assert falsey_result == {"outcome": "falsey", "output": {"value": False}}
+
+
+def test_runtime_error_raises_with_message_and_details() -> None:
+    registry = build_registry(runtime_error)
+
+    with pytest.raises(RuntimeError, match="bad branch"):
+        registry["authoring.runtime_error"](
+            {"message": "bad branch", "details": {"step": "demo"}},
+            RuntimeContext(current_node_id="runtime_error"),
+        )