diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs
index dc356a00..4329695f 100644
--- a/control_plane/src/backend_client.rs
+++ b/control_plane/src/backend_client.rs
@@ -268,12 +268,16 @@ impl BackendClient {
     }
 
     /// Publish one authoritative catalog generation and all plans that reference it.
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(plan_id = publication.transmission_plan.envelope.plan_id,
+            plan_version = publication.transmission_plan.envelope.plan_version))]
     pub async fn post_catalog_plan_typed(
         &self,
         publication: &crate::physical::publication::PhysicalPlanPublication,
         storage_routing: Option<serde_json::Value>,
         adaptation_evidence: &[crate::physical::compiler::RuntimeAdaptationEvidence],
     ) -> std::result::Result<(), BackendPostError> {
+        tracing::debug!(target: "asap_runtime_debug", "backend plan stage request started");
         let body = publication
             .install_request(storage_routing, adaptation_evidence.to_vec())
             .map_err(|error| BackendPostError::Permanent(anyhow::anyhow!(error)))?;
@@ -285,6 +289,8 @@ impl BackendClient {
             .await
             .map_err(classify_reqwest_error)?;
         let status = response.status();
+        tracing::debug!(target: "asap_runtime_debug", http_status = %status,
+            "backend plan stage response received");
         if status.is_success() {
             Ok(())
         } else {
@@ -297,11 +303,18 @@ impl BackendClient {
         }
     }
 
+    #[tracing::instrument(
+        level = "debug",
+        target = "asap_runtime_debug",
+        skip_all,
+        fields(plan_id, plan_version)
+    )]
     pub async fn discard_staged_physical_plan(
         &self,
         plan_id: u64,
         plan_version: u64,
     ) -> std::result::Result<(), BackendPostError> {
+        tracing::debug!(target: "asap_runtime_debug", "staged backend plan cleanup requested");
         let response = self
             .http
             .post(format!(
@@ -324,11 +337,18 @@ impl BackendClient {
         }
     }
 
+    #[tracing::instrument(
+        level = "debug",
+        target = "asap_runtime_debug",
+        skip_all,
+        fields(plan_id, plan_version)
+    )]
     pub async fn activate_physical_plan(
         &self,
         plan_id: u64,
         plan_version: u64,
     ) -> std::result::Result<(), BackendPostError> {
+        tracing::debug!(target: "asap_runtime_debug", "backend plan activation request started");
         let url = format!("{}/activate", derive_physical_plan_url(&self.endpoint));
         let response = self
             .http
@@ -341,6 +361,8 @@ impl BackendClient {
             .await
             .map_err(classify_reqwest_error)?;
         let status = response.status();
+        tracing::debug!(target: "asap_runtime_debug", http_status = %status,
+            "backend plan activation response received");
         if status.is_success() {
             Ok(())
         } else {
diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs
index 41dd1505..e8c3cdd8 100644
--- a/control_plane/src/main.rs
+++ b/control_plane/src/main.rs
@@ -18,10 +18,13 @@ use axum::{
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use tracing::info;
 
+static NEXT_PLAN_CALL_ID: AtomicU64 = AtomicU64::new(1);
+
 use opamp::OpampServer;
 use physical::deployment_cost::online as online_cost_model;
 use physical::deployment_cost::online::{init_store as init_online_store, OnlineMetricsStore};
@@ -50,7 +53,16 @@ struct AppState {
 
 #[tokio::main]
 async fn main() {
-    tracing_subscriber::fmt::init();
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        .with_file(true)
+        .with_line_number(true)
+        .with_target(true)
+        .with_writer(std::io::stdout)
+        .init();
 
     let api_addr = std::env::var("CONTROLLER_ADDR").unwrap_or_else(|_| "0.0.0.0:8080".into());
     let opamp_addr =
@@ -266,11 +278,16 @@ async fn compile_and_publish_physical_plan(
     mut request: CompileAndPublishPhysicalPlanRequest,
     frontend: QueryFrontend,
 ) -> Response {
+    let call_id = NEXT_PLAN_CALL_ID.fetch_add(1, Ordering::Relaxed);
+    let started = std::time::Instant::now();
+    tracing::debug!(target: "asap_runtime_debug", call_id, ?frontend,
+        "physical plan compilation requested");
     // Serialize typed activations so an older response cannot overwrite the
     // catalog recorded after a newer backend activation.
     let mut active_catalog = st.active_summary_catalog.lock().await;
     if let Some(erp) = &mut request.erp {
         if let Err(error) = erp.hydrate_observed_shape(&st.runtime_samples) {
+            tracing::warn!(call_id, %error, "ERP observation hydration failed");
             return (StatusCode::UNPROCESSABLE_ENTITY, error).into_response();
         }
         let catalog = active_catalog.clone();
@@ -282,16 +299,37 @@ async fn compile_and_publish_physical_plan(
                 (bundle, ids, timeout, adaptation, manifests)
             }
             Ok((None, ..)) => {
+                tracing::error!(
+                    call_id,
+                    "physical plan compilation produced no deployable plan"
+                );
                 return (
                     StatusCode::INTERNAL_SERVER_ERROR,
                     "publication requires a selected plan",
                 )
-                    .into_response()
+                    .into_response();
+            }
+            Err(response) => {
+                tracing::warn!(call_id, status = %response.0, error = %response.1,
+                    "physical plan compilation failed");
+                return physical_compile_failure(response);
             }
-            Err(response) => return physical_compile_failure(response),
         };
+    info!(
+        call_id,
+        plan_id = bundle.envelope.plan_id,
+        plan_version = bundle.envelope.plan_version,
+        collector_count = bundle.collector_plans.len(),
+        "physical plan compiled"
+    );
 
     let Some(backend) = st.backend_client.as_ref() else {
+        tracing::warn!(
+            call_id,
+            plan_id = bundle.envelope.plan_id,
+            plan_version = bundle.envelope.plan_version,
+            "backend endpoint is not configured"
+        );
         return (
             StatusCode::SERVICE_UNAVAILABLE,
             "CONTROLLER_BACKEND_ENDPOINT is required for physical-plan publication".to_string(),
@@ -303,6 +341,8 @@ async fn compile_and_publish_physical_plan(
         .ensure_collector_plan_targets(&bundle.collector_plans, apply_timeout)
         .await
     {
+        tracing::warn!(call_id, plan_id = bundle.envelope.plan_id, plan_version = bundle.envelope.plan_version,
+            %error, "collector plan preflight failed");
         return (
             StatusCode::BAD_GATEWAY,
             format!("collector physical-plan preflight failed: {error}"),
@@ -312,11 +352,14 @@ async fn compile_and_publish_physical_plan(
     let publication = match bundle.to_publication_artifact() {
         Ok(publication) => publication,
         Err(error) => {
+            tracing::error!(call_id, plan_id = bundle.envelope.plan_id,
+                plan_version = bundle.envelope.plan_version, %error,
+                "catalog publication artifact failed");
             return (
                 StatusCode::INTERNAL_SERVER_ERROR,
                 format!("invalid catalog publication: {error}"),
             )
-                .into_response()
+                .into_response();
         }
     };
     if let Err(error) = backend
@@ -327,17 +370,27 @@ async fn compile_and_publish_physical_plan(
         )
         .await
     {
+        tracing::warn!(call_id, plan_id = bundle.envelope.plan_id, plan_version = bundle.envelope.plan_version,
+            %error, "backend plan staging failed");
         return (
             StatusCode::BAD_GATEWAY,
             format!("backend rejected physical plan: {error}"),
         )
             .into_response();
     }
+    info!(
+        call_id,
+        plan_id = bundle.envelope.plan_id,
+        plan_version = bundle.envelope.plan_version,
+        "backend physical plan staged"
+    );
     if let Err(error) = st
         .opamp
         .publish_collector_plans(&bundle.collector_plans, apply_timeout)
         .await
     {
+        tracing::warn!(call_id, plan_id = bundle.envelope.plan_id, plan_version = bundle.envelope.plan_version,
+            %error, "collector plan publication failed");
         let cleanup = backend
             .discard_staged_physical_plan(bundle.envelope.plan_id, bundle.envelope.plan_version)
             .await;
@@ -347,12 +400,26 @@ async fn compile_and_publish_physical_plan(
         )
             .into_response();
     }
+    info!(
+        call_id,
+        plan_id = bundle.envelope.plan_id,
+        plan_version = bundle.envelope.plan_version,
+        "collector plans published"
+    );
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .unwrap_or_default()
         .as_millis() as u64;
     let activation_wait = bundle.envelope.activation_unix_ms.saturating_sub(now);
     if activation_wait > apply_timeout.as_millis() as u64 {
+        tracing::warn!(
+            call_id,
+            plan_id = bundle.envelope.plan_id,
+            plan_version = bundle.envelope.plan_version,
+            activation_wait_ms = activation_wait,
+            timeout_ms = apply_timeout.as_millis() as u64,
+            "scheduled activation exceeds apply timeout; backend plan remains staged"
+        );
         return (
             StatusCode::GATEWAY_TIMEOUT,
             "activation time exceeds apply_timeout_ms; backend remains staged".to_string(),
@@ -366,6 +433,8 @@ async fn compile_and_publish_physical_plan(
         .activate_physical_plan(bundle.envelope.plan_id, bundle.envelope.plan_version)
         .await
     {
+        tracing::warn!(call_id, plan_id = bundle.envelope.plan_id, plan_version = bundle.envelope.plan_version,
+            %error, "backend plan activation failed");
         return (
             StatusCode::BAD_GATEWAY,
             format!("backend physical-plan activation failed: {error}"),
@@ -374,6 +443,13 @@ async fn compile_and_publish_physical_plan(
     }
 
     *active_catalog = Some(Arc::new(bundle.summary_catalog));
+    info!(
+        call_id,
+        plan_id = bundle.envelope.plan_id,
+        plan_version = bundle.envelope.plan_version,
+        elapsed_ms = started.elapsed().as_millis() as u64,
+        "physical plan active"
+    );
 
     Json(CompileAndPublishPhysicalPlanResponse {
         cost_comparison: bundle.cost_comparison,
diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 390934cc..d73dfa98 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -381,11 +381,15 @@ pub fn gos_policy_from_accuracy_budget(
     })
 }
 
+#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+    fields(plan_id = envelope.plan_id, plan_version = envelope.plan_version,
+        producer_count = precompute.producers.len()))]
 pub fn build_transmission_plan(
     envelope: PlanEnvelope,
     precompute: &PrecomputePlan,
     runtime_policies: &BTreeMap<asap_types::PolicyFingerprint, RuntimeRulePolicy>,
 ) -> Result<TransmissionPlan, TransmissionPlanError> {
+    tracing::debug!(target: "asap_runtime_debug", "transmission plan construction started");
     if envelope != precompute.envelope {
         return Err(TransmissionPlanError::EnvelopeMismatch);
     }
@@ -957,12 +961,16 @@ impl PhysicalPlanCompiler {
         self.compile_for_frontend(request, environment, QueryFrontend::MetricsQl)
     }
 
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(frontend = ?frontend, plan_version = environment.plan_version,
+            query_count = request.queries.len()))]
     pub fn compile_for_frontend(
         &self,
         mut request: PhysicalCompilationRequest,
         environment: PhysicalDeploymentContext,
         frontend: QueryFrontend,
     ) -> Result<CompiledPhysicalPlan, CompileError> {
+        tracing::debug!(target: "asap_runtime_debug", "physical plan compiler entered");
         if let Some(data) = &request.data_workload {
             data.validate()
                 .map_err(|error| CompileError::Snapshot(error.to_string()))?;
@@ -2158,6 +2166,8 @@ pub fn select_logical_roots_with_error_resource_profiles(
     select_logical_roots_with_trace(queries, roots, evidence, exact_costs, erp).map(|_| ())
 }
 
+#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+    fields(query_count = queries.len(), root_count = roots.len()))]
 pub fn select_logical_roots_with_trace(
     queries: &mut [QueryCompilationInput],
     roots: Vec<Rc<QueryExpr>>,
@@ -2165,6 +2175,7 @@ pub fn select_logical_roots_with_trace(
     exact_costs: &HashMap<String, Vec<ExactCompositionCostEvidence>>,
     erp: Option<&super::erp::ErpPlanningInput>,
 ) -> Result<Vec<serde_json::Value>, CompileError> {
+    tracing::debug!(target: "asap_runtime_debug", "logical root selection started");
     let mut traces = Vec::new();
     if roots.len() != queries.len() {
         return Err(CompileError::Snapshot(
diff --git a/crates/asap_types/src/query_plan.rs b/crates/asap_types/src/query_plan.rs
index 51204ce0..d0fbaa42 100644
--- a/crates/asap_types/src/query_plan.rs
+++ b/crates/asap_types/src/query_plan.rs
@@ -629,6 +629,138 @@ pub enum QueryPlanNode {
 }
 
 impl QueryPlanNode {
+    /// Operator label for logs: the serialized `op` tag, plus the residual
+    /// `kind` for logical nodes, including the operation where applicable
+    /// (e.g. `logical/aggregate/sum`).
+    pub fn op_label(&self) -> &'static str {
+        use residual::ResidualQueryOperator as R;
+        match self {
+            Self::RelationalJoin { .. } => "relational_join",
+            Self::Relational { .. } => "relational",
+            Self::Logical { operator, .. } => match operator {
+                R::CurrentSeries { .. } => "logical/current_series",
+                R::ExactSubquery { .. } => "logical/exact_subquery",
+                R::CandidateExactSubquery { .. } => "logical/candidate_exact_subquery",
+                R::Scan { .. } => "logical/scan",
+                R::UnaryNegate => "logical/unary_negate",
+                R::VectorToScalar => "logical/vector_to_scalar",
+                R::Aggregate { operation, .. } => match operation {
+                    residual::Aggregation::Sum => "logical/aggregate/sum",
+                    residual::Aggregation::Max => "logical/aggregate/max",
+                    residual::Aggregation::Min => "logical/aggregate/min",
+                    residual::Aggregation::Avg => "logical/aggregate/avg",
+                    residual::Aggregation::Count => "logical/aggregate/count",
+                },
+                R::Limit { .. } => "logical/limit",
+                R::Binary { .. } => "logical/binary",
+                R::Temporal { .. } => "logical/temporal",
+                R::Sort { .. } => "logical/sort",
+                R::HistogramQuantile => "logical/histogram_quantile",
+                R::Subquery { .. } => "logical/subquery",
+            },
+            Self::Scalar { .. } => "scalar",
+            Self::Binary { .. } => "binary",
+            Self::ReduceSum { .. } => "reduce_sum",
+            Self::ReadMaterialization { .. } => "read_materialization",
+            Self::SummaryEstimate { .. } => "summary_estimate",
+            Self::ExactReadout { readout, .. } => match readout {
+                ExactReadout::Sum => "exact_readout/sum",
+                ExactReadout::Count => "exact_readout/count",
+                ExactReadout::Increase => "exact_readout/increase",
+                ExactReadout::Rate => "exact_readout/rate",
+                ExactReadout::Min => "exact_readout/min",
+                ExactReadout::Max => "exact_readout/max",
+            },
+            Self::SummaryMerge { .. } => "summary_merge",
+            Self::ExternalExact { .. } => "external_exact",
+            Self::ExactFallback { .. } => "exact_fallback",
+        }
+    }
+
+    /// Bounded, query-text-free operator arguments for execution logs.
+    /// The query ID links these details to the full installed plan when needed.
+    pub fn log_syntax(&self) -> String {
+        use residual::ResidualQueryOperator as R;
+        match self {
+            Self::RelationalJoin { join_kind, .. } => format!("join_kind={join_kind:?}"),
+            Self::Relational { .. } => String::new(),
+            Self::Logical { operator, .. } => match operator {
+                R::CurrentSeries { readout, .. } => format!("readout={readout:?}"),
+                R::ExactSubquery { .. } => String::new(),
+                R::CandidateExactSubquery { .. } => String::new(),
+                R::Scan {
+                    metric,
+                    matchers,
+                    range_ms,
+                    offset_ms,
+                } => format!(
+                    "metric={} matcher_count={} range_ms={range_ms:?} offset_ms={offset_ms}",
+                    metric
+                        .as_deref()
+                        .map(|name| name.chars().take(64).collect::<String>())
+                        .unwrap_or_default(),
+                    matchers.len(),
+                ),
+                R::UnaryNegate | R::VectorToScalar | R::HistogramQuantile => String::new(),
+                R::Aggregate {
+                    operation,
+                    grouping,
+                } => format!(
+                    "operation={operation:?} grouping={}",
+                    log_grouping(&grouping.labels, grouping.without)
+                ),
+                R::Limit {
+                    n,
+                    offset,
+                    grouping,
+                } => format!(
+                    "n={n} offset={offset} grouping={}",
+                    log_grouping(&grouping.labels, grouping.without)
+                ),
+                R::Binary {
+                    operation,
+                    return_bool,
+                } => format!("operation={operation:?} return_bool={return_bool}"),
+                R::Temporal { operation } => format!("operation={operation:?}"),
+                R::Sort {
+                    descending,
+                    grouping,
+                } => format!(
+                    "descending={descending} grouping={}",
+                    log_grouping(&grouping.labels, grouping.without)
+                ),
+                R::Subquery {
+                    range_ms,
+                    step_ms,
+                    offset_ms,
+                } => format!("range_ms={range_ms} step_ms={step_ms} offset_ms={offset_ms}"),
+            },
+            Self::Scalar { value } => format!("value={value}"),
+            Self::Binary { operator, .. } => format!("operation={operator:?}"),
+            Self::ReduceSum { grouping, .. } => match grouping {
+                PhysicalGrouping::PerEntity => "grouping=per_entity".into(),
+                PhysicalGrouping::Reduce(labels) => {
+                    format!("grouping=reduce({})", log_labels(labels))
+                }
+            },
+            Self::ReadMaterialization { binding } => format!(
+                "window_ms={} lookback_ms={:?}",
+                binding.window_ms, binding.readout_lookback_ms
+            ),
+            Self::SummaryEstimate { query, .. } => match query {
+                QueryReadout::FrequencyL2 => "readout=frequency_l2".into(),
+                QueryReadout::FrequencyEntropy => "readout=frequency_entropy".into(),
+                QueryReadout::Quantile { q } => format!("readout=quantile q={q}"),
+                QueryReadout::PointCount { .. } => "readout=point_count".into(),
+                QueryReadout::Cardinality => "readout=cardinality".into(),
+                QueryReadout::TopK { k } => format!("readout=top_k k={k}"),
+            },
+            Self::ExactReadout { readout, .. } => format!("readout={readout:?}"),
+            Self::SummaryMerge { .. } => String::new(),
+            Self::ExternalExact { .. } | Self::ExactFallback { .. } => String::new(),
+        }
+    }
+
     pub fn inputs(&self) -> &[QueryNodeId] {
         match self {
             Self::Scalar { .. } | Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => {
@@ -646,6 +778,26 @@ impl QueryPlanNode {
     }
 }
 
+fn log_labels(labels: &[String]) -> String {
+    let mut names = labels
+        .iter()
+        .take(8)
+        .map(|label| label.chars().take(64).collect::<String>())
+        .collect::<Vec<_>>();
+    if labels.len() > 8 {
+        names.push("...".into());
+    }
+    names.join(",")
+}
+
+fn log_grouping(labels: &[String], without: bool) -> String {
+    format!(
+        "{}({})",
+        if without { "without" } else { "by" },
+        log_labels(labels)
+    )
+}
+
 pub use planner_types::post_asap::CandidateCompleteness;
 
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
@@ -739,6 +891,8 @@ pub fn canonical_promql(query: &str) -> Result<String, QueryPlanError> {
 
 #[cfg(test)]
 mod contract_tests {
+    use super::{residual, QueryPlanNode};
+
     // Installed plans cross producer/query threads without Planner Rc state.
     #[test]
     fn installed_query_contract_is_send_sync() {
@@ -746,6 +900,66 @@ mod contract_tests {
         assert_send_sync::<super::QueryPlan>();
         assert_send_sync::<super::QueryPlanEntry>();
     }
+
+    #[test]
+    fn aggregate_log_labels_identify_the_operation() {
+        for (operation, expected) in [
+            (residual::Aggregation::Sum, "logical/aggregate/sum"),
+            (residual::Aggregation::Count, "logical/aggregate/count"),
+            (residual::Aggregation::Avg, "logical/aggregate/avg"),
+        ] {
+            let node = QueryPlanNode::Logical {
+                operator: residual::ResidualQueryOperator::Aggregate {
+                    operation,
+                    grouping: residual::Grouping {
+                        labels: vec!["service".into()],
+                        without: false,
+                    },
+                },
+                inputs: vec![],
+            };
+            assert_eq!(node.op_label(), expected);
+            assert!(node.log_syntax().contains("grouping=by(service)"));
+        }
+        for (readout, expected) in [
+            (super::ExactReadout::Sum, "exact_readout/sum"),
+            (super::ExactReadout::Count, "exact_readout/count"),
+        ] {
+            assert_eq!(
+                QueryPlanNode::ExactReadout {
+                    input: super::QueryNodeId(1),
+                    readout,
+                }
+                .op_label(),
+                expected
+            );
+        }
+    }
+
+    #[test]
+    fn execution_log_syntax_identifies_operator_without_query_text() {
+        let binary = QueryPlanNode::Logical {
+            operator: residual::ResidualQueryOperator::Binary {
+                operation: residual::BinaryOperation::CheckedDiv,
+                return_bool: false,
+            },
+            inputs: vec![super::QueryNodeId(1), super::QueryNodeId(2)],
+        };
+        assert_eq!(binary.op_label(), "logical/binary");
+        assert_eq!(
+            binary.log_syntax(),
+            "operation=CheckedDiv return_bool=false"
+        );
+
+        let exact = QueryPlanNode::Logical {
+            operator: residual::ResidualQueryOperator::ExactSubquery {
+                query: "secret_metric{credential=\"secret\"}".into(),
+            },
+            inputs: vec![],
+        };
+        assert_eq!(exact.op_label(), "logical/exact_subquery");
+        assert!(exact.log_syntax().is_empty());
+    }
 }
 
 /// Bind portable relation semantics before an installed plan can access its sources.
diff --git a/data_plane/docker-compose.yml.j2 b/data_plane/docker-compose.yml.j2
index cacf95ef..7a650240 100644
--- a/data_plane/docker-compose.yml.j2
+++ b/data_plane/docker-compose.yml.j2
@@ -27,7 +27,6 @@ services:
       "--prometheus-server", "http://{{ prometheus_host }}:{{ prometheus_port }}",
       "--prometheus-scrape-interval", "{{ prometheus_scrape_interval }}",
       "--delete-existing-db",
-      "--log-level", "{{ log_level }}",
       "--output-dir", "/app/outputs",
       "--streaming-engine", "{{ streaming_engine }}",
       "--query-language", "{{ query_language }}",
diff --git a/data_plane/query-engine-rust-cli-compose.yml.j2 b/data_plane/query-engine-rust-cli-compose.yml.j2
index 32fe8f96..d54bb112 100644
--- a/data_plane/query-engine-rust-cli-compose.yml.j2
+++ b/data_plane/query-engine-rust-cli-compose.yml.j2
@@ -25,7 +25,6 @@ services:
       "--prometheus-server", "http://prometheus:9090",
       "--prometheus-scrape-interval", "{{ prometheus_scrape_interval }}",
       "--delete-existing-db",
-      "--log-level", "{{ log_level }}",
       "--output-dir", "/app/outputs",
       "--streaming-engine", "{{ streaming_engine }}",
       "--query-language", "{{ query_language }}",
diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs
index e4b31355..2a577bb4 100644
--- a/data_plane/src/drivers/query/servers/http.rs
+++ b/data_plane/src/drivers/query/servers/http.rs
@@ -9,11 +9,14 @@ use axum::{
 };
 use serde_json::Value;
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::net::TcpListener;
 use tracing::{debug, info, warn};
 
+static NEXT_PLAN_REQUEST_ID: AtomicU64 = AtomicU64::new(1);
+
 use crate::drivers::query::adapters::{AdapterConfig, HttpProtocolAdapter, PrometheusHttpAdapter};
 use crate::drivers::query::servers::metrics as srv_metrics;
 use crate::query_engines::routing::{
@@ -1986,7 +1989,7 @@ async fn handle_runtime_info(
     let mut forwarding_headers = HashMap::new();
     if let Some(auth) = headers.get(axum::http::header::AUTHORIZATION) {
         if let Ok(auth_str) = auth.to_str() {
-            debug!("Found Authorization header for runtime info: {}", auth_str);
+            debug!("Found Authorization header for runtime info");
             forwarding_headers.insert("Authorization".to_string(), auth_str.to_string());
         }
     } else {
@@ -5294,10 +5297,14 @@ pub use asap_types::plan_publication::PhysicalPlanInstallRequest;
 
 /// Decode and cross-validate every backend view before it can become visible.
 /// Used by both startup artifact loading and the staged HTTP install path.
+#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+    fields(plan_id = request.transmission_plan.envelope.plan_id,
+        plan_version = request.transmission_plan.envelope.plan_version))]
 pub fn validate_and_build_runtime_plan(
     request: PhysicalPlanInstallRequest,
     default_routing: Arc<crate::storage_engines::types::BackendStorageRouting>,
 ) -> Result<crate::storage_engines::types::RuntimePhysicalPlan, String> {
+    tracing::debug!(target: "asap_runtime_debug", "backend runtime plan validation started");
     use std::collections::BTreeSet;
     request
         .precompute_plan
@@ -5380,8 +5387,25 @@ async fn handle_post_physical_plan(
 ) -> axum::response::Response {
     use axum::http::StatusCode;
     use axum::response::IntoResponse;
+    let call_id = NEXT_PLAN_REQUEST_ID.fetch_add(1, Ordering::Relaxed);
+    let started = Instant::now();
+    let requested_plan_id = request.transmission_plan.envelope.plan_id;
+    let requested_plan_version = request.transmission_plan.envelope.plan_version;
+    tracing::debug!(target: "asap_runtime_debug",
+        call_id,
+        plan_id = requested_plan_id,
+        plan_version = requested_plan_version,
+        "physical plan staging requested"
+    );
 
     let Some(active_handle) = state.active_physical_plan.as_ref() else {
+        tracing::warn!(
+            call_id,
+            plan_id = requested_plan_id,
+            plan_version = requested_plan_version,
+            error = "physical-plan hot-reload handles are not attached",
+            "physical plan staging failed"
+        );
         return (
             StatusCode::SERVICE_UNAVAILABLE,
             axum::Json(serde_json::json!({
@@ -5391,6 +5415,13 @@ async fn handle_post_physical_plan(
             .into_response();
     };
     let Some(lifecycle) = state.physical_plan_lifecycle.as_ref() else {
+        tracing::warn!(
+            call_id,
+            plan_id = requested_plan_id,
+            plan_version = requested_plan_version,
+            error = "physical-plan lifecycle is not attached",
+            "physical plan staging failed"
+        );
         return (
             StatusCode::SERVICE_UNAVAILABLE,
             axum::Json(serde_json::json!({
@@ -5406,6 +5437,9 @@ async fn handle_post_physical_plan(
             &request.adaptation_evidence,
             unix_time_ms(),
         ) {
+            tracing::warn!(call_id, plan_id = requested_plan_id,
+                plan_version = requested_plan_version,
+                %error, "physical plan successor rejected");
             return (
                 StatusCode::UNPROCESSABLE_ENTITY,
                 axum::Json(serde_json::json!({
@@ -5420,6 +5454,9 @@ async fn handle_post_physical_plan(
     let active = match validate_and_build_runtime_plan(request, current.storage_routing.clone()) {
         Ok(active) => active,
         Err(error) => {
+            tracing::warn!(call_id, plan_id = requested_plan_id,
+                plan_version = requested_plan_version, %error,
+                "physical plan validation failed");
             return (
                 StatusCode::UNPROCESSABLE_ENTITY,
                 axum::Json(serde_json::json!({"status": "error", "error": error})),
@@ -5435,6 +5472,13 @@ async fn handle_post_physical_plan(
             )
             || active.precompute_plan.ingest.endpoint_path != "/api/v1/write")
     {
+        tracing::warn!(
+            call_id,
+            plan_id = requested_plan_id,
+            plan_version = requested_plan_version,
+            error = "Remote Write listener requires a non-bootstrap prometheus_remote_write_v1 plan at /api/v1/write",
+            "physical plan staging failed"
+        );
         return (
             StatusCode::UNPROCESSABLE_ENTITY,
             axum::Json(serde_json::json!({
@@ -5461,12 +5505,23 @@ async fn handle_post_physical_plan(
     let plan_version = active.plan_version();
     let now = unix_time_ms();
     if let Err(error) = lifecycle.stage(active, now) {
+        tracing::warn!(call_id, plan_id, plan_version, %error, "physical plan staging failed");
         return (
             StatusCode::CONFLICT,
             axum::Json(serde_json::json!({"status": "error", "error": error.to_string()})),
         )
             .into_response();
     }
+    tracing::info!(
+        call_id,
+        plan_id,
+        plan_version,
+        elapsed_ms = started.elapsed().as_millis() as u64,
+        materialization_count,
+        metricsql_query_count,
+        clickhouse_plan_count,
+        "physical plan staged"
+    );
     (
         StatusCode::ACCEPTED,
         axum::Json(serde_json::json!({
@@ -5491,10 +5546,25 @@ async fn handle_activate_physical_plan(
     axum::Json(request): axum::Json<ActivatePhysicalPlanRequest>,
 ) -> axum::response::Response {
     use axum::response::IntoResponse;
+    let call_id = NEXT_PLAN_REQUEST_ID.fetch_add(1, Ordering::Relaxed);
+    let started = Instant::now();
+    tracing::debug!(target: "asap_runtime_debug",
+        call_id,
+        plan_id = request.plan_id,
+        plan_version = request.plan_version,
+        "physical plan activation requested"
+    );
     let (Some(lifecycle), Some(active_handle)) = (
         state.physical_plan_lifecycle.as_ref(),
         state.active_physical_plan.as_ref(),
     ) else {
+        tracing::warn!(
+            call_id,
+            plan_id = request.plan_id,
+            plan_version = request.plan_version,
+            error = "physical-plan lifecycle is not attached",
+            "physical plan activation failed"
+        );
         return (
             StatusCode::SERVICE_UNAVAILABLE,
             axum::Json(serde_json::json!({
@@ -5527,16 +5597,27 @@ async fn handle_activate_physical_plan(
     ) {
         Ok(old) => old,
         Err(error) => {
+            tracing::warn!(call_id, plan_id = request.plan_id, plan_version = request.plan_version,
+                %error, "physical plan activation failed");
             return (
                 StatusCode::CONFLICT,
                 axum::Json(serde_json::json!({
                     "status": "error", "error": error.to_string()
                 })),
             )
-                .into_response()
+                .into_response();
         }
     };
     let activated = active_handle.active_snapshot();
+    tracing::info!(
+        call_id,
+        plan_id = request.plan_id,
+        plan_version = request.plan_version,
+        elapsed_ms = started.elapsed().as_millis() as u64,
+        previous_plan_id = old.plan_id(),
+        previous_plan_version = old.plan_version(),
+        "physical plan activated; summary catalog installed"
+    );
     let clickhouse_plan_count = activated
         .query_plan
         .entries
diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs
index 6e3d38bd..5125a89d 100644
--- a/data_plane/src/main.rs
+++ b/data_plane/src/main.rs
@@ -171,10 +171,6 @@ struct Args {
     #[arg(long, default_value = "/var/log/asap")]
     output_dir: String,
 
-    /// Log level
-    #[arg(long, default_value = "INFO")]
-    log_level: String,
-
     /// Enable profiling (currently unused, kept for compatibility)
     #[arg(long)]
     do_profiling: bool,
@@ -519,7 +515,7 @@ async fn main() -> Result<()> {
 
     // Initialize logging similar to Python's create_loggers function
     // Keep the guard alive for the entire lifetime of the application
-    let _log_guard = setup_logging(&args.output_dir, &args.log_level)?;
+    let _log_guard = setup_logging(&args.output_dir)?;
 
     info!("Starting Query Engine Rust");
     info!("Output directory: {}", args.output_dir);
@@ -1324,16 +1320,10 @@ async fn spawn_memory_diagnostics(
     }
 }
 
-fn setup_logging(
-    output_dir: &str,
-    log_level: &str,
-) -> Result<tracing_appender::non_blocking::WorkerGuard> {
+fn setup_logging(output_dir: &str) -> Result<tracing_appender::non_blocking::WorkerGuard> {
     use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 
-    // Create env filter that respects RUST_LOG, with fallback to command line arg
-    let env_filter = EnvFilter::try_from_default_env()
-        .or_else(|_| EnvFilter::try_new(log_level))
-        .unwrap_or_else(|_| EnvFilter::new("info"));
+    let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
 
     // Create file appender for logging to file
     let file_appender = tracing_appender::rolling::never(output_dir, "query_engine.log");
diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs
index 33782c92..7a80308b 100644
--- a/data_plane/src/precompute_engine/output_sink.rs
+++ b/data_plane/src/precompute_engine/output_sink.rs
@@ -5,7 +5,7 @@ use crate::storage_engines::types::hot_reload_config::InstalledPrecomputePlanHan
 use crate::storage_engines::types::{AggregateCore, PrecomputedOutput};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
-use tracing::{debug_span, warn};
+use tracing::{debug, debug_span, warn};
 
 /// CQ-6 — process-global fallback for the output-sink policy-miss
 /// counter, used when a `SketchStoreSink` was constructed without an
@@ -233,11 +233,15 @@ impl OutputSink for SketchStoreSink {
         if outputs.is_empty() {
             return Ok(());
         }
-        let _span = debug_span!("sketch_index_insert", batch_size = outputs.len()).entered();
+        let _span = debug_span!(target: "asap_runtime_debug", "sketch_index_insert",
+            batch_size = outputs.len())
+        .entered();
         let output_count = outputs.len();
         let failed = consume_in_order(outputs, |(output, accumulator)| {
             self.append_to_index(output, accumulator.as_ref())
         });
+        debug!(target: "asap_runtime_debug", output_count, failed,
+            "precompute output batch stored");
         if failed > 0 {
             return Err(format!(
                 "SketchStore rejected {failed} of {} completed outputs",
diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs
index 6085fcab..3e8c8387 100644
--- a/data_plane/src/precompute_engine/subdag_scheduler.rs
+++ b/data_plane/src/precompute_engine/subdag_scheduler.rs
@@ -82,6 +82,33 @@ where
     Ok(outputs.remove(0))
 }
 
+fn node_syntax(payload: &ExecutableOperatorPayload) -> String {
+    let details = match payload {
+        ExecutableOperatorPayload::Fallback { .. } => String::new(),
+        ExecutableOperatorPayload::Binary { operator } => {
+            format!("operator={operator:?}")
+        }
+        ExecutableOperatorPayload::Value { operation } => {
+            format!("operation={operation:?}")
+        }
+        ExecutableOperatorPayload::RelationalJoin { join_kind, .. } => {
+            format!("join_kind={join_kind:?}")
+        }
+        ExecutableOperatorPayload::SummaryAgg {
+            family,
+            input,
+            reduction,
+            ..
+        } => format!("family={family:?} input={input:?} reduction={reduction:?}"),
+        ExecutableOperatorPayload::SummaryJoin { family, .. } => format!("family={family:?}"),
+        ExecutableOperatorPayload::SummarySubtract => String::new(),
+        ExecutableOperatorPayload::SummaryDelete { .. } => String::new(),
+        ExecutableOperatorPayload::SummaryEstimate { query } => format!("readout={query:?}"),
+        ExecutableOperatorPayload::SummaryMerge => String::new(),
+    };
+    details.chars().take(256).collect()
+}
+
 /// Evaluate all selected stored outputs with one dependency cache. Keys must
 /// describe the same input revision and window; only their output identity may
 /// differ. Validation finishes before executing or committing any output.
@@ -270,10 +297,12 @@ where
                         }
                     };
                     if committed.contains(&node.0) {
+                        tracing::debug!(target: "asap_runtime_debug", sink_node_id = node.0, "precompute DAG reused committed sink");
                         Ok(value)
                     } else {
-                        sink.commit_if_absent(key.clone(), value)
-                            .map_err(ScheduleError::Sink)
+                        let value = sink.commit_if_absent(key.clone(), value).map_err(ScheduleError::Sink)?;
+                        tracing::debug!(target: "asap_runtime_debug", sink_node_id = node.0, "precompute DAG sink commit completed");
+                        Ok(value)
                     }
                 }
             }),
@@ -310,6 +339,7 @@ impl<V, R: PrecomputeOperatorRegistry<V>>
     ) -> Result<execution::OutputStream<'a, Arc<V>>, execution::Error> {
         Ok(futures::stream::once(async move {
             if let Some(source) = &self.source {
+                tracing::debug!(target: "asap_runtime_debug", node_id = self.node.id.0, "precompute node used materialized input");
                 return Ok(Arc::clone(source));
             }
             let inputs =
@@ -323,10 +353,16 @@ impl<V, R: PrecomputeOperatorRegistry<V>>
                 .iter()
                 .map(|value| Arc::clone(value.value()))
                 .collect::<Vec<_>>();
+            let started = std::time::Instant::now();
+            tracing::debug!(target: "asap_runtime_debug", node_id = self.node.id.0, phase = ?self.node.output_state.timing, syntax = %node_syntax(&self.node.payload), "precompute node started");
             self.registry
                 .execute(self.node, &values, context)
-                .map(Arc::new)
+                .map(|value| {
+                    tracing::debug!(target: "asap_runtime_debug", node_id = self.node.id.0, elapsed_us = started.elapsed().as_micros() as u64, "precompute node completed");
+                    Arc::new(value)
+                })
                 .map_err(|e| {
+                    tracing::warn!(node_id = self.node.id.0, elapsed_us = started.elapsed().as_micros() as u64, "precompute node failed");
                     *self.error.borrow_mut() = Some(e);
                     execution::Error::Operator(format!("ingestion node {} failed", self.node.id.0))
                 })
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index c23b9fd5..95747447 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -335,8 +335,11 @@ impl Worker {
                 } => {
                     let sample_count = samples.len();
                     let _span = debug_span!(
+                        target: "asap_runtime_debug",
                         "worker_process_group",
                         worker_id = self.id,
+                        plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id),
+                        plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version),
                         sid,
                         policy_fp = %policy_fp,
                         group = %group_key,
@@ -346,12 +349,13 @@ impl Worker {
                     if let Err(e) = self.process_group_samples(sid, policy_fp, &group_key, samples)
                     {
                         processing_error = Some(e.to_string());
-                        warn!(
-                            "Worker {} error processing sid={} (policy_fp={}, group={}): {}",
-                            self.id, sid, policy_fp, group_key, e
-                        );
+                        warn!(worker_id = self.id, sid, policy_fp = %policy_fp,
+                            plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id),
+                            plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version),
+                            error = %e, "worker group processing failed");
                     }
                     debug!(
+                        target: "asap_runtime_debug",
                         e2e_latency_us = ingest_received_at.elapsed().as_micros() as u64,
                         "e2e: ingest->worker complete"
                     );
@@ -362,17 +366,24 @@ impl Worker {
                     ingest_received_at,
                 } => {
                     let _span = debug_span!(
+                        target: "asap_runtime_debug",
                         "worker_process_raw",
                         worker_id = self.id,
+                        plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id),
+                        plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version),
                         series = %series_key,
                         sample_count = samples.len(),
                     )
                     .entered();
                     if let Err(e) = self.process_samples_raw(&series_key, samples) {
                         processing_error = Some(e.to_string());
-                        warn!("Worker {} raw error for {}: {}", self.id, series_key, e);
+                        warn!(worker_id = self.id, series = %series_key,
+                            plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id),
+                            plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version),
+                            error = %e, "worker raw processing failed");
                     }
                     debug!(
+                        target: "asap_runtime_debug",
                         e2e_latency_us = ingest_received_at.elapsed().as_micros() as u64,
                         "e2e: ingest->worker complete (raw)"
                     );
@@ -386,8 +397,11 @@ impl Worker {
                     ingest_received_at,
                 } => {
                     let _span = debug_span!(
+                        target: "asap_runtime_debug",
                         "worker_process_accumulator",
                         worker_id = self.id,
+                        plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id),
+                        plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version),
                         sid,
                         policy_fp = %policy_fp,
                         group = %group_key,
@@ -403,12 +417,13 @@ impl Worker {
                         accumulator,
                     ) {
                         processing_error = Some(e.to_string());
-                        warn!(
-                            "Worker {} accumulator input error for sid={} (policy_fp={}, group={}): {}",
-                            self.id, sid, policy_fp, group_key, e
-                        );
+                        warn!(worker_id = self.id, sid, policy_fp = %policy_fp,
+                            plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id),
+                            plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version),
+                            error = %e, "worker accumulator processing failed");
                     }
                     debug!(
+                        target: "asap_runtime_debug",
                         e2e_latency_us = ingest_received_at.elapsed().as_micros() as u64,
                         "e2e: ingest->worker complete (accumulator)"
                     );
@@ -528,6 +543,8 @@ impl Worker {
     /// `PrecomputeMaterialization` fingerprint used to resolve the bucket's
     /// config on first sight; `group_key` is held on the resulting
     /// `GroupState` for emit-time label rendering.
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(worker_id = self.id, sid, policy_fp = %policy_fp, sample_count = samples.len()))]
     pub fn process_group_samples(
         &mut self,
         sid: u64,
@@ -535,6 +552,7 @@ impl Worker {
         group_key: &Arc<GroupKey>,
         samples: Vec<(String, i64, f64)>, // (series_key, timestamp_ms, value)
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        debug!(target: "asap_runtime_debug", "worker group processing started");
         let input_revision = self.current_input_revision.clone();
         let worker_id = self.id;
         let allowed_lateness_ms = self.allowed_lateness_ms;
@@ -552,6 +570,8 @@ impl Worker {
             return Ok(());
         }
         let state = self.group_states.get_mut(&sid).unwrap();
+        tracing::debug!(target: "asap_runtime_debug", worker_id = self.id, policy_fp = %policy_fp,
+            "precompute update route resolved");
         #[cfg(not(test))]
         if state.program.is_none() {
             return Err("raw precompute requires an installed post-ASAP DAG producer".into());
@@ -851,6 +871,8 @@ impl Worker {
     ///
     /// `policy_fp` / `group_key` carry the same semantics as on
     /// `process_group_samples` — policy lookup + emit-time label rendering.
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(worker_id = self.id, sid, policy_fp = %policy_fp, timestamp_ms))]
     pub fn process_accumulator_input(
         &mut self,
         sid: u64,
@@ -859,6 +881,7 @@ impl Worker {
         timestamp_ms: i64,
         incoming: Box<dyn AggregateCore>,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        debug!(target: "asap_runtime_debug", "worker accumulator processing started");
         let worker_id = self.id;
         let allowed_lateness_ms = self.allowed_lateness_ms;
         let late_data_policy = self.late_data_policy;
@@ -877,6 +900,10 @@ impl Worker {
         let state = self.group_states.get_mut(&sid).unwrap();
 
         let previous_event_time = state.max_event_time_ms;
+        debug!(target: "asap_runtime_debug", aggregation_type = ?state.config.aggregation_type,
+            window_secs = state.config.window_size, slide_secs = state.config.slide_interval,
+            layout = ?state.config.window_layout,
+            "worker accumulator maintenance configuration selected");
         let current_event_time = if timestamp_ms > previous_event_time {
             timestamp_ms
         } else {
@@ -1021,11 +1048,14 @@ impl Worker {
     }
 
     /// Raw fast-path: emit each sample as a standalone `SumAccumulator`.
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(worker_id = self.id, sample_count = samples.len()))]
     pub fn process_samples_raw(
         &self,
         series_key: &str,
         samples: Vec<(i64, f64)>,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        debug!(target: "asap_runtime_debug", "worker raw processing started");
         let mut emit_batch: Vec<(PrecomputedOutput, Box<dyn AggregateCore>)> =
             Vec::with_capacity(samples.len());
 
@@ -1097,7 +1127,10 @@ impl Worker {
         }
     }
 
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(worker_id = self.id, group_count = self.group_states.len()))]
     fn flush_all(&mut self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        debug!(target: "asap_runtime_debug", "worker flush started");
         if self.pass_raw_samples {
             return Ok(());
         }
diff --git a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs
index 8b454a6b..c565845a 100644
--- a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs
+++ b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs
@@ -119,40 +119,48 @@ pub(crate) fn validate_payload(
     let mut resolved = BTreeMap::new();
     for id in entry.topological_order().map_err(|e| miss(e.to_string()))? {
         let node = &entry.nodes[&id];
-        let state_ids = match node {
-            QueryPlanNode::ReadMaterialization { binding } => {
-                if let std::collections::btree_map::Entry::Vacant(entry) =
-                    resolved.entry(binding.materialization)
-                {
-                    entry.insert(resolve(catalog, binding.materialization)?);
+        let state_ids = (|| -> Result<BTreeSet<SummaryDefinitionId>, EngineError> {
+            Ok(match node {
+                QueryPlanNode::ReadMaterialization { binding } => {
+                    if let std::collections::btree_map::Entry::Vacant(entry) =
+                        resolved.entry(binding.materialization)
+                    {
+                        entry.insert(resolve(catalog, binding.materialization)?);
+                    }
+                    BTreeSet::from([binding.materialization])
                 }
-                BTreeSet::from([binding.materialization])
-            }
-            QueryPlanNode::SummaryMerge { inputs } => {
-                let mut ids = BTreeSet::new();
-                for input in inputs {
-                    let children: &BTreeSet<SummaryDefinitionId> = states
-                        .get(input)
-                        .ok_or_else(|| miss("summary merge has no state input"))?;
-                    if children.is_empty() {
-                        return Err(miss("summary merge has value input"));
+                QueryPlanNode::SummaryMerge { inputs } => {
+                    let mut ids = BTreeSet::new();
+                    for input in inputs {
+                        let children: &BTreeSet<SummaryDefinitionId> = states
+                            .get(input)
+                            .ok_or_else(|| miss("summary merge has no state input"))?;
+                        if children.is_empty() {
+                            return Err(miss("summary merge has value input"));
+                        }
+                        ids.extend(children);
                     }
-                    ids.extend(children);
+                    ids
                 }
-                ids
-            }
-            QueryPlanNode::SummaryEstimate { input, .. }
-            | QueryPlanNode::ExactReadout { input, .. } => {
-                let ids: &BTreeSet<SummaryDefinitionId> = states
-                    .get(input)
-                    .ok_or_else(|| miss("readout has no state input"))?;
-                if ids.is_empty() || ids.iter().any(|id| !resolved[id].supports(node)) {
-                    return Err(miss("catalog descriptor cannot satisfy installed readout"));
+                QueryPlanNode::SummaryEstimate { input, .. }
+                | QueryPlanNode::ExactReadout { input, .. } => {
+                    let ids: &BTreeSet<SummaryDefinitionId> = states
+                        .get(input)
+                        .ok_or_else(|| miss("readout has no state input"))?;
+                    if ids.is_empty() || ids.iter().any(|id| !resolved[id].supports(node)) {
+                        return Err(miss("catalog descriptor cannot satisfy installed readout"));
+                    }
+                    BTreeSet::new()
                 }
-                BTreeSet::new()
-            }
-            _ => BTreeSet::new(),
-        };
+                _ => BTreeSet::new(),
+            })
+        })()
+        .map_err(|error| {
+            tracing::debug!(target: "asap_runtime_debug", query_id = %entry.query_id,
+                node_id = ?id, op = node.op_label(), %error,
+                "installed query node failed catalog validation");
+            error
+        })?;
         states.insert(id, state_ids);
     }
     Ok(())
diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs
index 528c4806..ae83f375 100644
--- a/data_plane/src/query_engines/asap_query_engine/engine.rs
+++ b/data_plane/src/query_engines/asap_query_engine/engine.rs
@@ -1,5 +1,40 @@
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
-use tracing::debug;
+use std::time::Instant;
+use tracing::{debug, Instrument};
+
+static NEXT_QUERY_CALL_ID: AtomicU64 = AtomicU64::new(1);
+
+fn query_call_id() -> u64 {
+    NEXT_QUERY_CALL_ID.fetch_add(1, Ordering::Relaxed)
+}
+
+fn log_query_outcome<T>(
+    call_id: u64,
+    operation: &'static str,
+    started: Instant,
+    remote_stats: Option<(usize, usize)>,
+    result: &Result<T, crate::query_engines::EngineError>,
+) {
+    match result {
+        Ok(_) => match remote_stats {
+            Some((remote_evaluations, remote_rpcs)) => debug!(target: "asap_runtime_debug",
+                call_id, operation, elapsed_ms = started.elapsed().as_millis() as u64,
+                remote_evaluations, remote_rpcs, "query call completed"),
+            None => debug!(target: "asap_runtime_debug",
+                call_id, operation, elapsed_ms = started.elapsed().as_millis() as u64,
+                "query call completed"),
+        },
+        Err(error @ crate::query_engines::EngineError::CapabilityMiss { .. }) => {
+            debug!(target: "asap_runtime_debug",
+            call_id, operation, elapsed_ms = started.elapsed().as_millis() as u64,
+            %error, "query call could not be served by ASAP tier")
+        }
+        Err(error) => tracing::warn!(call_id, operation,
+            elapsed_ms = started.elapsed().as_millis() as u64, %error,
+            "query call failed"),
+    }
+}
 
 use asap_types::query_requirements::QueryRequirements;
 use asap_types::KeyByLabelNames;
@@ -190,27 +225,49 @@ impl ASAPQueryEngine {
         now_ms: u64,
     ) -> Result<crate::query_engines::query_result::QueryResult, crate::query_engines::EngineError>
     {
-        let physical = self.active_physical_plan_snapshot().ok_or_else(|| {
-            crate::query_engines::EngineError::capability_miss(
-                "query_plan",
-                "no active physical plan",
-            )
-        })?;
-        let planned = physical
-            .query_plan
-            .lookup_canonical(asap_types::query_plan::QueryLanguage::MetricsQl, identity)
-            .map_err(|error| {
-                crate::query_engines::EngineError::capability_miss("query_plan", error.to_string())
+        let call_id = query_call_id();
+        let started = Instant::now();
+        let mut remote_stats = None;
+        debug!(target: "asap_runtime_debug",
+            call_id,
+            operation = "metricsql_instant",
+            evaluation_ms = now_ms,
+            "query call started"
+        );
+        let result = async {
+            let physical = self.active_physical_plan_snapshot().ok_or_else(|| {
+                crate::query_engines::EngineError::capability_miss(
+                    "query_plan",
+                    "no active physical plan",
+                )
             })?;
-        let leaves = self
-            .prepare_query_inputs(&physical, planned, &[now_ms])
-            .await?;
-        let (mut result, mut stats) =
-            self.execute_logical_entry(&physical, planned, &leaves, now_ms)?;
-        stats.remote_evaluations = leaves.values().map(|leaf| leaf.remote_evaluations).sum();
-        stats.remote_rpcs = leaves.values().map(|leaf| leaf.remote_rpcs).sum();
-        annotate_logical_execution(&mut result, &stats);
-        Ok(result)
+            let planned = physical
+                .query_plan
+                .lookup_canonical(asap_types::query_plan::QueryLanguage::MetricsQl, identity)
+                .map_err(|error| {
+                    crate::query_engines::EngineError::capability_miss(
+                        "query_plan",
+                        error.to_string(),
+                    )
+                })?;
+            debug!(target: "asap_runtime_debug", plan_id = physical.plan_id(), plan_version = physical.plan_version(),
+            query_id = %planned.query_id, evaluation_ms = now_ms,
+            "installed MetricsQL instant query selected");
+            let leaves = self
+                .prepare_query_inputs(&physical, planned, &[now_ms])
+                .await?;
+            let (mut result, mut stats) =
+                self.execute_logical_entry(&physical, planned, &leaves, now_ms)?;
+            stats.remote_evaluations = leaves.values().map(|leaf| leaf.remote_evaluations).sum();
+            stats.remote_rpcs = leaves.values().map(|leaf| leaf.remote_rpcs).sum();
+            annotate_logical_execution(&mut result, &stats);
+            remote_stats = Some((stats.remote_evaluations, stats.remote_rpcs));
+            Ok(result)
+        }
+        .instrument(tracing::info_span!("query_call", call_id))
+        .await;
+        log_query_outcome(call_id, "metricsql_instant", started, remote_stats, &result);
+        result
     }
 
     pub async fn execute_metricsql_range(
@@ -221,20 +278,42 @@ impl ASAPQueryEngine {
         step_ms: u64,
     ) -> Result<crate::query_engines::query_result::QueryResult, crate::query_engines::EngineError>
     {
-        let physical = self.active_physical_plan_snapshot().ok_or_else(|| {
-            crate::query_engines::EngineError::capability_miss(
-                "query_plan",
-                "no active physical plan",
-            )
-        })?;
-        let planned = physical
-            .query_plan
-            .lookup_canonical(asap_types::query_plan::QueryLanguage::MetricsQl, identity)
-            .map_err(|error| {
-                crate::query_engines::EngineError::capability_miss("query_plan", error.to_string())
+        let call_id = query_call_id();
+        let started = Instant::now();
+        debug!(target: "asap_runtime_debug",
+            call_id,
+            operation = "metricsql_range",
+            start_ms,
+            end_ms,
+            step_ms,
+            "query call started"
+        );
+        let result = async {
+            let physical = self.active_physical_plan_snapshot().ok_or_else(|| {
+                crate::query_engines::EngineError::capability_miss(
+                    "query_plan",
+                    "no active physical plan",
+                )
             })?;
-        self.execute_logical_range(&physical, planned, start_ms, end_ms, step_ms)
-            .await
+            let planned = physical
+                .query_plan
+                .lookup_canonical(asap_types::query_plan::QueryLanguage::MetricsQl, identity)
+                .map_err(|error| {
+                    crate::query_engines::EngineError::capability_miss(
+                        "query_plan",
+                        error.to_string(),
+                    )
+                })?;
+            debug!(target: "asap_runtime_debug", plan_id = physical.plan_id(), plan_version = physical.plan_version(),
+            query_id = %planned.query_id, start_ms, end_ms, step_ms,
+            "installed MetricsQL range query selected");
+            self.execute_logical_range(&physical, planned, start_ms, end_ms, step_ms)
+                .await
+        }
+        .instrument(tracing::info_span!("query_call", call_id))
+        .await;
+        log_query_outcome(call_id, "metricsql_range", started, None, &result);
+        result
     }
 
     /// Construct the query executor. Runtime configuration is read only from
@@ -271,12 +350,16 @@ impl ASAPQueryEngine {
         self.query_forwarding_policy = policy;
         self
     }
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(plan_id = physical.plan_id(), plan_version = physical.plan_version(),
+            query_id = %entry.query_id, evaluation_count = times.len()))]
     async fn prepare_query_inputs(
         &self,
         physical: &crate::storage_engines::types::RuntimePhysicalPlan,
         entry: &asap_types::query_plan::QueryPlanEntry,
         times: &[u64],
     ) -> Result<super::logical_dag::PreparedLeaves, crate::query_engines::EngineError> {
+        debug!(target: "asap_runtime_debug", "installed query input preparation started");
         super::catalog_resolver::validate_entry(
             physical.summary_catalog.as_deref(),
             entry,
@@ -333,7 +416,7 @@ impl ASAPQueryEngine {
                 )
             })
         {
-            debug!(
+            debug!(target: "asap_runtime_debug",
                 language = ?entry.language,
                 query_id = %entry.query_id,
                 "query forwarding disabled; external exact subquery blocked"
@@ -365,6 +448,9 @@ impl ASAPQueryEngine {
         .await
     }
 
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(plan_id = physical.plan_id(), plan_version = physical.plan_version(),
+            query_id = %entry.query_id, evaluation_ms = at))]
     fn execute_logical_entry(
         &self,
         physical: &crate::storage_engines::types::RuntimePhysicalPlan,
@@ -545,6 +631,9 @@ impl ASAPQueryEngine {
         result
     }
 
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(plan_id = physical.plan_id(), plan_version = physical.plan_version(),
+            query_id = %entry.query_id, start_ms = start, end_ms = end, step_ms = step))]
     async fn execute_logical_range(
         &self,
         physical: &crate::storage_engines::types::RuntimePhysicalPlan,
@@ -558,6 +647,7 @@ impl ASAPQueryEngine {
             query_result::{QueryResult, RangeVectorElement},
             EngineError,
         };
+        debug!(target: "asap_runtime_debug", "installed query range execution started");
         if step == 0 || start > end || (end - start) / step >= 11_000 {
             return Err(EngineError::capability_miss(
                 "installed_logical_dag",
@@ -748,11 +838,24 @@ impl ASAPQueryEngine {
         step_ms: u64,
     ) -> Result<crate::query_engines::query_result::QueryResult, crate::query_engines::EngineError>
     {
+        let call_id = query_call_id();
+        let started = Instant::now();
+        debug!(target: "asap_runtime_debug",
+            call_id,
+            operation = "promql_range",
+            start_ms,
+            end_ms,
+            step_ms,
+            "query call started"
+        );
+        let result = async {
         if let Some(physical) = self.active_physical_plan_snapshot() {
             if let Ok(entry) = physical.query_plan.lookup(query) {
                 if entry.nodes.values().any(|node| {
                     matches!(node, asap_types::query_plan::QueryPlanNode::Logical { .. })
                 }) {
+                    debug!(target: "asap_runtime_debug", plan_id = physical.plan_id(), plan_version = physical.plan_version(),
+                        query_id = %entry.query_id, "installed query DAG selected");
                     return self
                         .execute_logical_range(&physical, entry, start_ms, end_ms, step_ms)
                         .await;
@@ -856,6 +959,9 @@ impl ASAPQueryEngine {
         // Complete coverage is a prerequisite above. Hybrid stitching is
         // retained for legacy/test callers without an active QueryPlan only.
         Ok(warm_qr)
+        }.instrument(tracing::info_span!("query_call", call_id)).await;
+        log_query_outcome(call_id, "promql_range", started, None, &result);
+        result
     }
 }
 
@@ -999,25 +1105,43 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
         now_ms: u64,
     ) -> Result<crate::query_engines::query_result::QueryResult, crate::query_engines::EngineError>
     {
+        let call_id = query_call_id();
+        let started = Instant::now();
+        let mut remote_stats = None;
+        debug!(target: "asap_runtime_debug",
+            call_id,
+            operation = "promql_instant",
+            evaluation_ms = now_ms,
+            "query call started"
+        );
+        let result = async {
         if let Some(physical) = self.active_physical_plan_snapshot() {
             if let Ok(entry) = physical.query_plan.lookup(query) {
+                debug!(target: "asap_runtime_debug", plan_id = physical.plan_id(), plan_version = physical.plan_version(),
+                    query_id = %entry.query_id, evaluation_ms = now_ms,
+                    "installed query DAG selected");
                 let leaves = self
                     .prepare_query_inputs(&physical, entry, &[now_ms])
                     .await
                     .map_err(|error| {
-                        tracing::warn!(query, error = %error, "installed query DAG preparation failed");
+                        tracing::warn!(plan_id = physical.plan_id(),
+                            plan_version = physical.plan_version(), query_id = %entry.query_id,
+                            error = %error, "installed query DAG preparation failed");
                         error
                     })?;
                 let (mut result, mut stats) = self
                     .execute_logical_entry(&physical, entry, &leaves, now_ms)
                     .map_err(|error| {
-                        tracing::warn!(query, error = %error, "installed query DAG execution failed");
+                        tracing::warn!(plan_id = physical.plan_id(),
+                            plan_version = physical.plan_version(), query_id = %entry.query_id,
+                            error = %error, "installed query DAG execution failed");
                         error
                     })?;
                 stats.remote_evaluations =
                     leaves.values().map(|leaf| leaf.remote_evaluations).sum();
                 stats.remote_rpcs = leaves.values().map(|leaf| leaf.remote_rpcs).sum();
                 annotate_logical_execution(&mut result, &stats);
+                remote_stats = Some((stats.remote_evaluations, stats.remote_rpcs));
                 return Ok(result);
             }
         }
@@ -1112,6 +1236,9 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu
             crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(),
             format!("ASAPQueryEngine: no sketch index for `{query}` — failing over to archive"),
         ))
+        }.instrument(tracing::info_span!("query_call", call_id)).await;
+        log_query_outcome(call_id, "promql_instant", started, remote_stats, &result);
+        result
     }
 
     /// Range-query entry point for the [`EngineRouter`] failover loop.
diff --git a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs
index c99856af..add9887e 100644
--- a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs
+++ b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs
@@ -296,147 +296,154 @@ pub(super) async fn prepare_external(
     // Equivalent exact cuts at the same time share one actual remote request.
     let mut remote_cache = HashMap::<(QueryLanguage, String, i64), Value>::new();
     for ((id, at), leaf) in leaves(entry, times)? {
-        u64::try_from(at).map_err(|_| miss("subquery predates epoch"))?;
-        let (language, query, candidate_input) = match &leaf {
-            ExactLeaf::Legacy(ResidualQueryOperator::ExactSubquery { query }) => {
-                (QueryLanguage::PromQl, query.clone(), None)
-            }
-            ExactLeaf::Legacy(ResidualQueryOperator::CandidateExactSubquery {
-                query,
-                item_label,
-            }) => (
-                QueryLanguage::PromQl,
-                query.clone(),
-                Some((entry.nodes[&id].inputs()[0], item_label.as_str())),
-            ),
-            ExactLeaf::External(request) => {
-                if !matches!(
-                    request.language,
-                    QueryLanguage::PromQl | QueryLanguage::MetricsQl
-                ) {
-                    return Err(miss(format!(
-                        "external exact language {:?} has no installed adapter",
-                        request.language
-                    )));
+        let result = async {
+            u64::try_from(at).map_err(|_| miss("subquery predates epoch"))?;
+            let (language, query, candidate_input) = match &leaf {
+                ExactLeaf::Legacy(ResidualQueryOperator::ExactSubquery { query }) => {
+                    (QueryLanguage::PromQl, query.clone(), None)
                 }
-                let candidate = match request.input_contracts.as_slice() {
-                    [] => None,
-                    [ExternalExactInput::CandidateMembership { item_label }] => {
-                        Some((entry.nodes[&id].inputs()[0], item_label.as_str()))
+                ExactLeaf::Legacy(ResidualQueryOperator::CandidateExactSubquery {
+                    query,
+                    item_label,
+                }) => (
+                    QueryLanguage::PromQl,
+                    query.clone(),
+                    Some((entry.nodes[&id].inputs()[0], item_label.as_str())),
+                ),
+                ExactLeaf::External(request) => {
+                    if !matches!(
+                        request.language,
+                        QueryLanguage::PromQl | QueryLanguage::MetricsQl
+                    ) {
+                        return Err(miss(format!(
+                            "external exact language {:?} has no installed adapter",
+                            request.language
+                        )));
                     }
-                    _ => return Err(miss("unsupported external exact input contract")),
-                };
-                (request.language, request.expression.clone(), candidate)
-            }
-            _ => return Err(miss("prepared leaf is not an exact subtree")),
-        };
-        let (query, candidate_filtered) = if let Some((candidate_input, item_label)) =
-            candidate_input
-        {
-            if language == QueryLanguage::MetricsQl {
-                return Err(miss(
-                    "candidate-filtered MetricsQL exact subqueries are not implemented",
-                ));
-            }
-            let candidate = prepared
-                .get(&(candidate_input, at))
-                .ok_or_else(|| miss("candidate membership was not prepared"))?;
-            let Value::Vector(rows) = &candidate.value else {
-                return Err(miss("candidate membership is not an instant vector"));
+                    let candidate = match request.input_contracts.as_slice() {
+                        [] => None,
+                        [ExternalExactInput::CandidateMembership { item_label }] => {
+                            Some((entry.nodes[&id].inputs()[0], item_label.as_str()))
+                        }
+                        _ => return Err(miss("unsupported external exact input contract")),
+                    };
+                    (request.language, request.expression.clone(), candidate)
+                }
+                _ => return Err(miss("prepared leaf is not an exact subtree")),
             };
-            let mut values = rows
-                .iter()
-                .map(|(labels, _)| {
-                    labels.get(item_label).cloned().ok_or_else(|| {
-                        miss(format!(
-                            "candidate membership is missing item label {item_label}"
-                        ))
+            let (query, candidate_filtered) = if let Some((candidate_input, item_label)) =
+                candidate_input
+            {
+                if language == QueryLanguage::MetricsQl {
+                    return Err(miss(
+                        "candidate-filtered MetricsQL exact subqueries are not implemented",
+                    ));
+                }
+                let candidate = prepared
+                    .get(&(candidate_input, at))
+                    .ok_or_else(|| miss("candidate membership was not prepared"))?;
+                let Value::Vector(rows) = &candidate.value else {
+                    return Err(miss("candidate membership is not an instant vector"));
+                };
+                let mut values = rows
+                    .iter()
+                    .map(|(labels, _)| {
+                        labels.get(item_label).cloned().ok_or_else(|| {
+                            miss(format!(
+                                "candidate membership is missing item label {item_label}"
+                            ))
+                        })
                     })
-                })
-                .collect::<Result<Vec<_>, _>>()?;
-            values.sort();
-            values.dedup();
-            if values.is_empty() {
-                prepared.insert(
-                    (id, at),
-                    PreparedLeaf {
+                    .collect::<Result<Vec<_>, _>>()?;
+                values.sort();
+                values.dedup();
+                if values.is_empty() {
+                    return Ok(PreparedLeaf {
                         value: Value::Vector(Vec::new()),
                         remote: true,
                         remote_evaluations: 0,
                         remote_rpcs: 0,
-                    },
-                );
-                continue;
-            }
-            if values.len() > MAX_CANDIDATE_VALUES {
-                return Err(miss(format!(
-                    "candidate set has {} values, exceeding limit {MAX_CANDIDATE_VALUES}",
-                    values.len()
-                )));
-            }
-            let restricted = inject_candidate_matcher(&query, item_label, &values)?;
-            if restricted.len() > MAX_CANDIDATE_QUERY_BYTES {
-                return Err(miss(format!(
-                        "candidate-filtered exact query has {} bytes, exceeding limit {MAX_CANDIDATE_QUERY_BYTES}",
-                        restricted.len()
+                    });
+                }
+                if values.len() > MAX_CANDIDATE_VALUES {
+                    return Err(miss(format!(
+                        "candidate set has {} values, exceeding limit {MAX_CANDIDATE_VALUES}",
+                        values.len()
                     )));
-            }
-            (restricted, true)
-        } else {
-            (query, false)
-        };
-        let key = (language, query.clone(), at);
-        let cached = remote_cache.contains_key(&key);
-        let value = if let Some(value) = remote_cache.get(&key) {
-            value.clone()
-        } else {
-            let endpoint = match language {
-                QueryLanguage::PromQl => prometheus_endpoint
-                    .ok_or_else(|| miss("Prometheus exact endpoint unavailable"))?,
-                QueryLanguage::MetricsQl => metricsql_endpoint
-                    .ok_or_else(|| miss("VictoriaMetrics exact endpoint unavailable"))?,
-                QueryLanguage::ClickHouseSql => {
-                    return Err(miss("ClickHouse exact subquery needs its SQL adapter"))
                 }
+                let restricted = inject_candidate_matcher(&query, item_label, &values)?;
+                if restricted.len() > MAX_CANDIDATE_QUERY_BYTES {
+                    return Err(miss(format!(
+                            "candidate-filtered exact query has {} bytes, exceeding limit {MAX_CANDIDATE_QUERY_BYTES}",
+                            restricted.len()
+                        )));
+                }
+                (restricted, true)
+            } else {
+                (query, false)
             };
-            let url = format!("{}/api/v1/query", endpoint.trim_end_matches('/'));
-            let time = format!("{:.3}", at as f64 / 1000.0);
-            // Candidate sets can be large enough to exceed proxy URL limits;
-            // Prometheus accepts the instant-query parameters as an encoded
-            // form body. Static exact cuts keep their existing GET contract.
-            let request = if candidate_filtered {
-                client
-                    .post(url)
-                    .form(&[("query", query.as_str()), ("time", time.as_str())])
+            let key = (language, query.clone(), at);
+            let cached = remote_cache.contains_key(&key);
+            let value = if let Some(value) = remote_cache.get(&key) {
+                value.clone()
             } else {
-                client
-                    .get(url)
-                    .query(&[("query", query.as_str()), ("time", time.as_str())])
+                let endpoint = match language {
+                    QueryLanguage::PromQl => prometheus_endpoint
+                        .ok_or_else(|| miss("Prometheus exact endpoint unavailable"))?,
+                    QueryLanguage::MetricsQl => metricsql_endpoint
+                        .ok_or_else(|| miss("VictoriaMetrics exact endpoint unavailable"))?,
+                    QueryLanguage::ClickHouseSql => {
+                        return Err(miss("ClickHouse exact subquery needs its SQL adapter"))
+                    }
+                };
+                let url = format!("{}/api/v1/query", endpoint.trim_end_matches('/'));
+                let time = format!("{:.3}", at as f64 / 1000.0);
+                // Candidate sets can be large enough to exceed proxy URL limits;
+                // Prometheus accepts the instant-query parameters as an encoded
+                // form body. Static exact cuts keep their existing GET contract.
+                let request = if candidate_filtered {
+                    client
+                        .post(url)
+                        .form(&[("query", query.as_str()), ("time", time.as_str())])
+                } else {
+                    client
+                        .get(url)
+                        .query(&[("query", query.as_str()), ("time", time.as_str())])
+                };
+                let response = request
+                    .send()
+                    .await
+                    .map_err(|e| miss(format!("exact request failed: {e}")))?;
+                if !response.status().is_success() {
+                    return Err(miss(format!("exact endpoint HTTP {}", response.status())));
+                }
+                let body: serde_json::Value = response
+                    .json()
+                    .await
+                    .map_err(|e| miss(format!("invalid exact response: {e}")))?;
+                let value = parse_result(&body, at)?;
+                remote_cache.insert(key, value.clone());
+                value
             };
-            let response = request
-                .send()
-                .await
-                .map_err(|e| miss(format!("exact request failed: {e}")))?;
-            if !response.status().is_success() {
-                return Err(miss(format!("exact endpoint HTTP {}", response.status())));
-            }
-            let body: serde_json::Value = response
-                .json()
-                .await
-                .map_err(|e| miss(format!("invalid exact response: {e}")))?;
-            let value = parse_result(&body, at)?;
-            remote_cache.insert(key, value.clone());
-            value
-        };
-        prepared.insert(
-            (id, at),
-            PreparedLeaf {
+            Ok(PreparedLeaf {
                 value,
                 remote: true,
                 remote_evaluations: usize::from(!cached),
                 remote_rpcs: usize::from(!cached),
-            },
-        );
+            })
+        }
+        .await;
+        match result {
+            Ok(leaf) => {
+                prepared.insert((id, at), leaf);
+            }
+            Err(error) => {
+                tracing::debug!(target: "asap_runtime_debug", query_id = %entry.query_id,
+                    node_id = ?id, op = entry.nodes[&id].op_label(), evaluation_ms = at, %error,
+                    "installed query exact leaf could not be prepared");
+                return Err(error);
+            }
+        }
     }
     Ok(prepared)
 }
diff --git a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs
index 42cbec9b..bf60deec 100644
--- a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs
+++ b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs
@@ -135,6 +135,7 @@ where
         for dependency in &dependencies {
             if let Some(id) = identities.get(dependency) {
                 runtime.borrow_mut().stats.memo_hits += 1;
+                tracing::debug!(target: "asap_runtime_debug", query_id = %entry.query_id, node_id = ?dependency.0, evaluation_ms = dependency.1, "installed query node reused within request");
                 input_ids.push(*id);
             } else {
                 if identities.len() >= 200_000 {
@@ -573,6 +574,8 @@ impl<F: FnMut(QueryNodeId, u64) -> Result<QueryResult, EngineError>>
                 }))
                 .await?;
             let values = values.iter().map(|v| v.value().clone()).collect::<Vec<_>>();
+            let started = std::time::Instant::now();
+            tracing::debug!(target: "asap_runtime_debug", node_id = ?self.id, evaluation_ms = self.time, op = self.node.op_label(), syntax = %self.node.log_syntax(), "installed query node started");
             self.runtime
                 .borrow_mut()
                 .execute_node(
@@ -583,7 +586,11 @@ impl<F: FnMut(QueryNodeId, u64) -> Result<QueryResult, EngineError>>
                     &self.dependencies,
                     &context,
                 )
+                .inspect(|_| {
+                    tracing::debug!(target: "asap_runtime_debug", node_id = ?self.id, op = self.node.op_label(), elapsed_us = started.elapsed().as_micros() as u64, "installed query node completed");
+                })
                 .map_err(|error| {
+                    tracing::warn!(node_id = ?self.id, op = self.node.op_label(), elapsed_us = started.elapsed().as_micros() as u64, %error, "installed query node failed");
                     *self.error.borrow_mut() = Some(error);
                     physical::Error::Operator(format!(
                         "query node {} at {} failed",
diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs
index b05261d2..2197d863 100644
--- a/data_plane/src/storage_engines/sketch_db/index/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs
@@ -974,11 +974,14 @@ impl SketchStore {
         self.install_catalog_outputs(catalog, outputs)
     }
 
+    #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all,
+        fields(plan_id = catalog.plan_id, plan_version = catalog.plan_version))]
     fn install_catalog_outputs(
         &self,
         catalog: Arc<asap_types::summary_catalog::SummaryCatalog>,
         outputs: BTreeMap<SummaryDefinitionId, StoredOutputReference>,
     ) -> Result<(), String> {
+        tracing::debug!(target: "asap_runtime_debug", "summary store catalog installation started");
         for (definition, output) in &outputs {
             output.validate().map_err(|e| e.to_string())?;
             if output.definition_id != *definition || !catalog.definitions.contains_key(definition)
@@ -1068,12 +1071,24 @@ impl SketchStore {
         // producer cannot mutate a new generation before its receipt is rejected.
         let mut inventory = self.admission.write().unwrap();
         if inventory.validate_publication(generation, coordinate, first_revision, revision)? {
+            tracing::debug!(target: "asap_runtime_debug", plan_id = generation.plan_id,
+                plan_version = generation.plan_version,
+                summary_definition = coordinate.summary_definition_id.as_u64(),
+                window_start_ms = coordinate.time_range.start_ms,
+                window_end_ms = coordinate.time_range.end_ms, revision,
+                "SDS summary publication replay acknowledged");
             return Ok(());
         }
         let series_id =
             persist(&SummaryPublicationWriter(self)).ok_or("summary state publication failed")?;
         inventory.record_series(generation, coordinate, series_id)?;
         inventory.acknowledge(generation, coordinate, revision)?;
+        tracing::debug!(target: "asap_runtime_debug", plan_id = generation.plan_id,
+            plan_version = generation.plan_version, sid = series_id,
+            summary_definition = coordinate.summary_definition_id.as_u64(),
+            window_start_ms = coordinate.time_range.start_ms,
+            window_end_ms = coordinate.time_range.end_ms, revision,
+            "SDS summary publication acknowledged");
         self.admitted_mutations
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
         let floor = coordinate
@@ -1758,6 +1773,7 @@ impl SketchStore {
         start_unix_ms: u64,
         end_unix_ms: u64,
     ) -> Vec<SketchTimeSeries> {
+        let started = std::time::Instant::now();
         if self
             .incomplete_summary_lineages
             .get(&sid)
@@ -1768,6 +1784,8 @@ impl SketchStore {
                 })
             })
         {
+            tracing::debug!(target: "asap_runtime_debug", sid, start_unix_ms, end_unix_ms,
+                reason = "incomplete_summary_lineage", "SketchStore sketch range read skipped");
             return Vec::new();
         }
         // Result is keyed by the resolved label MAP so the in-memory tier
@@ -1961,14 +1979,18 @@ impl SketchStore {
         // state; the guard is belt-and-suspenders).
         self.union_disk_parts_into(sid, start_unix_ms, end_unix_ms, &mut by_label_map);
 
-        by_label_map
+        let result: Vec<_> = by_label_map
             .into_iter()
             .map(|(label_values, samples)| SketchTimeSeries {
                 sid,
                 series_label_values: label_values,
                 samples,
             })
-            .collect()
+            .collect();
+        tracing::debug!(target: "asap_runtime_debug", sid, start_unix_ms, end_unix_ms,
+            series_count = result.len(), elapsed_us = started.elapsed().as_micros() as u64,
+            "SketchStore sketch range read completed");
+        result
     }
 
     /// Resolve the sorted group-by KEYS for a sid from its instance
@@ -2203,6 +2225,7 @@ impl SketchStore {
         BTreeMap<String, String>,
         BTreeMap<i64, Arc<dyn crate::storage_engines::types::AggregateCore>>,
     )> {
+        let started = std::time::Instant::now();
         // Key by the resolved label MAP (not `LabelValuesId`) so the
         // in-memory tier and the durable disk tier — which carry
         // independent intern spaces — union by label identity. Mirrors
@@ -2263,7 +2286,11 @@ impl SketchStore {
         // of the range. In-memory wins on a window-end collision.
         self.union_disk_exact_agg_into(sid, start_unix_ms, end_unix_ms, &mut by_label_map);
 
-        by_label_map.into_iter().collect()
+        let result: Vec<_> = by_label_map.into_iter().collect();
+        tracing::debug!(target: "asap_runtime_debug", sid, start_unix_ms, end_unix_ms,
+            series_count = result.len(), elapsed_us = started.elapsed().as_micros() as u64,
+            "SketchStore exact range read completed");
+        result
     }
 
     /// Union the durable disk tier's exact-aggregation entries into
@@ -3387,6 +3414,13 @@ impl SketchStore {
                 accumulator.clone_boxed_core(),
             ),
         };
+        tracing::debug!(target: "asap_runtime_debug", sid,
+            policy_fp = %output.policy_fp,
+            aggregation_type = ?agg_cfg.aggregation_type,
+            window_start_ms = output.start_timestamp,
+            window_end_ms = output.end_timestamp,
+            accepted,
+            "SketchStore precompute window append completed");
         accepted.then_some(sid)
     }
 }
diff --git a/data_plane/src/storage_engines/sketch_db/sds.rs b/data_plane/src/storage_engines/sketch_db/sds.rs
index 99f023a7..7d05d14e 100644
--- a/data_plane/src/storage_engines/sketch_db/sds.rs
+++ b/data_plane/src/storage_engines/sketch_db/sds.rs
@@ -143,8 +143,14 @@ impl SummaryDescriptorRegistry {
             asap_types::sds::StoredOutputReference,
         >,
     ) -> Result<(), asap_types::summary_catalog::SummaryCatalogError> {
-        catalog.validate()?;
-        let reference = catalog.reference()?;
+        catalog.validate().inspect_err(|error| {
+            tracing::warn!(plan_id = catalog.plan_id, plan_version = catalog.plan_version,
+                %error, "storage descriptor catalog validation failed");
+        })?;
+        let reference = catalog.reference().inspect_err(|error| {
+            tracing::warn!(plan_id = catalog.plan_id, plan_version = catalog.plan_version,
+                %error, "storage descriptor catalog reference failed");
+        })?;
         let generation = Arc::new(asap_types::sds::CatalogGeneration {
             schema_version: reference.schema_version,
             plan_id: reference.plan_id,
@@ -152,6 +158,11 @@ impl SummaryDescriptorRegistry {
             snapshot_sha256: reference.snapshot_sha256,
         });
         *self.authoritative_catalog.write().unwrap() = Some((catalog, generation, outputs));
+        tracing::info!(
+            plan_id = reference.plan_id,
+            plan_version = reference.plan_version,
+            "storage descriptor catalog installed"
+        );
         Ok(())
     }
 
@@ -257,6 +268,16 @@ impl SummaryDescriptorRegistry {
             }
         };
 
+        let catalog_generation = authoritative
+            .as_ref()
+            .map(|(_, generation, _)| generation.clone());
+        tracing::debug!(target: "asap_runtime_debug", sid = metadata.storage_handle,
+            policy_fp = %metadata.policy_fp,
+            summary_descriptor_hash = format_args!("{:016x}", xxhash_rust::xxh64::xxh64(summary_descriptor.id().canonical().as_bytes(), 0)),
+            data_descriptor_hash = format_args!("{:016x}", xxhash_rust::xxh64::xxh64(data_id.canonical().as_bytes(), 0)),
+            plan_id = catalog_generation.as_ref().map(|g| g.plan_id),
+            plan_version = catalog_generation.as_ref().map(|g| g.plan_version),
+            "SDS descriptor binding resolved");
         Ok(SdsBinding {
             stored_output_reference,
             metadata: Arc::new(metadata),
diff --git a/docs/runtime-debugging.md b/docs/runtime-debugging.md
new file mode 100644
index 00000000..29384fcd
--- /dev/null
+++ b/docs/runtime-debugging.md
@@ -0,0 +1,9 @@
+# Runtime debugging logs
+
+Set `RUST_LOG=info,asap_runtime_debug=debug` before starting both controller and backend to enable workflow debug events while keeping other targets at `info`. Set `RUST_LOG=info` to hide debug events. Both processes use `RUST_LOG` and default to `info` when it is unset or invalid. Their console streams use the same `tracing_subscriber` text layout with timestamp, level, target, source file and line, span context, and event fields. The controller writes to stdout. The backend writes to stdout and `<output_dir>/query_engine.log`; the file layer disables terminal colors.
+
+Filter both streams by `plan_id` and `plan_version`. A `call_id` distinguishes concurrent plan requests or query calls within one process; it is local to that process and is not shared between controller and backend. The controller reports compilation, collector preflight and publication, backend staging, and activation. The backend reports staging validation, activation, and summary catalog installation. At `debug`, the query engine reports call start and completion or failure with elapsed time, the selected installed query DAG, and each evaluated node with its `node_id`, operator label `op` (for example `logical/aggregate/sum` or `exact_readout/count`), and inclusive duration. Node start events also include `inputs` and `syntax`: bounded operator arguments such as `operation=Sum grouping=by(service)`, `operation=Div return_bool=false`, or window parameters. The syntax field omits complete query strings, matchers, and samples. Prepared leaves and memo hits have separate events with the node type in `op`. When input preparation fails, the backend logs the node that failed catalog validation or the exact leaf that could not be prepared, with its `node_id` and `op`; a failing DAG logs the first failing node rather than repeating its error at every ancestor. Instant-query completion also includes remote evaluation and RPC counts. Failed query calls that the ASAP tier cannot serve are logged at `debug`; backend failures and failed plan operations are logged at `warn` or `error`. Error messages can contain query text, so protect access to these logs accordingly.
+
+Events share a rendered layout, but their fields depend on the operation. Plan events use `plan_id` and `plan_version`; query events add `query_id`, `call_id`, and elapsed time when available; node events add `node_id` and `op`. Inner spans cover compiler selection and transmission construction, backend publication and validation, worker processing, summary catalog installation, and installed query preparation and execution. They carry identifiers and counts without dumping request bodies or samples. Existing messages elsewhere in the system still include free-form text. These logs are text output, not a JSON schema or a distributed trace ID.
+
+For deeper precompute and storage work, use `RUST_LOG=info,asap_runtime_debug=debug,data_plane::precompute_engine=debug,data_plane::storage_engines=debug` on the backend. The precompute DAG span carries plan ID, version, sink node, summary definition, and window bounds. Node events add the operator type, bounded syntax arguments, input node IDs, and duration; a reused materialization or committed sink has a separate event. Worker processing logs the aggregation type and window layout. SketchStore logs accepted or rejected window appends and range read counts. SDS logs descriptor bindings by a short hash of each canonical descriptor ID, and admitted publication receipts with their catalog generation and summary definition. The new event fields omit raw samples and full fallback expressions; existing worker spans may include group label values. An error before backend staging points to plan compilation or publication; an error after activation with a matching plan identifier points to runtime query or storage behavior.