diff --git a/core/cu29_derive/src/lib.rs b/core/cu29_derive/src/lib.rs
index cf51f193d9c..c92a8d605da 100644
--- a/core/cu29_derive/src/lib.rs
+++ b/core/cu29_derive/src/lib.rs
@@ -16,13 +16,13 @@ use crate::utils::{config_id_to_bridge_const, config_id_to_enum, config_id_to_st
 use cu29_build::COPPER_CFG_FEATURES_ENV;
 use cu29_runtime::config::CuConfig;
 use cu29_runtime::config::{
-    BridgeChannelConfigRepresentation, ConfigGraphs, CuGraph, Flavor, HandleContent, Node, NodeId,
-    RT_POOL, ResourceBundleConfig, read_configuration_with_features,
-    read_configuration_with_resolved_ron_and_features,
+    BridgeChannelConfigRepresentation, ConfigGraphs, CorePlacement, CuGraph, Flavor, HandleContent,
+    Node, NodeId, PlanPolicy, PlanProfile, RT_POOL, ResourceBundleConfig,
+    read_configuration_with_features, read_configuration_with_resolved_ron_and_features,
 };
 use cu29_runtime::curuntime::{
     CuExecutionLoop, CuExecutionStep, CuExecutionUnit, CuTaskType, compute_runtime_plan,
-    find_task_type_for_id,
+    find_task_type_for_id, place_steps_on_cores,
 };
 use cu29_traits::{CuError, CuResult};
 use proc_macro2::{Ident, Span};
@@ -678,16 +678,22 @@ fn build_gen_cumsgs_support(
     let task_specs = CuTaskSpecSet::from_graph(graph)?;
     let channel_usage = collect_bridge_channel_usage(graph);
     let mut bridge_specs = build_bridge_specs(cuconfig, graph, &channel_usage);
-    let (culist_plan, exec_entities, plan_to_original) =
-        build_execution_plan(graph, &task_specs, &mut bridge_specs).map_err(|e| {
-            if let Some(mission) = mission_label {
-                CuError::from(format!(
-                    "Could not compute copperlist plan for mission '{mission}': {e}"
-                ))
-            } else {
-                CuError::from(format!("Could not compute copperlist plan: {e}"))
-            }
-        })?;
+    let (culist_plan, exec_entities, plan_to_original) = build_execution_plan(
+        graph,
+        &task_specs,
+        &mut bridge_specs,
+        cuconfig.plan_policy(),
+        &cuconfig.plan_profile(),
+    )
+    .map_err(|e| {
+        if let Some(mission) = mission_label {
+            CuError::from(format!(
+                "Could not compute copperlist plan for mission '{mission}': {e}"
+            ))
+        } else {
+            CuError::from(format!("Could not compute copperlist plan: {e}"))
+        }
+    })?;
     let task_names = collect_task_names(graph);
     let (culist_order, node_output_positions) = collect_culist_metadata(
         &culist_plan,
@@ -1714,7 +1720,13 @@ pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream {
         let mut culist_bridge_specs =
             build_bridge_specs(&copper_config, graph, &culist_channel_usage);
         let (culist_plan, culist_exec_entities, culist_plan_to_original) =
-            match build_execution_plan(graph, &task_specs, &mut culist_bridge_specs) {
+            match build_execution_plan(
+                graph,
+                &task_specs,
+                &mut culist_bridge_specs,
+                copper_config.plan_policy(),
+                &copper_config.plan_profile(),
+            ) {
                 Ok(plan) => plan,
                 Err(e) => return return_error(format!("Could not compute copperlist plan: {e}")),
             };
@@ -3478,12 +3490,25 @@ pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream {
                         }
                     })
                     .collect();
+            // Which affinity slot each stage worker pins to. Computed here, at
+            // compile time, because both inputs (the profile and the "rt" pool's
+            // affinity list) are in the config.
+            let stage_affinity_slots =
+                match build_stage_affinity_slots(&copper_config, &culist_plan) {
+                    Ok(slots) => slots,
+                    Err(e) => return return_error(format!("Could not place parallel stages: {e}")),
+                };
             let parallel_stage_worker_spawns: Vec<proc_macro2::TokenStream> =
                 parallel_process_step_idents
                     .iter()
                     .enumerate()
                     .map(|(stage_index, step_ident)| {
-                        let stage_index_lit = syn::Index::from(stage_index);
+                        let stage_index_lit = syn::Index::from(
+                            stage_affinity_slots
+                                .as_ref()
+                                .and_then(|slots| slots.get(stage_index).copied())
+                                .unwrap_or(stage_index),
+                        );
                         let receiver_ident =
                             format_ident!("__cu_parallel_stage_rx_{stage_index}");
                         quote! {
@@ -3508,8 +3533,9 @@ pub fn copper_runtime(args: TokenStream, input: TokenStream) -> TokenStream {
                                 let rt_pool = std::sync::Arc::clone(&rt_pool);
                                 scope.spawn(move || {
                                     // Apply the "rt" pool's CPU affinity / scheduling policy to
-                                    // this stage worker (Spread by stage index). On a Strict pool
-                                    // this fails the worker, which aborts the pipeline.
+                                    // this stage worker, on the affinity slot the configured
+                                    // core placement picked. On a Strict pool this fails the
+                                    // worker, which aborts the pipeline.
                                     if let Some(rt_pool) = rt_pool.as_ref()
                                         && cu29::thread_pool::apply_current_thread_scheduling(
                                             rt_pool,
@@ -7726,6 +7752,8 @@ fn build_execution_plan(
     graph: &CuGraph,
     task_specs: &CuTaskSpecSet,
     bridge_specs: &mut [BridgeSpec],
+    plan_policy: PlanPolicy,
+    plan_profile: &PlanProfile,
 ) -> CuResult<(
     CuExecutionLoop,
     Vec<ExecutionEntity>,
@@ -7897,7 +7925,7 @@ fn build_execution_plan(
             .map_err(|e| CuError::from(e.to_string()))?;
     }
 
-    let runtime_plan = compute_runtime_plan(&plan_graph)?;
+    let runtime_plan = compute_runtime_plan(&plan_graph, plan_policy, plan_profile)?;
     Ok((runtime_plan, exec_entities, plan_to_original))
 }
 
@@ -7994,6 +8022,61 @@ fn build_monitor_culist_component_mapping(
     Ok(mapping)
 }
 
+/// Resolves the configured [`CorePlacement`] into one affinity slot per plan
+/// step, or `None` when there is nothing to place: no `rt` pool, or an `rt`
+/// pool that declares no CPU affinity. `None` leaves the historical
+/// spread-by-stage-index behavior untouched.
+///
+/// A non-default placement with nothing to place is a config mistake, not a
+/// silent fallback: it errors out the same way an empty profile does.
+fn build_stage_affinity_slots(
+    config: &CuConfig,
+    plan: &CuExecutionLoop,
+) -> CuResult<Option<Vec<usize>>> {
+    let placement = config.core_placement();
+    let no_slots = |reason: String| -> CuResult<Option<Vec<usize>>> {
+        if placement == CorePlacement::default() {
+            return Ok(None);
+        }
+        Err(CuError::from(format!(
+            "The core placement {placement:?} has no CPU affinity slots to place onto: {reason}. \
+             Declare the cores in the '{RT_POOL}' thread pool, for example \
+             `runtime: (thread_pools: [(id: \"{RT_POOL}\", threads: 4, affinity: [0, 1, 2, 3])])`, \
+             or drop `core_placement` to keep the default {:?}.",
+            CorePlacement::default()
+        )))
+    };
+
+    let Some(runtime) = config.runtime.as_ref() else {
+        return no_slots("the config has no `runtime` section".to_string());
+    };
+    let Some(rt_pool) = runtime.thread_pools.iter().find(|pool| pool.id == RT_POOL) else {
+        return no_slots(format!("the config declares no '{RT_POOL}' thread pool"));
+    };
+    let Some(cores) = rt_pool.affinity.as_ref().filter(|cores| !cores.is_empty()) else {
+        return no_slots(format!(
+            "the '{RT_POOL}' thread pool declares an empty CPU affinity list"
+        ));
+    };
+
+    let slots = place_steps_on_cores(plan, placement, &config.plan_profile(), cores.len())?;
+    // The generated pipeline spawns exactly one stage worker per plan step, so a
+    // shorter slot list would silently leave the tail on its stage index.
+    if slots.len() != plan.steps.len() {
+        return Err(CuError::from(format!(
+            "Core placement produced {} slots for {} plan steps",
+            slots.len(),
+            plan.steps.len()
+        )));
+    }
+    #[cfg(feature = "macro_debug")]
+    eprintln!(
+        "[core placement: {placement:?} over {} slots -> {slots:?}]",
+        cores.len()
+    );
+    Ok(Some(slots))
+}
+
 fn build_parallel_rt_stage_entries(
     runtime_plan: &CuExecutionLoop,
     exec_entities: &[ExecutionEntity],
@@ -9642,7 +9725,8 @@ mod tests {
         let graph = config.get_graph(None).expect("missing graph");
         let src_id = graph.get_node_id_by_name("src").expect("missing src node");
 
-        let runtime = compute_runtime_plan(graph).expect("runtime plan failed");
+        let runtime = compute_runtime_plan(graph, config.plan_policy(), &config.plan_profile())
+            .expect("runtime plan failed");
         let src_step = runtime
             .steps
             .iter()
@@ -9658,6 +9742,105 @@ mod tests {
         );
     }
 
+    #[test]
+    fn core_placement_packs_stages_onto_the_rt_pool_affinity() {
+        use super::*;
+        use cu29::config::CuConfig;
+
+        let mut config: CuConfig =
+            read_config("tests/config/core_placement_valid.ron").expect("failed to read config");
+        let graph = config.get_graph(None).expect("missing graph");
+        let plan = compute_runtime_plan(graph, config.plan_policy(), &config.plan_profile())
+            .expect("runtime plan failed");
+
+        // Plan order is src(1000) heavy(9000) light(500) sink(4000) over two
+        // cores: LPT packs heavy alone and the other three together.
+        let slots = build_stage_affinity_slots(&config, &plan)
+            .expect("placement failed")
+            .expect("an rt pool with affinity must produce a placement");
+        assert_eq!(slots, vec![1, 0, 1, 1]);
+
+        // Without a placement request the historical spread is kept implicit:
+        // every stage keeps its own index.
+        config.runtime.as_mut().unwrap().core_placement = cu29::config::CorePlacement::Spread;
+        let spread = build_stage_affinity_slots(&config, &plan)
+            .expect("placement failed")
+            .expect("an rt pool with affinity must produce a placement");
+        assert_eq!(spread, vec![0, 1, 0, 1]);
+    }
+
+    #[test]
+    fn default_core_placement_is_skipped_without_an_affinity_list() {
+        use super::*;
+        use cu29::config::{CorePlacement, CuConfig};
+
+        let mut config: CuConfig =
+            read_config("tests/config/core_placement_valid.ron").expect("failed to read config");
+        let graph = config.get_graph(None).expect("missing graph");
+        let plan = compute_runtime_plan(graph, config.plan_policy(), &config.plan_profile())
+            .expect("runtime plan failed");
+        config.runtime.as_mut().unwrap().core_placement = CorePlacement::Spread;
+
+        // An rt pool that pins nothing has no slots to balance across.
+        config.runtime.as_mut().unwrap().thread_pools[0].affinity = None;
+        assert!(
+            build_stage_affinity_slots(&config, &plan)
+                .expect("placement failed")
+                .is_none()
+        );
+
+        // Neither has a config without an rt pool at all.
+        config.runtime.as_mut().unwrap().thread_pools.clear();
+        assert!(
+            build_stage_affinity_slots(&config, &plan)
+                .expect("placement failed")
+                .is_none()
+        );
+
+        // ... nor one without a runtime section.
+        config.runtime = None;
+        assert!(
+            build_stage_affinity_slots(&config, &plan)
+                .expect("placement failed")
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn requested_core_placement_without_slots_fails_the_build() {
+        use super::*;
+        use cu29::config::CuConfig;
+
+        let mut config: CuConfig =
+            read_config("tests/config/core_placement_valid.ron").expect("failed to read config");
+        let graph = config.get_graph(None).expect("missing graph");
+        let plan = compute_runtime_plan(graph, config.plan_policy(), &config.plan_profile())
+            .expect("runtime plan failed");
+
+        // `core_placement: LongestFirst` with nothing to place onto is a config
+        // mistake, not a silent fallback to the historical spread. Dropping the
+        // whole `runtime` section drops the request with it, so it is not in
+        // this list.
+        for strip in [
+            (|config: &mut CuConfig| {
+                config.runtime.as_mut().unwrap().thread_pools[0].affinity = None
+            }) as fn(&mut CuConfig),
+            |config: &mut CuConfig| {
+                config.runtime.as_mut().unwrap().thread_pools[0].affinity = Some(Vec::new())
+            },
+            |config: &mut CuConfig| config.runtime.as_mut().unwrap().thread_pools.clear(),
+        ] {
+            let mut config = config.clone();
+            strip(&mut config);
+            let err = build_stage_affinity_slots(&config, &plan)
+                .expect_err("a requested placement with no slots must fail");
+            assert!(
+                err.to_string().contains("LongestFirst"),
+                "unexpected error: {err}"
+            );
+        }
+    }
+
     #[test]
     fn matching_task_ids_are_flattened_per_output_message() {
         use super::*;
@@ -9670,9 +9853,14 @@ mod tests {
         let task_specs = CuTaskSpecSet::from_graph(graph).expect("task specs");
         let channel_usage = collect_bridge_channel_usage(graph);
         let mut bridge_specs = build_bridge_specs(&config, graph, &channel_usage);
-        let (runtime_plan, exec_entities, plan_to_original) =
-            build_execution_plan(graph, &task_specs, &mut bridge_specs)
-                .expect("runtime plan failed");
+        let (runtime_plan, exec_entities, plan_to_original) = build_execution_plan(
+            graph,
+            &task_specs,
+            &mut bridge_specs,
+            config.plan_policy(),
+            &config.plan_profile(),
+        )
+        .expect("runtime plan failed");
         let output_packs = extract_output_packs(&runtime_plan);
         let task_names = collect_task_names(graph);
         let (_, node_output_positions) = collect_culist_metadata(
diff --git a/core/cu29_derive/tests/config/core_placement_valid.ron b/core/cu29_derive/tests/config/core_placement_valid.ron
new file mode 100644
index 00000000000..b87f54c367e
--- /dev/null
+++ b/core/cu29_derive/tests/config/core_placement_valid.ron
@@ -0,0 +1,25 @@
+(
+    tasks: [
+        (id: "src", type: "tasks::Source"),
+        (id: "heavy", type: "tasks::Heavy"),
+        (id: "light", type: "tasks::Light"),
+        (id: "sink", type: "tasks::Sink"),
+    ],
+    cnx: [
+        (src: "src", dst: "heavy", msg: "i32"),
+        (src: "heavy", dst: "light", msg: "i32"),
+        (src: "light", dst: "sink", msg: "i32"),
+    ],
+    runtime: (
+        thread_pools: [
+            (id: "rt", threads: 2, affinity: [2, 3]),
+        ],
+        core_placement: LongestFirst,
+        plan_profile: (task_duration_ns: {
+            "src": 1000,
+            "heavy": 9000,
+            "light": 500,
+            "sink": 4000,
+        }),
+    ),
+)
diff --git a/core/cu29_export/Cargo.toml b/core/cu29_export/Cargo.toml
index f09819e8857..66da92e9162 100644
--- a/core/cu29_export/Cargo.toml
+++ b/core/cu29_export/Cargo.toml
@@ -33,6 +33,7 @@ pyo3 = { version = "0.29", optional = true, default-features = false, features =
 ] }
 
 serde_json = { version = "1.0", default-features = false }
+ron = { version = "0.12", default-features = false }
 
 mcap = { version = "0.25", optional = true, default-features = false }
 erased-serde = { version = "0.4", optional = true, default-features = false }
diff --git a/core/cu29_export/src/lib.rs b/core/cu29_export/src/lib.rs
index 02cc02d2917..2fc67aea788 100644
--- a/core/cu29_export/src/lib.rs
+++ b/core/cu29_export/src/lib.rs
@@ -15,6 +15,7 @@
 
 mod fsck;
 pub mod logstats;
+pub mod schedule_profile;
 
 #[cfg(feature = "mcap")]
 pub mod mcap_export;
@@ -33,7 +34,8 @@ use cu29_intern_strs::read_interned_strings;
 use fsck::check;
 #[cfg(feature = "mcap")]
 use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle};
-use logstats::{compute_logstats, write_logstats};
+use logstats::{compute_logstats, format_bottleneck, write_logstats};
+use schedule_profile::{ProfileStat, compute_schedule_profile, write_schedule_profile};
 use serde::Serialize;
 use std::fmt::{Display, Formatter};
 #[cfg(feature = "mcap")]
@@ -146,6 +148,21 @@ pub enum Command {
         #[arg(long)]
         mission: Option<String>,
     },
+    /// Export a measured `plan_profile: (...)` RON snippet (see doc/sched-v0.md)
+    ScheduleProfile {
+        /// Output RON file path
+        #[arg(short, long, default_value = "schedule_profile.ron")]
+        output: PathBuf,
+        /// Config file used to map outputs to tasks
+        #[arg(long, default_value = "copperconfig.ron")]
+        config: PathBuf,
+        /// Mission id to use when reading the config
+        #[arg(long)]
+        mission: Option<String>,
+        /// Which statistic of the sampled durations fills the profile
+        #[arg(long, value_enum, default_value_t = ProfileStat::Mean)]
+        stat: ProfileStat,
+    },
     /// Export copperlists to MCAP format (requires 'mcap' feature)
     #[cfg(feature = "mcap")]
     ExportMcap {
@@ -231,7 +248,8 @@ where
 {
     let args = LogReaderCli::parse();
     let unifiedlog_base = args.unifiedlog_base;
-    let _ = cu29::logcodec::seed_effective_config_from_log::<P>(&unifiedlog_base)?;
+    let embedded_config_ron =
+        cu29::logcodec::seed_effective_config_from_log::<P>(&unifiedlog_base)?;
 
     let mut dl = build_read_logger(&unifiedlog_base)?;
 
@@ -296,7 +314,22 @@ where
             config,
             mission,
         } => {
-            run_logstats::<P>(dl, output, config, mission)?;
+            run_logstats::<P>(dl, output, config, mission, embedded_config_ron.as_deref())?;
+        }
+        Command::ScheduleProfile {
+            output,
+            config,
+            mission,
+            stat,
+        } => {
+            run_schedule_profile::<P>(
+                dl,
+                output,
+                config,
+                mission,
+                stat,
+                embedded_config_ron.as_deref(),
+            )?;
         }
         #[cfg(feature = "mcap")]
         Command::ExportMcap {
@@ -348,7 +381,8 @@ where
 {
     let args = LogReaderCli::parse();
     let unifiedlog_base = args.unifiedlog_base;
-    let _ = cu29::logcodec::seed_effective_config_from_log::<P>(&unifiedlog_base)?;
+    let embedded_config_ron =
+        cu29::logcodec::seed_effective_config_from_log::<P>(&unifiedlog_base)?;
 
     let mut dl = build_read_logger(&unifiedlog_base)?;
 
@@ -413,7 +447,22 @@ where
             config,
             mission,
         } => {
-            run_logstats::<P>(dl, output, config, mission)?;
+            run_logstats::<P>(dl, output, config, mission, embedded_config_ron.as_deref())?;
+        }
+        Command::ScheduleProfile {
+            output,
+            config,
+            mission,
+            stat,
+        } => {
+            run_schedule_profile::<P>(
+                dl,
+                output,
+                config,
+                mission,
+                stat,
+                embedded_config_ron.as_deref(),
+            )?;
         }
     }
 
@@ -425,6 +474,7 @@ fn run_logstats<P>(
     output: PathBuf,
     config: PathBuf,
     mission: Option<String>,
+    embedded_config_ron: Option<&str>,
 ) -> CuResult<()>
 where
     P: CopperListTuple + CuPayloadRawBytes,
@@ -434,9 +484,93 @@ where
         .ok_or_else(|| CuError::from("Config path is not valid UTF-8"))?;
     let cfg = read_configuration(config_path)
         .map_err(|e| CuError::new_with_cause("Failed to read configuration", e))?;
+    warn_on_plan_drift(&cfg, embedded_config_ron, config_path);
     let reader = UnifiedLoggerIOReader::new(dl, UnifiedLogType::CopperList);
     let stats = compute_logstats::<P>(reader, &cfg, mission.as_deref())?;
-    write_logstats(&stats, &output)
+    write_logstats(&stats, &output)?;
+    println!("{}", format_bottleneck(&stats.pipeline));
+    Ok(())
+}
+
+fn run_schedule_profile<P>(
+    dl: UnifiedLoggerRead,
+    output: PathBuf,
+    config: PathBuf,
+    mission: Option<String>,
+    stat: ProfileStat,
+    embedded_config_ron: Option<&str>,
+) -> CuResult<()>
+where
+    P: CopperListTuple + CuPayloadRawBytes,
+{
+    let config_path = config
+        .to_str()
+        .ok_or_else(|| CuError::from("Config path is not valid UTF-8"))?;
+    let cfg = read_configuration(config_path)
+        .map_err(|e| CuError::new_with_cause("Failed to read configuration", e))?;
+    warn_on_plan_drift(&cfg, embedded_config_ron, config_path);
+    let reader = UnifiedLoggerIOReader::new(dl, UnifiedLogType::CopperList);
+    let profile = compute_schedule_profile::<P>(reader, &cfg, mission.as_deref(), stat)?;
+    if profile.is_empty() {
+        eprintln!(
+            "Warning: no task in this recording had a process_time window, so the profile is \
+             empty. A profile-guided policy rejects an empty `plan_profile`; record a log with \
+             message logging enabled before exporting."
+        );
+    }
+    write_schedule_profile(&profile, &output)?;
+    println!(
+        "Wrote {}. Paste its content as the config's `runtime.plan_profile` value, set \
+         `runtime.plan_policy` to a profile-guided policy, and rebuild.",
+        output.display()
+    );
+    Ok(())
+}
+
+/// Warns when the config on disk plans differently from the one the log was
+/// recorded with.
+///
+/// Both tools map copperlist slots back to tasks by recomputing the plan from
+/// `--config`. The unified log embeds the config the binary actually ran, so a
+/// profile pasted (or a policy switched) without a rebuild is detectable here —
+/// and it silently misattributes every slot if it goes unnoticed.
+fn warn_on_plan_drift(cfg: &CuConfig, embedded_config_ron: Option<&str>, config_path: &str) {
+    if let Some(warning) = plan_drift_warning(cfg, embedded_config_ron, config_path) {
+        eprintln!("{warning}");
+    }
+}
+
+/// The warning text of [`warn_on_plan_drift`], or `None` when the on-disk config
+/// still plans the way the recording did.
+fn plan_drift_warning(
+    cfg: &CuConfig,
+    embedded_config_ron: Option<&str>,
+    config_path: &str,
+) -> Option<String> {
+    let recorded = read_configuration_str(embedded_config_ron?.to_string(), None).ok()?;
+    let mut drift = Vec::new();
+    if recorded.plan_policy() != cfg.plan_policy() {
+        drift.push(format!(
+            "plan_policy: log has {:?}, '{config_path}' has {:?}",
+            recorded.plan_policy(),
+            cfg.plan_policy()
+        ));
+    }
+    if recorded.plan_profile() != cfg.plan_profile() {
+        drift.push(format!(
+            "plan_profile: '{config_path}' was edited since the log"
+        ));
+    }
+    if drift.is_empty() {
+        return None;
+    }
+    Some(format!(
+        "Warning: '{config_path}' no longer plans the way the recorded binary did ({}). \
+         Steps are mapped to tasks with the on-disk plan, so the results below are \
+         misattributed. Rebuild and re-record, or export against the config the log was \
+         made with.",
+        drift.join("; ")
+    ))
 }
 
 /// Helper function for MCAP export.
@@ -1374,6 +1508,49 @@ mod tests {
     use std::sync::{Arc, Mutex};
     use tempfile::{TempDir, tempdir};
 
+    /// A src -> sink config, optionally carrying a `runtime:` section.
+    fn plan_config(runtime: &str) -> CuConfig {
+        let txt = format!(
+            r#"(
+                tasks: [(id: "src", type: "a"), (id: "sink", type: "b")],
+                cnx: [(src: "src", dst: "sink", msg: "msg::A")],
+                {runtime}
+            )"#
+        );
+        read_configuration_str(txt, None).expect("config should parse")
+    }
+
+    #[test]
+    fn plan_drift_is_reported_against_the_config_embedded_in_the_log() {
+        let recorded = r#"(
+            tasks: [(id: "src", type: "a"), (id: "sink", type: "b")],
+            cnx: [(src: "src", dst: "sink", msg: "msg::A")],
+        )"#;
+
+        // Same plan on both sides: nothing to say.
+        let same = plan_config("");
+        assert!(plan_drift_warning(&same, Some(recorded), "copperconfig.ron").is_none());
+        // No embedded config (an older log) cannot be compared.
+        assert!(plan_drift_warning(&same, None, "copperconfig.ron").is_none());
+
+        // A policy switched without re-recording.
+        let switched = plan_config(
+            "runtime: (plan_policy: CriticalPathFirst, plan_profile: (task_duration_ns: {\"src\": 5})),",
+        );
+        let warning = plan_drift_warning(&switched, Some(recorded), "copperconfig.ron")
+            .expect("a switched policy must be reported");
+        assert!(warning.contains("plan_policy"), "{warning}");
+        assert!(warning.contains("plan_profile"), "{warning}");
+
+        // A profile pasted but the policy left alone still changes nothing about
+        // the plan, yet it means the file no longer matches the recording.
+        let pasted = plan_config("runtime: (plan_profile: (task_duration_ns: {\"src\": 5})),");
+        let warning = plan_drift_warning(&pasted, Some(recorded), "copperconfig.ron")
+            .expect("an edited profile must be reported");
+        assert!(warning.contains("plan_profile"), "{warning}");
+        assert!(!warning.contains("plan_policy"), "{warning}");
+    }
+
     fn copy_stringindex_to_temp(tmpdir: &TempDir) -> PathBuf {
         // Build a minimal index on the fly so tests don't depend on build-time artifacts.
         let fake_out_dir = tmpdir.path().join("build").join("out").join("dir");
diff --git a/core/cu29_export/src/logstats.rs b/core/cu29_export/src/logstats.rs
index 3b4cbeea9ab..e97ffe5426d 100644
--- a/core/cu29_export/src/logstats.rs
+++ b/core/cu29_export/src/logstats.rs
@@ -1,6 +1,6 @@
 use crate::copperlists_reader;
 use cu29::clock::{CuDuration, OptionCuTime};
-use cu29::config::{CuConfig, CuGraph, Flavor};
+use cu29::config::{CuConfig, CuGraph, Flavor, LOGSTATS_SCHEMA_VERSION, PlanPolicy, PlanProfile};
 use cu29::curuntime::{CuExecutionLoop, CuExecutionUnit, compute_runtime_plan};
 use cu29::monitoring::CuDurationStatistics;
 use cu29::prelude::{CopperListTuple, CuMsgMetadataTrait, CuPayloadRawBytes};
@@ -11,7 +11,6 @@ use std::fs::File;
 use std::io::Read;
 use std::path::Path;
 
-const LOGSTATS_SCHEMA_VERSION: u32 = 1;
 const MAX_LATENCY_NS: u64 = 10_000_000_000;
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -21,6 +20,60 @@ pub struct LogStats {
     pub mission: Option<String>,
     pub edges: Vec<EdgeLogStats>,
     pub perf: PerfStats,
+    /// Added in schema version 2, so a version 1 document still reads back.
+    #[serde(default)]
+    pub pipeline: PipelineStats,
+}
+
+/// Per-plan-step cost, and the throughput ceiling each execution engine can
+/// reach with it.
+///
+/// The serial engine runs every step of a CopperList back to back, so its
+/// cycle is `serial_cycle_ns`. `parallel-rt` runs one worker per step and
+/// pipelines CopperLists through them, so its cycle is the slowest single
+/// step: `bottleneck`. `max_pipeline_speedup` is the ratio, i.e. the most
+/// `parallel-rt` can buy on this recording before the bottleneck step has to
+/// be split or made faster.
+#[derive(Debug, Default, Serialize, Deserialize)]
+pub struct PipelineStats {
+    /// One entry per plan step, in plan order.
+    pub stages: Vec<StageStats>,
+    /// Sum of the mean step durations: the serial engine's cycle.
+    pub serial_cycle_ns: Option<f64>,
+    /// The slowest step: the `parallel-rt` pipeline's cycle.
+    pub bottleneck: Option<Bottleneck>,
+    /// `serial_cycle_ns / bottleneck.mean_ns`, capped by the step count.
+    pub max_pipeline_speedup: Option<f64>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct StageStats {
+    /// Task or bridge id owning this plan step.
+    pub task: String,
+    /// Index of this step in the plan computed from the config graph, which is
+    /// also its copperlist output slot.
+    ///
+    /// This matches the `parallel-rt` worker index only for a graph without
+    /// bridges: the runtime plans over a graph the derive first expands with one
+    /// node per bridge channel, and those extra steps shift every later index.
+    pub index: usize,
+    pub samples: u64,
+    /// Measured `process()` window of this step: min/max/mean/stddev, all
+    /// exact. For an exact percentile instead, use
+    /// `cu29_export <log> schedule-profile --stat p99`, which keeps the raw
+    /// samples; the live histogram behind these stats is too coarse for one.
+    pub duration: DurationStats,
+    /// Share of `serial_cycle_ns` spent in this step, in `0.0..=1.0`.
+    pub share: Option<f64>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Bottleneck {
+    pub task: String,
+    pub index: usize,
+    pub mean_ns: f64,
+    /// `1e9 / mean_ns`: the CopperList rate this step alone allows.
+    pub max_rate_hz: f64,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -49,7 +102,7 @@ pub struct PerfStats {
     pub jitter: DurationStats,
 }
 
-#[derive(Debug, Default, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct DurationStats {
     pub min_ns: Option<u64>,
     pub max_ns: Option<u64>,
@@ -145,6 +198,128 @@ impl EdgeAccumulator {
     }
 }
 
+/// One plan step's flattened slot range in the copperlist message vector.
+pub(crate) struct PackRange {
+    pub(crate) start: usize,
+    pub(crate) len: usize,
+    pub(crate) task: String,
+}
+
+/// The copperlist message vector flattens the output packs in slot order, so a
+/// step owns one contiguous range of it.
+pub(crate) fn build_pack_ranges(packs: &[OutputPackInfo]) -> Vec<PackRange> {
+    let mut ranges = Vec::with_capacity(packs.len());
+    let mut base = 0usize;
+    for pack in packs {
+        ranges.push(PackRange {
+            start: base,
+            len: pack.msg_types.len(),
+            task: pack.src.clone(),
+        });
+        base += pack.msg_types.len();
+    }
+    ranges
+}
+
+/// One step's `process()` window in one copperlist: the widest start/end pair
+/// over the slots it owns. `None` when the recording has neither bound.
+pub(crate) fn sample_step_duration_ns(
+    cumsgs: &[&dyn cu29::prelude::ErasedCuStampedData],
+    range: &PackRange,
+) -> Option<u64> {
+    let end_slot = (range.start + range.len).min(cumsgs.len());
+    let mut start_ns: Option<u64> = None;
+    let mut end_ns: Option<u64> = None;
+    for msg in &cumsgs[range.start.min(end_slot)..end_slot] {
+        let meta = msg.metadata();
+        if let Some(start) = extract_start_time_ns(meta) {
+            start_ns = Some(start_ns.map_or(start, |current| current.min(start)));
+        }
+        if let Some(end) = extract_end_time_ns(meta) {
+            end_ns = Some(end_ns.map_or(end, |current| current.max(end)));
+        }
+    }
+    end_ns?.checked_sub(start_ns?)
+}
+
+/// Accumulates one plan step's `process()` durations across the recording.
+#[derive(Debug)]
+struct StageAccumulator {
+    task: String,
+    stats: CuDurationStatistics,
+}
+
+impl StageAccumulator {
+    fn new(task: String) -> Self {
+        Self {
+            task,
+            stats: CuDurationStatistics::new(CuDuration(MAX_LATENCY_NS)),
+        }
+    }
+
+    fn record_sample(&mut self, duration_ns: u64) {
+        self.stats.record(CuDuration(duration_ns));
+    }
+
+    fn finalize(&self, index: usize, serial_cycle_ns: Option<f64>) -> StageStats {
+        let duration = duration_stats_from(&self.stats);
+        let share = match (duration.mean_ns, serial_cycle_ns) {
+            (Some(mean), Some(cycle)) if cycle > 0.0 => Some(mean / cycle),
+            _ => None,
+        };
+        StageStats {
+            task: self.task.clone(),
+            index,
+            samples: self.stats.len(),
+            duration,
+            share,
+        }
+    }
+}
+
+fn finalize_pipeline(accumulators: &[StageAccumulator]) -> PipelineStats {
+    let means: Vec<Option<f64>> = accumulators
+        .iter()
+        .map(|acc| duration_stats_from(&acc.stats).mean_ns)
+        .collect();
+
+    // Steps the recording never sampled leave the cycle unknown rather than
+    // making it look cheaper than it is.
+    let serial_cycle_ns = means
+        .iter()
+        .copied()
+        .try_fold(0.0, |total, mean| mean.map(|mean| total + mean));
+
+    let bottleneck = means
+        .iter()
+        .enumerate()
+        .filter_map(|(index, mean)| mean.map(|mean| (index, mean)))
+        .max_by(|(_, a), (_, b)| a.total_cmp(b))
+        .filter(|(_, mean)| *mean > 0.0)
+        .map(|(index, mean)| Bottleneck {
+            task: accumulators[index].task.clone(),
+            index,
+            mean_ns: mean,
+            max_rate_hz: 1e9 / mean,
+        });
+
+    let max_pipeline_speedup = match (serial_cycle_ns, &bottleneck) {
+        (Some(cycle), Some(slowest)) if slowest.mean_ns > 0.0 => Some(cycle / slowest.mean_ns),
+        _ => None,
+    };
+
+    PipelineStats {
+        stages: accumulators
+            .iter()
+            .enumerate()
+            .map(|(index, acc)| acc.finalize(index, serial_cycle_ns))
+            .collect(),
+        serial_cycle_ns,
+        bottleneck,
+        max_pipeline_speedup,
+    }
+}
+
 #[derive(Debug)]
 struct PerfAccumulator {
     stats: CuDurationStatistics,
@@ -192,7 +367,13 @@ where
 {
     let graph = config.get_graph(mission)?;
     let signature = build_graph_signature(graph, mission);
-    let output_slots = build_output_slots(graph)?;
+    let packs = collect_output_packs(graph, config.plan_policy(), &config.plan_profile())?;
+    let stage_ranges = build_pack_ranges(&packs);
+    let output_slots = build_output_slots(&packs, graph);
+    let mut stage_accumulators: Vec<StageAccumulator> = stage_ranges
+        .iter()
+        .map(|range| StageAccumulator::new(range.task.clone()))
+        .collect();
     let mut edge_accumulators = build_edge_accumulators(graph);
     let mut perf = PerfAccumulator::new();
     let mut warned_lengths = false;
@@ -228,6 +409,12 @@ where
             }
         }
 
+        for (range, acc) in stage_ranges.iter().zip(stage_accumulators.iter_mut()) {
+            if let Some(duration_ns) = sample_step_duration_ns(&cumsgs, range) {
+                acc.record_sample(duration_ns);
+            }
+        }
+
         perf.record_sample(compute_end_to_end_latency(&cumsgs));
     }
 
@@ -242,9 +429,36 @@ where
         mission: mission.map(|value| value.to_string()),
         edges,
         perf: perf.finalize(),
+        pipeline: finalize_pipeline(&stage_accumulators),
     })
 }
 
+/// One line naming the step that caps the CopperList rate, and what a wider
+/// engine could still buy. Printed next to the JSON so the ceiling is visible
+/// without opening the file.
+pub fn format_bottleneck(pipeline: &PipelineStats) -> String {
+    let Some(slowest) = &pipeline.bottleneck else {
+        let reason = if pipeline.stages.iter().any(|stage| stage.samples > 0) {
+            "every sampled step measured 0 ns"
+        } else {
+            "no step had a recorded process_time window"
+        };
+        return format!("Bottleneck: unknown ({reason}).");
+    };
+    let speedup = match pipeline.max_pipeline_speedup {
+        Some(speedup) => format!("{speedup:.2}x"),
+        None => "unknown".to_string(),
+    };
+    format!(
+        "Bottleneck: step {} '{}' at {:.3} ms mean caps the rate at {:.1} Hz; \
+         pipelining every step can buy at most {speedup}.",
+        slowest.index,
+        slowest.task,
+        slowest.mean_ns / 1e6,
+        slowest.max_rate_hz,
+    )
+}
+
 pub fn write_logstats(stats: &LogStats, path: &Path) -> CuResult<()> {
     let file = File::create(path)
         .map_err(|e| CuError::new_with_cause("Failed to create logstats output", e))?;
@@ -253,14 +467,13 @@ pub fn write_logstats(stats: &LogStats, path: &Path) -> CuResult<()> {
     Ok(())
 }
 
-fn build_output_slots(graph: &CuGraph) -> CuResult<Vec<OutputSlot>> {
-    let packs = collect_output_packs(graph)?;
+fn build_output_slots(packs: &[OutputPackInfo], graph: &CuGraph) -> Vec<OutputSlot> {
     let edges_by_src = build_edges_by_src_msg(graph);
     let total_msgs: usize = packs.iter().map(|pack| pack.msg_types.len()).sum();
     let mut slots = Vec::with_capacity(total_msgs);
 
     for pack in packs {
-        for msg in pack.msg_types {
+        for msg in &pack.msg_types {
             let edges = edges_by_src
                 .get(&SrcMsgKey {
                     src: pack.src.clone(),
@@ -272,7 +485,7 @@ fn build_output_slots(graph: &CuGraph) -> CuResult<Vec<OutputSlot>> {
         }
     }
 
-    Ok(slots)
+    slots
 }
 
 fn build_edge_accumulators(graph: &CuGraph) -> HashMap<EdgeKey, EdgeAccumulator> {
@@ -310,14 +523,18 @@ fn build_edges_by_src_msg(graph: &CuGraph) -> HashMap<SrcMsgKey, Vec<EdgeKey>> {
 }
 
 #[derive(Debug)]
-struct OutputPackInfo {
-    culist_index: u32,
-    src: String,
-    msg_types: Vec<String>,
+pub(crate) struct OutputPackInfo {
+    pub(crate) culist_index: u32,
+    pub(crate) src: String,
+    pub(crate) msg_types: Vec<String>,
 }
 
-fn collect_output_packs(graph: &CuGraph) -> CuResult<Vec<OutputPackInfo>> {
-    let plan = compute_runtime_plan(graph)?;
+pub(crate) fn collect_output_packs(
+    graph: &CuGraph,
+    plan_policy: PlanPolicy,
+    plan_profile: &PlanProfile,
+) -> CuResult<Vec<OutputPackInfo>> {
+    let plan = compute_runtime_plan(graph, plan_policy, plan_profile)?;
     let mut packs = Vec::new();
     collect_output_packs_from_loop(&plan, graph, &mut packs)?;
     packs.sort_by_key(|pack| pack.culist_index);
@@ -363,11 +580,11 @@ fn compute_end_to_end_latency(
     end.checked_sub(start).map(CuDuration::from_nanos)
 }
 
-fn extract_start_time_ns(meta: &dyn CuMsgMetadataTrait) -> Option<u64> {
+pub(crate) fn extract_start_time_ns(meta: &dyn CuMsgMetadataTrait) -> Option<u64> {
     option_time_ns(meta.process_time().start)
 }
 
-fn extract_end_time_ns(meta: &dyn CuMsgMetadataTrait) -> Option<u64> {
+pub(crate) fn extract_end_time_ns(meta: &dyn CuMsgMetadataTrait) -> Option<u64> {
     option_time_ns(meta.process_time().end)
 }
 
@@ -497,6 +714,84 @@ mod tests {
         assert!(stats.throughput_bytes_per_sec.is_none());
     }
 
+    fn stage(task: &str, durations: &[u64]) -> StageAccumulator {
+        let mut acc = StageAccumulator::new(task.to_string());
+        for &duration in durations {
+            acc.record_sample(duration);
+        }
+        acc
+    }
+
+    #[test]
+    fn pipeline_names_the_slowest_step_and_its_rate() {
+        // 1 ms + 4 ms + 1 ms serial; the 4 ms step alone allows 250 Hz.
+        let stages = vec![
+            stage("cam", &[1_000_000, 1_000_000]),
+            stage("detect", &[4_000_000, 4_000_000]),
+            stage("brake", &[1_000_000, 1_000_000]),
+        ];
+        let pipeline = finalize_pipeline(&stages);
+
+        let slowest = pipeline.bottleneck.unwrap();
+        assert_eq!(slowest.task, "detect");
+        assert_eq!(slowest.index, 1);
+        assert!((slowest.max_rate_hz - 250.0).abs() < 1e-6);
+        assert!((pipeline.serial_cycle_ns.unwrap() - 6_000_000.0).abs() < 1e-6);
+        // 6 ms serial / 4 ms bottleneck: pipelining buys 1.5x, not 3x.
+        assert!((pipeline.max_pipeline_speedup.unwrap() - 1.5).abs() < 1e-6);
+
+        let shares: Vec<f64> = pipeline
+            .stages
+            .iter()
+            .map(|stage| stage.share.unwrap())
+            .collect();
+        assert!((shares.iter().sum::<f64>() - 1.0).abs() < 1e-6);
+        assert!((shares[1] - 4.0 / 6.0).abs() < 1e-6);
+    }
+
+    #[test]
+    fn pipeline_cycle_is_unknown_when_a_step_was_never_sampled() {
+        // An unsampled step must not make the serial cycle look cheaper.
+        let stages = vec![stage("cam", &[1_000_000]), stage("detect", &[])];
+        let pipeline = finalize_pipeline(&stages);
+
+        assert!(pipeline.serial_cycle_ns.is_none());
+        assert!(pipeline.max_pipeline_speedup.is_none());
+        assert_eq!(pipeline.bottleneck.unwrap().task, "cam");
+        assert_eq!(pipeline.stages[1].samples, 0);
+        assert!(pipeline.stages[1].share.is_none());
+        assert!(pipeline.stages[1].duration.mean_ns.is_none());
+    }
+
+    #[test]
+    fn pipeline_without_any_sample_reports_no_bottleneck() {
+        let pipeline = finalize_pipeline(&[stage("cam", &[])]);
+        assert!(pipeline.bottleneck.is_none());
+        assert!(format_bottleneck(&pipeline).contains("unknown"));
+    }
+
+    #[test]
+    fn pack_ranges_follow_the_flattened_slot_order() {
+        let packs = vec![
+            OutputPackInfo {
+                culist_index: 0,
+                src: "cam".to_string(),
+                msg_types: vec!["Image".to_string(), "Meta".to_string()],
+            },
+            OutputPackInfo {
+                culist_index: 1,
+                src: "detect".to_string(),
+                msg_types: vec!["Boxes".to_string()],
+            },
+        ];
+        let ranges = build_pack_ranges(&packs);
+
+        assert_eq!(ranges.len(), 2);
+        assert_eq!((ranges[0].start, ranges[0].len), (0, 2));
+        assert_eq!((ranges[1].start, ranges[1].len), (2, 1));
+        assert_eq!(ranges[1].task, "detect");
+    }
+
     #[test]
     fn perf_stats_skip_missing_latency() {
         let mut perf = PerfAccumulator::new();
diff --git a/core/cu29_export/src/schedule_profile.rs b/core/cu29_export/src/schedule_profile.rs
new file mode 100644
index 00000000000..a30736f240f
--- /dev/null
+++ b/core/cu29_export/src/schedule_profile.rs
@@ -0,0 +1,148 @@
+//! Builds a measured [`PlanProfile`] from a recorded log.
+//!
+//! Per-task durations come from the `process_time` window every message
+//! metadata already carries; no extra instrumentation is involved. The output
+//! RON is the exact value of the config's `runtime.plan_profile` field, which
+//! any profile-guided [`PlanPolicy`](cu29::config::PlanPolicy) then reads
+//! (see `doc/sched-v0.md`).
+
+use crate::copperlists_reader;
+use crate::logstats::{build_pack_ranges, collect_output_packs, sample_step_duration_ns};
+use cu29::config::{CuConfig, PlanProfile};
+use cu29::prelude::{CopperListTuple, CuPayloadRawBytes};
+use cu29::{CuError, CuResult};
+use std::collections::BTreeMap;
+use std::io::Read;
+use std::path::Path;
+
+/// Which statistic of the sampled durations fills the profile.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, clap::ValueEnum)]
+pub enum ProfileStat {
+    /// Arithmetic mean of the samples.
+    #[default]
+    Mean,
+    /// 99th percentile: robust to outliers, still pessimistic.
+    P99,
+    /// Largest observed sample.
+    Max,
+}
+
+pub fn compute_schedule_profile<P>(
+    mut reader: impl Read,
+    config: &CuConfig,
+    mission: Option<&str>,
+    stat: ProfileStat,
+) -> CuResult<PlanProfile>
+where
+    P: CopperListTuple + CuPayloadRawBytes,
+{
+    let graph = config.get_graph(mission)?;
+    let packs = collect_output_packs(graph, config.plan_policy(), &config.plan_profile())?;
+    let ranges = build_pack_ranges(&packs);
+
+    let mut samples: BTreeMap<String, Vec<u64>> = BTreeMap::new();
+    for culist in copperlists_reader::<P>(&mut reader) {
+        let cumsgs = culist.msgs.cumsgs();
+        for range in &ranges {
+            if let Some(duration) = sample_step_duration_ns(&cumsgs, range) {
+                samples
+                    .entry(range.task.clone())
+                    .or_default()
+                    .push(duration);
+            }
+        }
+    }
+
+    Ok(PlanProfile {
+        task_duration_ns: finalize_samples(samples, stat),
+    })
+}
+
+fn finalize_samples(
+    samples: BTreeMap<String, Vec<u64>>,
+    stat: ProfileStat,
+) -> BTreeMap<String, u64> {
+    let mut task_duration_ns = BTreeMap::new();
+    for (task, mut durations) in samples {
+        if durations.is_empty() {
+            continue;
+        }
+        durations.sort_unstable();
+        let value = match stat {
+            ProfileStat::Mean => {
+                (durations.iter().map(|&d| d as u128).sum::<u128>() / durations.len() as u128)
+                    as u64
+            }
+            // Nearest rank: the smallest sample at or above 99% of the set.
+            // `(len - 1) * 99 / 100` rounds the other way and returns the
+            // minimum for two samples, which is never a p99.
+            ProfileStat::P99 => {
+                let rank = (durations.len() as u128 * 99).div_ceil(100).max(1) as usize;
+                durations[rank - 1]
+            }
+            ProfileStat::Max => *durations.last().unwrap(),
+        };
+        task_duration_ns.insert(task, value);
+    }
+    task_duration_ns
+}
+
+/// Writes the profile as pretty RON: the pasteable `plan_profile:` value.
+pub fn write_schedule_profile(profile: &PlanProfile, path: &Path) -> CuResult<()> {
+    let ron = ron::ser::to_string_pretty(profile, ron::ser::PrettyConfig::default())
+        .map_err(|e| CuError::new_with_cause("Failed to serialize schedule profile", e))?;
+    std::fs::write(path, ron)
+        .map_err(|e| CuError::new_with_cause("Failed to write schedule profile", e))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn samples(entries: &[(&str, &[u64])]) -> BTreeMap<String, Vec<u64>> {
+        entries
+            .iter()
+            .map(|(task, durations)| (task.to_string(), durations.to_vec()))
+            .collect()
+    }
+
+    #[test]
+    fn finalize_picks_the_requested_statistic() {
+        let input = samples(&[("cam", &[100, 200, 300]), ("imu", &[10])]);
+        let mean = finalize_samples(input.clone(), ProfileStat::Mean);
+        assert_eq!(mean.get("cam"), Some(&200));
+        assert_eq!(mean.get("imu"), Some(&10));
+
+        let max = finalize_samples(input.clone(), ProfileStat::Max);
+        assert_eq!(max.get("cam"), Some(&300));
+
+        // Nearest rank over 3 samples lands on the largest one.
+        let p99 = finalize_samples(input, ProfileStat::P99);
+        assert_eq!(p99.get("cam"), Some(&300));
+        assert_eq!(p99.get("imu"), Some(&10));
+    }
+
+    #[test]
+    fn p99_never_returns_the_minimum() {
+        // Nearest rank on tiny sets: a p99 must stay pessimistic.
+        let two = finalize_samples(samples(&[("cam", &[10, 900])]), ProfileStat::P99);
+        assert_eq!(two.get("cam"), Some(&900));
+
+        let hundred: Vec<u64> = (1..=100).collect();
+        let wide = finalize_samples(samples(&[("cam", &hundred)]), ProfileStat::P99);
+        assert_eq!(wide.get("cam"), Some(&99));
+    }
+
+    #[test]
+    fn profile_snippet_round_trips_through_ron() {
+        let profile = PlanProfile {
+            task_duration_ns: samples(&[("cam", &[0])])
+                .into_keys()
+                .map(|task| (task, 1234u64))
+                .collect(),
+        };
+        let ron = ron::ser::to_string_pretty(&profile, ron::ser::PrettyConfig::default()).unwrap();
+        let parsed: PlanProfile = ron::from_str(&ron).unwrap();
+        assert_eq!(parsed, profile);
+    }
+}
diff --git a/core/cu29_runtime/copper-crash-1785255090794-661645.txt b/core/cu29_runtime/copper-crash-1785255090794-661645.txt
new file mode 100644
index 00000000000..f155576450a
--- /dev/null
+++ b/core/cu29_runtime/copper-crash-1785255090794-661645.txt
@@ -0,0 +1,80 @@
+Copper panic
+time_unix_ms: 1785255090794
+thread: generated_anytime_runtime_refines_and_uses_earliest_range_tov
+location: core/cu29_runtime/tests/anytime_generated.rs:152:5
+message: assertion `left == right` failed
+  left: 2
+ right: 1
+crash_report: /home/ubuntu/copper-rs/core/cu29_runtime/copper-crash-1785255090794-661645.txt
+
+Backtrace:
+   0: cu29_runtime::monitoring::PanicReport::capture
+             at ./src/monitoring.rs:812:24
+   1: cu29_runtime::monitoring::ensure_runtime_panic_hook_installed::{{closure}}::{{closure}}
+             at ./src/monitoring.rs:911:30
+   2: <alloc::boxed::Box<dyn for<'a, 'b> core::ops::function::Fn<(&'a std::panic::PanicHookInfo<'b>,), Output = ()> + core::marker::Sync + core::marker::Send> as core::ops::function::Fn<(&std::panic::PanicHookInfo,)>>::call
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/alloc/src/boxed.rs:2254:9
+   3: std::panicking::panic_with_hook
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panicking.rs:833:13
+   4: std::panicking::panic_handler::{closure#0}
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panicking.rs:698:13
+   5: std::sys::backtrace::__rust_end_short_backtrace::<std::panicking::panic_handler::{closure#0}, !>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/sys/backtrace.rs:182:18
+   6: __rustc::rust_begin_unwind
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panicking.rs:689:5
+   7: core::panicking::panic_fmt
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/panicking.rs:80:14
+   8: core::panicking::assert_failed_inner
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/panicking.rs:439:17
+   9: core::panicking::assert_failed::<usize, usize>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/panicking.rs:394:5
+  10: anytime_generated::generated_anytime_runtime_refines_and_uses_earliest_range_tov
+             at ./tests/anytime_generated.rs:152:5
+  11: anytime_generated::generated_anytime_runtime_refines_and_uses_earliest_range_tov::{{closure}}
+             at ./tests/anytime_generated.rs:128:71
+  12: core::ops::function::FnOnce::call_once
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/ops/function.rs:250:5
+  13: <fn() -> core::result::Result<(), alloc::string::String> as core::ops::function::FnOnce<()>>::call_once
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/ops/function.rs:250:5
+  14: test::__rust_begin_short_backtrace::<core::result::Result<(), alloc::string::String>, fn() -> core::result::Result<(), alloc::string::String>>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/test/src/lib.rs:663:18
+  15: test::run_test_in_process::{closure#0}
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/test/src/lib.rs:686:74
+  16: <core::panic::unwind_safe::AssertUnwindSafe<test::run_test_in_process::{closure#0}> as core::ops::function::FnOnce<()>>::call_once
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/panic/unwind_safe.rs:274:9
+  17: std::panicking::catch_unwind::do_call::<core::panic::unwind_safe::AssertUnwindSafe<test::run_test_in_process::{closure#0}>, core::result::Result<(), alloc::string::String>>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panicking.rs:581:40
+  18: std::panicking::catch_unwind::<core::result::Result<(), alloc::string::String>, core::panic::unwind_safe::AssertUnwindSafe<test::run_test_in_process::{closure#0}>>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panicking.rs:544:19
+  19: std::panic::catch_unwind::<core::panic::unwind_safe::AssertUnwindSafe<test::run_test_in_process::{closure#0}>, core::result::Result<(), alloc::string::String>>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panic.rs:359:14
+  20: test::run_test_in_process
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/test/src/lib.rs:686:27
+  21: test::run_test::{closure#0}
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/test/src/lib.rs:607:43
+  22: test::run_test::{closure#1}
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/test/src/lib.rs:637:41
+  23: std::sys::backtrace::__rust_begin_short_backtrace::<test::run_test::{closure#1}, ()>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/sys/backtrace.rs:166:18
+  24: std::thread::lifecycle::spawn_unchecked::<test::run_test::{closure#1}, ()>::{closure#1}::{closure#0}
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/thread/lifecycle.rs:91:13
+  25: <core::panic::unwind_safe::AssertUnwindSafe<std::thread::lifecycle::spawn_unchecked<test::run_test::{closure#1}, ()>::{closure#1}::{closure#0}> as core::ops::function::FnOnce<()>>::call_once
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/panic/unwind_safe.rs:274:9
+  26: std::panicking::catch_unwind::do_call::<core::panic::unwind_safe::AssertUnwindSafe<std::thread::lifecycle::spawn_unchecked<test::run_test::{closure#1}, ()>::{closure#1}::{closure#0}>, ()>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panicking.rs:581:40
+  27: std::panicking::catch_unwind::<(), core::panic::unwind_safe::AssertUnwindSafe<std::thread::lifecycle::spawn_unchecked<test::run_test::{closure#1}, ()>::{closure#1}::{closure#0}>>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panicking.rs:544:19
+  28: std::panic::catch_unwind::<core::panic::unwind_safe::AssertUnwindSafe<std::thread::lifecycle::spawn_unchecked<test::run_test::{closure#1}, ()>::{closure#1}::{closure#0}>, ()>
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/panic.rs:359:14
+  29: std::thread::lifecycle::spawn_unchecked::<test::run_test::{closure#1}, ()>::{closure#1}
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/thread/lifecycle.rs:89:26
+  30: <std::thread::lifecycle::spawn_unchecked<test::run_test::{closure#1}, ()>::{closure#1} as core::ops::function::FnOnce<()>>::call_once::{shim:vtable#0}
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/ops/function.rs:250:5
+  31: <alloc::boxed::Box<dyn core::ops::function::FnOnce<(), Output = ()> + core::marker::Send> as core::ops::function::FnOnce<()>>::call_once
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/alloc/src/boxed.rs:2240:9
+  32: <std::sys::thread::unix::Thread>::new::thread_start
+             at /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/sys/thread/unix.rs:118:17
+  33: start_thread
+             at ./nptl/pthread_create.c:447:8
+  34: clone3
+             at ./misc/../sysdeps/unix/sysv/linux/x86_64/clone3.S:78:0
diff --git a/core/cu29_runtime/src/config.rs b/core/cu29_runtime/src/config.rs
index 1ff4eb55667..d3de37ef2fc 100644
--- a/core/cu29_runtime/src/config.rs
+++ b/core/cu29_runtime/src/config.rs
@@ -642,6 +642,14 @@ pub const DEFAULT_BACKGROUND_POOL: &str = "background";
 #[allow(dead_code)] // consumed by cu29_derive; unused in some binary targets
 pub const RT_POOL: &str = "rt";
 
+/// Schema version of the `log-stats` JSON document. Lives here because both the
+/// writer (`cu29_export::logstats`) and the reader (the `cu29-rendercfg` bin)
+/// must agree on it; keeping one copy per crate let them drift.
+///
+/// 1: edges + perf. 2: adds the `pipeline` section.
+#[allow(dead_code)] // consumed by cu29_export; unused in some binary targets
+pub const LOGSTATS_SCHEMA_VERSION: u32 = 2;
+
 /// How a task is backgrounded.
 ///
 /// Either a simple on/off flag (`background: true`), which runs the task on the
@@ -1989,6 +1997,98 @@ pub struct LoggingCodecSpec {
     pub config: Option<ComponentConfig>,
 }
 
+/// Ordering algorithm for the compile-time execution plan.
+///
+/// Every variant names an algorithm and nothing else; measured data lives in
+/// [`PlanProfile`], next to this field. Every variant emits a valid
+/// topological order of the task graph, so a policy can only pick a better or
+/// worse order, never a wrong one. See `doc/sched-v0.md` for the roadmap.
+#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PlanPolicy {
+    /// The historical heuristic: BFS from the sources, a node entering the
+    /// order once all of its producers are ordered. The default. Ignores the
+    /// profile.
+    #[default]
+    TopoBfs,
+    /// List scheduling that always picks the ready node with the longest
+    /// remaining critical path. Needs a non-empty [`PlanProfile`].
+    CriticalPathFirst,
+}
+
+impl PlanPolicy {
+    fn is_default(&self) -> bool {
+        *self == Self::default()
+    }
+
+    /// Whether this algorithm reads [`PlanProfile`] and therefore needs a
+    /// non-empty one.
+    #[allow(dead_code)] // The rendercfg bin doesn't plan, only the lib does.
+    pub fn needs_profile(&self) -> bool {
+        matches!(self, Self::CriticalPathFirst)
+    }
+}
+
+/// How `parallel-rt` stage workers map onto the `rt` pool's CPU affinity list.
+///
+/// Like [`PlanPolicy`], every variant names an algorithm; the measurement it
+/// reads lives in [`PlanProfile`]. This is placement, not ordering: it decides
+/// *where* a step runs, never *when*.
+///
+/// Ignored when the `parallel-rt` feature is off, or when the `rt` pool
+/// declares no affinity — there is nothing to place in either case.
+#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum CorePlacement {
+    /// Worker `i` pins to `affinity[i % affinity.len()]`. The historical
+    /// behavior and the default. Ignores the profile.
+    #[default]
+    Spread,
+    /// Longest-processing-time-first bin packing: order the steps heaviest
+    /// first and give each to the least loaded core. Balances per-core load
+    /// when steps outnumber cores. Needs a non-empty [`PlanProfile`].
+    LongestFirst,
+}
+
+impl CorePlacement {
+    fn is_default(&self) -> bool {
+        *self == Self::default()
+    }
+
+    /// Whether this algorithm reads [`PlanProfile`] and therefore needs a
+    /// non-empty one.
+    #[allow(dead_code)] // The rendercfg bin doesn't plan, only the lib does.
+    pub fn needs_profile(&self) -> bool {
+        matches!(self, Self::LongestFirst)
+    }
+}
+
+/// Measured task timings that feed profile-guided planning.
+///
+/// The profile is an input, not a policy: it is orthogonal to which algorithm
+/// reads it, and later stages (`parallel-rt` core packing) read the same
+/// numbers. Written by the `cu29_export` `schedule-profile` subcommand and
+/// pasted inline into the config, so the unified log embeds it and offline
+/// tools recompute the exact same plan.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Default)]
+pub struct PlanProfile {
+    /// Measured `process()` duration per task id, in nanoseconds. Tasks
+    /// absent from the map weigh zero.
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub task_duration_ns: BTreeMap<String, u64>,
+}
+
+impl PlanProfile {
+    /// No measurement at all: every policy that needs a profile rejects this.
+    pub fn is_empty(&self) -> bool {
+        self.task_duration_ns.is_empty()
+    }
+
+    /// Measured duration of `task_id`, zero when the task was never sampled.
+    #[allow(dead_code)] // The rendercfg bin doesn't plan, only the lib does.
+    pub fn task_duration_ns(&self, task_id: &str) -> u64 {
+        self.task_duration_ns.get(task_id).copied().unwrap_or(0)
+    }
+}
+
 #[derive(Serialize, Deserialize, Default, Debug, Clone)]
 pub struct RuntimeConfig {
     /// Set a CopperList execution rate target in Hz
@@ -2007,6 +2107,21 @@ pub struct RuntimeConfig {
     /// threads and this section is ignored.
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub thread_pools: Vec<ThreadPoolConfig>,
+
+    /// Ordering algorithm for the compile-time execution plan (see `doc/sched-v0.md`).
+    #[serde(default, skip_serializing_if = "PlanPolicy::is_default")]
+    pub plan_policy: PlanPolicy,
+
+    /// Measured task timings feeding [`PlanPolicy::CriticalPathFirst`] and
+    /// [`CorePlacement::LongestFirst`]. Written by
+    /// `cu29_export ... schedule-profile`.
+    #[serde(default, skip_serializing_if = "PlanProfile::is_empty")]
+    pub plan_profile: PlanProfile,
+
+    /// How `parallel-rt` stage workers map onto the `rt` pool's CPU affinity
+    /// list (see `doc/sched-v0.md`).
+    #[serde(default, skip_serializing_if = "CorePlacement::is_default")]
+    pub core_placement: CorePlacement,
 }
 
 /// Smallest valid real-time priority for [`SchedulingPolicy::Fifo`]/[`SchedulingPolicy::RoundRobin`].
@@ -2963,6 +3078,31 @@ impl CuConfig {
         self.runtime.as_ref()
     }
 
+    /// The execution-plan ordering policy of this config (`runtime.plan_policy`).
+    #[allow(dead_code)]
+    pub fn plan_policy(&self) -> PlanPolicy {
+        self.runtime
+            .as_ref()
+            .map(|runtime| runtime.plan_policy)
+            .unwrap_or_default()
+    }
+
+    #[allow(dead_code)] // The rendercfg bin doesn't plan, only the lib does.
+    pub fn plan_profile(&self) -> PlanProfile {
+        self.runtime
+            .as_ref()
+            .map(|runtime| runtime.plan_profile.clone())
+            .unwrap_or_default()
+    }
+
+    #[allow(dead_code)] // The rendercfg bin doesn't plan, only the lib does.
+    pub fn core_placement(&self) -> CorePlacement {
+        self.runtime
+            .as_ref()
+            .map(|runtime| runtime.core_placement)
+            .unwrap_or_default()
+    }
+
     #[allow(dead_code)]
     pub fn find_task_node(&self, mission_id: Option<&str>, task_id: &str) -> Option<&Node> {
         self.get_graph(mission_id)
@@ -5644,6 +5784,59 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_runtime_plan_policy_parses_and_defaults() {
+        let with_policy = r#"(
+            tasks: [(id: "src", type: "a"), (id: "sink", type: "b")],
+            cnx: [(src: "src", dst: "sink", msg: "msg::A")],
+            runtime: (plan_policy: TopoBfs)
+        )"#;
+        let config = read_configuration_str(with_policy.to_string(), None).unwrap();
+        assert_eq!(config.plan_policy(), PlanPolicy::TopoBfs);
+
+        let without_policy = r#"(
+            tasks: [(id: "src", type: "a"), (id: "sink", type: "b")],
+            cnx: [(src: "src", dst: "sink", msg: "msg::A")],
+        )"#;
+        let config = read_configuration_str(without_policy.to_string(), None).unwrap();
+        assert_eq!(config.plan_policy(), PlanPolicy::default());
+    }
+
+    #[test]
+    fn test_runtime_plan_profile_parses_independently_of_the_policy() {
+        let txt = r#"(
+            tasks: [(id: "src", type: "a"), (id: "sink", type: "b")],
+            cnx: [(src: "src", dst: "sink", msg: "msg::A")],
+            runtime: (
+                plan_policy: CriticalPathFirst,
+                plan_profile: (task_duration_ns: {"src": 1200, "sink": 300}),
+            )
+        )"#;
+        let config = read_configuration_str(txt.to_string(), None).unwrap();
+        assert_eq!(config.plan_policy(), PlanPolicy::CriticalPathFirst);
+        let profile = config.plan_profile();
+        assert_eq!(profile.task_duration_ns("src"), 1200);
+        assert_eq!(profile.task_duration_ns("sink"), 300);
+        // An unsampled task weighs zero rather than failing the lookup.
+        assert_eq!(profile.task_duration_ns("absent"), 0);
+    }
+
+    #[test]
+    fn test_runtime_plan_profile_is_orthogonal_to_the_policy() {
+        // The default policy ignores the profile, but still round-trips it, so
+        // switching policy needs no re-measurement.
+        let txt = r#"(
+            tasks: [(id: "src", type: "a"), (id: "sink", type: "b")],
+            cnx: [(src: "src", dst: "sink", msg: "msg::A")],
+            runtime: (
+                plan_profile: (task_duration_ns: {"src": 42}),
+            )
+        )"#;
+        let config = read_configuration_str(txt.to_string(), None).unwrap();
+        assert_eq!(config.plan_policy(), PlanPolicy::TopoBfs);
+        assert_eq!(config.plan_profile().task_duration_ns("src"), 42);
+    }
+
     /// Builds a src -> any -> sink config with the given `anytime:` policy body,
     /// extra node attributes (e.g. `, background: true`) and top-level extras
     /// (e.g. `runtime: (rate_target_hz: 100),`).
diff --git a/core/cu29_runtime/src/curuntime.rs b/core/cu29_runtime/src/curuntime.rs
index b9d23b11f82..1f1211ff40f 100644
--- a/core/cu29_runtime/src/curuntime.rs
+++ b/core/cu29_runtime/src/curuntime.rs
@@ -5,7 +5,8 @@
 use crate::app::Subsystem;
 use crate::config::{ComponentConfig, CuDirection, DEFAULT_KEYFRAME_INTERVAL, Node, TaskKind};
 use crate::config::{
-    CuConfig, CuGraph, MAX_RATE_TARGET_HZ, NodeId, RuntimeConfig, resolve_task_kind_for_id,
+    CorePlacement, CuConfig, CuGraph, MAX_RATE_TARGET_HZ, NodeId, PlanPolicy, PlanProfile,
+    RuntimeConfig, resolve_task_kind_for_id,
 };
 use crate::copperlist::{CopperList, CopperListState, CuListZeroedInit, CuListsManager};
 use crate::cutask::{BincodeAdapter, Freezable};
@@ -50,9 +51,10 @@ use cu29_value::to_value;
 #[cfg(all(feature = "std", any(feature = "async-cl-io", feature = "parallel-rt")))]
 use alloc::alloc::{alloc_zeroed, handle_alloc_error};
 use alloc::boxed::Box;
-use alloc::collections::{BTreeSet, VecDeque};
+use alloc::collections::{BTreeMap, BTreeSet, BinaryHeap, VecDeque};
 use alloc::format;
 use alloc::string::{String, ToString};
+use alloc::vec;
 use alloc::vec::Vec;
 use bincode::enc::EncoderImpl;
 use bincode::enc::write::{SizeWriter, SliceWriter};
@@ -60,6 +62,7 @@ use bincode::error::EncodeError;
 use bincode::{Decode, Encode};
 #[cfg(all(feature = "std", any(feature = "async-cl-io", feature = "parallel-rt")))]
 use core::alloc::Layout;
+use core::cmp::Reverse;
 use core::fmt::Result as FmtResult;
 use core::fmt::{Debug, Formatter};
 use core::marker::PhantomData;
@@ -1722,188 +1725,63 @@ fn sort_inputs_by_connection_order(input_msg_indices_types: &mut [CuInputMsg]) {
     input_msg_indices_types.sort_by_key(|input| input.connection_order);
 }
 
-/// Explores a subbranch and build the partial plan out of it.
-fn plan_tasks_tree_branch(
+/// Explores a subbranch and appends every node whose producers are all
+/// ordered to the order.
+fn topo_bfs_branch(
     graph: &CuGraph,
-    mut next_culist_output_index: u32,
     starting_point: NodeId,
-    plan: &mut Vec<CuExecutionUnit>,
-) -> CuResult<(u32, bool)> {
+    order: &mut Vec<NodeId>,
+) -> CuResult<bool> {
     #[cfg(all(feature = "std", feature = "macro_debug"))]
     eprintln!("-- starting branch from node {starting_point}");
 
     let mut handled = false;
 
     for id in graph.bfs_nodes(starting_point) {
-        let node_ref = graph.get_node(id).unwrap();
         #[cfg(all(feature = "std", feature = "macro_debug"))]
-        eprintln!("  Visiting node: {node_ref:?}");
+        eprintln!("  Visiting node: {id}");
 
-        let mut input_msg_indices_types: Vec<CuInputMsg> = Vec::new();
-        let output_msg_pack: Option<CuOutputPack>;
         let task_type = find_task_type_for_id(graph, id)?;
 
-        match task_type {
-            CuTaskType::Source => {
-                #[cfg(all(feature = "std", feature = "macro_debug"))]
-                eprintln!("    → Source node, assign output index {next_culist_output_index}");
-                let msg_types = graph.get_node_output_msg_types_by_id(id)?;
-                if msg_types.is_empty() {
-                    return Err(CuError::from(format!(
-                        "Source node '{}' has no declared outputs",
-                        node_ref.get_id()
-                    )));
-                }
-                output_msg_pack = Some(CuOutputPack {
-                    culist_index: next_culist_output_index,
-                    msg_types,
-                });
-                next_culist_output_index += 1;
-            }
-            CuTaskType::Sink => {
-                let mut edge_ids = graph.get_dst_edges(id).unwrap_or_default();
-                edge_ids.sort();
-                #[cfg(all(feature = "std", feature = "macro_debug"))]
-                eprintln!("    → Sink with incoming edges: {edge_ids:?}");
-                for edge_id in edge_ids {
-                    let edge = graph
-                        .edge(edge_id)
-                        .unwrap_or_else(|| panic!("Missing edge {edge_id} for node {id}"));
-                    let pid = graph
-                        .get_node_id_by_name(edge.src.as_str())
-                        .unwrap_or_else(|| {
-                            panic!("Missing source node '{}' for edge {edge_id}", edge.src)
-                        });
-                    let output_pack = find_output_pack_from_nodeid(pid, plan);
-                    if let Some(output_pack) = output_pack {
-                        #[cfg(all(feature = "std", feature = "macro_debug"))]
-                        eprintln!("      ✓ Input from {pid} ready: {output_pack:?}");
-                        let msg_type = edge.msg.as_str();
-                        let src_port = output_pack
-                            .msg_types
-                            .iter()
-                            .position(|msg| msg == msg_type)
-                            .unwrap_or_else(|| {
-                                panic!(
-                                    "Missing output port for message type '{msg_type}' on node {pid}"
-                                )
-                            });
-                        input_msg_indices_types.push(CuInputMsg {
-                            culist_index: output_pack.culist_index,
-                            msg_type: msg_type.to_string(),
-                            src_port,
-                            edge_id,
-                            connection_order: edge.order,
-                        });
-                    } else {
-                        #[cfg(all(feature = "std", feature = "macro_debug"))]
-                        eprintln!("      ✗ Input from {pid} not ready, returning");
-                        return Ok((next_culist_output_index, handled));
-                    }
+        if task_type != CuTaskType::Source {
+            let mut edge_ids = graph.get_dst_edges(id).unwrap_or_default();
+            edge_ids.sort();
+            for edge_id in edge_ids {
+                let edge = graph
+                    .edge(edge_id)
+                    .unwrap_or_else(|| panic!("Missing edge {edge_id} for node {id}"));
+                let pid = graph
+                    .get_node_id_by_name(edge.src.as_str())
+                    .unwrap_or_else(|| {
+                        panic!("Missing source node '{}' for edge {edge_id}", edge.src)
+                    });
+                if !order.contains(&pid) {
+                    #[cfg(all(feature = "std", feature = "macro_debug"))]
+                    eprintln!("      ✗ Input from {pid} not ready, returning");
+                    return Ok(handled);
                 }
-                output_msg_pack = Some(CuOutputPack {
-                    culist_index: next_culist_output_index,
-                    msg_types: Vec::from(["()".to_string()]),
-                });
-                next_culist_output_index += 1;
-            }
-            CuTaskType::Regular => {
-                let mut edge_ids = graph.get_dst_edges(id).unwrap_or_default();
-                edge_ids.sort();
-                #[cfg(all(feature = "std", feature = "macro_debug"))]
-                eprintln!("    → Regular task with incoming edges: {edge_ids:?}");
-                for edge_id in edge_ids {
-                    let edge = graph
-                        .edge(edge_id)
-                        .unwrap_or_else(|| panic!("Missing edge {edge_id} for node {id}"));
-                    let pid = graph
-                        .get_node_id_by_name(edge.src.as_str())
-                        .unwrap_or_else(|| {
-                            panic!("Missing source node '{}' for edge {edge_id}", edge.src)
-                        });
-                    let output_pack = find_output_pack_from_nodeid(pid, plan);
-                    if let Some(output_pack) = output_pack {
-                        #[cfg(all(feature = "std", feature = "macro_debug"))]
-                        eprintln!("      ✓ Input from {pid} ready: {output_pack:?}");
-                        let msg_type = edge.msg.as_str();
-                        let src_port = output_pack
-                            .msg_types
-                            .iter()
-                            .position(|msg| msg == msg_type)
-                            .unwrap_or_else(|| {
-                                panic!(
-                                    "Missing output port for message type '{msg_type}' on node {pid}"
-                                )
-                            });
-                        input_msg_indices_types.push(CuInputMsg {
-                            culist_index: output_pack.culist_index,
-                            msg_type: msg_type.to_string(),
-                            src_port,
-                            edge_id,
-                            connection_order: edge.order,
-                        });
-                    } else {
-                        #[cfg(all(feature = "std", feature = "macro_debug"))]
-                        eprintln!("      ✗ Input from {pid} not ready, returning");
-                        return Ok((next_culist_output_index, handled));
-                    }
-                }
-                let msg_types = graph.get_node_output_msg_types_by_id(id)?;
-                if msg_types.is_empty() {
-                    return Err(CuError::from(format!(
-                        "Regular node '{}' has no declared outputs",
-                        node_ref.get_id()
-                    )));
-                }
-                output_msg_pack = Some(CuOutputPack {
-                    culist_index: next_culist_output_index,
-                    msg_types,
-                });
-                next_culist_output_index += 1;
             }
         }
 
-        sort_inputs_by_connection_order(&mut input_msg_indices_types);
-
-        if let Some(pos) = plan
-            .iter()
-            .position(|step| matches!(step, CuExecutionUnit::Step(s) if s.node_id == id))
-        {
+        if let Some(pos) = order.iter().position(|&ordered| ordered == id) {
             #[cfg(all(feature = "std", feature = "macro_debug"))]
-            eprintln!("    → Already in plan, modifying existing step");
-            let mut step = plan.remove(pos);
-            if let CuExecutionUnit::Step(ref mut s) = step {
-                s.input_msg_indices_types = input_msg_indices_types;
-            }
-            plan.push(step);
-        } else {
-            #[cfg(all(feature = "std", feature = "macro_debug"))]
-            eprintln!("    → New step added to plan");
-            let step = CuExecutionStep {
-                node_id: id,
-                node: node_ref.clone(),
-                task_type,
-                input_msg_indices_types,
-                output_msg_pack,
-            };
-            plan.push(CuExecutionUnit::Step(Box::new(step)));
+            eprintln!("    → Already ordered, moving to the back");
+            order.remove(pos);
         }
-
+        order.push(id);
         handled = true;
     }
 
     #[cfg(all(feature = "std", feature = "macro_debug"))]
     eprintln!("-- finished branch from node {starting_point} with handled={handled}");
-    Ok((next_culist_output_index, handled))
+    Ok(handled)
 }
 
-/// This is the main heuristics to compute an execution plan at compilation time.
-/// TODO(gbin): Make that heuristic pluggable.
-pub fn compute_runtime_plan(graph: &CuGraph) -> CuResult<CuExecutionLoop> {
-    #[cfg(all(feature = "std", feature = "macro_debug"))]
-    eprintln!("[runtime plan]");
-    let mut plan = Vec::new();
-    let mut next_culist_output_index = 0u32;
+/// The historical ordering heuristic ([`PlanPolicy::TopoBfs`]): repeated BFS
+/// from each source, a node entering the order once all its producers are
+/// ordered.
+fn topo_bfs_order(graph: &CuGraph) -> CuResult<Vec<NodeId>> {
+    let mut order = Vec::new();
 
     let mut queue: VecDeque<NodeId> = VecDeque::new();
     for node_id in graph.node_ids() {
@@ -1919,22 +1797,15 @@ pub fn compute_runtime_plan(graph: &CuGraph) -> CuResult<CuExecutionLoop> {
         #[cfg(all(feature = "std", feature = "macro_debug"))]
         eprintln!("→ Starting BFS from source {start_node}");
         for node_id in graph.bfs_nodes(start_node) {
-            let already_in_plan = plan
-                .iter()
-                .any(|unit| matches!(unit, CuExecutionUnit::Step(s) if s.node_id == node_id));
-            if already_in_plan {
+            if order.contains(&node_id) {
                 #[cfg(all(feature = "std", feature = "macro_debug"))]
-                eprintln!("    → Node {node_id} already planned, skipping");
+                eprintln!("    → Node {node_id} already ordered, skipping");
                 continue;
             }
 
             #[cfg(all(feature = "std", feature = "macro_debug"))]
-            eprintln!("    Planning from node {node_id}");
-            let (new_index, handled) =
-                plan_tasks_tree_branch(graph, next_culist_output_index, node_id, &mut plan)?;
-            next_culist_output_index = new_index;
-
-            if !handled {
+            eprintln!("    Ordering from node {node_id}");
+            if !topo_bfs_branch(graph, node_id, &mut order)? {
                 #[cfg(all(feature = "std", feature = "macro_debug"))]
                 eprintln!("    ✗ Node {node_id} was not handled, skipping enqueue of neighbors");
                 continue;
@@ -1950,13 +1821,229 @@ pub fn compute_runtime_plan(graph: &CuGraph) -> CuResult<CuExecutionLoop> {
         }
     }
 
-    let mut planned_nodes = BTreeSet::new();
-    for unit in &plan {
-        if let CuExecutionUnit::Step(step) = unit {
-            planned_nodes.insert(step.node_id);
+    Ok(order)
+}
+
+/// Builds the executable steps for a step order: assigns copperlist slots in
+/// order and wires each step's inputs to its producers' packs.
+///
+/// Errors out if the order is not topological (an input's producer does not
+/// appear earlier), so a buggy ordering policy fails the build instead of
+/// generating a broken runtime.
+fn build_plan_from_order(graph: &CuGraph, order: &[NodeId]) -> CuResult<Vec<CuExecutionUnit>> {
+    let mut plan: Vec<CuExecutionUnit> = Vec::with_capacity(order.len());
+
+    for (step_index, &id) in order.iter().enumerate() {
+        // Every step consumes exactly one copperlist slot: slot == step rank.
+        let next_culist_output_index = step_index as u32;
+        let node_ref = graph.get_node(id).unwrap();
+        let mut input_msg_indices_types: Vec<CuInputMsg> = Vec::new();
+        let task_type = find_task_type_for_id(graph, id)?;
+
+        if task_type != CuTaskType::Source {
+            let mut edge_ids = graph.get_dst_edges(id).unwrap_or_default();
+            edge_ids.sort();
+            #[cfg(all(feature = "std", feature = "macro_debug"))]
+            eprintln!("    → {task_type:?} with incoming edges: {edge_ids:?}");
+            for edge_id in edge_ids {
+                let edge = graph
+                    .edge(edge_id)
+                    .unwrap_or_else(|| panic!("Missing edge {edge_id} for node {id}"));
+                let pid = graph
+                    .get_node_id_by_name(edge.src.as_str())
+                    .unwrap_or_else(|| {
+                        panic!("Missing source node '{}' for edge {edge_id}", edge.src)
+                    });
+                let output_pack = find_output_pack_from_nodeid(pid, &plan).ok_or_else(|| {
+                    CuError::from(format!(
+                        "Invalid execution order: '{}' consumes '{}' which is not ordered earlier",
+                        node_ref.get_id(),
+                        edge.src
+                    ))
+                })?;
+                #[cfg(all(feature = "std", feature = "macro_debug"))]
+                eprintln!("      ✓ Input from {pid} ready: {output_pack:?}");
+                let msg_type = edge.msg.as_str();
+                let src_port = output_pack
+                    .msg_types
+                    .iter()
+                    .position(|msg| msg == msg_type)
+                    .unwrap_or_else(|| {
+                        panic!("Missing output port for message type '{msg_type}' on node {pid}")
+                    });
+                input_msg_indices_types.push(CuInputMsg {
+                    culist_index: output_pack.culist_index,
+                    msg_type: msg_type.to_string(),
+                    src_port,
+                    edge_id,
+                    connection_order: edge.order,
+                });
+            }
+        }
+
+        let output_msg_pack = match task_type {
+            CuTaskType::Source => {
+                #[cfg(all(feature = "std", feature = "macro_debug"))]
+                eprintln!("    → Source node, assign output index {next_culist_output_index}");
+                let msg_types = graph.get_node_output_msg_types_by_id(id)?;
+                if msg_types.is_empty() {
+                    return Err(CuError::from(format!(
+                        "Source node '{}' has no declared outputs",
+                        node_ref.get_id()
+                    )));
+                }
+                Some(CuOutputPack {
+                    culist_index: next_culist_output_index,
+                    msg_types,
+                })
+            }
+            CuTaskType::Sink => Some(CuOutputPack {
+                culist_index: next_culist_output_index,
+                msg_types: Vec::from(["()".to_string()]),
+            }),
+            CuTaskType::Regular => {
+                let msg_types = graph.get_node_output_msg_types_by_id(id)?;
+                if msg_types.is_empty() {
+                    return Err(CuError::from(format!(
+                        "Regular node '{}' has no declared outputs",
+                        node_ref.get_id()
+                    )));
+                }
+                Some(CuOutputPack {
+                    culist_index: next_culist_output_index,
+                    msg_types,
+                })
+            }
+        };
+
+        sort_inputs_by_connection_order(&mut input_msg_indices_types);
+
+        plan.push(CuExecutionUnit::Step(Box::new(CuExecutionStep {
+            node_id: id,
+            node: node_ref.clone(),
+            task_type,
+            input_msg_indices_types,
+            output_msg_pack,
+        })));
+    }
+
+    Ok(plan)
+}
+
+/// Critical-path-first list scheduling over measured task durations
+/// ([`PlanPolicy::CriticalPathFirst`]): among the ready nodes, always order the
+/// one with the longest remaining critical path. Ties break on the smaller node
+/// id so the order stays deterministic. Tasks absent from the profile weigh
+/// zero.
+fn critical_path_first_order(graph: &CuGraph, profile: &PlanProfile) -> CuResult<Vec<NodeId>> {
+    let node_ids = graph.node_ids();
+    let duration = |id: NodeId| -> u64 {
+        graph
+            .get_node(id)
+            .map(|node| profile.task_duration_ns(node.get_id().as_str()))
+            .unwrap_or(0)
+    };
+    // Distinct neighbors: parallel edges between two nodes count once.
+    let consumers = |id: NodeId| -> BTreeSet<NodeId> {
+        graph
+            .get_neighbor_ids(id, CuDirection::Outgoing)
+            .into_iter()
+            .collect()
+    };
+    let producers = |id: NodeId| -> BTreeSet<NodeId> {
+        graph
+            .get_neighbor_ids(id, CuDirection::Incoming)
+            .into_iter()
+            .collect()
+    };
+
+    // Critical path per node: its duration plus the longest path through its
+    // consumers, computed sink-to-source. Nodes on a cycle are never reached;
+    // they keep weight zero, stay unready below, and the missing-node check
+    // in `compute_runtime_plan` reports them.
+    let mut critical_path: BTreeMap<NodeId, u64> = BTreeMap::new();
+    let mut pending_consumers: BTreeMap<NodeId, usize> = BTreeMap::new();
+    let mut queue: VecDeque<NodeId> = VecDeque::new();
+    for &id in &node_ids {
+        let count = consumers(id).len();
+        pending_consumers.insert(id, count);
+        if count == 0 {
+            queue.push_back(id);
+        }
+    }
+    while let Some(id) = queue.pop_front() {
+        let downstream = consumers(id)
+            .iter()
+            .map(|consumer| critical_path.get(consumer).copied().unwrap_or(0))
+            .max()
+            .unwrap_or(0);
+        critical_path.insert(id, duration(id) + downstream);
+        for producer in producers(id) {
+            let pending = pending_consumers.get_mut(&producer).unwrap();
+            *pending -= 1;
+            if *pending == 0 {
+                queue.push_back(producer);
+            }
+        }
+    }
+
+    // List scheduling: a max-heap on (critical path, smaller node id).
+    let mut pending_producers: BTreeMap<NodeId, usize> = BTreeMap::new();
+    let mut ready: BinaryHeap<(u64, Reverse<NodeId>)> = BinaryHeap::new();
+    for &id in &node_ids {
+        let count = producers(id).len();
+        pending_producers.insert(id, count);
+        if count == 0 {
+            ready.push((critical_path.get(&id).copied().unwrap_or(0), Reverse(id)));
+        }
+    }
+
+    let mut order = Vec::with_capacity(node_ids.len());
+    while let Some((_, Reverse(id))) = ready.pop() {
+        order.push(id);
+        for consumer in consumers(id) {
+            let pending = pending_producers.get_mut(&consumer).unwrap();
+            *pending -= 1;
+            if *pending == 0 {
+                ready.push((
+                    critical_path.get(&consumer).copied().unwrap_or(0),
+                    Reverse(consumer),
+                ));
+            }
         }
     }
 
+    Ok(order)
+}
+
+/// This is the main entry point to compute an execution plan at compilation
+/// time. The policy picks the step order, reading `profile` when it is a
+/// profile-guided one; the build phase is shared by every policy (see
+/// `doc/sched-v0.md`).
+pub fn compute_runtime_plan(
+    graph: &CuGraph,
+    policy: PlanPolicy,
+    profile: &PlanProfile,
+) -> CuResult<CuExecutionLoop> {
+    #[cfg(all(feature = "std", feature = "macro_debug"))]
+    eprintln!("[runtime plan: {policy:?}]");
+
+    if policy.needs_profile() && profile.is_empty() {
+        return Err(CuError::from(format!(
+            "The plan policy {policy:?} needs a measured profile, but runtime.plan_profile is \
+             empty. Record a log with the default policy, then run \
+             `cu29_export <log> schedule-profile` and paste its output as runtime.plan_profile."
+        )));
+    }
+
+    let order = match policy {
+        PlanPolicy::TopoBfs => topo_bfs_order(graph)?,
+        PlanPolicy::CriticalPathFirst => critical_path_first_order(graph, profile)?,
+    };
+
+    let plan = build_plan_from_order(graph, &order)?;
+
+    let planned_nodes: BTreeSet<NodeId> = order.iter().copied().collect();
     let mut missing = Vec::new();
     for node_id in graph.node_ids() {
         if !planned_nodes.contains(&node_id) {
@@ -1982,6 +2069,93 @@ pub fn compute_runtime_plan(graph: &CuGraph) -> CuResult<CuExecutionLoop> {
     })
 }
 
+/// Assigns every plan step to a slot of the `rt` pool's CPU affinity list.
+///
+/// Returns exactly `plan.steps.len()` slot indices, in plan order, each in
+/// `0..slots` — one per stage worker, whatever the placement. The caller hands
+/// that slot to `apply_current_thread_scheduling` in place of the step index,
+/// so the historical `index % slots` spread stays reachable.
+///
+/// [`CorePlacement::LongestFirst`] is longest-processing-time-first bin
+/// packing: walk the steps heaviest first and give each to the slot with the
+/// least accumulated load. Ties break on the slot holding fewer steps, then on
+/// the lower slot index; step ties break on the lower step index. The result
+/// is therefore deterministic, and with an all-zero profile it degenerates
+/// back to the plain spread.
+///
+/// This is the second consumer of [`PlanProfile`] and it is deliberately not a
+/// [`PlanPolicy`]: reordering steps cannot change a pipeline's slowest stage,
+/// so ordering and placement answer different questions.
+pub fn place_steps_on_cores(
+    plan: &CuExecutionLoop,
+    placement: CorePlacement,
+    profile: &PlanProfile,
+    slots: usize,
+) -> CuResult<Vec<usize>> {
+    let step_count = plan.steps.len();
+    if slots == 0 {
+        return Err(CuError::from(
+            "Core placement needs at least one CPU affinity slot.",
+        ));
+    }
+    if placement.needs_profile() && profile.is_empty() {
+        return Err(CuError::from(format!(
+            "The core placement {placement:?} needs a measured profile, but runtime.plan_profile \
+             is empty. Record a log with the default placement, then run \
+             `cu29_export <log> schedule-profile` and paste its output as runtime.plan_profile."
+        )));
+    }
+
+    if placement == CorePlacement::Spread {
+        return Ok((0..step_count).map(|step| step % slots).collect());
+    }
+
+    // Always one weight per plan step, so every placement returns the same
+    // number of slots as the pipeline spawns stage workers.
+    let durations = step_weights(plan, profile);
+    debug_assert_eq!(durations.len(), step_count);
+
+    // Heaviest first; equal weights keep plan order so the packing is stable.
+    let mut by_weight: Vec<usize> = (0..step_count).collect();
+    by_weight.sort_by_key(|&step| (core::cmp::Reverse(durations[step]), step));
+
+    let mut load = vec![0u128; slots];
+    let mut assigned = vec![0usize; slots];
+    let mut placement_of_step = vec![0usize; step_count];
+    for step in by_weight {
+        let slot = (0..slots)
+            .min_by_key(|&slot| (load[slot], assigned[slot], slot))
+            .expect("slots is non-zero");
+        load[slot] += u128::from(durations[step]);
+        assigned[slot] += 1;
+        placement_of_step[step] = slot;
+    }
+
+    Ok(placement_of_step)
+}
+
+/// One measured weight per top-level plan unit, in execution order.
+///
+/// The generated pipeline spawns one stage worker per top-level unit, so the
+/// result is always `plan.steps.len()` long: a unit's position is its worker
+/// index. A nested loop counts as the sum of the steps it runs.
+fn step_weights(plan: &CuExecutionLoop, profile: &PlanProfile) -> Vec<u64> {
+    fn weight(unit: &CuExecutionUnit, profile: &PlanProfile) -> u64 {
+        match unit {
+            CuExecutionUnit::Step(step) => profile.task_duration_ns(step.node.get_id().as_str()),
+            CuExecutionUnit::Loop(inner) => inner
+                .steps
+                .iter()
+                .map(|inner_unit| weight(inner_unit, profile))
+                .sum(),
+        }
+    }
+    plan.steps
+        .iter()
+        .map(|unit| weight(unit, profile))
+        .collect()
+}
+
 //tests
 #[cfg(test)]
 mod tests {
@@ -2601,7 +2775,8 @@ mod tests {
         assert_eq!(src1_edge_id, 1);
         assert_eq!(src2_edge_id, 0);
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let sink_step = runtime
             .steps
             .iter()
@@ -2617,6 +2792,245 @@ mod tests {
         assert_eq!(sink_step.input_msg_indices_types[1].msg_type, src1_type);
     }
 
+    #[test]
+    fn test_runtime_plan_default_policy_golden_order() {
+        let mut config = CuConfig::default();
+        let graph = config.get_graph_mut(None).unwrap();
+        let s1 = graph.add_node(Node::new("s1", "Source1")).unwrap();
+        let s2 = graph.add_node(Node::new("s2", "Source2")).unwrap();
+        let fusion = graph.add_node(Node::new("fusion", "Fusion")).unwrap();
+        let sink = graph.add_node(Node::new("sink", "Sink")).unwrap();
+
+        graph.connect(s1, fusion, "m1").unwrap();
+        graph.connect(s2, fusion, "m2").unwrap();
+        graph.connect(fusion, sink, "m3").unwrap();
+
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
+
+        let order_and_slots: Vec<(NodeId, u32)> = runtime
+            .steps
+            .iter()
+            .map(|unit| match unit {
+                CuExecutionUnit::Step(step) => (
+                    step.node_id,
+                    step.output_msg_pack.as_ref().unwrap().culist_index,
+                ),
+                CuExecutionUnit::Loop(_) => panic!("unexpected loop in a flat plan"),
+            })
+            .collect();
+
+        // Golden default-policy plan: the fusion waits for both sources, the
+        // copperlist slots follow the step order.
+        assert_eq!(
+            order_and_slots,
+            vec![(s1, 0), (s2, 1), (fusion, 2), (sink, 3)]
+        );
+    }
+
+    fn plan_node_order(plan: &CuExecutionLoop) -> Vec<NodeId> {
+        plan.steps
+            .iter()
+            .map(|unit| match unit {
+                CuExecutionUnit::Step(step) => step.node_id,
+                CuExecutionUnit::Loop(_) => panic!("unexpected loop in a flat plan"),
+            })
+            .collect()
+    }
+
+    #[test]
+    fn test_runtime_plan_critical_path_first_prioritizes_critical_path() {
+        let mut config = CuConfig::default();
+        let graph = config.get_graph_mut(None).unwrap();
+        let s1 = graph.add_node(Node::new("s1", "Source1")).unwrap();
+        let slow = graph.add_node(Node::new("slow", "Slow")).unwrap();
+        let sink1 = graph.add_node(Node::new("sink1", "Sink1")).unwrap();
+        let s2 = graph.add_node(Node::new("s2", "Source2")).unwrap();
+        let fast = graph.add_node(Node::new("fast", "Fast")).unwrap();
+        let sink2 = graph.add_node(Node::new("sink2", "Sink2")).unwrap();
+
+        graph.connect(s1, slow, "m1").unwrap();
+        graph.connect(slow, sink1, "m2").unwrap();
+        graph.connect(s2, fast, "m3").unwrap();
+        graph.connect(fast, sink2, "m4").unwrap();
+
+        // The default policy exhausts one source branch before the next.
+        let default_plan =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
+        assert_eq!(
+            plan_node_order(&default_plan),
+            vec![s1, slow, sink1, s2, fast, sink2]
+        );
+
+        let profile = PlanProfile {
+            task_duration_ns: [
+                ("s1".to_string(), 1),
+                ("slow".to_string(), 1000),
+                ("s2".to_string(), 1),
+                ("fast".to_string(), 10),
+            ]
+            .into_iter()
+            .collect(),
+        };
+
+        // The critical-path-first policy runs the long chain first; the equal-weight
+        // sinks fall back to node-id order.
+        let cpf_plan =
+            compute_runtime_plan(graph, PlanPolicy::CriticalPathFirst, &profile).unwrap();
+        assert_eq!(
+            plan_node_order(&cpf_plan),
+            vec![s1, slow, s2, fast, sink1, sink2]
+        );
+    }
+
+    #[test]
+    fn test_runtime_plan_critical_path_first_partial_profile_is_deterministic() {
+        let mut config = CuConfig::default();
+        let graph = config.get_graph_mut(None).unwrap();
+        let s1 = graph.add_node(Node::new("s1", "Source1")).unwrap();
+        let s2 = graph.add_node(Node::new("s2", "Source2")).unwrap();
+        let fusion = graph.add_node(Node::new("fusion", "Fusion")).unwrap();
+        let sink = graph.add_node(Node::new("sink", "Sink")).unwrap();
+
+        graph.connect(s1, fusion, "m1").unwrap();
+        graph.connect(s2, fusion, "m2").unwrap();
+        graph.connect(fusion, sink, "m3").unwrap();
+
+        // Only `sink` was sampled, so both sources weigh zero and share the
+        // same critical path: ties resolve on node id and the order is stable.
+        let profile = PlanProfile {
+            task_duration_ns: [("sink".to_string(), 5)].into_iter().collect(),
+        };
+        let plan = compute_runtime_plan(graph, PlanPolicy::CriticalPathFirst, &profile).unwrap();
+        assert_eq!(plan_node_order(&plan), vec![s1, s2, fusion, sink]);
+    }
+
+    #[test]
+    fn test_runtime_plan_critical_path_first_rejects_empty_profile() {
+        let mut config = CuConfig::default();
+        let graph = config.get_graph_mut(None).unwrap();
+        let src = graph.add_node(Node::new("src", "Source")).unwrap();
+        let sink = graph.add_node(Node::new("sink", "Sink")).unwrap();
+        graph.connect(src, sink, "m1").unwrap();
+
+        // A profile-guided policy without a profile is a config mistake, not a
+        // silent fallback to another order.
+        let err = compute_runtime_plan(
+            graph,
+            PlanPolicy::CriticalPathFirst,
+            &PlanProfile::default(),
+        )
+        .expect_err("empty profile should be rejected");
+        assert!(err.to_string().contains("schedule-profile"));
+    }
+
+    /// A chain of `weights.len()` steps, so the plan order is the given order.
+    fn chain_plan(weights: &[(&str, u64)]) -> (CuConfig, PlanProfile) {
+        let mut config = CuConfig::default();
+        let graph = config.get_graph_mut(None).unwrap();
+        let mut previous = None;
+        for (index, (id, _)) in weights.iter().enumerate() {
+            let node = graph.add_node(Node::new(id, "T")).unwrap();
+            if let Some(previous) = previous {
+                graph.connect(previous, node, &format!("m{index}")).unwrap();
+            }
+            previous = Some(node);
+        }
+        let profile = PlanProfile {
+            task_duration_ns: weights
+                .iter()
+                .map(|(id, ns)| (id.to_string(), *ns))
+                .collect(),
+        };
+        (config, profile)
+    }
+
+    #[test]
+    fn test_core_placement_spread_is_round_robin() {
+        let (mut config, profile) = chain_plan(&[("a", 1), ("b", 2), ("c", 3), ("d", 4)]);
+        let graph = config.get_graph_mut(None).unwrap();
+        let plan =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
+
+        // Spread ignores the profile entirely: this is the historical mapping.
+        let slots =
+            place_steps_on_cores(&plan, CorePlacement::Spread, &PlanProfile::default(), 3).unwrap();
+        assert_eq!(slots, vec![0, 1, 2, 0]);
+        let with_profile = place_steps_on_cores(&plan, CorePlacement::Spread, &profile, 3).unwrap();
+        assert_eq!(with_profile, slots);
+    }
+
+    #[test]
+    fn test_core_placement_longest_first_balances_load() {
+        // 8+1 and 5+4 both make 9, against 9+1+5+4 = 19 spread as 9+5, 1, 4.
+        let (mut config, profile) = chain_plan(&[("a", 9), ("b", 1), ("c", 5), ("d", 4)]);
+        let graph = config.get_graph_mut(None).unwrap();
+        let plan =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
+
+        let slots = place_steps_on_cores(&plan, CorePlacement::LongestFirst, &profile, 2).unwrap();
+        assert_eq!(slots, vec![0, 0, 1, 1]);
+
+        let per_slot =
+            slots
+                .iter()
+                .zip([9u64, 1, 5, 4])
+                .fold(vec![0u64; 2], |mut load, (&slot, weight)| {
+                    load[slot] += weight;
+                    load
+                });
+        assert_eq!(per_slot, vec![10, 9]);
+    }
+
+    #[test]
+    fn test_core_placement_longest_first_is_deterministic_on_ties() {
+        // All-equal weights must not pile onto one core; ties fall back to the
+        // plain spread.
+        let (mut config, profile) = chain_plan(&[("a", 7), ("b", 7), ("c", 7), ("d", 7)]);
+        let graph = config.get_graph_mut(None).unwrap();
+        let plan =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
+
+        let slots = place_steps_on_cores(&plan, CorePlacement::LongestFirst, &profile, 2).unwrap();
+        assert_eq!(slots, vec![0, 1, 0, 1]);
+    }
+
+    #[test]
+    fn test_core_placement_longest_first_ignores_unmeasured_steps() {
+        // 'b' was never sampled: it weighs zero and lands wherever there is room.
+        let (mut config, _) = chain_plan(&[("a", 0), ("b", 0), ("c", 0)]);
+        let graph = config.get_graph_mut(None).unwrap();
+        let plan =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
+        let partial = PlanProfile {
+            task_duration_ns: [("a".to_string(), 100), ("c".to_string(), 10)]
+                .into_iter()
+                .collect(),
+        };
+
+        let slots = place_steps_on_cores(&plan, CorePlacement::LongestFirst, &partial, 2).unwrap();
+        assert_eq!(slots, vec![0, 1, 1]);
+    }
+
+    #[test]
+    fn test_core_placement_rejects_empty_profile_and_zero_slots() {
+        let (mut config, profile) = chain_plan(&[("a", 1), ("b", 2)]);
+        let graph = config.get_graph_mut(None).unwrap();
+        let plan =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
+
+        let err = place_steps_on_cores(
+            &plan,
+            CorePlacement::LongestFirst,
+            &PlanProfile::default(),
+            2,
+        )
+        .expect_err("a profile-guided placement needs a profile");
+        assert!(err.to_string().contains("schedule-profile"));
+
+        assert!(place_steps_on_cores(&plan, CorePlacement::LongestFirst, &profile, 0).is_err());
+    }
+
     #[test]
     fn test_runtime_output_ports_unique_ordered() {
         let mut config = CuConfig::default();
@@ -2632,7 +3046,8 @@ mod tests {
         graph.connect(src_id, dst_a2_id, "msg::A").unwrap();
         graph.connect(src_id, dst_c_id, "msg::C").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let src_step = runtime
             .steps
             .iter()
@@ -2695,7 +3110,8 @@ mod tests {
         graph.connect(src_id, dst_a_id, "i32").unwrap();
         graph.connect(src_id, dst_b_id, "i32").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let src_step = runtime
             .steps
             .iter()
@@ -2721,7 +3137,8 @@ mod tests {
             .expect("missing source node")
             .add_nc_output("msg::B", usize::MAX);
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let src_step = runtime
             .steps
             .iter()
@@ -2760,7 +3177,8 @@ mod tests {
         let graph = config.get_graph(None).unwrap();
         let regular_id = graph.get_node_id_by_name("regular").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let regular_step = runtime
             .steps
             .iter()
@@ -2791,7 +3209,8 @@ mod tests {
         let src_id = graph.get_node_id_by_name("src").unwrap();
         let dst_id = graph.get_node_id_by_name("sink").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let src_step = runtime
             .steps
             .iter()
@@ -2831,7 +3250,8 @@ mod tests {
         let src_id = graph.get_node_id_by_name("src").unwrap();
         let dst_id = graph.get_node_id_by_name("sink").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let src_step = runtime
             .steps
             .iter()
@@ -2868,7 +3288,8 @@ mod tests {
         let src_id = graph.get_node_id_by_name("src").unwrap();
         let dst_id = graph.get_node_id_by_name("sink").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let src_step = runtime
             .steps
             .iter()
@@ -2917,7 +3338,8 @@ mod tests {
         assert_eq!(edge_cam0_to_inf0, 0);
         assert_eq!(edge_cam0_to_broadcast, 1);
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let broadcast_step = runtime
             .steps
             .iter()
@@ -2957,7 +3379,8 @@ mod tests {
         assert_eq!(edge_cam0_to_broadcast, 0);
         assert_eq!(edge_cam0_to_inf0, 1);
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let broadcast_step = runtime
             .steps
             .iter()
diff --git a/core/cu29_runtime/src/rendercfg.rs b/core/cu29_runtime/src/rendercfg.rs
index 80d43d1da9d..725ef5505fc 100644
--- a/core/cu29_runtime/src/rendercfg.rs
+++ b/core/cu29_runtime/src/rendercfg.rs
@@ -1,7 +1,8 @@
 mod config;
 use clap::Parser;
 use config::{
-    ConfigGraphs, PortLookup, build_render_topology, read_configuration, read_multi_configuration,
+    ConfigGraphs, LOGSTATS_SCHEMA_VERSION, PortLookup, build_render_topology, read_configuration,
+    read_multi_configuration,
 };
 pub use cu29_traits::*;
 use hashbrown::HashMap;
@@ -46,7 +47,6 @@ const MODULE_TRUNC_MARKER: &str = "…";
 const MODULE_SEPARATOR: &str = "⠶";
 const PLACEHOLDER_TEXT: &str = "\u{2014}";
 const COPPER_LOGO_SVG: &str = include_str!("../assets/cu29.svg");
-const LOGSTATS_SCHEMA_VERSION: u32 = 1;
 
 // Color palette and fills.
 const BORDER_COLOR: &str = "#999999";
@@ -187,7 +187,7 @@ struct Args {
 }
 
 enum RenderInput {
-    Single(config::CuConfig),
+    Single(Box<config::CuConfig>),
     Multi(config::MultiCopperConfig),
 }
 
@@ -303,7 +303,7 @@ fn load_render_input(path: &Path) -> CuResult<RenderInput> {
     match read_multi_configuration(path_str) {
         Ok(config) => Ok(RenderInput::Multi(config)),
         Err(multi_err) => match read_configuration(path_str) {
-            Ok(config) => Ok(RenderInput::Single(config)),
+            Ok(config) => Ok(RenderInput::Single(Box::new(config))),
             Err(single_err) => Err(CuError::from(format!(
                 "Failed to read '{}' as either a Copper config or a multi-Copper config.\nCopper config: {single_err}\nMulti-Copper config: {multi_err}",
                 path.display()
diff --git a/core/cu29_runtime/src/thread_pool.rs b/core/cu29_runtime/src/thread_pool.rs
index 6a0b0db06f8..021a2828037 100644
--- a/core/cu29_runtime/src/thread_pool.rs
+++ b/core/cu29_runtime/src/thread_pool.rs
@@ -98,10 +98,14 @@ fn apply_scheduling(_pool: &ThreadPool, spec: &ThreadPoolConfig) -> CuResult<()>
 }
 
 /// Applies a pool's CPU affinity and scheduling policy to the **current** thread,
-/// as worker `index` (Spread: pinned to `affinity[index % affinity.len()]`).
+/// as worker `index` (pinned to `affinity[index % affinity.len()]`).
 ///
 /// This is for worker threads that are not part of a rayon pool — notably the
 /// `parallel-rt` stage workers, which are plain `std::thread::scope` threads.
+/// Those pass the affinity slot that
+/// [`CorePlacement`](crate::config::CorePlacement) picked for their stage
+/// rather than their own stage index, so a balanced placement reaches this
+/// function already resolved and the modulo is a no-op.
 /// Behavior mirrors [`build_pool`]: [`OnError::Warn`](crate::config::OnError::Warn)
 /// logs and returns `Ok`, [`OnError::Strict`](crate::config::OnError::Strict)
 /// returns `Err`. When the `rt-scheduling` feature is off the request is ignored
diff --git a/core/cu29_runtime/tests/anytime_generated.rs b/core/cu29_runtime/tests/anytime_generated.rs
new file mode 100644
index 00000000000..356b11f59c7
--- /dev/null
+++ b/core/cu29_runtime/tests/anytime_generated.rs
@@ -0,0 +1,160 @@
+#![cfg(all(test, feature = "std"))]
+
+use cu29::cutask_anytime::{AnytimeStatus, CuAnytimeTask};
+use cu29::prelude::copper_runtime;
+use cu29::prelude::*;
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
+
+static BASE_CALLS: AtomicUsize = AtomicUsize::new(0);
+static REFINE_CALLS: AtomicUsize = AtomicUsize::new(0);
+static SINK_CALLS: AtomicUsize = AtomicUsize::new(0);
+static FIRST_PAYLOAD: AtomicU32 = AtomicU32::new(u32::MAX);
+static FIRST_STOPPED_AT_MAX: AtomicBool = AtomicBool::new(false);
+static SECOND_HAS_PAYLOAD: AtomicBool = AtomicBool::new(true);
+static SECOND_SKIPPED_STALE: AtomicBool = AtomicBool::new(false);
+
+#[derive(Reflect)]
+struct RangeSource {
+    iteration: usize,
+}
+
+impl Freezable for RangeSource {}
+
+impl CuSrcTask for RangeSource {
+    type Resources<'r> = ();
+    type Output<'m> = output_msg!(u32);
+
+    fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self> {
+        Ok(Self { iteration: 0 })
+    }
+
+    fn process(&mut self, ctx: &CuContext, output: &mut Self::Output<'_>) -> CuResult<()> {
+        let now = ctx.clock.now();
+        let start = if self.iteration == 0 {
+            now - CuDuration::from_millis(20)
+        } else {
+            CuTime::default()
+        };
+        self.iteration += 1;
+
+        output.set_payload(1);
+        output.tov = Tov::Range(CuTimeRange { start, end: now });
+        Ok(())
+    }
+}
+
+#[derive(Reflect)]
+struct Refiner;
+
+impl Freezable for Refiner {}
+
+impl CuAnytimeTask for Refiner {
+    type Input<'m> = input_msg!(u32);
+    type Output<'m> = output_msg!(u32);
+    type Resources<'r> = ();
+    type Quality = ();
+
+    fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self> {
+        Ok(Self)
+    }
+
+    fn base(
+        &mut self,
+        _ctx: &CuContext,
+        input: &Self::Input<'_>,
+        output: &mut Self::Output<'_>,
+    ) -> CuResult<AnytimeStatus<()>> {
+        BASE_CALLS.fetch_add(1, Ordering::SeqCst);
+        assert_eq!(input.payload(), Some(&1));
+        output.set_payload(0);
+        Ok(AnytimeStatus::Improved(()))
+    }
+
+    fn refine(
+        &mut self,
+        _ctx: &CuContext,
+        output: &mut Self::Output<'_>,
+    ) -> CuResult<AnytimeStatus<()>> {
+        REFINE_CALLS.fetch_add(1, Ordering::SeqCst);
+        let next = output.payload().copied().unwrap_or_default() + 1;
+        output.set_payload(next);
+        Ok(AnytimeStatus::Improved(()))
+    }
+}
+
+#[derive(Reflect)]
+struct RecordingSink;
+
+impl Freezable for RecordingSink {}
+
+impl CuSinkTask for RecordingSink {
+    type Resources<'r> = ();
+    type Input<'m> = input_msg!(u32);
+
+    fn new(_config: Option<&ComponentConfig>, _resources: Self::Resources<'_>) -> CuResult<Self> {
+        Ok(Self)
+    }
+
+    fn process(&mut self, _ctx: &CuContext, input: &Self::Input<'_>) -> CuResult<()> {
+        let call = SINK_CALLS.fetch_add(1, Ordering::SeqCst);
+        match call {
+            0 => {
+                FIRST_PAYLOAD.store(
+                    input.payload().copied().unwrap_or(u32::MAX),
+                    Ordering::SeqCst,
+                );
+                FIRST_STOPPED_AT_MAX.store(
+                    input.metadata.status_txt.0.as_str() == "any:3it max",
+                    Ordering::SeqCst,
+                );
+            }
+            1 => {
+                SECOND_HAS_PAYLOAD.store(input.payload().is_some(), Ordering::SeqCst);
+                SECOND_SKIPPED_STALE.store(
+                    input.metadata.status_txt.0.as_str() == "any:0it stale!",
+                    Ordering::SeqCst,
+                );
+            }
+            _ => panic!("unexpected sink invocation {call}"),
+        }
+        Ok(())
+    }
+}
+
+#[copper_runtime(config = "tests/anytime_generated_config.ron")]
+struct AnytimeGeneratedApp {}
+
+#[test]
+fn generated_anytime_runtime_refines_and_uses_earliest_range_tov() -> CuResult<()> {
+    BASE_CALLS.store(0, Ordering::SeqCst);
+    REFINE_CALLS.store(0, Ordering::SeqCst);
+    SINK_CALLS.store(0, Ordering::SeqCst);
+    FIRST_PAYLOAD.store(u32::MAX, Ordering::SeqCst);
+    FIRST_STOPPED_AT_MAX.store(false, Ordering::SeqCst);
+    SECOND_HAS_PAYLOAD.store(true, Ordering::SeqCst);
+    SECOND_SKIPPED_STALE.store(false, Ordering::SeqCst);
+
+    let (clock, clock_mock) = RobotClock::mock();
+    let mut app = AnytimeGeneratedApp::builder().with_clock(clock).build()?;
+
+    app.start_all_tasks()?;
+
+    clock_mock.set_value(CuDuration::from_millis(40).as_nanos());
+    app.run_one_iteration()?;
+
+    // The range's end is current, but its earliest Tov is 100 ms old and
+    // therefore exceeds max_age_ms.
+    clock_mock.set_value(CuDuration::from_millis(100).as_nanos());
+    app.run_one_iteration()?;
+
+    app.stop_all_tasks()?;
+
+    assert_eq!(BASE_CALLS.load(Ordering::SeqCst), 1);
+    assert_eq!(REFINE_CALLS.load(Ordering::SeqCst), 3);
+    assert_eq!(SINK_CALLS.load(Ordering::SeqCst), 2);
+    assert_eq!(FIRST_PAYLOAD.load(Ordering::SeqCst), 3);
+    assert!(FIRST_STOPPED_AT_MAX.load(Ordering::SeqCst));
+    assert!(!SECOND_HAS_PAYLOAD.load(Ordering::SeqCst));
+    assert!(SECOND_SKIPPED_STALE.load(Ordering::SeqCst));
+    Ok(())
+}
diff --git a/core/cu29_runtime/tests/anytime_generated_config.ron b/core/cu29_runtime/tests/anytime_generated_config.ron
new file mode 100644
index 00000000000..96b425b4fd5
--- /dev/null
+++ b/core/cu29_runtime/tests/anytime_generated_config.ron
@@ -0,0 +1,32 @@
+(
+    tasks: [
+        (
+            id: "range_source",
+            type: "RangeSource",
+        ),
+        (
+            id: "refiner",
+            type: "Refiner",
+            anytime: (
+                max_age_ms: 50.0,
+                max_refines: 3,
+            ),
+        ),
+        (
+            id: "sink",
+            type: "RecordingSink",
+        ),
+    ],
+    cnx: [
+        (
+            src: "range_source",
+            dst: "refiner",
+            msg: "u32",
+        ),
+        (
+            src: "refiner",
+            dst: "sink",
+            msg: "u32",
+        ),
+    ],
+)
\ No newline at end of file
diff --git a/core/cu29_runtime/tests/loopback.rs b/core/cu29_runtime/tests/loopback.rs
index c7447e32941..24137098a02 100644
--- a/core/cu29_runtime/tests/loopback.rs
+++ b/core/cu29_runtime/tests/loopback.rs
@@ -1,6 +1,6 @@
 #[cfg(all(test, feature = "std"))]
 mod tests {
-    use cu29_runtime::config::read_configuration;
+    use cu29_runtime::config::{PlanPolicy, PlanProfile, read_configuration};
     use cu29_runtime::curuntime::compute_runtime_plan;
     use std::path::PathBuf;
 
@@ -12,7 +12,8 @@ mod tests {
         let config =
             read_configuration(config_path.to_str().unwrap()).expect("config should parse");
         let graph = config.get_graph(None).expect("graph should load");
-        let err = compute_runtime_plan(graph).expect_err("loopback should fail");
+        let err = compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default())
+            .expect_err("loopback should fail");
         let msg = err.to_string();
         assert!(msg.contains("loopback"), "unexpected error: {msg}");
         assert!(msg.contains("Missing"), "unexpected error: {msg}");
diff --git a/core/cu29_runtime/tests/missions.rs b/core/cu29_runtime/tests/missions.rs
index b6369299e8f..7502a56dc65 100644
--- a/core/cu29_runtime/tests/missions.rs
+++ b/core/cu29_runtime/tests/missions.rs
@@ -1,6 +1,6 @@
 #[cfg(all(test, feature = "std"))]
 mod tests {
-    use cu29_runtime::config::read_configuration;
+    use cu29_runtime::config::{PlanPolicy, PlanProfile, read_configuration};
     use cu29_runtime::curuntime::{CuExecutionUnit, compute_runtime_plan};
     use std::fs::{create_dir_all, write};
     use tempfile::tempdir;
@@ -51,8 +51,12 @@ mod tests {
         let mission_a = config.get_graph(Some("A")).expect("mission A graph");
         let mission_b = config.get_graph(Some("B")).expect("mission B graph");
 
-        let runtime_a = compute_runtime_plan(mission_a).expect("mission A runtime plan");
-        let runtime_b = compute_runtime_plan(mission_b).expect("mission B runtime plan");
+        let runtime_a =
+            compute_runtime_plan(mission_a, PlanPolicy::default(), &PlanProfile::default())
+                .expect("mission A runtime plan");
+        let runtime_b =
+            compute_runtime_plan(mission_b, PlanPolicy::default(), &PlanProfile::default())
+                .expect("mission B runtime plan");
 
         let sink_a = runtime_a
             .steps
diff --git a/doc/sched-v0.md b/doc/sched-v0.md
new file mode 100644
index 00000000000..dea02df6ed6
--- /dev/null
+++ b/doc/sched-v0.md
@@ -0,0 +1,192 @@
+# Pluggable execution-plan scheduling — design (v0)
+
+`compute_runtime_plan()` turns the task graph into the fixed step sequence the
+`#[copper_runtime]` macro bakes into the generated loop. Today the ordering
+heuristic is hard-coded (a BFS from the sources) and carries a
+`TODO(gbin): Make that heuristic pluggable`. This design makes the heuristic a
+config-selected policy, and lays the groundwork for a profile-guided policy.
+
+## The core invariant
+
+Any topological order of the (per-mission, bridge-expanded) task graph is a
+*correct* plan: a step only needs every producer of its inputs to run earlier
+in the same iteration. Everything else — copperlist slot assignment, input
+wiring, validation — is mechanical and identical for every order.
+
+So a scheduling policy answers exactly one question: **which topological order
+do we emit?** That keeps every policy small and safe: a policy cannot corrupt
+message wiring, it can only pick a better or worse order.
+
+## The split
+
+`compute_runtime_plan(graph, policy)` becomes two phases:
+
+| phase | function | role |
+|---|---|---|
+| order | `topo_bfs_order(graph)` (one per policy) | pick the step order: `Vec<NodeId>` |
+| build | `build_plan_from_order(graph, &order)` | assign copperlist slots in order, wire inputs, validate |
+
+The build phase rejects an order where an input's producer does not appear
+earlier — a buggy policy fails the build with a clear error instead of
+generating a broken runtime.
+
+## Config surface
+
+Two independent fields in `config.rs`: *which algorithm* and *what it
+measured*. A variant names an algorithm and nothing else; measurement is a
+separate struct, because the same numbers feed more than one consumer.
+
+```rust
+#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PlanPolicy {
+    /// The historical source-BFS order. The default. Ignores the profile.
+    #[default]
+    TopoBfs,
+    /// Longest-remaining-critical-path first. Needs a profile.
+    CriticalPathFirst,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Default)]
+pub struct PlanProfile {
+    pub task_duration_ns: BTreeMap<String, u64>,
+}
+```
+
+RON:
+
+```ron
+runtime: (
+    plan_policy: TopoBfs,
+    plan_profile: (task_duration_ns: {"cam": 1200, "detect": 8400}),
+),
+```
+
+**Why the profile is not a variant field.** Folding the measurement into
+`CriticalPathFirst(task_duration_ns: ...)` names the *input* where a variant
+should name the *algorithm*, and it forces every future profile-guided
+algorithm to repeat the same field. Keeping them apart means adding
+`LongestTaskFirst` or `MinSlack` costs one unit variant, switching policy
+needs no re-measurement, and `parallel-rt` placement can read the same
+`PlanProfile` without going through a policy at all.
+
+Both fields are optional. `plan_policy` defaults to `TopoBfs`, whose output is
+byte-identical to the historical planner (same order, same copperlist indices).
+A policy with `needs_profile()` and an empty `plan_profile` fails the build with
+a message pointing at `schedule-profile` — a config mistake, not a silent
+fallback to another order. The same holds for `core_placement`: a non-default
+placement with no `rt` affinity list to place onto fails the build.
+
+**Why the RON config and not a macro attribute:** the unified log embeds the
+config. Offline tools (`cu29_export` logstats) recompute the plan from that
+embedded config to map copperlist slots back to tasks. If the policy lived
+outside the config, an offline reader could reconstruct the wrong slot layout.
+Policy-in-config keeps one source of truth for everything that derives from
+the plan.
+
+## Profile-guided ordering (v1)
+
+This is a chicken-egg problem only in the way classic PGO is: the first build
+cannot have a profile. The loop that resolves it:
+
+```
+build (TopoBfs) → run robot or resim → export profile
+      → paste plan_profile into RON, set plan_policy → rebuild
+      → compare logstats → repeat
+```
+
+- **No new instrumentation.** Every `CuMsg` already records the
+  before/after `process()` window in its metadata; the unified log has the
+  per-task durations for every recorded cycle.
+- **Exporter.** `cu29_export <log> schedule-profile [--config copperconfig.ron]
+  [--mission M] [--stat mean|p99|max]` reads the log and writes
+  `schedule_profile.ron`: the exact RON value of the config's
+  `runtime.plan_profile` field. It writes a profile, never a policy — picking
+  the algorithm stays the user's decision.
+- **Policy.** `CriticalPathFirst` is critical-path-first list scheduling:
+  among the ready nodes, always order the one with the longest remaining
+  critical path, weighing each node with `plan_profile.task_duration_ns`. Ties
+  break on the smaller node id, so the order is deterministic. Tasks absent
+  from the map (including generated bridge channel nodes) weigh zero, which
+  keeps a partial profile usable.
+- **Durations live inline in the config**, not in a separate build-time file.
+
+Inlining the profile is what keeps offline readers exact: the unified log
+embeds the config, so logstats recomputes the same plan from the same data.
+The pasted snippet is committed with the config, so CI reproduces the build.
+
+What a better order can and cannot buy on the single-threaded runtime: total
+work per cycle is fixed; the order only moves *latency* — it shortens the
+sensor→actuator path of the chains it favors and reduces input staleness.
+Later consumers of the same profile data are worth more: placing anytime
+refine quanta into measured gaps (once anytime tasks land), and core packing
+for `parallel-rt`.
+
+## What the profile buys `parallel-rt` (v2)
+
+`parallel-rt` is a stage-affine pipeline: one worker thread per plan step
+(`cu29_derive/src/lib.rs`), each pinned to `cores[stage_index % cores.len()]`
+(`thread_pool.rs`). Two consequences decide what the profile is worth there:
+
+- Pipeline throughput is the duration of the **slowest single step**, not the
+  sum. Reordering steps cannot change a maximum, so `CriticalPathFirst` buys
+  `parallel-rt` essentially nothing. Ordering is a latency tool; parallel-rt
+  needs a *balance* tool.
+- Worker count equals step count, independent of core count. A 20-step graph
+  on 4 cores spawns 20 threads.
+
+Three uses of the same `PlanProfile`, cheapest first. None of them belongs in
+`PlanPolicy` — they are placement, not ordering.
+
+1. **Report the ceiling (done).** `cu29_export <log> log-stats` now emits a
+   `pipeline` section: per-step duration stats, `serial_cycle_ns` (the serial
+   engine's cycle), `bottleneck` (the slowest step, i.e. the `parallel-rt`
+   cycle), and `max_pipeline_speedup = serial_cycle_ns / bottleneck.mean_ns`.
+   The CLI prints the bottleneck line to stdout. This is diagnosis, not
+   scheduling: it says whether steps 2 and 3 are worth doing at all. A graph
+   whose `max_pipeline_speedup` is 1.2x will not repay a pipelining engine.
+2. **Profile-driven core packing (done).** `runtime.core_placement` selects
+   how stage workers map onto the `rt` pool's `affinity` list:
+   `Spread` (default, `stage % cores`) or `LongestFirst`, an LPT bin-pack over
+   `task_duration_ns` — walk the steps heaviest first, give each to the least
+   loaded core. `place_steps_on_cores` computes it at compile time, since both
+   inputs are in the config, and the generated worker passes the resolved slot
+   to `apply_current_thread_scheduling` instead of its stage index. Ties break
+   on the emptier core then the lower index, so an all-zero profile degenerates
+   back to `Spread` exactly. Steps the profile never measured weigh zero, as
+   with `CriticalPathFirst`.
+3. **Stage fusion.** Merge cheap adjacent steps into one worker until the step
+   count is near the core count: fewer queue hops, lower latency, less
+   oversubscription. This breaks the stage-index = plan-index identity in
+   `build_parallel_rt_stage_entries`, so it is the largest of the three.
+
+Note the histogram behind `CuDurationStatistics` is 1024 linear buckets over
+its configured max, so its `percentile()` is useless at microsecond scale. The
+`pipeline` section therefore reports only exact stats (min/max/mean/stddev);
+an exact percentile comes from `schedule-profile --stat p99`, which keeps the
+raw samples.
+
+## Caveats
+
+- A different order changes the copperlist slot layout, hence the generated
+  types. Logs recorded under one plan do not resim under another. The unified
+  log embeds the config the binary ran, so `log-stats` and `schedule-profile`
+  compare it against the `--config` file they plan with and warn when the two
+  disagree. Without that check, editing the config before re-exporting would
+  silently misattribute every step.
+- `logstats` plans over the config graph, while the runtime plans over the graph
+  the derive expands with one node per bridge channel. On a config with bridges
+  the `pipeline` step indices therefore do not line up with the `parallel-rt`
+  worker indices. Naming stays correct; only the numbering shifts.
+- A profile change requires a rebuild. Inherent to compile-time planning; the
+  determinism and zero-alloc properties of the generated loop depend on it.
+
+## Status
+
+- v0 (done): the order/build split, the `PlanPolicy` config surface, golden
+  tests pinning the default order.
+- v1 (done): the `CriticalPathFirst` policy and the `schedule-profile` exporter.
+- v2 (partial): the `pipeline` section of logstats reports the bottleneck and
+  the ceiling; `core_placement: LongestFirst` packs stage workers onto cores by
+  measured load. Stage fusion for `parallel-rt` is still open.
+- Later: profile-driven placement of anytime refine quanta, once anytime tasks
+  land.
diff --git a/examples/cu_baremetal_safety/src/lib.rs b/examples/cu_baremetal_safety/src/lib.rs
index c9f8158df03..15fe2fd33cb 100644
--- a/examples/cu_baremetal_safety/src/lib.rs
+++ b/examples/cu_baremetal_safety/src/lib.rs
@@ -3,7 +3,9 @@ pub use cu29::serde;
 
 #[cfg(feature = "safety-ids")]
 pub mod harness {
-    use cu29::config::{CuConfig, Node, TaskKind, resolve_task_kind_for_id};
+    use cu29::config::{
+        CuConfig, Node, PlanPolicy, PlanProfile, TaskKind, resolve_task_kind_for_id,
+    };
     use cu29::curuntime::{CuExecutionLoop, CuExecutionUnit, CuTaskType, compute_runtime_plan};
     use cu29::prelude::*;
     use std::path::PathBuf;
@@ -699,7 +701,8 @@ pub mod harness {
         graph.connect(src2_id, sink_id, "src2_type").unwrap();
         graph.connect(src1_id, sink_id, "src1_type").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let sink_step = step_for(&runtime, sink_id);
         let sink_inputs: Vec<String> = sink_step
             .input_msg_indices_types
@@ -729,7 +732,8 @@ pub mod harness {
         graph.connect(cam0_id, inf0_id, "i32").unwrap();
         graph.connect(inf0_id, broadcast_id, "f32").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let broadcast_step = step_for(&runtime, broadcast_id);
         let case1_inputs: Vec<String> = broadcast_step
             .input_msg_indices_types
@@ -759,7 +763,8 @@ pub mod harness {
         graph.connect(cam0_id, broadcast_id, "i32").unwrap();
         graph.connect(inf0_id, broadcast_id, "f32").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let broadcast_step = step_for(&runtime, broadcast_id);
         let case2_inputs: Vec<String> = broadcast_step
             .input_msg_indices_types
@@ -821,7 +826,8 @@ pub mod harness {
         graph.connect(src_id, dst_a, "i32").unwrap();
         graph.connect(src_id, dst_b, "i32").unwrap();
 
-        let runtime = compute_runtime_plan(graph).unwrap();
+        let runtime =
+            compute_runtime_plan(graph, PlanPolicy::default(), &PlanProfile::default()).unwrap();
         let src_step = step_for(&runtime, src_id);
 
         safety_check_eq!(
@@ -844,7 +850,12 @@ pub mod harness {
         let ordered_graph = ordered.get_graph(None).unwrap();
         let ordered_src = ordered_graph.get_node_id_by_name("src").unwrap();
         let ordered_sink = ordered_graph.get_node_id_by_name("sink").unwrap();
-        let runtime = compute_runtime_plan(ordered_graph).unwrap();
+        let runtime = compute_runtime_plan(
+            ordered_graph,
+            PlanPolicy::default(),
+            &PlanProfile::default(),
+        )
+        .unwrap();
         let src_step = step_for(&runtime, ordered_src);
         let sink_step = step_for(&runtime, ordered_sink);
 
@@ -873,7 +884,12 @@ pub mod harness {
         .unwrap();
         let inferred_graph = inferred.get_graph(None).unwrap();
         let regular_id = inferred_graph.get_node_id_by_name("regular").unwrap();
-        let runtime = compute_runtime_plan(inferred_graph).unwrap();
+        let runtime = compute_runtime_plan(
+            inferred_graph,
+            PlanPolicy::default(),
+            &PlanProfile::default(),
+        )
+        .unwrap();
         let regular_step = step_for(&runtime, regular_id);
 
         safety_check_eq!(