diff --git a/obp-api/src/main/resources/props/sample.props.template b/obp-api/src/main/resources/props/sample.props.template
index cd4ea1034b..2f4d3d6a1e 100644
--- a/obp-api/src/main/resources/props/sample.props.template
+++ b/obp-api/src/main/resources/props/sample.props.template
@@ -516,6 +516,9 @@ mail.smtp.port=25
 mail.smtp.auth=false
 mail.smtp.user=
 mail.smtp.password=
+## Milliseconds to wait for the SMTP server to connect, answer and accept data (Default: 30000).
+## Mail such as the consent challenge is sent inside the request, so a server that goes quiet must not hold it.
+#mail.smtp.timeout_millis=30000
 
 ## TLS/SSL Configuration
 ## ---------------------
diff --git a/obp-api/src/main/scala/code/api/ResourceDocs1_4_0/SwaggerJSONFactory.scala b/obp-api/src/main/scala/code/api/ResourceDocs1_4_0/SwaggerJSONFactory.scala
index f7ad80eb3f..940eb5d100 100644
--- a/obp-api/src/main/scala/code/api/ResourceDocs1_4_0/SwaggerJSONFactory.scala
+++ b/obp-api/src/main/scala/code/api/ResourceDocs1_4_0/SwaggerJSONFactory.scala
@@ -707,7 +707,13 @@ object SwaggerJSONFactory extends MdcLoggable {
         case _ => paramValue
       }
 
-      val definition = buildSwaggerSchema(paramType, exampleValue)
+      // Name the entity and field, so an unusable example can be found from the message alone.
+      val definition =
+        try buildSwaggerSchema(paramType, exampleValue)
+        catch {
+          case e: RuntimeException =>
+            throw new RuntimeException(s"${entityType.typeSymbol.name}.$paramName: ${e.getMessage}", e)
+        }
 
       s""" "$paramName": $definition """
     })
diff --git a/obp-api/src/main/scala/code/api/constant/constant.scala b/obp-api/src/main/scala/code/api/constant/constant.scala
index e684eb40d8..e32020497c 100644
--- a/obp-api/src/main/scala/code/api/constant/constant.scala
+++ b/obp-api/src/main/scala/code/api/constant/constant.scala
@@ -374,6 +374,10 @@ object Constant extends MdcLoggable {
   // The Glossary as each instance holds it in memory. Bumping it reloads the Glossary at once and
   // rebuilds every cached resource-docs document, because their keys carry the Glossary version.
   final val GLOSSARY_NAMESPACE = "glossary"
+  // Dynamic entity definitions: the definition lists memoized in Redis and the definitions map each
+  // instance builds from them. A definition change bumps it once committed, so every instance reads
+  // the definitions again.
+  final val DYNAMIC_ENTITY_NAMESPACE = "dynamic_entity"
 
   // List of all versioned cache namespaces
   final val ALL_CACHE_NAMESPACES = List(
@@ -393,7 +397,8 @@ object Constant extends MdcLoggable {
     FINANCIAL_PRODUCTS_NAMESPACE,
     API_PRODUCTS_NAMESPACE,
     MESSAGE_DOCS_NAMESPACE,
-    GLOSSARY_NAMESPACE
+    GLOSSARY_NAMESPACE,
+    DYNAMIC_ENTITY_NAMESPACE
   )
 
   // Cache key prefixes with global namespace and versioning for easy invalidation
diff --git a/obp-api/src/main/scala/code/api/dynamic/entity/helper/DefinitionsMapCache.scala b/obp-api/src/main/scala/code/api/dynamic/entity/helper/DefinitionsMapCache.scala
new file mode 100644
index 0000000000..1cc717cb64
--- /dev/null
+++ b/obp-api/src/main/scala/code/api/dynamic/entity/helper/DefinitionsMapCache.scala
@@ -0,0 +1,68 @@
+/**
+Open Bank Project - API
+Copyright (C) 2011-2026, TESOBE GmbH.
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+Email: contact@tesobe.com
+TESOBE GmbH.
+Osloer Strasse 16/17
+Berlin 13359, Germany
+
+This product includes software developed at
+TESOBE (http://www.tesobe.com/)
+
+  */
+
+package code.api.dynamic.entity.helper
+
+/**
+ * Keeps one built value (the dynamic entity definitions map) for up to `ttlMillis`, and only while
+ * `currentVersion` still returns the version it was built under.
+ *
+ * The version is read before building, so a value built while the version changed is kept under the
+ * old version and dropped on the next read. [[forget]] drops the value on this instance alone, for when
+ * the version cannot change (Redis unavailable); a build that started before it is not kept after it.
+ * Nothing is kept, and the version is not read, when `ttlMillis` is not positive.
+ */
+private[helper] final class DefinitionsMapCache[M](ttlMillis: Long, currentVersion: () => Long, build: () => M) {
+
+  // (built at, version it was built under, value)
+  @volatile private var entry: Option[(Long, Long, M)] = None
+  // Replaced by forget; a build keeps its value only if the token it started under is still current.
+  @volatile private var token: AnyRef = new Object
+
+  def get: M =
+    if (ttlMillis <= 0) build()
+    else {
+      val now = System.currentTimeMillis()
+      val version = currentVersion()
+      entry match {
+        case Some((builtAt, builtUnder, value)) if builtUnder == version && now - builtAt < ttlMillis => value
+        case _ =>
+          val startedUnder = token
+          val value = build()
+          if (token eq startedUnder) entry = Some((now, version, value))
+          value
+      }
+    }
+
+  def forget(): Unit = {
+    token = new Object
+    entry = None
+  }
+
+  /** The value kept now, if any, whatever its version. */
+  def kept: Option[M] = entry.map(_._3)
+}
diff --git a/obp-api/src/main/scala/code/api/dynamic/entity/helper/DynamicEntityHelper.scala b/obp-api/src/main/scala/code/api/dynamic/entity/helper/DynamicEntityHelper.scala
index ef657f47e0..262d5809ce 100644
--- a/obp-api/src/main/scala/code/api/dynamic/entity/helper/DynamicEntityHelper.scala
+++ b/obp-api/src/main/scala/code/api/dynamic/entity/helper/DynamicEntityHelper.scala
@@ -27,6 +27,7 @@ TESOBE (http://www.tesobe.com/)
 
 package code.api.dynamic.entity.helper
 
+import code.api.Constant
 import code.api.util.APIUtil.{EmptyBody, ResourceDoc, userAuthenticationMessage}
 import code.api.util.ApiRole.getOrCreateDynamicApiRole
 import code.api.util.ApiTag._
@@ -202,35 +203,44 @@ object DynamicEntityHelper {
   // It is asked for many times per request (the access checks, each field's read check, the resource
   // docs), and building it reads every definition and parses each one's JSON: with 120 definitions
   // that was ~5 ms a time, and a Dynamic Query returning 11 records spent ~0.5 s rebuilding it. So
-  // the built map is kept for dynamicEntity.definitions_map.cache.ttl.seconds (default 300), and
-  // forgotten the moment a definition is created, updated or deleted on this node
-  // (MappedDynamicEntityProvider calls forgetDefinitions). Another node sees such a change when its
-  // copy expires. Not kept in test mode, where a test may change the table behind the provider.
+  // the built map is kept for dynamicEntity.definitions_map.cache.ttl.seconds (default 300), and only
+  // while the dynamic_entity cache namespace version is the one it was built under: a definition change
+  // bumps that version once committed (forgetDefinitions), and every instance, which re-reads the version
+  // at most once a second, then builds the map again. Not kept in test mode, where a test may change the
+  // table behind the provider.
   private val definitionsMapTtlMillis: Long =
     if (net.liftweb.util.Props.testMode) 0L
     else APIUtil.getPropsAsLongValue("dynamicEntity.definitions_map.cache.ttl.seconds", 300L) * 1000L
-  // (built at, map). Each change replaces the token, and a build keeps its map only while the token it
-  // started under is still the current one, so a build that started before a change is not kept after it.
-  @volatile private var definitionsMapCache: Option[(Long, Map[(String, String), DynamicEntityInfo])] = None
-  @volatile private var definitionsToken: AnyRef = new Object
-
-  /** Forget the kept definitions map, so the next use reads the definitions again. */
-  def forgetDefinitions(): Unit = {
-    definitionsToken = new Object
-    definitionsMapCache = None
-  }
 
-  def definitionsMap: Map[(String, String), DynamicEntityInfo] = {
-    val now = System.currentTimeMillis()
-    definitionsMapCache match {
-      case Some((builtAt, map)) if now - builtAt < definitionsMapTtlMillis => map
-      case _ =>
-        val token = definitionsToken
-        val map = buildDefinitionsMap
-        if (definitionsMapTtlMillis > 0 && (definitionsToken eq token)) definitionsMapCache = Some((now, map))
-        map
+  private val definitionsMapCache = new DefinitionsMapCache[Map[(String, String), DynamicEntityInfo]](
+    definitionsMapTtlMillis,
+    () => Constant.recentCacheNamespaceVersion(Constant.DYNAMIC_ENTITY_NAMESPACE),
+    () => buildDefinitionsMap)
+
+  /**
+   * Tell every instance that a definition was created, updated or deleted, once the request's transaction
+   * has committed (at once outside a transaction): bump the dynamic_entity cache namespace version, which
+   * the definitions map and the definition lists NewStyle.getDynamicEntities memoizes in Redis both carry,
+   * so all of them read the definitions again, and then the dynamic resource-doc caches, which hold the
+   * docs generated from the definitions. Not earlier, because until the commit the change is not visible
+   * to other connections: an instance that saw a bump first could read the old rows and keep them under the
+   * new version. Nothing happens if the transaction rolls back.
+   *
+   * This instance also drops its map directly, which is what keeps it current when Redis is unavailable
+   * and the version cannot change. Another instance follows the dynamic_entity version within a second, so
+   * for that second it can still build resource docs from its previous map.
+   */
+  def forgetDefinitions(): Unit =
+    code.api.util.http4s.RequestScopeConnection.afterCommit {
+      Constant.incrementCacheNamespaceVersion(Constant.DYNAMIC_ENTITY_NAMESPACE)
+      definitionsMapCache.forget()
+      NewStyle.function.invalidateDynamicResourceDocCaches()
     }
-  }
+
+  def definitionsMap: Map[(String, String), DynamicEntityInfo] = definitionsMapCache.get
+
+  /** The number of definitions in this instance's kept map, for the cache info page. */
+  def keptDefinitionCount: Int = definitionsMapCache.kept.map(_.size).getOrElse(0)
 
   private def buildDefinitionsMap: Map[(String, String), DynamicEntityInfo] = NewStyle.function.getDynamicEntities(None, true).map(it => ((DynamicEntitySpace.bankIdOrSystem(it.bankId), it.entityName), DynamicEntityInfo(it.metadataJson, it.entityName, it.bankId, it.hasPersonalEntity, it.hasPublicAccess, it.hasCommunityAccess, it.personalRequiresRole, it.useRowLevelAccess, it.authMode))).toMap
 
diff --git a/obp-api/src/main/scala/code/api/util/CommonsEmailWrapper.scala b/obp-api/src/main/scala/code/api/util/CommonsEmailWrapper.scala
index 306947d158..65f96a048d 100644
--- a/obp-api/src/main/scala/code/api/util/CommonsEmailWrapper.scala
+++ b/obp-api/src/main/scala/code/api/util/CommonsEmailWrapper.scala
@@ -51,9 +51,17 @@ object CommonsEmailWrapper extends MdcLoggable {
     useTLS: Boolean = true,
     useSSL: Boolean = false,
     debug: Boolean = false,
-    tlsProtocols: String = "TLSv1.2"
+    tlsProtocols: String = "TLSv1.2",
+    timeoutMillis: Int = CommonsEmailWrapper.defaultTimeoutMillis
   )
 
+  val defaultTimeoutMillis = 30000
+
+  // mail.smtp.timeout_millis as a positive number of milliseconds; anything else falls back to the default,
+  // because Jakarta Mail reads 0 (and Lift reads an unparseable value as 0) as "wait forever".
+  def timeoutMillisFrom(value: Option[String]): Int =
+    value.flatMap(v => scala.util.Try(v.trim.toInt).toOption).filter(_ > 0).getOrElse(defaultTimeoutMillis)
+
   case class EmailContent(
     from: String,
     to: List[String],
@@ -80,13 +88,33 @@ object CommonsEmailWrapper extends MdcLoggable {
       useTLS = APIUtil.getPropsValue("mail.smtp.starttls.enable", "false").toBoolean,
       useSSL = APIUtil.getPropsValue("mail.smtp.ssl.enable", "false").toBoolean,
       debug = APIUtil.getPropsValue("mail.debug", "false").toBoolean,
-      tlsProtocols = APIUtil.getPropsValue("mail.smtp.ssl.protocols", "TLSv1.2")
+      tlsProtocols = APIUtil.getPropsValue("mail.smtp.ssl.protocols", "TLSv1.2"),
+      timeoutMillis = timeoutMillisFrom(APIUtil.getPropsValue("mail.smtp.timeout_millis").toOption)
     )
   }
 
   def isTestMode: Boolean = APIUtil.getPropsValue("mail.test.mode", "false").toBoolean
 
+  // What logEmailInTestMode logs (sendTextEmail, sendHtmlEmail, sendEmailWithAttachments), oldest first and
+  // bounded, so tests can check what OBP has sent and when. Kept only when OBP runs its tests (Props.testMode): a
+  // sandbox in mail test mode must not hold challenge codes and reset links in memory.
+  private val testModeMailLimit = 200
+  private val testModeMail = new java.util.concurrent.ConcurrentLinkedQueue[EmailContent]()
+  def testModeSentMail: List[EmailContent] = {
+    import scala.jdk.CollectionConverters._
+    testModeMail.asScala.toList
+  }
+  def clearTestModeSentMail(): Unit = testModeMail.clear()
+
   private def logEmailInTestMode(content: EmailContent): Box[String] = {
+    if (net.liftweb.util.Props.testMode) {
+      // mail.test.mode.delay_millis makes a send take that long, as an SMTP round trip does, so a test can tell
+      // whether a request waited for its mail.
+      val delay = APIUtil.getPropsAsIntValue("mail.test.mode.delay_millis", 0)
+      if (delay > 0) Thread.sleep(delay.toLong)
+      testModeMail.add(content)
+      while (testModeMail.size > testModeMailLimit) testModeMail.poll()
+    }
     logger.info("=" * 80)
     logger.info("EMAIL TEST MODE - Email would be sent with the following content:")
     logger.info("=" * 80)
@@ -291,7 +319,7 @@ object CommonsEmailWrapper extends MdcLoggable {
     }
   }
 
-  private def createSession(config: EmailConfig): Session = {
+  def sessionProperties(config: EmailConfig): Properties = {
     val props = new Properties()
     props.put("mail.smtp.host", config.smtpHost)
     props.put("mail.smtp.port", config.smtpPort.toString)
@@ -300,6 +328,15 @@ object CommonsEmailWrapper extends MdcLoggable {
     props.put("mail.smtp.ssl.enable", config.useSSL.toString)
     props.put("mail.debug", config.debug.toString)
     props.put("mail.smtp.ssl.protocols", config.tlsProtocols)
+    // Jakarta Mail waits without limit by default; mail is sent inside requests, which must not hang on SMTP.
+    props.put("mail.smtp.connectiontimeout", config.timeoutMillis.toString)
+    props.put("mail.smtp.timeout", config.timeoutMillis.toString)
+    props.put("mail.smtp.writetimeout", config.timeoutMillis.toString)
+    props
+  }
+
+  private def createSession(config: EmailConfig): Session = {
+    val props = sessionProperties(config)
     val authenticator = new Authenticator() {
       override def getPasswordAuthentication: PasswordAuthentication =
         new PasswordAuthentication(config.username, config.password)
diff --git a/obp-api/src/main/scala/code/api/util/NewStyle.scala b/obp-api/src/main/scala/code/api/util/NewStyle.scala
index 40f7489a6a..e0f9842f7e 100644
--- a/obp-api/src/main/scala/code/api/util/NewStyle.scala
+++ b/obp-api/src/main/scala/code/api/util/NewStyle.scala
@@ -3543,12 +3543,14 @@ object NewStyle extends MdcLoggable{
      * next `/resource-docs` request regenerates from the database instead of
      * serving a pre-change snapshot (TTL is 1 hour by default).
      *
-     * Call after a Dynamic Entity, Dynamic Endpoint or Dynamic Resource Doc is
-     * created, updated, or deleted: each one generates resource docs. The static
-     * resource-doc / swagger caches are not touched because dynamic docs only
-     * reach them through the content=all documents, which are left to their TTL.
+     * Call after a Dynamic Endpoint or Dynamic Resource Doc is created, updated,
+     * or deleted: each one generates resource docs. For a Dynamic Entity,
+     * DynamicEntityHelper.forgetDefinitions calls it once the change has
+     * committed. The static resource-doc / swagger caches are not touched because
+     * dynamic docs only reach them through the content=all documents, which are
+     * left to their TTL.
      */
-    private def invalidateDynamicResourceDocCaches(): Unit = {
+    private[code] def invalidateDynamicResourceDocCaches(): Unit = {
       Constant.incrementCacheNamespaceVersion(Constant.RD_DYNAMIC_NAMESPACE)
       Constant.incrementCacheNamespaceVersion(Constant.RD_ALL_NAMESPACE)
     }
@@ -3589,9 +3591,9 @@ object NewStyle extends MdcLoggable{
       }
 
       Future {
-        val result = DynamicEntityProvider.connectorMethodProvider.vend.createOrUpdate(dynamicEntity)
-        if (result.isDefined) invalidateDynamicResourceDocCaches()
-        result
+        // The provider calls DynamicEntityHelper.forgetDefinitions, which also drops the dynamic
+        // resource-doc caches once the change has committed.
+        DynamicEntityProvider.connectorMethodProvider.vend.createOrUpdate(dynamicEntity)
       }
     }
 
@@ -3630,9 +3632,9 @@ object NewStyle extends MdcLoggable{
       }
 
       Future {
-        val result = DynamicEntityProvider.connectorMethodProvider.vend.createOrUpdate(dynamicEntity)
-        if (result.isDefined) invalidateDynamicResourceDocCaches()
-        result
+        // The provider calls DynamicEntityHelper.forgetDefinitions, which also drops the dynamic
+        // resource-doc caches once the change has committed.
+        DynamicEntityProvider.connectorMethodProvider.vend.createOrUpdate(dynamicEntity)
       }
 
     }
@@ -3674,8 +3676,8 @@ object NewStyle extends MdcLoggable{
           }
           deleteEntitleMentResult
         }
-        // The entity (and its generated resource docs) are gone — drop the stale doc caches.
-        if (result.exists(deleted => deleted)) invalidateDynamicResourceDocCaches()
+        // The entity's generated resource docs go with it: the provider's delete calls
+        // DynamicEntityHelper.forgetDefinitions, which drops the doc caches once the delete has committed.
         result
       }
     }
@@ -3704,9 +3706,27 @@ object NewStyle extends MdcLoggable{
     }
 
     def getDynamicEntities(bankId: Option[String], returnBothBankAndSystemLevel: Boolean): List[DynamicEntityT] = {
-      import scala.concurrent.duration._
-
       validateBankId(bankId, None)
+      getDynamicEntities(bankId, returnBothBankAndSystemLevel, dynamicEntityDefinitionsVersion)
+    }
+
+    /**
+     * The dynamic_entity cache namespace version, which the memoized definition lists below carry in their
+     * keys. DynamicEntityHelper.forgetDefinitions bumps it once a definition change has committed, so every
+     * list memoized before the change becomes unreachable on every instance, including one that a read still
+     * in flight writes after the bump: it is written under the version that read started with.
+     */
+    private def dynamicEntityDefinitionsVersion: Long =
+      if (dynamicEntityTTL == 0) 0L // nothing is memoized, so no need to ask Redis
+      else Constant.recentCacheNamespaceVersion(Constant.DYNAMIC_ENTITY_NAMESPACE)
+
+    // The version is a parameter so that CacheKeyFromArguments renders it into the key; the name stays
+    // getDynamicEntities so the key and its Telemetry label stay the same apart from the version. Keep
+    // buildCacheKey as the tail expression (see getEndpointMappingsCached): bound to a val first, the macro
+    // renders no arguments and every list would share one key whatever the version. Test mode memoizes
+    // nothing, so no test would notice; the rendering was checked with javap.
+    private def getDynamicEntities(bankId: Option[String], returnBothBankAndSystemLevel: Boolean, definitionsVersion: Long): List[DynamicEntityT] = {
+      import scala.concurrent.duration._
 
       var cacheKey = (randomUUID().toString, randomUUID().toString, randomUUID().toString)
       CacheKeyFromArguments.buildCacheKey {
@@ -3715,8 +3735,11 @@ object NewStyle extends MdcLoggable{
         }
       }
     }
-    
-    def getDynamicEntitiesByUserId(userId: String): List[DynamicEntityT] = {
+
+    def getDynamicEntitiesByUserId(userId: String): List[DynamicEntityT] =
+      getDynamicEntitiesByUserId(userId, dynamicEntityDefinitionsVersion)
+
+    private def getDynamicEntitiesByUserId(userId: String, definitionsVersion: Long): List[DynamicEntityT] = {
       import scala.concurrent.duration._
 
       var cacheKey = (randomUUID().toString, randomUUID().toString, randomUUID().toString)
diff --git a/obp-api/src/main/scala/code/api/v5_0_0/Http4s500.scala b/obp-api/src/main/scala/code/api/v5_0_0/Http4s500.scala
index 74612b799b..9eceda13ed 100644
--- a/obp-api/src/main/scala/code/api/v5_0_0/Http4s500.scala
+++ b/obp-api/src/main/scala/code/api/v5_0_0/Http4s500.scala
@@ -104,7 +104,7 @@ object Http4s500 {
   val versionStatus: String = ApiVersionStatus.STABLE.toString
   val resourceDocs: ArrayBuffer[ResourceDoc] = ArrayBuffer[ResourceDoc]()
 
-  object Implementations5_0_0 {
+  object Implementations5_0_0 extends code.util.Helper.MdcLoggable {
 
     val prefixPath = Root / ApiPathZero.toString / implementedInApiVersion.toString
 
@@ -1058,34 +1058,33 @@ object Http4s500 {
     // POST /consumer/consent-requests/CONSENT_REQUEST_ID/{EMAIL|SMS|IMPLICIT}/consents → 201
     // Three ResourceDoc registrations (one per SCA literal) but one HttpRoutes pattern.
 
-    private def sendEmailConsentNotification(
-      callContextOpt: Option[code.api.util.CallContext],
+    // Where the challenge for a consent goes: the SCA method and its recipient, or None when this method sends
+    // nothing. Fails with a 400 when the challenge has nowhere to go.
+    private def resolveChallengeRecipient(
+      scaMethod: String,
       consentRequestJson: PostConsentRequestJsonV500,
-      challengeText: String
-    ): Future[String] =
-      for {
-        consentScaEmail <- NewStyle.function.tryons(
-          s"$InvalidJsonFormat The Json body must contain the field email", 400, callContextOpt) {
-          consentRequestJson.email.head
-        }
-        (status, _) <- NewStyle.function.sendCustomerNotification(
-          StrongCustomerAuthentication.EMAIL, consentScaEmail,
-          Some("OBP Consent Challenge"), challengeText, callContextOpt)
-      } yield status
-
-    private def sendSmsConsentNotification(
-      callContextOpt: Option[code.api.util.CallContext],
-      consentRequestJson: PostConsentRequestJsonV500,
-      challengeText: String
-    ): Future[String] =
-      for {
-        consentScaPhoneNumber <- NewStyle.function.tryons(
-          s"$InvalidJsonFormat The Json body must contain the field phone_number", 400, callContextOpt) {
-          consentRequestJson.phone_number.head
-        }
-        (status, _) <- NewStyle.function.sendCustomerNotification(
-          StrongCustomerAuthentication.SMS, consentScaPhoneNumber, None, challengeText, callContextOpt)
-      } yield status
+      user: User,
+      callContextOpt: Option[code.api.util.CallContext]
+    ): Future[Option[(StrongCustomerAuthentication, String)]] = {
+      def required(method: StrongCustomerAuthentication, recipient: Option[String], failMsg: String) =
+        Helper.booleanToFuture(failMsg, 400, callContextOpt)(recipient.exists(r => r != null && r.trim.nonEmpty))
+          .map(_ => Some((method, recipient.get)))
+      scaMethod match {
+        case v if v == StrongCustomerAuthentication.EMAIL.toString =>
+          required(StrongCustomerAuthentication.EMAIL, consentRequestJson.email, s"$InvalidJsonFormat The Json body must contain the field email")
+        case v if v == StrongCustomerAuthentication.SMS.toString =>
+          required(StrongCustomerAuthentication.SMS, consentRequestJson.phone_number, s"$InvalidJsonFormat The Json body must contain the field phone_number")
+        case v if v == StrongCustomerAuthentication.IMPLICIT.toString =>
+          NewStyle.function.getConsentImplicitSCA(user, callContextOpt).flatMap { case (implicitSca, _) =>
+            implicitSca.scaMethod match {
+              case m @ (StrongCustomerAuthentication.EMAIL | StrongCustomerAuthentication.SMS) =>
+                required(m, Option(implicitSca.recipient), s"$GetConsentImplicitSCAError The user has no $m address")
+              case _ => Future.successful(None)
+            }
+          }
+        case _ => Future.successful(None)
+      }
+    }
 
     val createConsentByConsentRequestId: Http4sRoute = Http4sRoute {
       case req @ POST -> `prefixPath` / "consumer" / "consent-requests" / consentRequestId / scaMethod / "consents"
@@ -1121,6 +1120,28 @@ object Http4s500 {
                   json.parse(createdConsentRequest.payload).extract[PostConsentRequestJsonV500]
                 }.map(p => (p, false))
               }
+            // Every Consent names the Consumer it is for: the body's consumer_id, else the Consumer that
+            // lodged the consent request, else the caller. NewStyle.function.resolveConsentConsumer says
+            // why a Consent may not be created without one. One resolution feeds both the JWT's aud and
+            // the stored row, so the two can no longer disagree about who the Consent is for.
+            consentConsumer <- NewStyle.function.resolveConsentConsumer(
+              consentRequestJson.consumer_id.orElse(Option(createdConsentRequest.consumerId)), callContextOpt)
+            grantorConsumerId = callContextOpt.flatMap(_.consumer.toOption.map(_.consumerId.get)).getOrElse("Unknown")
+            granteeConsumerId = consentConsumer.consumerId.get
+            shouldSkipConsentScaForConsumerIdPair = APIUtil.skipConsentScaForConsumerIdPairs.contains(
+              APIUtil.ConsumerIdPair(grantorConsumerId, granteeConsumerId))
+            // Where the challenge goes, settled before anything is written (the VRP view, counterparty and limit
+            // below, then the consent): a request that fails later still keeps what it wrote, and a consent whose
+            // challenge has nowhere to go could never be answered.
+            challengeRecipient <- if (shouldSkipConsentScaForConsumerIdPair) Future.successful(None)
+                                  else resolveChallengeRecipient(scaMethod, consentRequestJson, user, callContextOpt)
+            maxTimeToLive = APIUtil.getPropsAsIntValue(nameOfProperty = "consents.max_time_to_live", defaultValue = DEFAULT_CONSENT_TTL)
+            _ <- Helper.booleanToFuture(s"$ConsentMaxTTL ($maxTimeToLive)", cc = callContextOpt) {
+              consentRequestJson.time_to_live match {
+                case Some(ttl) => ttl <= maxTimeToLive
+                case _         => true
+              }
+            }
             (bankId, accountId, viewId, counterpartyId) <- if (isVRPConsentRequest) {
               val postConsentRequestJsonV510 = json.parse(createdConsentRequest.payload).extract[code.api.v5_1_0.PostVRPConsentRequestJsonV510]
               val vrpViewId = s"${VRP_VIEW_ID_PREFIX}${UUID.randomUUID.toString}".dropRight(5)
@@ -1234,13 +1255,6 @@ object Http4s500 {
             } else {
               Future.successful((BankId(""), AccountId(""), ViewId(""), CounterpartyId(""))): Future[(BankId, AccountId, ViewId, CounterpartyId)]
             }
-            maxTimeToLive = APIUtil.getPropsAsIntValue(nameOfProperty = "consents.max_time_to_live", defaultValue = DEFAULT_CONSENT_TTL)
-            _ <- Helper.booleanToFuture(s"$ConsentMaxTTL ($maxTimeToLive)", cc = callContextOpt) {
-              consentRequestJson.time_to_live match {
-                case Some(ttl) => ttl <= maxTimeToLive
-                case _         => true
-              }
-            }
             requestedEntitlements = consentRequestJson.entitlements.getOrElse(Nil)
             myEntitlements <- Entitlement.entitlement.vend.getEntitlementsByUserIdFuture(user.userId)
             // CanCreateEntitlementAtAnyBank stays out of consents. CanCreateEntitlementAtOneBank is
@@ -1276,12 +1290,6 @@ object Http4s500 {
                          e.view_id == rv.view_id && e.bank_id == rv.bank_id && e.account_id == rv.account_id))
                    }
                  } yield ()
-            // Every Consent names the Consumer it is for: the body's consumer_id, else the Consumer that
-            // lodged the consent request, else the caller. NewStyle.function.resolveConsentConsumer says
-            // why a Consent may not be created without one. One resolution feeds both the JWT's aud and
-            // the stored row, so the two can no longer disagree about who the Consent is for.
-            consentConsumer <- NewStyle.function.resolveConsentConsumer(
-              consentRequestJson.consumer_id.orElse(Option(createdConsentRequest.consumerId)), callContextOpt)
             consumerIdOpt = Some(consentConsumer.consumerId.get)
             applicationText = consentConsumer.description
             challengeAnswer = Props.mode match {
@@ -1324,10 +1332,6 @@ object Http4s500 {
             validUntil = Helper.calculateValidTo(postConsentBodyCommonJson.valid_from, postConsentBodyCommonJson.time_to_live.getOrElse(3600))
             _ <- Future(Consents.consentProvider.vend.setValidUntil(createdConsent.consentId, validUntil))
               .map(i => connectorEmptyResponse(i, callContextOpt))
-            grantorConsumerId = callContextOpt.flatMap(_.consumer.toOption.map(_.consumerId.get)).getOrElse("Unknown")
-            granteeConsumerId = consentConsumer.consumerId.get
-            shouldSkipConsentScaForConsumerIdPair = APIUtil.skipConsentScaForConsumerIdPairs.contains(
-              APIUtil.ConsumerIdPair(grantorConsumerId, granteeConsumerId))
             mappedConsent <- if (shouldSkipConsentScaForConsumerIdPair) {
               Future {
                 // Atomic guarded auto-accept: only move INITIATED -> ACCEPTED. If the consent was
@@ -1340,25 +1344,33 @@ object Http4s500 {
               }
             } else {
               val challengeText = s"Your consent challenge : ${challengeAnswer}, Application: $applicationText"
-              scaMethod match {
-                case v if v == StrongCustomerAuthentication.EMAIL.toString =>
-                  sendEmailConsentNotification(callContextOpt, consentRequestJson, challengeText)
-                case v if v == StrongCustomerAuthentication.SMS.toString =>
-                  sendSmsConsentNotification(callContextOpt, consentRequestJson, challengeText)
-                case v if v == StrongCustomerAuthentication.IMPLICIT.toString =>
-                  for {
-                    (consentImplicitSCA, _) <- NewStyle.function.getConsentImplicitSCA(user, callContextOpt)
-                    _ <- consentImplicitSCA.scaMethod match {
-                      case v if v == StrongCustomerAuthentication.EMAIL =>
-                        sendEmailConsentNotification(callContextOpt, consentRequestJson.copy(email = Some(consentImplicitSCA.recipient)), challengeText)
-                      case v if v == StrongCustomerAuthentication.SMS =>
-                        sendSmsConsentNotification(callContextOpt, consentRequestJson.copy(phone_number = Some(consentImplicitSCA.recipient)), challengeText)
-                      case _ => Future.successful("Success")
-                    }
-                  } yield "Success"
-                case _ => Future.successful("Success")
+              // The challenge is sent within the request, as by the other Create Consent endpoints: the connector
+              // looks up its method routing on the request's database connection, which closes with the request.
+              // A challenge that cannot be sent fails the request, so no consent is kept that no one can answer.
+              val challengeSent: Future[Unit] = challengeRecipient match {
+                case Some((method, recipient)) =>
+                  val subject = if (method == StrongCustomerAuthentication.EMAIL) Some("OBP Consent Challenge") else None
+                  // Started inside the Future, so a connector that throws before building its own Future fails
+                  // this one rather than the request.
+                  Future.unit.flatMap(_ =>
+                    NewStyle.function.sendCustomerNotification(method, recipient, subject, challengeText, callContextOpt)).map(_ => ())
+                case None => Future.successful(())
+              }
+              // The consent is already written when a send fails (an SMS provider that is missing or refuses), and the
+              // failed request keeps it: reject it, and release a VRP mandate the way a revoke does, so no one is left
+              // with a consent they can never answer or with a payment authority it no longer stands for. The send's
+              // error is what the caller gets, even if cleaning up fails too.
+              challengeSent.map(_ => createdConsent).recoverWith { case e =>
+                Future {
+                  Consents.consentProvider.vend.updateConsentStatus(createdConsent.consentId, ConsentStatus.REJECTED)
+                  // Read again: the JWT that names the VRP view was saved on another instance.
+                  Consents.consentProvider.vend.getConsentByConsentId(createdConsent.consentId).foreach(Consent.revokeConsentAccountAccess)
+                }.transformWith { cleanup =>
+                  cleanup.failed.foreach(f => logger.warn(s"createConsentByConsentRequestId says: could not reject consent " +
+                    s"${createdConsent.consentId} after its challenge failed: ${f.getClass.getSimpleName}"))
+                  Future.failed(e)
+                }
               }
-              Future(createdConsent)
             }
           } yield ConsentJsonV500(
             mappedConsent.consentId,
@@ -1418,6 +1430,8 @@ object Http4s500 {
       |
       |An SCA challenge code is sent to the phone number that was supplied in the Create Consent Request body. The User then completes SCA via Answer Consent Challenge, which moves the Consent from INITIATED to ACCEPTED.
       |
+      |The challenge is sent before this call answers. If it cannot be sent (no SMS provider configured, or the provider refuses), the call fails and the Consent it created is REJECTED; the Consent Request is then used up, and the TPP has to lodge a new one.
+      |
       |Pinning: the resulting Consent is pinned to a single consumer at creation. The pinned consumer is taken from the `consumer_id` field of the original Create Consent Request body if present, otherwise from the consumer that created the Request, otherwise from the consumer making this call. If none of those names a consumer the Consent is not created and this endpoint returns ConsentConsumerIsRequired — a Consent that names no consumer could be presented by nobody. After creation, only the pinned consumer can present the resulting Consent JWT — any other consumer presenting it gets ConsentNotFound (consumer mismatch).
       |
       |Each Consent Request can be answered exactly once. A second call returns ConsentRequestIsInvalid.
@@ -1439,11 +1453,11 @@ object Http4s500 {
       "/consumer/consent-requests/CONSENT_REQUEST_ID/IMPLICIT/consents",
       "Create Consent By CONSENT_REQUEST_ID (IMPLICIT)",
       s"""
-      |Answer a Consent Request and create the resulting Consent, without a Strong Customer Authentication challenge — the Consent is moved directly from INITIATED to ACCEPTED.
+      |Answer a Consent Request and create the resulting Consent, with the Strong Customer Authentication challenge the User's implicit SCA method names.
       |
       |After the TPP has called Create Consent Request (Client Credentials), the User authenticates and answers the request via this endpoint. This creates the Consent (the credential the consumer will use to access OBP on the User's behalf).
       |
-      |IMPLICIT means no SCA challenge is sent. The Consent is immediately ACCEPTED. Use only in flows where the User has already been strongly authenticated by upstream means; for production use behind a public TPP, prefer EMAIL or SMS.
+      |IMPLICIT lets the instance choose how the challenge reaches the User: the connector's getConsentImplicitSCA names the method and the recipient (on the mapped connector, an EMAIL to the User's email address). The challenge is sent before this call answers, and the User completes SCA via Answer Consent Challenge, which moves the Consent from INITIATED to ACCEPTED. If the User has no address for that method the call fails with GetConsentImplicitSCAError before any Consent is created.
       |
       |Pinning: the resulting Consent is pinned to a single consumer at creation. The pinned consumer is taken from the `consumer_id` field of the original Create Consent Request body if present, otherwise from the consumer that created the Request, otherwise from the consumer making this call. If none of those names a consumer the Consent is not created and this endpoint returns ConsentConsumerIsRequired — a Consent that names no consumer could be presented by nobody. After creation, only the pinned consumer can present the resulting Consent JWT — any other consumer presenting it gets ConsentNotFound (consumer mismatch).
       |
@@ -1455,7 +1469,7 @@ object Http4s500 {
       |
       |""",
       EmptyBody, consentJsonV500,
-      ConsentRequestIsInvalid :: MissingPropsValueAtThisInstance :: SmsServerNotResponding :: createConsentByConsentRequestIdCommonErrors,
+      ConsentRequestIsInvalid :: MissingPropsValueAtThisInstance :: SmsServerNotResponding :: GetConsentImplicitSCAError :: createConsentByConsentRequestIdCommonErrors,
       apiTagConsent :: apiTagPSD2AIS :: apiTagPsd2 :: Nil,
       None,
       http4sPartialFunction = Some(createConsentByConsentRequestId)
diff --git a/obp-api/src/main/scala/code/api/v6_0_0/JSONFactory6.0.0.scala b/obp-api/src/main/scala/code/api/v6_0_0/JSONFactory6.0.0.scala
index 98539177ca..3823a8bce4 100644
--- a/obp-api/src/main/scala/code/api/v6_0_0/JSONFactory6.0.0.scala
+++ b/obp-api/src/main/scala/code/api/v6_0_0/JSONFactory6.0.0.scala
@@ -2444,14 +2444,16 @@ object JSONFactory600 extends CustomJsonFormats with MdcLoggable {
       Constant.FINANCIAL_PRODUCTS_NAMESPACE -> ("Financial product list (bank-scoped and all-banks)", "Products"),
       Constant.API_PRODUCTS_NAMESPACE -> ("Api product list (all banks)", "Products"),
       Constant.MESSAGE_DOCS_NAMESPACE -> ("Message docs and connector JSON Schemas (Redis, and each instance's memory)", "API Documentation"),
-      Constant.GLOSSARY_NAMESPACE -> ("Glossary items held in each instance's memory; bumping also rebuilds cached resource docs", "API Documentation")
+      Constant.GLOSSARY_NAMESPACE -> ("Glossary items held in each instance's memory; bumping also rebuilds cached resource docs", "API Documentation"),
+      Constant.DYNAMIC_ENTITY_NAMESPACE -> ("Dynamic entity definitions: the definition lists in Redis and each instance's definitions map", "Dynamic Entities")
     )
 
     // Caches held in this instance's memory outside the shared in-memory store, which follow their
     // namespace's version through their own keys. Counted here so the page shows their real size.
     val inProcessEntries: Map[String, () => Long] = Map(
       Constant.MESSAGE_DOCS_NAMESPACE -> (() => code.api.v2_2_0.MessageDocsJsonCache.size + code.api.util.JsonSchemaGenerator.cacheSize),
-      Constant.GLOSSARY_NAMESPACE -> (() => code.api.util.Glossary.loadedItemCount.toLong)
+      Constant.GLOSSARY_NAMESPACE -> (() => code.api.util.Glossary.loadedItemCount.toLong),
+      Constant.DYNAMIC_ENTITY_NAMESPACE -> (() => code.api.dynamic.entity.helper.DynamicEntityHelper.keptDefinitionCount.toLong)
     )
 
     var redisAvailable = true
diff --git a/obp-api/src/main/scala/code/api/v7_0_0/Http4s700.scala b/obp-api/src/main/scala/code/api/v7_0_0/Http4s700.scala
index cff7c92f5c..3b04f334a7 100644
--- a/obp-api/src/main/scala/code/api/v7_0_0/Http4s700.scala
+++ b/obp-api/src/main/scala/code/api/v7_0_0/Http4s700.scala
@@ -6852,7 +6852,8 @@ object Http4s700 {
       payload_hash = "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08",
       current_payload_hash = "",
       proposed_payload = Extraction.decompose(jsonDynamicResourceDoc),
-      current_payload = JNothing,
+      // A CREATE has nothing to replace. The swagger export needs an object here, not JNothing.
+      current_payload = JObject(Nil),
       requestor_user_id = code.api.util.ExampleValue.userIdExample.value,
       business_justification = "Expose the new loan quote endpoint for the mobile app.",
       checker_user_id = "",
diff --git a/obp-api/src/main/scala/deletion/DeleteAccountCascade.scala b/obp-api/src/main/scala/deletion/DeleteAccountCascade.scala
index f3756634ea..0a1f1eb81f 100644
--- a/obp-api/src/main/scala/deletion/DeleteAccountCascade.scala
+++ b/obp-api/src/main/scala/deletion/DeleteAccountCascade.scala
@@ -33,8 +33,13 @@ import code.api.util.APIUtil.fullBoxOrException
 import code.api.util.ErrorMessages.CouldNotDeleteCascade
 import code.bankconnectors.Connector
 import code.cards.MappedPhysicalCard
+import code.counterpartyattribute.CounterpartyAttribute
+import code.counterpartylimit.CounterpartyLimit
 import code.entitlement.MappedEntitlement
-import code.model.dataAccess.{BankAccountRouting, MappedBankAccount, MappedBankAccountData}
+import code.metadata.counterparties.{MappedCounterparty, MappedCounterpartyBespoke, MappedCounterpartyMetadata, MappedCounterpartyWhereTag}
+import code.model.dataAccess.{BankAccountRouting, DoubleEntryBookTransaction, MappedBankAccount, MappedBankAccountData}
+import code.transactionChallenge.MappedExpectedChallengeAnswer
+import code.transactionrequests.MappedTransactionRequest
 import code.views.system.{AccountAccess, ViewDefinition}
 import code.webhook.MappedAccountWebhook
 import com.openbankproject.commons.model.{AccountId, BankId}
@@ -49,16 +54,23 @@ import scala.collection.immutable.List
 object DeleteAccountCascade {
 
   def delete(bankId: BankId, accountId: AccountId): Boolean = {
+    // The tasks run in list order: the challenges are found through the account's transaction
+    // requests, so they must go before deleteTransactionRequests and deleteTransactions remove them.
     val doneTasks =
-      deleteTransactions(bankId, accountId) ::
+      deleteTransactionRequestChallenges(bankId, accountId) ::
+        deleteTransactionRequests(bankId, accountId) ::
+        deleteDoubleEntryBookTransactions(bankId, accountId) ::
+        deleteTransactions(bankId, accountId) ::
         deleteEntitlements(bankId, accountId) ::
         deleteAccountAccess(bankId, accountId) ::
         deleteCustomViews(bankId, accountId) ::
         deleteAccountAttributes(bankId, accountId) ::
         deleteAccountWebhooks(bankId, accountId) ::
         deleteBankAccountData(bankId, accountId) ::
-        deleteCards(accountId) ::
+        deleteCards(bankId, accountId) ::
         deleteAccountRoutings(bankId, accountId) ::
+        deleteCounterparties(bankId, accountId) ::
+        deleteCounterpartyMetadata(bankId, accountId) ::
         deleteAccount(bankId, accountId) ::
         Nil
     doneTasks.forall(_ == true)
@@ -81,15 +93,19 @@ object DeleteAccountCascade {
     )
   }
   private def deleteEntitlements(bankId: BankId, accountId: AccountId): Boolean = {
-    val userIds = AccountAccess.findAll(By(AccountAccess.account_id, accountId.value)).map(_.user_fk.foreign.map(_.userId).getOrElse(""))
+    val userIds = AccountAccess.findAll(
+      By(AccountAccess.bank_id, bankId.value),
+      By(AccountAccess.account_id, accountId.value)
+    ).map(_.user_fk.foreign.map(_.userId).getOrElse(""))
     MappedEntitlement.bulkDelete_!!(
       By(MappedEntitlement.mBankId, bankId.value),
       ByList(MappedEntitlement.mUserId, userIds)
     )
   }
   
-  private def deleteCards(accountId: AccountId): Boolean = {
+  private def deleteCards(bankId: BankId, accountId: AccountId): Boolean = {
     MappedBankAccount.findAll(
+      By(MappedBankAccount.bank, bankId.value),
       By(MappedBankAccount.theAccountId, accountId.value)
     ) map (
       account =>
@@ -136,6 +152,46 @@ object DeleteAccountCascade {
     )
   }
 
+  // The counterparties this account created, with the rows that hang off each of them.
+  // Counterparties of other accounts that point at this account are theirs and are kept.
+  private def deleteCounterparties(bankId: BankId, accountId: AccountId): Boolean = {
+    val counterparties = MappedCounterparty.findAll(
+      By(MappedCounterparty.mThisBankId, bankId.value),
+      By(MappedCounterparty.mThisAccountId, accountId.value)
+    )
+    val primaryKeys = counterparties.map(_.id.get)
+    val counterpartyIds = counterparties.map(_.counterpartyId)
+    val bespokesDeleted = primaryKeys.isEmpty ||
+      MappedCounterpartyBespoke.bulkDelete_!!(ByList(MappedCounterpartyBespoke.mCounterparty, primaryKeys))
+    val attributesDeleted = counterpartyIds.isEmpty ||
+      CounterpartyAttribute.bulkDelete_!!(ByList(CounterpartyAttribute.CounterpartyId_, counterpartyIds))
+    val limitsDeleted = CounterpartyLimit.bulkDelete_!!(
+      By(CounterpartyLimit.BankId, bankId.value),
+      By(CounterpartyLimit.AccountId, accountId.value)
+    )
+    val counterpartiesDeleted = MappedCounterparty.bulkDelete_!!(
+      By(MappedCounterparty.mThisBankId, bankId.value),
+      By(MappedCounterparty.mThisAccountId, accountId.value)
+    )
+    bespokesDeleted && attributesDeleted && limitsDeleted && counterpartiesDeleted
+  }
+
+  // Metadata covers both the counterparties created on this account and the ones derived from its transactions.
+  private def deleteCounterpartyMetadata(bankId: BankId, accountId: AccountId): Boolean = {
+    val metadata = MappedCounterpartyMetadata.findAll(
+      By(MappedCounterpartyMetadata.thisBankId, bankId.value),
+      By(MappedCounterpartyMetadata.thisAccountId, accountId.value)
+    )
+    val locationIds = metadata.flatMap(m => List(m.physicalLocation.get, m.corporateLocation.get)).filter(_ > 0)
+    val locationsDeleted = locationIds.isEmpty ||
+      MappedCounterpartyWhereTag.bulkDelete_!!(ByList(MappedCounterpartyWhereTag.id, locationIds))
+    val metadataDeleted = MappedCounterpartyMetadata.bulkDelete_!!(
+      By(MappedCounterpartyMetadata.thisBankId, bankId.value),
+      By(MappedCounterpartyMetadata.thisAccountId, accountId.value)
+    )
+    locationsDeleted && metadataDeleted
+  }
+
   private def deleteTransactions(bankId: BankId, accountId: AccountId): Boolean = {
     val deletedTransactions: Box[List[Boolean]] =
       for(
@@ -153,4 +209,31 @@ object DeleteAccountCascade {
         false
     }
   }
+
+  private def deleteTransactionRequestChallenges(bankId: BankId, accountId: AccountId): Boolean = {
+    val transactionRequestIds = MappedTransactionRequest.findAll(
+      By(MappedTransactionRequest.mFrom_BankId, bankId.value),
+      By(MappedTransactionRequest.mFrom_AccountId, accountId.value)
+    ).map(_.mTransactionRequestId.get)
+    MappedExpectedChallengeAnswer.bulkDelete_!!(
+      ByList(MappedExpectedChallengeAnswer.TransactionRequestId, transactionRequestIds)
+    )
+  }
+  private def deleteTransactionRequests(bankId: BankId, accountId: AccountId): Boolean = {
+    MappedTransactionRequest.bulkDelete_!!(
+      By(MappedTransactionRequest.mFrom_BankId, bankId.value),
+      By(MappedTransactionRequest.mFrom_AccountId, accountId.value)
+    )
+  }
+  private def deleteDoubleEntryBookTransactions(bankId: BankId, accountId: AccountId): Boolean = {
+    val debitSide = DoubleEntryBookTransaction.bulkDelete_!!(
+      By(DoubleEntryBookTransaction.DebitTransactionBankId, bankId.value),
+      By(DoubleEntryBookTransaction.DebitTransactionAccountId, accountId.value)
+    )
+    val creditSide = DoubleEntryBookTransaction.bulkDelete_!!(
+      By(DoubleEntryBookTransaction.CreditTransactionBankId, bankId.value),
+      By(DoubleEntryBookTransaction.CreditTransactionAccountId, accountId.value)
+    )
+    debitSide && creditSide
+  }
 }
diff --git a/obp-api/src/test/scala/code/api/ResourceDocs1_4_0/SwaggerDocsTest.scala b/obp-api/src/test/scala/code/api/ResourceDocs1_4_0/SwaggerDocsTest.scala
index f4c6342ce9..4fc4e223cc 100644
--- a/obp-api/src/test/scala/code/api/ResourceDocs1_4_0/SwaggerDocsTest.scala
+++ b/obp-api/src/test/scala/code/api/ResourceDocs1_4_0/SwaggerDocsTest.scala
@@ -110,6 +110,15 @@ class SwaggerDocsTest extends ResourceDocsV140ServerSetup with PropsReset with D
       errors.isEmpty should be (true)
     }
     
+    scenario(s"We will test ${ApiEndpoint1.name} Api - v7.0.0 swagger export succeeds", ApiEndpoint1, VersionOfApi) {
+      val requestGetSwagger = (ResourceDocsV5_1Request / "resource-docs" / "v7.0.0" / "swagger").GET
+      val responseGetSwagger = makeGetRequest(requestGetSwagger)
+      And("We should get 200, not OBP-50000 because one entity has no example")
+      withClue(responseGetSwagger.body.toString) {
+        responseGetSwagger.code should equal(200)
+      }
+    }
+
     scenario(s"We will test ${ApiEndpoint1.name} Api - v4.0.0", ApiEndpoint1, VersionOfApi) {
       val requestGetObp = (ResourceDocsV4_0Request / "resource-docs" / "v4.0.0" / "swagger").GET
       val responseGetObp = makeGetRequest(requestGetObp)
diff --git a/obp-api/src/test/scala/code/api/dynamic/entity/helper/DefinitionsCacheInvalidationTest.scala b/obp-api/src/test/scala/code/api/dynamic/entity/helper/DefinitionsCacheInvalidationTest.scala
new file mode 100644
index 0000000000..516ee8e3e2
--- /dev/null
+++ b/obp-api/src/test/scala/code/api/dynamic/entity/helper/DefinitionsCacheInvalidationTest.scala
@@ -0,0 +1,197 @@
+/**
+Open Bank Project - API
+Copyright (C) 2011-2026, TESOBE GmbH.
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+Email: contact@tesobe.com
+TESOBE GmbH.
+Osloer Strasse 16/17
+Berlin 13359, Germany
+
+This product includes software developed at
+TESOBE (http://www.tesobe.com/)
+
+  */
+
+package code.api.dynamic.entity.helper
+
+import code.api.Constant
+import code.api.cache.Redis
+import code.api.util.http4s.RequestScopeConnection
+import code.setup.RedisTestTarget
+import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers}
+
+import java.util.concurrent.ConcurrentLinkedQueue
+
+/**
+ * A changed dynamic entity definition must reach every instance's definition caches.
+ *
+ * Two caches hold definitions: the definition lists NewStyle.getDynamicEntities memoizes in Redis
+ * (dynamicEntity.cache.ttl.seconds, default 30) and the definitions map each instance builds from them
+ * (dynamicEntity.definitions_map.cache.ttl.seconds, default 300). Both follow the dynamic_entity cache
+ * namespace version, which a definition change bumps once its transaction has committed. When a change
+ * only dropped this instance's map, the rebuild read the list memoized before the change and kept it for
+ * the map's whole time-to-live, and other instances never heard of the change at all: a new entity's
+ * record endpoints answered 404 for up to five minutes.
+ */
+class DefinitionsCacheInvalidationTest extends FlatSpec with Matchers with BeforeAndAfterEach {
+
+  private val namespace = "dynamic_entity"
+
+  private def version(): Long = Constant.getCacheNamespaceVersion(namespace)
+
+  private def requireRedis(): Unit =
+    RedisTestTarget.requireReachable(Redis.isRedisReady, "the dynamic entity definition cache checks")
+
+  private def runPending(pending: ConcurrentLinkedQueue[() => Unit]): Unit =
+    Iterator.continually(pending.poll()).takeWhile(_ != null).foreach(_.apply())
+
+  override def afterEach(): Unit = {
+    RequestScopeConnection.currentAfterCommit.remove()
+    super.afterEach()
+  }
+
+  "The dynamic_entity namespace" should "be listed with the other cache namespaces" in {
+    // Listed, it shows on the cache info page and an administrator can bump it by hand.
+    Constant.ALL_CACHE_NAMESPACES should contain (namespace)
+  }
+
+  "forgetDefinitions" should "bump the dynamic_entity version once the transaction commits" in {
+    // Until the commit the change is not visible to other connections: an instance that saw the bump
+    // earlier could read the old rows and cache them under the new version.
+    requireRedis()
+    val pending = new ConcurrentLinkedQueue[() => Unit]()
+    RequestScopeConnection.currentAfterCommit.set(pending)
+    val before = version()
+
+    DynamicEntityHelper.forgetDefinitions()
+
+    withClue("the version was bumped before the commit ") {
+      version() shouldBe before
+    }
+    runPending(pending)
+    withClue("the commit did not bump the dynamic_entity version, so no instance learns of the change ") {
+      version() shouldBe before + 1
+    }
+  }
+
+  it should "bump the dynamic resource-doc caches only once the transaction commits" in {
+    // Dynamic entities generate resource docs. Bumped before the commit, another instance could rebuild
+    // /resource-docs from the old definitions and keep that for the resource-doc time-to-live (an hour).
+    requireRedis()
+    val pending = new ConcurrentLinkedQueue[() => Unit]()
+    RequestScopeConnection.currentAfterCommit.set(pending)
+    val dynamicBefore = Constant.getCacheNamespaceVersion(Constant.RD_DYNAMIC_NAMESPACE)
+    val allBefore = Constant.getCacheNamespaceVersion(Constant.RD_ALL_NAMESPACE)
+
+    DynamicEntityHelper.forgetDefinitions()
+
+    withClue("the resource-doc caches were bumped before the commit ") {
+      Constant.getCacheNamespaceVersion(Constant.RD_DYNAMIC_NAMESPACE) shouldBe dynamicBefore
+      Constant.getCacheNamespaceVersion(Constant.RD_ALL_NAMESPACE) shouldBe allBefore
+    }
+    runPending(pending)
+    withClue("the commit did not bump the resource-doc caches, so /resource-docs keeps the old entities ") {
+      Constant.getCacheNamespaceVersion(Constant.RD_DYNAMIC_NAMESPACE) shouldBe dynamicBefore + 1
+      Constant.getCacheNamespaceVersion(Constant.RD_ALL_NAMESPACE) shouldBe allBefore + 1
+    }
+  }
+
+  it should "bump it at once outside a transaction" in {
+    requireRedis()
+    val before = version()
+    DynamicEntityHelper.forgetDefinitions()
+    version() shouldBe before + 1
+  }
+
+  it should "leave it alone when the transaction rolls back" in {
+    requireRedis()
+    val pending = new ConcurrentLinkedQueue[() => Unit]()
+    RequestScopeConnection.currentAfterCommit.set(pending)
+    val before = version()
+
+    DynamicEntityHelper.forgetDefinitions()
+    pending.clear() // what a rollback does with the after-commit actions
+
+    version() shouldBe before
+  }
+
+  "The definitions map cache" should "keep a built map while the version is unchanged" in {
+    val probe = new Probe
+    probe.cache.get shouldBe 1
+    probe.cache.get shouldBe 1
+    probe.builds shouldBe 1
+  }
+
+  it should "build it again once the version changes, as when another instance bumps it" in {
+    val probe = new Probe
+    probe.cache.get
+    probe.version += 1
+    withClue("a map built under the previous version was served after the bump ") {
+      probe.cache.get shouldBe 2
+    }
+  }
+
+  it should "not keep a map built while the version changed under it" in {
+    // The build read the definitions before the change committed; it is kept under the version read
+    // before the build, which is already behind, so the next read builds again.
+    val probe = new Probe
+    probe.duringBuild = () => probe.version += 1
+    probe.cache.get shouldBe 1
+    probe.duringBuild = () => ()
+    probe.cache.get shouldBe 2
+  }
+
+  it should "drop the map on forget even though the version did not change (Redis unavailable)" in {
+    val probe = new Probe
+    probe.cache.get
+    probe.cache.forget()
+    probe.cache.get shouldBe 2
+  }
+
+  it should "not keep a map whose build started before forget" in {
+    val probe = new Probe
+    probe.duringBuild = () => probe.cache.forget()
+    probe.cache.get shouldBe 1
+    probe.duringBuild = () => ()
+    probe.cache.get shouldBe 2
+  }
+
+  it should "keep nothing, and not read the version, when its time-to-live is zero" in {
+    val probe = new Probe(ttlMillis = 0L)
+    probe.cache.get shouldBe 1
+    probe.cache.get shouldBe 2
+    probe.versionReads shouldBe 0
+  }
+
+  it should "build it again once its time-to-live has passed" in {
+    val probe = new Probe(ttlMillis = 50L)
+    probe.cache.get
+    Thread.sleep(80)
+    probe.cache.get shouldBe 2
+  }
+
+  /** A map cache whose version and builds the test controls; each build returns its own number. */
+  private class Probe(ttlMillis: Long = 60000L) {
+    @volatile var version: Long = 1L
+    @volatile var builds: Int = 0
+    @volatile var versionReads: Int = 0
+    @volatile var duringBuild: () => Unit = () => ()
+    val cache = new DefinitionsMapCache[Int](
+      ttlMillis,
+      () => { versionReads += 1; version },
+      () => { builds += 1; duringBuild(); builds })
+  }
+}
diff --git a/obp-api/src/test/scala/code/api/util/CommonsEmailWrapperTest.scala b/obp-api/src/test/scala/code/api/util/CommonsEmailWrapperTest.scala
new file mode 100644
index 0000000000..ded66e1bad
--- /dev/null
+++ b/obp-api/src/test/scala/code/api/util/CommonsEmailWrapperTest.scala
@@ -0,0 +1,36 @@
+package code.api.util
+
+import org.scalatest.{FeatureSpec, GivenWhenThen, Matchers}
+
+/**
+ * Mail is sent inside requests (the consent challenge, for one), so an SMTP server that accepts the connection and
+ * then goes quiet must not hold the request, and its database connection, forever. Jakarta Mail waits without limit
+ * unless the session says otherwise.
+ */
+class CommonsEmailWrapperTest extends FeatureSpec with Matchers with GivenWhenThen {
+
+  private val config = CommonsEmailWrapper.EmailConfig(smtpHost = "localhost", smtpPort = 1025, username = "", password = "")
+
+  feature("SMTP sessions have timeouts") {
+    scenario("connect, read and write are bounded by default") {
+      val props = CommonsEmailWrapper.sessionProperties(config)
+      for (key <- List("mail.smtp.connectiontimeout", "mail.smtp.timeout", "mail.smtp.writetimeout")) {
+        val value = Option(props.getProperty(key))
+        withClue(key) { value.map(_.toInt).getOrElse(0) should be > 0 }
+      }
+    }
+
+    scenario("mail.smtp.timeout_millis sets the timeout, and a value Jakarta Mail would read as 'wait forever' does not") {
+      CommonsEmailWrapper.timeoutMillisFrom(Some("2500")) shouldBe 2500
+      for (unusable <- List(None, Some("0"), Some("-1"), Some("abc"), Some("")))
+        withClue(unusable) { CommonsEmailWrapper.timeoutMillisFrom(unusable) shouldBe CommonsEmailWrapper.defaultTimeoutMillis }
+    }
+
+    scenario("the timeouts follow the configuration") {
+      val props = CommonsEmailWrapper.sessionProperties(config.copy(timeoutMillis = 2500))
+      props.getProperty("mail.smtp.connectiontimeout") shouldBe "2500"
+      props.getProperty("mail.smtp.timeout") shouldBe "2500"
+      props.getProperty("mail.smtp.writetimeout") shouldBe "2500"
+    }
+  }
+}
diff --git a/obp-api/src/test/scala/code/api/v4_0_0/DeleteAccountCascadeTest.scala b/obp-api/src/test/scala/code/api/v4_0_0/DeleteAccountCascadeTest.scala
index 2bd939c43a..f9b5b2f8bf 100644
--- a/obp-api/src/test/scala/code/api/v4_0_0/DeleteAccountCascadeTest.scala
+++ b/obp-api/src/test/scala/code/api/v4_0_0/DeleteAccountCascadeTest.scala
@@ -32,15 +32,19 @@ import code.api.Constant
 import code.api.ResourceDocs1_4_0.SwaggerDefinitionsJSON
 import code.api.ResourceDocs1_4_0.SwaggerDefinitionsJSON.createViewJsonV300
 import code.api.util.APIUtil.OAuth._
-import code.api.util.ApiRole
+import code.api.util.{APIUtil, ApiRole}
 import code.api.util.ApiRole.CanDeleteAccountCascade
 import code.api.util.ErrorMessages.{UserHasMissingRoles, AuthenticatedUserIsRequired}
 import code.api.v3_1_0.CreateAccountResponseJsonV310
 import code.api.v4_0_0.Http4s400.Implementations4_0_0
-import code.entitlement.Entitlement
+import code.cards.MappedPhysicalCard
+import code.entitlement.{Entitlement, MappedEntitlement}
+import code.model.dataAccess.MappedBankAccount
 import com.github.dwickern.macros.NameOf.nameOf
-import com.openbankproject.commons.model.{AmountOfMoneyJsonV121, ErrorMessage}
+import com.openbankproject.commons.model.{AccountId, AmountOfMoneyJsonV121, BankId, ErrorMessage}
 import com.openbankproject.commons.util.ApiVersion
+import net.liftweb.mapper.By
+import net.liftweb.util.Helpers.randomString
 import org.json4s.native.Serialization.write
 import org.scalatest.Tag
 
@@ -138,7 +142,105 @@ class DeleteAccountCascadeTest extends V400ServerSetup {
       When("We try to delete one more time")
       makeDeleteRequest(request400).code should equal(404)
     }
+    scenario("We delete an account that has paid by transaction requests", ApiEndpoint1, VersionOfApi) {
+      val paymentsBankId = createBank(APIUtil.generateUUID()).bankId.value
+      val (fromAccountId, toAccountId, transactionRequestIds) = createPaymentsForDeleteCascade(paymentsBankId)
+      doubleEntryBookTransactionsOfAccount(paymentsBankId, fromAccountId) should not be empty
+      val (transactionRequestsBefore, challengesBefore) = transactionRequestsAndChallenges(transactionRequestIds)
+      transactionRequestsBefore.size should equal(2)
+      challengesBefore should not be empty
+
+      When("We grant the role")
+      Entitlement.entitlement.vend.addEntitlement(paymentsBankId, resourceUser1.userId, ApiRole.canDeleteAccountCascade.toString)
+      And("We make a delete cascade request v4.0.0")
+      val deleteRequest400 = (v4_0_0_Request / "management" / "cascading" / "banks" / paymentsBankId /
+        "accounts" / fromAccountId).DELETE <@(user1)
+      makeDeleteRequest(deleteRequest400).code should equal(200)
+
+      Then("The double-entry book rows, transaction requests and challenges of the account are deleted")
+      doubleEntryBookTransactionsOfAccount(paymentsBankId, fromAccountId) shouldBe empty
+      val (transactionRequestsAfter, challengesAfter) = transactionRequestsAndChallenges(transactionRequestIds)
+      transactionRequestsAfter shouldBe empty
+      challengesAfter shouldBe empty
+      And("The double-entry book row is gone for the receiving account as well, since it was one row for both sides")
+      doubleEntryBookTransactionsOfAccount(paymentsBankId, toAccountId) shouldBe empty
+    }
+  }
+
+  feature(s"test $ApiEndpoint1 version $VersionOfApi - cards") {
+    scenario("Deleting an account keeps the cards of an account with the same ACCOUNT_ID at another bank", ApiEndpoint1, VersionOfApi) {
+      Given("Two banks that each have an account with the same ACCOUNT_ID, each with a card")
+      val bankId = BankId(randomString(10))
+      val otherBankId = BankId(randomString(10))
+      createBank(bankId.value)
+      createBank(otherBankId.value)
+      val accountId = AccountId(randomString(10))
+      createAccountRelevantResource(Some(resourceUser1), bankId, accountId, "EUR")
+      createAccountRelevantResource(Some(resourceUser1), otherBankId, accountId, "EUR")
+
+      def addCard(bank: BankId): String = {
+        val account = MappedBankAccount.find(
+          By(MappedBankAccount.bank, bank.value),
+          By(MappedBankAccount.theAccountId, accountId.value)
+        ).openOrThrowException("test account not found")
+        MappedPhysicalCard.create
+          .mBankId(bank.value)
+          .mAccount(account.id.get)
+          .mBankCardNumber(randomString(16))
+          .mSerialNumber(randomString(10))
+          .saveMe()
+          .mCardId.get
+      }
+      def cardExists(cardId: String): Boolean = MappedPhysicalCard.find(By(MappedPhysicalCard.mCardId, cardId)).isDefined
+      val card = addCard(bankId)
+      val otherBankCard = addCard(otherBankId)
+
+      When("We delete the account at the first bank with the cascading endpoint")
+      Entitlement.entitlement.vend.addEntitlement(bankId.value, resourceUser1.userId, ApiRole.canDeleteAccountCascade.toString)
+      val request = (v4_0_0_Request / "management" / "cascading" / "banks" / bankId.value /
+        "accounts" / accountId.value).DELETE <@ (user1)
+      val response = makeDeleteRequest(request)
+
+      Then("We should get a 200")
+      response.code should equal(200)
+
+      And("The deleted account's card is gone and the other bank's card is kept")
+      cardExists(card) should equal(false)
+      cardExists(otherBankCard) should equal(true)
+    }
+  }
+
+  feature(s"test $ApiEndpoint1 version $VersionOfApi - entitlements") {
+    scenario("Deleting an account keeps the entitlements of users who only use an account with the same ACCOUNT_ID at another bank", ApiEndpoint1, VersionOfApi) {
+      Given("Two banks that each have an account with the same ACCOUNT_ID, owned by different users")
+      val bankId = BankId(randomString(10))
+      val otherBankId = BankId(randomString(10))
+      createBank(bankId.value)
+      createBank(otherBankId.value)
+      val accountId = AccountId(randomString(10))
+      createAccountRelevantResource(Some(resourceUser1), bankId, accountId, "EUR")
+      createAccountRelevantResource(Some(resourceUser2), otherBankId, accountId, "EUR")
+
+      And("The other bank's account owner has an entitlement at the first bank")
+      Entitlement.entitlement.vend.addEntitlement(bankId.value, resourceUser2.userId, ApiRole.canCreateAccount.toString)
+      def otherOwnerEntitlements: Long = MappedEntitlement.count(
+        By(MappedEntitlement.mBankId, bankId.value),
+        By(MappedEntitlement.mUserId, resourceUser2.userId)
+      )
+      otherOwnerEntitlements should equal(1L)
+
+      When("We delete the account at the first bank with the cascading endpoint")
+      Entitlement.entitlement.vend.addEntitlement(bankId.value, resourceUser1.userId, ApiRole.canDeleteAccountCascade.toString)
+      val request = (v4_0_0_Request / "management" / "cascading" / "banks" / bankId.value /
+        "accounts" / accountId.value).DELETE <@ (user1)
+      val response = makeDeleteRequest(request)
+
+      Then("We should get a 200")
+      response.code should equal(200)
+
+      And("The other bank's account owner keeps the entitlement")
+      otherOwnerEntitlements should equal(1L)
+    }
   }
-  
 
 }
diff --git a/obp-api/src/test/scala/code/api/v4_0_0/DeleteBankCascadeTest.scala b/obp-api/src/test/scala/code/api/v4_0_0/DeleteBankCascadeTest.scala
index 0e4807db50..65b990002d 100644
--- a/obp-api/src/test/scala/code/api/v4_0_0/DeleteBankCascadeTest.scala
+++ b/obp-api/src/test/scala/code/api/v4_0_0/DeleteBankCascadeTest.scala
@@ -158,6 +158,25 @@ class DeleteBankCascadeTest extends V400ServerSetup {
       val responseBody = response.body.extract[CustomersMinimalJsonV400]
       responseBody.customers.length equals 0
     }
+    scenario("We delete a bank whose accounts have paid by transaction requests", ApiEndpoint1, VersionOfApi) {
+      val bankId = createBank(APIUtil.generateUUID()).bankId.value
+      val (fromAccountId, toAccountId, transactionRequestIds) = createPaymentsForDeleteCascade(bankId)
+      doubleEntryBookTransactionsOfAccount(bankId, fromAccountId) should not be empty
+      transactionRequestsAndChallenges(transactionRequestIds)._2 should not be empty
+
+      When("We grant the role")
+      Entitlement.entitlement.vend.addEntitlement(bankId, resourceUser1.userId, ApiRole.canDeleteBankCascade.toString)
+      And("We make a delete cascade request v4.0.0")
+      val deleteRequest400 = (v4_0_0_Request / "management" / "cascading" / "banks" / bankId ).DELETE <@(user1)
+      makeDeleteRequest(deleteRequest400).code should equal(200)
+
+      Then("The double-entry book rows, transaction requests and challenges of its accounts are deleted")
+      doubleEntryBookTransactionsOfAccount(bankId, fromAccountId) shouldBe empty
+      doubleEntryBookTransactionsOfAccount(bankId, toAccountId) shouldBe empty
+      val (transactionRequestsAfter, challengesAfter) = transactionRequestsAndChallenges(transactionRequestIds)
+      transactionRequestsAfter shouldBe empty
+      challengesAfter shouldBe empty
+    }
   }
   
 
diff --git a/obp-api/src/test/scala/code/api/v4_0_0/DeleteCascadeCounterpartiesTest.scala b/obp-api/src/test/scala/code/api/v4_0_0/DeleteCascadeCounterpartiesTest.scala
new file mode 100644
index 0000000000..f2cee8514f
--- /dev/null
+++ b/obp-api/src/test/scala/code/api/v4_0_0/DeleteCascadeCounterpartiesTest.scala
@@ -0,0 +1,201 @@
+/**
+Open Bank Project - API
+Copyright (C) 2011-2026, TESOBE GmbH.
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+Email: contact@tesobe.com
+TESOBE GmbH.
+Osloer Strasse 16/17
+Berlin 13359, Germany
+
+This product includes software developed at
+TESOBE (http://www.tesobe.com/)
+
+  */
+
+package code.api.v4_0_0
+
+import code.api.Constant
+import code.api.util.APIUtil.OAuth._
+import code.api.util.ApiRole
+import code.api.util.ErrorMessages.attemptedToOpenAnEmptyBox
+import code.api.v4_0_0.Http4s400.Implementations4_0_0
+import code.counterpartyattribute.{CounterpartyAttribute, CounterpartyAttributeX}
+import code.counterpartylimit.{CounterpartyLimit, CounterpartyLimitProvider}
+import code.entitlement.Entitlement
+import code.metadata.counterparties.{Counterparties, MappedCounterparty, MappedCounterpartyBespoke, MappedCounterpartyMetadata, MappedCounterpartyWhereTag}
+import com.github.dwickern.macros.NameOf.nameOf
+import com.openbankproject.commons.model.enums.CounterpartyAttributeType
+import com.openbankproject.commons.model.{AccountId, BankId, CounterpartyBespoke, CounterpartyId, CounterpartyTrait}
+import com.openbankproject.commons.util.ApiVersion
+import net.liftweb.mapper.{By, ByList}
+import net.liftweb.util.Helpers.randomString
+import org.scalatest.Tag
+
+import java.util.Date
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+class DeleteCascadeCounterpartiesTest extends V400ServerSetup {
+
+  object VersionOfApi extends Tag(ApiVersion.v4_0_0.toString)
+  object ApiEndpoint1 extends Tag(nameOf(Implementations4_0_0.deleteAccountCascade))
+  object ApiEndpoint2 extends Tag(nameOf(Implementations4_0_0.deleteBankCascade))
+
+  private val ownerView = Constant.SYSTEM_OWNER_VIEW_ID
+
+  private case class Seeded(bankId: BankId, accountId: AccountId, counterparty: CounterpartyTrait, locationIds: List[Long])
+
+  /** A counterparty on the given account, with rows in every table that hangs off it. */
+  private def createCounterpartyWithChildren(bankId: BankId, accountId: AccountId): Seeded = {
+    val counterparty = Counterparties.counterparties.vend.createCounterparty(
+      createdByUserId = resourceUser1.userId,
+      thisBankId = bankId.value,
+      thisAccountId = accountId.value,
+      thisViewId = ownerView,
+      name = randomString(20),
+      otherAccountRoutingScheme = "IBAN",
+      otherAccountRoutingAddress = "DE89370400440532013000",
+      otherBankRoutingScheme = "BIC",
+      otherBankRoutingAddress = "COBADEFFXXX",
+      otherBranchRoutingScheme = "OBP",
+      otherBranchRoutingAddress = "Berlin",
+      isBeneficiary = true,
+      otherAccountSecondaryRoutingScheme = "IBAN",
+      otherAccountSecondaryRoutingAddress = "DE89370400440532013000",
+      description = "cascade test payee",
+      currency = "EUR",
+      bespoke = List(CounterpartyBespoke("key", "value"))
+    ).openOrThrowException(attemptedToOpenAnEmptyBox)
+
+    val metadata = Counterparties.counterparties.vend
+      .getOrCreateMetadata(bankId, accountId, counterparty.counterpartyId, counterparty.name)
+      .openOrThrowException(attemptedToOpenAnEmptyBox)
+    metadata.addPhysicalLocation(resourceUser1.userPrimaryKey, new Date(), 13.4, 52.5) should be(true)
+    metadata.addCorporateLocation(resourceUser1.userPrimaryKey, new Date(), 11.6, 48.1) should be(true)
+    Await.result(
+      CounterpartyAttributeX.counterpartyAttributeProvider.vend.createOrUpdateCounterpartyAttribute(
+        CounterpartyId(counterparty.counterpartyId), None, "TAX_NUMBER", CounterpartyAttributeType.STRING, "123", Some(true)),
+      10.seconds
+    ).openOrThrowException(attemptedToOpenAnEmptyBox)
+    Await.result(
+      CounterpartyLimitProvider.counterpartyLimit.vend.createOrUpdateCounterpartyLimit(
+        bankId.value, accountId.value, ownerView, counterparty.counterpartyId,
+        "EUR", BigDecimal(100), BigDecimal(1000), 10, BigDecimal(10000), 100, BigDecimal(100000), 1000),
+      10.seconds
+    ).openOrThrowException(attemptedToOpenAnEmptyBox)
+
+    val storedMetadata = MappedCounterpartyMetadata.find(By(MappedCounterpartyMetadata.counterpartyId, counterparty.counterpartyId))
+      .openOrThrowException(attemptedToOpenAnEmptyBox)
+    Seeded(bankId, accountId, counterparty, List(storedMetadata.physicalLocation.get, storedMetadata.corporateLocation.get))
+  }
+
+  /** Rows left in each counterparty table for one seeded counterparty. */
+  private def remainingRows(seeded: Seeded): Map[String, Long] = {
+    val Seeded(bankId, accountId, counterparty, locationIds) = seeded
+    val counterpartyId = counterparty.counterpartyId
+    val primaryKey = counterparty.asInstanceOf[MappedCounterparty].id.get
+    Map(
+      "counterparty" -> MappedCounterparty.count(By(MappedCounterparty.mCounterPartyId, counterpartyId)),
+      "bespoke" -> MappedCounterpartyBespoke.count(By(MappedCounterpartyBespoke.mCounterparty, primaryKey)),
+      "metadata" -> MappedCounterpartyMetadata.count(
+        By(MappedCounterpartyMetadata.thisBankId, bankId.value),
+        By(MappedCounterpartyMetadata.thisAccountId, accountId.value),
+        By(MappedCounterpartyMetadata.counterpartyId, counterpartyId)),
+      "attribute" -> CounterpartyAttribute.count(By(CounterpartyAttribute.CounterpartyId_, counterpartyId)),
+      "limit" -> CounterpartyLimit.count(By(CounterpartyLimit.CounterpartyId, counterpartyId)),
+      "location" -> MappedCounterpartyWhereTag.count(ByList(MappedCounterpartyWhereTag.id, locationIds))
+    )
+  }
+
+  private val nothingLeft =
+    Map("counterparty" -> 0L, "bespoke" -> 0L, "metadata" -> 0L, "attribute" -> 0L, "limit" -> 0L, "location" -> 0L)
+  private val untouched =
+    Map("counterparty" -> 1L, "bespoke" -> 1L, "metadata" -> 1L, "attribute" -> 1L, "limit" -> 1L, "location" -> 2L)
+
+  feature(s"test $ApiEndpoint1 version $VersionOfApi - counterparties") {
+    scenario("Deleting an account deletes its own counterparties and leaves other accounts' counterparties alone", ApiEndpoint1, VersionOfApi) {
+      Given("Two banks that each have an account with the same ACCOUNT_ID, plus a second account in the first bank")
+      val bankId = BankId(randomString(10))
+      val otherBankId = BankId(randomString(10))
+      createBank(bankId.value)
+      createBank(otherBankId.value)
+      val accountId = AccountId(randomString(10))
+      val siblingAccountId = AccountId(randomString(10))
+      createAccountRelevantResource(Some(resourceUser1), bankId, accountId, "EUR")
+      createAccountRelevantResource(Some(resourceUser1), otherBankId, accountId, "EUR")
+      createAccountRelevantResource(Some(resourceUser1), bankId, siblingAccountId, "EUR")
+
+      And("A counterparty with bespoke fields, metadata with two locations, an attribute and a limit on each account")
+      val target = createCounterpartyWithChildren(bankId, accountId)
+      val sameIdOtherBank = createCounterpartyWithChildren(otherBankId, accountId)
+      val sibling = createCounterpartyWithChildren(bankId, siblingAccountId)
+      remainingRows(target) should equal(untouched)
+
+      When("We delete the first account with the cascading endpoint")
+      Entitlement.entitlement.vend.addEntitlement(bankId.value, resourceUser1.userId, ApiRole.canDeleteAccountCascade.toString)
+      val request = (v4_0_0_Request / "management" / "cascading" / "banks" / bankId.value /
+        "accounts" / accountId.value).DELETE <@ (user1)
+      val response = makeDeleteRequest(request)
+
+      Then("We should get a 200")
+      response.code should equal(200)
+
+      And("The account's counterparty and every row that hangs off it are gone")
+      remainingRows(target) should equal(nothingLeft)
+
+      And("The counterparty of the same ACCOUNT_ID at another bank is untouched")
+      remainingRows(sameIdOtherBank) should equal(untouched)
+
+      And("The counterparty of the other account at the same bank is untouched")
+      remainingRows(sibling) should equal(untouched)
+    }
+  }
+
+  feature(s"test $ApiEndpoint2 version $VersionOfApi - counterparties") {
+    scenario("Deleting a bank deletes the counterparties of all its accounts and leaves other banks alone", ApiEndpoint2, VersionOfApi) {
+      Given("A bank with two accounts, and another bank with an account that has the same ACCOUNT_ID")
+      val bankId = BankId(randomString(10))
+      val otherBankId = BankId(randomString(10))
+      createBank(bankId.value)
+      createBank(otherBankId.value)
+      val accountId = AccountId(randomString(10))
+      val secondAccountId = AccountId(randomString(10))
+      createAccountRelevantResource(Some(resourceUser1), bankId, accountId, "EUR")
+      createAccountRelevantResource(Some(resourceUser1), bankId, secondAccountId, "EUR")
+      createAccountRelevantResource(Some(resourceUser1), otherBankId, accountId, "EUR")
+
+      val first = createCounterpartyWithChildren(bankId, accountId)
+      val second = createCounterpartyWithChildren(bankId, secondAccountId)
+      val otherBank = createCounterpartyWithChildren(otherBankId, accountId)
+
+      When("We delete the bank with the cascading endpoint")
+      Entitlement.entitlement.vend.addEntitlement(bankId.value, resourceUser1.userId, ApiRole.canDeleteBankCascade.toString)
+      val request = (v4_0_0_Request / "management" / "cascading" / "banks" / bankId.value).DELETE <@ (user1)
+      val response = makeDeleteRequest(request)
+
+      Then("We should get a 200")
+      response.code should equal(200)
+
+      And("The counterparties of both accounts and their child rows are gone")
+      remainingRows(first) should equal(nothingLeft)
+      remainingRows(second) should equal(nothingLeft)
+
+      And("The counterparty at the other bank is untouched")
+      remainingRows(otherBank) should equal(untouched)
+    }
+  }
+
+}
diff --git a/obp-api/src/test/scala/code/api/v4_0_0/V400ServerSetup.scala b/obp-api/src/test/scala/code/api/v4_0_0/V400ServerSetup.scala
index 4f97200a61..37b24ad00e 100644
--- a/obp-api/src/test/scala/code/api/v4_0_0/V400ServerSetup.scala
+++ b/obp-api/src/test/scala/code/api/v4_0_0/V400ServerSetup.scala
@@ -50,12 +50,16 @@ import code.metadata.transactionimages.MappedTransactionImage
 import code.metadata.wheretags.MappedWhereTag
 import code.api.util.ErrorMessages.{AuthenticatedUserIsRequired, UserHasMissingRoles}
 import code.setup.{APIResponse, DefaultUsers, ServerSetupWithTestData}
+import code.model.dataAccess.DoubleEntryBookTransaction
+import code.transactionChallenge.MappedExpectedChallengeAnswer
 import code.transactionattribute.MappedTransactionAttribute
+import code.transactionrequests.MappedTransactionRequest
+import com.openbankproject.commons.model.enums.TransactionRequestStatus
 import com.openbankproject.commons.model.{AccountId, AccountRoutingJsonV121, AmountOfMoneyJsonV121, BankId, CreateViewJson, ErrorMessage, UpdateViewJSON}
 import com.openbankproject.commons.util.ApiShortVersions
 import code.setup.OBPReq
 import org.json4s.native.Serialization.write
-import net.liftweb.mapper.By
+import net.liftweb.mapper.{By, ByList}
 import net.liftweb.util.Helpers.randomString
 
 import java.util.concurrent.TimeUnit
@@ -457,6 +461,61 @@ trait V400ServerSetup extends ServerSetupWithTestData with DefaultUsers {
     (bank.bankId.value, fromAccount.account_id, transactionId)
   }
 
+  /**
+   * Creates two accounts at the bank and two SANDBOX_TAN transaction requests from the first to the second:
+   * one below the challenge threshold, which is paid at once and booked in the double-entry book,
+   * and one above it, which stays INITIATED with an expected challenge answer and no transaction.
+   * Returns (fromAccountId, toAccountId, transactionRequestIds).
+   */
+  def createPaymentsForDeleteCascade(bankId: String): (String, String, List[String]) = {
+    def addAccountJson = SwaggerDefinitionsJSON.createAccountRequestJsonV310
+      .copy(user_id = resourceUser1.userId, balance = AmountOfMoneyJsonV121("EUR","0"),
+        account_routings = List(AccountRoutingJsonV121(Random.nextString(10), Random.nextString(10))))
+    val fromAccount = createAccountViaEndpoint(bankId, addAccountJson, user1)
+    val toAccount = createAccountViaEndpoint(bankId, addAccountJson, user1)
+    val customViewJson = createViewJsonV300.copy(name = "_cascade_delete", metadata_view = "_cascade_delete", is_public = false).toCreateViewJson
+    val customView = createViewViaEndpoint(bankId, fromAccount.account_id, customViewJson, user1)
+    grantUserAccessToViewViaEndpoint(
+      bankId,
+      fromAccount.account_id,
+      resourceUser1.userId,
+      user1,
+      PostViewJsonV400(view_id = customView.id, is_system = false)
+    )
+    def pay(amount: String) = createTransactionRequestViaEndpoint(
+      fromBankId = bankId,
+      fromAccountId = fromAccount.account_id,
+      fromCurrency = fromAccount.balance.currency,
+      fromViewId = customView.id,
+      amount = amount,
+      toBankId = bankId,
+      toAccountId = toAccount.account_id,
+      user1
+    )
+    val completed = pay("10")
+    completed.status should equal(TransactionRequestStatus.COMPLETED.toString)
+    val initiated = pay("5000")
+    initiated.status should equal(TransactionRequestStatus.INITIATED.toString)
+    (fromAccount.account_id, toAccount.account_id, List(completed.id, initiated.id))
+  }
+
+  def doubleEntryBookTransactionsOfAccount(bankId: String, accountId: String): List[DoubleEntryBookTransaction] = {
+    DoubleEntryBookTransaction.findAll(
+      By(DoubleEntryBookTransaction.DebitTransactionBankId, bankId),
+      By(DoubleEntryBookTransaction.DebitTransactionAccountId, accountId)
+    ) ++ DoubleEntryBookTransaction.findAll(
+      By(DoubleEntryBookTransaction.CreditTransactionBankId, bankId),
+      By(DoubleEntryBookTransaction.CreditTransactionAccountId, accountId)
+    )
+  }
+
+  def transactionRequestsAndChallenges(transactionRequestIds: List[String]): (List[MappedTransactionRequest], List[MappedExpectedChallengeAnswer]) = {
+    (
+      MappedTransactionRequest.findAll(ByList(MappedTransactionRequest.mTransactionRequestId, transactionRequestIds)),
+      MappedExpectedChallengeAnswer.findAll(ByList(MappedExpectedChallengeAnswer.TransactionRequestId, transactionRequestIds))
+    )
+  }
+
   def saveHistoricalTransactionViaEndpoint(fromBankId: BankId,
                                           fromAccountId: AccountId,
                                           toBankId: BankId,
diff --git a/obp-api/src/test/scala/code/api/v5_0_0/ConsentRequestTest.scala b/obp-api/src/test/scala/code/api/v5_0_0/ConsentRequestTest.scala
index 1cd30efbbd..2bd931566d 100644
--- a/obp-api/src/test/scala/code/api/v5_0_0/ConsentRequestTest.scala
+++ b/obp-api/src/test/scala/code/api/v5_0_0/ConsentRequestTest.scala
@@ -30,7 +30,7 @@ import code.api.{Constant, RequestHeader}
 import code.api.ResourceDocs1_4_0.SwaggerDefinitionsJSON
 import code.api.util.APIUtil.OAuth._
 import code.api.util.ApiRole._
-import code.api.util.{APIUtil, Consent}
+import code.api.util.{APIUtil, CommonsEmailWrapper, Consent}
 import code.api.util.ErrorMessages._
 import code.api.v3_1_0.{PostConsentChallengeJsonV310, PostConsentEntitlementJsonV310}
 import code.api.v4_0_0.Http4s400.Implementations4_0_0
@@ -85,6 +85,7 @@ class ConsentRequestTest extends V500ServerSetup with PropsReset{
   def getConsentRequestUrl(requestId:String) = (v5_0_0_Request / "consumer"/ "consent-requests"/requestId).GET<@(user1)
   def createConsentByConsentRequestIdEmail(requestId:String) = (v5_0_0_Request / "consumer"/ "consent-requests"/requestId/"EMAIL"/"consents").POST<@(user1)
   def createConsentByConsentRequestIdImplicit(requestId:String) = (v5_0_0_Request / "consumer"/ "consent-requests"/requestId/"IMPLICIT"/"consents").POST<@(user1)
+  def createConsentByConsentRequestIdSms(requestId:String) = (v5_0_0_Request / "consumer"/ "consent-requests"/requestId/"SMS"/"consents").POST<@(user1)
   def getConsentByRequestIdUrl(requestId:String) = (v5_0_0_Request / "consumer"/ "consent-requests"/requestId/"consents").GET<@(user1)
 
   feature("Create/Get Consent Request v5.0.0") {
@@ -283,5 +284,88 @@ class ConsentRequestTest extends V500ServerSetup with PropsReset{
     }
 
   }
-  
+
+  feature("The consent challenge is sent before Create Consent By CONSENT_REQUEST_ID answers v5.0.0") {
+    // The PSU needs the challenge to accept the consent. When it was sent from a Future the request did not wait
+    // for, it could run after the request's database connection was closed and never be sent at all.
+    def challengeMailTo(recipient: String) = CommonsEmailWrapper.testModeSentMail.filter(mail =>
+      mail.to == List(recipient) && mail.subject == "OBP Consent Challenge" &&
+        mail.textContent.exists(_.contains(Consent.challengeAnswerAtTestEnvironment)))
+
+    // Returns the consent request id and the response of creating its consent.
+    def createConsent(createConsentUrl: String => code.setup.OBPReq, consentRequest: PostConsentRequestJsonV500 = postConsentRequestJson,
+                      sendDelayMillis: Int = 1500) = {
+      // A send that takes as long as an SMTP round trip: one the request does not wait for is still pending when
+      // the response arrives.
+      setPropsValues("mail.test.mode" -> "true", "mail.test.mode.delay_millis" -> sendDelayMillis.toString)
+      CommonsEmailWrapper.clearTestModeSentMail()
+      Entitlement.entitlement.vend.addEntitlement("", resourceUser1.userId, CanGetAnyUser.toString)
+      val createConsentRequestResponse = makePostRequest(createConsentRequestUrl, write(consentRequest))
+      createConsentRequestResponse.code should equal(201)
+      val consentRequestId = createConsentRequestResponse.body.extract[ConsentRequestResponseJson].consent_request_id
+      (consentRequestId, makePostRequest(createConsentUrl(consentRequestId), write("")))
+    }
+
+    scenario("EMAIL: the challenge goes to the email address of the consent request", ApiEndpoint1, ApiEndpoint3, VersionOfApi) {
+      When("We create a consent with an EMAIL challenge")
+      val (_, response) = createConsent(createConsentByConsentRequestIdEmail)
+      response.code should equal(201)
+      Then("The challenge mail has already been sent, once, to the consent request's email address")
+      challengeMailTo(postConsentRequestJson.email.head).size should equal(1)
+    }
+
+    scenario("IMPLICIT: the challenge goes to the email address of the user", ApiEndpoint1, ApiEndpoint3, VersionOfApi) {
+      When("We create a consent with an IMPLICIT challenge")
+      val (_, response) = createConsent(createConsentByConsentRequestIdImplicit)
+      response.code should equal(201)
+      Then("The challenge mail has already been sent, once, to the user's email address")
+      challengeMailTo(resourceUser1.emailAddress).size should equal(1)
+    }
+
+    scenario("EMAIL without an email address: no consent that no one could answer", ApiEndpoint1, ApiEndpoint2, ApiEndpoint3, VersionOfApi) {
+      When("We create a consent with an EMAIL challenge from a consent request that names no email address")
+      val (consentRequestId, response) =
+        createConsent(createConsentByConsentRequestIdEmail, postConsentRequestJson.copy(email = None), sendDelayMillis = 0)
+      Then("We get a 400, as from the other Create Consent endpoints, and no challenge is sent")
+      response.code should equal(400)
+      response.body.extract[ErrorMessage].message should include ("email")
+      CommonsEmailWrapper.testModeSentMail.filter(_.subject == "OBP Consent Challenge") shouldBe empty
+      And("No consent was kept for the consent request")
+      // Get Consent By CONSENT_REQUEST_ID answers ConsentRequestNotFound when no consent was created from the request.
+      val getConsent = makeGetRequest(getConsentByRequestIdUrl(consentRequestId))
+      getConsent.code should equal(400)
+      getConsent.body.extract[ErrorMessage].message should equal(ConsentRequestNotFound)
+    }
+
+    scenario("A send that throws before it starts: the consent is rejected as well", ApiEndpoint1, ApiEndpoint2, ApiEndpoint3, VersionOfApi) {
+      When("We create a consent with an EMAIL challenge while the SMTP settings cannot be read")
+      // Outside mail test mode the mapped connector reads the SMTP settings before it builds its Future, so a bad
+      // port throws at once rather than failing a Future. No SMTP connection is attempted.
+      setPropsValues("mail.test.mode" -> "false", "mail.smtp.port" -> "not-a-port")
+      Entitlement.entitlement.vend.addEntitlement("", resourceUser1.userId, CanGetAnyUser.toString)
+      val createConsentRequestResponse = makePostRequest(createConsentRequestUrl, write(postConsentRequestJson))
+      createConsentRequestResponse.code should equal(201)
+      val consentRequestId = createConsentRequestResponse.body.extract[ConsentRequestResponseJson].consent_request_id
+      val response = makePostRequest(createConsentByConsentRequestIdEmail(consentRequestId), write(""))
+      Then("The request fails and the consent it wrote is rejected, not left INITIATED")
+      response.code should not equal(201)
+      val getConsent = makeGetRequest(getConsentByRequestIdUrl(consentRequestId))
+      getConsent.code should equal(200)
+      getConsent.body.extract[ConsentJsonV500].status should equal(ConsentStatus.REJECTED.toString)
+    }
+
+    scenario("SMS that cannot be sent: the consent is rejected rather than left waiting for an answer", ApiEndpoint1, ApiEndpoint2, ApiEndpoint3, VersionOfApi) {
+      // With an SMS provider configured this would send a real message; the scenario needs none.
+      assume(APIUtil.getPropsValue("sca_phone_api_key").isEmpty, "sca_phone_api_key is set, so an SMS would be sent")
+      When("We create a consent with an SMS challenge on an instance with no SMS provider configured")
+      val (consentRequestId, response) = createConsent(createConsentByConsentRequestIdSms, sendDelayMillis = 0)
+      Then("We get the 400 of the failed send")
+      response.code should equal(400)
+      And("The consent written before the send is rejected, so no one is left with a consent they cannot answer")
+      val getConsent = makeGetRequest(getConsentByRequestIdUrl(consentRequestId))
+      getConsent.code should equal(200)
+      getConsent.body.extract[ConsentJsonV500].status should equal(ConsentStatus.REJECTED.toString)
+    }
+  }
+
 }
diff --git a/obp-api/src/test/scala/code/api/v5_1_0/VRPConsentRequestTest.scala b/obp-api/src/test/scala/code/api/v5_1_0/VRPConsentRequestTest.scala
index a1e87501f9..bb4173f20f 100644
--- a/obp-api/src/test/scala/code/api/v5_1_0/VRPConsentRequestTest.scala
+++ b/obp-api/src/test/scala/code/api/v5_1_0/VRPConsentRequestTest.scala
@@ -34,7 +34,7 @@ import code.api.util.APIUtil.OAuth._
 import code.api.util.ApiRole._
 import code.api.Constant
 import com.openbankproject.commons.model.{AccountId, BankId, ViewId}
-import code.api.util.Consent
+import code.api.util.{APIUtil, Consent}
 import code.consent.Consents
 import code.counterpartylimit.CounterpartyLimitProvider
 import code.metadata.counterparties.Counterparties
@@ -90,6 +90,7 @@ class VRPConsentRequestTest extends V510ServerSetup with PropsReset{
   def getConsentRequestUrl(requestId:String) = (v5_1_0_Request / "consumer"/ "consent-requests"/requestId).GET<@(user1)
   def createConsentByConsentRequestIdEmail(requestId:String) = (v5_1_0_Request / "consumer"/ "consent-requests"/requestId/"EMAIL"/"consents").POST<@(user1)
   def createConsentByConsentRequestIdImplicit(requestId:String) = (v5_1_0_Request / "consumer"/ "consent-requests"/requestId/"IMPLICIT"/"consents").POST<@(user1)
+  def createConsentBySmsUrl(requestId:String) = (v5_1_0_Request / "consumer"/ "consent-requests"/requestId/"SMS"/"consents").POST<@(user1)
   def getConsentByRequestIdUrl(requestId:String) = (v5_1_0_Request / "consumer"/ "consent-requests"/requestId/"consents").GET<@(user1)
 
   val fromAccountJson = ConsentRequestFromAccountJson (
@@ -243,6 +244,65 @@ class VRPConsentRequestTest extends V510ServerSetup with PropsReset{
     }
   
 
+    scenario("A VRP consent whose challenge has nowhere to go leaves no mandate behind", ApiEndpoint1, ApiEndpoint3, VersionOfApi) {
+      When("the PSU converts a VRP consent request that names no email address with an EMAIL challenge")
+      val createConsentResponse = makePostRequest(createVRPConsentRequestUrl, write(postVRPConsentRequestMonthlyGuardJson.copy(email = None)))
+      createConsentResponse.code should equal(201)
+      val consentRequestId = createConsentResponse.body.extract[ConsentRequestResponseJson].consent_request_id
+      Entitlement.entitlement.vend.addEntitlement("", resourceUser1.userId, CanGetAnyUser.toString)
+      def vrpViews = ViewDefinition.findAll().count(_.viewId.value.startsWith(Constant.VRP_VIEW_ID_PREFIX))
+      def counterparties = code.metadata.counterparties.MappedCounterparty.count
+      val (viewsBefore, counterpartiesBefore) = (vrpViews, counterparties)
+      val response = makePostRequest(createConsentByConsentRequestIdEmail(consentRequestId), write(""))
+
+      Then("the request is refused before the view, its grant, the counterparty or its limit are written")
+      response.code should equal(400)
+      response.body.extract[ErrorMessage].message should include ("email")
+      vrpViews should equal(viewsBefore)
+      counterparties should equal(counterpartiesBefore)
+    }
+
+    scenario("A VRP consent with too long a time to live leaves no mandate behind", ApiEndpoint1, ApiEndpoint3, VersionOfApi) {
+      When("consents.max_time_to_live is lowered after a VRP consent request was lodged, and the PSU then converts it")
+      // Lodging the request checks the same limit, so only a limit lowered in between reaches this check.
+      val createConsentResponse = makePostRequest(createVRPConsentRequestUrl, write(postVRPConsentRequestMonthlyGuardJson.copy(time_to_live = Some(1000L))))
+      createConsentResponse.code should equal(201)
+      val consentRequestId = createConsentResponse.body.extract[ConsentRequestResponseJson].consent_request_id
+      setPropsValues("consents.max_time_to_live" -> "999")
+      Entitlement.entitlement.vend.addEntitlement("", resourceUser1.userId, CanGetAnyUser.toString)
+      def vrpViews = ViewDefinition.findAll().count(_.viewId.value.startsWith(Constant.VRP_VIEW_ID_PREFIX))
+      def counterparties = code.metadata.counterparties.MappedCounterparty.count
+      val (viewsBefore, counterpartiesBefore) = (vrpViews, counterparties)
+      val response = makePostRequest(createConsentByConsentRequestIdEmail(consentRequestId), write(""))
+
+      Then("the request is refused before the mandate is written")
+      response.code should equal(400)
+      response.body.extract[ErrorMessage].message should include (ConsentMaxTTL)
+      vrpViews should equal(viewsBefore)
+      counterparties should equal(counterpartiesBefore)
+    }
+
+    scenario("A VRP consent whose SMS challenge cannot be sent is rejected and its mandate released", ApiEndpoint1, ApiEndpoint3, VersionOfApi) {
+      // With an SMS provider configured this would send a real message; the scenario needs none.
+      assume(APIUtil.getPropsValue("sca_phone_api_key").isEmpty, "sca_phone_api_key is set, so an SMS would be sent")
+      When("the PSU converts a VRP consent request with an SMS challenge on an instance with no SMS provider")
+      val createConsentResponse = makePostRequest(createVRPConsentRequestUrl, write(postVRPConsentRequestMonthlyGuardJson))
+      createConsentResponse.code should equal(201)
+      val consentRequestId = createConsentResponse.body.extract[ConsentRequestResponseJson].consent_request_id
+      Entitlement.entitlement.vend.addEntitlement("", resourceUser1.userId, CanGetAnyUser.toString)
+      def vrpViews = ViewDefinition.findAll().count(_.viewId.value.startsWith(Constant.VRP_VIEW_ID_PREFIX))
+      def vrpGrants = AccountAccess.findAll().count(_.view_id.get.startsWith(Constant.VRP_VIEW_ID_PREFIX))
+      val (viewsBefore, grantsBefore) = (vrpViews, vrpGrants)
+      val response = makePostRequest(createConsentBySmsUrl(consentRequestId), write(""))
+
+      Then("the send fails, the consent is rejected, and the view and the PSU's grant on it are released")
+      response.code should equal(400)
+      val getConsent = makeGetRequest(getConsentByRequestIdUrl(consentRequestId))
+      getConsent.body.extract[ConsentJsonV500].status should equal(ConsentStatus.REJECTED.toString)
+      vrpGrants should equal(grantsBefore)
+      vrpViews should equal(viewsBefore)
+    }
+
     scenario("Revoking a VRP consent releases the mandate it created", ApiEndpoint1, ApiEndpoint3, VersionOfApi) {
       When("the PSU creates a VRP consent request and converts it")
       val createConsentResponse = makePostRequest(createVRPConsentRequestUrl, write(postVRPConsentRequestMonthlyGuardJson))
diff --git a/scripts/resource_doc_baseline/parity_allowlist.json b/scripts/resource_doc_baseline/parity_allowlist.json
index fec398b409..40827e7e7b 100644
--- a/scripts/resource_doc_baseline/parity_allowlist.json
+++ b/scripts/resource_doc_baseline/parity_allowlist.json
@@ -1047,9 +1047,9 @@
       "version": "v5_0_0",
       "endpoint": "createConsentByConsentRequestIdImplicit",
       "field": "errorResponseBodies",
-      "reason": "Same shared-val expansion artifact as createConsentByConsentRequestIdEmail; expanded sets are identical.",
+      "reason": "Same shared-val expansion artifact as createConsentByConsentRequestIdEmail; expanded sets are identical. The http4s list also names GetConsentImplicitSCAError, which IMPLICIT returns when the user has no address for the implicit SCA method.",
       "lift_digest": "1cb5a60be98c8388253c1855e0b351469e254c5dee7ab2f42d388fe6c213da14",
-      "http4s_digest": "7affd9a379cb014033cdab59e2f4c7df80fbcb89ee3b14b126ae3fc11f20747f"
+      "http4s_digest": "70b04a9494e0047a99f36e6a4b3479ffe1958ab185684307bd6d1e33fb3e0c33"
     },
     {
       "version": "v5_0_0",
@@ -1759,17 +1759,17 @@
       "version": "v5_0_0",
       "endpoint": "createConsentByConsentRequestIdImplicit",
       "field": "description",
-      "reason": "Documents the consumer pin as implemented: adds the caller as the final fallback and states that a Consent naming no consumer is refused with ConsentConsumerIsRequired rather than stored with a null consumer_id. Text-only.",
+      "reason": "Documents the consumer pin as implemented: adds the caller as the final fallback and states that a Consent naming no consumer is refused with ConsentConsumerIsRequired rather than stored with a null consumer_id. Text-only. Also states that IMPLICIT sends the challenge the user's implicit SCA method names (an EMAIL on the mapped connector) before the call answers, and that a user with no address for it gets GetConsentImplicitSCAError before any Consent is created.",
       "lift_digest": "b262a1d68717cdbcc51a7c035741e5da4cf5124123683e3dd015c85ed8baa231",
-      "http4s_digest": "95623d7eda5322dbacd11e947887da53859b646ea16edaa5dd0b34782021f46e"
+      "http4s_digest": "d5dc004e521ebc35830f13b55ca24ca1e01a2f761360b69d8bdbe021fae74c6b"
     },
     {
       "version": "v5_0_0",
       "endpoint": "createConsentByConsentRequestIdSms",
       "field": "description",
-      "reason": "Documents the consumer pin as implemented: adds the caller as the final fallback and states that a Consent naming no consumer is refused with ConsentConsumerIsRequired rather than stored with a null consumer_id. Text-only.",
+      "reason": "Documents the consumer pin as implemented: adds the caller as the final fallback and states that a Consent naming no consumer is refused with ConsentConsumerIsRequired rather than stored with a null consumer_id. Text-only. Also states that the challenge is sent before the call answers, and that a challenge that cannot be sent fails the call, rejects the Consent and uses up the Consent Request.",
       "lift_digest": "7397560b818a5d9f4eb31a701499aea10c0ecd8d89d15496c0904d90984b84a6",
-      "http4s_digest": "881b4fe9e74e92efbe1177f1548d7ad4aa85be90f0ec44789ecf779267335f6b"
+      "http4s_digest": "339420f215d5c413ee4037e4cf5b26a3d906b3e60bd5535a41c24f39778d9765"
     },
     {
       "version": "v4_0_0",