refactor(controller): retire algebra/+planner/+config/+query_language/+analyzer.rs (Steps 2b/c/d/e/f)

controller/src/deployment_model.rs

@@ -0,0 +1,75 @@
+//! `DeploymentModelRegistry` + `DeploymentModelId` — placeholder
+//! scaffolding for the design.md §5 `core::registry` surface.
+//!
+//! Refactor 2026-05 created this module so the design.md target layout
+//! is materialised in code. The single-crate today (no separate
+//! `deployment-model-asaplifecycle` / `deployment-model-asapquery` /
+//! `deployment-model-asapfusion` crates) means there is exactly one
+//! deployment model and the registry is degenerate. The shape is
+//! defined here so the future multi-crate migration is purely additive.
+
+#![allow(dead_code)]
+
+use std::collections::HashMap;
+
+/// Stable identifier for a deployment model — `asaplifecycle`,
+/// `asapquery`, `asapfusion`, etc. The string is the wire identifier
+/// used in `QuerySpec::deployment_model` (see `crate::pipeline::QuerySpec`)
+/// and in the per-deployment-model crate selection on a future `bin/`
+/// build.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct DeploymentModelId(pub String);
+
+impl DeploymentModelId {
+    /// Default DC deployment id — the controller's historical name for
+    /// the lifecycle deployment model (edge / gateway / backend OTel
+    /// collectors).
+    pub fn asaplifecycle() -> Self {
+        Self("asaplifecycle".to_string())
+    }
+
+    /// `asapquery` — the ASAPQuery-backend in-process planner for
+    /// `streaming_config.yaml` + `inference_config.yaml` emission.
+    pub fn asapquery() -> Self {
+        Self("asapquery".to_string())
+    }
+
+    /// `asapfusion` — the DataFusion `LogicalPlan` rewriter.
+    pub fn asapfusion() -> Self {
+        Self("asapfusion".to_string())
+    }
+}
+
+/// Registry of available deployment models.
+///
+/// Today this is a stub — the single-crate ships only the
+/// `asaplifecycle` deployment model and the registry contains one
+/// entry. The shape is here so the future multi-crate migration can
+/// populate it with concrete `Box<dyn DeploymentModel>` entries without
+/// reworking the controller's startup flow.
+#[derive(Default)]
+pub struct DeploymentModelRegistry {
+    /// Map of deployment model id → opaque marker. Reserved for the
+    /// `dyn DeploymentModel` trait object that will land with the
+    /// multi-crate split.
+    entries: HashMap<DeploymentModelId, ()>,
+}
+
+impl DeploymentModelRegistry {
+    /// Register a deployment model id. Returns `true` when the id was
+    /// added; `false` when it was already present (the registry is a
+    /// set today, not a map).
+    pub fn register(&mut self, id: DeploymentModelId) -> bool {
+        self.entries.insert(id, ()).is_none()
+    }
+
+    /// Whether the registry knows about a deployment model id.
+    pub fn contains(&self, id: &DeploymentModelId) -> bool {
+        self.entries.contains_key(id)
+    }
+
+    /// Iterate over the known deployment model ids.
+    pub fn ids(&self) -> impl Iterator<Item = &DeploymentModelId> {
+        self.entries.keys()
+    }
+}

controller/src/config/agent.rs → controller/src/emit/agent.rs

@@ -3,7 +3,7 @@ use serde::Serialize;
 use serde_yaml::{Mapping, Value};
 use std::collections::HashMap;
 
-use crate::analyzer::format_duration;
+use crate::pipeline::format_duration;
 use crate::types::*;
 
 // ── YAML structural types ─────────────────────────────────────────────────────

controller/src/config/mod.rs → controller/src/emit/mod.rs

@@ -1,11 +1,28 @@
+//! `emit/` — per-deployment-model plan emitters (L5 output side).
+//!
+//! Per `controller/docs/design.md` §5 `core::emit`. The 2026-05
+//! layered-cleanup refactor consolidated the former
+//! `controller/src/config/` directory here. Mapping:
+//!
+//! | Old path | New path |
+//! |---|---|
+//! | `config/agent.rs` | [`agent`] |
+//! | `config/backend.rs` | [`backend`] |
+//! | `config/asapquery_backend.rs` | [`asapquery_backend`] |
+//! | `config/precompute.rs` | [`precompute`] |
+//! | `config/stage_config.rs` | [`stage_config`] (TODO: split into `opamp` + `streaming_config` + `inference_config` per design.md §5; deferred from refactor 2026-05 because the 3,020-line monolith mixes OTel-collector YAML emit, ASAPQuery-backend JSON emit, and shared internals — clean split needs ownership reorganisation, not file renames) |
+//! | `config/stage_config_otap.rs` | [`otap`] |
+//! | `config/stage_config_telegraf.rs` | [`telegraf`] |
+//! | `config/workloads.rs` | [`crate::workload`] (top-level — design.md §5 puts `workload` next to `emit`, not inside it) |
+
 pub mod agent;
 pub mod asapquery_backend;
 pub mod backend;
 pub mod precompute;
 pub mod stage_config;
-pub mod stage_config_otap;
-pub mod stage_config_telegraf;
-pub mod workloads;
+pub mod otap;
+pub mod telegraf;
+pub mod trait_def;
 
 pub use agent::generate_agent_config;
 pub use asapquery_backend::generate_streaming_config_yaml;
@@ -17,11 +34,18 @@ pub use stage_config::{
     emit_backend_storage_routing_with_prometheus_for_tenant, emit_edge_yaml, emit_gateway_yaml,
     DEFAULT_TENANT,
 };
-pub use stage_config_otap::emit_otap_dag_yaml;
-pub use stage_config_telegraf::emit_telegraf_toml;
-pub use workloads::WorkloadRegistry;
-
-use crate::stage_split::emitter::EdgeStageConfig;
+pub use otap::emit_otap_dag_yaml;
+pub use telegraf::emit_telegraf_toml;
+pub use trait_def::PlanEmitter;
+
+// Refactor 2026-05: design.md §5 puts `WorkloadRegistry` next to
+// `emit`, not inside it. The new home is `crate::workload`; we re-export
+// here so historical `crate::config::WorkloadRegistry` and
+// `controller::config::WorkloadRegistry` references via the `config`
+// back-compat alias keep working without churn.
+pub use crate::workload::WorkloadRegistry;
+
+use crate::physical::colored_dag::emitter::EdgeStageConfig;
 use crate::sketch_algebra::SketchExpr;
 use crate::sketch_algebra::params::SketchKind;
 use crate::store::WorkloadStore;
@@ -148,7 +172,7 @@ pub fn extend_edge_with_demo_plumbing(
     edge_cfg: &mut EdgeStageConfig,
     workload_registry_metrics: impl IntoIterator<Item = String>,
 ) {
-    use crate::stage_split::emitter::ArchiveTierMetric;
+    use crate::physical::colored_dag::emitter::ArchiveTierMetric;
 
     // 1. Freshness probes → archive tier with the tight 10s window.
     for m in FRESHNESS_PROBE_METRICS.iter() {
@@ -300,8 +324,8 @@ mod runtime_tests {
     #[test]
     fn emit_for_runtime_default_matches_emit_edge_yaml() {
         use crate::sketch_algebra::params::{DDSketchParams, SketchParams};
-        use crate::stage_split::emitter::{EdgeSketchProcessor, ExportTarget};
-        use crate::stage_split::stage_id::StageId;
+        use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, ExportTarget};
+        use crate::physical::colored_dag::stage_id::StageId;
         use crate::sketch_algebra::params::SketchKind;
 
         let cfg = EdgeStageConfig {
@@ -330,8 +354,8 @@ mod runtime_tests {
 
     #[test]
     fn emit_for_runtime_otap_yields_dag_yaml() {
-        use crate::stage_split::emitter::ExportTarget;
-        use crate::stage_split::stage_id::StageId;
+        use crate::physical::colored_dag::emitter::ExportTarget;
+        use crate::physical::colored_dag::stage_id::StageId;
 
         let cfg = EdgeStageConfig {
             source_metric: Some("m".to_string()),
@@ -353,8 +377,8 @@ mod runtime_tests {
 
     #[test]
     fn emit_for_runtime_telegraf_yields_toml() {
-        use crate::stage_split::emitter::ExportTarget;
-        use crate::stage_split::stage_id::StageId;
+        use crate::physical::colored_dag::emitter::ExportTarget;
+        use crate::physical::colored_dag::stage_id::StageId;
 
         let cfg = EdgeStageConfig {
             source_metric: Some("m".to_string()),
@@ -392,7 +416,7 @@ mod runtime_tests {
         registry: &WorkloadRegistry,
         store: &WorkloadStore,
     ) {
-        use crate::analyzer::{Analyzer, QuerySpec};
+        use crate::pipeline::{Analyzer, QuerySpec};
         use crate::types;
         use crate::types_v2;
         let analyzer = Analyzer::new();
@@ -464,11 +488,11 @@ mod runtime_tests {
   assign_to_role: agent
   sketch_family_override: CountMinSketch
 "#;
-        let entries: Vec<crate::config::workloads::WorkloadEntry> =
+        let entries: Vec<crate::workload::WorkloadEntry> =
             serde_yaml::from_str(yaml).expect("parse workload yaml");
         assert_eq!(entries.len(), 6, "all 6 contract metrics must deserialize");
 
-        let registry = crate::config::workloads::WorkloadRegistry::from_entries(entries);
+        let registry = crate::workload::WorkloadRegistry::from_entries(entries);
         let store = WorkloadStore::new();
         populate_store_from_registry(&registry, &store);
 

controller/src/config/stage_config_otap.rs → controller/src/emit/otap.rs

@@ -39,8 +39,8 @@ use serde_yaml::{Mapping, Value};
 use std::collections::BTreeMap;
 
 use crate::sketch_algebra::params::{SketchKind, SketchParams};
-use crate::stage_split::emitter::{EdgeStageConfig, EdgeSketchProcessor, ExportTarget};
-use crate::stage_split::stage_id::StageId;
+use crate::physical::colored_dag::emitter::{EdgeStageConfig, EdgeSketchProcessor, ExportTarget};
+use crate::physical::colored_dag::stage_id::StageId;
 
 /// Default URL for Prometheus's native OTLP HTTP receiver.
 /// Matches `super::stage_config::emit_edge_yaml`'s placeholder so the
@@ -337,7 +337,7 @@ fn sketch_kind_tag(kind: &SketchKind) -> &'static str {
 mod tests {
     use super::*;
     use crate::sketch_algebra::params::DDSketchParams;
-    use crate::stage_split::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric};
+    use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric};
 
     /// Minimal struct-stub used to validate the emitted DAG parses as the
     /// otap-dataflow schema. We don't pull in the otap-df-config crate

controller/src/config/precompute.rs → controller/src/emit/precompute.rs

@@ -3,7 +3,7 @@ use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use std::time::Duration;
 
-use crate::analyzer::format_duration;
+use crate::pipeline::format_duration;
 use crate::types::*;
 
 // ── Scheduling rule ───────────────────────────────────────────────────────────

controller/src/config/stage_config.rs → controller/src/emit/stage_config.rs

@@ -1,5 +1,5 @@
 //! Phase B (MVP v6) — turn a typed L5 [`StageConfig`] map (produced by
-//! [`crate::stage_split::ThreeStageEmitter`]) into the **wire bytes** the
+//! [`crate::physical::colored_dag::ThreeStageEmitter`]) into the **wire bytes** the
 //! three executors actually consume:
 //!
 //! - [`emit_edge_yaml`] → OTel-collector YAML for the edge agent (OTLP
@@ -40,12 +40,12 @@ use std::collections::HashMap;
 
 use crate::sketch_algebra::params::{SketchKind, SketchParams};
 use crate::sketch_algebra::sketch_expr::EstimateOp;
-use crate::stage_split::emitter::{
+use crate::physical::colored_dag::emitter::{
     AggregationInput, ArchiveTierMetric, BackendAggregation, BackendReadout, BackendStageConfig,
     EdgeSketchProcessor, EdgeStageConfig, ExportTarget, GatewayMergeProcessor, GatewayStageConfig,
     PrometheusArchiveMetric,
 };
-use crate::stage_split::stage_id::StageId;
+use crate::physical::colored_dag::stage_id::StageId;
 
 // ── YAML structural types ─────────────────────────────────────────────────────
 //
@@ -1134,7 +1134,7 @@ tsdb_block_duration: {window_secs}s\n",
 
 /// Map a `SketchKind` to the OTel processor name registered by the
 /// patched contrib build's factory. Keep in sync with
-/// `crate::stage_split::emitter::edge_processor_name`.
+/// `crate::physical::colored_dag::emitter::edge_processor_name`.
 fn sketch_kind_to_processor_name(kind: &SketchKind) -> &'static str {
     match kind {
         SketchKind::DDSketch => "ddsketch",

controller/src/config/stage_config_telegraf.rs → controller/src/emit/telegraf.rs

@@ -38,8 +38,8 @@
 use anyhow::{Context, Result};
 
 use crate::sketch_algebra::params::{SketchKind, SketchParams};
-use crate::stage_split::emitter::{EdgeStageConfig, EdgeSketchProcessor, ExportTarget};
-use crate::stage_split::stage_id::StageId;
+use crate::physical::colored_dag::emitter::{EdgeStageConfig, EdgeSketchProcessor, ExportTarget};
+use crate::physical::colored_dag::stage_id::StageId;
 
 /// Default Prometheus remote-write URL for Mode 3 — Telegraf doesn't
 /// support OTLP-HTTP egress, so we land in the same Prometheus archive
@@ -296,7 +296,7 @@ mod toml_minimal {
 mod tests {
     use super::*;
     use crate::sketch_algebra::params::{DDSketchParams, KllParams};
-    use crate::stage_split::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric};
+    use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric};
 
     fn ddsketch_edge_cfg_mode1() -> EdgeStageConfig {
         EdgeStageConfig {

controller/src/emit/trait_def.rs

@@ -0,0 +1,37 @@
+//! `PlanEmitter` trait — placeholder scaffolding for the L5 output
+//! surface declared in `controller/docs/design.md` §5 / §6
+//! `core::emit::PlanEmitter`.
+//!
+//! Refactor 2026-05 created this module so the design.md target layout
+//! is materialised in code. The existing per-runtime emitters
+//! ([`super::stage_config::emit_edge_yaml`], [`super::otap::emit_otap_dag_yaml`],
+//! [`super::telegraf::emit_telegraf_toml`], [`super::asapquery_backend::generate_streaming_config_yaml`],
+//! [`super::agent::generate_agent_config`], [`super::backend::generate_backend_config`])
+//! still operate as free functions with deployment-specific signatures.
+//! Migrating each onto this trait is a follow-up — until then this
+//! module is intentionally minimal so it has zero behavior impact.
+
+#![allow(dead_code)]
+
+use anyhow::Result;
+
+/// `PlanEmitter` — every per-deployment-model plan emitter implements
+/// this so a future controller pipeline can call them polymorphically.
+///
+/// The `Input` associated type is the typed L5 stage config the emitter
+/// consumes (e.g. `EdgeStageConfig` for OTel YAML emitters,
+/// `BackendStageConfig` for the ASAPQuery-backend `StreamingConfig`
+/// emitter). `Output` is the wire-format string / JSON the deployment
+/// model's transport expects.
+pub trait PlanEmitter: Send + Sync {
+    /// Typed L5 stage config this emitter consumes.
+    type Input;
+    /// Wire-format output produced for the deployment model's transport.
+    type Output;
+
+    /// Stable name for diagnostics + the emitter registry.
+    fn name(&self) -> &'static str;
+
+    /// Emit the wire-format payload for the given stage config.
+    fn emit(&self, input: &Self::Input) -> Result<Self::Output>;
+}