diff --git a/spring-webflow/src/main/java/org/springframework/webflow/support/persistence/HibernateFlowExecutionListener.java b/spring-webflow/src/main/java/org/springframework/webflow/support/persistence/HibernateFlowExecutionListener.java
index 9589ddee..8cb56164 100644
--- a/spring-webflow/src/main/java/org/springframework/webflow/support/persistence/HibernateFlowExecutionListener.java
+++ b/spring-webflow/src/main/java/org/springframework/webflow/support/persistence/HibernateFlowExecutionListener.java
@@ -26,6 +26,7 @@ import org.springframework.transaction.support.TransactionCallbackWithoutResult;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.webflow.core.collection.AttributeMap;
+import org.springframework.webflow.definition.FlowDefinition;
 import org.springframework.webflow.execution.FlowExecutionException;
 import org.springframework.webflow.execution.FlowExecutionListener;
 import org.springframework.webflow.execution.FlowExecutionListenerAdapter;
@@ -34,11 +35,12 @@ import org.springframework.webflow.execution.RequestContext;
 import org.springframework.webflow.execution.ViewSelection;
 
 /**
- * A {@link FlowExecutionListener} that implements the Session-per-Conversation pattern using the native Hibernate API.
+ * A {@link FlowExecutionListener} that implements the Flow Managed Persistence Context (FMPC) pattern using the native
+ * Hibernate API.
  * <p>
  * The general pattern is as follows:
  * <ul>
- * <li>When a flow execution starts, create a new Hibernate Session and bind it to conversation scope.
+ * <li>When a flow execution starts, create a new Hibernate Session and bind it to flow scope.
  * <li>Before processing a flow execution request, expose the conversationally-bound session as the "current session"
  * for the current thread.
  * <li>When an existing flow pauses, unbind the session from the current thread.
@@ -50,8 +52,8 @@ import org.springframework.webflow.execution.ViewSelection;
  * <ul>
  * <li> Create a new persistence context when a new flow execution with the 'persistenceContext' attribute starts
  * <li> Load some objects into this persistence context
- * <li> Perform edits to those objects over a series of conversational requests
- * <li> On successful conversation completion, commit and flush those edits to the database, applying a version check if
+ * <li> Perform edits to those objects over a series of requests into the flow
+ * <li> On successful flow ccompletion, commit and flush those edits to the database, applying a version check if
  * necessary.
  * </ul>
  * 
@@ -61,10 +63,10 @@ import org.springframework.webflow.execution.ViewSelection;
  * transaction. In that case, the session's flush mode will be set to Manual and no intermediate changes will be
  * flushed.
  * <p>
- * Care should be taken to prevent premature commits of conversational data while the conversation is in progress. You
- * would generally not want intermediate flushing to happen, as the nature of a conversation implies a transient,
- * isolated resource that can be canceled before it ends. Generally, the only time a read-write transaction should be
- * started is upon successful completion of the conversation, triggered by reaching a 'commit' end state.
+ * Care should be taken to prevent premature commits of flow data while the flow is in progress. You would generally not
+ * want intermediate flushing to happen, as the nature of a flow implies a transient, isolated resource that can be
+ * canceled before it ends. Generally, the only time a read-write transaction should be started is upon successful
+ * completion of the conversation, triggered by reaching a 'commit' end state.
  * 
  * @author Ben Hale
  * @author Keith Donald
@@ -73,6 +75,8 @@ import org.springframework.webflow.execution.ViewSelection;
  */
 public class HibernateFlowExecutionListener extends FlowExecutionListenerAdapter {
 
+    private static final String PERSISTENCE_CONTEXT_ATTRIBUTE = "persistenceContext";
+
     private static final String HIBERNATE_SESSION_ATTRIBUTE = "session";
 
     private TransactionTemplate transactionTemplate;
@@ -99,25 +103,28 @@ public class HibernateFlowExecutionListener extends FlowExecutionListenerAdapter
     }
 
     public void sessionCreated(RequestContext context, FlowSession session) {
-	if (session.isRoot() && session.getDefinition().getAttributes().contains("persistenceContext")) {
+	if (isPersistenceContext(session.getDefinition())) {
 	    Session hibernateSession = createSession(context);
-	    context.getConversationScope().put(HIBERNATE_SESSION_ATTRIBUTE, hibernateSession);
+	    session.getScope().put(HIBERNATE_SESSION_ATTRIBUTE, hibernateSession);
 	    bind(hibernateSession, context);
 	}
     }
 
     public void resumed(RequestContext context) {
-	bind(getHibernateSession(context), context);
+	if (isPersistenceContext(context.getActiveFlow())) {
+	    bind(getHibernateSession(context), context);
+	}
     }
 
     public void paused(RequestContext context, ViewSelection selectedView) {
-	unbind(getHibernateSession(context), context);
+	if (isPersistenceContext(context.getActiveFlow())) {
+	    unbind(getHibernateSession(context), context);
+	}
     }
 
     public void sessionEnded(RequestContext context, FlowSession session, AttributeMap output) {
-	if (session.isRoot()) {
-	    final Session hibernateSession = (Session) context.getConversationScope().remove(
-		    HIBERNATE_SESSION_ATTRIBUTE);
+	if (isPersistenceContext(session.getDefinition())) {
+	    final Session hibernateSession = (Session) session.getScope().remove(HIBERNATE_SESSION_ATTRIBUTE);
 	    Boolean commitStatus = session.getState().getAttributes().getBoolean("commit");
 	    if (Boolean.TRUE.equals(commitStatus)) {
 		// this is a commit end state - start a new transaction that quickly commits
@@ -129,17 +136,23 @@ public class HibernateFlowExecutionListener extends FlowExecutionListenerAdapter
 		    }
 		});
 	    }
-	    hibernateSession.close();
 	    unbind(hibernateSession, context);
+	    hibernateSession.close();
 	}
     }
 
     public void exceptionThrown(RequestContext context, FlowExecutionException exception) {
-	unbind(getHibernateSession(context), context);
+	if (isPersistenceContext(context.getActiveFlow())) {
+	    unbind(getHibernateSession(context), context);
+	}
     }
 
     // internal helpers
 
+    private boolean isPersistenceContext(FlowDefinition flow) {
+	return flow.getAttributes().contains(PERSISTENCE_CONTEXT_ATTRIBUTE);
+    }
+
     private Session createSession(RequestContext context) {
 	Session session = (entityInterceptor != null ? sessionFactory.openSession(entityInterceptor) : sessionFactory
 		.openSession());
@@ -148,7 +161,7 @@ public class HibernateFlowExecutionListener extends FlowExecutionListenerAdapter
     }
 
     private Session getHibernateSession(RequestContext context) {
-	return (Session) context.getConversationScope().get(HIBERNATE_SESSION_ATTRIBUTE);
+	return (Session) context.getFlowScope().get(HIBERNATE_SESSION_ATTRIBUTE);
     }
 
     private void bind(Session session, RequestContext context) {
diff --git a/spring-webflow/src/main/java/org/springframework/webflow/support/persistence/JpaFlowExecutionListener.java b/spring-webflow/src/main/java/org/springframework/webflow/support/persistence/JpaFlowExecutionListener.java
index da5fcbbd..455b3aae 100644
--- a/spring-webflow/src/main/java/org/springframework/webflow/support/persistence/JpaFlowExecutionListener.java
+++ b/spring-webflow/src/main/java/org/springframework/webflow/support/persistence/JpaFlowExecutionListener.java
@@ -25,6 +25,7 @@ import org.springframework.transaction.support.TransactionCallbackWithoutResult;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.webflow.core.collection.AttributeMap;
+import org.springframework.webflow.definition.FlowDefinition;
 import org.springframework.webflow.execution.FlowExecutionException;
 import org.springframework.webflow.execution.FlowExecutionListener;
 import org.springframework.webflow.execution.FlowExecutionListenerAdapter;
@@ -33,14 +34,14 @@ import org.springframework.webflow.execution.RequestContext;
 import org.springframework.webflow.execution.ViewSelection;
 
 /**
- * A {@link FlowExecutionListener} that implements the Session-per-Conversation using the standard Java Persistence API
- * (JPA).
+ * A {@link FlowExecutionListener} that implements the Flow Managed Persistence Context (FMPC) pattern using the
+ * standard Java Persistence API (JPA).
  * <p>
  * This implementation uses standard JPA APIs. The general pattern is as follows:
  * <ul>
- * <li>When a flow execution starts, create a new JPA persistence context and bind it to conversation scope.
- * <li>Before processing a flow execution request, expose the conversationally-bound persistence context as the
- * "current" persistence context for the current thread.
+ * <li>When a flow execution starts, create a new JPA persistence context and bind it to flow scope.
+ * <li>Before processing a flow execution request, expose the flow-scoped persistence context as the "current"
+ * persistence context for the current thread.
  * <li>When an existing flow pauses, unbind the persistence context from the current thread.
  * <li>When an existing flow ends, commit the changes made to the persistence context in a transaction if the ending
  * state is a commit state. Then, unbind the context and close it.
@@ -50,7 +51,7 @@ import org.springframework.webflow.execution.ViewSelection;
  * <ul>
  * <li> Create a new persistence context when a new flow execution with the 'persistenceContext' attribute starts
  * <li> Load some objects into this persistence context
- * <li> Perform edits to those objects over a series of conversational requests
+ * <li> Perform edits to those objects over a series of requests into the flow
  * <li> On successful conversation completion, commit and flush those edits to the database, applying a version check if
  * necessary.
  * </ul>
@@ -63,10 +64,10 @@ import org.springframework.webflow.execution.ViewSelection;
  * to MANUAL to ensure any in-progress changes to managed persistent entities are not flushed, while reads of new
  * objects occur transactionally.
  * <p>
- * Care should be taken to prevent premature commits of conversational data while the conversation is in progress. You
- * would generally not want intermediate flushing to happen, as the nature of a conversation implies a transient,
- * isolated resource that can be canceled before it ends. Generally, the only time a read-write transaction should be
- * started is upon successful completion of the conversation, triggered by reaching a 'commit' end state.
+ * Care should be taken to prevent premature commits of flow data while the flow is in progress. You would generally not
+ * want intermediate flushing to happen, as the nature of a flow implies a transient, isolated resource that can be
+ * canceled before it ends. Generally, the only time a read-write transaction should be started is upon successful
+ * completion of the flow, triggered by reaching a 'commit' end state.
  * 
  * @author Keith Donald
  * @author Juergen Hoeller
@@ -74,6 +75,8 @@ import org.springframework.webflow.execution.ViewSelection;
  */
 public class JpaFlowExecutionListener extends FlowExecutionListenerAdapter {
 
+    private static final String PERSISTENCE_CONTEXT_ATTRIBUTE = "persistenceContext";
+
     private static final String ENTITY_MANAGER_ATTRIBUTE = "entityManager";
 
     private EntityManagerFactory entityManagerFactory;
@@ -81,7 +84,7 @@ public class JpaFlowExecutionListener extends FlowExecutionListenerAdapter {
     private TransactionTemplate transactionTemplate;
 
     /**
-     * Create a new Session-per-Conversation listener using given JPA Entity Manager factory.
+     * Create a new JPA flow execution listener using given JPA Entity Manager factory.
      * @param entityManagerFactory the entity manager factory to use
      */
     public JpaFlowExecutionListener(EntityManagerFactory entityManagerFactory,
@@ -91,7 +94,7 @@ public class JpaFlowExecutionListener extends FlowExecutionListenerAdapter {
     }
 
     public void sessionCreated(RequestContext context, FlowSession session) {
-	if (session.getDefinition().getAttributes().contains("persistenceContext")) {
+	if (isPersistenceContext(session.getDefinition())) {
 	    EntityManager em = entityManagerFactory.createEntityManager();
 	    session.getScope().put(ENTITY_MANAGER_ATTRIBUTE, em);
 	    bind(em);
@@ -99,19 +102,19 @@ public class JpaFlowExecutionListener extends FlowExecutionListenerAdapter {
     }
 
     public void resumed(RequestContext context) {
-	if (context.getActiveFlow().getAttributes().contains("persistenceContext")) {
+	if (isPersistenceContext(context.getActiveFlow())) {
 	    bind(getEntityManager(context));
 	}
     }
 
     public void paused(RequestContext context, ViewSelection selectedView) {
-	if (context.getActiveFlow().getAttributes().contains("persistenceContext")) {
+	if (isPersistenceContext(context.getActiveFlow())) {
 	    unbind(getEntityManager(context));
 	}
     }
 
     public void sessionEnded(RequestContext context, FlowSession session, AttributeMap output) {
-	if (session.getDefinition().getAttributes().contains("persistenceContext")) {
+	if (isPersistenceContext(session.getDefinition())) {
 	    final EntityManager em = (EntityManager) session.getScope().remove(ENTITY_MANAGER_ATTRIBUTE);
 	    Boolean commitStatus = session.getState().getAttributes().getBoolean("commit");
 	    if (Boolean.TRUE.equals(commitStatus)) {
@@ -133,13 +136,17 @@ public class JpaFlowExecutionListener extends FlowExecutionListenerAdapter {
     }
 
     public void exceptionThrown(RequestContext context, FlowExecutionException exception) {
-	if (context.getActiveFlow().getAttributes().contains("persistenceContext")) {
+	if (isPersistenceContext(context.getActiveFlow())) {
 	    unbind(getEntityManager(context));
 	}
     }
 
     // internal helpers
 
+    private boolean isPersistenceContext(FlowDefinition flow) {
+	return flow.getAttributes().contains(PERSISTENCE_CONTEXT_ATTRIBUTE);
+    }
+
     private EntityManager getEntityManager(RequestContext context) {
 	return (EntityManager) context.getFlowScope().get(ENTITY_MANAGER_ATTRIBUTE);
     }
diff --git a/spring-webflow/src/test/java/org/springframework/webflow/support/persistence/HibernateFlowExecutionListenerTests.java b/spring-webflow/src/test/java/org/springframework/webflow/support/persistence/HibernateFlowExecutionListenerTests.java
index 1fe9e219..c4c824bc 100644
--- a/spring-webflow/src/test/java/org/springframework/webflow/support/persistence/HibernateFlowExecutionListenerTests.java
+++ b/spring-webflow/src/test/java/org/springframework/webflow/support/persistence/HibernateFlowExecutionListenerTests.java
@@ -71,10 +71,11 @@ public class HibernateFlowExecutionListenerTests extends TestCase {
 	MockFlowSession flowSession = new MockFlowSession();
 	flowSession.getDefinitionInternal().getAttributeMap().put("persistenceContext", "true");
 	listener.sessionCreated(context, flowSession);
+	context.setActiveSession(flowSession);
 	assertSessionBound();
 
 	// Session created and bound to conversation
-	final Session hibSession = (Session) context.getConversationScope().get("session");
+	final Session hibSession = (Session) flowSession.getScope().get("session");
 	assertNotNull("Should have been populated", hibSession);
 	listener.paused(context, ViewSelection.NULL_VIEW);
 	assertSessionNotBound();
@@ -106,6 +107,7 @@ public class HibernateFlowExecutionListenerTests extends TestCase {
 	MockFlowSession flowSession = new MockFlowSession();
 	flowSession.getDefinitionInternal().getAttributeMap().put("persistenceContext", "true");
 	listener.sessionCreated(context, flowSession);
+	context.setActiveSession(flowSession);
 	assertSessionBound();
 
 	TestBean bean = new TestBean("Keith Donald");
@@ -128,6 +130,7 @@ public class HibernateFlowExecutionListenerTests extends TestCase {
 	MockFlowSession flowSession = new MockFlowSession();
 	flowSession.getDefinitionInternal().getAttributeMap().put("persistenceContext", "true");
 	listener.sessionCreated(context, flowSession);
+	context.setActiveSession(flowSession);
 	assertSessionBound();
 
 	TestBean bean1 = new TestBean("Keith Donald");
@@ -161,6 +164,7 @@ public class HibernateFlowExecutionListenerTests extends TestCase {
 	MockFlowSession flowSession = new MockFlowSession();
 	flowSession.getDefinitionInternal().getAttributeMap().put("persistenceContext", "true");
 	listener.sessionCreated(context, flowSession);
+	context.setActiveSession(flowSession);
 	assertSessionBound();
 
 	TestBean bean = new TestBean("Keith Donald");
@@ -182,6 +186,7 @@ public class HibernateFlowExecutionListenerTests extends TestCase {
 	MockFlowSession flowSession = new MockFlowSession();
 	flowSession.getDefinitionInternal().getAttributeMap().put("persistenceContext", "true");
 	listener.sessionCreated(context, flowSession);
+	context.setActiveSession(flowSession);
 	assertSessionBound();
 
 	EndState endState = new EndState(flowSession.getDefinitionInternal(), "cancel");
@@ -201,6 +206,7 @@ public class HibernateFlowExecutionListenerTests extends TestCase {
 	MockFlowSession flowSession = new MockFlowSession();
 	flowSession.getDefinitionInternal().getAttributeMap().put("persistenceContext", "true");
 	listener.sessionCreated(context, flowSession);
+	context.setActiveSession(flowSession);
 	assertSessionBound();
 
 	TestBean bean1 = new TestBean("Keith Donald");
diff --git a/spring-webflow/src/test/java/org/springframework/webflow/support/persistence/JpaFlowExecutionListenerTests.java b/spring-webflow/src/test/java/org/springframework/webflow/support/persistence/JpaFlowExecutionListenerTests.java
index eef71bfb..c96d2f65 100644
--- a/spring-webflow/src/test/java/org/springframework/webflow/support/persistence/JpaFlowExecutionListenerTests.java
+++ b/spring-webflow/src/test/java/org/springframework/webflow/support/persistence/JpaFlowExecutionListenerTests.java
@@ -16,6 +16,7 @@ import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
 import org.springframework.orm.jpa.vendor.HibernateJpaVendorAdapter;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.webflow.engine.EndState;
+import org.springframework.webflow.execution.ViewSelection;
 import org.springframework.webflow.test.MockFlowSession;
 import org.springframework.webflow.test.MockRequestContext;
 
@@ -56,6 +57,7 @@ public class JpaFlowExecutionListenerTests extends TestCase {
 	MockFlowSession flowSession = new MockFlowSession();
 	flowSession.getDefinitionInternal().getAttributeMap().put("persistenceContext", "true");
 	jpaListener.sessionCreated(context, flowSession);
+	context.setActiveSession(flowSession);
 	assertSessionBound();
 
 	TestBean bean = new TestBean(1, "Keith Donald");
@@ -72,6 +74,40 @@ public class JpaFlowExecutionListenerTests extends TestCase {
 	assertFalse(flowSession.getScope().contains("hibernate.session"));
     }
 
+    public void testFlowCommitsAfterMultipleRequests() {
+	assertEquals("Table should only have one row", 1, jdbcTemplate.queryForInt("select count(*) from T_BEAN"));
+	MockRequestContext context = new MockRequestContext();
+	MockFlowSession flowSession = new MockFlowSession();
+	flowSession.getDefinitionInternal().getAttributeMap().put("persistenceContext", "true");
+	jpaListener.sessionCreated(context, flowSession);
+	context.setActiveSession(flowSession);
+	assertSessionBound();
+
+	TestBean bean1 = new TestBean(1, "Keith Donald");
+	jpaTemplate.persist(bean1);
+	assertEquals("Table should still only have one row", 1, jdbcTemplate.queryForInt("select count(*) from T_BEAN"));
+	jpaListener.paused(context, ViewSelection.NULL_VIEW);
+	assertSessionNotBound();
+
+	jpaListener.resumed(context);
+	TestBean bean2 = new TestBean(2, "Keith Donald");
+	jpaTemplate.persist(bean2);
+	assertEquals("Table should still only have one row", 1, jdbcTemplate.queryForInt("select count(*) from T_BEAN"));
+	assertSessionBound();
+
+	EndState endState = new EndState(flowSession.getDefinitionInternal(), "success");
+	endState.getAttributeMap().put("commit", Boolean.TRUE);
+	flowSession.setState(endState);
+
+	jpaListener.sessionEnded(context, flowSession, null);
+	assertEquals("Table should only have three rows", 3, jdbcTemplate.queryForInt("select count(*) from T_BEAN"));
+	assertFalse(flowSession.getScope().contains("hibernate.session"));
+
+	assertSessionNotBound();
+	assertFalse(flowSession.getScope().contains("hibernate.session"));
+
+    }
+
     private DataSource getDataSource() {
 	DriverManagerDataSource dataSource = new DriverManagerDataSource();
 	dataSource.setDriverClassName("org.hsqldb.jdbcDriver");