feat(tf): integrate heterogeneous local app stacks

Cargo.lock

@@ -1700,10 +1700,12 @@ name = "kvstore-examples"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "async-trait",
  "kvstore-node",
  "kvstore-runtime-ext",
  "kvstore-runtime-workloads",
  "serde",
+ "testing-framework-app",
  "testing-framework-core",
  "testing-framework-runner-compose",
  "testing-framework-runner-k8s",
@@ -1735,6 +1737,7 @@ dependencies = [
  "async-trait",
  "kvstore-node",
  "serde",
+ "testing-framework-app",
  "testing-framework-core",
  "testing-framework-runner-compose",
  "testing-framework-runner-k8s",
@@ -1749,6 +1752,7 @@ dependencies = [
  "kvstore-node",
  "kvstore-runtime-ext",
  "serde",
+ "testing-framework-app",
  "testing-framework-core",
  "tokio",
  "tracing",
@@ -1920,6 +1924,23 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "multi-app-examples"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "kvstore-runtime-ext",
+ "openraft-kv-runtime-ext",
+ "openraft-kv-runtime-workloads",
+ "serde",
+ "testing-framework-app",
+ "testing-framework-core",
+ "tokio",
+ "tracing",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "native-tls"
 version = "0.2.14"
@@ -2085,9 +2106,11 @@ name = "openraft-kv-examples"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "async-trait",
  "openraft-kv-node",
  "openraft-kv-runtime-ext",
  "openraft-kv-runtime-workloads",
+ "testing-framework-app",
  "testing-framework-core",
  "testing-framework-runner-k8s",
  "tokio",
@@ -2120,6 +2143,7 @@ dependencies = [
  "async-trait",
  "openraft-kv-node",
  "reqwest",
+ "testing-framework-app",
  "testing-framework-core",
  "testing-framework-runner-compose",
  "testing-framework-runner-k8s",
@@ -2134,6 +2158,7 @@ dependencies = [
  "async-trait",
  "openraft-kv-node",
  "openraft-kv-runtime-ext",
+ "testing-framework-app",
  "testing-framework-core",
  "thiserror 2.0.18",
  "tokio",

Cargo.toml

@@ -12,6 +12,7 @@ members = [
   "examples/metrics_counter/metrics-counter-node",
   "examples/metrics_counter/testing/integration",
   "examples/metrics_counter/testing/workloads",
+  "examples/multi_app/examples",
   "examples/nats/examples",
   "examples/nats/testing/integration",
   "examples/nats/testing/workloads",

examples/README.md

@@ -0,0 +1,25 @@
+# Examples
+
+Use the app layer first when the test is about an application or composed
+system interface.
+
+Canonical app-layer examples:
+
+- `kvstore_app_host_convergence`: one local app cluster exposed through
+  `AppHost`
+- `openraft_kv_app_host_smoke`: one richer local app cluster exposed through
+  `AppHost`
+- `multi_app_typed_stack_smoke`: composed app stack exposing typed child and
+  stack handles
+
+The older direct `ScenarioBuilder<AppEnv>` examples are still useful for
+backend-specific coverage:
+
+- local uniform cluster behavior
+- compose runner behavior
+- k8s runner behavior
+- manual cluster control
+
+Do not use those older examples as the pattern for new composed systems. New
+multi-app tests should define an `AppDeployment`, expose typed handles, and run
+through `AppHost::scenario().with_app(...)`.

examples/kvstore/examples/Cargo.toml

@@ -4,6 +4,10 @@ license.workspace = true
 name              = "kvstore-examples"
 version.workspace = true
 
+[[bin]]
+name = "kvstore_app_host_convergence"
+path = "src/bin/app_host_convergence.rs"
+
 [[bin]]
 name = "kvstore_basic_convergence"
 path = "src/bin/basic_convergence.rs"
@@ -21,9 +25,11 @@ name = "kvstore_k8s_manual_convergence"
 path = "src/bin/k8s_manual_convergence.rs"
 
 [dependencies]
+async-trait                      = { workspace = true }
 kvstore-node                     = { path = "../kvstore-node" }
 kvstore-runtime-ext              = { path = "../testing/integration" }
 kvstore-runtime-workloads        = { path = "../testing/workloads" }
+testing-framework-app            = { workspace = true }
 testing-framework-core           = { workspace = true }
 testing-framework-runner-compose = { workspace = true }
 testing-framework-runner-k8s     = { workspace = true }

examples/kvstore/examples/src/bin/app_host_convergence.rs

@@ -0,0 +1,117 @@
+use std::time::Duration;
+
+use async_trait::async_trait;
+use kvstore_runtime_ext::{KvEnv, KvLocalApp};
+use serde::{Deserialize, Serialize};
+use testing_framework_app::{
+    AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt, AppScenarioBuilderExt,
+    LocalAppCluster,
+};
+use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload};
+use tracing::info;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .init();
+
+    let mut scenario = AppHost::scenario()
+        .with_app(KvLocalApp::nodes(3))
+        .with_run_duration(Duration::from_secs(5))
+        .with_workload(KvAppHostConvergence::new(3))
+        .build()?;
+
+    let deployer = AppHostLocalDeployer::default();
+    let runner = deployer.deploy(&scenario).await?;
+    runner.run(&mut scenario).await?;
+
+    Ok(())
+}
+
+#[derive(Clone)]
+struct KvAppHostConvergence {
+    expected_nodes: usize,
+}
+
+impl KvAppHostConvergence {
+    const fn new(expected_nodes: usize) -> Self {
+        Self { expected_nodes }
+    }
+}
+
+#[async_trait]
+impl Workload<AppHostEnv> for KvAppHostConvergence {
+    fn name(&self) -> &str {
+        "kv_app_host_convergence"
+    }
+
+    async fn start(&self, ctx: &RunContext<AppHostEnv>) -> Result<(), DynError> {
+        let cluster = ctx.require_app::<LocalAppCluster<KvEnv>>()?;
+
+        ensure_cluster_shape(&cluster, self.expected_nodes)?;
+        put_value(&cluster, "before-restart").await?;
+        cluster.restart_node("node-0").await?;
+        cluster.wait_node_ready("node-0").await?;
+        put_value(&cluster, "after-restart").await?;
+
+        info!(
+            nodes = self.expected_nodes,
+            "kv app host cluster exposes app-local node control and clients"
+        );
+
+        Ok(())
+    }
+}
+
+fn ensure_cluster_shape(
+    cluster: &LocalAppCluster<KvEnv>,
+    expected_nodes: usize,
+) -> Result<(), DynError> {
+    if cluster.node_count() != expected_nodes || cluster.clients().len() != expected_nodes {
+        return Err(format!("kv app host expected {expected_nodes} nodes").into());
+    }
+
+    if cluster.node_client("node-0").is_none() {
+        return Err("kv app host cannot access node-0 client".into());
+    }
+
+    if cluster.node_pid("node-0").is_none() {
+        return Err("kv app host cannot access node-0 process id".into());
+    }
+
+    Ok(())
+}
+
+async fn put_value(cluster: &LocalAppCluster<KvEnv>, value: &str) -> Result<(), DynError> {
+    let Some(client) = cluster.first_client() else {
+        return Err("kv app host has no clients".into());
+    };
+
+    let response: KvPutResponse = client
+        .put(
+            "/kv/app-host-smoke",
+            &KvPutRequest {
+                value: value.to_owned(),
+                expected_version: None,
+            },
+        )
+        .await?;
+
+    if !response.applied {
+        return Err(format!("kv app host write was rejected for value {value}").into());
+    }
+
+    Ok(())
+}
+
+#[derive(Serialize)]
+struct KvPutRequest {
+    value: String,
+    expected_version: Option<u64>,
+}
+
+#[derive(Deserialize)]
+struct KvPutResponse {
+    applied: bool,
+}

examples/kvstore/examples/src/bin/basic_convergence.rs

@@ -2,7 +2,8 @@ use std::time::Duration;
 
 use kvstore_runtime_ext::KvLocalDeployer;
 use kvstore_runtime_workloads::{
-    KvBuilderExt, KvConverges, KvScenarioBuilder, KvTopology, KvWriteWorkload,
+    KvBuilderExt, KvClusterAccessible, KvConverges, KvExistingClusterApp, KvScenarioBuilder,
+    KvWriteWorkload,
 };
 use testing_framework_core::scenario::Deployer;
 
@@ -12,8 +13,9 @@ async fn main() -> anyhow::Result<()> {
         .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
         .init();
 
-    let mut scenario = KvScenarioBuilder::deployment_with(|_| KvTopology::new(3))
+    let mut scenario = KvScenarioBuilder::with_existing_kvstore_app(KvExistingClusterApp::nodes(3))
         .with_run_duration(Duration::from_secs(30))
+        .with_workload(KvClusterAccessible::new(3))
         .with_workload(
             KvWriteWorkload::new()
                 .operations(300)

examples/kvstore/testing/integration/Cargo.toml

@@ -5,6 +5,7 @@ name              = "kvstore-runtime-ext"
 version.workspace = true
 
 [dependencies]
+testing-framework-app            = { workspace = true }
 testing-framework-core           = { workspace = true }
 testing-framework-runner-compose = { workspace = true }
 testing-framework-runner-k8s     = { workspace = true }

examples/kvstore/testing/integration/src/app.rs

@@ -3,9 +3,13 @@ use std::io::Error;
 use async_trait::async_trait;
 use kvstore_node::KvHttpClient;
 use serde::{Deserialize, Serialize};
-use testing_framework_core::scenario::{
-    Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError,
-    NodeAccess, serialize_cluster_yaml_config,
+use testing_framework_app::{AppDeployment, AppHostEnv, DeployContext, LocalAppCluster};
+use testing_framework_core::{
+    scenario::{
+        Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError,
+        NodeAccess, NodeClients, serialize_cluster_yaml_config,
+    },
+    topology::DeploymentDescriptor,
 };
 
 pub type KvTopology = testing_framework_core::topology::ClusterTopology;
@@ -40,6 +44,102 @@ impl Application for KvEnv {
     }
 }
 
+#[derive(Clone)]
+pub struct KvStoreCluster {
+    deployment: KvTopology,
+    node_clients: NodeClients<KvEnv>,
+}
+
+impl KvStoreCluster {
+    #[must_use]
+    pub const fn new(deployment: KvTopology, node_clients: NodeClients<KvEnv>) -> Self {
+        Self {
+            deployment,
+            node_clients,
+        }
+    }
+
+    #[must_use]
+    pub fn topology(&self) -> &KvTopology {
+        &self.deployment
+    }
+
+    #[must_use]
+    pub fn node_count(&self) -> usize {
+        self.deployment.node_count()
+    }
+
+    #[must_use]
+    pub fn clients(&self) -> Vec<KvHttpClient> {
+        self.node_clients.snapshot()
+    }
+
+    #[must_use]
+    pub fn first_client(&self) -> Option<KvHttpClient> {
+        self.node_clients
+            .with_clients(|clients| clients.first().cloned())
+    }
+}
+
+#[derive(Clone)]
+pub struct KvExistingClusterApp {
+    topology: KvTopology,
+}
+
+impl KvExistingClusterApp {
+    #[must_use]
+    pub fn nodes(nodes: usize) -> Self {
+        Self {
+            topology: KvTopology::new(nodes),
+        }
+    }
+
+    #[must_use]
+    pub fn topology(&self) -> KvTopology {
+        self.topology.clone()
+    }
+}
+
+#[async_trait]
+impl AppDeployment<KvEnv> for KvExistingClusterApp {
+    type Handle = KvStoreCluster;
+
+    async fn deploy(self, ctx: &mut DeployContext<KvEnv>) -> Result<Self::Handle, DynError> {
+        Ok(KvStoreCluster::new(
+            ctx.deployment().clone(),
+            ctx.node_clients().clone(),
+        ))
+    }
+}
+
+#[derive(Clone)]
+pub struct KvLocalApp {
+    deployment: KvTopology,
+}
+
+impl KvLocalApp {
+    #[must_use]
+    pub fn nodes(nodes: usize) -> Self {
+        Self {
+            deployment: KvTopology::new(nodes),
+        }
+    }
+
+    #[must_use]
+    pub fn deployment(&self) -> KvTopology {
+        self.deployment.clone()
+    }
+}
+
+#[async_trait]
+impl AppDeployment<AppHostEnv> for KvLocalApp {
+    type Handle = LocalAppCluster<KvEnv>;
+
+    async fn deploy(self, ctx: &mut DeployContext<AppHostEnv>) -> Result<Self::Handle, DynError> {
+        ctx.deploy_local_cluster::<KvEnv>(self.deployment).await
+    }
+}
+
 impl ClusterNodeConfigApplication for KvEnv {
     type ConfigError = Error;
 

examples/kvstore/testing/integration/src/local_env.rs

@@ -1,8 +1,9 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, path::PathBuf, sync::Arc};
 
 use testing_framework_core::scenario::{DynError, StartNodeOptions};
 use testing_framework_runner_local::{
-    LocalBinaryApp, LocalNodePorts, LocalPeerNode, LocalProcessSpec,
+    BinaryProviderRef, BuildBinaryProvider, BuildCommand, EnvBinaryProvider,
+    FallbackBinaryProvider, LocalBinaryApp, LocalNodePorts, LocalPeerNode, LocalProcessSpec,
     build_local_cluster_node_config, yaml_node_config,
 };
 
@@ -28,7 +29,9 @@ impl LocalBinaryApp for KvEnv {
     }
 
     fn local_process_spec() -> LocalProcessSpec {
-        LocalProcessSpec::new("KVSTORE_NODE_BIN").with_rust_log("kvstore_node=info")
+        LocalProcessSpec::new("KVSTORE_NODE_BIN")
+            .with_binary_provider(kvstore_binary_provider())
+            .with_rust_log("kvstore_node=info")
     }
 
     fn render_local_config(config: &KvNodeConfig) -> Result<Vec<u8>, DynError> {
@@ -39,3 +42,31 @@ impl LocalBinaryApp for KvEnv {
         config.http_port
     }
 }
+
+fn kvstore_binary_provider() -> FallbackBinaryProvider {
+    let workspace = workspace_root();
+    let providers: [BinaryProviderRef; 2] = [
+        Arc::new(EnvBinaryProvider::new("KVSTORE_NODE_BIN")),
+        Arc::new(BuildBinaryProvider {
+            command: BuildCommand::new("cargo").with_args([
+                "build",
+                "-p",
+                "kvstore-node",
+                "--bin",
+                "kvstore-node",
+            ]),
+            output_path: PathBuf::from(format!(
+                "target/debug/kvstore-node{}",
+                std::env::consts::EXE_SUFFIX
+            )),
+            working_dir: Some(workspace),
+            lock_dir: None,
+        }),
+    ];
+
+    FallbackBinaryProvider::new(providers)
+}
+
+fn workspace_root() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../..")
+}

examples/kvstore/testing/integration/src/scenario.rs

@@ -1,15 +1,22 @@
+use testing_framework_app::AppScenarioBuilderExt;
 use testing_framework_core::scenario::ScenarioBuilder;
 
-use crate::{KvEnv, KvTopology};
+use crate::{KvEnv, KvExistingClusterApp, KvTopology};
 
 pub type KvScenarioBuilder = ScenarioBuilder<KvEnv>;
 
 pub trait KvBuilderExt: Sized {
     fn deployment_with(f: impl FnOnce(KvTopology) -> KvTopology) -> Self;
+
+    fn with_existing_kvstore_app(app: KvExistingClusterApp) -> Self;
 }
 
 impl KvBuilderExt for KvScenarioBuilder {
     fn deployment_with(f: impl FnOnce(KvTopology) -> KvTopology) -> Self {
         KvScenarioBuilder::with_deployment(f(KvTopology::new(3)))
     }
+
+    fn with_existing_kvstore_app(app: KvExistingClusterApp) -> Self {
+        KvScenarioBuilder::with_deployment(app.topology()).with_app(app)
+    }
 }

examples/kvstore/testing/workloads/Cargo.toml

@@ -7,6 +7,7 @@ version.workspace = true
 [dependencies]
 kvstore-node           = { path = "../../kvstore-node" }
 kvstore-runtime-ext    = { path = "../integration" }
+testing-framework-app  = { workspace = true }
 testing-framework-core = { workspace = true }
 
 async-trait = { workspace = true }

examples/kvstore/testing/workloads/src/lib.rs

@@ -2,5 +2,7 @@ mod expectations;
 mod write;
 
 pub use expectations::KvConverges;
-pub use kvstore_runtime_ext::{KvBuilderExt, KvEnv, KvScenarioBuilder, KvTopology};
-pub use write::KvWriteWorkload;
+pub use kvstore_runtime_ext::{
+    KvBuilderExt, KvEnv, KvExistingClusterApp, KvScenarioBuilder, KvStoreCluster, KvTopology,
+};
+pub use write::{KvClusterAccessible, KvWriteWorkload};

examples/kvstore/testing/workloads/src/write.rs

@@ -1,8 +1,9 @@
 use std::time::Duration;
 
 use async_trait::async_trait;
-use kvstore_runtime_ext::KvEnv;
+use kvstore_runtime_ext::{KvEnv, KvStoreCluster};
 use serde::{Deserialize, Serialize};
+use testing_framework_app::AppRunContextExt;
 use testing_framework_core::scenario::{DynError, RunContext, Workload};
 use tracing::info;
 
@@ -68,6 +69,18 @@ impl Default for KvWriteWorkload {
     }
 }
 
+#[derive(Clone)]
+pub struct KvClusterAccessible {
+    expected_nodes: usize,
+}
+
+impl KvClusterAccessible {
+    #[must_use]
+    pub const fn new(expected_nodes: usize) -> Self {
+        Self { expected_nodes }
+    }
+}
+
 #[async_trait]
 impl Workload<KvEnv> for KvWriteWorkload {
     fn name(&self) -> &str {
@@ -126,6 +139,42 @@ impl Workload<KvEnv> for KvWriteWorkload {
     }
 }
 
+#[async_trait]
+impl Workload<KvEnv> for KvClusterAccessible {
+    fn name(&self) -> &str {
+        "kv_cluster_accessible"
+    }
+
+    async fn start(&self, ctx: &RunContext<KvEnv>) -> Result<(), DynError> {
+        let cluster = ctx.require_app::<KvStoreCluster>()?;
+        let client_count = cluster.clients().len();
+
+        if cluster.node_count() != self.expected_nodes {
+            return Err(format!(
+                "kv app topology has {} nodes, expected {}",
+                cluster.node_count(),
+                self.expected_nodes
+            )
+            .into());
+        }
+
+        if client_count != self.expected_nodes {
+            return Err(format!(
+                "kv app handle has {client_count} clients, expected {}",
+                self.expected_nodes
+            )
+            .into());
+        }
+
+        info!(
+            nodes = self.expected_nodes,
+            "kv app handle is accessible from workload"
+        );
+
+        Ok(())
+    }
+}
+
 fn compute_interval(rate_per_sec: usize) -> Option<Duration> {
     if rate_per_sec == 0 {
         return None;

examples/multi_app/README.md

@@ -0,0 +1,56 @@
+# Multi-App Examples
+
+This directory shows the canonical app-layer pattern for composed systems.
+
+Use this shape when a scenario needs several apps or resources that should feel
+like one system to the workload:
+
+```rust
+let mut scenario = AppHost::scenario()
+    .with_app(ExampleStackApp::new())
+    .with_workload(ExampleStackWorkload::new())
+    .build()?;
+```
+
+The stack app deploys child apps, exposes their typed handles, and returns a
+composed stack handle:
+
+```rust
+let store = StoreHandle::new(ctx.deploy_and_expose(self.store).await?);
+let consensus = ConsensusHandle::new(ctx.deploy_and_expose(self.consensus).await?);
+let wallet = WalletHandle::new(store.clone(), consensus.clone());
+let stack = ExampleStackHandle::new(store.clone(), consensus.clone(), wallet.clone());
+
+ctx.expose(store)?;
+ctx.expose(consensus)?;
+ctx.expose(wallet)?;
+ctx.expose(stack.clone())?;
+```
+
+Workloads then request concrete handles:
+
+```rust
+let stack = ctx.require_app::<ExampleStackHandle>()?;
+let wallet = ctx.require_app::<WalletHandle>()?;
+```
+
+Resource lifecycle comes from the TF adapter used by a child deployment:
+
+- `LocalProcessApp` manages one local binary process.
+- `LocalAppCluster` manages a uniform local cluster.
+
+Attached and external sources are not registered again at the app layer. The
+outer scenario resolves them through its existing source providers, and the app
+deployment receives the resulting node clients and control profile through
+`DeployContext`. When the active deployer provides node control or cluster
+readiness, the same handles are available through `DeployContext::node_control`
+and `DeployContext::cluster_wait`; the app layer does not create replacements.
+
+Managed adapters register cleanup during deployment. Cleanup runs in reverse
+acquisition order and does not depend on the last handle clone being dropped.
+An app-specific deployment only describes composition; it does not implement a
+second lifecycle interface.
+
+For a single uniform cluster, the core `ScenarioBuilder<AppEnv>` flow remains
+valid. For composed systems, prefer this app-layer shape instead of building a
+fake outer cluster or adding app-specific code to TF.

examples/multi_app/examples/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+edition.workspace = true
+license.workspace = true
+name              = "multi-app-examples"
+version.workspace = true
+
+[[bin]]
+name = "multi_app_typed_stack_smoke"
+path = "src/bin/typed_stack_smoke.rs"
+
+[dependencies]
+anyhow                        = "1.0"
+async-trait                   = { workspace = true }
+kvstore-runtime-ext           = { path = "../../kvstore/testing/integration" }
+openraft-kv-runtime-ext       = { path = "../../openraft_kv/testing/integration" }
+openraft-kv-runtime-workloads = { path = "../../openraft_kv/testing/workloads" }
+serde                         = { workspace = true }
+testing-framework-app         = { workspace = true }
+testing-framework-core        = { workspace = true }
+tokio                         = { workspace = true, features = ["full"] }
+tracing                       = { workspace = true }
+tracing-subscriber            = { version = "0.3", features = ["env-filter"] }

examples/multi_app/examples/src/bin/typed_stack_smoke.rs

@@ -0,0 +1,269 @@
+use std::time::Duration;
+
+use async_trait::async_trait;
+use kvstore_runtime_ext::{KvEnv, KvLocalApp};
+use openraft_kv_runtime_ext::{OpenRaftKvEnv, OpenRaftKvLocalApp};
+use openraft_kv_runtime_workloads::{
+    OpenRaftMembership, ensure_cluster_size, expected_kv, resolve_client_for_node, wait_for_leader,
+    wait_for_membership, wait_for_replication, write_batch,
+};
+use serde::{Deserialize, Serialize};
+use testing_framework_app::{
+    AppDeployment, AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt,
+    AppScenarioBuilderExt, DeployContext, LocalAppCluster,
+};
+use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload};
+use tracing::info;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .init();
+
+    let mut scenario = AppHost::scenario()
+        .with_app(ExampleStackApp::new())
+        .with_run_duration(Duration::from_secs(5))
+        .with_workload(ExampleStackWorkload::new())
+        .build()?;
+
+    let deployer = AppHostLocalDeployer::default();
+    let runner = deployer.deploy(&scenario).await?;
+    runner.run(&mut scenario).await?;
+
+    Ok(())
+}
+
+#[derive(Clone)]
+struct ExampleStackApp {
+    store: KvLocalApp,
+    consensus: OpenRaftKvLocalApp,
+}
+
+impl ExampleStackApp {
+    fn new() -> Self {
+        Self {
+            store: KvLocalApp::nodes(2),
+            consensus: OpenRaftKvLocalApp::nodes(3),
+        }
+    }
+}
+
+#[async_trait]
+impl AppDeployment<AppHostEnv> for ExampleStackApp {
+    type Handle = ExampleStackHandle;
+
+    async fn deploy(self, ctx: &mut DeployContext<AppHostEnv>) -> Result<Self::Handle, DynError> {
+        let store = StoreHandle::new(ctx.deploy_and_expose(self.store).await?);
+        let consensus = ConsensusHandle::new(ctx.deploy_and_expose(self.consensus).await?);
+        let wallet = WalletHandle::new(store.clone(), consensus.clone());
+        let stack = ExampleStackHandle::new(store.clone(), consensus.clone(), wallet.clone());
+
+        ctx.expose(store)?;
+        ctx.expose(consensus)?;
+        ctx.expose(wallet)?;
+        ctx.expose(stack.clone())?;
+
+        Ok(stack)
+    }
+}
+
+#[derive(Clone)]
+struct ExampleStackHandle {
+    store: StoreHandle,
+    consensus: ConsensusHandle,
+    wallet: WalletHandle,
+}
+
+impl ExampleStackHandle {
+    const fn new(store: StoreHandle, consensus: ConsensusHandle, wallet: WalletHandle) -> Self {
+        Self {
+            store,
+            consensus,
+            wallet,
+        }
+    }
+
+    const fn store(&self) -> &StoreHandle {
+        &self.store
+    }
+
+    const fn consensus(&self) -> &ConsensusHandle {
+        &self.consensus
+    }
+
+    const fn wallet(&self) -> &WalletHandle {
+        &self.wallet
+    }
+}
+
+#[derive(Clone)]
+struct StoreHandle {
+    cluster: LocalAppCluster<KvEnv>,
+}
+
+impl StoreHandle {
+    const fn new(cluster: LocalAppCluster<KvEnv>) -> Self {
+        Self { cluster }
+    }
+
+    fn node_count(&self) -> usize {
+        self.cluster.node_count()
+    }
+
+    async fn put(&self, key: &str, value: &str) -> Result<(), DynError> {
+        let Some(client) = self.cluster.first_client() else {
+            return Err("store handle has no kv clients".into());
+        };
+
+        let response: KvPutResponse = client
+            .put(
+                key,
+                &KvPutRequest {
+                    value: value.to_owned(),
+                    expected_version: None,
+                },
+            )
+            .await?;
+
+        if !response.applied {
+            return Err(format!("store write for {key} was rejected").into());
+        }
+
+        Ok(())
+    }
+}
+
+#[derive(Clone)]
+struct ConsensusHandle {
+    cluster: LocalAppCluster<OpenRaftKvEnv>,
+}
+
+impl ConsensusHandle {
+    const fn new(cluster: LocalAppCluster<OpenRaftKvEnv>) -> Self {
+        Self { cluster }
+    }
+
+    fn node_count(&self) -> usize {
+        self.cluster.node_count()
+    }
+
+    async fn bootstrap_and_write(&self, prefix: &str, writes: usize) -> Result<(), DynError> {
+        let clients = self.cluster.clients();
+
+        ensure_cluster_size(&clients, self.node_count())?;
+        clients[0].init_self().await?;
+
+        let leader_id = wait_for_leader(&clients, Duration::from_secs(30), None).await?;
+        let membership = OpenRaftMembership::discover(&clients).await?;
+        let leader = resolve_client_for_node(&clients, leader_id, Duration::from_secs(30)).await?;
+
+        for learner in membership.learner_targets(leader_id) {
+            leader
+                .add_learner(learner.node_id, &learner.public_addr)
+                .await?;
+        }
+
+        let voter_ids = membership.voter_ids();
+        leader.change_membership(voter_ids.iter().copied()).await?;
+        wait_for_membership(&clients, &voter_ids, Duration::from_secs(30)).await?;
+
+        write_batch(&leader, prefix, 0, writes).await?;
+        wait_for_replication(
+            &clients,
+            &expected_kv(prefix, writes),
+            Duration::from_secs(30),
+        )
+        .await?;
+
+        Ok(())
+    }
+}
+
+#[derive(Clone)]
+struct WalletHandle {
+    store: StoreHandle,
+    consensus: ConsensusHandle,
+}
+
+impl WalletHandle {
+    const fn new(store: StoreHandle, consensus: ConsensusHandle) -> Self {
+        Self { store, consensus }
+    }
+
+    async fn submit_transfer(&self, id: &str) -> Result<(), DynError> {
+        self.store.put("/kv/wallet-transfer", id).await?;
+        self.consensus
+            .bootstrap_and_write("wallet-transfer", 3)
+            .await?;
+
+        Ok(())
+    }
+}
+
+#[derive(Clone)]
+struct ExampleStackWorkload;
+
+impl ExampleStackWorkload {
+    const fn new() -> Self {
+        Self
+    }
+}
+
+#[async_trait]
+impl Workload<AppHostEnv> for ExampleStackWorkload {
+    fn name(&self) -> &str {
+        "multi_app_typed_stack_smoke"
+    }
+
+    async fn start(&self, ctx: &RunContext<AppHostEnv>) -> Result<(), DynError> {
+        let stack = ctx.require_app::<ExampleStackHandle>()?;
+        let store = ctx.require_app::<StoreHandle>()?;
+        let consensus = ctx.require_app::<ConsensusHandle>()?;
+        let wallet = ctx.require_app::<WalletHandle>()?;
+
+        ensure_stack_handles_match(&stack, &store, &consensus, &wallet)?;
+        store.put("/kv/typed-stack-smoke", "store-ready").await?;
+        wallet.submit_transfer("transfer-1").await?;
+
+        info!(
+            store_nodes = store.node_count(),
+            consensus_nodes = consensus.node_count(),
+            "typed multi-app stack handles are available to workloads"
+        );
+
+        Ok(())
+    }
+}
+
+fn ensure_stack_handles_match(
+    stack: &ExampleStackHandle,
+    store: &StoreHandle,
+    consensus: &ConsensusHandle,
+    wallet: &WalletHandle,
+) -> Result<(), DynError> {
+    if stack.store().node_count() != store.node_count() {
+        return Err("stack store handle does not match exposed store handle".into());
+    }
+
+    if stack.consensus().node_count() != consensus.node_count() {
+        return Err("stack consensus handle does not match exposed consensus handle".into());
+    }
+
+    if stack.wallet().store.node_count() != wallet.store.node_count() {
+        return Err("stack wallet handle does not match exposed wallet handle".into());
+    }
+
+    Ok(())
+}
+
+#[derive(Serialize)]
+struct KvPutRequest {
+    value: String,
+    expected_version: Option<u64>,
+}
+
+#[derive(Deserialize)]
+struct KvPutResponse {
+    applied: bool,
+}

examples/openraft_kv/examples/Cargo.toml

@@ -4,6 +4,10 @@ license.workspace = true
 name              = "openraft-kv-examples"
 version.workspace = true
 
+[[bin]]
+name = "openraft_kv_app_host_smoke"
+path = "src/bin/app_host_smoke.rs"
+
 [[bin]]
 name = "openraft_kv_basic_failover"
 path = "src/bin/basic_failover.rs"
@@ -18,9 +22,11 @@ path = "src/bin/k8s_failover.rs"
 
 [dependencies]
 anyhow                        = "1.0"
+async-trait                   = { workspace = true }
 openraft-kv-node              = { path = "../openraft-kv-node" }
 openraft-kv-runtime-ext       = { path = "../testing/integration" }
 openraft-kv-runtime-workloads = { path = "../testing/workloads" }
+testing-framework-app         = { workspace = true }
 testing-framework-core        = { workspace = true }
 testing-framework-runner-k8s  = { workspace = true }
 tokio                         = { workspace = true, features = ["full"] }

examples/openraft_kv/examples/src/bin/app_host_smoke.rs

@@ -0,0 +1,118 @@
+use std::time::Duration;
+
+use async_trait::async_trait;
+use openraft_kv_runtime_ext::{OpenRaftKvEnv, OpenRaftKvLocalApp};
+use openraft_kv_runtime_workloads::{
+    OpenRaftMembership, ensure_cluster_size, expected_kv, resolve_client_for_node, wait_for_leader,
+    wait_for_membership, wait_for_replication, write_batch,
+};
+use testing_framework_app::{
+    AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt, AppScenarioBuilderExt,
+    LocalAppCluster,
+};
+use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload};
+use tracing::info;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .init();
+
+    let mut scenario = AppHost::scenario()
+        .with_app(OpenRaftKvLocalApp::nodes(3))
+        .with_run_duration(Duration::from_secs(5))
+        .with_workload(OpenRaftKvAppHostSmoke::new(3))
+        .build()?;
+
+    let deployer = AppHostLocalDeployer::default();
+    let runner = deployer.deploy(&scenario).await?;
+    runner.run(&mut scenario).await?;
+
+    Ok(())
+}
+
+#[derive(Clone)]
+struct OpenRaftKvAppHostSmoke {
+    expected_nodes: usize,
+    timeout: Duration,
+}
+
+impl OpenRaftKvAppHostSmoke {
+    const fn new(expected_nodes: usize) -> Self {
+        Self {
+            expected_nodes,
+            timeout: Duration::from_secs(30),
+        }
+    }
+}
+
+#[async_trait]
+impl Workload<AppHostEnv> for OpenRaftKvAppHostSmoke {
+    fn name(&self) -> &str {
+        "openraft_kv_app_host_smoke"
+    }
+
+    async fn start(&self, ctx: &RunContext<AppHostEnv>) -> Result<(), DynError> {
+        let cluster = ctx.require_app::<LocalAppCluster<OpenRaftKvEnv>>()?;
+
+        ensure_cluster_shape(&cluster, self.expected_nodes)?;
+        bootstrap_voter_cluster(&cluster, self.timeout).await?;
+
+        info!(
+            nodes = self.expected_nodes,
+            "openraft app host cluster exposes app-local clients and process access"
+        );
+
+        Ok(())
+    }
+}
+
+fn ensure_cluster_shape(
+    cluster: &LocalAppCluster<OpenRaftKvEnv>,
+    expected_nodes: usize,
+) -> Result<(), DynError> {
+    if cluster.node_count() != expected_nodes {
+        return Err(format!("openraft app host expected {expected_nodes} nodes").into());
+    }
+
+    if cluster.node_client("node-0").is_none() {
+        return Err("openraft app host cannot access node-0 client".into());
+    }
+
+    if cluster.node_pid("node-0").is_none() {
+        return Err("openraft app host cannot access node-0 process id".into());
+    }
+
+    ensure_cluster_size(&cluster.clients(), expected_nodes)?;
+
+    Ok(())
+}
+
+async fn bootstrap_voter_cluster(
+    cluster: &LocalAppCluster<OpenRaftKvEnv>,
+    timeout: Duration,
+) -> Result<(), DynError> {
+    let clients = cluster.clients();
+
+    clients[0].init_self().await?;
+
+    let initial_leader = wait_for_leader(&clients, timeout, None).await?;
+    let membership = OpenRaftMembership::discover(&clients).await?;
+    let leader = resolve_client_for_node(&clients, initial_leader, timeout).await?;
+
+    for learner in membership.learner_targets(initial_leader) {
+        leader
+            .add_learner(learner.node_id, &learner.public_addr)
+            .await?;
+    }
+
+    let voter_ids = membership.voter_ids();
+    leader.change_membership(voter_ids.iter().copied()).await?;
+    wait_for_membership(&clients, &voter_ids, timeout).await?;
+
+    write_batch(&leader, "app-host-raft-key", 0, 3).await?;
+    wait_for_replication(&clients, &expected_kv("app-host-raft-key", 3), timeout).await?;
+
+    Ok(())
+}

examples/openraft_kv/examples/src/lib.rs

@@ -1,7 +1,11 @@
 use std::time::Duration;
 
-use openraft_kv_runtime_ext::{OpenRaftKvBuilderExt, OpenRaftKvEnv, OpenRaftKvScenarioBuilder};
-use openraft_kv_runtime_workloads::{OpenRaftKvConverges, OpenRaftKvFailoverWorkload};
+use openraft_kv_runtime_ext::{
+    OpenRaftKvBuilderExt, OpenRaftKvEnv, OpenRaftKvExistingClusterApp, OpenRaftKvScenarioBuilder,
+};
+use openraft_kv_runtime_workloads::{
+    OpenRaftKvClusterAccessible, OpenRaftKvConverges, OpenRaftKvFailoverWorkload,
+};
 use testing_framework_core::scenario::{NodeControlCapability, Scenario};
 
 /// Number of writes issued before the leader restart.
@@ -19,23 +23,23 @@ pub fn build_failover_scenario(
     run_duration: Duration,
     workload_timeout: Duration,
 ) -> anyhow::Result<Scenario<OpenRaftKvEnv, NodeControlCapability>> {
-    Ok(
-        OpenRaftKvScenarioBuilder::deployment_with(|deployment| deployment)
-            .with_cluster_observer()
-            .enable_node_control()
-            .with_run_duration(run_duration)
-            .with_workload(
-                OpenRaftKvFailoverWorkload::new()
-                    .first_batch(INITIAL_WRITE_BATCH)
-                    .second_batch(SECOND_WRITE_BATCH)
-                    .timeout(workload_timeout)
-                    .key_prefix(RAFT_KEY_PREFIX),
-            )
-            .with_expectation(
-                OpenRaftKvConverges::new(TOTAL_WRITES)
-                    .timeout(run_duration)
-                    .key_prefix(RAFT_KEY_PREFIX),
-            )
-            .build()?,
+    Ok(OpenRaftKvScenarioBuilder::with_existing_openraft_kv_app(
+        OpenRaftKvExistingClusterApp::nodes(3),
     )
+    .enable_node_control()
+    .with_run_duration(run_duration)
+    .with_workload(OpenRaftKvClusterAccessible::new(3))
+    .with_workload(
+        OpenRaftKvFailoverWorkload::new()
+            .first_batch(INITIAL_WRITE_BATCH)
+            .second_batch(SECOND_WRITE_BATCH)
+            .timeout(workload_timeout)
+            .key_prefix(RAFT_KEY_PREFIX),
+    )
+    .with_expectation(
+        OpenRaftKvConverges::new(TOTAL_WRITES)
+            .timeout(run_duration)
+            .key_prefix(RAFT_KEY_PREFIX),
+    )
+    .build()?)
 }

examples/openraft_kv/testing/integration/Cargo.toml

@@ -8,6 +8,7 @@ version.workspace = true
 async-trait                      = { workspace = true }
 openraft-kv-node                 = { path = "../../openraft-kv-node" }
 reqwest                          = { workspace = true }
+testing-framework-app            = { workspace = true }
 testing-framework-core           = { workspace = true }
 testing-framework-runner-compose = { workspace = true }
 testing-framework-runner-k8s     = { workspace = true }

examples/openraft_kv/testing/integration/src/app.rs

@@ -1,9 +1,14 @@
 use std::io::Error;
 
+use async_trait::async_trait;
 use openraft_kv_node::{OpenRaftKvClient, OpenRaftKvNodeConfig};
-use testing_framework_core::scenario::{
-    Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError,
-    NodeAccess, serialize_cluster_yaml_config,
+use testing_framework_app::{AppDeployment, AppHostEnv, DeployContext, LocalAppCluster};
+use testing_framework_core::{
+    scenario::{
+        Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError,
+        NodeAccess, NodeClients, serialize_cluster_yaml_config,
+    },
+    topology::DeploymentDescriptor,
 };
 
 /// Three-node topology used by the OpenRaft example scenarios.
@@ -57,3 +62,116 @@ impl ClusterNodeConfigApplication for OpenRaftKvEnv {
         serialize_cluster_yaml_config(config).map_err(Error::other)
     }
 }
+
+/// Runtime handle exposed by the OpenRaft app deployment.
+#[derive(Clone)]
+pub struct OpenRaftKvCluster {
+    deployment: OpenRaftKvTopology,
+    node_clients: NodeClients<OpenRaftKvEnv>,
+}
+
+impl OpenRaftKvCluster {
+    #[must_use]
+    pub const fn new(
+        deployment: OpenRaftKvTopology,
+        node_clients: NodeClients<OpenRaftKvEnv>,
+    ) -> Self {
+        Self {
+            deployment,
+            node_clients,
+        }
+    }
+
+    #[must_use]
+    pub fn topology(&self) -> &OpenRaftKvTopology {
+        &self.deployment
+    }
+
+    #[must_use]
+    pub fn node_count(&self) -> usize {
+        self.deployment.node_count()
+    }
+
+    #[must_use]
+    pub fn clients(&self) -> Vec<OpenRaftKvClient> {
+        self.node_clients.snapshot()
+    }
+
+    pub async fn states(&self) -> Result<Vec<openraft_kv_node::OpenRaftKvState>, DynError> {
+        let clients = self.clients();
+        let mut states = Vec::with_capacity(clients.len());
+
+        for client in clients {
+            states.push(client.state().await?);
+        }
+
+        states.sort_by_key(|state| state.node_id);
+
+        Ok(states)
+    }
+}
+
+/// App preset for the OpenRaft key-value example.
+#[derive(Clone)]
+pub struct OpenRaftKvExistingClusterApp {
+    topology: OpenRaftKvTopology,
+}
+
+impl OpenRaftKvExistingClusterApp {
+    #[must_use]
+    pub fn nodes(nodes: usize) -> Self {
+        Self {
+            topology: OpenRaftKvTopology::new(nodes),
+        }
+    }
+
+    #[must_use]
+    pub fn topology(&self) -> OpenRaftKvTopology {
+        self.topology.clone()
+    }
+}
+
+#[async_trait]
+impl AppDeployment<OpenRaftKvEnv> for OpenRaftKvExistingClusterApp {
+    type Handle = OpenRaftKvCluster;
+
+    async fn deploy(
+        self,
+        ctx: &mut DeployContext<OpenRaftKvEnv>,
+    ) -> Result<Self::Handle, DynError> {
+        Ok(OpenRaftKvCluster::new(
+            ctx.deployment().clone(),
+            ctx.node_clients().clone(),
+        ))
+    }
+}
+
+/// Local app preset that starts its own OpenRaft cluster.
+#[derive(Clone)]
+pub struct OpenRaftKvLocalApp {
+    deployment: OpenRaftKvTopology,
+}
+
+impl OpenRaftKvLocalApp {
+    #[must_use]
+    pub fn nodes(nodes: usize) -> Self {
+        Self {
+            deployment: OpenRaftKvTopology::new(nodes),
+        }
+    }
+
+    #[must_use]
+    pub fn deployment(&self) -> OpenRaftKvTopology {
+        self.deployment.clone()
+    }
+}
+
+#[async_trait]
+impl AppDeployment<AppHostEnv> for OpenRaftKvLocalApp {
+    type Handle = LocalAppCluster<OpenRaftKvEnv>;
+
+    async fn deploy(self, ctx: &mut DeployContext<AppHostEnv>) -> Result<Self::Handle, DynError> {
+        ctx.deploy_local_cluster::<OpenRaftKvEnv>(self.deployment)
+            .await
+    }
+}

examples/openraft_kv/testing/integration/src/local_env.rs

@@ -1,4 +1,8 @@
-use std::collections::{BTreeMap, HashMap};
+use std::{
+    collections::{BTreeMap, HashMap},
+    path::PathBuf,
+    sync::Arc,
+};
 
 use openraft_kv_node::OpenRaftKvNodeConfig;
 use testing_framework_core::{
@@ -6,7 +10,8 @@ use testing_framework_core::{
     topology::DeploymentDescriptor,
 };
 use testing_framework_runner_local::{
-    BuiltNodeConfig, LocalDeployerEnv, LocalNodePorts, LocalProcessSpec, NodeConfigEntry,
+    BinaryProviderRef, BuildBinaryProvider, BuildCommand, BuiltNodeConfig, EnvBinaryProvider,
+    FallbackBinaryProvider, LocalDeployerEnv, LocalNodePorts, LocalProcessSpec, NodeConfigEntry,
     reserve_local_node_ports, yaml_node_config,
 };
 
@@ -80,7 +85,11 @@ impl LocalDeployerEnv for OpenRaftKvEnv {
     }
 
     fn local_process_spec() -> Option<LocalProcessSpec> {
-        Some(LocalProcessSpec::new("OPENRAFT_KV_NODE_BIN").with_rust_log("info"))
+        Some(
+            LocalProcessSpec::new("OPENRAFT_KV_NODE_BIN")
+                .with_binary_provider(openraft_binary_provider())
+                .with_rust_log("info"),
+        )
     }
 
     fn render_local_config(config: &OpenRaftKvNodeConfig) -> Result<Vec<u8>, DynError> {
@@ -92,6 +101,34 @@ impl LocalDeployerEnv for OpenRaftKvEnv {
     }
 }
 
+fn openraft_binary_provider() -> FallbackBinaryProvider {
+    let workspace = workspace_root();
+    let providers: [BinaryProviderRef; 2] = [
+        Arc::new(EnvBinaryProvider::new("OPENRAFT_KV_NODE_BIN")),
+        Arc::new(BuildBinaryProvider {
+            command: BuildCommand::new("cargo").with_args([
+                "build",
+                "-p",
+                "openraft-kv-node",
+                "--bin",
+                "openraft-kv-node",
+            ]),
+            output_path: PathBuf::from(format!(
+                "target/debug/openraft-kv-node{}",
+                std::env::consts::EXE_SUFFIX
+            )),
+            working_dir: Some(workspace),
+            lock_dir: None,
+        }),
+    ];
+
+    FallbackBinaryProvider::new(providers)
+}
+
+fn workspace_root() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../..")
+}
+
 fn local_node_config(
     index: usize,
     network_port: u16,

examples/openraft_kv/testing/integration/src/scenario.rs

@@ -1,7 +1,9 @@
+use testing_framework_app::AppScenarioBuilderExt;
 use testing_framework_core::scenario::{CoreBuilderExt, ScenarioBuilder};
 
 use crate::{
-    OpenRaftClusterObserver, OpenRaftKvEnv, OpenRaftKvTopology, openraft_cluster_source_provider,
+    OpenRaftClusterObserver, OpenRaftKvEnv, OpenRaftKvExistingClusterApp, OpenRaftKvTopology,
+    openraft_cluster_source_provider,
 };
 
 /// Scenario builder alias used by the OpenRaft example binaries.
@@ -15,6 +17,10 @@ pub trait OpenRaftKvBuilderExt: Sized {
 
     /// Attaches the default OpenRaft cluster observer to the scenario.
     fn with_cluster_observer(self) -> Self;
+
+    /// Starts from an OpenRaft app preset and exposes its cluster handle to
+    /// workloads.
+    fn with_existing_openraft_kv_app(app: OpenRaftKvExistingClusterApp) -> Self;
 }
 
 impl OpenRaftKvBuilderExt for OpenRaftKvScenarioBuilder {
@@ -29,4 +35,10 @@ impl OpenRaftKvBuilderExt for OpenRaftKvScenarioBuilder {
             OpenRaftClusterObserver::config(),
         )
     }
+
+    fn with_existing_openraft_kv_app(app: OpenRaftKvExistingClusterApp) -> Self {
+        OpenRaftKvScenarioBuilder::with_deployment(app.topology())
+            .with_app(app)
+            .with_cluster_observer()
+    }
 }

examples/openraft_kv/testing/workloads/Cargo.toml

@@ -9,6 +9,7 @@ anyhow                  = "1.0"
 async-trait             = { workspace = true }
 openraft-kv-node        = { path = "../../openraft-kv-node" }
 openraft-kv-runtime-ext = { path = "../integration" }
+testing-framework-app   = { workspace = true }
 testing-framework-core  = { workspace = true }
 thiserror               = "2.0"
 tokio                   = { workspace = true, features = ["full"] }

examples/openraft_kv/testing/workloads/src/handle_access.rs

@@ -0,0 +1,63 @@
+use async_trait::async_trait;
+use openraft_kv_runtime_ext::{OpenRaftKvCluster, OpenRaftKvEnv};
+use testing_framework_app::AppRunContextExt;
+use testing_framework_core::scenario::{DynError, RunContext, Workload};
+use tracing::info;
+
+#[derive(Clone)]
+pub struct OpenRaftKvClusterAccessible {
+    expected_nodes: usize,
+}
+
+impl OpenRaftKvClusterAccessible {
+    #[must_use]
+    pub const fn new(expected_nodes: usize) -> Self {
+        Self { expected_nodes }
+    }
+}
+
+#[async_trait]
+impl Workload<OpenRaftKvEnv> for OpenRaftKvClusterAccessible {
+    fn name(&self) -> &str {
+        "openraft_kv_cluster_accessible"
+    }
+
+    async fn start(&self, ctx: &RunContext<OpenRaftKvEnv>) -> Result<(), DynError> {
+        let cluster = ctx.require_app::<OpenRaftKvCluster>()?;
+        let states = cluster.states().await?;
+        let client_count = cluster.clients().len();
+
+        if cluster.node_count() != self.expected_nodes {
+            return Err(format!(
+                "openraft app topology has {} nodes, expected {}",
+                cluster.node_count(),
+                self.expected_nodes
+            )
+            .into());
+        }
+
+        if client_count != self.expected_nodes {
+            return Err(format!(
+                "openraft app handle has {client_count} clients, expected {}",
+                self.expected_nodes
+            )
+            .into());
+        }
+
+        if states.len() != self.expected_nodes {
+            return Err(format!(
+                "openraft app handle read {} node states, expected {}",
+                states.len(),
+                self.expected_nodes
+            )
+            .into());
+        }
+
+        info!(
+            nodes = self.expected_nodes,
+            "openraft app handle is accessible from workload"
+        );
+
+        Ok(())
+    }
+}

examples/openraft_kv/testing/workloads/src/lib.rs

@@ -1,11 +1,14 @@
 mod convergence;
 mod failover;
+mod handle_access;
 mod support;
 
 /// Replication expectation used by the OpenRaft example binaries.
 pub use convergence::OpenRaftKvConverges;
 /// Failover workload used by the OpenRaft example binaries.
 pub use failover::OpenRaftKvFailoverWorkload;
+/// Workload that proves app handles are available to OpenRaft workloads.
+pub use handle_access::OpenRaftKvClusterAccessible;
 /// Shared cluster helpers used by the OpenRaft workload and manual k8s example.
 pub use support::{
     FULL_VOTER_SET, OpenRaftClusterError, OpenRaftMembership, ensure_cluster_size, expected_kv,