From eeb0573798a098e8ddf748bee5481c09fe650dc5 Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 13:56:12 +0100
Subject: [PATCH 1/9] Route source access through semantic helpers

---
 .../core/src/scenario/definition.rs           | 10 ++++
 .../core/src/scenario/sources/model.rs        | 40 ++++++++++++++++
 .../compose/src/deployer/attach_provider.rs   | 28 +++++------
 .../compose/src/deployer/orchestrator.rs      | 45 +++++++-----------
 .../k8s/src/deployer/attach_provider.rs       | 20 +++-----
 .../k8s/src/deployer/orchestrator.rs          | 47 +++++++++----------
 6 files changed, 109 insertions(+), 81 deletions(-)

diff --git a/testing-framework/core/src/scenario/definition.rs b/testing-framework/core/src/scenario/definition.rs
index b16b6a7..15c7ea9 100644
--- a/testing-framework/core/src/scenario/definition.rs
+++ b/testing-framework/core/src/scenario/definition.rs
@@ -117,6 +117,16 @@ impl<E: Application, Caps> Scenario<E, Caps> {
         &self.sources
     }
 
+    #[must_use]
+    pub fn attached_source(&self) -> Option<&AttachSource> {
+        self.sources.attached_source()
+    }
+
+    #[must_use]
+    pub fn external_nodes(&self) -> &[ExternalNodeSource] {
+        self.sources.external_nodes()
+    }
+
     #[must_use]
     pub const fn source_orchestration_plan(&self) -> &SourceOrchestrationPlan {
         &self.source_orchestration_plan
diff --git a/testing-framework/core/src/scenario/sources/model.rs b/testing-framework/core/src/scenario/sources/model.rs
index 98c60ad..5b0ec2c 100644
--- a/testing-framework/core/src/scenario/sources/model.rs
+++ b/testing-framework/core/src/scenario/sources/model.rs
@@ -43,6 +43,38 @@ impl AttachSource {
             services,
         }
     }
+
+    #[must_use]
+    pub fn compose_project(&self) -> Option<&str> {
+        match self {
+            Self::Compose { project, .. } => project.as_deref(),
+            Self::K8s { .. } => None,
+        }
+    }
+
+    #[must_use]
+    pub fn compose_services(&self) -> Option<&[String]> {
+        match self {
+            Self::Compose { services, .. } => Some(services),
+            Self::K8s { .. } => None,
+        }
+    }
+
+    #[must_use]
+    pub fn k8s_namespace(&self) -> Option<&str> {
+        match self {
+            Self::K8s { namespace, .. } => namespace.as_deref(),
+            Self::Compose { .. } => None,
+        }
+    }
+
+    #[must_use]
+    pub fn k8s_label_selector(&self) -> Option<&str> {
+        match self {
+            Self::K8s { label_selector, .. } => Some(label_selector),
+            Self::Compose { .. } => None,
+        }
+    }
 }
 
 /// Static external node endpoint that should be included in the runtime
@@ -140,6 +172,14 @@ impl ScenarioSources {
         Self::ExternalOnly { external }
     }
 
+    #[must_use]
+    pub fn attached_source(&self) -> Option<&AttachSource> {
+        match self {
+            Self::Attached { attach, .. } => Some(attach),
+            Self::Managed { .. } | Self::ExternalOnly { .. } => None,
+        }
+    }
+
     #[must_use]
     pub fn external_nodes(&self) -> &[ExternalNodeSource] {
         match self {
diff --git a/testing-framework/deployers/compose/src/deployer/attach_provider.rs b/testing-framework/deployers/compose/src/deployer/attach_provider.rs
index 87358aa..06c8d61 100644
--- a/testing-framework/deployers/compose/src/deployer/attach_provider.rs
+++ b/testing-framework/deployers/compose/src/deployer/attach_provider.rs
@@ -87,14 +87,15 @@ fn to_discovery_error(source: DynError) -> AttachProviderError {
 fn compose_attach_request(
     source: &AttachSource,
 ) -> Result<ComposeAttachRequest<'_>, AttachProviderError> {
-    let AttachSource::Compose { project, services } = source else {
-        return Err(AttachProviderError::UnsupportedSource {
-            attach_source: source.clone(),
-        });
-    };
+    let services =
+        source
+            .compose_services()
+            .ok_or_else(|| AttachProviderError::UnsupportedSource {
+                attach_source: source.clone(),
+            })?;
 
-    let project = project
-        .as_deref()
+    let project = source
+        .compose_project()
         .ok_or_else(|| AttachProviderError::Discovery {
             source: ComposeAttachDiscoveryError::MissingProjectName.into(),
         })?;
@@ -173,13 +174,12 @@ impl<E: ComposeDeployEnv> ClusterWaitHandle<E> for ComposeAttachedClusterWait<E>
 }
 
 fn compose_wait_request(source: &AttachSource) -> Result<ComposeAttachRequest<'_>, DynError> {
-    let AttachSource::Compose { project, services } = source else {
-        return Err("compose cluster wait requires a compose attach source".into());
-    };
-
-    let project = project
-        .as_deref()
-        .ok_or(ComposeAttachDiscoveryError::MissingProjectName)?;
+    let project = source
+        .compose_project()
+        .ok_or_else(|| DynError::from("compose cluster wait requires a compose attach source"))?;
+    let services = source
+        .compose_services()
+        .ok_or_else(|| DynError::from("compose cluster wait requires a compose attach source"))?;
 
     Ok(ComposeAttachRequest { project, services })
 }
diff --git a/testing-framework/deployers/compose/src/deployer/orchestrator.rs b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
index ddd88f0..c4fdc4f 100644
--- a/testing-framework/deployers/compose/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
@@ -6,9 +6,9 @@ use testing_framework_core::{
         ApplicationExternalProvider, AttachSource, CleanupGuard, ClusterWaitHandle,
         DeploymentPolicy, FeedHandle, FeedRuntime, HttpReadinessRequirement, Metrics, NodeClients,
         NodeControlHandle, ObservabilityCapabilityProvider, ObservabilityInputs,
-        RequiresNodeControl, RunContext, Runner, Scenario, ScenarioSources,
-        SourceOrchestrationPlan, SourceProviders, StaticManagedProvider,
-        build_source_orchestration_plan, orchestrate_sources_with_providers,
+        RequiresNodeControl, RunContext, Runner, Scenario, SourceOrchestrationPlan,
+        SourceProviders, StaticManagedProvider, build_source_orchestration_plan,
+        orchestrate_sources_with_providers,
     },
     topology::DeploymentDescriptor,
 };
@@ -214,21 +214,15 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
             return Ok(None);
         }
 
-        let ScenarioSources::Attached { attach, .. } = scenario.sources() else {
-            return Err(ComposeRunnerError::InternalInvariant {
+        let attach = scenario
+            .attached_source()
+            .ok_or(ComposeRunnerError::InternalInvariant {
                 message: "attached node control requested outside attached source mode",
-            });
-        };
+            })?;
 
-        let AttachSource::Compose { project, .. } = attach else {
-            return Err(ComposeRunnerError::InternalInvariant {
-                message: "compose deployer requires compose attach source for node control",
-            });
-        };
-
-        let Some(project_name) = project
-            .as_ref()
-            .map(|value| value.trim())
+        let Some(project_name) = attach
+            .compose_project()
+            .map(str::trim)
             .filter(|value| !value.is_empty())
         else {
             return Err(ComposeRunnerError::InternalInvariant {
@@ -248,11 +242,11 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
     where
         Caps: Send + Sync,
     {
-        let ScenarioSources::Attached { attach, .. } = scenario.sources() else {
-            return Err(ComposeRunnerError::InternalInvariant {
+        let attach = scenario
+            .attached_source()
+            .ok_or(ComposeRunnerError::InternalInvariant {
                 message: "compose attached cluster wait requested outside attached source mode",
-            });
-        };
+            })?;
 
         Ok(Arc::new(ComposeAttachedClusterWait::<E>::new(
             compose_runner_host(),
@@ -378,13 +372,10 @@ where
     E: ComposeDeployEnv,
     Caps: Send + Sync,
 {
-    let project_name = match scenario.sources() {
-        ScenarioSources::Attached {
-            attach: AttachSource::Compose { project, .. },
-            ..
-        } => project.clone(),
-        _ => None,
-    };
+    let project_name = scenario
+        .attached_source()
+        .and_then(|attach| attach.compose_project())
+        .map(ToOwned::to_owned);
 
     ComposeDeploymentMetadata { project_name }
 }
diff --git a/testing-framework/deployers/k8s/src/deployer/attach_provider.rs b/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
index b10a24a..bae5692 100644
--- a/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
+++ b/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
@@ -91,11 +91,7 @@ fn to_discovery_error(source: DynError) -> AttachProviderError {
 }
 
 fn k8s_attach_request(source: &AttachSource) -> Result<K8sAttachRequest<'_>, AttachProviderError> {
-    let AttachSource::K8s {
-        namespace,
-        label_selector,
-    } = source
-    else {
+    let Some(label_selector) = source.k8s_label_selector() else {
         return Err(AttachProviderError::UnsupportedSource {
             attach_source: source.clone(),
         });
@@ -108,7 +104,7 @@ fn k8s_attach_request(source: &AttachSource) -> Result<K8sAttachRequest<'_>, Att
     }
 
     Ok(K8sAttachRequest {
-        namespace: namespace.as_deref().unwrap_or("default"),
+        namespace: source.k8s_namespace().unwrap_or("default"),
         label_selector,
     })
 }
@@ -247,20 +243,16 @@ impl<E: K8sDeployEnv> ClusterWaitHandle<E> for K8sAttachedClusterWait<E> {
 }
 
 fn k8s_wait_request(source: &AttachSource) -> Result<K8sAttachRequest<'_>, DynError> {
-    let AttachSource::K8s {
-        namespace,
-        label_selector,
-    } = source
-    else {
-        return Err("k8s cluster wait requires a k8s attach source".into());
-    };
+    let label_selector = source
+        .k8s_label_selector()
+        .ok_or_else(|| DynError::from("k8s cluster wait requires a k8s attach source"))?;
 
     if label_selector.trim().is_empty() {
         return Err(K8sAttachDiscoveryError::EmptyLabelSelector.into());
     }
 
     Ok(K8sAttachRequest {
-        namespace: namespace.as_deref().unwrap_or("default"),
+        namespace: source.k8s_namespace().unwrap_or("default"),
         label_selector,
     })
 }
diff --git a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
index eb39b33..ec1cb80 100644
--- a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
@@ -5,12 +5,11 @@ use kube::Client;
 use reqwest::Url;
 use testing_framework_core::{
     scenario::{
-        Application, ApplicationExternalProvider, AttachSource, CleanupGuard, ClusterWaitHandle,
-        Deployer, DynError, FeedHandle, FeedRuntime, HttpReadinessRequirement, Metrics,
-        MetricsError, NodeClients, ObservabilityCapabilityProvider, ObservabilityInputs,
-        RequiresNodeControl, RunContext, Runner, Scenario, ScenarioSources,
-        SourceOrchestrationPlan, SourceProviders, StaticManagedProvider,
-        build_source_orchestration_plan, orchestrate_sources_with_providers,
+        Application, ApplicationExternalProvider, CleanupGuard, ClusterWaitHandle, Deployer,
+        DynError, FeedHandle, FeedRuntime, HttpReadinessRequirement, Metrics, MetricsError,
+        NodeClients, ObservabilityCapabilityProvider, ObservabilityInputs, RequiresNodeControl,
+        RunContext, Runner, Scenario, SourceOrchestrationPlan, SourceProviders,
+        StaticManagedProvider, build_source_orchestration_plan, orchestrate_sources_with_providers,
     },
     topology::DeploymentDescriptor,
 };
@@ -250,22 +249,18 @@ where
     E: K8sDeployEnv,
     Caps: Send + Sync,
 {
-    match scenario.sources() {
-        ScenarioSources::Attached {
-            attach:
-                AttachSource::K8s {
-                    namespace,
-                    label_selector,
-                },
-            ..
-        } => K8sDeploymentMetadata {
-            namespace: namespace.clone(),
-            label_selector: Some(label_selector.clone()),
-        },
-        _ => K8sDeploymentMetadata {
-            namespace: None,
-            label_selector: None,
-        },
+    let namespace = scenario
+        .attached_source()
+        .and_then(|attach| attach.k8s_namespace())
+        .map(ToOwned::to_owned);
+    let label_selector = scenario
+        .attached_source()
+        .and_then(|attach| attach.k8s_label_selector())
+        .map(ToOwned::to_owned);
+
+    K8sDeploymentMetadata {
+        namespace,
+        label_selector,
     }
 }
 
@@ -277,11 +272,11 @@ where
     E: K8sDeployEnv,
     Caps: Send + Sync,
 {
-    let ScenarioSources::Attached { attach, .. } = scenario.sources() else {
-        return Err(K8sRunnerError::InternalInvariant {
+    let attach = scenario
+        .attached_source()
+        .ok_or_else(|| K8sRunnerError::InternalInvariant {
             message: "k8s attached cluster wait requested outside attached source mode".to_owned(),
-        });
-    };
+        })?;
 
     Ok(Arc::new(K8sAttachedClusterWait::<E>::new(
         client,

From 05b907d8eff0f2a6c84f27e73c7b22c025ec5325 Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 13:57:15 +0100
Subject: [PATCH 2/9] Name existing cluster semantics explicitly

---
 .../tests/compose_attach_node_control.rs      |  2 +-
 .../examples/tests/k8s_attach_node_control.rs |  2 +-
 .../core/src/scenario/definition.rs           | 26 ++++++++++++++++---
 .../core/src/scenario/sources/model.rs        |  7 ++++-
 .../compose/src/deployer/orchestrator.rs      |  8 +++---
 .../k8s/src/deployer/orchestrator.rs          |  8 +++---
 6 files changed, 38 insertions(+), 15 deletions(-)

diff --git a/logos/examples/tests/compose_attach_node_control.rs b/logos/examples/tests/compose_attach_node_control.rs
index fa3befd..7b3e5fe 100644
--- a/logos/examples/tests/compose_attach_node_control.rs
+++ b/logos/examples/tests/compose_attach_node_control.rs
@@ -23,7 +23,7 @@ async fn compose_attach_mode_queries_node_api_opt_in() -> Result<()> {
     let attach_source = metadata.attach_source().map_err(|err| anyhow!("{err}"))?;
     let attached = ScenarioBuilder::deployment_with(|d| d.with_node_count(1))
         .with_run_duration(Duration::from_secs(5))
-        .with_attach_source(attach_source)
+        .with_existing_cluster(attach_source)
         .build()?;
 
     let attached_deployer = LbcComposeDeployer::default();
diff --git a/logos/examples/tests/k8s_attach_node_control.rs b/logos/examples/tests/k8s_attach_node_control.rs
index 72b2c9d..f309eee 100644
--- a/logos/examples/tests/k8s_attach_node_control.rs
+++ b/logos/examples/tests/k8s_attach_node_control.rs
@@ -23,7 +23,7 @@ async fn k8s_attach_mode_queries_node_api_opt_in() -> Result<()> {
     let attach_source = metadata.attach_source().map_err(|err| anyhow!("{err}"))?;
     let attached = ScenarioBuilder::deployment_with(|d| d.with_node_count(1))
         .with_run_duration(Duration::from_secs(5))
-        .with_attach_source(attach_source)
+        .with_existing_cluster(attach_source)
         .build()?;
 
     let attached_deployer = LbcK8sDeployer::default();
diff --git a/testing-framework/core/src/scenario/definition.rs b/testing-framework/core/src/scenario/definition.rs
index 15c7ea9..9739141 100644
--- a/testing-framework/core/src/scenario/definition.rs
+++ b/testing-framework/core/src/scenario/definition.rs
@@ -118,8 +118,14 @@ impl<E: Application, Caps> Scenario<E, Caps> {
     }
 
     #[must_use]
+    pub fn existing_cluster(&self) -> Option<&AttachSource> {
+        self.sources.existing_cluster()
+    }
+
+    #[must_use]
+    #[doc(hidden)]
     pub fn attached_source(&self) -> Option<&AttachSource> {
-        self.sources.attached_source()
+        self.existing_cluster()
     }
 
     #[must_use]
@@ -243,8 +249,14 @@ macro_rules! impl_common_builder_methods {
             }
 
             #[must_use]
+            pub fn with_existing_cluster(self, cluster: AttachSource) -> Self {
+                self.map_core_builder(|builder| builder.with_existing_cluster(cluster))
+            }
+
+            #[must_use]
+            #[doc(hidden)]
             pub fn with_attach_source(self, attach: AttachSource) -> Self {
-                self.map_core_builder(|builder| builder.with_attach_source(attach))
+                self.with_existing_cluster(attach)
             }
 
             #[must_use]
@@ -556,11 +568,17 @@ impl<E: Application, Caps> Builder<E, Caps> {
     }
 
     #[must_use]
-    pub fn with_attach_source(mut self, attach: AttachSource) -> Self {
-        self.sources = self.sources.with_attach(attach);
+    pub fn with_existing_cluster(mut self, cluster: AttachSource) -> Self {
+        self.sources = self.sources.with_attach(cluster);
         self
     }
 
+    #[must_use]
+    #[doc(hidden)]
+    pub fn with_attach_source(self, attach: AttachSource) -> Self {
+        self.with_existing_cluster(attach)
+    }
+
     #[must_use]
     pub fn with_external_node(mut self, node: ExternalNodeSource) -> Self {
         self.sources = self.sources.with_external_node(node);
diff --git a/testing-framework/core/src/scenario/sources/model.rs b/testing-framework/core/src/scenario/sources/model.rs
index 5b0ec2c..df444f6 100644
--- a/testing-framework/core/src/scenario/sources/model.rs
+++ b/testing-framework/core/src/scenario/sources/model.rs
@@ -173,7 +173,7 @@ impl ScenarioSources {
     }
 
     #[must_use]
-    pub fn attached_source(&self) -> Option<&AttachSource> {
+    pub fn existing_cluster(&self) -> Option<&AttachSource> {
         match self {
             Self::Attached { attach, .. } => Some(attach),
             Self::Managed { .. } | Self::ExternalOnly { .. } => None,
@@ -199,6 +199,11 @@ impl ScenarioSources {
         matches!(self, Self::Attached { .. })
     }
 
+    #[must_use]
+    pub const fn uses_existing_cluster(&self) -> bool {
+        self.is_attached()
+    }
+
     #[must_use]
     pub const fn is_external_only(&self) -> bool {
         matches!(self, Self::ExternalOnly { .. })
diff --git a/testing-framework/deployers/compose/src/deployer/orchestrator.rs b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
index c4fdc4f..9619e86 100644
--- a/testing-framework/deployers/compose/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
@@ -71,7 +71,7 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
             }
         })?;
 
-        if scenario.sources().is_attached() {
+        if scenario.sources().uses_existing_cluster() {
             return self
                 .deploy_attached_only::<Caps>(scenario, source_plan)
                 .await
@@ -215,7 +215,7 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
         }
 
         let attach = scenario
-            .attached_source()
+            .existing_cluster()
             .ok_or(ComposeRunnerError::InternalInvariant {
                 message: "attached node control requested outside attached source mode",
             })?;
@@ -243,7 +243,7 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
         Caps: Send + Sync,
     {
         let attach = scenario
-            .attached_source()
+            .existing_cluster()
             .ok_or(ComposeRunnerError::InternalInvariant {
                 message: "compose attached cluster wait requested outside attached source mode",
             })?;
@@ -373,7 +373,7 @@ where
     Caps: Send + Sync,
 {
     let project_name = scenario
-        .attached_source()
+        .existing_cluster()
         .and_then(|attach| attach.compose_project())
         .map(ToOwned::to_owned);
 
diff --git a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
index ec1cb80..a41e8d5 100644
--- a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
@@ -179,7 +179,7 @@ where
 
     let observability = resolve_observability_inputs(scenario.capabilities())?;
 
-    if scenario.sources().is_attached() {
+    if scenario.sources().uses_existing_cluster() {
         let runner = deploy_attached_only::<E, Caps>(scenario, source_plan, observability).await?;
         return Ok((runner, attached_metadata(scenario)));
     }
@@ -250,11 +250,11 @@ where
     Caps: Send + Sync,
 {
     let namespace = scenario
-        .attached_source()
+        .existing_cluster()
         .and_then(|attach| attach.k8s_namespace())
         .map(ToOwned::to_owned);
     let label_selector = scenario
-        .attached_source()
+        .existing_cluster()
         .and_then(|attach| attach.k8s_label_selector())
         .map(ToOwned::to_owned);
 
@@ -273,7 +273,7 @@ where
     Caps: Send + Sync,
 {
     let attach = scenario
-        .attached_source()
+        .existing_cluster()
         .ok_or_else(|| K8sRunnerError::InternalInvariant {
             message: "k8s attached cluster wait requested outside attached source mode".to_owned(),
         })?;

From f79eb34a507d98034f0bef5e094e91e0d1feb4d5 Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 14:00:34 +0100
Subject: [PATCH 3/9] Hide raw scenario sources behind semantic accessors

---
 testing-framework/core/src/scenario/definition.rs           | 6 ++++++
 .../src/scenario/runtime/orchestration/source_resolver.rs   | 2 +-
 .../deployers/compose/src/deployer/orchestrator.rs          | 2 +-
 .../deployers/k8s/src/deployer/orchestrator.rs              | 2 +-
 4 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/testing-framework/core/src/scenario/definition.rs b/testing-framework/core/src/scenario/definition.rs
index 9739141..56190db 100644
--- a/testing-framework/core/src/scenario/definition.rs
+++ b/testing-framework/core/src/scenario/definition.rs
@@ -113,6 +113,7 @@ impl<E: Application, Caps> Scenario<E, Caps> {
     }
 
     #[must_use]
+    #[doc(hidden)]
     pub fn sources(&self) -> &ScenarioSources {
         &self.sources
     }
@@ -122,6 +123,11 @@ impl<E: Application, Caps> Scenario<E, Caps> {
         self.sources.existing_cluster()
     }
 
+    #[must_use]
+    pub const fn uses_existing_cluster(&self) -> bool {
+        self.sources.uses_existing_cluster()
+    }
+
     #[must_use]
     #[doc(hidden)]
     pub fn attached_source(&self) -> Option<&AttachSource> {
diff --git a/testing-framework/core/src/scenario/runtime/orchestration/source_resolver.rs b/testing-framework/core/src/scenario/runtime/orchestration/source_resolver.rs
index ebf2d2d..d6f822b 100644
--- a/testing-framework/core/src/scenario/runtime/orchestration/source_resolver.rs
+++ b/testing-framework/core/src/scenario/runtime/orchestration/source_resolver.rs
@@ -41,7 +41,7 @@ pub enum SourceResolveError {
 pub fn build_source_orchestration_plan<E: Application, Caps>(
     scenario: &Scenario<E, Caps>,
 ) -> Result<SourceOrchestrationPlan, SourceOrchestrationPlanError> {
-    SourceOrchestrationPlan::try_from_sources(scenario.sources())
+    Ok(scenario.source_orchestration_plan().clone())
 }
 
 /// Resolves runtime source nodes via unified providers from orchestration plan.
diff --git a/testing-framework/deployers/compose/src/deployer/orchestrator.rs b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
index 9619e86..ef93c7f 100644
--- a/testing-framework/deployers/compose/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
@@ -71,7 +71,7 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
             }
         })?;
 
-        if scenario.sources().uses_existing_cluster() {
+        if scenario.uses_existing_cluster() {
             return self
                 .deploy_attached_only::<Caps>(scenario, source_plan)
                 .await
diff --git a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
index a41e8d5..1565a40 100644
--- a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
@@ -179,7 +179,7 @@ where
 
     let observability = resolve_observability_inputs(scenario.capabilities())?;
 
-    if scenario.sources().uses_existing_cluster() {
+    if scenario.uses_existing_cluster() {
         let runner = deploy_attached_only::<E, Caps>(scenario, source_plan, observability).await?;
         return Ok((runner, attached_metadata(scenario)));
     }

From e04832f62c8215fa1435eb2c158e13a48459454d Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 14:03:50 +0100
Subject: [PATCH 4/9] Rename metadata attach flow around existing clusters

---
 .../tests/compose_attach_node_control.rs      |  4 +++-
 .../examples/tests/k8s_attach_node_control.rs |  4 +++-
 .../deployers/compose/src/deployer/mod.rs     | 23 +++++++++++++++----
 .../deployers/k8s/src/deployer/mod.rs         |  9 ++++++--
 .../k8s/src/deployer/orchestrator.rs          |  2 +-
 5 files changed, 32 insertions(+), 10 deletions(-)

diff --git a/logos/examples/tests/compose_attach_node_control.rs b/logos/examples/tests/compose_attach_node_control.rs
index 7b3e5fe..4f909a6 100644
--- a/logos/examples/tests/compose_attach_node_control.rs
+++ b/logos/examples/tests/compose_attach_node_control.rs
@@ -20,7 +20,9 @@ async fn compose_attach_mode_queries_node_api_opt_in() -> Result<()> {
             Err(error) => return Err(Error::new(error)),
         };
 
-    let attach_source = metadata.attach_source().map_err(|err| anyhow!("{err}"))?;
+    let attach_source = metadata
+        .existing_cluster()
+        .map_err(|err| anyhow!("{err}"))?;
     let attached = ScenarioBuilder::deployment_with(|d| d.with_node_count(1))
         .with_run_duration(Duration::from_secs(5))
         .with_existing_cluster(attach_source)
diff --git a/logos/examples/tests/k8s_attach_node_control.rs b/logos/examples/tests/k8s_attach_node_control.rs
index f309eee..47bbe35 100644
--- a/logos/examples/tests/k8s_attach_node_control.rs
+++ b/logos/examples/tests/k8s_attach_node_control.rs
@@ -20,7 +20,9 @@ async fn k8s_attach_mode_queries_node_api_opt_in() -> Result<()> {
             Err(error) => return Err(Error::new(error)),
         };
 
-    let attach_source = metadata.attach_source().map_err(|err| anyhow!("{err}"))?;
+    let attach_source = metadata
+        .existing_cluster()
+        .map_err(|err| anyhow!("{err}"))?;
     let attached = ScenarioBuilder::deployment_with(|d| d.with_node_count(1))
         .with_run_duration(Duration::from_secs(5))
         .with_existing_cluster(attach_source)
diff --git a/testing-framework/deployers/compose/src/deployer/mod.rs b/testing-framework/deployers/compose/src/deployer/mod.rs
index 2f809c7..fd6f44a 100644
--- a/testing-framework/deployers/compose/src/deployer/mod.rs
+++ b/testing-framework/deployers/compose/src/deployer/mod.rs
@@ -43,9 +43,9 @@ impl ComposeDeploymentMetadata {
         self.project_name.as_deref()
     }
 
-    /// Builds an attach source for the same compose project using deployer
-    /// discovery to resolve services.
-    pub fn attach_source(&self) -> Result<AttachSource, DynError> {
+    /// Builds an existing-cluster descriptor for the same compose project
+    /// using deployer discovery to resolve services.
+    pub fn existing_cluster(&self) -> Result<AttachSource, DynError> {
         let project_name = self
             .project_name()
             .ok_or(ComposeMetadataError::MissingProjectName)?;
@@ -56,8 +56,8 @@ impl ComposeDeploymentMetadata {
         ))
     }
 
-    /// Builds an attach source for the same compose project.
-    pub fn attach_source_for_services(
+    /// Builds an existing-cluster descriptor for the same compose project.
+    pub fn existing_cluster_for_services(
         &self,
         services: Vec<String>,
     ) -> Result<AttachSource, DynError> {
@@ -70,6 +70,19 @@ impl ComposeDeploymentMetadata {
             project_name.to_owned(),
         ))
     }
+
+    #[doc(hidden)]
+    pub fn attach_source(&self) -> Result<AttachSource, DynError> {
+        self.existing_cluster()
+    }
+
+    #[doc(hidden)]
+    pub fn attach_source_for_services(
+        &self,
+        services: Vec<String>,
+    ) -> Result<AttachSource, DynError> {
+        self.existing_cluster_for_services(services)
+    }
 }
 
 impl<E: ComposeDeployEnv> Default for ComposeDeployer<E> {
diff --git a/testing-framework/deployers/k8s/src/deployer/mod.rs b/testing-framework/deployers/k8s/src/deployer/mod.rs
index 43f0ff5..a43b83c 100644
--- a/testing-framework/deployers/k8s/src/deployer/mod.rs
+++ b/testing-framework/deployers/k8s/src/deployer/mod.rs
@@ -34,8 +34,8 @@ impl K8sDeploymentMetadata {
         self.label_selector.as_deref()
     }
 
-    /// Builds an attach source for the same k8s deployment scope.
-    pub fn attach_source(&self) -> Result<AttachSource, DynError> {
+    /// Builds an existing-cluster descriptor for the same k8s deployment scope.
+    pub fn existing_cluster(&self) -> Result<AttachSource, DynError> {
         let namespace = self.namespace().ok_or(K8sMetadataError::MissingNamespace)?;
         let label_selector = self
             .label_selector()
@@ -46,4 +46,9 @@ impl K8sDeploymentMetadata {
             namespace.to_owned(),
         ))
     }
+
+    #[doc(hidden)]
+    pub fn attach_source(&self) -> Result<AttachSource, DynError> {
+        self.existing_cluster()
+    }
 }
diff --git a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
index 1565a40..836a892 100644
--- a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
@@ -290,7 +290,7 @@ fn managed_cluster_wait<E: K8sDeployEnv>(
 ) -> Result<Arc<dyn ClusterWaitHandle<E>>, K8sRunnerError> {
     let client = client_from_cluster(cluster)?;
     let attach_source = metadata
-        .attach_source()
+        .existing_cluster()
         .map_err(|source| K8sRunnerError::SourceOrchestration { source })?;
 
     Ok(Arc::new(K8sAttachedClusterWait::<E>::new(

From 4d8349679e74e05d2c2a7d5120cfc841f1d38a8b Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 14:05:28 +0100
Subject: [PATCH 5/9] Rename attach source around existing clusters

---
 .../core/src/scenario/definition.rs              | 14 +++++++-------
 testing-framework/core/src/scenario/mod.rs       |  4 +++-
 .../orchestration/source_orchestration_plan.rs   |  9 +++++----
 .../runtime/providers/attach_provider.rs         |  8 ++++----
 .../core/src/scenario/sources/mod.rs             |  4 +++-
 .../core/src/scenario/sources/model.rs           | 13 ++++++++-----
 .../compose/src/deployer/attach_provider.rs      | 14 +++++++-------
 .../deployers/compose/src/deployer/mod.rs        | 14 +++++++-------
 .../compose/src/deployer/orchestrator.rs         |  6 +++---
 .../k8s/src/deployer/attach_provider.rs          | 16 +++++++++-------
 .../deployers/k8s/src/deployer/mod.rs            |  8 ++++----
 11 files changed, 60 insertions(+), 50 deletions(-)

diff --git a/testing-framework/core/src/scenario/definition.rs b/testing-framework/core/src/scenario/definition.rs
index 56190db..83ebed1 100644
--- a/testing-framework/core/src/scenario/definition.rs
+++ b/testing-framework/core/src/scenario/definition.rs
@@ -4,7 +4,7 @@ use thiserror::Error;
 use tracing::{debug, info};
 
 use super::{
-    Application, AttachSource, DeploymentPolicy, DynError, ExternalNodeSource,
+    Application, DeploymentPolicy, DynError, ExistingCluster, ExternalNodeSource,
     HttpReadinessRequirement, NodeControlCapability, ObservabilityCapability, ScenarioSources,
     builder_ops::CoreBuilderAccess,
     expectation::Expectation,
@@ -119,7 +119,7 @@ impl<E: Application, Caps> Scenario<E, Caps> {
     }
 
     #[must_use]
-    pub fn existing_cluster(&self) -> Option<&AttachSource> {
+    pub fn existing_cluster(&self) -> Option<&ExistingCluster> {
         self.sources.existing_cluster()
     }
 
@@ -130,7 +130,7 @@ impl<E: Application, Caps> Scenario<E, Caps> {
 
     #[must_use]
     #[doc(hidden)]
-    pub fn attached_source(&self) -> Option<&AttachSource> {
+    pub fn attached_source(&self) -> Option<&ExistingCluster> {
         self.existing_cluster()
     }
 
@@ -255,13 +255,13 @@ macro_rules! impl_common_builder_methods {
             }
 
             #[must_use]
-            pub fn with_existing_cluster(self, cluster: AttachSource) -> Self {
+            pub fn with_existing_cluster(self, cluster: ExistingCluster) -> Self {
                 self.map_core_builder(|builder| builder.with_existing_cluster(cluster))
             }
 
             #[must_use]
             #[doc(hidden)]
-            pub fn with_attach_source(self, attach: AttachSource) -> Self {
+            pub fn with_attach_source(self, attach: ExistingCluster) -> Self {
                 self.with_existing_cluster(attach)
             }
 
@@ -574,14 +574,14 @@ impl<E: Application, Caps> Builder<E, Caps> {
     }
 
     #[must_use]
-    pub fn with_existing_cluster(mut self, cluster: AttachSource) -> Self {
+    pub fn with_existing_cluster(mut self, cluster: ExistingCluster) -> Self {
         self.sources = self.sources.with_attach(cluster);
         self
     }
 
     #[must_use]
     #[doc(hidden)]
-    pub fn with_attach_source(self, attach: AttachSource) -> Self {
+    pub fn with_attach_source(self, attach: ExistingCluster) -> Self {
         self.with_existing_cluster(attach)
     }
 
diff --git a/testing-framework/core/src/scenario/mod.rs b/testing-framework/core/src/scenario/mod.rs
index 8b570ad..37dc1ef 100644
--- a/testing-framework/core/src/scenario/mod.rs
+++ b/testing-framework/core/src/scenario/mod.rs
@@ -51,7 +51,9 @@ pub use runtime::{
     wait_for_http_ports_with_host_and_requirement, wait_for_http_ports_with_requirement,
     wait_http_readiness, wait_until_stable,
 };
-pub use sources::{AttachSource, ExternalNodeSource, ScenarioSources};
+#[doc(hidden)]
+pub use sources::AttachSource;
+pub use sources::{ExistingCluster, ExternalNodeSource, ScenarioSources};
 pub use workload::Workload;
 
 pub use crate::env::Application;
diff --git a/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs b/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
index f1dd9b9..2c10958 100644
--- a/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
+++ b/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
@@ -1,4 +1,4 @@
-use crate::scenario::{AttachSource, ExternalNodeSource, ScenarioSources};
+use crate::scenario::{ExistingCluster, ExternalNodeSource, ScenarioSources};
 
 /// Explicit descriptor for managed node sourcing.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -19,7 +19,7 @@ pub(crate) enum SourceOrchestrationMode {
         external: Vec<ExternalNodeSource>,
     },
     Attached {
-        attach: AttachSource,
+        attach: ExistingCluster,
         external: Vec<ExternalNodeSource>,
     },
     ExternalOnly {
@@ -69,11 +69,12 @@ impl SourceOrchestrationPlan {
 #[cfg(test)]
 mod tests {
     use super::{SourceOrchestrationMode, SourceOrchestrationPlan};
-    use crate::scenario::{AttachSource, ScenarioSources};
+    use crate::scenario::{ExistingCluster, ScenarioSources};
 
     #[test]
     fn attached_sources_are_planned() {
-        let sources = ScenarioSources::attached(AttachSource::compose(vec!["node-0".to_string()]));
+        let sources =
+            ScenarioSources::attached(ExistingCluster::compose(vec!["node-0".to_string()]));
         let plan = SourceOrchestrationPlan::try_from_sources(&sources)
             .expect("attached sources should build a source orchestration plan");
 
diff --git a/testing-framework/core/src/scenario/runtime/providers/attach_provider.rs b/testing-framework/core/src/scenario/runtime/providers/attach_provider.rs
index d31bd78..b239193 100644
--- a/testing-framework/core/src/scenario/runtime/providers/attach_provider.rs
+++ b/testing-framework/core/src/scenario/runtime/providers/attach_provider.rs
@@ -1,6 +1,6 @@
 use async_trait::async_trait;
 
-use crate::scenario::{Application, AttachSource, DynError};
+use crate::scenario::{Application, DynError, ExistingCluster};
 
 /// Attached node discovered from an existing external cluster source.
 #[derive(Clone, Debug)]
@@ -15,7 +15,7 @@ pub struct AttachedNode<E: Application> {
 #[derive(Debug, thiserror::Error)]
 pub enum AttachProviderError {
     #[error("attach source is not supported by this provider: {attach_source:?}")]
-    UnsupportedSource { attach_source: AttachSource },
+    UnsupportedSource { attach_source: ExistingCluster },
     #[error("attach discovery failed: {source}")]
     Discovery {
         #[source]
@@ -32,7 +32,7 @@ pub trait AttachProvider<E: Application>: Send + Sync {
     /// Discovers node clients for the requested attach source.
     async fn discover(
         &self,
-        source: &AttachSource,
+        source: &ExistingCluster,
     ) -> Result<Vec<AttachedNode<E>>, AttachProviderError>;
 }
 
@@ -44,7 +44,7 @@ pub struct NoopAttachProvider;
 impl<E: Application> AttachProvider<E> for NoopAttachProvider {
     async fn discover(
         &self,
-        source: &AttachSource,
+        source: &ExistingCluster,
     ) -> Result<Vec<AttachedNode<E>>, AttachProviderError> {
         Err(AttachProviderError::UnsupportedSource {
             attach_source: source.clone(),
diff --git a/testing-framework/core/src/scenario/sources/mod.rs b/testing-framework/core/src/scenario/sources/mod.rs
index 98324dd..26075f2 100644
--- a/testing-framework/core/src/scenario/sources/mod.rs
+++ b/testing-framework/core/src/scenario/sources/mod.rs
@@ -1,3 +1,5 @@
 mod model;
 
-pub use model::{AttachSource, ExternalNodeSource, ScenarioSources};
+#[doc(hidden)]
+pub use model::AttachSource;
+pub use model::{ExistingCluster, ExternalNodeSource, ScenarioSources};
diff --git a/testing-framework/core/src/scenario/sources/model.rs b/testing-framework/core/src/scenario/sources/model.rs
index df444f6..8e62419 100644
--- a/testing-framework/core/src/scenario/sources/model.rs
+++ b/testing-framework/core/src/scenario/sources/model.rs
@@ -1,6 +1,6 @@
 /// Typed attach source for existing clusters.
 #[derive(Clone, Debug, Eq, PartialEq)]
-pub enum AttachSource {
+pub enum ExistingCluster {
     K8s {
         namespace: Option<String>,
         label_selector: String,
@@ -11,7 +11,7 @@ pub enum AttachSource {
     },
 }
 
-impl AttachSource {
+impl ExistingCluster {
     #[must_use]
     pub fn k8s(label_selector: String) -> Self {
         Self::K8s {
@@ -77,6 +77,9 @@ impl AttachSource {
     }
 }
 
+#[doc(hidden)]
+pub type AttachSource = ExistingCluster;
+
 /// Static external node endpoint that should be included in the runtime
 /// inventory.
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -110,7 +113,7 @@ pub enum ScenarioSources {
         external: Vec<ExternalNodeSource>,
     },
     Attached {
-        attach: AttachSource,
+        attach: ExistingCluster,
         external: Vec<ExternalNodeSource>,
     },
     ExternalOnly {
@@ -135,7 +138,7 @@ impl ScenarioSources {
     }
 
     #[must_use]
-    pub fn attached(attach: AttachSource) -> Self {
+    pub fn attached(attach: ExistingCluster) -> Self {
         Self::Attached {
             attach,
             external: Vec::new(),
@@ -159,7 +162,7 @@ impl ScenarioSources {
     }
 
     #[must_use]
-    pub fn with_attach(self, attach: AttachSource) -> Self {
+    pub fn with_attach(self, attach: ExistingCluster) -> Self {
         let external = self.external_nodes().to_vec();
 
         Self::Attached { attach, external }
diff --git a/testing-framework/deployers/compose/src/deployer/attach_provider.rs b/testing-framework/deployers/compose/src/deployer/attach_provider.rs
index 06c8d61..314ae3d 100644
--- a/testing-framework/deployers/compose/src/deployer/attach_provider.rs
+++ b/testing-framework/deployers/compose/src/deployer/attach_provider.rs
@@ -2,8 +2,8 @@ use std::marker::PhantomData;
 
 use async_trait::async_trait;
 use testing_framework_core::scenario::{
-    AttachProvider, AttachProviderError, AttachSource, AttachedNode, ClusterWaitHandle, DynError,
-    ExternalNodeSource, HttpReadinessRequirement, wait_http_readiness,
+    AttachProvider, AttachProviderError, AttachedNode, ClusterWaitHandle, DynError,
+    ExistingCluster, ExternalNodeSource, HttpReadinessRequirement, wait_http_readiness,
 };
 use url::Url;
 
@@ -22,7 +22,7 @@ pub(super) struct ComposeAttachProvider<E: ComposeDeployEnv> {
 
 pub(super) struct ComposeAttachedClusterWait<E: ComposeDeployEnv> {
     host: String,
-    source: AttachSource,
+    source: ExistingCluster,
     _env: PhantomData<E>,
 }
 
@@ -42,7 +42,7 @@ impl<E: ComposeDeployEnv> ComposeAttachProvider<E> {
 }
 
 impl<E: ComposeDeployEnv> ComposeAttachedClusterWait<E> {
-    pub(super) fn new(host: String, source: AttachSource) -> Self {
+    pub(super) fn new(host: String, source: ExistingCluster) -> Self {
         Self {
             host,
             source,
@@ -60,7 +60,7 @@ struct ComposeAttachRequest<'a> {
 impl<E: ComposeDeployEnv> AttachProvider<E> for ComposeAttachProvider<E> {
     async fn discover(
         &self,
-        source: &AttachSource,
+        source: &ExistingCluster,
     ) -> Result<Vec<AttachedNode<E>>, AttachProviderError> {
         let request = compose_attach_request(source)?;
         let services = resolve_services(request.project, request.services)
@@ -85,7 +85,7 @@ fn to_discovery_error(source: DynError) -> AttachProviderError {
 }
 
 fn compose_attach_request(
-    source: &AttachSource,
+    source: &ExistingCluster,
 ) -> Result<ComposeAttachRequest<'_>, AttachProviderError> {
     let services =
         source
@@ -173,7 +173,7 @@ impl<E: ComposeDeployEnv> ClusterWaitHandle<E> for ComposeAttachedClusterWait<E>
     }
 }
 
-fn compose_wait_request(source: &AttachSource) -> Result<ComposeAttachRequest<'_>, DynError> {
+fn compose_wait_request(source: &ExistingCluster) -> Result<ComposeAttachRequest<'_>, DynError> {
     let project = source
         .compose_project()
         .ok_or_else(|| DynError::from("compose cluster wait requires a compose attach source"))?;
diff --git a/testing-framework/deployers/compose/src/deployer/mod.rs b/testing-framework/deployers/compose/src/deployer/mod.rs
index fd6f44a..66859c1 100644
--- a/testing-framework/deployers/compose/src/deployer/mod.rs
+++ b/testing-framework/deployers/compose/src/deployer/mod.rs
@@ -9,7 +9,7 @@ use std::marker::PhantomData;
 
 use async_trait::async_trait;
 use testing_framework_core::scenario::{
-    AttachSource, CleanupGuard, Deployer, DynError, FeedHandle, ObservabilityCapabilityProvider,
+    CleanupGuard, Deployer, DynError, ExistingCluster, FeedHandle, ObservabilityCapabilityProvider,
     RequiresNodeControl, Runner, Scenario,
 };
 
@@ -45,12 +45,12 @@ impl ComposeDeploymentMetadata {
 
     /// Builds an existing-cluster descriptor for the same compose project
     /// using deployer discovery to resolve services.
-    pub fn existing_cluster(&self) -> Result<AttachSource, DynError> {
+    pub fn existing_cluster(&self) -> Result<ExistingCluster, DynError> {
         let project_name = self
             .project_name()
             .ok_or(ComposeMetadataError::MissingProjectName)?;
 
-        Ok(AttachSource::compose_in_project(
+        Ok(ExistingCluster::compose_in_project(
             Vec::new(),
             project_name.to_owned(),
         ))
@@ -60,19 +60,19 @@ impl ComposeDeploymentMetadata {
     pub fn existing_cluster_for_services(
         &self,
         services: Vec<String>,
-    ) -> Result<AttachSource, DynError> {
+    ) -> Result<ExistingCluster, DynError> {
         let project_name = self
             .project_name()
             .ok_or(ComposeMetadataError::MissingProjectName)?;
 
-        Ok(AttachSource::compose_in_project(
+        Ok(ExistingCluster::compose_in_project(
             services,
             project_name.to_owned(),
         ))
     }
 
     #[doc(hidden)]
-    pub fn attach_source(&self) -> Result<AttachSource, DynError> {
+    pub fn attach_source(&self) -> Result<ExistingCluster, DynError> {
         self.existing_cluster()
     }
 
@@ -80,7 +80,7 @@ impl ComposeDeploymentMetadata {
     pub fn attach_source_for_services(
         &self,
         services: Vec<String>,
-    ) -> Result<AttachSource, DynError> {
+    ) -> Result<ExistingCluster, DynError> {
         self.existing_cluster_for_services(services)
     }
 }
diff --git a/testing-framework/deployers/compose/src/deployer/orchestrator.rs b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
index ef93c7f..d9cf18c 100644
--- a/testing-framework/deployers/compose/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
@@ -3,8 +3,8 @@ use std::{env, sync::Arc, time::Duration};
 use reqwest::Url;
 use testing_framework_core::{
     scenario::{
-        ApplicationExternalProvider, AttachSource, CleanupGuard, ClusterWaitHandle,
-        DeploymentPolicy, FeedHandle, FeedRuntime, HttpReadinessRequirement, Metrics, NodeClients,
+        ApplicationExternalProvider, CleanupGuard, ClusterWaitHandle, DeploymentPolicy,
+        ExistingCluster, FeedHandle, FeedRuntime, HttpReadinessRequirement, Metrics, NodeClients,
         NodeControlHandle, ObservabilityCapabilityProvider, ObservabilityInputs,
         RequiresNodeControl, RunContext, Runner, Scenario, SourceOrchestrationPlan,
         SourceProviders, StaticManagedProvider, build_source_orchestration_plan,
@@ -315,7 +315,7 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
     fn managed_cluster_wait(&self, project_name: String) -> Arc<dyn ClusterWaitHandle<E>> {
         Arc::new(ComposeAttachedClusterWait::<E>::new(
             compose_runner_host(),
-            AttachSource::compose_in_project(Vec::new(), project_name),
+            ExistingCluster::compose_in_project(Vec::new(), project_name),
         ))
     }
 
diff --git a/testing-framework/deployers/k8s/src/deployer/attach_provider.rs b/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
index bae5692..cfa0dc2 100644
--- a/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
+++ b/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
@@ -7,8 +7,8 @@ use kube::{
     api::{ListParams, ObjectList},
 };
 use testing_framework_core::scenario::{
-    AttachProvider, AttachProviderError, AttachSource, AttachedNode, ClusterWaitHandle, DynError,
-    ExternalNodeSource, HttpReadinessRequirement, wait_http_readiness,
+    AttachProvider, AttachProviderError, AttachedNode, ClusterWaitHandle, DynError,
+    ExistingCluster, ExternalNodeSource, HttpReadinessRequirement, wait_http_readiness,
 };
 use url::Url;
 
@@ -37,7 +37,7 @@ pub(super) struct K8sAttachProvider<E: K8sDeployEnv> {
 
 pub(super) struct K8sAttachedClusterWait<E: K8sDeployEnv> {
     client: Client,
-    source: AttachSource,
+    source: ExistingCluster,
     _env: PhantomData<E>,
 }
 
@@ -56,7 +56,7 @@ impl<E: K8sDeployEnv> K8sAttachProvider<E> {
 }
 
 impl<E: K8sDeployEnv> K8sAttachedClusterWait<E> {
-    pub(super) fn new(client: Client, source: AttachSource) -> Self {
+    pub(super) fn new(client: Client, source: ExistingCluster) -> Self {
         Self {
             client,
             source,
@@ -69,7 +69,7 @@ impl<E: K8sDeployEnv> K8sAttachedClusterWait<E> {
 impl<E: K8sDeployEnv> AttachProvider<E> for K8sAttachProvider<E> {
     async fn discover(
         &self,
-        source: &AttachSource,
+        source: &ExistingCluster,
     ) -> Result<Vec<AttachedNode<E>>, AttachProviderError> {
         let request = k8s_attach_request(source)?;
         let services = discover_services(&self.client, request.namespace, request.label_selector)
@@ -90,7 +90,9 @@ fn to_discovery_error(source: DynError) -> AttachProviderError {
     AttachProviderError::Discovery { source }
 }
 
-fn k8s_attach_request(source: &AttachSource) -> Result<K8sAttachRequest<'_>, AttachProviderError> {
+fn k8s_attach_request(
+    source: &ExistingCluster,
+) -> Result<K8sAttachRequest<'_>, AttachProviderError> {
     let Some(label_selector) = source.k8s_label_selector() else {
         return Err(AttachProviderError::UnsupportedSource {
             attach_source: source.clone(),
@@ -242,7 +244,7 @@ impl<E: K8sDeployEnv> ClusterWaitHandle<E> for K8sAttachedClusterWait<E> {
     }
 }
 
-fn k8s_wait_request(source: &AttachSource) -> Result<K8sAttachRequest<'_>, DynError> {
+fn k8s_wait_request(source: &ExistingCluster) -> Result<K8sAttachRequest<'_>, DynError> {
     let label_selector = source
         .k8s_label_selector()
         .ok_or_else(|| DynError::from("k8s cluster wait requires a k8s attach source"))?;
diff --git a/testing-framework/deployers/k8s/src/deployer/mod.rs b/testing-framework/deployers/k8s/src/deployer/mod.rs
index a43b83c..eae7507 100644
--- a/testing-framework/deployers/k8s/src/deployer/mod.rs
+++ b/testing-framework/deployers/k8s/src/deployer/mod.rs
@@ -2,7 +2,7 @@ mod attach_provider;
 mod orchestrator;
 
 pub use orchestrator::{K8sDeployer, K8sRunnerError};
-use testing_framework_core::scenario::{AttachSource, DynError};
+use testing_framework_core::scenario::{DynError, ExistingCluster};
 
 /// Kubernetes deployment metadata returned by k8s-specific deployment APIs.
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -35,20 +35,20 @@ impl K8sDeploymentMetadata {
     }
 
     /// Builds an existing-cluster descriptor for the same k8s deployment scope.
-    pub fn existing_cluster(&self) -> Result<AttachSource, DynError> {
+    pub fn existing_cluster(&self) -> Result<ExistingCluster, DynError> {
         let namespace = self.namespace().ok_or(K8sMetadataError::MissingNamespace)?;
         let label_selector = self
             .label_selector()
             .ok_or(K8sMetadataError::MissingLabelSelector)?;
 
-        Ok(AttachSource::k8s_in_namespace(
+        Ok(ExistingCluster::k8s_in_namespace(
             label_selector.to_owned(),
             namespace.to_owned(),
         ))
     }
 
     #[doc(hidden)]
-    pub fn attach_source(&self) -> Result<AttachSource, DynError> {
+    pub fn attach_source(&self) -> Result<ExistingCluster, DynError> {
         self.existing_cluster()
     }
 }

From a5230242797380bacaf6263e12b5dadb5aa754ee Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 14:08:42 +0100
Subject: [PATCH 6/9] Hide raw source types behind semantic scenario API

---
 testing-framework/core/src/scenario/definition.rs | 15 +++++++++++++++
 testing-framework/core/src/scenario/mod.rs        |  4 ++--
 .../core/src/scenario/sources/mod.rs              |  4 ++--
 3 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/testing-framework/core/src/scenario/definition.rs b/testing-framework/core/src/scenario/definition.rs
index 83ebed1..fe6dad3 100644
--- a/testing-framework/core/src/scenario/definition.rs
+++ b/testing-framework/core/src/scenario/definition.rs
@@ -139,6 +139,21 @@ impl<E: Application, Caps> Scenario<E, Caps> {
         self.sources.external_nodes()
     }
 
+    #[must_use]
+    pub const fn is_managed(&self) -> bool {
+        self.sources.is_managed()
+    }
+
+    #[must_use]
+    pub const fn is_external_only(&self) -> bool {
+        self.sources.is_external_only()
+    }
+
+    #[must_use]
+    pub fn has_external_nodes(&self) -> bool {
+        !self.sources.external_nodes().is_empty()
+    }
+
     #[must_use]
     pub const fn source_orchestration_plan(&self) -> &SourceOrchestrationPlan {
         &self.source_orchestration_plan
diff --git a/testing-framework/core/src/scenario/mod.rs b/testing-framework/core/src/scenario/mod.rs
index 37dc1ef..5fcd013 100644
--- a/testing-framework/core/src/scenario/mod.rs
+++ b/testing-framework/core/src/scenario/mod.rs
@@ -52,8 +52,8 @@ pub use runtime::{
     wait_http_readiness, wait_until_stable,
 };
 #[doc(hidden)]
-pub use sources::AttachSource;
-pub use sources::{ExistingCluster, ExternalNodeSource, ScenarioSources};
+pub use sources::{AttachSource, ScenarioSources};
+pub use sources::{ExistingCluster, ExternalNodeSource};
 pub use workload::Workload;
 
 pub use crate::env::Application;
diff --git a/testing-framework/core/src/scenario/sources/mod.rs b/testing-framework/core/src/scenario/sources/mod.rs
index 26075f2..15ddbf0 100644
--- a/testing-framework/core/src/scenario/sources/mod.rs
+++ b/testing-framework/core/src/scenario/sources/mod.rs
@@ -1,5 +1,5 @@
 mod model;
 
 #[doc(hidden)]
-pub use model::AttachSource;
-pub use model::{ExistingCluster, ExternalNodeSource, ScenarioSources};
+pub use model::{AttachSource, ScenarioSources};
+pub use model::{ExistingCluster, ExternalNodeSource};

From f18820b8d1009117bc44b4939bfd72b7cfbedc2d Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 14:13:45 +0100
Subject: [PATCH 7/9] Hide raw source storage behind existing-cluster API

---
 .../core/src/scenario/definition.rs           |   9 +-
 testing-framework/core/src/scenario/mod.rs    |   2 -
 .../source_orchestration_plan.rs              |  10 +-
 .../core/src/scenario/sources/mod.rs          |   2 +-
 .../core/src/scenario/sources/model.rs        | 119 +++++++++---------
 5 files changed, 64 insertions(+), 78 deletions(-)

diff --git a/testing-framework/core/src/scenario/definition.rs b/testing-framework/core/src/scenario/definition.rs
index fe6dad3..0eea6f2 100644
--- a/testing-framework/core/src/scenario/definition.rs
+++ b/testing-framework/core/src/scenario/definition.rs
@@ -5,13 +5,14 @@ use tracing::{debug, info};
 
 use super::{
     Application, DeploymentPolicy, DynError, ExistingCluster, ExternalNodeSource,
-    HttpReadinessRequirement, NodeControlCapability, ObservabilityCapability, ScenarioSources,
+    HttpReadinessRequirement, NodeControlCapability, ObservabilityCapability,
     builder_ops::CoreBuilderAccess,
     expectation::Expectation,
     runtime::{
         context::RunMetrics,
         orchestration::{SourceOrchestrationPlan, SourceOrchestrationPlanError},
     },
+    sources::ScenarioSources,
     workload::Workload,
 };
 use crate::topology::{DeploymentDescriptor, DeploymentProvider, DeploymentSeed, DynTopologyError};
@@ -112,12 +113,6 @@ impl<E: Application, Caps> Scenario<E, Caps> {
         self.deployment_policy
     }
 
-    #[must_use]
-    #[doc(hidden)]
-    pub fn sources(&self) -> &ScenarioSources {
-        &self.sources
-    }
-
     #[must_use]
     pub fn existing_cluster(&self) -> Option<&ExistingCluster> {
         self.sources.existing_cluster()
diff --git a/testing-framework/core/src/scenario/mod.rs b/testing-framework/core/src/scenario/mod.rs
index 5fcd013..ca358a5 100644
--- a/testing-framework/core/src/scenario/mod.rs
+++ b/testing-framework/core/src/scenario/mod.rs
@@ -51,8 +51,6 @@ pub use runtime::{
     wait_for_http_ports_with_host_and_requirement, wait_for_http_ports_with_requirement,
     wait_http_readiness, wait_until_stable,
 };
-#[doc(hidden)]
-pub use sources::{AttachSource, ScenarioSources};
 pub use sources::{ExistingCluster, ExternalNodeSource};
 pub use workload::Workload;
 
diff --git a/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs b/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
index 2c10958..c9c99f9 100644
--- a/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
+++ b/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
@@ -1,4 +1,4 @@
-use crate::scenario::{ExistingCluster, ExternalNodeSource, ScenarioSources};
+use crate::scenario::{ExistingCluster, ExternalNodeSource, sources::ScenarioSources};
 
 /// Explicit descriptor for managed node sourcing.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -43,7 +43,7 @@ pub enum SourceOrchestrationPlanError {
 }
 
 impl SourceOrchestrationPlan {
-    pub fn try_from_sources(
+    pub(crate) fn try_from_sources(
         sources: &ScenarioSources,
     ) -> Result<Self, SourceOrchestrationPlanError> {
         let mode = mode_from_sources(sources);
@@ -69,12 +69,12 @@ impl SourceOrchestrationPlan {
 #[cfg(test)]
 mod tests {
     use super::{SourceOrchestrationMode, SourceOrchestrationPlan};
-    use crate::scenario::{ExistingCluster, ScenarioSources};
+    use crate::scenario::{ExistingCluster, sources::ScenarioSources};
 
     #[test]
     fn attached_sources_are_planned() {
-        let sources =
-            ScenarioSources::attached(ExistingCluster::compose(vec!["node-0".to_string()]));
+        let sources = ScenarioSources::default()
+            .with_attach(ExistingCluster::compose(vec!["node-0".to_string()]));
         let plan = SourceOrchestrationPlan::try_from_sources(&sources)
             .expect("attached sources should build a source orchestration plan");
 
diff --git a/testing-framework/core/src/scenario/sources/mod.rs b/testing-framework/core/src/scenario/sources/mod.rs
index 15ddbf0..506a8aa 100644
--- a/testing-framework/core/src/scenario/sources/mod.rs
+++ b/testing-framework/core/src/scenario/sources/mod.rs
@@ -1,5 +1,5 @@
 mod model;
 
+pub(crate) use model::ScenarioSources;
 #[doc(hidden)]
-pub use model::{AttachSource, ScenarioSources};
 pub use model::{ExistingCluster, ExternalNodeSource};
diff --git a/testing-framework/core/src/scenario/sources/model.rs b/testing-framework/core/src/scenario/sources/model.rs
index 8e62419..566e2ee 100644
--- a/testing-framework/core/src/scenario/sources/model.rs
+++ b/testing-framework/core/src/scenario/sources/model.rs
@@ -1,6 +1,11 @@
-/// Typed attach source for existing clusters.
+/// Typed descriptor for an existing cluster.
 #[derive(Clone, Debug, Eq, PartialEq)]
-pub enum ExistingCluster {
+pub struct ExistingCluster {
+    kind: ExistingClusterKind,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+enum ExistingClusterKind {
     K8s {
         namespace: Option<String>,
         label_selector: String,
@@ -13,73 +18,86 @@ pub enum ExistingCluster {
 
 impl ExistingCluster {
     #[must_use]
+    #[doc(hidden)]
     pub fn k8s(label_selector: String) -> Self {
-        Self::K8s {
-            namespace: None,
-            label_selector,
+        Self {
+            kind: ExistingClusterKind::K8s {
+                namespace: None,
+                label_selector,
+            },
         }
     }
 
     #[must_use]
+    #[doc(hidden)]
     pub fn k8s_in_namespace(label_selector: String, namespace: String) -> Self {
-        Self::K8s {
-            namespace: Some(namespace),
-            label_selector,
+        Self {
+            kind: ExistingClusterKind::K8s {
+                namespace: Some(namespace),
+                label_selector,
+            },
         }
     }
 
     #[must_use]
+    #[doc(hidden)]
     pub fn compose(services: Vec<String>) -> Self {
-        Self::Compose {
-            project: None,
-            services,
+        Self {
+            kind: ExistingClusterKind::Compose {
+                project: None,
+                services,
+            },
         }
     }
 
     #[must_use]
+    #[doc(hidden)]
     pub fn compose_in_project(services: Vec<String>, project: String) -> Self {
-        Self::Compose {
-            project: Some(project),
-            services,
+        Self {
+            kind: ExistingClusterKind::Compose {
+                project: Some(project),
+                services,
+            },
         }
     }
 
     #[must_use]
+    #[doc(hidden)]
     pub fn compose_project(&self) -> Option<&str> {
-        match self {
-            Self::Compose { project, .. } => project.as_deref(),
-            Self::K8s { .. } => None,
+        match &self.kind {
+            ExistingClusterKind::Compose { project, .. } => project.as_deref(),
+            ExistingClusterKind::K8s { .. } => None,
         }
     }
 
     #[must_use]
+    #[doc(hidden)]
     pub fn compose_services(&self) -> Option<&[String]> {
-        match self {
-            Self::Compose { services, .. } => Some(services),
-            Self::K8s { .. } => None,
+        match &self.kind {
+            ExistingClusterKind::Compose { services, .. } => Some(services),
+            ExistingClusterKind::K8s { .. } => None,
         }
     }
 
     #[must_use]
+    #[doc(hidden)]
     pub fn k8s_namespace(&self) -> Option<&str> {
-        match self {
-            Self::K8s { namespace, .. } => namespace.as_deref(),
-            Self::Compose { .. } => None,
+        match &self.kind {
+            ExistingClusterKind::K8s { namespace, .. } => namespace.as_deref(),
+            ExistingClusterKind::Compose { .. } => None,
         }
     }
 
     #[must_use]
+    #[doc(hidden)]
     pub fn k8s_label_selector(&self) -> Option<&str> {
-        match self {
-            Self::K8s { label_selector, .. } => Some(label_selector),
-            Self::Compose { .. } => None,
+        match &self.kind {
+            ExistingClusterKind::K8s { label_selector, .. } => Some(label_selector),
+            ExistingClusterKind::Compose { .. } => None,
         }
     }
 }
 
-#[doc(hidden)]
-pub type AttachSource = ExistingCluster;
-
 /// Static external node endpoint that should be included in the runtime
 /// inventory.
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -108,7 +126,7 @@ impl ExternalNodeSource {
 /// Source model that makes invalid managed+attached combinations
 /// unrepresentable by type.
 #[derive(Clone, Debug, Eq, PartialEq)]
-pub enum ScenarioSources {
+pub(crate) enum ScenarioSources {
     Managed {
         external: Vec<ExternalNodeSource>,
     },
@@ -131,27 +149,7 @@ impl Default for ScenarioSources {
 
 impl ScenarioSources {
     #[must_use]
-    pub const fn managed() -> Self {
-        Self::Managed {
-            external: Vec::new(),
-        }
-    }
-
-    #[must_use]
-    pub fn attached(attach: ExistingCluster) -> Self {
-        Self::Attached {
-            attach,
-            external: Vec::new(),
-        }
-    }
-
-    #[must_use]
-    pub fn external_only(external: Vec<ExternalNodeSource>) -> Self {
-        Self::ExternalOnly { external }
-    }
-
-    #[must_use]
-    pub fn with_external_node(mut self, node: ExternalNodeSource) -> Self {
+    pub(crate) fn with_external_node(mut self, node: ExternalNodeSource) -> Self {
         match &mut self {
             Self::Managed { external }
             | Self::Attached { external, .. }
@@ -162,21 +160,21 @@ impl ScenarioSources {
     }
 
     #[must_use]
-    pub fn with_attach(self, attach: ExistingCluster) -> Self {
+    pub(crate) fn with_attach(self, attach: ExistingCluster) -> Self {
         let external = self.external_nodes().to_vec();
 
         Self::Attached { attach, external }
     }
 
     #[must_use]
-    pub fn into_external_only(self) -> Self {
+    pub(crate) fn into_external_only(self) -> Self {
         let external = self.external_nodes().to_vec();
 
         Self::ExternalOnly { external }
     }
 
     #[must_use]
-    pub fn existing_cluster(&self) -> Option<&AttachSource> {
+    pub(crate) fn existing_cluster(&self) -> Option<&ExistingCluster> {
         match self {
             Self::Attached { attach, .. } => Some(attach),
             Self::Managed { .. } | Self::ExternalOnly { .. } => None,
@@ -184,7 +182,7 @@ impl ScenarioSources {
     }
 
     #[must_use]
-    pub fn external_nodes(&self) -> &[ExternalNodeSource] {
+    pub(crate) fn external_nodes(&self) -> &[ExternalNodeSource] {
         match self {
             Self::Managed { external }
             | Self::Attached { external, .. }
@@ -193,22 +191,17 @@ impl ScenarioSources {
     }
 
     #[must_use]
-    pub const fn is_managed(&self) -> bool {
+    pub(crate) const fn is_managed(&self) -> bool {
         matches!(self, Self::Managed { .. })
     }
 
     #[must_use]
-    pub const fn is_attached(&self) -> bool {
+    pub(crate) const fn uses_existing_cluster(&self) -> bool {
         matches!(self, Self::Attached { .. })
     }
 
     #[must_use]
-    pub const fn uses_existing_cluster(&self) -> bool {
-        self.is_attached()
-    }
-
-    #[must_use]
-    pub const fn is_external_only(&self) -> bool {
+    pub(crate) const fn is_external_only(&self) -> bool {
         matches!(self, Self::ExternalOnly { .. })
     }
 }

From fbede7f535da53b57b6f45641e3931a537552f13 Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 14:17:56 +0100
Subject: [PATCH 8/9] Confine backend cluster details to deployer adapters

---
 .../compose/src/deployer/attach_provider.rs   | 10 +--
 .../deployers/compose/src/deployer/mod.rs     | 16 +++++
 .../compose/src/deployer/orchestrator.rs      | 61 ++++++++-----------
 .../deployers/compose/src/docker/control.rs   | 18 +++++-
 .../k8s/src/deployer/attach_provider.rs       | 10 +--
 .../deployers/k8s/src/deployer/mod.rs         | 12 ++++
 .../k8s/src/deployer/orchestrator.rs          | 28 +++------
 7 files changed, 90 insertions(+), 65 deletions(-)

diff --git a/testing-framework/deployers/compose/src/deployer/attach_provider.rs b/testing-framework/deployers/compose/src/deployer/attach_provider.rs
index 314ae3d..8874fd1 100644
--- a/testing-framework/deployers/compose/src/deployer/attach_provider.rs
+++ b/testing-framework/deployers/compose/src/deployer/attach_provider.rs
@@ -42,12 +42,14 @@ impl<E: ComposeDeployEnv> ComposeAttachProvider<E> {
 }
 
 impl<E: ComposeDeployEnv> ComposeAttachedClusterWait<E> {
-    pub(super) fn new(host: String, source: ExistingCluster) -> Self {
-        Self {
+    pub(super) fn try_new(host: String, source: &ExistingCluster) -> Result<Self, DynError> {
+        let _ = compose_wait_request(source)?;
+
+        Ok(Self {
             host,
-            source,
+            source: source.clone(),
             _env: PhantomData,
-        }
+        })
     }
 }
 
diff --git a/testing-framework/deployers/compose/src/deployer/mod.rs b/testing-framework/deployers/compose/src/deployer/mod.rs
index 66859c1..2eab7c6 100644
--- a/testing-framework/deployers/compose/src/deployer/mod.rs
+++ b/testing-framework/deployers/compose/src/deployer/mod.rs
@@ -36,6 +36,22 @@ enum ComposeMetadataError {
 }
 
 impl ComposeDeploymentMetadata {
+    #[must_use]
+    pub fn for_project(project_name: String) -> Self {
+        Self {
+            project_name: Some(project_name),
+        }
+    }
+
+    #[must_use]
+    pub fn from_existing_cluster(cluster: Option<&ExistingCluster>) -> Self {
+        Self {
+            project_name: cluster
+                .and_then(ExistingCluster::compose_project)
+                .map(ToOwned::to_owned),
+        }
+    }
+
     /// Returns project name when deployment is bound to a specific compose
     /// project.
     #[must_use]
diff --git a/testing-framework/deployers/compose/src/deployer/orchestrator.rs b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
index d9cf18c..29b58c3 100644
--- a/testing-framework/deployers/compose/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/compose/src/deployer/orchestrator.rs
@@ -3,12 +3,11 @@ use std::{env, sync::Arc, time::Duration};
 use reqwest::Url;
 use testing_framework_core::{
     scenario::{
-        ApplicationExternalProvider, CleanupGuard, ClusterWaitHandle, DeploymentPolicy,
-        ExistingCluster, FeedHandle, FeedRuntime, HttpReadinessRequirement, Metrics, NodeClients,
-        NodeControlHandle, ObservabilityCapabilityProvider, ObservabilityInputs,
-        RequiresNodeControl, RunContext, Runner, Scenario, SourceOrchestrationPlan,
-        SourceProviders, StaticManagedProvider, build_source_orchestration_plan,
-        orchestrate_sources_with_providers,
+        ApplicationExternalProvider, CleanupGuard, ClusterWaitHandle, DeploymentPolicy, FeedHandle,
+        FeedRuntime, HttpReadinessRequirement, Metrics, NodeClients, NodeControlHandle,
+        ObservabilityCapabilityProvider, ObservabilityInputs, RequiresNodeControl, RunContext,
+        Runner, Scenario, SourceOrchestrationPlan, SourceProviders, StaticManagedProvider,
+        build_source_orchestration_plan, orchestrate_sources_with_providers,
     },
     topology::DeploymentDescriptor,
 };
@@ -219,20 +218,10 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
             .ok_or(ComposeRunnerError::InternalInvariant {
                 message: "attached node control requested outside attached source mode",
             })?;
+        let node_control = ComposeAttachedNodeControl::try_from_existing_cluster(attach)
+            .map_err(|source| ComposeRunnerError::SourceOrchestration { source })?;
 
-        let Some(project_name) = attach
-            .compose_project()
-            .map(str::trim)
-            .filter(|value| !value.is_empty())
-        else {
-            return Err(ComposeRunnerError::InternalInvariant {
-                message: "attached compose mode requires explicit project name for node control",
-            });
-        };
-
-        Ok(Some(Arc::new(ComposeAttachedNodeControl {
-            project_name: project_name.to_owned(),
-        }) as Arc<dyn NodeControlHandle<E>>))
+        Ok(Some(Arc::new(node_control) as Arc<dyn NodeControlHandle<E>>))
     }
 
     fn attached_cluster_wait<Caps>(
@@ -247,11 +236,10 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
             .ok_or(ComposeRunnerError::InternalInvariant {
                 message: "compose attached cluster wait requested outside attached source mode",
             })?;
+        let cluster_wait = ComposeAttachedClusterWait::<E>::try_new(compose_runner_host(), attach)
+            .map_err(|source| ComposeRunnerError::SourceOrchestration { source })?;
 
-        Ok(Arc::new(ComposeAttachedClusterWait::<E>::new(
-            compose_runner_host(),
-            attach.clone(),
-        )))
+        Ok(Arc::new(cluster_wait))
     }
 
     async fn build_runner<Caps>(
@@ -268,7 +256,8 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
     {
         let telemetry = observability.telemetry_handle()?;
         let node_control = self.maybe_node_control::<Caps>(&prepared.environment);
-        let cluster_wait = self.managed_cluster_wait(project_name);
+        let cluster_wait =
+            self.managed_cluster_wait(ComposeDeploymentMetadata::for_project(project_name))?;
 
         log_observability_endpoints(&observability);
         log_profiling_urls(&deployed.host, &deployed.host_ports);
@@ -312,11 +301,18 @@ impl<E: ComposeDeployEnv> DeploymentOrchestrator<E> {
         })
     }
 
-    fn managed_cluster_wait(&self, project_name: String) -> Arc<dyn ClusterWaitHandle<E>> {
-        Arc::new(ComposeAttachedClusterWait::<E>::new(
-            compose_runner_host(),
-            ExistingCluster::compose_in_project(Vec::new(), project_name),
-        ))
+    fn managed_cluster_wait(
+        &self,
+        metadata: ComposeDeploymentMetadata,
+    ) -> Result<Arc<dyn ClusterWaitHandle<E>>, ComposeRunnerError> {
+        let existing_cluster = metadata
+            .existing_cluster()
+            .map_err(|source| ComposeRunnerError::SourceOrchestration { source })?;
+        let cluster_wait =
+            ComposeAttachedClusterWait::<E>::try_new(compose_runner_host(), &existing_cluster)
+                .map_err(|source| ComposeRunnerError::SourceOrchestration { source })?;
+
+        Ok(Arc::new(cluster_wait))
     }
 
     fn log_deploy_start<Caps>(
@@ -372,12 +368,7 @@ where
     E: ComposeDeployEnv,
     Caps: Send + Sync,
 {
-    let project_name = scenario
-        .existing_cluster()
-        .and_then(|attach| attach.compose_project())
-        .map(ToOwned::to_owned);
-
-    ComposeDeploymentMetadata { project_name }
+    ComposeDeploymentMetadata::from_existing_cluster(scenario.existing_cluster())
 }
 
 struct DeployedNodes<E: ComposeDeployEnv> {
diff --git a/testing-framework/deployers/compose/src/docker/control.rs b/testing-framework/deployers/compose/src/docker/control.rs
index 8e98948..90b8a3d 100644
--- a/testing-framework/deployers/compose/src/docker/control.rs
+++ b/testing-framework/deployers/compose/src/docker/control.rs
@@ -5,7 +5,7 @@ use std::{
 
 use testing_framework_core::{
     adjust_timeout,
-    scenario::{Application, DynError, NodeControlHandle},
+    scenario::{Application, DynError, ExistingCluster, NodeControlHandle},
 };
 use tokio::{process::Command, time::timeout};
 use tracing::info;
@@ -160,6 +160,22 @@ pub struct ComposeAttachedNodeControl {
     pub(crate) project_name: String,
 }
 
+impl ComposeAttachedNodeControl {
+    pub fn try_from_existing_cluster(source: &ExistingCluster) -> Result<Self, DynError> {
+        let Some(project_name) = source
+            .compose_project()
+            .map(str::trim)
+            .filter(|value| !value.is_empty())
+        else {
+            return Err("attached compose node control requires explicit project name".into());
+        };
+
+        Ok(Self {
+            project_name: project_name.to_owned(),
+        })
+    }
+}
+
 #[async_trait::async_trait]
 impl<E: Application> NodeControlHandle<E> for ComposeAttachedNodeControl {
     async fn restart_node(&self, name: &str) -> Result<(), DynError> {
diff --git a/testing-framework/deployers/k8s/src/deployer/attach_provider.rs b/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
index cfa0dc2..43925bc 100644
--- a/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
+++ b/testing-framework/deployers/k8s/src/deployer/attach_provider.rs
@@ -56,12 +56,14 @@ impl<E: K8sDeployEnv> K8sAttachProvider<E> {
 }
 
 impl<E: K8sDeployEnv> K8sAttachedClusterWait<E> {
-    pub(super) fn new(client: Client, source: ExistingCluster) -> Self {
-        Self {
+    pub(super) fn try_new(client: Client, source: &ExistingCluster) -> Result<Self, DynError> {
+        let _ = k8s_wait_request(source)?;
+
+        Ok(Self {
             client,
-            source,
+            source: source.clone(),
             _env: PhantomData,
-        }
+        })
     }
 }
 
diff --git a/testing-framework/deployers/k8s/src/deployer/mod.rs b/testing-framework/deployers/k8s/src/deployer/mod.rs
index eae7507..e20fcc5 100644
--- a/testing-framework/deployers/k8s/src/deployer/mod.rs
+++ b/testing-framework/deployers/k8s/src/deployer/mod.rs
@@ -22,6 +22,18 @@ enum K8sMetadataError {
 }
 
 impl K8sDeploymentMetadata {
+    #[must_use]
+    pub fn from_existing_cluster(cluster: Option<&ExistingCluster>) -> Self {
+        Self {
+            namespace: cluster
+                .and_then(ExistingCluster::k8s_namespace)
+                .map(ToOwned::to_owned),
+            label_selector: cluster
+                .and_then(ExistingCluster::k8s_label_selector)
+                .map(ToOwned::to_owned),
+        }
+    }
+
     /// Returns namespace when deployment is bound to a specific namespace.
     #[must_use]
     pub fn namespace(&self) -> Option<&str> {
diff --git a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
index 836a892..5fa21fe 100644
--- a/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
+++ b/testing-framework/deployers/k8s/src/deployer/orchestrator.rs
@@ -249,19 +249,7 @@ where
     E: K8sDeployEnv,
     Caps: Send + Sync,
 {
-    let namespace = scenario
-        .existing_cluster()
-        .and_then(|attach| attach.k8s_namespace())
-        .map(ToOwned::to_owned);
-    let label_selector = scenario
-        .existing_cluster()
-        .and_then(|attach| attach.k8s_label_selector())
-        .map(ToOwned::to_owned);
-
-    K8sDeploymentMetadata {
-        namespace,
-        label_selector,
-    }
+    K8sDeploymentMetadata::from_existing_cluster(scenario.existing_cluster())
 }
 
 fn attached_cluster_wait<E, Caps>(
@@ -277,11 +265,10 @@ where
         .ok_or_else(|| K8sRunnerError::InternalInvariant {
             message: "k8s attached cluster wait requested outside attached source mode".to_owned(),
         })?;
+    let cluster_wait = K8sAttachedClusterWait::<E>::try_new(client, attach)
+        .map_err(|source| K8sRunnerError::SourceOrchestration { source })?;
 
-    Ok(Arc::new(K8sAttachedClusterWait::<E>::new(
-        client,
-        attach.clone(),
-    )))
+    Ok(Arc::new(cluster_wait))
 }
 
 fn managed_cluster_wait<E: K8sDeployEnv>(
@@ -292,11 +279,10 @@ fn managed_cluster_wait<E: K8sDeployEnv>(
     let attach_source = metadata
         .existing_cluster()
         .map_err(|source| K8sRunnerError::SourceOrchestration { source })?;
+    let cluster_wait = K8sAttachedClusterWait::<E>::try_new(client, &attach_source)
+        .map_err(|source| K8sRunnerError::SourceOrchestration { source })?;
 
-    Ok(Arc::new(K8sAttachedClusterWait::<E>::new(
-        client,
-        attach_source,
-    )))
+    Ok(Arc::new(cluster_wait))
 }
 
 fn client_from_cluster(cluster: &Option<ClusterEnvironment>) -> Result<Client, K8sRunnerError> {

From ad288e742162959ea440ba5a794695b933f6505e Mon Sep 17 00:00:00 2001
From: andrussal <salumets.andrus@gmail.com>
Date: Sun, 8 Mar 2026 14:22:09 +0100
Subject: [PATCH 9/9] Add semantic existing-cluster constructors

---
 .../orchestration/source_orchestration_plan.rs   |  7 +++++--
 .../core/src/scenario/sources/model.rs           | 16 ++++++----------
 .../deployers/compose/src/deployer/mod.rs        |  7 +++----
 .../deployers/k8s/src/deployer/mod.rs            |  4 ++--
 4 files changed, 16 insertions(+), 18 deletions(-)

diff --git a/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs b/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
index c9c99f9..2c3cb4e 100644
--- a/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
+++ b/testing-framework/core/src/scenario/runtime/orchestration/source_orchestration_plan.rs
@@ -73,8 +73,11 @@ mod tests {
 
     #[test]
     fn attached_sources_are_planned() {
-        let sources = ScenarioSources::default()
-            .with_attach(ExistingCluster::compose(vec!["node-0".to_string()]));
+        let sources =
+            ScenarioSources::default().with_attach(ExistingCluster::for_compose_services(
+                "test-project".to_string(),
+                vec!["node-0".to_string()],
+            ));
         let plan = SourceOrchestrationPlan::try_from_sources(&sources)
             .expect("attached sources should build a source orchestration plan");
 
diff --git a/testing-framework/core/src/scenario/sources/model.rs b/testing-framework/core/src/scenario/sources/model.rs
index 566e2ee..a64b010 100644
--- a/testing-framework/core/src/scenario/sources/model.rs
+++ b/testing-framework/core/src/scenario/sources/model.rs
@@ -18,8 +18,7 @@ enum ExistingClusterKind {
 
 impl ExistingCluster {
     #[must_use]
-    #[doc(hidden)]
-    pub fn k8s(label_selector: String) -> Self {
+    pub fn for_k8s_selector(label_selector: String) -> Self {
         Self {
             kind: ExistingClusterKind::K8s {
                 namespace: None,
@@ -29,8 +28,7 @@ impl ExistingCluster {
     }
 
     #[must_use]
-    #[doc(hidden)]
-    pub fn k8s_in_namespace(label_selector: String, namespace: String) -> Self {
+    pub fn for_k8s_selector_in_namespace(namespace: String, label_selector: String) -> Self {
         Self {
             kind: ExistingClusterKind::K8s {
                 namespace: Some(namespace),
@@ -40,19 +38,17 @@ impl ExistingCluster {
     }
 
     #[must_use]
-    #[doc(hidden)]
-    pub fn compose(services: Vec<String>) -> Self {
+    pub fn for_compose_project(project: String) -> Self {
         Self {
             kind: ExistingClusterKind::Compose {
-                project: None,
-                services,
+                project: Some(project),
+                services: Vec::new(),
             },
         }
     }
 
     #[must_use]
-    #[doc(hidden)]
-    pub fn compose_in_project(services: Vec<String>, project: String) -> Self {
+    pub fn for_compose_services(project: String, services: Vec<String>) -> Self {
         Self {
             kind: ExistingClusterKind::Compose {
                 project: Some(project),
diff --git a/testing-framework/deployers/compose/src/deployer/mod.rs b/testing-framework/deployers/compose/src/deployer/mod.rs
index 2eab7c6..a0c35d5 100644
--- a/testing-framework/deployers/compose/src/deployer/mod.rs
+++ b/testing-framework/deployers/compose/src/deployer/mod.rs
@@ -66,8 +66,7 @@ impl ComposeDeploymentMetadata {
             .project_name()
             .ok_or(ComposeMetadataError::MissingProjectName)?;
 
-        Ok(ExistingCluster::compose_in_project(
-            Vec::new(),
+        Ok(ExistingCluster::for_compose_project(
             project_name.to_owned(),
         ))
     }
@@ -81,9 +80,9 @@ impl ComposeDeploymentMetadata {
             .project_name()
             .ok_or(ComposeMetadataError::MissingProjectName)?;
 
-        Ok(ExistingCluster::compose_in_project(
-            services,
+        Ok(ExistingCluster::for_compose_services(
             project_name.to_owned(),
+            services,
         ))
     }
 
diff --git a/testing-framework/deployers/k8s/src/deployer/mod.rs b/testing-framework/deployers/k8s/src/deployer/mod.rs
index e20fcc5..d97615b 100644
--- a/testing-framework/deployers/k8s/src/deployer/mod.rs
+++ b/testing-framework/deployers/k8s/src/deployer/mod.rs
@@ -53,9 +53,9 @@ impl K8sDeploymentMetadata {
             .label_selector()
             .ok_or(K8sMetadataError::MissingLabelSelector)?;
 
-        Ok(ExistingCluster::k8s_in_namespace(
-            label_selector.to_owned(),
+        Ok(ExistingCluster::for_k8s_selector_in_namespace(
             namespace.to_owned(),
+            label_selector.to_owned(),
         ))
     }