fix(flow): propagate nested FlowNode input requests

docs/pyagentspec/source/agentspec/language_spec_nightly.rst

@@ -1742,6 +1742,9 @@ A more detailed description of each node follows.
         * The flow is started giving the specified inputs
         * The flow provides the specified outputs
         * The flow is run as it was inlined with the overall flow
+        * If the child flow enters an ``input-required`` state, that pending input
+          is propagated through the ``FlowNode`` by default so the parent flow also
+          waits for the same input before continuing.
       - .. list-table::
             :header-rows: 1
             :widths: 20 35 15 15 15
@@ -1757,6 +1760,15 @@ A more detailed description of each node follows.
               - Flow
               - Yes
               - -
+            * - propagate_pending_input
+              - Whether pending input requests raised by the subflow propagate through the
+                ``FlowNode``. When ``true``, the parent flow must surface the child prompt
+                and resume only after the child input is supplied. When ``false``, a runtime
+                may keep the pending input local to the child execution, but must fail
+                explicitly if it cannot preserve that isolation.
+              - bool
+              - No
+              - true
 
       - Inferred from the inner structure.
         It's the sets of inputs required by the StartNode of the inner flow

docs/pyagentspec/source/changelog.rst

@@ -7,6 +7,13 @@ Agent Spec |release|
 Improvements
 ^^^^^^^^^^^^
 
+* **FlowNode pending input propagation**
+
+  ``FlowNode`` now exposes ``propagate_pending_input`` to make nested
+  ``input-required`` behavior explicit. By default, pending input raised inside
+  a child flow is surfaced through the parent ``FlowNode`` so parent flows do not
+  continue or hang while a child is waiting for input.
+
 * **LangGraph adapter timeout and retry improvements**
 
   The LangGraph adapter now applies ``RetryPolicy.request_timeout`` and

pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py

@@ -21,6 +21,7 @@
     BaseChatModel,
     BaseMessage,
     Checkpointer,
+    Command,
     CompiledStateGraph,
     ExecuteOutput,
     FlowStateSchema,
@@ -768,18 +769,76 @@ def __init__(
         self.subflow = subflow
         self.config = config
 
+    def _disabled_pending_input_error(self) -> RuntimeError:
+        return RuntimeError(
+            f"FlowNode `{self.node.name}` received a pending input request from its "
+            "subflow, but propagate_pending_input is disabled."
+        )
+
+    def _invoke_subflow(self, inputs: Dict[str, Any], messages: Messages) -> Dict[str, Any]:
+        from langgraph.errors import GraphInterrupt
+
+        try:
+            return self.subflow.invoke({"messages": messages, "inputs": inputs}, self.config)
+        except GraphInterrupt as e:
+            if self.node.propagate_pending_input:
+                raise
+            raise self._disabled_pending_input_error() from e
+
+    async def _ainvoke_subflow(self, inputs: Dict[str, Any], messages: Messages) -> Dict[str, Any]:
+        from langgraph.errors import GraphInterrupt
+
+        try:
+            return await self.subflow.ainvoke(
+                {"messages": messages, "inputs": inputs}, self.config
+            )
+        except GraphInterrupt as e:
+            if self.node.propagate_pending_input:
+                raise
+            raise self._disabled_pending_input_error() from e
+
+    def _resume_or_raise_for_nested_interrupt(self, flow_output: Dict[str, Any]) -> Any:
+        if not self.node.propagate_pending_input:
+            raise self._disabled_pending_input_error()
+        nested_interrupts = flow_output["__interrupt__"]
+        if (
+            isinstance(nested_interrupts, (list, tuple))
+            and len(nested_interrupts) == 1
+            and hasattr(nested_interrupts[0], "value")
+        ):
+            return interrupt(nested_interrupts[0].value)
+        return interrupt(nested_interrupts)
+
+    def _get_generated_messages(
+        self, initial_messages: Messages, flow_output: Dict[str, Any]
+    ) -> Messages:
+        flow_messages = flow_output.get("messages", [])
+        if not isinstance(flow_messages, list):
+            return []
+        if isinstance(initial_messages, list) and flow_messages[: len(initial_messages)] == list(
+            initial_messages
+        ):
+            return flow_messages[len(initial_messages) :]
+        return flow_messages
+
     def _execute(self, inputs: Dict[str, Any], messages: Messages) -> ExecuteOutput:
-        flow_output = self.subflow.invoke({"messages": messages, "inputs": inputs}, self.config)
+        flow_output = self._invoke_subflow(inputs, messages)
+        while "__interrupt__" in flow_output:
+            resume_value = self._resume_or_raise_for_nested_interrupt(flow_output)
+            flow_output = self.subflow.invoke(Command(resume=resume_value), self.config)
         return flow_output["outputs"], NodeExecutionDetails(
-            branch=flow_output["node_execution_details"]["branch"]
+            branch=flow_output["node_execution_details"]["branch"],
+            generated_messages=self._get_generated_messages(messages, flow_output),
         )
 
     async def _aexecute(self, inputs: Dict[str, Any], messages: Messages) -> ExecuteOutput:
-        flow_output = await self.subflow.ainvoke(
-            {"messages": messages, "inputs": inputs}, self.config
-        )
+        flow_output = await self._ainvoke_subflow(inputs, messages)
+        while "__interrupt__" in flow_output:
+            resume_value = self._resume_or_raise_for_nested_interrupt(flow_output)
+            flow_output = await self.subflow.ainvoke(Command(resume=resume_value), self.config)
         return flow_output["outputs"], NodeExecutionDetails(
-            branch=flow_output["node_execution_details"]["branch"]
+            branch=flow_output["node_execution_details"]["branch"],
+            generated_messages=self._get_generated_messages(messages, flow_output),
         )
 
 

pyagentspec/src/pyagentspec/adapters/langgraph/_types.py

@@ -31,7 +31,7 @@
     from langgraph.graph._node import StateNodeSpec
     from langgraph.graph.message import Messages
     from langgraph.graph.state import CompiledStateGraph
-    from langgraph.types import Checkpointer, interrupt
+    from langgraph.types import Checkpointer, Command, interrupt
 
 else:
     langgraph_swarm = LazyLoader("langgraph_swarm")
@@ -42,6 +42,7 @@
     BaseTool = LazyType("langchain_core.tools", "BaseTool")
     StructuredTool = LazyType("langchain_core.tools", "StructuredTool")
     Checkpointer = LazyType("langgraph.types", "Checkpointer")
+    Command = LazyType("langgraph.types", "Command")
     interrupt = LazyLoader("langgraph.types", "interrupt")
     StateGraph = LazyType("langgraph.graph", "StateGraph")
     Messages = LazyLoader("langgraph.graph.message", "Messages")
@@ -137,6 +138,7 @@ class FlowOutputSchema(TypedDict):
     "BaseChatModel",
     "AgentState",
     "Checkpointer",
+    "Command",
     "interrupt",
     "RunnableConfig",
     "Messages",

pyagentspec/src/pyagentspec/flows/nodes/flownode.py

@@ -12,6 +12,7 @@
 from pyagentspec.flows.node import Node
 from pyagentspec.flows.nodes.endnode import EndNode
 from pyagentspec.property import Property
+from pyagentspec.versioning import AgentSpecVersionEnum
 
 
 class FlowNode(Node):
@@ -26,6 +27,11 @@ class FlowNode(Node):
     - **Branches**
         Inferred from the inner flow, one per each different value of the attribute
         ``branch_name`` of the nodes of type EndNode in the inner flow.
+    - **Pending input**
+        If the subflow execution pauses to request input from a user or external caller,
+        the pending input state is propagated through the ``FlowNode`` by default. This
+        preserves the inline semantics of the subflow and lets the parent flow surface
+        the child prompt and resume only after the child input is supplied.
 
     Example
     -------
@@ -259,6 +265,23 @@ class FlowNode(Node):
     subflow: Flow
     """The flow that should be executed"""
 
+    propagate_pending_input: bool = True
+    """Whether pending input requests raised inside the subflow propagate through this FlowNode."""
+
+    def _versioned_model_fields_to_exclude(
+        self, agentspec_version: AgentSpecVersionEnum
+    ) -> set[str]:
+        fields_to_exclude = super()._versioned_model_fields_to_exclude(agentspec_version)
+        if agentspec_version < AgentSpecVersionEnum.v26_2_0:
+            fields_to_exclude.add("propagate_pending_input")
+        return fields_to_exclude
+
+    def _infer_min_agentspec_version_from_configuration(self) -> AgentSpecVersionEnum:
+        min_version = super()._infer_min_agentspec_version_from_configuration()
+        if not self.propagate_pending_input:
+            min_version = max(min_version, AgentSpecVersionEnum.v26_2_0)
+        return min_version
+
     def _get_inferred_branches(self) -> List[str]:
         if hasattr(self, "subflow"):
             end_nodes = sorted(

pyagentspec/tests/adapters/langgraph/flows/test_flownode.py

@@ -8,7 +8,7 @@
 
 from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge
 from pyagentspec.flows.flow import Flow
-from pyagentspec.flows.nodes import EndNode, FlowNode, StartNode
+from pyagentspec.flows.nodes import EndNode, FlowNode, InputMessageNode, StartNode
 from pyagentspec.property import StringProperty
 
 
@@ -88,6 +88,109 @@ def test_flownode_can_be_imported_and_executed(flow_node_flow: Flow) -> None:
     assert outputs["custom_prop"] == "custom"
 
 
+def _flow_with_nested_input_node(*, propagate_pending_input: bool = True) -> Flow:
+    custom_property = StringProperty(title="custom_input")
+    input_message_node = InputMessageNode(
+        name="input_message",
+        outputs=[custom_property],
+    )
+    subflow_start_node = StartNode(name="subflow_start")
+    subflow_end_node = EndNode(name="subflow_end", outputs=[custom_property])
+    subflow = Flow(
+        name="subflow",
+        start_node=subflow_start_node,
+        nodes=[subflow_start_node, input_message_node, subflow_end_node],
+        control_flow_connections=[
+            ControlFlowEdge(
+                name="subflow_start_to_input",
+                from_node=subflow_start_node,
+                to_node=input_message_node,
+            ),
+            ControlFlowEdge(
+                name="input_to_subflow_end",
+                from_node=input_message_node,
+                to_node=subflow_end_node,
+            ),
+        ],
+        data_flow_connections=[
+            DataFlowEdge(
+                name="input_to_subflow_end",
+                source_node=input_message_node,
+                source_output=custom_property.title,
+                destination_node=subflow_end_node,
+                destination_input=custom_property.title,
+            ),
+        ],
+        outputs=[custom_property],
+    )
+
+    flow_node = FlowNode(
+        name="flow_node",
+        subflow=subflow,
+        propagate_pending_input=propagate_pending_input,
+    )
+    start_node = StartNode(name="start")
+    end_node = EndNode(name="end", outputs=[custom_property])
+    return Flow(
+        name="flow",
+        start_node=start_node,
+        nodes=[start_node, flow_node, end_node],
+        control_flow_connections=[
+            ControlFlowEdge(name="start_to_flow_node", from_node=start_node, to_node=flow_node),
+            ControlFlowEdge(name="flow_node_to_end", from_node=flow_node, to_node=end_node),
+        ],
+        data_flow_connections=[
+            DataFlowEdge(
+                name="flow_node_to_end",
+                source_node=flow_node,
+                source_output=custom_property.title,
+                destination_node=end_node,
+                destination_input=custom_property.title,
+            ),
+        ],
+        outputs=[custom_property],
+    )
+
+
+def test_flownode_propagates_nested_input_required_state() -> None:
+    from langchain_core.messages import HumanMessage
+    from langchain_core.runnables import RunnableConfig
+    from langgraph.checkpoint.memory import MemorySaver
+    from langgraph.types import Command
+
+    from pyagentspec.adapters.langgraph import AgentSpecLoader
+
+    flow = _flow_with_nested_input_node()
+    agent = AgentSpecLoader(checkpointer=MemorySaver()).load_component(flow)
+
+    config = RunnableConfig({"configurable": {"thread_id": "nested-flow-node-input"}})
+    result = agent.invoke({}, config=config)
+    assert "__interrupt__" in result
+    assert result["__interrupt__"][0].value == ""
+
+    result = agent.invoke(Command(resume="nested value"), config=config)
+
+    assert "outputs" in result
+    assert result["outputs"]["custom_input"] == "nested value"
+    assert len(result["messages"]) == 1
+    assert isinstance(result["messages"][0], HumanMessage)
+    assert result["messages"][0].content == "nested value"
+
+
+def test_flownode_disabled_pending_input_propagation_raises() -> None:
+    from langchain_core.runnables import RunnableConfig
+    from langgraph.checkpoint.memory import MemorySaver
+
+    from pyagentspec.adapters.langgraph import AgentSpecLoader
+
+    flow = _flow_with_nested_input_node(propagate_pending_input=False)
+    agent = AgentSpecLoader(checkpointer=MemorySaver()).load_component(flow)
+
+    config = RunnableConfig({"configurable": {"thread_id": "nested-flow-node-input-disabled"}})
+    with pytest.raises(RuntimeError, match="propagate_pending_input is disabled"):
+        agent.invoke({}, config=config)
+
+
 @pytest.mark.anyio
 async def test_flownode_can_be_executed_async(flow_node_flow: Flow) -> None:
     from pyagentspec.adapters.langgraph import AgentSpecLoader

pyagentspec/tests/serialization/test_flow_node.py

@@ -0,0 +1,67 @@
+# Copyright © 2025 Oracle and/or its affiliates.
+#
+# This software is under the Apache License 2.0
+# (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License
+# (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option.
+
+import pytest
+
+from pyagentspec.flows.flow import Flow
+from pyagentspec.flows.nodes.flownode import FlowNode
+from pyagentspec.serialization import AgentSpecDeserializer, AgentSpecSerializer
+from pyagentspec.versioning import AgentSpecVersionEnum
+
+
+def test_flow_node_propagates_pending_input_by_default(simplest_flow: Flow) -> None:
+    flow_node = FlowNode(name="flow_node", subflow=simplest_flow)
+
+    assert flow_node.propagate_pending_input is True
+
+
+def test_flow_node_serializes_pending_input_propagation_in_26_2_0(
+    simplest_flow: Flow,
+) -> None:
+    flow_node = FlowNode(name="flow_node", subflow=simplest_flow)
+
+    serialized = AgentSpecSerializer().to_dict(
+        flow_node,
+        agentspec_version=AgentSpecVersionEnum.v26_2_0,
+    )
+
+    assert serialized["propagate_pending_input"] is True
+
+
+def test_flow_node_excludes_default_pending_input_propagation_before_26_2_0(
+    simplest_flow: Flow,
+) -> None:
+    flow_node = FlowNode(name="flow_node", subflow=simplest_flow)
+
+    serialized = AgentSpecSerializer().to_dict(
+        flow_node,
+        agentspec_version=AgentSpecVersionEnum.v26_1_2,
+    )
+
+    assert "propagate_pending_input" not in serialized
+
+
+def test_flow_node_can_disable_pending_input_propagation(simplest_flow: Flow) -> None:
+    flow_node = FlowNode(
+        name="flow_node",
+        subflow=simplest_flow,
+        propagate_pending_input=False,
+    )
+
+    assert flow_node.min_agentspec_version == AgentSpecVersionEnum.v26_2_0
+    with pytest.raises(ValueError, match="Invalid agentspec_version"):
+        AgentSpecSerializer().to_dict(
+            flow_node,
+            agentspec_version=AgentSpecVersionEnum.v26_1_2,
+        )
+
+    serialized = AgentSpecSerializer().to_dict(flow_node)
+    assert serialized["agentspec_version"] == AgentSpecVersionEnum.v26_2_0.value
+    assert serialized["propagate_pending_input"] is False
+
+    deserialized = AgentSpecDeserializer().from_dict(serialized)
+    assert isinstance(deserialized, FlowNode)
+    assert deserialized.propagate_pending_input is False