feat(tf): add composable app deployments

Cargo.lock

@@ -111,9 +111,9 @@ dependencies = [
 
 [[package]]
 name = "anyhow"
-version = "1.0.101"
+version = "1.0.103"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5f0e0fee31ef5ed1ba1316088939cea399010ed7731dba877ed44aeb407a75ea"
+checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3"
 
 [[package]]
 name = "arc-swap"
@@ -660,9 +660,9 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-epoch"
-version = "0.9.18"
+version = "0.9.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
+checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
 dependencies = [
  "crossbeam-utils",
 ]
@@ -3602,6 +3602,17 @@ dependencies = [
  "unicode-segmentation",
 ]
 
+[[package]]
+name = "testing-framework-app"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "testing-framework-core",
+ "testing-framework-runner-local",
+ "thiserror 2.0.18",
+ "tokio",
+]
+
 [[package]]
 name = "testing-framework-core"
 version = "0.1.0"

Cargo.toml

@@ -30,6 +30,7 @@ members = [
   "examples/redis_streams/examples",
   "examples/redis_streams/testing/integration",
   "examples/redis_streams/testing/workloads",
+  "testing-framework/app",
   "testing-framework/core",
   "testing-framework/deployers/compose",
   "testing-framework/deployers/k8s",
@@ -58,6 +59,7 @@ all = "allow"
 cfgsync-adapter                  = { default-features = false, path = "cfgsync/adapter" }
 cfgsync-artifacts                = { default-features = false, path = "cfgsync/artifacts" }
 cfgsync-core                     = { default-features = false, path = "cfgsync/core" }
+testing-framework-app            = { default-features = false, path = "testing-framework/app" }
 testing-framework-core           = { default-features = false, path = "testing-framework/core" }
 testing-framework-runner-compose = { default-features = false, path = "testing-framework/deployers/compose" }
 testing-framework-runner-k8s     = { default-features = false, path = "testing-framework/deployers/k8s" }

testing-framework/app/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+description.workspace = true
+edition.workspace     = true
+license.workspace     = true
+name                  = "testing-framework-app"
+readme.workspace      = true
+repository.workspace  = true
+version.workspace     = true
+
+[lints]
+workspace = true
+
+[dependencies]
+async-trait                    = { workspace = true }
+testing-framework-core         = { workspace = true }
+testing-framework-runner-local = { workspace = true }
+thiserror                      = { workspace = true }
+tokio                          = { features = ["rt", "sync"], workspace = true }
+
+[dev-dependencies]
+tokio = { features = ["macros", "rt"], workspace = true }

testing-framework/app/src/cleanup.rs

@@ -0,0 +1,81 @@
+use testing_framework_core::scenario::internal::CleanupGuard;
+
+/// LIFO cleanup owned by one application deployment preparation.
+///
+/// Managed adapters register guards here as soon as they acquire a resource.
+/// If deployment fails, dropping the stack releases everything already
+/// acquired. On success, the stack is transferred to the scenario runner.
+#[derive(Default)]
+pub(crate) struct AppCleanupStack {
+    guards: Vec<Box<dyn CleanupGuard>>,
+}
+
+impl AppCleanupStack {
+    pub(crate) fn push(&mut self, guard: Box<dyn CleanupGuard>) {
+        self.guards.push(guard);
+    }
+
+    pub(crate) fn into_guard(self) -> Option<Box<dyn CleanupGuard>> {
+        if self.guards.is_empty() {
+            None
+        } else {
+            Some(Box::new(self))
+        }
+    }
+
+    fn cleanup_all(&mut self) {
+        while let Some(guard) = self.guards.pop() {
+            guard.cleanup();
+        }
+    }
+}
+
+impl CleanupGuard for AppCleanupStack {
+    fn cleanup(mut self: Box<Self>) {
+        self.cleanup_all();
+    }
+}
+
+impl Drop for AppCleanupStack {
+    fn drop(&mut self) {
+        self.cleanup_all();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use testing_framework_core::scenario::internal::CleanupGuard;
+
+    use super::AppCleanupStack;
+
+    struct Probe {
+        label: &'static str,
+        order: Arc<Mutex<Vec<&'static str>>>,
+    }
+
+    impl CleanupGuard for Probe {
+        fn cleanup(self: Box<Self>) {
+            self.order.lock().unwrap().push(self.label);
+        }
+    }
+
+    #[test]
+    fn managed_resources_clean_up_in_reverse_acquisition_order() {
+        let order = Arc::new(Mutex::new(Vec::new()));
+        let mut cleanup = AppCleanupStack::default();
+        cleanup.push(Box::new(Probe {
+            label: "dependency",
+            order: Arc::clone(&order),
+        }));
+        cleanup.push(Box::new(Probe {
+            label: "dependent",
+            order: Arc::clone(&order),
+        }));
+
+        drop(cleanup);
+
+        assert_eq!(*order.lock().unwrap(), ["dependent", "dependency"]);
+    }
+}

testing-framework/app/src/context.rs

@@ -0,0 +1,323 @@
+use testing_framework_core::scenario::{
+    Application, DynError, NodeClients, internal::CleanupGuard,
+};
+use testing_framework_runner_local::LocalDeployerEnv;
+
+use crate::{
+    AppDeployError, AppDeployment, AppHandle, HandleRegistry, LocalAppCluster,
+    cleanup::AppCleanupStack,
+};
+
+/// Mutable deployment context used to compose applications and expose handles.
+///
+/// A context belongs to one scenario preparation. It provides access to the
+/// outer scenario deployment and node clients while collecting application
+/// handles for later use by workloads. Exposed handles are dropped in reverse
+/// exposure order when the scenario runtime is released.
+pub struct DeployContext<E: Application> {
+    deployment: E::Deployment,
+    node_clients: NodeClients<E>,
+    handles: HandleRegistry,
+    cleanup: AppCleanupStack,
+}
+
+impl<E> DeployContext<E>
+where
+    E: Application,
+{
+    /// Creates a context for an outer scenario deployment and its node clients.
+    pub fn new(deployment: E::Deployment, node_clients: NodeClients<E>) -> Self {
+        Self {
+            deployment,
+            node_clients,
+            handles: HandleRegistry::new(),
+            cleanup: AppCleanupStack::default(),
+        }
+    }
+
+    /// Deploys a child application without automatically exposing its handle.
+    ///
+    /// Use this when the returned handle is only needed while constructing a
+    /// higher-level application handle.
+    pub async fn deploy<A>(&mut self, app: A) -> Result<A::Handle, DynError>
+    where
+        A: AppDeployment<E>,
+    {
+        app.deploy(self).await
+    }
+
+    /// Deploys a child application and exposes a clone of its handle.
+    ///
+    /// The exposed clone makes the handle available to workloads through
+    /// [`crate::AppRunContextExt`]. Managed resource lifetime remains owned by
+    /// scenario cleanup.
+    pub async fn deploy_and_expose<A>(&mut self, app: A) -> Result<A::Handle, DynError>
+    where
+        A: AppDeployment<E>,
+    {
+        let handle = self.deploy(app).await?;
+
+        self.expose(handle.clone())?;
+
+        Ok(handle)
+    }
+
+    /// Exposes the default unnamed handle for its concrete type.
+    ///
+    /// Returns [`AppDeployError::DuplicateHandle`] when that type already has a
+    /// default handle.
+    pub fn expose<T>(&mut self, handle: T) -> Result<(), AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.handles.expose(handle)
+    }
+
+    /// Exposes a named handle, allowing multiple instances of the same type.
+    ///
+    /// A type and name pair must be unique within the deployment context.
+    pub fn expose_named<T>(
+        &mut self,
+        name: impl Into<String>,
+        handle: T,
+    ) -> Result<(), AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.handles.expose_named(name, handle)
+    }
+
+    /// Returns a clone of the default handle for `T`, if it is exposed.
+    pub fn get<T>(&self) -> Option<T>
+    where
+        T: AppHandle,
+    {
+        self.handles.get()
+    }
+
+    /// Returns a clone of the named handle for `T`, if it is exposed.
+    pub fn get_named<T>(&self, name: &str) -> Option<T>
+    where
+        T: AppHandle,
+    {
+        self.handles.get_named(name)
+    }
+
+    /// Returns the default handle for `T` or a typed missing-handle error.
+    pub fn require<T>(&self) -> Result<T, AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.handles.require()
+    }
+
+    /// Returns the named handle for `T` or a typed missing-handle error.
+    pub fn require_named<T>(&self, name: &str) -> Result<T, AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.handles.require_named(name)
+    }
+
+    /// Returns whether the default handle for `T` is exposed.
+    pub fn contains<T>(&self) -> bool
+    where
+        T: AppHandle,
+    {
+        self.handles.contains::<T>()
+    }
+
+    /// Borrows the handles exposed so far.
+    pub fn handles(&self) -> &HandleRegistry {
+        &self.handles
+    }
+
+    /// Starts every node of an additional uniform local cluster.
+    ///
+    /// The cluster is registered with scenario cleanup before its handle is
+    /// returned. Cleanup stops all nodes independently of handle clones.
+    pub async fn deploy_local_cluster<App>(
+        &mut self,
+        deployment: App::Deployment,
+    ) -> Result<LocalAppCluster<App>, DynError>
+    where
+        App: LocalDeployerEnv,
+    {
+        let cluster = LocalAppCluster::start(deployment).await?;
+        self.register_cleanup(cluster.cleanup_guard());
+        Ok(cluster)
+    }
+
+    /// Returns the outer scenario deployment descriptor.
+    pub fn deployment(&self) -> &E::Deployment {
+        &self.deployment
+    }
+
+    /// Returns the outer scenario clients resolved from all configured sources.
+    pub fn node_clients(&self) -> &NodeClients<E> {
+        &self.node_clients
+    }
+
+    pub(crate) fn register_cleanup(&mut self, guard: Box<dyn CleanupGuard>) {
+        self.cleanup.push(guard);
+    }
+
+    pub(crate) fn into_runtime_parts(self) -> (HandleRegistry, Option<Box<dyn CleanupGuard>>) {
+        let Self {
+            handles, cleanup, ..
+        } = self;
+        (handles, cleanup.into_guard())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use async_trait::async_trait;
+    use testing_framework_core::{
+        scenario::{Application, DynError, NodeClients},
+        topology::DeploymentDescriptor,
+    };
+
+    use super::DeployContext;
+    use crate::AppDeployment;
+
+    #[derive(Clone)]
+    struct TestDeployment;
+
+    impl DeploymentDescriptor for TestDeployment {
+        fn node_count(&self) -> usize {
+            0
+        }
+    }
+
+    struct TestEnv;
+
+    #[async_trait]
+    impl Application for TestEnv {
+        type Deployment = TestDeployment;
+        type NodeClient = ();
+        type NodeConfig = ();
+    }
+
+    #[derive(Clone, Debug, Eq, PartialEq)]
+    struct ChildHandle {
+        id: &'static str,
+    }
+
+    struct ChildApp;
+
+    #[async_trait]
+    impl AppDeployment<TestEnv> for ChildApp {
+        type Handle = ChildHandle;
+
+        async fn deploy(self, _ctx: &mut DeployContext<TestEnv>) -> Result<Self::Handle, DynError> {
+            Ok(ChildHandle { id: "child" })
+        }
+    }
+
+    #[derive(Clone, Debug, Eq, PartialEq)]
+    struct SiblingHandle {
+        id: &'static str,
+    }
+
+    struct SiblingApp;
+
+    #[async_trait]
+    impl AppDeployment<TestEnv> for SiblingApp {
+        type Handle = SiblingHandle;
+
+        async fn deploy(self, _ctx: &mut DeployContext<TestEnv>) -> Result<Self::Handle, DynError> {
+            Ok(SiblingHandle { id: "sibling" })
+        }
+    }
+
+    #[derive(Clone, Debug, Eq, PartialEq)]
+    struct ParentHandle {
+        child: ChildHandle,
+    }
+
+    struct ParentApp;
+
+    #[async_trait]
+    impl AppDeployment<TestEnv> for ParentApp {
+        type Handle = ParentHandle;
+
+        async fn deploy(self, ctx: &mut DeployContext<TestEnv>) -> Result<Self::Handle, DynError> {
+            let child = ctx.deploy(ChildApp).await?;
+            let parent = ParentHandle { child };
+
+            ctx.expose(parent.clone())?;
+
+            Ok(parent)
+        }
+    }
+
+    #[tokio::test]
+    async fn deploy_returns_handle_without_exposing_it() {
+        let mut ctx = test_context();
+        let child = ctx.deploy(ChildApp).await.expect("deploy child app");
+
+        assert_eq!(child, ChildHandle { id: "child" });
+        assert!(ctx.get::<ChildHandle>().is_none());
+    }
+
+    #[tokio::test]
+    async fn exposed_handle_can_be_required() {
+        let mut ctx = test_context();
+
+        ctx.expose(ChildHandle { id: "child" })
+            .expect("expose child");
+
+        let child = ctx.require::<ChildHandle>().expect("require exposed child");
+        assert_eq!(child, ChildHandle { id: "child" });
+    }
+
+    #[tokio::test]
+    async fn nested_deployment_exposes_parent_handle_only() {
+        let mut ctx = test_context();
+        let parent = ctx.deploy(ParentApp).await.expect("deploy parent app");
+
+        assert_eq!(parent.child, ChildHandle { id: "child" });
+        assert!(ctx.get::<ChildHandle>().is_none());
+        assert!(ctx.get::<ParentHandle>().is_some());
+    }
+
+    #[derive(Clone, Debug, Eq, PartialEq)]
+    struct StackHandle {
+        child: ChildHandle,
+        sibling: SiblingHandle,
+    }
+
+    struct StackApp;
+
+    #[async_trait]
+    impl AppDeployment<TestEnv> for StackApp {
+        type Handle = StackHandle;
+
+        async fn deploy(self, ctx: &mut DeployContext<TestEnv>) -> Result<Self::Handle, DynError> {
+            let child = ctx.deploy_and_expose(ChildApp).await?;
+            let sibling = ctx.deploy_and_expose(SiblingApp).await?;
+            let stack = StackHandle { child, sibling };
+
+            ctx.expose(stack.clone())?;
+
+            Ok(stack)
+        }
+    }
+
+    #[tokio::test]
+    async fn nested_deployment_can_expose_multiple_typed_handles() {
+        let mut ctx = test_context();
+        let stack = ctx.deploy(StackApp).await.expect("deploy stack app");
+
+        assert_eq!(stack.child, ChildHandle { id: "child" });
+        assert_eq!(stack.sibling, SiblingHandle { id: "sibling" });
+        assert!(ctx.get::<ChildHandle>().is_some());
+        assert!(ctx.get::<SiblingHandle>().is_some());
+        assert!(ctx.get::<StackHandle>().is_some());
+    }
+
+    fn test_context() -> DeployContext<TestEnv> {
+        DeployContext::new(TestDeployment, NodeClients::default())
+    }
+}

testing-framework/app/src/deployment.rs

@@ -0,0 +1,30 @@
+use async_trait::async_trait;
+use testing_framework_core::scenario::{Application, DynError};
+
+use crate::DeployContext;
+
+/// Typed runtime capability returned by an application deployment.
+///
+/// Handles are cloned when retrieved from the runtime registry. Managed
+/// resources must be acquired through TF adapters, which register scenario
+/// cleanup independently from these clones.
+pub trait AppHandle: Clone + Send + Sync + 'static {}
+
+impl<T> AppHandle for T where T: Clone + Send + Sync + 'static {}
+
+/// Deploys one reusable application preset and returns its typed handle.
+#[async_trait]
+pub trait AppDeployment<E>: Send + 'static
+where
+    E: Application,
+{
+    /// Runtime capability produced by this deployment.
+    type Handle: AppHandle;
+
+    /// Prepares the application and returns its typed runtime access handle.
+    ///
+    /// Child deployments can be composed with [`DeployContext::deploy`] or
+    /// [`DeployContext::deploy_and_expose`]. Any handle exposed through the
+    /// context remains available through the scenario runtime.
+    async fn deploy(self, ctx: &mut DeployContext<E>) -> Result<Self::Handle, DynError>;
+}

testing-framework/app/src/error.rs

@@ -0,0 +1,29 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+/// Error returned when a typed application handle cannot be registered or
+/// found.
+pub enum AppDeployError {
+    /// No handle exists for the requested type and optional name.
+    #[error("app handle is not exposed: {type_name}{suffix}", suffix = format_name(.name))]
+    HandleMissing {
+        /// Rust type name of the requested handle.
+        type_name: &'static str,
+        /// Instance name, or `None` for the default unnamed handle.
+        name: Option<String>,
+    },
+    /// A handle already exists for the same type and optional name.
+    #[error("app handle is already exposed: {type_name}{suffix}", suffix = format_name(.name))]
+    DuplicateHandle {
+        /// Rust type name of the duplicate handle.
+        type_name: &'static str,
+        /// Instance name, or `None` for the default unnamed handle.
+        name: Option<String>,
+    },
+}
+
+fn format_name(name: &Option<String>) -> String {
+    name.as_ref()
+        .map(|name| format!(" named {name:?}"))
+        .unwrap_or_default()
+}

testing-framework/app/src/extension.rs

@@ -0,0 +1,233 @@
+use async_trait::async_trait;
+use testing_framework_core::scenario::{
+    Application, CoreBuilderExt, DynError, NodeClients, PreparedRuntimeExtension, RunContext,
+    RuntimeExtensionFactory, internal::CoreBuilderAccess,
+};
+
+use crate::{AppDeployment, AppHandle, AppRuntime, DeployContext};
+
+/// Adapts an [`AppDeployment`] to the scenario runtime extension lifecycle.
+///
+/// A factory is normally installed through [`AppScenarioBuilderExt::with_app`]
+/// rather than constructed directly.
+pub struct AppDeploymentFactory<A> {
+    app: A,
+}
+
+impl<A> AppDeploymentFactory<A> {
+    /// Creates a runtime extension factory for `app`.
+    pub const fn new(app: A) -> Self {
+        Self { app }
+    }
+}
+
+#[async_trait]
+impl<E, A> RuntimeExtensionFactory<E> for AppDeploymentFactory<A>
+where
+    E: Application,
+    A: AppDeployment<E> + Clone + Sync,
+{
+    async fn prepare(
+        &self,
+        deployment: &E::Deployment,
+        node_clients: NodeClients<E>,
+    ) -> Result<PreparedRuntimeExtension, DynError> {
+        let mut ctx = DeployContext::new(deployment.clone(), node_clients);
+
+        let handle = ctx.deploy(self.app.clone()).await?;
+
+        if !ctx.contains::<A::Handle>() {
+            ctx.expose(handle)?;
+        }
+
+        let (handles, cleanup) = ctx.into_runtime_parts();
+        let runtime = AppRuntime::new(handles);
+
+        Ok(match cleanup {
+            Some(cleanup) => PreparedRuntimeExtension::with_cleanup(runtime, cleanup),
+            None => PreparedRuntimeExtension::new(runtime),
+        })
+    }
+}
+
+/// Adds composable application deployments to scenario builders.
+pub trait AppScenarioBuilderExt: CoreBuilderAccess + Sized {
+    /// Registers an application deployment to prepare before workloads start.
+    ///
+    /// If the root deployment does not expose its returned handle itself, the
+    /// factory exposes it as the default handle automatically.
+    #[must_use]
+    fn with_app<A>(self, app: A) -> Self
+    where
+        A: AppDeployment<Self::Env> + Clone + Sync,
+    {
+        self.with_runtime_extension_factory(Box::new(AppDeploymentFactory::new(app)))
+    }
+}
+
+impl<T> AppScenarioBuilderExt for T where T: CoreBuilderAccess {}
+
+/// Retrieves application handles from a running scenario.
+pub trait AppRunContextExt<E: Application> {
+    /// Returns the default handle for `T`, or `None` if it is not exposed.
+    fn app<T>(&self) -> Option<T>
+    where
+        T: AppHandle;
+
+    /// Returns a named handle for `T`, or `None` if it is not exposed.
+    fn app_named<T>(&self, name: &str) -> Option<T>
+    where
+        T: AppHandle;
+
+    /// Returns the default handle for `T` or an error when it is unavailable.
+    fn require_app<T>(&self) -> Result<T, DynError>
+    where
+        T: AppHandle;
+
+    /// Returns a named handle for `T` or an error when it is unavailable.
+    fn require_app_named<T>(&self, name: &str) -> Result<T, DynError>
+    where
+        T: AppHandle;
+}
+
+impl<E> AppRunContextExt<E> for RunContext<E>
+where
+    E: Application,
+{
+    fn app<T>(&self) -> Option<T>
+    where
+        T: AppHandle,
+    {
+        self.extension::<AppRuntime>()?.get()
+    }
+
+    fn app_named<T>(&self, name: &str) -> Option<T>
+    where
+        T: AppHandle,
+    {
+        self.extension::<AppRuntime>()?.get_named(name)
+    }
+
+    fn require_app<T>(&self) -> Result<T, DynError>
+    where
+        T: AppHandle,
+    {
+        self.require_extension::<AppRuntime>()?
+            .require()
+            .map_err(Into::into)
+    }
+
+    fn require_app_named<T>(&self, name: &str) -> Result<T, DynError>
+    where
+        T: AppHandle,
+    {
+        self.require_extension::<AppRuntime>()?
+            .require_named(name)
+            .map_err(Into::into)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    };
+
+    use async_trait::async_trait;
+    use testing_framework_core::{
+        scenario::{Application, DynError, NodeClients, RuntimeExtensionFactory},
+        topology::DeploymentDescriptor,
+    };
+
+    use super::AppDeploymentFactory;
+    use crate::{AppDeployment, DeployContext};
+
+    #[derive(Clone)]
+    struct TestDeployment;
+
+    impl DeploymentDescriptor for TestDeployment {
+        fn node_count(&self) -> usize {
+            0
+        }
+    }
+
+    struct TestEnv;
+
+    #[async_trait]
+    impl Application for TestEnv {
+        type Deployment = TestDeployment;
+        type NodeClient = ();
+        type NodeConfig = ();
+    }
+
+    #[derive(Clone)]
+    struct FailingApp {
+        dropped: Arc<AtomicBool>,
+    }
+
+    #[derive(Clone)]
+    struct OwnedHandle {
+        _resource: Arc<DropProbe>,
+    }
+
+    struct DropProbe {
+        dropped: Arc<AtomicBool>,
+    }
+
+    impl Drop for DropProbe {
+        fn drop(&mut self) {
+            self.dropped.store(true, Ordering::SeqCst);
+        }
+    }
+
+    #[async_trait]
+    impl AppDeployment<TestEnv> for FailingApp {
+        type Handle = OwnedHandle;
+
+        async fn deploy(self, ctx: &mut DeployContext<TestEnv>) -> Result<Self::Handle, DynError> {
+            ctx.expose(OwnedHandle {
+                _resource: Arc::new(DropProbe {
+                    dropped: Arc::clone(&self.dropped),
+                }),
+            })?;
+            Err("deployment failed".into())
+        }
+    }
+
+    #[tokio::test]
+    async fn deployment_failure_cleans_up_partial_resources() {
+        let dropped = Arc::new(AtomicBool::new(false));
+        let factory = AppDeploymentFactory::new(FailingApp {
+            dropped: Arc::clone(&dropped),
+        });
+
+        let result = factory
+            .prepare(test_deployment(), NodeClients::default())
+            .await;
+
+        assert!(result.is_err(), "deployment must fail");
+
+        assert!(dropped.load(Ordering::SeqCst));
+    }
+
+    #[tokio::test]
+    async fn app_factory_can_prepare_more_than_once() {
+        let dropped = Arc::new(AtomicBool::new(false));
+        let factory = AppDeploymentFactory::new(FailingApp {
+            dropped: Arc::clone(&dropped),
+        });
+
+        for _ in 0..2 {
+            let result = factory
+                .prepare(test_deployment(), NodeClients::default())
+                .await;
+            assert!(result.is_err(), "deployment must fail");
+        }
+    }
+
+    fn test_deployment() -> &'static TestDeployment {
+        static DEPLOYMENT: TestDeployment = TestDeployment;
+        &DEPLOYMENT
+    }
+}

testing-framework/app/src/host.rs

@@ -0,0 +1,55 @@
+use async_trait::async_trait;
+use testing_framework_core::{
+    scenario::{Application, DynError, NodeAccess, ScenarioBuilder},
+    topology::DeploymentDescriptor,
+};
+use testing_framework_runner_local::{LocalDeployerEnv, ProcessDeployer};
+
+#[derive(Clone, Default)]
+/// Empty outer topology for scenarios whose entire system is deployed as apps.
+///
+/// Application deployments registered with [`crate::AppScenarioBuilderExt`]
+/// provide all processes and clusters, so this topology contains no nodes.
+pub struct AppHostTopology;
+
+impl DeploymentDescriptor for AppHostTopology {
+    fn node_count(&self) -> usize {
+        0
+    }
+}
+
+/// Testing-framework application environment for [`AppHost`] scenarios.
+///
+/// This environment intentionally has no outer node client.
+/// Application-specific clients are exposed as typed handles instead.
+pub struct AppHostEnv;
+
+#[async_trait]
+impl Application for AppHostEnv {
+    type Deployment = AppHostTopology;
+    type NodeClient = ();
+    type NodeConfig = ();
+
+    fn build_node_client(_access: &NodeAccess) -> Result<Self::NodeClient, DynError> {
+        Err(std::io::Error::other("app host does not expose node clients").into())
+    }
+}
+
+#[async_trait]
+impl LocalDeployerEnv for AppHostEnv {}
+
+/// Entry point for a scenario composed entirely from application deployments.
+pub struct AppHost;
+
+impl AppHost {
+    /// Creates an empty scenario builder ready for `.with_app(...)` calls.
+    #[must_use]
+    pub fn scenario() -> AppHostScenarioBuilder {
+        ScenarioBuilder::with_deployment(AppHostTopology)
+    }
+}
+
+/// Scenario builder for an application-hosted heterogeneous stack.
+pub type AppHostScenarioBuilder = ScenarioBuilder<AppHostEnv>;
+/// Local process deployer used to execute an [`AppHost`] scenario.
+pub type AppHostLocalDeployer = ProcessDeployer<AppHostEnv>;

testing-framework/app/src/lib.rs

@@ -0,0 +1,35 @@
+//! Composable application deployment for heterogeneous test systems.
+//!
+//! The core testing framework models one
+//! [`Application`](testing_framework_core::scenario::Application)
+//! and its uniform node topology. This crate adds an application layer for
+//! scenarios that also need singleton processes, additional clusters,
+//! attached services, or a stack of several different applications.
+//!
+//! Implement [`AppDeployment`] in the application repository, compose child
+//! deployments through [`DeployContext`], and expose typed handles to workloads
+//! with [`AppRunContextExt`]. TF adapters register managed resources with the
+//! scenario cleanup path, while attached and external apps remain explicit.
+
+#![warn(missing_docs)]
+
+mod cleanup;
+mod context;
+mod deployment;
+mod error;
+mod extension;
+mod host;
+mod local;
+mod process;
+mod registry;
+
+pub use context::DeployContext;
+pub use deployment::{AppDeployment, AppHandle};
+pub use error::AppDeployError;
+pub use extension::{AppDeploymentFactory, AppRunContextExt, AppScenarioBuilderExt};
+pub use host::{
+    AppHost, AppHostEnv, AppHostLocalDeployer, AppHostScenarioBuilder, AppHostTopology,
+};
+pub use local::LocalAppCluster;
+pub use process::{LocalProcessApp, LocalProcessHandle};
+pub use registry::{AppRuntime, HandleRegistry};

testing-framework/app/src/local.rs

@@ -0,0 +1,238 @@
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, Ordering},
+};
+
+use testing_framework_core::{
+    scenario::{DynError, NodeClients, StartNodeOptions, StartedNode, internal::CleanupGuard},
+    topology::DeploymentDescriptor,
+};
+use testing_framework_runner_local::{LocalDeployerEnv, ManualCluster, ProcessDeployer};
+
+/// Control handle for an additional uniform cluster of local processes.
+///
+/// Unlike the outer scenario cluster, this cluster is deployed as one child of
+/// a heterogeneous application stack. The scenario owns its lifetime and stops
+/// every node during cleanup even if a handle clone remains elsewhere.
+pub struct LocalAppCluster<E>
+where
+    E: LocalDeployerEnv,
+{
+    deployment: E::Deployment,
+    owner: Arc<LocalClusterOwner<E>>,
+}
+
+struct LocalClusterOwner<E>
+where
+    E: LocalDeployerEnv,
+{
+    cluster: ManualCluster<E>,
+    closed: AtomicBool,
+}
+
+impl<E> Drop for LocalClusterOwner<E>
+where
+    E: LocalDeployerEnv,
+{
+    fn drop(&mut self) {
+        self.close();
+    }
+}
+
+impl<E> LocalClusterOwner<E>
+where
+    E: LocalDeployerEnv,
+{
+    fn close(&self) {
+        if !self.closed.swap(true, Ordering::AcqRel) {
+            self.cluster.stop_all();
+        }
+    }
+
+    fn ensure_open(&self) -> Result<(), DynError> {
+        if self.closed.load(Ordering::Acquire) {
+            Err("local app cluster is no longer owned by an active run".into())
+        } else {
+            Ok(())
+        }
+    }
+}
+
+struct LocalClusterCleanup<E>
+where
+    E: LocalDeployerEnv,
+{
+    owner: Arc<LocalClusterOwner<E>>,
+}
+
+impl<E> CleanupGuard for LocalClusterCleanup<E>
+where
+    E: LocalDeployerEnv,
+{
+    fn cleanup(self: Box<Self>) {
+        self.owner.close();
+    }
+}
+
+impl<E> Clone for LocalAppCluster<E>
+where
+    E: LocalDeployerEnv,
+{
+    fn clone(&self) -> Self {
+        Self {
+            deployment: self.deployment.clone(),
+            owner: Arc::clone(&self.owner),
+        }
+    }
+}
+
+impl<E> LocalAppCluster<E>
+where
+    E: LocalDeployerEnv,
+{
+    /// Starts every node described by `deployment` and waits for readiness.
+    pub(crate) async fn start(deployment: E::Deployment) -> Result<Self, DynError> {
+        let cluster =
+            ProcessDeployer::<E>::default().manual_cluster_from_descriptors(deployment.clone());
+        let owner = Arc::new(LocalClusterOwner {
+            cluster,
+            closed: AtomicBool::new(false),
+        });
+
+        start_all_nodes(&owner.cluster, deployment.node_count()).await?;
+        owner.cluster.wait_network_ready().await?;
+
+        Ok(Self { deployment, owner })
+    }
+
+    /// Returns this cluster's deployment descriptor.
+    #[must_use]
+    pub fn deployment(&self) -> &E::Deployment {
+        &self.deployment
+    }
+
+    /// Returns the number of nodes described by the deployment.
+    #[must_use]
+    pub fn node_count(&self) -> usize {
+        self.deployment.node_count()
+    }
+
+    /// Returns a shared snapshot-capable collection of node clients.
+    #[must_use]
+    pub fn node_clients(&self) -> NodeClients<E> {
+        self.owner.cluster.node_clients()
+    }
+
+    /// Returns the client for `name`, if that node has started.
+    #[must_use]
+    pub fn node_client(&self, name: &str) -> Option<E::NodeClient> {
+        self.owner.cluster.node_client(name)
+    }
+
+    /// Returns the operating-system process ID for `name`, if it is running.
+    #[must_use]
+    pub fn node_pid(&self, name: &str) -> Option<u32> {
+        self.owner.cluster.node_pid(name)
+    }
+
+    /// Returns a snapshot of all currently available node clients.
+    #[must_use]
+    pub fn clients(&self) -> Vec<E::NodeClient> {
+        self.node_clients().snapshot()
+    }
+
+    /// Returns the first available node client.
+    #[must_use]
+    pub fn first_client(&self) -> Option<E::NodeClient> {
+        self.node_clients()
+            .with_clients(|clients| clients.first().cloned())
+    }
+
+    /// Starts the named node with default start options.
+    pub async fn start_node(&self, name: &str) -> Result<StartedNode<E>, DynError> {
+        self.owner.ensure_open()?;
+        Ok(self.owner.cluster.start_node(name).await?)
+    }
+
+    /// Starts the named node with explicit start options.
+    pub async fn start_node_with(
+        &self,
+        name: &str,
+        options: StartNodeOptions<E>,
+    ) -> Result<StartedNode<E>, DynError> {
+        self.owner.ensure_open()?;
+        Ok(self.owner.cluster.start_node_with(name, options).await?)
+    }
+
+    /// Stops the named node.
+    pub async fn stop_node(&self, name: &str) -> Result<(), DynError> {
+        self.owner.ensure_open()?;
+        Ok(self.owner.cluster.stop_node(name).await?)
+    }
+
+    /// Restarts the named node with its existing configuration.
+    pub async fn restart_node(&self, name: &str) -> Result<(), DynError> {
+        self.owner.ensure_open()?;
+        Ok(self.owner.cluster.restart_node(name).await?)
+    }
+
+    /// Restarts the named node with explicit start options.
+    pub async fn restart_node_with(
+        &self,
+        name: &str,
+        options: StartNodeOptions<E>,
+    ) -> Result<(), DynError> {
+        self.owner.ensure_open()?;
+        Ok(self.owner.cluster.restart_node_with(name, options).await?)
+    }
+
+    /// Waits until the cluster-level network readiness condition succeeds.
+    pub async fn wait_network_ready(&self) -> Result<(), DynError> {
+        self.owner.ensure_open()?;
+        Ok(self.owner.cluster.wait_network_ready().await?)
+    }
+
+    /// Waits until the named node reports ready.
+    pub async fn wait_node_ready(&self, name: &str) -> Result<(), DynError> {
+        self.owner.ensure_open()?;
+        Ok(self.owner.cluster.wait_node_ready(name).await?)
+    }
+
+    /// Stops every node while keeping the cluster available for a later start.
+    pub fn stop_all(&self) -> Result<(), DynError> {
+        self.owner.ensure_open()?;
+        self.owner.cluster.stop_all();
+        Ok(())
+    }
+
+    /// Starts every node with its original deployment and waits for readiness.
+    pub async fn start_all(&self) -> Result<(), DynError> {
+        self.owner.ensure_open()?;
+        start_all_nodes(&self.owner.cluster, self.node_count()).await?;
+        self.owner.cluster.wait_network_ready().await?;
+        Ok(())
+    }
+
+    /// Stops and starts every node, then waits for network readiness.
+    pub async fn restart_all(&self) -> Result<(), DynError> {
+        self.stop_all()?;
+        self.start_all().await
+    }
+
+    pub(crate) fn cleanup_guard(&self) -> Box<dyn CleanupGuard> {
+        Box::new(LocalClusterCleanup {
+            owner: Arc::clone(&self.owner),
+        })
+    }
+}
+
+async fn start_all_nodes<E>(cluster: &ManualCluster<E>, node_count: usize) -> Result<(), DynError>
+where
+    E: LocalDeployerEnv,
+{
+    for index in 0..node_count {
+        cluster.start_node(&format!("node-{index}")).await?;
+    }
+
+    Ok(())
+}

testing-framework/app/src/process.rs

@@ -0,0 +1,387 @@
+use std::{
+    future::Future,
+    path::PathBuf,
+    pin::Pin,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+};
+
+use async_trait::async_trait;
+use testing_framework_core::scenario::{Application, DynError, internal::CleanupGuard};
+use testing_framework_runner_local::{LaunchSpec, NodeEndpoints, ProcessNode};
+use tokio::sync::Mutex;
+
+use crate::{AppDeployment, DeployContext};
+
+type ReadinessFuture = Pin<Box<dyn Future<Output = Result<(), DynError>> + Send>>;
+type Readiness<C> = Arc<dyn Fn(NodeEndpoints, C) -> ReadinessFuture + Send + Sync>;
+
+/// One local binary process that can be composed with other app deployments.
+///
+/// This is the small path for heterogeneous stacks. It does not require the
+/// process to pretend to be a uniform TF node cluster. The app repository owns
+/// its launch files, client type, and readiness check; TF owns process
+/// lifetime, working directories, and teardown.
+pub struct LocalProcessApp<C> {
+    label: String,
+    launch: LaunchSpec,
+    endpoints: NodeEndpoints,
+    client: C,
+    readiness: Option<Readiness<C>>,
+    keep_tempdir: bool,
+    persist_dir: Option<PathBuf>,
+    snapshot_dir: Option<PathBuf>,
+}
+
+impl<C> LocalProcessApp<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    /// Describes a process and the client returned through its runtime handle.
+    ///
+    /// `endpoints` describes the addresses assigned by the application-specific
+    /// launch configuration; the generic process layer does not allocate them.
+    #[must_use]
+    pub fn new(
+        label: impl Into<String>,
+        launch: LaunchSpec,
+        endpoints: NodeEndpoints,
+        client: C,
+    ) -> Self {
+        Self {
+            label: label.into(),
+            launch,
+            endpoints,
+            client,
+            readiness: None,
+            keep_tempdir: false,
+            persist_dir: None,
+            snapshot_dir: None,
+        }
+    }
+
+    /// Adds an asynchronous readiness check that runs after the process starts.
+    ///
+    /// A readiness failure stops the process and fails deployment.
+    #[must_use]
+    pub fn with_readiness<F, Fut>(mut self, readiness: F) -> Self
+    where
+        F: Fn(NodeEndpoints, C) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), DynError>> + Send + 'static,
+    {
+        self.readiness = Some(Arc::new(move |endpoints, client| {
+            Box::pin(readiness(endpoints, client))
+        }));
+        self
+    }
+
+    /// Controls whether the generated process working directory survives
+    /// teardown.
+    #[must_use]
+    pub fn keep_tempdir(mut self, keep: bool) -> Self {
+        self.keep_tempdir = keep;
+        self
+    }
+
+    /// Copies persisted state from `path` into the process working directory.
+    #[must_use]
+    pub fn with_persist_dir(mut self, path: impl Into<PathBuf>) -> Self {
+        self.persist_dir = Some(path.into());
+        self
+    }
+
+    /// Copies snapshot state from `path` into the process working directory.
+    #[must_use]
+    pub fn with_snapshot_dir(mut self, path: impl Into<PathBuf>) -> Self {
+        self.snapshot_dir = Some(path.into());
+        self
+    }
+}
+
+#[async_trait]
+impl<E, C> AppDeployment<E> for LocalProcessApp<C>
+where
+    E: Application,
+    C: Clone + Send + Sync + 'static,
+{
+    type Handle = LocalProcessHandle<C>;
+
+    async fn deploy(self, ctx: &mut DeployContext<E>) -> Result<Self::Handle, DynError> {
+        let Self {
+            label,
+            launch,
+            endpoints,
+            client,
+            readiness,
+            keep_tempdir,
+            persist_dir,
+            snapshot_dir,
+        } = self;
+
+        let mut process = ProcessNode::spawn(
+            &label,
+            (),
+            move |(), _working_dir, _label| Ok(launch),
+            move |()| Ok(endpoints),
+            keep_tempdir,
+            persist_dir.as_deref(),
+            snapshot_dir.as_deref(),
+            move |_endpoints| Ok(client),
+        )
+        .await?;
+
+        let endpoints = process.endpoints().clone();
+        let client = process.client();
+
+        if let Some(readiness) = &readiness {
+            if let Err(error) = readiness(endpoints.clone(), client.clone()).await {
+                process.stop().await;
+                return Err(error);
+            }
+        }
+
+        let state = Arc::new(LocalProcessState {
+            process: Mutex::new(process),
+            endpoints: endpoints.clone(),
+            client: client.clone(),
+            readiness,
+            closed: AtomicBool::new(false),
+        });
+        let handle = LocalProcessHandle {
+            state,
+            endpoints,
+            client,
+        };
+        ctx.register_cleanup(handle.cleanup_guard());
+
+        Ok(handle)
+    }
+}
+
+struct LocalProcessState<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    process: Mutex<ProcessNode<(), C>>,
+    endpoints: NodeEndpoints,
+    client: C,
+    readiness: Option<Readiness<C>>,
+    closed: AtomicBool,
+}
+
+impl<C> LocalProcessState<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    fn ensure_open(&self) -> Result<(), DynError> {
+        if self.closed.load(Ordering::Acquire) {
+            Err("local process is no longer owned by an active run".into())
+        } else {
+            Ok(())
+        }
+    }
+
+    fn close(&self) {
+        if self.closed.swap(true, Ordering::AcqRel) {
+            return;
+        }
+
+        if let Ok(mut process) = self.process.try_lock() {
+            process.stop_blocking();
+            return;
+        }
+
+        std::thread::scope(|scope| {
+            let worker = scope.spawn(|| self.process.blocking_lock().stop_blocking());
+            let _ = worker.join();
+        });
+    }
+
+    async fn wait_ready(&self) -> Result<(), DynError> {
+        if let Some(readiness) = &self.readiness {
+            readiness(self.endpoints.clone(), self.client.clone()).await?;
+        }
+        Ok(())
+    }
+}
+
+struct LocalProcessCleanup<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    state: Arc<LocalProcessState<C>>,
+}
+
+impl<C> CleanupGuard for LocalProcessCleanup<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    fn cleanup(self: Box<Self>) {
+        self.state.close();
+    }
+}
+
+/// Shared ownership and control handle for one local application process.
+///
+/// Clones refer to the same process. The scenario owns its lifetime, so cleanup
+/// stops it even if a handle clone remains elsewhere.
+pub struct LocalProcessHandle<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    state: Arc<LocalProcessState<C>>,
+    endpoints: NodeEndpoints,
+    client: C,
+}
+
+impl<C> Clone for LocalProcessHandle<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    fn clone(&self) -> Self {
+        Self {
+            state: Arc::clone(&self.state),
+            endpoints: self.endpoints.clone(),
+            client: self.client.clone(),
+        }
+    }
+}
+
+impl<C> LocalProcessHandle<C>
+where
+    C: Clone + Send + Sync + 'static,
+{
+    /// Returns a clone of the application-specific client.
+    #[must_use]
+    pub fn client(&self) -> C {
+        self.client.clone()
+    }
+
+    /// Returns the process endpoints supplied at deployment time.
+    #[must_use]
+    pub fn endpoints(&self) -> &NodeEndpoints {
+        &self.endpoints
+    }
+
+    /// Returns the operating-system process ID.
+    pub async fn pid(&self) -> u32 {
+        self.state.process.lock().await.pid()
+    }
+
+    /// Returns whether the child process is still running.
+    pub async fn is_running(&self) -> bool {
+        if self.state.closed.load(Ordering::Acquire) {
+            return false;
+        }
+        self.state.process.lock().await.is_running()
+    }
+
+    /// Returns the generated working directory used by the process.
+    pub async fn working_dir(&self) -> PathBuf {
+        self.state.process.lock().await.working_dir().to_owned()
+    }
+
+    /// Starts a process that was previously stopped.
+    pub async fn start(&self) -> Result<(), DynError> {
+        self.state.ensure_open()?;
+        let mut process = self.state.process.lock().await;
+        if !process.is_running() {
+            process.restart().await?;
+        }
+        drop(process);
+        self.state.wait_ready().await?;
+        Ok(())
+    }
+
+    /// Restarts the process with its original launch specification.
+    pub async fn restart(&self) -> Result<(), DynError> {
+        self.state.ensure_open()?;
+        self.state.process.lock().await.restart().await?;
+        self.state.wait_ready().await?;
+        Ok(())
+    }
+
+    /// Stops the process without waiting for the handle to be dropped.
+    pub async fn stop(&self) -> Result<(), DynError> {
+        self.state.ensure_open()?;
+        self.state.process.lock().await.stop().await;
+        Ok(())
+    }
+
+    /// Prevents deletion of the temporary working directory on teardown.
+    pub async fn keep_tempdir(&self) -> std::io::Result<()> {
+        self.state.process.lock().await.keep_tempdir()
+    }
+
+    pub(crate) fn cleanup_guard(&self) -> Box<dyn CleanupGuard> {
+        Box::new(LocalProcessCleanup {
+            state: Arc::clone(&self.state),
+        })
+    }
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use async_trait::async_trait;
+    use testing_framework_core::{
+        scenario::{Application, NodeClients},
+        topology::DeploymentDescriptor,
+    };
+    use testing_framework_runner_local::{LaunchSpec, NodeEndpoints};
+
+    use super::LocalProcessApp;
+    use crate::DeployContext;
+
+    #[derive(Clone)]
+    struct TestDeployment;
+
+    impl DeploymentDescriptor for TestDeployment {
+        fn node_count(&self) -> usize {
+            0
+        }
+    }
+
+    struct TestEnv;
+
+    #[async_trait]
+    impl Application for TestEnv {
+        type Deployment = TestDeployment;
+        type NodeClient = ();
+        type NodeConfig = ();
+    }
+
+    #[tokio::test]
+    async fn process_lifetime_follows_scenario_ownership() {
+        let launch = LaunchSpec {
+            binary: "/bin/sh".into(),
+            args: vec!["-c".into(), "sleep 30".into()],
+            ..LaunchSpec::default()
+        };
+        let app = LocalProcessApp::new(
+            "test-process",
+            launch,
+            NodeEndpoints::default(),
+            "client".to_owned(),
+        )
+        .with_readiness(|_endpoints, client| async move {
+            assert_eq!(client, "client");
+            Ok(())
+        });
+        let mut ctx = DeployContext::<TestEnv>::new(TestDeployment, NodeClients::default());
+
+        let handle = ctx.deploy(app).await.expect("deploy process");
+        assert!(handle.is_running().await);
+        assert_eq!(handle.client(), "client");
+
+        handle.stop().await.unwrap();
+        assert!(!handle.is_running().await);
+        handle.start().await.unwrap();
+        assert!(handle.is_running().await);
+
+        drop(ctx);
+        assert!(!handle.is_running().await);
+        assert!(handle.restart().await.is_err());
+    }
+}

testing-framework/app/src/registry.rs

@@ -0,0 +1,281 @@
+use std::{
+    any::{Any, TypeId, type_name},
+    sync::Arc,
+};
+
+use crate::{AppDeployError, AppHandle};
+
+const DEFAULT_HANDLE_NAME: &str = "";
+
+#[derive(Default)]
+/// Type-indexed storage for application runtime handles.
+///
+/// One unnamed handle may be stored per concrete type. Named handles allow
+/// multiple instances of the same type. Entries retain exposure order and are
+/// released in reverse order. Managed resource teardown is registered
+/// separately by TF adapters and does not depend on handle clone counts.
+pub struct HandleRegistry {
+    handles: Vec<(HandleKey, Box<StoredHandle>)>,
+}
+
+impl HandleRegistry {
+    /// Creates an empty registry.
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Stores the default unnamed handle for `T`.
+    ///
+    /// Returns [`AppDeployError::DuplicateHandle`] if one is already stored.
+    pub fn expose<T>(&mut self, handle: T) -> Result<(), AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.expose_named(DEFAULT_HANDLE_NAME, handle)
+    }
+
+    /// Stores `handle` under its concrete type and `name`.
+    ///
+    /// Returns [`AppDeployError::DuplicateHandle`] for an existing type/name
+    /// pair.
+    pub fn expose_named<T>(
+        &mut self,
+        name: impl Into<String>,
+        handle: T,
+    ) -> Result<(), AppDeployError>
+    where
+        T: AppHandle,
+    {
+        let key = HandleKey::new::<T>(name);
+        if self.handles.iter().any(|(existing, _)| existing == &key) {
+            return Err(AppDeployError::DuplicateHandle {
+                type_name: type_name::<T>(),
+                name: key.display_name().map(ToOwned::to_owned),
+            });
+        }
+
+        self.handles.push((key, Box::new(handle)));
+        Ok(())
+    }
+
+    /// Returns a clone of the default handle for `T`, if present.
+    #[must_use]
+    pub fn get<T>(&self) -> Option<T>
+    where
+        T: AppHandle,
+    {
+        self.get_named(DEFAULT_HANDLE_NAME)
+    }
+
+    /// Returns a clone of the named handle for `T`, if present.
+    #[must_use]
+    pub fn get_named<T>(&self, name: &str) -> Option<T>
+    where
+        T: AppHandle,
+    {
+        let key = HandleKey::new::<T>(name);
+        self.handles
+            .iter()
+            .find(|(existing, _)| existing == &key)
+            .and_then(|(_, handle)| handle.as_ref().downcast_ref::<T>())
+            .cloned()
+    }
+
+    /// Returns the default handle for `T` or [`AppDeployError::HandleMissing`].
+    pub fn require<T>(&self) -> Result<T, AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.require_named(DEFAULT_HANDLE_NAME)
+    }
+
+    /// Returns the named handle for `T` or [`AppDeployError::HandleMissing`].
+    pub fn require_named<T>(&self, name: &str) -> Result<T, AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.get_named(name)
+            .ok_or_else(|| AppDeployError::HandleMissing {
+                type_name: type_name::<T>(),
+                name: (!name.is_empty()).then(|| name.to_owned()),
+            })
+    }
+
+    /// Returns whether the default handle for `T` is present.
+    #[must_use]
+    pub fn contains<T>(&self) -> bool
+    where
+        T: AppHandle,
+    {
+        self.contains_named::<T>(DEFAULT_HANDLE_NAME)
+    }
+
+    /// Returns whether the named handle for `T` is present.
+    #[must_use]
+    pub fn contains_named<T>(&self, name: &str) -> bool
+    where
+        T: AppHandle,
+    {
+        let key = HandleKey::new::<T>(name);
+        self.handles.iter().any(|(existing, _)| existing == &key)
+    }
+
+    /// Returns whether the registry contains no handles.
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.handles.is_empty()
+    }
+}
+
+impl Drop for HandleRegistry {
+    fn drop(&mut self) {
+        // Release arbitrary handle state deterministically. Managed resources
+        // use the deployment cleanup stack, whose order follows acquisition.
+        while self.handles.pop().is_some() {}
+    }
+}
+
+#[derive(Clone, Default)]
+/// Scenario runtime extension containing exposed application handles.
+///
+/// Runtime clones share one registry. Managed resources have a separate
+/// scenario-owned cleanup stack.
+pub struct AppRuntime {
+    handles: Arc<HandleRegistry>,
+}
+
+impl AppRuntime {
+    /// Wraps a prepared handle registry as a scenario runtime extension.
+    #[must_use]
+    pub fn new(handles: HandleRegistry) -> Self {
+        Self {
+            handles: Arc::new(handles),
+        }
+    }
+
+    /// Returns a clone of the default handle for `T`, if present.
+    #[must_use]
+    pub fn get<T>(&self) -> Option<T>
+    where
+        T: AppHandle,
+    {
+        self.handles.get()
+    }
+
+    /// Returns a clone of the named handle for `T`, if present.
+    #[must_use]
+    pub fn get_named<T>(&self, name: &str) -> Option<T>
+    where
+        T: AppHandle,
+    {
+        self.handles.get_named(name)
+    }
+
+    /// Returns the default handle for `T` or [`AppDeployError::HandleMissing`].
+    pub fn require<T>(&self) -> Result<T, AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.handles.require()
+    }
+
+    /// Returns the named handle for `T` or [`AppDeployError::HandleMissing`].
+    pub fn require_named<T>(&self, name: &str) -> Result<T, AppDeployError>
+    where
+        T: AppHandle,
+    {
+        self.handles.require_named(name)
+    }
+}
+
+#[derive(Eq, Hash, PartialEq)]
+struct HandleKey {
+    type_id: TypeId,
+    name: String,
+}
+
+impl HandleKey {
+    fn new<T: 'static>(name: impl Into<String>) -> Self {
+        Self {
+            type_id: TypeId::of::<T>(),
+            name: name.into(),
+        }
+    }
+
+    fn display_name(&self) -> Option<&str> {
+        (!self.name.is_empty()).then_some(self.name.as_str())
+    }
+}
+
+type StoredHandle = dyn Any + Send + Sync;
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use super::HandleRegistry;
+
+    #[derive(Clone, Debug, Eq, PartialEq)]
+    struct Handle(u8);
+
+    #[test]
+    fn named_handles_allow_multiple_instances_of_one_type() {
+        let mut handles = HandleRegistry::new();
+        handles.expose_named("left", Handle(1)).unwrap();
+        handles.expose_named("right", Handle(2)).unwrap();
+
+        assert_eq!(handles.get_named("left"), Some(Handle(1)));
+        assert_eq!(handles.get_named("right"), Some(Handle(2)));
+    }
+
+    #[test]
+    fn duplicate_handle_is_rejected() {
+        let mut handles = HandleRegistry::new();
+        handles.expose(Handle(1)).unwrap();
+
+        let error = handles.expose(Handle(2)).unwrap_err();
+        assert!(error.to_string().contains("already exposed"));
+        assert_eq!(handles.get(), Some(Handle(1)));
+    }
+
+    #[test]
+    fn handles_drop_in_reverse_exposure_order() {
+        let order = Arc::new(Mutex::new(Vec::new()));
+        let mut handles = HandleRegistry::new();
+        handles
+            .expose_named("first", OwnedHandle::new("first", Arc::clone(&order)))
+            .unwrap();
+        handles
+            .expose_named("second", OwnedHandle::new("second", Arc::clone(&order)))
+            .unwrap();
+
+        drop(handles);
+
+        assert_eq!(*order.lock().unwrap(), ["second", "first"]);
+    }
+
+    #[derive(Clone)]
+    struct OwnedHandle {
+        _resource: Arc<OwnedResource>,
+    }
+
+    impl OwnedHandle {
+        fn new(label: &'static str, order: Arc<Mutex<Vec<&'static str>>>) -> Self {
+            Self {
+                _resource: Arc::new(OwnedResource { label, order }),
+            }
+        }
+    }
+
+    struct OwnedResource {
+        label: &'static str,
+        order: Arc<Mutex<Vec<&'static str>>>,
+    }
+
+    impl Drop for OwnedResource {
+        fn drop(&mut self) {
+            self.order.lock().unwrap().push(self.label);
+        }
+    }
+}

testing-framework/core/src/scenario/runtime/extensions.rs

@@ -84,6 +84,11 @@ impl RuntimeExtensions {
             .and_then(|value| value.downcast_ref::<T>())
             .cloned()
     }
+
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.values.is_empty()
+    }
 }
 
 #[derive(Default)]

testing-framework/deployers/local/src/deployer/orchestrator.rs

@@ -27,7 +27,7 @@ use tokio_retry::{
 use tracing::{debug, info, warn};
 
 use crate::{
-    env::{LocalDeployerEnv, Node, wait_local_http_readiness},
+    env::{LocalDeployerEnv, Node, wait_local_readiness},
     external::build_external_client,
     keep_tempdir_from_env,
     manual::ManualCluster,
@@ -440,7 +440,7 @@ async fn run_readiness_for_attempt<E: LocalDeployerEnv>(
         return Ok(nodes);
     }
 
-    match wait_local_http_readiness::<E>(&nodes, execution.readiness_requirement).await {
+    match wait_local_readiness::<E>(&nodes, execution.readiness_requirement).await {
         Ok(()) => {
             info!(attempt, "local nodes are ready");
             Ok(nodes)
@@ -519,7 +519,7 @@ async fn run_context_for<E: Application>(
     runtime_cleanup: Option<Box<dyn CleanupGuard>>,
     node_control: Option<Arc<dyn NodeControlHandle<E>>>,
 ) -> Result<RuntimeContext<E>, ProcessDeployerError> {
-    if node_clients.is_empty() {
+    if node_clients.is_empty() && runtime_extensions.is_empty() {
         return Err(ProcessDeployerError::RuntimePreflight);
     }
 

testing-framework/deployers/local/src/env/helpers.rs

@@ -10,7 +10,7 @@ use testing_framework_core::{
 };
 
 use crate::{
-    binary::{BinaryProvider, BinaryProviderRef, EnvBinaryProvider},
+    binary::{BinaryProvider, BinaryProviderRef, EnvBinaryProvider, PathBinaryProvider},
     env::LocalBuildContext,
     process::{LaunchSpec, NodeEndpointPort, NodeEndpoints, ProcessSpawnError},
 };
@@ -63,8 +63,8 @@ impl LocalNodePorts {
     }
 }
 
-#[derive(Clone, Debug)]
 /// Peer node view used while constructing local configs.
+#[derive(Clone, Debug)]
 pub struct LocalPeerNode {
     index: usize,
     network_port: u16,
@@ -96,6 +96,16 @@ impl LocalPeerNode {
     }
 }
 
+/// How a rendered local config file path is passed to the child process.
+#[derive(Clone, Default)]
+pub enum LocalConfigArgMode {
+    /// Pass the config file as a flag pair, for example `--config config.yaml`.
+    #[default]
+    Flag,
+    /// Pass the config file path as a positional argument.
+    Positional,
+}
+
 /// Standard local process description for one node binary plus one config file.
 #[derive(Clone)]
 pub struct LocalProcessSpec {
@@ -105,6 +115,8 @@ pub struct LocalProcessSpec {
     pub config_file_name: String,
     /// CLI flag used to point the process at `config_file_name`.
     pub config_arg: String,
+    /// Controls whether `config_arg` is emitted or the file is positional.
+    pub config_arg_mode: LocalConfigArgMode,
     /// Extra CLI arguments passed after the config flag.
     pub extra_args: Vec<String>,
     /// Extra environment variables for the child process.
@@ -119,16 +131,32 @@ impl LocalProcessSpec {
             binary: Arc::new(EnvBinaryProvider::new(binary_env_var)),
             config_file_name: "config.yaml".to_owned(),
             config_arg: "--config".to_owned(),
+            config_arg_mode: LocalConfigArgMode::Flag,
             extra_args: Vec::new(),
             env: Vec::new(),
         }
     }
 
+    /// Sets an explicit binary path for this launch.
+    #[must_use]
+    pub fn with_binary_path(self, path: impl Into<std::path::PathBuf>) -> Self {
+        self.with_binary_provider(PathBinaryProvider::new(path))
+    }
+
     /// Overrides the config file name and CLI flag used to pass it.
     #[must_use]
     pub fn with_config_file(mut self, file_name: &str, arg: &str) -> Self {
         self.config_file_name = file_name.to_owned();
         self.config_arg = arg.to_owned();
+        self.config_arg_mode = LocalConfigArgMode::Flag;
+        self
+    }
+
+    /// Overrides the config file name and passes it as a positional argument.
+    #[must_use]
+    pub fn with_positional_config_file(mut self, file_name: &str) -> Self {
+        self.config_file_name = file_name.to_owned();
+        self.config_arg_mode = LocalConfigArgMode::Positional;
         self
     }
 
@@ -404,7 +432,7 @@ pub(crate) fn rendered_config_launch_spec(
     spec: &LocalProcessSpec,
 ) -> Result<LaunchSpec, DynError> {
     let binary = spec.binary.resolve()?;
-    let mut args = vec![spec.config_arg.clone(), spec.config_file_name.clone()];
+    let mut args = config_file_args(spec);
     args.extend(spec.extra_args.iter().cloned());
 
     Ok(LaunchSpec {
@@ -417,3 +445,39 @@ pub(crate) fn rendered_config_launch_spec(
         env: spec.env.clone(),
     })
 }
+fn config_file_args(spec: &LocalProcessSpec) -> Vec<String> {
+    match spec.config_arg_mode {
+        LocalConfigArgMode::Flag => vec![spec.config_arg.clone(), spec.config_file_name.clone()],
+        LocalConfigArgMode::Positional => vec![spec.config_file_name.clone()],
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{LocalProcessSpec, text_config_launch_spec};
+
+    #[test]
+    fn launch_spec_uses_flag_config_by_default() {
+        let temp = tempfile::tempdir().expect("temp dir");
+        let binary = temp.path().join("app");
+        std::fs::write(&binary, b"binary").expect("test binary");
+        let spec = LocalProcessSpec::new("APP_BIN").with_binary_path(binary);
+        let launch = text_config_launch_spec("config", &spec).expect("launch spec");
+
+        assert_eq!(launch.args, ["--config", "config.yaml"]);
+    }
+
+    #[test]
+    fn launch_spec_can_use_positional_config_path() {
+        let temp = tempfile::tempdir().expect("temp dir");
+        let binary = temp.path().join("app");
+        std::fs::write(&binary, b"binary").expect("test binary");
+        let spec = LocalProcessSpec::new("APP_BIN")
+            .with_binary_path(binary)
+            .with_positional_config_file("app.json")
+            .with_args(["--port".to_owned(), "8080".to_owned()]);
+        let launch = text_config_launch_spec("config", &spec).expect("launch spec");
+
+        assert_eq!(launch.args, ["app.json", "--port", "8080"]);
+    }
+}

testing-framework/deployers/local/src/env/mod.rs

@@ -1,13 +1,14 @@
 use std::{
     collections::HashMap,
-    net::{Ipv4Addr, SocketAddr},
+    net::{Ipv4Addr, SocketAddr, TcpStream},
     path::{Path, PathBuf},
+    time::{Duration, Instant},
 };
 
 use async_trait::async_trait;
 use testing_framework_core::scenario::{
     Application, DynError, HttpReadinessRequirement, ReadinessError, StartNodeOptions,
-    wait_for_http_ports_with_requirement,
+    wait_for_http_ports_with_requirement_and_timeout,
 };
 
 use crate::{
@@ -20,14 +21,27 @@ mod helpers;
 mod tests;
 
 pub use helpers::{
-    BuiltNodeConfig, LocalNodePorts, LocalPeerNode, LocalProcessSpec, NodeConfigEntry,
-    build_indexed_http_peers, build_indexed_node_configs, build_launch_spec_with_args,
-    build_local_cluster_node_config, build_local_peer_nodes, default_yaml_launch_spec,
-    discovered_node_access, preallocate_ports, reserve_local_node_ports,
+    BuiltNodeConfig, LocalConfigArgMode, LocalNodePorts, LocalPeerNode, LocalProcessSpec,
+    NodeConfigEntry, build_indexed_http_peers, build_indexed_node_configs,
+    build_launch_spec_with_args, build_local_cluster_node_config, build_local_peer_nodes,
+    default_yaml_launch_spec, discovered_node_access, preallocate_ports, reserve_local_node_ports,
     single_http_node_endpoints, text_config_launch_spec, text_node_config, yaml_config_launch_spec,
     yaml_node_config,
 };
 
+const TCP_READINESS_TIMEOUT: Duration = Duration::from_secs(60);
+const TCP_READINESS_POLL_INTERVAL: Duration = Duration::from_millis(200);
+const TCP_CONNECT_TIMEOUT: Duration = Duration::from_millis(100);
+
+/// Readiness probe shape used by the local process deployer.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum LocalReadinessProbe {
+    /// Probe `http://127.0.0.1:<api-port>/<path>` with GET.
+    HttpGet { path: &'static str },
+    /// Probe that the API port accepts TCP connections.
+    Tcp,
+}
+
 /// Context passed while building a local node config.
 pub struct LocalBuildContext<'a, E: Application> {
     /// Full deployment topology for the current scenario.
@@ -323,6 +337,13 @@ where
         <Self as Application>::node_readiness_path()
     }
 
+    /// Returns the readiness probe used for local process startup.
+    fn readiness_probe() -> LocalReadinessProbe {
+        LocalReadinessProbe::HttpGet {
+            path: Self::readiness_endpoint_path(),
+        }
+    }
+
     /// Waits for any additional cluster-specific stabilization after the HTTP
     /// readiness probe succeeds.
     async fn wait_readiness_stable(_nodes: &[Node<Self>]) -> Result<(), DynError> {
@@ -376,6 +397,13 @@ where
         <Self as Application>::node_readiness_path()
     }
 
+    /// Returns the readiness probe used for local process startup.
+    fn readiness_probe() -> LocalReadinessProbe {
+        LocalReadinessProbe::HttpGet {
+            path: Self::readiness_endpoint_path(),
+        }
+    }
+
     /// Waits for any additional cluster-specific stabilization after the HTTP
     /// readiness probe succeeds.
     async fn wait_readiness_stable(_nodes: &[Node<Self>]) -> Result<(), DynError> {
@@ -435,6 +463,10 @@ where
         T::readiness_endpoint_path()
     }
 
+    fn readiness_probe() -> LocalReadinessProbe {
+        T::readiness_probe()
+    }
+
     async fn wait_readiness_stable(nodes: &[Node<Self>]) -> Result<(), DynError> {
         T::wait_readiness_stable(nodes).await
     }
@@ -490,13 +522,9 @@ pub(crate) fn node_peer_port<E: LocalDeployerEnv>(node: &Node<E>) -> u16 {
     E::node_peer_port(node)
 }
 
-pub(crate) fn readiness_endpoint_path<E: LocalDeployerEnv>() -> &'static str {
-    E::readiness_endpoint_path()
-}
-
-/// Waits for local HTTP readiness across the provided nodes and then applies
+/// Waits for local readiness across the provided nodes and then applies
 /// any app-specific stabilization hook.
-pub async fn wait_local_http_readiness<E: LocalDeployerEnv>(
+pub async fn wait_local_readiness<E: LocalDeployerEnv>(
     nodes: &[Node<E>],
     requirement: HttpReadinessRequirement,
 ) -> Result<(), ReadinessError> {
@@ -505,13 +533,119 @@ pub async fn wait_local_http_readiness<E: LocalDeployerEnv>(
         .map(|node| node.endpoints().api.port())
         .collect();
 
-    wait_for_http_ports_with_requirement(&ports, E::readiness_endpoint_path(), requirement).await?;
+    wait_for_local_readiness_ports::<E>(&ports, requirement, None).await?;
 
     E::wait_readiness_stable(nodes)
         .await
         .map_err(|source| ReadinessError::ClusterStable { source })
 }
 
+/// Backward-compatible name for local readiness checks that used to be
+/// HTTP-only.
+pub async fn wait_local_http_readiness<E: LocalDeployerEnv>(
+    nodes: &[Node<E>],
+    requirement: HttpReadinessRequirement,
+) -> Result<(), ReadinessError> {
+    wait_local_readiness::<E>(nodes, requirement).await
+}
+
+pub(crate) async fn wait_for_local_readiness_ports<E: LocalDeployerEnv>(
+    ports: &[u16],
+    requirement: HttpReadinessRequirement,
+    timeout: Option<Duration>,
+) -> Result<(), ReadinessError> {
+    match E::readiness_probe() {
+        LocalReadinessProbe::HttpGet { path } => {
+            wait_for_http_ports_with_requirement_and_timeout(ports, path, requirement, timeout)
+                .await
+        }
+        LocalReadinessProbe::Tcp => {
+            wait_for_tcp_ports_with_requirement(ports, requirement, timeout).await
+        }
+    }
+}
+
+async fn wait_for_tcp_ports_with_requirement(
+    ports: &[u16],
+    requirement: HttpReadinessRequirement,
+    timeout: Option<Duration>,
+) -> Result<(), ReadinessError> {
+    if ports.is_empty() {
+        return Ok(());
+    }
+
+    let timeout = timeout.unwrap_or(TCP_READINESS_TIMEOUT);
+    let deadline = Instant::now() + timeout;
+
+    loop {
+        let statuses = collect_tcp_statuses(ports);
+        if tcp_requirement_satisfied(&statuses, requirement) {
+            return Ok(());
+        }
+
+        if Instant::now() >= deadline {
+            return Err(ReadinessError::ProbeTimeout {
+                message: format_tcp_timeout_message(&statuses, requirement),
+            });
+        }
+
+        tokio::time::sleep(TCP_READINESS_POLL_INTERVAL).await;
+    }
+}
+
+fn collect_tcp_statuses(ports: &[u16]) -> Vec<(u16, bool)> {
+    ports
+        .iter()
+        .map(|port| (*port, tcp_port_ready(*port)))
+        .collect()
+}
+
+fn tcp_port_ready(port: u16) -> bool {
+    let addr = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
+    TcpStream::connect_timeout(&addr, TCP_CONNECT_TIMEOUT).is_ok()
+}
+
+fn tcp_requirement_satisfied(
+    statuses: &[(u16, bool)],
+    requirement: HttpReadinessRequirement,
+) -> bool {
+    let ready = tcp_ready_count(statuses);
+
+    match requirement {
+        HttpReadinessRequirement::AllNodesReady => ready == statuses.len(),
+        HttpReadinessRequirement::AnyNodeReady => ready >= 1,
+        HttpReadinessRequirement::AtLeast(min_ready) => ready >= min_ready,
+    }
+}
+
+fn tcp_ready_count(statuses: &[(u16, bool)]) -> usize {
+    statuses.iter().filter(|(_, ready)| *ready).count()
+}
+
+fn format_tcp_timeout_message(
+    statuses: &[(u16, bool)],
+    requirement: HttpReadinessRequirement,
+) -> String {
+    let ready = tcp_ready_count(statuses);
+    let failing = statuses
+        .iter()
+        .filter_map(|(port, ok)| (!ok).then_some(port.to_string()))
+        .collect::<Vec<_>>()
+        .join(", ");
+
+    format!(
+        "timed out waiting for TCP readiness {:?}; ready={}, total={}, failing ports: {}",
+        requirement,
+        ready,
+        statuses.len(),
+        if failing.is_empty() {
+            "<none>"
+        } else {
+            &failing
+        }
+    )
+}
+
 /// Spawns a local process node from an already prepared config value.
 pub async fn spawn_node_from_config<E: LocalDeployerEnv>(
     label: String,

testing-framework/deployers/local/src/env/tests.rs

@@ -1,4 +1,8 @@
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::{
+    net::TcpListener,
+    sync::atomic::{AtomicUsize, Ordering},
+    time::Duration,
+};
 
 use testing_framework_core::{
     scenario::{Application, DynError, HttpReadinessRequirement},
@@ -22,6 +26,7 @@ impl DeploymentDescriptor for DummyTopology {
 }
 
 struct DummyEnv;
+struct TcpEnv;
 
 #[async_trait::async_trait]
 impl Application for DummyEnv {
@@ -30,6 +35,13 @@ impl Application for DummyEnv {
     type NodeConfig = DummyConfig;
 }
 
+#[async_trait::async_trait]
+impl Application for TcpEnv {
+    type Deployment = DummyTopology;
+    type NodeClient = ();
+    type NodeConfig = DummyConfig;
+}
+
 #[async_trait::async_trait]
 impl LocalDeployerEnv for DummyEnv {
     fn build_node_config(
@@ -69,6 +81,13 @@ impl LocalDeployerEnv for DummyEnv {
     }
 }
 
+#[async_trait::async_trait]
+impl LocalDeployerEnv for TcpEnv {
+    fn readiness_probe() -> LocalReadinessProbe {
+        LocalReadinessProbe::Tcp
+    }
+}
+
 fn build_dummy_node() -> Result<BuiltNodeConfig<DummyConfig>, DynError> {
     unreachable!("not used in this test")
 }
@@ -99,8 +118,22 @@ async fn dummy_wait_stable() -> Result<(), DynError> {
 async fn empty_cluster_still_runs_stability_hook() {
     STABLE_CALLS.store(0, Ordering::SeqCst);
     let nodes: Vec<Node<DummyEnv>> = Vec::new();
-    wait_local_http_readiness::<DummyEnv>(&nodes, HttpReadinessRequirement::AllNodesReady)
+    wait_local_readiness::<DummyEnv>(&nodes, HttpReadinessRequirement::AllNodesReady)
         .await
         .expect("empty cluster should be considered ready");
     assert_eq!(STABLE_CALLS.load(Ordering::SeqCst), 1);
 }
+
+#[tokio::test]
+async fn tcp_readiness_probe_accepts_bound_port() {
+    let listener = TcpListener::bind("127.0.0.1:0").expect("bind tcp listener");
+    let port = listener.local_addr().expect("listener addr").port();
+
+    wait_for_local_readiness_ports::<TcpEnv>(
+        &[port],
+        HttpReadinessRequirement::AllNodesReady,
+        Some(Duration::from_secs(1)),
+    )
+    .await
+    .expect("bound TCP port should be ready");
+}

testing-framework/deployers/local/src/lib.rs

@@ -14,12 +14,12 @@ pub use binary::{
 };
 pub use deployer::{ProcessDeployer, ProcessDeployerError};
 pub use env::{
-    BuiltNodeConfig, LocalBinaryApp, LocalBuildContext, LocalDeployerEnv, LocalNodePorts,
-    LocalPeerNode, LocalProcessSpec, NodeConfigEntry, build_indexed_http_peers,
-    build_indexed_node_configs, build_local_cluster_node_config, build_local_peer_nodes,
-    default_yaml_launch_spec, discovered_node_access, preallocate_ports, reserve_local_node_ports,
-    single_http_node_endpoints, text_config_launch_spec, text_node_config, yaml_config_launch_spec,
-    yaml_node_config,
+    BuiltNodeConfig, LocalBinaryApp, LocalBuildContext, LocalConfigArgMode, LocalDeployerEnv,
+    LocalNodePorts, LocalPeerNode, LocalProcessSpec, LocalReadinessProbe, NodeConfigEntry,
+    build_indexed_http_peers, build_indexed_node_configs, build_local_cluster_node_config,
+    build_local_peer_nodes, default_yaml_launch_spec, discovered_node_access, preallocate_ports,
+    reserve_local_node_ports, single_http_node_endpoints, text_config_launch_spec,
+    text_node_config, yaml_config_launch_spec, yaml_node_config,
 };
 pub use manual::{ManualCluster, ManualClusterError};
 pub use node_control::{NodeManager, NodeManagerError, NodeManagerSeed};

testing-framework/deployers/local/src/node_control/mod.rs

@@ -4,8 +4,8 @@ use std::{
 };
 
 use testing_framework_core::scenario::{
-    Application, DynError, NodeClients, NodeControlHandle, NodeRuntimeOptions, ReadinessError,
-    StartNodeOptions, StartedNode, wait_for_http_ports, wait_for_http_ports_with_timeout,
+    Application, DynError, HttpReadinessRequirement, NodeClients, NodeControlHandle,
+    NodeRuntimeOptions, ReadinessError, StartNodeOptions, StartedNode,
 };
 use thiserror::Error;
 
@@ -13,7 +13,7 @@ use crate::{
     env::{
         LocalDeployerEnv, Node, build_initial_node_configs, build_launch_spec_with_args,
         build_node_from_template, initial_persist_dir, initial_snapshot_dir, node_peer_port,
-        readiness_endpoint_path, spawn_node_from_config,
+        spawn_node_from_config, wait_for_local_readiness_ports,
     },
     process::ProcessSpawnError,
 };
@@ -214,15 +214,16 @@ impl<E: LocalDeployerEnv> NodeManager<E> {
             return Ok(());
         }
 
-        wait_for_http_ports(&ports, readiness_endpoint_path::<E>()).await
+        wait_for_local_readiness_ports::<E>(&ports, HttpReadinessRequirement::AllNodesReady, None)
+            .await
     }
 
     pub async fn wait_node_ready(&self, name: &str) -> Result<(), NodeManagerError> {
         let target = self.readiness_target(name)?;
 
-        wait_for_http_ports_with_timeout(
+        wait_for_local_readiness_ports::<E>(
             &[target.port],
-            readiness_endpoint_path::<E>(),
+            HttpReadinessRequirement::AllNodesReady,
             target.runtime.start_timeout,
         )
         .await