crabka-client-streams 0.3.6

//! Topology builder: public Processor-API surface.

use std::any::Any;
use std::borrow::Borrow;
use std::collections::{BTreeMap, HashMap};
use std::marker::PhantomData;

/// Factory that builds a fresh erased [`StateStore`] given the store name, pre-derived
/// changelog topic, and an already-opened byte backend. The factory only owns the serdes.
pub(crate) type StoreFactory = Box<
    dyn Fn(
            &str,
            String,
            Box<dyn crate::store::byte::ByteKeyValueStore>,
        ) -> Box<dyn crate::store::api::StateStore>
        + Send
        + Sync,
>;

use crabka_protocol::owned::streams_group_heartbeat_request::Topology as WireTopology;

use super::grouping::group_nodes;
use super::node::{NodeKind, NodeRegistry};
use super::wire::to_wire;
use crate::processor::api::ProcessorSupplier;
use crate::processor::erased::ProcessorError;
use crate::processor::factory::{MakeDeser, NodeFactory};
use crate::processor::graph::{Graph, GraphSource};
use crate::processor::node::{ErasedNode, ProcessorNode, SinkNode, SourceNode};
use crate::processor::serde::{Consumed, DefaultSerde, Produced, Serde};

// ──────────────────────────────────────────────────────────────────────────────
// TopologyError
// ──────────────────────────────────────────────────────────────────────────────

/// Error building a topology (bad node graph, invalid configuration, etc.).
///
/// Parent→child *type* mismatches are not represented here: typed
/// [`NodeHandle`] wiring makes them a compile error, so they never reach
/// `build()`.
#[derive(Debug, thiserror::Error)]
pub enum TopologyError {
    #[error("duplicate node name: {0}")]
    DuplicateNode(String),
    #[error("node {node} references unknown predecessor {predecessor}")]
    UnknownPredecessor { node: String, predecessor: String },
    #[error("topology has no source nodes")]
    Empty,
}

// Cloneable error subset.
#[derive(Debug, Clone)]
enum StoredError {
    DuplicateNode(String),
    UnknownPredecessor { node: String, predecessor: String },
    Empty,
}

impl From<StoredError> for TopologyError {
    fn from(e: StoredError) -> Self {
        match e {
            StoredError::DuplicateNode(n) => TopologyError::DuplicateNode(n),
            StoredError::UnknownPredecessor { node, predecessor } => {
                TopologyError::UnknownPredecessor { node, predecessor }
            }
            StoredError::Empty => TopologyError::Empty,
        }
    }
}

impl From<TopologyError> for StoredError {
    fn from(e: TopologyError) -> Self {
        match e {
            TopologyError::DuplicateNode(n) => StoredError::DuplicateNode(n),
            TopologyError::UnknownPredecessor { node, predecessor } => {
                StoredError::UnknownPredecessor { node, predecessor }
            }
            TopologyError::Empty => StoredError::Empty,
        }
    }
}

// ──────────────────────────────────────────────────────────────────────────────
// Topology
// ──────────────────────────────────────────────────────────────────────────────

/// A Processor-API topology under construction. Node insertion order is
/// significant — it determines subtopology indices (JVM-matching).
#[derive(Default)]
pub struct Topology {
    reg: NodeRegistry,
    error: Option<StoredError>,
    factories: HashMap<String, NodeFactory>,
    /// `(changelog_override, factory)` — override is `None` for the default
    /// `<app_id>-<store_name>-changelog` derivation.
    store_factories: HashMap<String, (Option<String>, StoreFactory)>,
    /// `GlobalKTable` store factories, keyed by store name. Kept SEPARATE from
    /// `store_factories` so per-task `instantiate` does NOT build them (a global
    /// store is fully replicated, not task-partitioned) and NO changelog topic is
    /// emitted. The override is always `None` here (global stores have no
    /// changelog) but the tuple mirrors the regular-store shape consumed by the
    /// global-store manager. Populated by [`Topology::add_global_store`].
    global_store_factories: HashMap<String, (Option<String>, StoreFactory)>,
    /// `global store name -> source topic` for each `GlobalKTable`. The shared
    /// global consumer reads each source topic (all partitions) to fully
    /// replicate the matching store. Populated by [`Topology::add_global_store`]
    /// alongside `global_store_factories`. Invisible in the wire output.
    global_store_topics: HashMap<String, String>,
    /// Materialized KV stores whose record cache is eligible (JVM `Materialized`
    /// caching, default on). Populated by [`Topology::mark_store_caching`] at the
    /// materialized KTable/aggregate lowering sites; consumed by `instantiate` to
    /// wrap the store in a [`NamedCache`] when `cache_max_bytes > 0`.
    caching_stores: std::collections::HashSet<String>,
}

impl std::fmt::Debug for Topology {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Topology")
            .field("reg", &self.reg)
            .field("error", &self.error)
            .field(
                "factories",
                &format!("<{} factories>", self.factories.len()),
            )
            .field(
                "store_factories",
                &format!("<{} store_factories>", self.store_factories.len()),
            )
            .field(
                "global_store_factories",
                &format!(
                    "<{} global_store_factories>",
                    self.global_store_factories.len()
                ),
            )
            .field("global_store_topics", &self.global_store_topics)
            .field("caching_stores", &self.caching_stores)
            .finish()
    }
}

/// A typed handle to a node in a [`Topology`], returned by
/// [`Topology::add_source`] and [`Topology::add_processor`]. Pass it (by
/// reference) as a parent when adding a child node.
///
/// Wiring by value rather than by string name means the compiler does two jobs
/// that `build()` used to: a parent that doesn't exist yet simply has no handle
/// to pass (so forward references and cycles can't be written), and the
/// parent's output type `(K, V)` is checked against the child's input type —
/// a mismatch is a **compile error**, not a runtime `build()` failure.
///
/// `K` / `V` are the key/value types the node *produces*. The handle is cheap
/// to [`Clone`] (it carries only the node name) so one parent can feed many
/// children.
pub struct NodeHandle<K, V> {
    name: String,
    _pd: PhantomData<fn() -> (K, V)>,
}

impl<K, V> NodeHandle<K, V> {
    fn new(name: String) -> Self {
        Self {
            name,
            _pd: PhantomData,
        }
    }

    /// Reconstruct a typed handle from a node name recorded during DSL lowering.
    ///
    /// The DSL lowers a type-erased logical graph: each lowering thunk knows its
    /// own concrete `K`/`V` statically and looks up its parent's
    /// Processor-API node name from `LowerState`, rebuilding a typed handle to
    /// pass to [`Topology::add_processor`] / [`Topology::add_sink`].
    pub(crate) fn from_name(name: String) -> Self {
        Self::new(name)
    }

    /// The node's name, as it appears in the wire topology. Useful for
    /// [`Topology::add_state_store`], which connects stores to processors by
    /// name.
    #[must_use]
    pub fn name(&self) -> &str {
        &self.name
    }
}

impl<K, V> Clone for NodeHandle<K, V> {
    fn clone(&self) -> Self {
        Self::new(self.name.clone())
    }
}

impl<K, V> std::fmt::Debug for NodeHandle<K, V> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NodeHandle")
            .field("name", &self.name)
            .finish()
    }
}

impl Topology {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Add a source node reading the given external topics with the default
    /// serdes for `K` and `V`, returning a typed [`NodeHandle`] used to wire
    /// children to it.
    ///
    /// Use [`Topology::add_source_explicit`] when either type has no default
    /// serde or when a topology needs custom serdes.
    pub fn add_source<K, V>(
        &mut self,
        name: impl Into<String>,
        topics: impl IntoIterator<Item = impl Into<String>>,
    ) -> NodeHandle<K, V>
    where
        K: Any + Send + Sync + Clone + DefaultSerde,
        V: Any + Send + Sync + Clone + DefaultSerde,
    {
        self.add_source_explicit::<K, V, K::Serde, V::Serde>(
            name,
            topics,
            Consumed::with(K::Serde::default(), V::Serde::default()),
        )
    }

    /// Add a source node reading the given external topics with explicit serdes,
    /// returning a typed [`NodeHandle`] used to wire children to it.
    ///
    /// `consumed` carries the key + value serdes used to deserialize incoming
    /// bytes into typed `Record<K, V>` values at runtime — written
    /// `Consumed::with(key_serde, value_serde)` so the two roles are visible.
    ///
    /// Prefer [`Topology::add_source`] (the default-serde form) for types that
    /// implement [`DefaultSerde`]. Reach for this escape hatch when a type has no
    /// default serde, or to override the default for a topic — e.g. a hand-rolled
    /// `Serde<T>`, a **key**-role schema serde (`AvroSerde::<T>::key(&cache)`), or
    /// a validation-on JSON serde (`JsonSerde::value(&cache, true)`).
    pub fn add_source_explicit<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        topics: impl IntoIterator<Item = impl Into<String>>,
        consumed: impl Into<Consumed<KS, VS>>,
    ) -> NodeHandle<K, V>
    where
        K: Any + Send + Clone,
        V: Any + Send + Clone,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let Consumed {
            key_serde,
            value_serde,
        } = consumed.into();
        let name: String = name.into();
        let topics: Vec<String> = topics.into_iter().map(Into::into).collect();
        // Let each serde pre-register any per-topic state (e.g. a schema-registry
        // subject) so membership pre-warm can resolve ids before processing.
        for t in &topics {
            key_serde.prepare(t, crate::processor::serde::SerdeRole::Key);
            value_serde.prepare(t, crate::processor::serde::SerdeRole::Value);
        }
        let r = self.reg.add_source(&name, topics.clone());
        self.record(r);

        // Build the make_deser factory closure.
        let make_deser: MakeDeser = {
            let ks = key_serde;
            let vs = value_serde;
            let n = name.clone();
            Box::new(move || {
                let node = SourceNode::new(n.clone(), ks.clone(), vs.clone());
                Box::new(move |k: Option<&[u8]>, v: &[u8], ts: i64| node.deserialize(k, v, ts))
                    as Box<dyn Fn(Option<&[u8]>, &[u8], i64) -> Result<_, ProcessorError> + Send>
            })
        };

        self.factories.insert(
            name.clone(),
            NodeFactory {
                make_node: None,
                make_deser: Some(make_deser),
            },
        );
        NodeHandle::new(name)
    }

    /// Add a processor node fed by the given parent [`NodeHandle`]s, returning a
    /// handle for its output.
    ///
    /// Wiring is by value: every parent's output type must equal this
    /// processor's input type `(KIn, VIn)`, enforced by the compiler. `supplier`
    /// produces a fresh `Processor` per task; the closure form `|| MyProc`
    /// satisfies [`ProcessorSupplier`] via a blanket impl and infers all four KV
    /// type parameters from the processor's `Processor` impl, so callers never
    /// annotate them.
    pub fn add_processor<KIn, VIn, KOut, VOut, S, P, I>(
        &mut self,
        name: impl Into<String>,
        supplier: S,
        parents: I,
    ) -> NodeHandle<KOut, VOut>
    where
        KIn: Any + Send,
        VIn: Any + Send,
        KOut: Any + Send + Clone,
        VOut: Any + Send + Clone,
        S: ProcessorSupplier<KIn, VIn, KOut, VOut> + Clone,
        I: IntoIterator<Item = P>,
        P: Borrow<NodeHandle<KIn, VIn>>,
    {
        let name: String = name.into();
        let preds: Vec<String> = parents
            .into_iter()
            .map(|p| p.borrow().name.clone())
            .collect();
        let r = self.reg.add_processor(&name, preds);
        self.record(r);

        let make_node: crate::processor::factory::MakeNode = {
            let n = name.clone();
            let s = supplier;
            Box::new(move || {
                Box::new(ProcessorNode::<KIn, VIn, KOut, VOut>::new(n.clone(), &s))
                    as Box<dyn ErasedNode>
            })
        };

        self.factories.insert(
            name.clone(),
            NodeFactory {
                make_node: Some(make_node),
                make_deser: None,
            },
        );
        NodeHandle::new(name)
    }

    /// Add a sink node writing to `topic` with the default serdes for `K` and
    /// `V`, fed by the given parent [`NodeHandle`]s. Every parent's output type
    /// must equal the sink's input type `(K, V)` — enforced by the compiler.
    ///
    /// Use [`Topology::add_sink_explicit`] when either type has no default serde
    /// or when a topology needs custom serdes.
    pub fn add_sink<K, V>(
        &mut self,
        name: impl Into<String>,
        topic: impl Into<String>,
        parents: impl IntoIterator<Item = impl Borrow<NodeHandle<K, V>>>,
    ) where
        K: Any + Send + Sync + DefaultSerde,
        V: Any + Send + Sync + DefaultSerde,
    {
        self.add_sink_explicit::<K, V, K::Serde, V::Serde, _, _>(
            name,
            topic,
            parents,
            Produced::with(K::Serde::default(), V::Serde::default()),
        );
    }

    /// Add a sink node writing to `topic` with explicit serdes, fed by the given
    /// parent [`NodeHandle`]s. Every parent's output type must equal the sink's
    /// input type `(K, V)` — enforced by the compiler.
    ///
    /// `produced` carries the key + value serdes used to serialize outgoing
    /// records — written `Produced::with(key_serde, value_serde)`. A sink is
    /// terminal, so nothing is returned.
    ///
    /// Prefer [`Topology::add_sink`] (the default-serde form) for types that
    /// implement [`DefaultSerde`]. Reach for this escape hatch when a type has no
    /// default serde, or to override the default for a topic — e.g. a hand-rolled
    /// `Serde<T>`, a **key**-role schema serde (`AvroSerde::<T>::key(&cache)`), or
    /// a validation-on JSON serde (`JsonSerde::value(&cache, true)`).
    pub fn add_sink_explicit<K, V, KS, VS, P, I>(
        &mut self,
        name: impl Into<String>,
        topic: impl Into<String>,
        parents: I,
        produced: impl Into<Produced<KS, VS>>,
    ) where
        K: Any + Send,
        V: Any + Send,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
        I: IntoIterator<Item = P>,
        P: Borrow<NodeHandle<K, V>>,
    {
        let Produced {
            key_serde,
            value_serde,
        } = produced.into();
        let name: String = name.into();
        let topic: String = topic.into();
        // Pre-register per-topic serde state (e.g. schema-registry subject).
        key_serde.prepare(&topic, crate::processor::serde::SerdeRole::Key);
        value_serde.prepare(&topic, crate::processor::serde::SerdeRole::Value);
        let preds: Vec<String> = parents
            .into_iter()
            .map(|p| p.borrow().name.clone())
            .collect();
        let r = self.reg.add_sink(&name, topic.clone(), preds);
        self.record(r);

        let make_node: crate::processor::factory::MakeNode = {
            let n = name.clone();
            let t = topic.clone();
            let ks = key_serde;
            let vs = value_serde;
            Box::new(move || {
                Box::new(SinkNode::<K, V, KS, VS>::new(
                    n.clone(),
                    t.clone(),
                    ks.clone(),
                    vs.clone(),
                )) as Box<dyn ErasedNode>
            })
        };

        self.factories.insert(
            name,
            NodeFactory {
                make_node: Some(make_node),
                make_deser: None,
            },
        );
    }

    /// Register a state store connected to the given processors (→ changelog).
    ///
    /// `key_serde` / `value_serde` define how records are serialized into the
    /// changelog topic (`<app_id>-<name>-changelog`) and the store's byte map.
    /// Stores connect processors of possibly-differing types, so the processor
    /// list is by name — pass [`NodeHandle::name`] for a handle you hold.
    pub fn add_state_store<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        processors: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        self.add_state_store_inner::<K, V, KS, VS>(name, key_serde, value_serde, processors, None)
    }

    /// Register a state store whose changelog is an existing **source topic**
    /// rather than the derived `<app_id>-<name>-changelog`.
    ///
    /// This backs the `REUSE_KTABLE_SOURCE_TOPICS` DSL optimizer: a
    /// `builder.table_explicit(topic, ...)` store can reuse `topic` as its changelog, so
    /// no separate `app-<store>-changelog` topic is created and the wire
    /// topology lists `topic` as the store's changelog. `changelog_topic` is the
    /// topic name used both in the wire `state_changelog_topics` entry and as the
    /// runtime store's changelog target.
    pub fn add_state_store_with_changelog<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        processors: impl IntoIterator<Item = impl Into<String>>,
        changelog_topic: impl Into<String>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        self.add_state_store_inner::<K, V, KS, VS>(
            name,
            key_serde,
            value_serde,
            processors,
            Some(changelog_topic.into()),
        )
    }

    fn add_state_store_inner<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        processors: impl IntoIterator<Item = impl Into<String>>,
        changelog_override: Option<String>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let name: String = name.into();
        let procs: Vec<String> = processors.into_iter().map(Into::into).collect();
        self.reg.add_store(&name, procs, changelog_override.clone());
        self.store_factories.insert(
            name,
            (
                changelog_override,
                Box::new(
                    move |store_name: &str,
                          changelog: String,
                          backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        Box::new(crate::store::kv::KeyValueBytesStore::<K, V>::new(
                            store_name.to_string(),
                            backend,
                            Box::new(key_serde.clone()),
                            Box::new(value_serde.clone()),
                            changelog,
                        )) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a windowed state store connected to the given processors.
    ///
    /// Like [`add_state_store`] but for windowed stores. The changelog topic
    /// carries `compact,delete` configs and a `retention.ms` derived from
    /// `size_ms + grace_ms + 86_400_000` (the JVM's
    /// `windowstore.changelog.additional.retention.ms` default of 1 day).
    ///
    /// [`add_state_store`]: Topology::add_state_store
    ///
    /// `size_ms` is the **retention basis**: it drives the changelog
    /// `retention.ms = size_ms + grace_ms + 86_400_000`. For tumbling/hopping
    /// windows it equals the window size; for sliding windows it is the
    /// retention span (`2 * timeDifferenceMs`), which is wider than the window.
    ///
    /// `window_size_ms` is the **true window length** used only by the store's
    /// cache flush to reconstruct the downstream `Windowed<K>` key's
    /// `end = start + window_size_ms` (the store-key bytes hold only the start).
    /// For sliding windows this is `timeDifferenceMs` (1×), NOT the retention
    /// span.
    #[allow(clippy::too_many_arguments)] // size_ms (retention basis) + window_size_ms (key-end) are distinct
    pub fn add_window_store<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        size_ms: i64,
        window_size_ms: i64,
        grace_ms: i64,
        processors: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let name: String = name.into();
        // windowstore.changelog.additional.retention.ms default = 1 day (86_400_000 ms)
        let retention_ms = size_ms + grace_ms + 86_400_000;
        let procs: Vec<String> = processors.into_iter().map(Into::into).collect();
        self.reg.add_window_store(&name, procs, None, retention_ms);
        self.store_factories.insert(
            name.clone(),
            (
                None,
                Box::new(
                    move |store_name: &str,
                          changelog: String,
                          backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        Box::new(crate::store::window::WindowBytesStore::<K, V>::new(
                            store_name.to_string(),
                            backend,
                            Box::new(key_serde.clone()),
                            Box::new(value_serde.clone()),
                            changelog,
                            // Aggregate window stores need the real window size so the
                            // cache flush can reconstruct `end = start + window_size_ms`
                            // for the downstream `Windowed` key (the store-key bytes hold
                            // only the start). This is distinct from `size_ms`, which is
                            // the retention basis — they diverge for sliding windows.
                            window_size_ms,
                        )) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a join window state store connected to the given processors.
    ///
    /// Like [`add_window_store`] but for join window stores (retainDuplicates).
    /// The changelog topic carries `delete`-only configs and a `retention.ms`
    /// derived from `before_ms + after_ms + grace_ms + 86_400_000`. Compaction is
    /// not applicable because the store retains duplicates.
    ///
    /// [`add_window_store`]: Topology::add_window_store
    #[allow(clippy::too_many_arguments)] // mirrors add_window_store + extra before_ms/after_ms split
    pub fn add_join_window_store<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        before_ms: i64,
        after_ms: i64,
        grace_ms: i64,
        processors: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let name: String = name.into();
        // windowstore.changelog.additional.retention.ms default = 1 day (86_400_000 ms)
        let retention_ms = before_ms + after_ms + grace_ms + 86_400_000;
        let procs: Vec<String> = processors.into_iter().map(Into::into).collect();
        self.reg
            .add_join_window_store(&name, procs, None, retention_ms);
        self.store_factories.insert(
            name.clone(),
            (
                None,
                Box::new(
                    move |store_name: &str,
                          changelog: String,
                          backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        Box::new(crate::store::join_window::JoinWindowBytesStore::<K, V>::new(
                            store_name.to_string(),
                            backend,
                            Box::new(key_serde.clone()),
                            Box::new(value_serde.clone()),
                            changelog,
                        )) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a session state store connected to the given processors.
    ///
    /// Like [`add_window_store`] but for session stores. Reuses the windowed
    /// (`compact,delete`) changelog config; the `retention.ms` is derived from
    /// `gap_ms + grace_ms + 86_400_000` (JVM `windowstore.changelog.additional.
    /// retention.ms` default of 1 day). The store holds the raw aggregate
    /// (`SessionBytesStore`).
    ///
    /// [`add_window_store`]: Topology::add_window_store
    pub fn add_session_store<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        gap_ms: i64,
        grace_ms: i64,
        processors: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let name: String = name.into();
        let retention_ms = gap_ms + grace_ms + 86_400_000;
        let procs: Vec<String> = processors.into_iter().map(Into::into).collect();
        // Session changelog == windowed changelog (compact,delete + retention);
        // reuse the AggWindow ChangelogKind via add_window_store.
        self.reg.add_window_store(&name, procs, None, retention_ms);
        self.store_factories.insert(
            name.clone(),
            (
                None,
                Box::new(
                    move |store_name: &str,
                          changelog: String,
                          backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        Box::new(crate::store::session::SessionBytesStore::<K, V>::new(
                            store_name.to_string(),
                            backend,
                            Box::new(key_serde.clone()),
                            Box::new(value_serde.clone()),
                            changelog,
                        )) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a versioned state store (KIP-889) connected to the given
    /// processors. The changelog topic carries `compact` + `min.compaction.lag.ms
    /// = history_retention_ms + 86_400_000`. The version-chain store is
    /// self-contained in memory; the supplied byte backend is unused.
    pub fn add_versioned_store<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        history_retention_ms: i64,
        processors: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self
    where
        K: Send + Sync + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        self.add_versioned_store_inner::<K, V, KS, VS>(
            name,
            key_serde,
            value_serde,
            history_retention_ms,
            processors,
            None,
        )
    }

    /// Like [`add_versioned_store`] but whose changelog is an existing **source
    /// topic** (the `REUSE_KTABLE_SOURCE_TOPICS` optimizer points a versioned
    /// `builder.table_explicit(topic, …)` store's changelog at its own source
    /// `topic`). Mirrors [`add_state_store_with_changelog`].
    ///
    /// [`add_versioned_store`]: Topology::add_versioned_store
    /// [`add_state_store_with_changelog`]: Topology::add_state_store_with_changelog
    pub fn add_versioned_store_with_changelog<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        history_retention_ms: i64,
        processors: impl IntoIterator<Item = impl Into<String>>,
        changelog_topic: impl Into<String>,
    ) -> &mut Self
    where
        K: Send + Sync + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        self.add_versioned_store_inner::<K, V, KS, VS>(
            name,
            key_serde,
            value_serde,
            history_retention_ms,
            processors,
            Some(changelog_topic.into()),
        )
    }

    fn add_versioned_store_inner<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        history_retention_ms: i64,
        processors: impl IntoIterator<Item = impl Into<String>>,
        changelog_override: Option<String>,
    ) -> &mut Self
    where
        K: Send + Sync + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let name: String = name.into();
        let min_compaction_lag_ms = history_retention_ms + 86_400_000;
        let procs: Vec<String> = processors.into_iter().map(Into::into).collect();
        self.reg.add_versioned_store(
            &name,
            procs,
            changelog_override.clone(),
            min_compaction_lag_ms,
        );
        self.store_factories.insert(
            name.clone(),
            (
                changelog_override,
                Box::new(
                    move |store_name: &str,
                          changelog: String,
                          _backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        Box::new(
                            crate::store::versioned::VersionedBytesStore::<K, V>::new(
                                store_name.to_string(),
                                history_retention_ms,
                                Box::new(key_serde.clone()),
                                Box::new(value_serde.clone()),
                                changelog,
                            ),
                        ) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a KIP-213 FK-join subscription store connected to the given
    /// processors.
    ///
    /// The subscription store is keyed by `combined_key(fk, pk)` bytes →
    /// `ValueAndTimestamp<SubscriptionWrapper>` bytes. Its changelog is a plain
    /// compacted KV changelog (`<app_id>-<name>-changelog`, like
    /// [`add_state_store`]) — NOT windowed retention. The store types are fixed
    /// (raw bytes in, `SubscriptionWrapper` out), so no key/value serdes are
    /// taken.
    ///
    /// [`add_state_store`]: Topology::add_state_store
    pub(crate) fn add_fk_subscription_store(
        &mut self,
        name: impl Into<String>,
        processors: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self {
        let name: String = name.into();
        let procs: Vec<String> = processors.into_iter().map(Into::into).collect();
        self.reg.add_store(&name, procs, None); // plain compact changelog
        self.store_factories.insert(
            name,
            (
                None,
                Box::new(
                    move |store_name: &str,
                          changelog: String,
                          backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        Box::new(crate::store::fk_subscription::SubscriptionBytesStore::new(
                            store_name.to_string(),
                            backend,
                            changelog,
                        )) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a suppress buffer store connected to the given processor.
    ///
    /// The suppress buffer ([`SuppressBytesStore`]) is a time-ordered in-memory
    /// buffer with its own storage (it does NOT use the pluggable byte backend),
    /// so the factory ignores the opened backend. `logging` toggles ONLY the
    /// changelog: when `true` the changelog topic is emitted in the wire topology
    /// (a plain `cleanup.policy=compact` changelog — the JVM suppress buffer is a
    /// compacted KV store) and the store logs/restores; when `false` the store
    /// stays in memory and NO changelog topic appears (so a logging-off suppress
    /// has the same wire shape as an unsuppressed topology.
    ///
    /// [`SuppressBytesStore`]: crate::store::suppress_store::SuppressBytesStore
    pub fn add_suppress_store<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        logging: bool,
        processors: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let name: String = name.into();
        let procs: Vec<String> = processors.into_iter().map(Into::into).collect();
        // Logging toggles ONLY the changelog topic; the runtime store is always
        // registered so the processor can buffer through it either way. The suppress
        // changelog is a plain compacted KV changelog (ChangelogKind::Kv).
        if logging {
            self.reg.add_store(&name, procs, None);
        }
        self.store_factories.insert(
            name.clone(),
            (
                None,
                Box::new(
                    move |store_name: &str,
                          changelog: String,
                          _backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        // logging off → empty changelog (never flushed) + flag off.
                        let cl = if logging { changelog } else { String::new() };
                        let mut store =
                            crate::store::suppress_store::SuppressBytesStore::<K, V>::new(
                                store_name.to_string(),
                                Box::new(key_serde.clone()),
                                Box::new(value_serde.clone()),
                                cl,
                            );
                        if !logging {
                            crate::store::api::StateStore::set_logging(&mut store, false);
                        }
                        Box::new(store) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a join-grace buffer store (KIP-923) connected to the given
    /// processor.
    ///
    /// The grace buffer ([`JoinGraceBufferStore`]) is a stream-side, time-ordered
    /// in-memory buffer with its own storage (it does NOT use the pluggable byte
    /// backend), so the factory ignores the opened backend. Its changelog is a
    /// COMPACTED KV changelog (`ChangelogKind::Kv` → `cleanup.policy=compact`,
    /// `message.timestamp.type=CreateTime`, NO explicit `retention.ms`), pinned
    /// from the JVM 4.1.0 capture. `logging` toggles ONLY the changelog: when
    /// `true` the changelog topic is emitted and the store logs/restores; when
    /// `false` the store stays in memory and NO changelog topic appears.
    ///
    /// [`JoinGraceBufferStore`]: crate::store::join_grace_buffer::JoinGraceBufferStore
    pub(crate) fn add_join_grace_store<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
        logging: bool,
        processors: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let name: String = name.into();
        let procs: Vec<String> = processors.into_iter().map(Into::into).collect();
        // Logging toggles ONLY the changelog topic; the runtime store is always
        // registered so the processor can buffer through it either way. The grace
        // buffer changelog is a plain compacted KV changelog (ChangelogKind::Kv).
        if logging {
            self.reg.add_store(&name, procs, None);
        }
        self.store_factories.insert(
            name.clone(),
            (
                None,
                Box::new(
                    move |store_name: &str,
                          changelog: String,
                          _backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        // logging off → empty changelog (never flushed) + flag off.
                        let cl = if logging { changelog } else { String::new() };
                        let mut store =
                            crate::store::join_grace_buffer::JoinGraceBufferStore::<K, V>::new(
                                store_name.to_string(),
                                Box::new(key_serde.clone()),
                                Box::new(value_serde.clone()),
                                cl,
                            );
                        if !logging {
                            crate::store::api::StateStore::set_logging(&mut store, false);
                        }
                        Box::new(store) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a state store **without** a changelog topic.
    ///
    /// The store is available at runtime (for in-memory state), but NO entry is
    /// emitted in the wire topology's `state_changelog_topics` array. This backs
    /// `Materialized::with_logging(false)` in the DSL.
    pub(crate) fn add_state_store_no_changelog<K, V, KS, VS>(
        &mut self,
        name: impl Into<String>,
        key_serde: KS,
        value_serde: VS,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let name: String = name.into();
        // Insert only into store_factories (runtime use) — NOT into reg.stores,
        // so no changelog topic appears in the wire topology.
        self.store_factories.insert(
            name,
            (
                None, // no changelog
                Box::new(
                    move |store_name: &str,
                          _changelog: String,
                          backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                        // No changelog: pass empty string so the store never flushes.
                        Box::new(crate::store::kv::KeyValueBytesStore::<K, V>::new(
                            store_name.to_string(),
                            backend,
                            Box::new(key_serde.clone()),
                            Box::new(value_serde.clone()),
                            String::new(),
                        )) as Box<dyn crate::store::api::StateStore>
                    },
                ),
            ),
        );
        self
    }

    /// Register a `GlobalKTable` source, update-processor, and KV store.
    ///
    /// A `GlobalKTable` is **invisible in the wire**: no subtopology of its own,
    /// no changelog topic. But its source node still occupies a node-group index
    /// during grouping (so other subtopology ids shift). This method:
    ///
    /// 1. registers a source node reading `topic` and a processor node fed by it
    ///    (the source→processor edge unites them into one node group);
    /// 2. marks `topic` global so the grouping pass skips it in the
    ///    source-bucketing pass — the resulting
    ///    source-less group is dropped by the final filter but already consumed
    ///    its index;
    /// 3. stores the global KV factory in a SEPARATE map (`global_store_factories`,
    ///    NOT `store_factories`) so per-task `instantiate` does not build it and NO
    ///    changelog topic is emitted. The factory builds a
    ///    [`KeyValueBytesStore`] with an empty changelog.
    ///
    /// The global store is not built by per-task `instantiate`; the
    /// fully-replicated global-store manager reads the factory.
    ///
    /// [`KeyValueBytesStore`]: crate::store::kv::KeyValueBytesStore
    pub fn add_global_store<K, V, KS, VS>(
        &mut self,
        store_name: impl Into<String>,
        source_name: impl Into<String>,
        topic: impl Into<String>,
        processor_name: impl Into<String>,
        consumed: impl Into<Consumed<KS, VS>>,
    ) -> &mut Self
    where
        K: Send + 'static,
        V: Send + 'static,
        KS: Serde<K> + Clone,
        VS: Serde<V> + Clone,
    {
        let store_name: String = store_name.into();
        let source_name: String = source_name.into();
        let topic: String = topic.into();
        let processor_name: String = processor_name.into();
        let Consumed {
            key_serde,
            value_serde,
        } = consumed.into();

        // (a) source node reading the global topic (consumes a node-group index).
        let r = self.reg.add_source(&source_name, vec![topic.clone()]);
        self.record(r);
        // (b) update-processor fed by the source — the edge unites them.
        let r = self
            .reg
            .add_processor(&processor_name, vec![source_name.clone()]);
        self.record(r);
        // (c) mark the topic global so grouping skips it (group ends source-less).
        self.reg.add_global_source(&topic);

        // (d) global KV factory, in the SEPARATE global map. The override is
        //     `None`; the factory uses an empty changelog so the store never
        //     flushes. `instantiate` only iterates `store_factories`, so it ignores
        //     this and no changelog topic is emitted.
        let factory: StoreFactory = Box::new(
            move |sn: &str,
                  _changelog: String,
                  backend: Box<dyn crate::store::byte::ByteKeyValueStore>| {
                Box::new(crate::store::kv::KeyValueBytesStore::<K, V>::new(
                    sn.to_string(),
                    backend,
                    Box::new(key_serde.clone()),
                    Box::new(value_serde.clone()),
                    String::new(),
                )) as Box<dyn crate::store::api::StateStore>
            },
        );
        self.global_store_topics.insert(store_name.clone(), topic);
        self.global_store_factories
            .insert(store_name, (None, factory));
        self
    }

    /// Connect an additional processor to an already-registered state store.
    ///
    /// Mirrors `InternalTopologyBuilder.connectProcessorAndStateStores`: a join
    /// processor needs to read the joined table's store even though it wasn't the
    /// store's original owner. The grouping pass unions all connected processors
    /// into one subtopology, so adding a second processor here is sufficient to
    /// pull the join processor into the same subtopology as the store.
    pub fn connect_processor_store(&mut self, processor: &str, store: &str) -> &mut Self {
        self.reg.connect_processor_store(processor, store);
        self
    }

    /// Mark a materialized store as record-cache eligible (JVM `Materialized`
    /// caching, default on). Called by the materialized KTable/aggregate lowering
    /// sites with the `Materialized`'s `caching_enabled()`; `on == false`
    /// (`with_caching(false)`) leaves the store uncached.
    pub(crate) fn mark_store_caching(&mut self, name: &str, on: bool) {
        if on {
            self.caching_stores.insert(name.to_string());
        }
    }

    /// Return the connected-processor list for a store (test helper).
    #[cfg(test)]
    pub(crate) fn store_entry_for_test(&self, store: &str) -> Option<Vec<String>> {
        self.reg
            .stores
            .iter()
            .find(|e| e.name == store)
            .map(|e| e.processors.clone())
    }

    /// Whether a global store factory is registered under `store` (test helper).
    #[cfg(test)]
    pub(crate) fn has_global_store_for_test(&self, store: &str) -> bool {
        self.global_store_factories.contains_key(store)
    }

    /// Register a topic name as an internal repartition topic.
    pub fn add_repartition_topic<S: Into<String>>(&mut self, name: S) -> &mut Self {
        self.reg.repartition_topics.insert(name.into());
        self
    }

    /// Declare a copartition group: the given member topics must share a
    /// partitioning. The grouping pass assigns the group to the subtopology that
    /// reads its members, and the wire layer encodes member names as `int16`
    /// indices into that subtopology's sorted source/repartition arrays. Required
    /// for joins (KIP-1071).
    pub fn add_copartition_group(
        &mut self,
        topics: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut Self {
        self.reg
            .add_copartition_group(topics.into_iter().map(Into::into).collect());
        self
    }

    /// Derive subtopologies and the wire topology. `application_id` drives
    /// internal-topic names (`<app>-<store>-changelog`).
    ///
    /// Parent→child KV types are already guaranteed to match by the typed
    /// [`NodeHandle`] wiring, so `build()` only checks structural invariants
    /// (no duplicate names, every predecessor exists, at least one source). The
    /// wire `Topology` is byte-identical to the untyped implementation.
    pub fn build<S: Into<String>>(
        mut self,
        application_id: S,
    ) -> Result<BuiltTopology, TopologyError> {
        if let Some(e) = self.error.take() {
            return Err(e.into());
        }
        self.reg.validate_predecessors()?;
        let groups = group_nodes(&self.reg);
        if groups.is_empty() {
            return Err(TopologyError::Empty);
        }
        let app = application_id.into();
        let wire = to_wire(&groups, &app);

        // Build source_topics map (unchanged from original).
        let mut source_topics: BTreeMap<String, Vec<String>> = BTreeMap::new();
        for g in &groups {
            let mut all = g.source_topics.clone();
            all.extend(g.repartition_source_topics.iter().cloned());
            source_topics.insert(g.id.clone(), all);
        }

        // ── Identify the GlobalKTable source + update-processor nodes ─────────
        // A `GlobalKTable` source/processor is invisible in the wire AND has no
        // per-task runtime factory (the fully-replicated global store is built by
        // the shared global manager or populated directly by the TestDriver, NOT
        // by per-task `instantiate`). Excluding them from
        // `node_specs` keeps `instantiate` from trying to build a node it has no
        // factory for. A node is global if it reads a global source topic, or if
        // any predecessor is global (nodes are in topological insertion order, so
        // one forward pass suffices).
        let mut global_nodes: std::collections::HashSet<&str> = std::collections::HashSet::new();
        for n in &self.reg.nodes {
            let is_global = match &n.kind {
                NodeKind::Source { topics } => topics
                    .iter()
                    .any(|t| self.reg.global_source_topics.contains(t)),
                NodeKind::Processor { predecessors } | NodeKind::Sink { predecessors, .. } => {
                    predecessors
                        .iter()
                        .any(|p| global_nodes.contains(p.as_str()))
                }
            };
            if is_global {
                global_nodes.insert(n.name.as_str());
            }
        }

        // ── Build node specs for instantiation (excluding global nodes) ───────
        let node_specs: Vec<NodeSpec> = self
            .reg
            .nodes
            .iter()
            .filter(|n| !global_nodes.contains(n.name.as_str()))
            .map(|n| {
                let (predecessors, kind_str, st, sink_t) = match &n.kind {
                    NodeKind::Source { topics } => (Vec::new(), "source", topics.clone(), None),
                    NodeKind::Processor { predecessors } => {
                        (predecessors.clone(), "processor", Vec::new(), None)
                    }
                    NodeKind::Sink {
                        predecessors,
                        topic,
                    } => (
                        predecessors.clone(),
                        "sink",
                        Vec::new(),
                        Some(topic.clone()),
                    ),
                };
                NodeSpec {
                    name: n.name.clone(),
                    kind: kind_str,
                    predecessors,
                    source_topics: st,
                    sink_topic: sink_t,
                }
            })
            .collect();

        // Store name → connected processor names (used by `instantiate` to root a
        // cached store's forwarded changes at its materializing processor's node).
        let store_processors: HashMap<String, Vec<String>> = self
            .reg
            .stores
            .iter()
            .map(|e| (e.name.clone(), e.processors.clone()))
            .collect();

        Ok(BuiltTopology {
            wire,
            source_topics,
            application_id: app,
            factories: self.factories,
            node_specs,
            store_factories: self.store_factories,
            global_store_factories: self.global_store_factories,
            global_store_topics: self.global_store_topics,
            caching_stores: self.caching_stores,
            store_processors,
        })
    }

    fn record(&mut self, r: Result<(), TopologyError>) {
        if self.error.is_none()
            && let Err(e) = r
        {
            self.error = Some(e.into());
        }
    }
}

// ──────────────────────────────────────────────────────────────────────────────
// NodeSpec
// ──────────────────────────────────────────────────────────────────────────────

/// Lightweight description of one node's wiring (no type parameters).
/// Used during `BuiltTopology::instantiate()`.
pub(crate) struct NodeSpec {
    pub name: String,
    /// `"source"` | `"processor"` | `"sink"`
    pub kind: &'static str,
    /// Predecessor names (empty for sources).
    pub predecessors: Vec<String>,
    /// Topics read (sources only; empty otherwise).
    pub source_topics: Vec<String>,
    /// Topic written (sinks only; `None` otherwise).
    pub sink_topic: Option<String>,
}

// ──────────────────────────────────────────────────────────────────────────────
// BuiltTopology
// ──────────────────────────────────────────────────────────────────────────────

/// A built topology: the wire `Topology` plus the per-subtopology source-topic
/// map used to resolve task assignments to concrete topic-partitions.
///
/// NOTE: `BuiltTopology` is **not** `Clone` because the node factories hold
/// `Box<dyn Fn…>` closures that are not cloneable. The membership client wraps
/// it in an `Arc<BuiltTopology>` — use that for sharing across tasks.
pub struct BuiltTopology {
    wire: WireTopology,
    source_topics: BTreeMap<String, Vec<String>>,
    application_id: String,
    factories: HashMap<String, NodeFactory>,
    node_specs: Vec<NodeSpec>,
    /// `(changelog_override, factory)` — override is `None` for the default
    /// `<app_id>-<store_name>-changelog` derivation.
    store_factories: HashMap<String, (Option<String>, StoreFactory)>,
    /// `GlobalKTable` store factories (separate from `store_factories`): NOT built
    /// by per-task `instantiate`. The fully-replicated global-store runtime
    /// (`StreamThread`) reads these to build + restore each global store once;
    /// the [`TopologyTestDriver`] reads them via
    /// [`global_store_factories`](Self::global_store_factories)
    /// to materialize global stores directly for join tests.
    global_store_factories: HashMap<String, (Option<String>, StoreFactory)>,
    /// `global store name -> source topic` for each `GlobalKTable`. Read by the
    /// shared global manager so the consumer knows which topic feeds each store.
    // Consumed by global-store wiring via the accessor.
    #[allow(dead_code)]
    global_store_topics: HashMap<String, String>,
    /// Materialized KV stores whose record cache is eligible (default-on JVM
    /// `Materialized` caching, opt-out via `with_caching(false)`). `instantiate`
    /// wraps each in a [`NamedCache`] when `cache_max_bytes > 0`.
    caching_stores: std::collections::HashSet<String>,
    /// Store name → its connected processor names. `instantiate` maps a cached
    /// store's (single, materializing) processor name to that node's graph index
    /// to root the store's `cache_owner` entry.
    store_processors: HashMap<String, Vec<String>>,
}

impl std::fmt::Debug for BuiltTopology {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BuiltTopology")
            .field("application_id", &self.application_id)
            .field("source_topics", &self.source_topics)
            .finish_non_exhaustive()
    }
}

impl BuiltTopology {
    /// The wire `Topology` as a serde-serializable view, for golden-frame
    /// interop assertions against captured JVM fixtures.
    #[must_use]
    pub fn to_wire(&self) -> super::wire::WireTopology {
        super::wire::WireTopology::from(&self.wire)
    }

    /// The raw protocol `Topology` to send in the `StreamsGroupHeartbeat` join.
    #[must_use]
    pub(crate) fn to_wire_request(&self) -> WireTopology {
        self.wire.clone()
    }

    /// The `GlobalKTable` store factories, keyed by store name. The
    /// [`TopologyTestDriver`] materializes these directly into its per-task store
    /// registry so a stream-globaltable join can find them; the real runtime
    /// (`StreamThread`) builds them once in a shared global manager, NOT per task.
    ///
    /// [`TopologyTestDriver`]: crate::TopologyTestDriver
    pub(crate) fn global_store_factories(
        &self,
    ) -> &HashMap<String, (Option<String>, StoreFactory)> {
        &self.global_store_factories
    }

    /// The `global store name -> source topic` map for each `GlobalKTable`.
    /// The shared global-store manager reads this so the global consumer knows
    /// which topic feeds each store. Invisible in the wire output.
    pub(crate) fn global_store_topics(&self) -> HashMap<String, String> {
        self.global_store_topics.clone()
    }

    /// The external + repartition source topics a subtopology's tasks read.
    #[must_use]
    pub fn source_topics_for(&self, subtopology_id: &str) -> &[String] {
        self.source_topics
            .get(subtopology_id)
            .map_or(&[], Vec::as_slice)
    }

    /// The application id (drives internal-topic names).
    #[must_use]
    pub fn application_id(&self) -> &str {
        &self.application_id
    }

    /// Topics that are sources in this topology (for test driver / repartition
    /// loopback).
    #[must_use]
    pub fn list_source_topics(&self) -> Vec<String> {
        self.node_specs
            .iter()
            .filter(|s| s.kind == "source")
            .flat_map(|s| s.source_topics.iter().cloned())
            .collect()
    }

    /// Topics that sinks in this topology write to.
    #[must_use]
    pub fn list_sink_topics(&self) -> Vec<String> {
        self.node_specs
            .iter()
            .filter_map(|s| s.sink_topic.clone())
            .collect()
    }

    /// Instantiate a runnable [`Graph`] for this topology.
    ///
    /// Each call produces an independent graph (its own processor instances and
    /// a fresh byte-store backend opened via `backend`).
    pub(crate) async fn instantiate(
        &self,
        backend: &crate::store::backend::StoreBackend,
        app_id: &str,
        cache_max_bytes: i64,
    ) -> Result<Graph, ProcessorError> {
        // 1. Collect the processor/sink nodes in spec order and build a name→idx map.
        //    Sources are NOT in the nodes vec — they become GraphSources.
        let non_source: Vec<&NodeSpec> = self
            .node_specs
            .iter()
            .filter(|s| s.kind != "source")
            .collect();

        let name_to_idx: HashMap<&str, usize> = non_source
            .iter()
            .enumerate()
            .map(|(i, s)| (s.name.as_str(), i))
            .collect();

        let nodes: Vec<Box<dyn ErasedNode>> = non_source
            .iter()
            .map(|s| {
                let factory = self
                    .factories
                    .get(&s.name)
                    .ok_or_else(|| ProcessorError::Serde {
                        node: s.name.clone(),
                        message: "factory missing".into(),
                    })?;
                let make = factory
                    .make_node
                    .as_ref()
                    .ok_or_else(|| ProcessorError::Serde {
                        node: s.name.clone(),
                        message: "make_node missing".into(),
                    })?;
                Ok((make)())
            })
            .collect::<Result<Vec<_>, ProcessorError>>()?;

        // 2. Build children[idx] = indices of nodes that have this node as a predecessor.
        let mut children: Vec<Vec<usize>> = vec![Vec::new(); nodes.len()];
        for (child_idx, s) in non_source.iter().enumerate() {
            for parent_name in &s.predecessors {
                // parent might be a source (not in name_to_idx) or a processor/sink
                if let Some(&parent_idx) = name_to_idx.get(parent_name.as_str()) {
                    children[parent_idx].push(child_idx);
                }
            }
        }

        // 3. Build GraphSources.
        let sources: Vec<GraphSource> = self
            .node_specs
            .iter()
            .filter(|s| s.kind == "source")
            .flat_map(|src_spec| {
                // Find which processor/sink nodes list this source as a predecessor.
                let src_name = src_spec.name.as_str();
                let src_children: Vec<usize> = non_source
                    .iter()
                    .enumerate()
                    .filter(|(_, s)| s.predecessors.iter().any(|p| p == src_name))
                    .map(|(i, _)| i)
                    .collect();

                let Some(factory) = self.factories.get(src_spec.name.as_str()) else {
                    return vec![];
                };
                let Some(make_deser) = &factory.make_deser else {
                    return vec![];
                };

                src_spec
                    .source_topics
                    .iter()
                    .map(|topic| {
                        let deser = (make_deser)();
                        GraphSource {
                            topic: topic.clone(),
                            deserialize: deser,
                            children: src_children.clone(),
                        }
                    })
                    .collect()
            })
            .collect();

        // Build the per-task store registry from the typed factories.
        let mut store_registry = crate::store::registry::StoreRegistry::default();
        for (store_name, (changelog_override, factory)) in &self.store_factories {
            let changelog = changelog_override
                .clone()
                .unwrap_or_else(|| format!("{app_id}-{store_name}-changelog"));
            let bytes = backend.open(app_id, store_name).await;
            store_registry.insert(factory(store_name, changelog, bytes));
        }

        // Build-time record-cache wiring (see `wire_record_caches`): a no-op when
        // `cache_max_bytes == 0` (TopologyTestDriver) or no store is marked.
        let (cache, cache_owner) =
            self.wire_record_caches(&mut store_registry, &name_to_idx, cache_max_bytes);

        Ok(Graph {
            nodes,
            children,
            sources,
            output: Vec::new(),
            stores: store_registry,
            // Default-empty here; app wiring / TopologyTestDriver builds
            // and assigns the shared GlobalStateManager. `instantiate` produces a
            // per-task graph and never owns the fully-replicated global stores.
            globals: crate::runtime::global::GlobalStateManager::default(),
            schedules: Vec::new(),
            stream_time: i64::MIN,
            wall_clock: 0,
            cache_max_bytes,
            cache_owner,
            cache,
        })
    }

    /// Build-time record-cache wiring: for each materialized KV store marked
    /// cache-eligible (default-on `Materialized` caching), register a
    /// [`NamedCache`](crate::store::cache::named::NamedCache) in the per-task
    /// [`ThreadCache`](crate::store::cache::thread::ThreadCache), wrap the typed
    /// store's backend (via the erased
    /// [`enable_cache_erased`](crate::store::api::StateStore::enable_cache_erased)
    /// hook), and root its forwarded changes at the materializing node.
    ///
    /// Returns the populated `ThreadCache` and the `store name → owning node
    /// index` map for `Graph::flush_caches`. With `cache_max_bytes == 0`
    /// (`TopologyTestDriver`) or no marked stores, the cache stays empty and the
    /// owner map is empty, so every store stays uncached (goldens unchanged).
    fn wire_record_caches(
        &self,
        store_registry: &mut crate::store::registry::StoreRegistry,
        name_to_idx: &HashMap<&str, usize>,
        cache_max_bytes: i64,
    ) -> (
        crate::store::cache::thread::ThreadCache,
        HashMap<String, usize>,
    ) {
        let mut cache = crate::store::cache::thread::ThreadCache::new(
            usize::try_from(cache_max_bytes.max(0)).unwrap_or(0),
        );
        let mut cache_owner: HashMap<String, usize> = HashMap::new();
        if cache_max_bytes <= 0 {
            return (cache, cache_owner);
        }
        for store_name in &self.caching_stores {
            // The store's single materializing processor → its graph node index.
            let owner_idx = self
                .store_processors
                .get(store_name)
                .and_then(|procs| procs.first())
                .and_then(|proc_name| name_to_idx.get(proc_name.as_str()).copied());
            let Some(owner_idx) = owner_idx else {
                // No connected processor in this subtopology (e.g. a store owned
                // by another partition's graph) — nothing to root here.
                continue;
            };
            let nc = cache.register(store_name);
            // KV, window, and session stores are all cache-aware: each overrides
            // `enable_cache_erased` to return `true` and drives its own
            // `flush_cache_into`. A store that declines caching (returns `false`)
            // is skipped so we don't root a `cache_owner` entry the flush
            // mechanism couldn't drive.
            if store_registry.enable_cache(store_name, nc) {
                cache_owner.insert(store_name.clone(), owner_idx);
            }
        }
        (cache, cache_owner)
    }
}

// ──────────────────────────────────────────────────────────────────────────────
// Tests
// ──────────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::processor::api::{Processor, ProcessorContext};
    use crate::processor::record::Record;
    use crate::processor::serde::StringSerde;
    use assert2::check;
    use async_trait::async_trait;

    struct Upper;
    #[async_trait]
    impl Processor<String, String, String, String> for Upper {
        async fn process(
            &mut self,
            ctx: &mut ProcessorContext<'_, '_, String, String>,
            r: Record<String, String>,
        ) {
            ctx.forward(Record::new(r.key, r.value.to_uppercase(), r.timestamp));
        }
    }

    #[test]
    fn copartition_group_emitted_in_wire() {
        let mut t = Topology::new();
        let a: NodeHandle<bytes::Bytes, bytes::Bytes> = t.add_source("sa", ["left"]);
        let b: NodeHandle<bytes::Bytes, bytes::Bytes> = t.add_source("sb", ["right"]);
        t.add_sink("snk", "out", [&a, &b]); // both → one subtopology
        t.add_copartition_group(["left", "right"]);
        let wire = t.build("app").unwrap().to_wire();
        let sub = &wire.subtopologies[0];
        check!(sub.copartition_groups.len() == 1);
        check!(sub.copartition_groups[0].source_topics == vec![0i16, 1i16]); // sorted ["left","right"]
    }

    #[test]
    fn global_store_is_invisible_and_bumps_stream_index() {
        // A GlobalKTable declared FIRST takes node-group index 0 but is invisible
        // in the wire (no subtopology, no changelog). The normal stream emits as
        // "1". The global topic appears in NO source_topics. The global factory is
        // recorded in the SEPARATE map (not built by `instantiate`).
        let mut t = Topology::new();
        // Global store declared first → index 0.
        t.add_global_store::<String, String, _, _>(
            "global-store",
            "gsrc",
            "global",
            "gproc",
            Consumed::with(StringSerde, StringSerde),
        );
        // Normal stream second → index 1.
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        t.add_sink("snk", "out", [&src]);
        check!(t.has_global_store_for_test("global-store"));
        let built = t.build("app").unwrap();
        let wire = built.to_wire();
        check!(wire.subtopologies.len() == 1);
        check!(wire.subtopologies[0].subtopology_id == "1");
        check!(wire.subtopologies[0].source_topics == vec!["in".to_string()]);
        // No changelog topic anywhere.
        check!(
            wire.subtopologies
                .iter()
                .all(|s| s.state_changelog_topics.is_empty())
        );
        // The global topic is not a wire source topic.
        check!(
            !wire.subtopologies[0]
                .source_topics
                .contains(&"global".to_string())
        );
    }

    #[test]
    fn add_source_and_sink_can_use_default_serdes() {
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        t.add_sink("out", "out-topic", [&src]);
        let built = t.build("app").unwrap();
        let wire = built.to_wire();

        check!(wire.subtopologies[0].source_topics == vec!["in".to_string()]);
        check!(built.source_topics_for("0") == ["in".to_string()]);
    }

    #[test]
    fn build_single_source_sink_wire_unchanged() {
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        let up = t.add_processor("up", || Upper, [&src]);
        t.add_sink("out", "out-topic", [&up]);
        let built = t.build("app").unwrap();
        let wire = built.to_wire();
        check!(wire.epoch == 0);
        check!(wire.subtopologies[0].subtopology_id == "0");
        check!(wire.subtopologies[0].source_topics == vec!["in".to_string()]);
        check!(built.source_topics_for("0") == ["in".to_string()]);
    }

    #[test]
    fn topology_error_converts_to_stored_error() {
        // `record()` stashes registry errors as the cloneable `StoredError`;
        // exercise every arm of the conversion so it stays total (the registry
        // only emits `DuplicateNode` at runtime, but the type covers all three).
        check!(matches!(
            StoredError::from(TopologyError::Empty),
            StoredError::Empty
        ));
        check!(matches!(
            StoredError::from(TopologyError::DuplicateNode("n".into())),
            StoredError::DuplicateNode(_)
        ));
        check!(matches!(
            StoredError::from(TopologyError::UnknownPredecessor {
                node: "a".into(),
                predecessor: "b".into(),
            }),
            StoredError::UnknownPredecessor { .. }
        ));
    }

    #[test]
    fn node_handle_clone_debug_and_owned_wiring() {
        // NodeHandle is Clone + Debug, and parents may be passed by value (an
        // owned handle), not only by reference — exercise all three.
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        let src2 = src.clone();
        check!(src2.name() == "src");
        check!(format!("{src2:?}").contains("src"));
        // `[src2]` wires by value (the owned-handle `Borrow` path).
        t.add_sink("out", "out", [src2]);
        check!(t.build("app").is_ok());
    }

    #[test]
    fn handle_from_another_topology_is_rejected_at_build() {
        // A handle's node name is only registered in the topology that created
        // it; wiring it into a different topology leaves a dangling predecessor
        // that `build()` rejects. (Within one topology, forward references and
        // cycles can't even be written — you need a parent's handle first.)
        let mut a = Topology::new();
        let foreign: NodeHandle<String, String> = a.add_source("src", ["in"]);

        let mut b = Topology::new();
        b.add_sink("out", "o", [&foreign]);
        check!(b.build("app").is_err());
    }

    #[test]
    fn instantiate_runs_records() {
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        let up = t.add_processor("up", || Upper, [&src]);
        t.add_sink("out", "out-topic", [&up]);
        let built = t.build("app").unwrap();
        let mut g = pollster::block_on(built.instantiate(
            &crate::store::backend::StoreBackend::InMemory,
            "app",
            0,
        ))
        .unwrap();
        pollster::block_on(g.pipe("in", Some(b"k"), b"hi", 0)).unwrap();
        let out = g.take_output();
        check!(out.len() == 1);
        check!(out[0].value.as_ref().unwrap().as_ref() == b"HI");
    }

    #[test]
    fn empty_topology_is_rejected() {
        let topo = Topology::new();
        check!(topo.build("app").is_err());
    }

    #[test]
    fn build_with_processor_store_and_repartition() {
        let mut t = Topology::new();
        t.add_repartition_topic("rp");
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        let proc = t.add_processor("proc", || Upper, [&src]);
        t.add_state_store("store", StringSerde, StringSerde, [proc.name()]);
        t.add_sink("rsink", "rp", [&proc]);
        let rsrc: NodeHandle<String, String> = t.add_source("rsrc", ["rp"]);
        t.add_sink("out", "out-topic", [&rsrc]);
        let built = t.build("my-app").unwrap();
        check!(built.application_id() == "my-app");
        let wire = built.to_wire();
        // repartition chain produces at least 2 subtopologies
        check!(wire.subtopologies.len() >= 2);
        // subtopology containing the state store has a changelog topic named my-app-store-changelog
        let has_changelog = wire.subtopologies.iter().any(|s| {
            s.state_changelog_topics
                .iter()
                .any(|c| c.name == "my-app-store-changelog")
        });
        check!(has_changelog);
    }

    #[test]
    fn application_id_accessor() {
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        t.add_sink("snk", "out", [&src]);
        let built = t.build("my-streams-app").unwrap();
        check!(built.application_id() == "my-streams-app");
    }

    #[test]
    fn source_topics_for_unknown_id_returns_empty() {
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        t.add_sink("snk", "out", [&src]);
        let built = t.build("app").unwrap();
        check!(built.source_topics_for("99").is_empty());
    }

    #[test]
    fn connect_processor_store_adds_processor_to_store() {
        let mut t = Topology::new();
        // register a store connected to processor "p1"
        t.add_state_store::<String, String, _, _>("s", StringSerde, StringSerde, ["p1"]);
        t.connect_processor_store("p2", "s");
        // the store's connected-processor list now has both p1 and p2
        let entry = t.store_entry_for_test("s").unwrap();
        check!(entry == vec!["p1".to_string(), "p2".to_string()]);
    }

    #[test]
    fn connect_processor_store_is_idempotent() {
        let mut t = Topology::new();
        t.add_state_store::<String, String, _, _>("s", StringSerde, StringSerde, ["p1"]);
        t.connect_processor_store("p1", "s"); // p1 already connected — must not duplicate
        let entry = t.store_entry_for_test("s").unwrap();
        check!(entry == vec!["p1".to_string()]);
    }

    #[test]
    fn connect_processor_store_unknown_store_is_noop() {
        let mut t = Topology::new();
        t.add_state_store::<String, String, _, _>("s", StringSerde, StringSerde, ["p1"]);
        t.connect_processor_store("p2", "no_such_store"); // should not panic
        let entry = t.store_entry_for_test("s").unwrap();
        check!(entry == vec!["p1".to_string()]); // unchanged
    }

    #[test]
    fn duplicate_node_name_propagates_error() {
        let mut t = Topology::new();
        let _: NodeHandle<String, String> = t.add_source("src", ["in"]);
        let _: NodeHandle<String, String> = t.add_source("src", ["other"]); // duplicate
        check!(t.build("app").is_err());
    }

    #[test]
    fn instantiate_repartition_topology_lists_topics() {
        let mut t = Topology::new();
        t.add_repartition_topic("rp");
        let s1: NodeHandle<String, String> = t.add_source("s1", ["in"]);
        let p = t.add_processor("p", || Upper, [&s1]);
        t.add_sink("to_rp", "rp", [&p]);
        let s2: NodeHandle<String, String> = t.add_source("s2", ["rp"]);
        t.add_sink("out", "out", [&s2]);
        let built = t.build("app").unwrap();
        let mut srcs = built.list_source_topics();
        srcs.sort();
        check!(srcs == vec!["in".to_string(), "rp".to_string()]);
        let mut sinks = built.list_sink_topics();
        sinks.sort();
        check!(sinks == vec!["out".to_string(), "rp".to_string()]);
        // instantiate must succeed and pipe through the first subtopology
        let mut g = pollster::block_on(built.instantiate(
            &crate::store::backend::StoreBackend::InMemory,
            "app",
            0,
        ))
        .unwrap();
        pollster::block_on(g.pipe("in", None, b"hi", 0)).unwrap();
        let out1 = g.take_output();
        check!(out1.iter().any(|o| o.topic == "rp"));
    }

    #[test]
    fn instantiate_builds_stores_and_processes_statefully() {
        use crate::processor::serde::I64Serde;
        struct Counter;
        #[async_trait]
        impl Processor<String, String, String, i64> for Counter {
            async fn process(
                &mut self,
                ctx: &mut ProcessorContext<'_, '_, String, i64>,
                r: Record<String, String>,
            ) {
                let n = {
                    let s = ctx.get_state_store::<String, i64>("counts").unwrap();
                    let n = s.get(&r.value).await.unwrap_or(0) + 1;
                    s.put(r.value.clone(), n).await;
                    n
                };
                ctx.forward(Record::new(Some(r.value), n, r.timestamp));
            }
        }
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        let c = t.add_processor("c", || Counter, [&src]);
        t.add_state_store("counts", StringSerde, I64Serde, [c.name()]);
        t.add_sink("out", "out", [&c]);
        let built = t.build("app").unwrap();
        // wire topology still has the changelog topic (golden frame contract)
        check!(built.to_wire().subtopologies.iter().any(|s| {
            s.state_changelog_topics
                .iter()
                .any(|c| c.name == "app-counts-changelog")
        }));
        let mut g = pollster::block_on(built.instantiate(
            &crate::store::backend::StoreBackend::InMemory,
            "app",
            0,
        ))
        .unwrap();
        pollster::block_on(g.pipe("in", None, b"x", 0)).unwrap();
        pollster::block_on(g.pipe("in", None, b"x", 1)).unwrap();
        // After two "x" records the count should be 2 (i64 big-endian = [0,0,0,0,0,0,0,2])
        check!(
            g.take_output()
                .last()
                .unwrap()
                .value
                .as_ref()
                .unwrap()
                .as_ref()
                == [0, 0, 0, 0, 0, 0, 0, 2]
        );
    }

    // ── Build-time record-cache wiring (sub-task 3b-ii) ───────────────────────
    //
    // A materializing processor that, per record, accumulates a count in its
    // store and forwards `Change<i64>` (so the store's owning node can feed a
    // `Change`-consuming sink). Mirrors the DSL aggregate's store-write + emit.
    struct CountingMaterializer;
    #[async_trait]
    impl Processor<String, String, String, crate::dsl::processors::change::Change<i64>>
        for CountingMaterializer
    {
        async fn process(
            &mut self,
            ctx: &mut ProcessorContext<'_, '_, String, crate::dsl::processors::change::Change<i64>>,
            r: Record<String, String>,
        ) {
            use crate::dsl::processors::change::Change;
            let (old, new) = {
                let s = ctx.get_state_store::<String, i64>("counts").unwrap();
                let old = s.get(&r.value).await;
                let new = old.unwrap_or(0) + 1;
                s.put(r.value.clone(), new).await;
                (old, Some(new))
            };
            ctx.forward(Record::new(Some(r.value), Change { old, new }, r.timestamp));
        }
    }

    /// Build a `source → CountingMaterializer(materializes "counts") → sink`
    /// topology, optionally marking the "counts" store cache-eligible.
    fn counting_topology(caching: bool) -> BuiltTopology {
        use crate::dsl::processors::change::Change;
        use crate::processor::serde::I64Serde;
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        let c = t.add_processor("c", || CountingMaterializer, [&src]);
        t.add_state_store("counts", StringSerde, I64Serde, [c.name()]);
        // The materializing node's `Change<i64>` output feeds the sink (a real
        // cached flush forwards `Change<i64>` rooted at "c").
        t.add_sink_explicit::<String, Change<i64>, _, _, _, _>(
            "out",
            "out",
            [&c],
            crate::processor::serde::Produced::with(StringSerde, ChangeI64Serde),
        );
        if caching {
            t.mark_store_caching("counts", true);
        }
        t.build("app").unwrap()
    }

    // A trivial serde so the `Change<i64>` sink can be wired; it never has to
    // round-trip in these tests (we assert on the changelog, not the sink bytes).
    #[derive(Clone)]
    struct ChangeI64Serde;
    impl crate::processor::serde::Serde<crate::dsl::processors::change::Change<i64>>
        for ChangeI64Serde
    {
        fn serialize(
            &self,
            _topic: &str,
            v: &crate::dsl::processors::change::Change<i64>,
        ) -> bytes::Bytes {
            // Encode just the `new` side (8 bytes BE) so the sink has bytes to emit.
            bytes::Bytes::copy_from_slice(&v.new.unwrap_or(0).to_be_bytes())
        }
        fn deserialize(
            &self,
            _topic: &str,
            _bytes: &[u8],
        ) -> Result<crate::dsl::processors::change::Change<i64>, crate::processor::serde::SerdeError>
        {
            unreachable!("Change<i64> sink is never deserialized in these tests")
        }
    }

    #[test]
    fn instantiate_caches_materialized_store_when_budget_positive() {
        // cache_max_bytes > 0 + marked store → the store is cached and
        // cache_owner roots it at the materializing node ("c"), which is graph
        // node index 0 (sources are not in the nodes vec; "c" is first non-source).
        let built = counting_topology(true);
        let mut g = pollster::block_on(built.instantiate(
            &crate::store::backend::StoreBackend::InMemory,
            "app",
            1024,
        ))
        .unwrap();
        check!(g.cache_owner.get("counts") == Some(&0));
        check!(g.stores.kv_is_cached("counts"));

        // Pipe two records for the SAME key, then flush: a cached store dedups the
        // two staged writes into ONE changelog entry. (Without caching the store
        // logs each put immediately → two entries; see the no-cache test below.)
        pollster::block_on(g.pipe("in", None, b"x", 0)).unwrap();
        pollster::block_on(g.pipe("in", None, b"x", 1)).unwrap();
        // No changelog buffered yet — cached writes defer logging to flush.
        check!(
            g.drain_changelogs(&std::collections::HashSet::new())
                .is_empty()
        );
        pollster::block_on(g.flush_caches()).unwrap();
        let cl = g.drain_changelogs(&std::collections::HashSet::new());
        check!(cl.len() == 1); // deduped to the latest count (2)
        // tuple is (changelog_topic, key, value, ts); value is the BE i64 count.
        check!(cl[0].2.as_ref().unwrap().as_ref() == [0, 0, 0, 0, 0, 0, 0, 2]);
    }

    #[test]
    fn instantiate_does_not_cache_when_budget_zero() {
        // cache_max_bytes == 0 (TopologyTestDriver default): nothing is wrapped,
        // cache_owner is empty, and the store logs each put immediately (2 entries
        // for two same-key writes — no dedup).
        let built = counting_topology(true);
        let mut g = pollster::block_on(built.instantiate(
            &crate::store::backend::StoreBackend::InMemory,
            "app",
            0,
        ))
        .unwrap();
        check!(g.cache_owner.is_empty());
        check!(!g.stores.kv_is_cached("counts"));

        pollster::block_on(g.pipe("in", None, b"x", 0)).unwrap();
        pollster::block_on(g.pipe("in", None, b"x", 1)).unwrap();
        // Uncached: each put logs immediately → two changelog entries (no dedup).
        let cl = g.drain_changelogs(&std::collections::HashSet::new());
        check!(cl.len() == 2);
    }

    #[test]
    fn instantiate_does_not_cache_when_marking_opted_out() {
        // with_caching(false) equivalent (store NOT marked) + budget > 0: the store
        // is NOT in cache_owner and stays uncached.
        let built = counting_topology(false);
        let g = pollster::block_on(built.instantiate(
            &crate::store::backend::StoreBackend::InMemory,
            "app",
            1024,
        ))
        .unwrap();
        check!(g.cache_owner.is_empty());
        check!(!g.stores.kv_is_cached("counts"));
    }

    #[test]
    fn add_versioned_store_registers_wire_spec() {
        use crate::processor::serde::{I64Serde, StringSerde};
        let mut t = Topology::new();
        let src: NodeHandle<String, String> = t.add_source("src", ["in"]);
        let proc = t.add_processor("proc", || Upper, [&src]);
        t.add_versioned_store::<String, i64, _, _>(
            "vstore",
            StringSerde,
            I64Serde,
            600_000,
            [proc.name()],
        );
        t.add_sink("out", "out-topic", [&proc]);
        let wire = t.build("app").unwrap().to_wire();
        let blob = serde_json::to_value(&wire).unwrap().to_string();
        assert!(blob.contains("vstore"), "changelog topic name not in wire");
        assert!(
            blob.contains("min.compaction.lag.ms"),
            "min.compaction.lag.ms key not in wire"
        );
        // history_retention_ms=600_000 → min_compaction_lag_ms = 600_000 + 86_400_000 = 87_000_000
        assert!(blob.contains("87000000"), "lag value not in wire");
    }

    #[tokio::test]
    async fn add_window_store_uses_window_size_not_retention_for_cached_key_end() {
        // Regression: a CACHED sliding aggregate registers its window store with a
        // retention basis of `2 * timeDifferenceMs` but a TRUE window size of
        // `1 * timeDifferenceMs`. The flushed `Windowed` key end must be
        // `start + windowSize`, NOT `start + retentionBasis`. Before the fix the
        // factory fed `size_ms` (the retention basis) to `WindowBytesStore::new`,
        // so the end was doubled (this asserts `end == 10`, which would be `20`).
        use crate::dsl::processors::change::Change;
        use crate::dsl::windows::{Window, Windowed};
        use crate::processor::record::RecordContext;
        use crate::processor::serde::I64Serde;
        use crate::store::api::StateStore;
        use crate::store::byte::InMemoryBytes;
        use crate::store::cache::named::NamedCache;
        use crate::store::window::{WindowBytesStore, WindowStore};
        use std::sync::{Arc, Mutex};

        const D: i64 = 10;
        let mut t = Topology::new();
        // size_ms = 2*D (retention basis), window_size_ms = D (true window size).
        t.add_window_store::<String, i64, _, _>("sw", StringSerde, I64Serde, D * 2, D, 0, ["p"]);

        // Pull the registered factory and instantiate the store over a fresh backend.
        let (_changelog, factory) = t.store_factories.get("sw").expect("factory registered");
        let backend: Box<dyn crate::store::byte::ByteKeyValueStore> =
            Box::new(InMemoryBytes::default());
        let mut store: Box<dyn crate::store::api::StateStore> =
            factory("sw", "sw-changelog".to_string(), backend);

        // Enable the record cache and stage a put for a window starting at 0.
        let typed = store
            .as_any_mut()
            .downcast_mut::<WindowBytesStore<String, i64>>()
            .expect("window store downcast");
        typed.enable_cache(Arc::new(Mutex::new(NamedCache::new("sw".into()))));
        typed.set_record_context(RecordContext {
            topic: "t".into(),
            partition: 0,
            offset: 0,
            timestamp: 7,
        });
        typed.put("a".into(), 0, 1, 7).await;

        let mut buffer = std::collections::VecDeque::new();
        store.flush_cache_into(&mut buffer, &[0]).await;
        assert_eq!(buffer.len(), 1);
        let (_child, rec) = &buffer[0];
        let key = rec
            .key
            .as_ref()
            .unwrap()
            .downcast_ref::<Windowed<String>>()
            .unwrap();
        assert_eq!(key.key, "a");
        // end == start + window_size (D), NOT start + retention basis (2*D).
        assert_eq!(key.window, Window { start: 0, end: D });
        let change = rec.value.downcast_ref::<Change<i64>>().unwrap();
        assert_eq!(change.new, Some(1));
    }
}