test pimcomp adversarial memory scheduling

backend-simulators/pim/pim-simulator/src/bin/pim-simulator/main.rs

@@ -11,6 +11,7 @@ use pimcore::cpu::crossbar::Crossbar;
 use pimcore::json_to_instruction::json_to_executor;
 use pimcore::memory_manager::CoreMemory;
 use pimcore::tracing::TRACER;
+use pimcore::{DiagnosticSchedulePolicy, DiagnosticScheduleTarget};
 use serde_json::Value;
 use std::collections::HashMap;
 use std::fs::{self, File};
@@ -60,6 +61,38 @@ struct Args {
     /// Optional directory for per-iteration output dumps
     #[arg(long)]
     batch_output_dir: Option<PathBuf>,
+
+    /// Optional JSONL shadow provenance trace
+    #[arg(long)]
+    provenance_trace: Option<PathBuf>,
+
+    /// Diagnostic-only barrier between global throughput iterations
+    #[arg(long)]
+    provenance_global_barrier: bool,
+
+    /// Diagnostic-only ready-core delay, formatted as CORE:CYCLES
+    #[arg(long, value_name = "CORE:CYCLES")]
+    provenance_core_stall: Option<String>,
+
+    /// Diagnostic scheduler policy; greedy is the unchanged default
+    #[arg(long, value_enum, default_value_t = DiagnosticSchedulePolicyArg::Greedy)]
+    diagnostic_schedule_policy: DiagnosticSchedulePolicyArg,
+
+    /// Deterministic seed for the randomized diagnostic scheduler
+    #[arg(long, default_value_t = 0)]
+    diagnostic_schedule_seed: u64,
+
+    /// Adversarial target: WCORE:WPC:RCORE:RPC:BEGIN:END[:READER_ITER:WRITER_MIN_ITER]
+    #[arg(long, value_name = "WCORE:WPC:RCORE:RPC:BEGIN:END[:RITER:WMIN]")]
+    diagnostic_schedule_target: Option<String>,
+
+    /// Maximum number of target-consumer deferrals
+    #[arg(long, default_value_t = 10_000)]
+    diagnostic_schedule_deferral_budget: u64,
+
+    /// Diagnostic delay applied when a target reader reaches its PC, formatted as CORE:PC:CYCLES or CORE:PC:ITERATION:CYCLES
+    #[arg(long, value_name = "CORE:PC[:ITERATION]:CYCLES")]
+    diagnostic_target_stall: Option<String>,
 }
 
 #[derive(Clone, Debug, ValueEnum)]
@@ -68,6 +101,13 @@ enum ExecutionMode {
     Throughput,
 }
 
+#[derive(Clone, Copy, Debug, ValueEnum)]
+enum DiagnosticSchedulePolicyArg {
+    Greedy,
+    Randomized,
+    Adversarial,
+}
+
 fn main() -> Result<()> {
     let args = Args::parse();
 
@@ -89,6 +129,33 @@ fn main() -> Result<()> {
         }
     };
     set_memory(&mut executor, memory);
+    if let Some(path) = &args.provenance_trace {
+        executor.enable_provenance(path)?;
+    }
+    executor.set_provenance_global_barrier(args.provenance_global_barrier);
+    if let Some(spec) = args.provenance_core_stall.as_deref() {
+        let (core, cycles) = parse_core_stall(spec)?;
+        executor.set_provenance_core_stall(core, cycles);
+    }
+    executor.set_diagnostic_schedule_policy(match args.diagnostic_schedule_policy {
+        DiagnosticSchedulePolicyArg::Greedy => DiagnosticSchedulePolicy::Greedy,
+        DiagnosticSchedulePolicyArg::Randomized => DiagnosticSchedulePolicy::Randomized,
+        DiagnosticSchedulePolicyArg::Adversarial => DiagnosticSchedulePolicy::Adversarial,
+    });
+    executor.set_diagnostic_schedule_seed(args.diagnostic_schedule_seed);
+    executor.set_diagnostic_schedule_deferral_budget(args.diagnostic_schedule_deferral_budget);
+    if let Some(spec) = args.diagnostic_target_stall.as_deref() {
+        let (core, pc, iteration, cycles) = parse_target_stall(spec)?;
+        executor.set_diagnostic_target_stall(core, pc, iteration, cycles);
+    }
+    if let Some(spec) = args.diagnostic_schedule_target.as_deref() {
+        executor.set_diagnostic_schedule_target(parse_schedule_target(spec)?);
+    } else if matches!(
+        args.diagnostic_schedule_policy,
+        DiagnosticSchedulePolicyArg::Adversarial
+    ) {
+        bail!("adversarial scheduling requires --diagnostic-schedule-target");
+    }
     TRACER
         .lock()
         .unwrap()
@@ -107,6 +174,75 @@ fn main() -> Result<()> {
     Ok(())
 }
 
+fn parse_core_stall(spec: &str) -> Result<(usize, u64)> {
+    let (core, cycles) = spec
+        .split_once(':')
+        .context("--provenance-core-stall must be CORE:CYCLES")?;
+    let core = core.parse().context("invalid stalled core")?;
+    let cycles = cycles.parse().context("invalid stall cycle count")?;
+    if cycles == 0 {
+        bail!("--provenance-core-stall cycles must be positive");
+    }
+    Ok((core, cycles))
+}
+
+fn parse_schedule_target(spec: &str) -> Result<DiagnosticScheduleTarget> {
+    let values: Vec<usize> = spec
+        .split(':')
+        .map(|value| {
+            value
+                .parse()
+                .with_context(|| format!("invalid schedule target field: {value}"))
+        })
+        .collect::<Result<_>>()?;
+    if values.len() != 6 && values.len() != 8 {
+        bail!(
+            "--diagnostic-schedule-target requires WCORE:WPC:RCORE:RPC:BEGIN:END with optional RITER:WMIN"
+        );
+    }
+    if values[5] <= values[4] {
+        bail!("schedule target address end must be greater than begin");
+    }
+    Ok(DiagnosticScheduleTarget {
+        writer_core: values[0],
+        writer_pc: values[1],
+        reader_core: values[2],
+        reader_pc: values[3],
+        address_begin: values[4],
+        address_end: values[5],
+        reader_iteration: (values.len() == 8).then_some(values[6] as u32),
+        writer_min_iteration: (values.len() == 8).then_some(values[7] as u32),
+    })
+}
+
+fn parse_target_stall(spec: &str) -> Result<(usize, usize, Option<u32>, u64)> {
+    let values: Vec<&str> = spec.split(':').collect();
+    if values.len() != 3 && values.len() != 4 {
+        bail!("--diagnostic-target-stall must be CORE:PC:CYCLES or CORE:PC:ITERATION:CYCLES");
+    }
+    let core = values[0].parse().context("invalid target-stall core")?;
+    let pc = values[1].parse().context("invalid target-stall PC")?;
+    let (iteration, cycle_field) = if values.len() == 4 {
+        (
+            Some(
+                values[2]
+                    .parse()
+                    .context("invalid target-stall iteration")?,
+            ),
+            values[3],
+        )
+    } else {
+        (None, values[2])
+    };
+    let cycles = cycle_field
+        .parse()
+        .context("invalid target-stall cycle count")?;
+    if cycles == 0 {
+        bail!("--diagnostic-target-stall cycles must be positive");
+    }
+    Ok((core, pc, iteration, cycles))
+}
+
 fn batch_size(args: &Args) -> Result<u32> {
     match (&args.mode, args.batch_size) {
         (ExecutionMode::Latency, None | Some(1)) => Ok(1),

backend-simulators/pim/pim-simulator/src/lib/cpu/mod.rs

@@ -2,11 +2,14 @@ use crate::utility::AddressArg;
 use anyhow::{Context, Result, ensure};
 use std::{collections::HashMap, fmt::Debug};
 
+use super::{DiagnosticSchedulePolicy, DiagnosticScheduleTarget};
 use crate::{
     cpu::crossbar::Crossbar,
     instruction_set::Instructions,
     memory_manager::{CoreMemory, MemoryStorable, type_traits::TryToUsize},
+    provenance::ProvenanceTracker,
 };
+use serde_json::json;
 
 pub mod crossbar;
 
@@ -14,6 +17,7 @@ pub mod crossbar;
 pub struct CPU<'a> {
     cores: Box<[Core<'a>]>,
     batch_outputs: Option<BatchOutputs>,
+    provenance: Option<ProvenanceTracker>,
 }
 
 #[derive(Debug, Clone)]
@@ -53,9 +57,265 @@ impl<'a> CPU<'a> {
         Self {
             cores: cores.into(),
             batch_outputs: None,
+            provenance: None,
         }
     }
 
+    pub(crate) fn enable_provenance(
+        &mut self,
+        path: impl AsRef<std::path::Path>,
+    ) -> std::io::Result<()> {
+        self.provenance = Some(ProvenanceTracker::new(self.cores.len(), path)?);
+        Ok(())
+    }
+
+    pub(crate) fn begin_provenance_batch(&mut self, batch_size: usize) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.begin_batch(batch_size);
+        }
+    }
+
+    pub(crate) fn set_execution_context(
+        &mut self,
+        cycle: u64,
+        core: usize,
+        pc: usize,
+        iteration: u32,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.set_context(cycle, core, pc, iteration);
+        }
+    }
+
+    pub(crate) fn provenance_input_store(&mut self, address: usize, size: usize, sample: u32) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.input_store(address, size, sample);
+        }
+    }
+
+    pub(crate) fn provenance_global_store_from_local(
+        &mut self,
+        core: usize,
+        global_address: usize,
+        local_address: usize,
+        size: usize,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.global_store_from_local(core, global_address, local_address, size);
+        }
+    }
+
+    pub(crate) fn provenance_global_load_to_local(
+        &mut self,
+        core: usize,
+        global_address: usize,
+        local_address: usize,
+        size: usize,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.global_load_to_local(core, global_address, local_address, size);
+        }
+    }
+
+    pub(crate) fn provenance_local_copy(
+        &mut self,
+        core: usize,
+        destination: usize,
+        source: usize,
+        size: usize,
+        operation: &'static str,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.local_copy(core, destination, source, size, operation);
+        }
+    }
+
+    pub(crate) fn provenance_local_strided_copy(
+        &mut self,
+        core: usize,
+        destination: usize,
+        source: usize,
+        element_size: usize,
+        stride: usize,
+        element_count: usize,
+        operation: &'static str,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.local_strided_copy(
+                core,
+                destination,
+                source,
+                element_size,
+                stride,
+                element_count,
+                operation,
+            );
+        }
+    }
+
+    pub(crate) fn provenance_local_transform(
+        &mut self,
+        core: usize,
+        destination: usize,
+        sources: &[(usize, usize)],
+        output_size: usize,
+        operation: &'static str,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.local_transform(core, destination, sources, output_size, operation);
+        }
+    }
+
+    pub(crate) fn provenance_local_broadcast_transform(
+        &mut self,
+        core: usize,
+        destination: usize,
+        source: usize,
+        source_size: usize,
+        output_size: usize,
+        operation: &'static str,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.local_broadcast_transform(
+                core,
+                destination,
+                source,
+                source_size,
+                output_size,
+                operation,
+            );
+        }
+    }
+
+    pub(crate) fn provenance_local_mvm_transform(
+        &mut self,
+        core: usize,
+        destination: usize,
+        source: usize,
+        element_size: usize,
+        output_size: usize,
+        used_rows: &[bool],
+        operation: &'static str,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.local_mvm_transform(
+                core,
+                destination,
+                source,
+                element_size,
+                output_size,
+                used_rows,
+                operation,
+            );
+        }
+    }
+
+    pub(crate) fn provenance_send_transfer(
+        &mut self,
+        sender: usize,
+        receiver: usize,
+        source: usize,
+        destination: usize,
+        size: usize,
+    ) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.send_transfer(sender, receiver, source, destination, size);
+        }
+    }
+
+    pub(crate) fn finish_provenance(&mut self) {
+        if let Some(provenance) = &mut self.provenance {
+            provenance.flush();
+        }
+    }
+
+    pub(crate) fn provenance_schedule_stall(&mut self, core: usize, remaining_cycles: u64) {
+        if let Some(provenance) = &self.provenance {
+            provenance.schedule_stall(core, remaining_cycles);
+        }
+    }
+
+    pub(crate) fn provenance_schedule_target_stall(
+        &mut self,
+        core: usize,
+        pc: usize,
+        remaining_cycles: u64,
+    ) {
+        if let Some(provenance) = &self.provenance {
+            provenance.schedule_target_stall(core, pc, remaining_cycles);
+        }
+    }
+
+    pub(crate) fn provenance_schedule_config(&mut self, config: super::DiagnosticScheduleConfig) {
+        if let Some(provenance) = &self.provenance {
+            provenance.schedule_config(
+                match config.policy {
+                    DiagnosticSchedulePolicy::Greedy => "greedy",
+                    DiagnosticSchedulePolicy::Randomized => "randomized",
+                    DiagnosticSchedulePolicy::Adversarial => "adversarial",
+                },
+                config.seed,
+                config.target.map(|target| {
+                    json!({
+                        "writer_core": target.writer_core,
+                        "writer_pc": target.writer_pc,
+                        "reader_core": target.reader_core,
+                        "reader_pc": target.reader_pc,
+                        "address_begin": target.address_begin,
+                        "address_end": target.address_end,
+                        "reader_iteration": target.reader_iteration,
+                        "writer_min_iteration": target.writer_min_iteration,
+                    })
+                }),
+                config.deferral_budget,
+                config.fixed_stall,
+                config.fixed_target_stall,
+            );
+        }
+    }
+
+    pub(crate) fn provenance_scheduler_event(
+        &mut self,
+        event: &'static str,
+        core: usize,
+        pc: usize,
+        iteration: u32,
+        reason: &'static str,
+        selected_core: Option<usize>,
+        target: Option<DiagnosticScheduleTarget>,
+        deferrals: u64,
+    ) {
+        if let Some(provenance) = &self.provenance {
+            let target = target.map(|target| {
+                json!({
+                    "writer_core": target.writer_core,
+                    "writer_pc": target.writer_pc,
+                    "reader_core": target.reader_core,
+                    "reader_pc": target.reader_pc,
+                    "address_begin": target.address_begin,
+                    "address_end": target.address_end,
+                    "reader_iteration": target.reader_iteration,
+                    "writer_min_iteration": target.writer_min_iteration,
+                })
+            });
+            provenance.scheduler_event(json!({
+                "event": event,
+                "cycle": self.provenance_cycle(),
+                "core": core,
+                "pc": pc,
+                "core_iteration": iteration,
+                "reason": reason,
+                "selected_core": selected_core,
+                "target": target,
+                "deferrals": deferrals,
+            }));
+        }
+    }
+
+    fn provenance_cycle(&self) -> u64 {
+        self.provenance.as_ref().map_or(0, ProvenanceTracker::cycle)
+    }
+
     pub(crate) fn set_current_iteration(&mut self, iteration: u32) {
         if let Some(batch_outputs) = &mut self.batch_outputs {
             batch_outputs.iteration = iteration as usize;
@@ -93,6 +353,7 @@ impl<'a> CPU<'a> {
         let Self {
             cores,
             batch_outputs,
+            ..
         } = self;
         let (host, cores) = cores.split_at_mut(1);
         let bytes = cores[core - 1].load::<u8>(core_address, size)?[0];

backend-simulators/pim/pim-simulator/src/lib/instruction_set/isa.rs

@@ -285,6 +285,10 @@ where
     let load = loads[0];
     let vec: Cow<[M]> = load.up();
     let matrix = crossbar.load::<M>(crossbar_stored_bytes)?[0];
+    let used_rows: Vec<bool> = matrix
+        .chunks_exact(crossbar_elem_width)
+        .map(|row| row.iter().any(|value| *value != M::from_f32(0.0)))
+        .collect();
 
     // --- FAER IMPLEMENTATION ---
 
@@ -323,6 +327,16 @@ where
 
     let res_up: Cow<[T]> = res.as_slice().up();
     core.execute_store(rd_val, res_up.as_ref());
+    let _ = core;
+    cores.provenance_local_mvm_transform(
+        core_indx as usize,
+        rd_val as usize,
+        r1_val as usize,
+        size_of::<F>(),
+        res_up.len() * size_of::<T>(),
+        &used_rows,
+        "mvmul",
+    );
 
     TRACER.lock().unwrap().post_mvm::<F, M, T>(cores, data);
     Ok(InstructionStatus::Completed)
@@ -389,6 +403,14 @@ where
     );
     let res_up: Cow<[T]> = res.as_slice().up();
     core.execute_store(rd_val, res_up.as_ref());
+    let _ = core;
+    cores.provenance_local_transform(
+        core_indx as usize,
+        rd_val,
+        &[(r1_val, byte_len), (r2_val, byte_len)],
+        byte_len,
+        "vvadd",
+    );
     TRACER.lock().unwrap().post_vvadd::<F, T>(cores, data);
     Ok(InstructionStatus::Completed)
 }
@@ -474,6 +496,13 @@ where
     );
     let res_up: Cow<[T]> = res.as_slice().up();
     core.execute_store(rd_val, res_up.as_ref());
+    cores.provenance_local_transform(
+        core_indx as usize,
+        rd_val,
+        &[(r1_val, byte_len), (r2_val, byte_len)],
+        byte_len,
+        "vvmul",
+    );
     Ok(InstructionStatus::Completed)
 }
 
@@ -780,6 +809,15 @@ where
         );
     }
     core.execute_store(destination, &result)?;
+    cores.provenance_local_strided_copy(
+        core_indx as usize,
+        destination as usize,
+        source,
+        size_of::<F>(),
+        stride,
+        element_count,
+        "vmv",
+    );
     Ok(InstructionStatus::Completed)
 }
 
@@ -799,16 +837,23 @@ pub fn vrsl(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus>
 #[inline(never)]
 pub fn ld(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus> {
     TRACER.lock().unwrap().pre_ld(cores, data);
-    let (core, rd, r1, _, imm_len, offset_select, offset_value) =
+    let (core_index, rd, r1, _, imm_len, offset_select, offset_value) =
         data.get_core_rd_r1_r2_immlen_offset();
-    ensure!(core != 0, "LD cannot be used to move from host to host");
-    let (host, core) = cores.host_and_cores(core);
-    let r1_val = core.register(r1);
-    let rd_val = core.register(rd);
-    let r1_val = add_offset_r1(r1_val, offset_select, offset_value);
-    let rd_val = add_offset_rd(rd_val, offset_select, offset_value);
-    let global_memory = host.load::<u8>(r1_val, imm_len)?;
-    core.execute_store(rd_val, global_memory[0])?;
+    ensure!(
+        core_index != 0,
+        "LD cannot be used to move from host to host"
+    );
+    let (r1_val, rd_val) = {
+        let (host, core) = cores.host_and_cores(core_index);
+        let r1_val = core.register(r1);
+        let rd_val = core.register(rd);
+        let r1_val = add_offset_r1(r1_val, offset_select, offset_value);
+        let rd_val = add_offset_rd(rd_val, offset_select, offset_value);
+        let global_memory = host.load::<u8>(r1_val, imm_len)?;
+        core.execute_store(rd_val, global_memory[0])?;
+        (r1_val, rd_val)
+    };
+    cores.provenance_global_load_to_local(core_index as usize, r1_val, rd_val, imm_len as usize);
     TRACER.lock().unwrap().post_ld(cores, data);
     Ok(InstructionStatus::Completed)
 }
@@ -828,6 +873,7 @@ pub fn st(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus> {
         (rd_val, r1_val)
     };
     cores.store_to_host(core, rd_val, r1_val, imm_len)?;
+    cores.provenance_global_store_from_local(core as usize, rd_val, r1_val, imm_len as usize);
     TRACER.lock().unwrap().post_st(cores, data);
     Ok(InstructionStatus::Completed)
 }
@@ -852,9 +898,9 @@ pub fn lldi(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus>
 #[inline(never)]
 pub fn lmv(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus> {
     TRACER.lock().unwrap().pre_lmv(cores, data);
-    let (core, rd, r1, _, imm_len, offset_select, offset_value) =
+    let (core_index, rd, r1, _, imm_len, offset_select, offset_value) =
         data.get_core_rd_r1_r2_immlen_offset();
-    let core = cores.core(core);
+    let core = cores.core(core_index);
     let r1_val = core.register(r1);
     let rd_val = core.register(rd);
     let r1_val = add_offset_r1(r1_val, offset_select, offset_value);
@@ -862,6 +908,8 @@ pub fn lmv(cores: &mut CPU, data: InstructionData) -> Result<InstructionStatus>
     let local_memory = core.load::<u8>(r1_val, imm_len)?;
     let tmp = local_memory[0].to_vec();
     core.execute_store(rd_val, tmp.as_slice());
+    let _ = core;
+    cores.provenance_local_copy(core_index as usize, rd_val, r1_val, imm_len as usize, "lmv");
     TRACER.lock().unwrap().post_lmv(cores, data);
     Ok(InstructionStatus::Completed)
 }

backend-simulators/pim/pim-simulator/src/lib/pimcore.rs

@@ -1,8 +1,10 @@
 #![allow(unused)]
 
 use anyhow::{Context, Result, bail};
+use serde_json::json;
 use std::{
     collections::{HashMap, HashSet},
+    path::Path,
     sync::{
         Mutex,
         atomic::{AtomicU32, Ordering},
@@ -25,6 +27,7 @@ pub mod cpu;
 pub mod instruction_set;
 pub mod json_to_instruction;
 pub mod memory_manager;
+pub mod provenance;
 pub mod send_recv;
 pub mod tracing;
 pub mod utility;
@@ -92,11 +95,315 @@ impl From<Instructions> for CoreInstructions {
     }
 }
 
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub enum DiagnosticSchedulePolicy {
+    #[default]
+    Greedy,
+    Randomized,
+    Adversarial,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct DiagnosticScheduleTarget {
+    pub writer_core: usize,
+    pub writer_pc: usize,
+    pub reader_core: usize,
+    pub reader_pc: usize,
+    pub address_begin: usize,
+    pub address_end: usize,
+    pub reader_iteration: Option<u32>,
+    pub writer_min_iteration: Option<u32>,
+}
+
+#[derive(Clone, Copy, Debug)]
+struct DiagnosticScheduleConfig {
+    policy: DiagnosticSchedulePolicy,
+    seed: u64,
+    target: Option<DiagnosticScheduleTarget>,
+    deferral_budget: u64,
+    fixed_stall: Option<(usize, u64)>,
+    fixed_target_stall: Option<(usize, usize, Option<u32>, u64)>,
+}
+
+impl Default for DiagnosticScheduleConfig {
+    fn default() -> Self {
+        Self {
+            policy: DiagnosticSchedulePolicy::Greedy,
+            seed: 0,
+            target: None,
+            deferral_budget: 10_000,
+            fixed_stall: None,
+            fixed_target_stall: None,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+struct ScheduleCoreState {
+    program_counter: usize,
+    instruction_count: usize,
+    current_iteration: u32,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum ScheduleAction {
+    Execute,
+    Defer {
+        next_core: usize,
+        reason: &'static str,
+    },
+    Stall {
+        remaining: u64,
+        target: bool,
+    },
+    Force {
+        reason: &'static str,
+    },
+}
+
+#[derive(Clone, Copy, Debug)]
+struct ScheduleChoice {
+    core: usize,
+    reason: &'static str,
+}
+
+#[derive(Debug)]
+struct DiagnosticScheduler {
+    config: DiagnosticScheduleConfig,
+    random_state: u64,
+    deferrals: u64,
+    writer_store_iteration: Option<u32>,
+    last_writer_state: Option<ScheduleCoreState>,
+    writer_stagnation: u64,
+}
+
+impl DiagnosticScheduler {
+    fn new(config: DiagnosticScheduleConfig) -> Self {
+        let random_state = if config.seed == 0 {
+            0x9e37_79b9_7f4a_7c15
+        } else {
+            config.seed
+        };
+        Self {
+            config,
+            random_state,
+            deferrals: 0,
+            writer_store_iteration: None,
+            last_writer_state: None,
+            writer_stagnation: 0,
+        }
+    }
+
+    fn state(cores: &[CoreInstructions]) -> Vec<ScheduleCoreState> {
+        cores
+            .iter()
+            .map(|core| ScheduleCoreState {
+                program_counter: core.program_counter,
+                instruction_count: core.instructions.len(),
+                current_iteration: core.current_iteration,
+            })
+            .collect()
+    }
+
+    fn has_work(core: ScheduleCoreState, batch_size: u32) -> bool {
+        !((core.instruction_count == 0)
+            || (core.program_counter == core.instruction_count
+                && core.current_iteration + 1 >= batch_size))
+    }
+
+    fn candidates(states: &[ScheduleCoreState], current: usize, batch_size: u32) -> Vec<usize> {
+        states
+            .iter()
+            .enumerate()
+            .filter_map(|(index, &core)| {
+                (index != current && Self::has_work(core, batch_size)).then_some(index)
+            })
+            .collect()
+    }
+
+    fn next_random(&mut self) -> u64 {
+        let mut value = self.random_state;
+        value ^= value << 13;
+        value ^= value >> 7;
+        value ^= value << 17;
+        self.random_state = value;
+        value
+    }
+
+    fn random_choice(&mut self, candidates: &[usize]) -> Option<ScheduleChoice> {
+        (!candidates.is_empty()).then(|| ScheduleChoice {
+            core: candidates[(self.next_random() as usize) % candidates.len()],
+            reason: "randomized_ready_core",
+        })
+    }
+
+    fn target_reader(&self, core: usize, state: ScheduleCoreState) -> bool {
+        self.config.target.is_some_and(|target| {
+            target.reader_core == core
+                && target.reader_pc == state.program_counter
+                && target
+                    .reader_iteration
+                    .is_none_or(|iteration| iteration == state.current_iteration)
+        })
+    }
+
+    fn target_writer_has_work(&self, states: &[ScheduleCoreState], batch_size: u32) -> bool {
+        self.config.target.is_some_and(|target| {
+            states
+                .get(target.writer_core)
+                .is_some_and(|&state| Self::has_work(state, batch_size))
+        })
+    }
+
+    fn reader_may_execute(&self, reader_iteration: u32) -> bool {
+        let minimum = self
+            .config
+            .target
+            .and_then(|target| target.writer_min_iteration)
+            .unwrap_or(reader_iteration + 1);
+        self.writer_store_iteration
+            .is_some_and(|iteration| iteration >= minimum)
+    }
+
+    fn before_instruction(
+        &mut self,
+        current: usize,
+        states: &[ScheduleCoreState],
+        batch_size: u32,
+    ) -> ScheduleAction {
+        if let Some((core, pc, iteration, remaining)) = self.config.fixed_target_stall
+            && core == current
+            && states.get(current).is_some_and(|state| {
+                state.program_counter == pc
+                    && iteration.is_none_or(|iteration| state.current_iteration == iteration)
+            })
+            && remaining > 0
+        {
+            self.config.fixed_target_stall = Some((core, pc, iteration, remaining - 1));
+            return ScheduleAction::Stall {
+                remaining: remaining - 1,
+                target: true,
+            };
+        }
+        if let Some((core, remaining)) = self.config.fixed_stall
+            && core == current
+            && remaining > 0
+        {
+            self.config.fixed_stall = Some((core, remaining - 1));
+            return ScheduleAction::Stall {
+                remaining: remaining - 1,
+                target: false,
+            };
+        }
+
+        let Some(state) = states.get(current).copied() else {
+            return ScheduleAction::Execute;
+        };
+        let candidates = Self::candidates(states, current, batch_size);
+        match self.config.policy {
+            DiagnosticSchedulePolicy::Greedy => ScheduleAction::Execute,
+            // Random choices are made after a blocking/ready boundary in
+            // `after_block`.  Deferring here would allow two ready cores to
+            // defer each other forever without executing an instruction.
+            DiagnosticSchedulePolicy::Randomized => ScheduleAction::Execute,
+            DiagnosticSchedulePolicy::Adversarial => {
+                let Some(target) = self.config.target else {
+                    return ScheduleAction::Execute;
+                };
+                if !self.target_reader(current, state)
+                    || self.reader_may_execute(state.current_iteration)
+                {
+                    return ScheduleAction::Execute;
+                }
+                if self.deferrals >= self.config.deferral_budget {
+                    return ScheduleAction::Force {
+                        reason: "DEFERRAL_LIMIT_REACHED",
+                    };
+                }
+                let writer_state = states.get(target.writer_core).copied();
+                let next_core = if Self::has_work(writer_state.unwrap_or(state), batch_size) {
+                    if self.last_writer_state == writer_state {
+                        self.writer_stagnation += 1;
+                    } else {
+                        self.last_writer_state = writer_state;
+                        self.writer_stagnation = 0;
+                    }
+                    if self.writer_stagnation >= 256 {
+                        return ScheduleAction::Force {
+                            reason: "PRODUCER_BLOCKED_BY_REAL_DEPENDENCY",
+                        };
+                    }
+                    target.writer_core
+                } else {
+                    candidates.first().copied().unwrap_or(current)
+                };
+                if next_core == current {
+                    ScheduleAction::Force {
+                        reason: "NO_ALTERNATIVE_READY_EVENT",
+                    }
+                } else {
+                    self.deferrals += 1;
+                    ScheduleAction::Defer {
+                        next_core,
+                        reason: "target_consumer",
+                    }
+                }
+            }
+        }
+    }
+
+    fn after_block(
+        &mut self,
+        current: usize,
+        states: &[ScheduleCoreState],
+        batch_size: u32,
+    ) -> Option<ScheduleChoice> {
+        let candidates = Self::candidates(states, current, batch_size);
+        match self.config.policy {
+            DiagnosticSchedulePolicy::Greedy => None,
+            DiagnosticSchedulePolicy::Randomized => self.random_choice(&candidates),
+            DiagnosticSchedulePolicy::Adversarial => {
+                let target = self.config.target?;
+                if self.target_writer_has_work(states, batch_size)
+                    && target.writer_core != current
+                    && candidates.contains(&target.writer_core)
+                {
+                    Some(ScheduleChoice {
+                        core: target.writer_core,
+                        reason: "target_producer",
+                    })
+                } else {
+                    candidates.first().copied().map(|core| ScheduleChoice {
+                        core,
+                        reason: "adversarial_ready_core",
+                    })
+                }
+            }
+        }
+    }
+
+    fn note_completed(&mut self, core: usize, pc: usize, iteration: u32) {
+        if self
+            .config
+            .target
+            .is_some_and(|target| target.writer_core == core && target.writer_pc == pc)
+        {
+            self.writer_store_iteration = Some(iteration);
+        }
+    }
+
+    fn config(&self) -> DiagnosticScheduleConfig {
+        self.config
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct Executable<'a> {
     cpu: CPU<'a>,
     core_instructions: Vec<CoreInstructions>,
     send_recv: SendRecv,
+    provenance_global_barrier: bool,
+    diagnostic_schedule: DiagnosticScheduleConfig,
 }
 
 struct DeadlockInfo {
@@ -134,9 +441,52 @@ impl<'a> Executable<'a> {
             cpu,
             core_instructions,
             send_recv,
+            provenance_global_barrier: false,
+            diagnostic_schedule: DiagnosticScheduleConfig::default(),
         }
     }
 
+    pub fn enable_provenance(&mut self, path: impl AsRef<Path>) -> Result<()> {
+        self.cpu
+            .enable_provenance(path)
+            .context("cannot enable provenance tracing")?;
+        Ok(())
+    }
+
+    pub fn set_provenance_global_barrier(&mut self, enabled: bool) {
+        self.provenance_global_barrier = enabled;
+    }
+
+    pub fn set_provenance_core_stall(&mut self, core: usize, cycles: u64) {
+        self.diagnostic_schedule.fixed_stall = Some((core, cycles));
+    }
+
+    pub fn set_diagnostic_target_stall(
+        &mut self,
+        core: usize,
+        pc: usize,
+        iteration: Option<u32>,
+        cycles: u64,
+    ) {
+        self.diagnostic_schedule.fixed_target_stall = Some((core, pc, iteration, cycles));
+    }
+
+    pub fn set_diagnostic_schedule_policy(&mut self, policy: DiagnosticSchedulePolicy) {
+        self.diagnostic_schedule.policy = policy;
+    }
+
+    pub fn set_diagnostic_schedule_seed(&mut self, seed: u64) {
+        self.diagnostic_schedule.seed = seed;
+    }
+
+    pub fn set_diagnostic_schedule_target(&mut self, target: DiagnosticScheduleTarget) {
+        self.diagnostic_schedule.target = Some(target);
+    }
+
+    pub fn set_diagnostic_schedule_deferral_budget(&mut self, budget: u64) {
+        self.diagnostic_schedule.deferral_budget = budget;
+    }
+
     pub fn execute<'b>(&'b mut self) -> Result<()>
     where
         'a: 'b,
@@ -169,28 +519,113 @@ impl<'a> Executable<'a> {
         let _execution_lock = EXECUTION_LOCK.lock().unwrap();
         let batch_size = u32::try_from(inputs.len().max(1)).context("batch size exceeds u32")?;
         GLOBAL_ITERATION.store(0, Ordering::SeqCst);
+        self.cpu.begin_provenance_batch(batch_size as usize);
         if let Some(input) = inputs.first() {
-            store_input(&mut self.cpu, input, input_regions)?;
+            store_input(&mut self.cpu, input, input_regions, 0)?;
         }
         self.cpu
             .begin_host_store_recording(batch_size as usize, dump_ranges)?;
 
+        let provenance_global_barrier = self.provenance_global_barrier;
+        let mut scheduler = DiagnosticScheduler::new(self.diagnostic_schedule);
+        self.cpu.provenance_schedule_config(scheduler.config());
         let Self {
             cpu,
             core_instructions: cores_instructions,
             send_recv,
+            ..
         } = self;
+        let active_cores: Vec<usize> = cores_instructions
+            .iter()
+            .enumerate()
+            .filter_map(|(index, core)| (!core.instructions.is_empty()).then_some(index))
+            .collect();
+        let mut barrier_iteration = None;
         let mut cpu_progressed = 0;
         let max_core = cpu.num_core();
         let mut sync_events: SyncEvents = vec![[0; 32]; max_core];
         let mut cpu_index = 0;
+        let mut cycle = 0;
+        let mut scheduler_no_progress = 0usize;
+        let scheduler_no_progress_limit = max_core.saturating_mul(4).max(8);
         let mut now = SystemTime::now();
 
         while (cpu_progressed > -2) {
             let mut core_result = InstructionStatus::Completed;
-            while core_result.is_completed()
-                && let Some(core_instruction) = cores_instructions.get_mut(cpu_index)
+            let mut scheduler_next = None;
+            if provenance_global_barrier
+                && barrier_iteration.is_some()
+                && active_cores.iter().all(|&index| {
+                    cores_instructions[index].current_iteration >= barrier_iteration.unwrap()
+                })
             {
+                barrier_iteration = None;
+            }
+            while core_result.is_completed() {
+                let barrier_ready = if provenance_global_barrier {
+                    let current_iteration = cores_instructions[cpu_index].current_iteration;
+                    active_cores.iter().all(|&index| {
+                        let core = &cores_instructions[index];
+                        core.program_counter == core.instructions.len()
+                            && core.current_iteration == current_iteration
+                    })
+                } else {
+                    false
+                };
+                let current_pc = cores_instructions[cpu_index].program_counter;
+                let current_iteration = cores_instructions[cpu_index].current_iteration;
+                let states = if scheduler.config().policy == DiagnosticSchedulePolicy::Greedy
+                    && scheduler.config().fixed_stall.is_none()
+                    && scheduler.config().fixed_target_stall.is_none()
+                {
+                    None
+                } else {
+                    Some(DiagnosticScheduler::state(cores_instructions))
+                };
+                let schedule_action = states.as_deref().map_or(ScheduleAction::Execute, |states| {
+                    scheduler.before_instruction(cpu_index, states, batch_size)
+                });
+                if let ScheduleAction::Defer { next_core, reason } = schedule_action {
+                    cpu.set_execution_context(cycle, cpu_index, current_pc, current_iteration);
+                    cpu.provenance_scheduler_event(
+                        "scheduler_defer",
+                        cpu_index,
+                        current_pc,
+                        current_iteration,
+                        reason,
+                        Some(next_core),
+                        scheduler.config().target,
+                        scheduler.deferrals,
+                    );
+                    scheduler_next = Some(next_core);
+                    break;
+                }
+                if let ScheduleAction::Stall { remaining, target } = schedule_action {
+                    cpu_progressed = 0;
+                    cpu.set_execution_context(cycle, cpu_index, current_pc, current_iteration);
+                    if target {
+                        cpu.provenance_schedule_target_stall(cpu_index, current_pc, remaining);
+                    } else {
+                        cpu.provenance_schedule_stall(cpu_index, remaining);
+                    }
+                    break;
+                }
+                if let ScheduleAction::Force { reason } = schedule_action {
+                    cpu.set_execution_context(cycle, cpu_index, current_pc, current_iteration);
+                    cpu.provenance_scheduler_event(
+                        "scheduler_force",
+                        cpu_index,
+                        current_pc,
+                        current_iteration,
+                        reason,
+                        None,
+                        scheduler.config().target,
+                        scheduler.deferrals,
+                    );
+                }
+                let Some(core_instruction) = cores_instructions.get_mut(cpu_index) else {
+                    break;
+                };
                 core_result = InstructionStatus::NotExecuted;
                 if core_instruction.program_counter == core_instruction.instructions.len() {
                     if core_instruction.instructions.is_empty()
@@ -198,48 +633,129 @@ impl<'a> Executable<'a> {
                     {
                         break;
                     }
+                    let next_iteration = core_instruction.current_iteration + 1;
+                    if provenance_global_barrier {
+                        if barrier_iteration != Some(next_iteration) {
+                            if !barrier_ready {
+                                break;
+                            }
+                            barrier_iteration = Some(next_iteration);
+                        }
+                    }
                     core_instruction.current_iteration += 1;
                     core_instruction.program_counter = 0;
                     let iteration = core_instruction.current_iteration;
                     if iteration > GLOBAL_ITERATION.fetch_max(iteration, Ordering::SeqCst) {
-                        store_input(cpu, inputs[iteration as usize], input_regions)?;
+                        cpu.set_execution_context(cycle, cpu_index, 0, iteration);
+                        store_input(cpu, inputs[iteration as usize], input_regions, iteration)?;
                     }
                 }
-                cpu.set_current_iteration(core_instruction.current_iteration);
-                let CoreInstructions {
-                    instructions,
-                    program_counter,
-                    ..
-                } = core_instruction;
-                core_result = instructions
-                    .get(*program_counter)
-                    .map_or(InstructionStatus::default(), |inst: &Instruction| {
-                        inst.execute(cpu)
-                    });
-                if core_result.is_completed() {
-                    cpu_progressed = 0;
-                    *program_counter += 1;
-                }
-                if (now.elapsed().unwrap() > Duration::from_secs(5)) {
-                    print_status(cores_instructions);
-                    if let Some(deadlock) = detect_deadlock(cores_instructions) {
-                        bail!(
-                            "Deadlock cycle detected: {} [{}]",
-                            deadlock.cycle,
-                            deadlock.states
+                if !matches!(
+                    schedule_action,
+                    ScheduleAction::Stall { .. } | ScheduleAction::Defer { .. }
+                ) {
+                    cpu.set_current_iteration(core_instruction.current_iteration);
+                    let CoreInstructions {
+                        instructions,
+                        program_counter,
+                        ..
+                    } = core_instruction;
+                    cpu.set_execution_context(
+                        cycle,
+                        cpu_index,
+                        *program_counter,
+                        core_instruction.current_iteration,
+                    );
+                    cycle += 1;
+                    core_result = instructions
+                        .get(*program_counter)
+                        .map_or(InstructionStatus::default(), |inst: &Instruction| {
+                            inst.execute(cpu)
+                        });
+                    if core_result.is_completed() {
+                        scheduler.note_completed(
+                            cpu_index,
+                            *program_counter,
+                            core_instruction.current_iteration,
                         );
+                        cpu_progressed = 0;
+                        scheduler_no_progress = 0;
+                        *program_counter += 1;
+                    }
+                    if (now.elapsed().unwrap() > Duration::from_secs(5)) {
+                        print_status(cores_instructions);
+                        if let Some(deadlock) = detect_deadlock(cores_instructions) {
+                            bail!(
+                                "Deadlock cycle detected: {} [{}]",
+                                deadlock.cycle,
+                                deadlock.states
+                            );
+                        }
+                        now = SystemTime::now();
                     }
-                    now = SystemTime::now();
                 }
             }
             if handle_wait_sync(cores_instructions, &mut sync_events, core_result) {
                 cpu_progressed = 0;
+                scheduler_no_progress = 0;
             }
-            match handle_send_recv(cpu, cores_instructions, send_recv, core_result) {
-                (true, other_cpu_index) => {
-                    cpu_progressed = 0;
-                    cpu_index = other_cpu_index;
-                }
+            if let Some(next_core) = scheduler_next {
+                cpu_index = next_core;
+                continue;
+            }
+            let send_recv_result =
+                handle_send_recv(cpu, cores_instructions, send_recv, core_result);
+            if let (true, other_cpu_index) = send_recv_result {
+                cpu_progressed = 0;
+                scheduler_no_progress = 0;
+                cpu_index = other_cpu_index;
+                continue;
+            }
+            if !core_result.is_completed() {
+                scheduler_no_progress += 1;
+            }
+            let states = if scheduler.config().policy == DiagnosticSchedulePolicy::Greedy {
+                None
+            } else {
+                Some(DiagnosticScheduler::state(cores_instructions))
+            };
+            let scheduler_choice = (scheduler_no_progress <= scheduler_no_progress_limit)
+                .then(|| {
+                    states
+                        .as_deref()
+                        .and_then(|states| scheduler.after_block(cpu_index, states, batch_size))
+                })
+                .flatten();
+            if let Some(choice) = scheduler_choice {
+                cpu.provenance_scheduler_event(
+                    "scheduler_prefer",
+                    cpu_index,
+                    cores_instructions[cpu_index].program_counter,
+                    cores_instructions[cpu_index].current_iteration,
+                    choice.reason,
+                    Some(choice.core),
+                    scheduler.config().target,
+                    scheduler.deferrals,
+                );
+                cpu_index = choice.core;
+                continue;
+            }
+            if scheduler_no_progress == scheduler_no_progress_limit + 1
+                && scheduler.config().policy != DiagnosticSchedulePolicy::Greedy
+            {
+                cpu.provenance_scheduler_event(
+                    "scheduler_force",
+                    cpu_index,
+                    cores_instructions[cpu_index].program_counter,
+                    cores_instructions[cpu_index].current_iteration,
+                    "NO_ALTERNATIVE_READY_EVENT",
+                    None,
+                    scheduler.config().target,
+                    scheduler.deferrals,
+                );
+            }
+            match send_recv_result {
+                (true, _) => unreachable!("completed SEND/RECV was handled above"),
                 (false, 0) => {
                     cpu_index = if cpu_index + 1 >= cores_instructions.len() {
                         cpu_progressed -= 1;
@@ -271,6 +787,7 @@ impl<'a> Executable<'a> {
 
         #[cfg(feature = "profile_time")]
         TRACER.lock().unwrap().report();
+        cpu.finish_provenance();
         Ok(cpu.finish_host_store_recording())
     }
 
@@ -306,11 +823,17 @@ fn validate_inputs(inputs: &[&[u8]], input_regions: &[(usize, usize)]) -> Result
     Ok(())
 }
 
-fn store_input(cpu: &mut CPU, input: &[u8], input_regions: &[(usize, usize)]) -> Result<()> {
+fn store_input(
+    cpu: &mut CPU,
+    input: &[u8],
+    input_regions: &[(usize, usize)],
+    sample: u32,
+) -> Result<()> {
     let mut offset = 0;
     for &(address, size) in input_regions {
         cpu.host()
             .execute_store(address, &input[offset..offset + size])?;
+        cpu.provenance_input_store(address, size, sample);
         offset += size;
     }
     Ok(())
@@ -465,3 +988,143 @@ fn handle_wait_sync(
         _ => false,
     }
 }
+
+#[cfg(test)]
+mod scheduler_tests {
+    use super::*;
+
+    fn target() -> DiagnosticScheduleTarget {
+        DiagnosticScheduleTarget {
+            writer_core: 0,
+            writer_pc: 3,
+            reader_core: 1,
+            reader_pc: 2,
+            address_begin: 100,
+            address_end: 200,
+            reader_iteration: Some(0),
+            writer_min_iteration: Some(1),
+        }
+    }
+
+    fn states() -> Vec<ScheduleCoreState> {
+        vec![
+            ScheduleCoreState {
+                program_counter: 3,
+                instruction_count: 8,
+                current_iteration: 1,
+            },
+            ScheduleCoreState {
+                program_counter: 2,
+                instruction_count: 8,
+                current_iteration: 0,
+            },
+        ]
+    }
+
+    #[test]
+    fn adversarial_policy_defers_reader_until_writer_store() {
+        let mut scheduler = DiagnosticScheduler::new(DiagnosticScheduleConfig {
+            policy: DiagnosticSchedulePolicy::Adversarial,
+            target: Some(target()),
+            deferral_budget: 4,
+            ..Default::default()
+        });
+        assert_eq!(
+            scheduler.before_instruction(1, &states(), 4),
+            ScheduleAction::Defer {
+                next_core: 0,
+                reason: "target_consumer"
+            }
+        );
+        scheduler.note_completed(0, 3, 1);
+        assert_eq!(
+            scheduler.before_instruction(1, &states(), 4),
+            ScheduleAction::Execute
+        );
+    }
+
+    #[test]
+    fn randomized_policy_is_deterministic_for_fixed_seed() {
+        let config = DiagnosticScheduleConfig {
+            policy: DiagnosticSchedulePolicy::Randomized,
+            seed: 17,
+            ..Default::default()
+        };
+        let mut left = DiagnosticScheduler::new(config);
+        let mut right = DiagnosticScheduler::new(config);
+        let states = vec![
+            ScheduleCoreState {
+                program_counter: 0,
+                instruction_count: 4,
+                current_iteration: 0,
+            },
+            ScheduleCoreState {
+                program_counter: 1,
+                instruction_count: 4,
+                current_iteration: 0,
+            },
+            ScheduleCoreState {
+                program_counter: 2,
+                instruction_count: 4,
+                current_iteration: 0,
+            },
+        ];
+        for current in [0, 1, 2, 0, 1] {
+            assert_eq!(
+                left.before_instruction(current, &states, 2),
+                right.before_instruction(current, &states, 2)
+            );
+        }
+    }
+
+    #[test]
+    fn target_stall_is_consumed_without_changing_program_order() {
+        let mut scheduler = DiagnosticScheduler::new(DiagnosticScheduleConfig {
+            fixed_target_stall: Some((1, 2, None, 2)),
+            ..Default::default()
+        });
+        let states = states();
+        assert_eq!(
+            scheduler.before_instruction(1, &states, 4),
+            ScheduleAction::Stall {
+                remaining: 1,
+                target: true
+            }
+        );
+        assert_eq!(
+            scheduler.before_instruction(1, &states, 4),
+            ScheduleAction::Stall {
+                remaining: 0,
+                target: true
+            }
+        );
+        assert_eq!(
+            scheduler.before_instruction(1, &states, 4),
+            ScheduleAction::Execute
+        );
+        assert_eq!(states[1].program_counter, 2);
+    }
+
+    #[test]
+    fn adversarial_policy_releases_reader_when_writer_state_stagnates() {
+        let mut scheduler = DiagnosticScheduler::new(DiagnosticScheduleConfig {
+            policy: DiagnosticSchedulePolicy::Adversarial,
+            target: Some(target()),
+            deferral_budget: 10_000,
+            ..Default::default()
+        });
+        let states = states();
+        for _ in 0..256 {
+            assert!(matches!(
+                scheduler.before_instruction(1, &states, 4),
+                ScheduleAction::Defer { .. }
+            ));
+        }
+        assert_eq!(
+            scheduler.before_instruction(1, &states, 4),
+            ScheduleAction::Force {
+                reason: "PRODUCER_BLOCKED_BY_REAL_DEPENDENCY"
+            }
+        );
+    }
+}

backend-simulators/pim/pim-simulator/src/lib/provenance.rs

@@ -0,0 +1,639 @@
+use serde_json::{Value, json};
+use std::{
+    collections::BTreeSet,
+    fs::File,
+    io::{BufWriter, Write},
+    path::Path,
+    sync::{Arc, Mutex},
+};
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
+pub enum Provenance {
+    #[default]
+    Uninitialized,
+    Unknown,
+    Samples(u64),
+}
+
+impl Provenance {
+    pub fn sample(sample: u32) -> Self {
+        if sample < 64 {
+            Self::Samples(1_u64 << sample)
+        } else {
+            Self::Unknown
+        }
+    }
+
+    fn merge(self, other: Self) -> Self {
+        match (self, other) {
+            (Self::Unknown, _) | (_, Self::Unknown) => Self::Unknown,
+            (Self::Uninitialized, value) | (value, Self::Uninitialized) => value,
+            (Self::Samples(left), Self::Samples(right)) => Self::Samples(left | right),
+        }
+    }
+
+    fn merge_all(values: impl IntoIterator<Item = Self>) -> Self {
+        let mut values = values.into_iter();
+        values
+            .next()
+            .map_or(Self::Uninitialized, |first| values.fold(first, Self::merge))
+    }
+
+    fn samples(self) -> Vec<u32> {
+        match self {
+            Self::Samples(mask) => (0..64)
+                .filter(|sample| mask & (1_u64 << sample) != 0)
+                .collect(),
+            Self::Unknown | Self::Uninitialized => Vec::new(),
+        }
+    }
+
+    fn state(self) -> &'static str {
+        match self {
+            Self::Samples(_) => "known",
+            Self::Unknown => "unknown",
+            Self::Uninitialized => "uninitialized",
+        }
+    }
+
+    fn is_mixed(self) -> bool {
+        matches!(self, Self::Samples(mask) if mask.count_ones() > 1)
+    }
+
+    fn json(self) -> Value {
+        json!({
+            "provenance": self.samples(),
+            "provenance_state": self.state(),
+            "mixed": self.is_mixed(),
+        })
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+struct ExecutionContext {
+    cycle: u64,
+    core: usize,
+    pc: usize,
+    iteration: u32,
+}
+
+impl Default for ExecutionContext {
+    fn default() -> Self {
+        Self {
+            cycle: 0,
+            core: 0,
+            pc: 0,
+            iteration: 0,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+struct Writer {
+    version: u64,
+    cycle: u64,
+    core: usize,
+    pc: usize,
+    iteration: u32,
+    provenance: Provenance,
+}
+
+#[derive(Clone, Copy, Debug, Default)]
+struct GlobalCell {
+    provenance: Provenance,
+    writer: Option<Writer>,
+}
+
+#[derive(Debug)]
+struct TraceSink {
+    output: BufWriter<File>,
+}
+
+impl TraceSink {
+    fn new(path: impl AsRef<Path>) -> std::io::Result<Self> {
+        let file = File::create(path)?;
+        Ok(Self {
+            output: BufWriter::new(file),
+        })
+    }
+
+    fn event(&mut self, value: Value) {
+        serde_json::to_writer(&mut self.output, &value).expect("write provenance event");
+        self.output
+            .write_all(b"\n")
+            .expect("write provenance newline");
+    }
+
+    fn flush(&mut self) {
+        self.output.flush().expect("flush provenance trace");
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct ProvenanceTracker {
+    global: Vec<GlobalCell>,
+    local: Vec<Vec<Provenance>>,
+    next_version: u64,
+    context: ExecutionContext,
+    sink: Arc<Mutex<TraceSink>>,
+}
+
+impl ProvenanceTracker {
+    pub fn new(core_count: usize, path: impl AsRef<Path>) -> std::io::Result<Self> {
+        if let Some(parent) = path.as_ref().parent() {
+            std::fs::create_dir_all(parent)?;
+        }
+        let sink = Arc::new(Mutex::new(TraceSink::new(path)?));
+        let tracker = Self {
+            global: Vec::new(),
+            local: vec![Vec::new(); core_count],
+            next_version: 0,
+            context: ExecutionContext::default(),
+            sink,
+        };
+        tracker.write(json!({
+            "event": "provenance_trace_start",
+            "schema": 1,
+            "core_count": core_count,
+        }));
+        Ok(tracker)
+    }
+
+    pub fn begin_batch(&mut self, batch_size: usize) {
+        self.global.fill(GlobalCell::default());
+        for memory in &mut self.local {
+            memory.fill(Provenance::Uninitialized);
+        }
+        self.next_version = 0;
+        self.write(json!({
+            "event": "batch_start",
+            "batch_size": batch_size,
+        }));
+    }
+
+    pub fn set_context(&mut self, cycle: u64, core: usize, pc: usize, iteration: u32) {
+        self.context = ExecutionContext {
+            cycle,
+            core,
+            pc,
+            iteration,
+        };
+    }
+
+    pub fn cycle(&self) -> u64 {
+        self.context.cycle
+    }
+
+    pub fn flush(&mut self) {
+        self.sink.lock().unwrap().flush();
+    }
+
+    pub fn schedule_stall(&self, core: usize, remaining_cycles: u64) {
+        self.write(json!({
+            "event": "diagnostic_core_stall",
+            "cycle": self.context.cycle,
+            "core": core,
+            "pc": self.context.pc,
+            "core_iteration": self.context.iteration,
+            "remaining_cycles": remaining_cycles,
+        }));
+    }
+
+    pub fn schedule_target_stall(&self, core: usize, pc: usize, remaining_cycles: u64) {
+        self.write(json!({
+            "event": "diagnostic_target_stall",
+            "cycle": self.context.cycle,
+            "core": core,
+            "pc": pc,
+            "core_iteration": self.context.iteration,
+            "remaining_cycles": remaining_cycles,
+        }));
+    }
+
+    pub fn schedule_config(
+        &self,
+        policy: &'static str,
+        seed: u64,
+        target: Option<Value>,
+        deferral_budget: u64,
+        fixed_stall: Option<(usize, u64)>,
+        fixed_target_stall: Option<(usize, usize, Option<u32>, u64)>,
+    ) {
+        self.write(json!({
+            "event": "scheduler_config",
+            "schedule_policy": policy,
+            "schedule_seed": seed,
+            "target_dependency": target,
+            "deferral_budget": deferral_budget,
+            "fixed_stall": fixed_stall.map(|(core, cycles)| json!({
+                "core": core,
+                "cycles": cycles,
+            })),
+            "fixed_target_stall": fixed_target_stall.map(|(core, pc, iteration, cycles)| json!({
+                "core": core,
+                "pc": pc,
+                "iteration": iteration,
+                "cycles": cycles,
+            })),
+        }));
+    }
+
+    pub fn scheduler_event(&self, event: Value) {
+        self.write(event);
+    }
+
+    fn write(&self, value: Value) {
+        self.sink.lock().unwrap().event(value);
+    }
+
+    fn ensure_global(&mut self, end: usize) {
+        if self.global.len() < end {
+            self.global.resize(end, GlobalCell::default());
+        }
+    }
+
+    fn ensure_local(&mut self, core: usize, end: usize) {
+        if let Some(memory) = self.local.get_mut(core)
+            && memory.len() < end
+        {
+            memory.resize(end, Provenance::Uninitialized);
+        }
+    }
+
+    fn local_tags(&mut self, core: usize, address: usize, size: usize) -> Vec<Provenance> {
+        let Some(end) = address.checked_add(size) else {
+            return vec![Provenance::Unknown; size];
+        };
+        self.ensure_local(core, end);
+        self.local[core][address..end].to_vec()
+    }
+
+    fn store_local_tags(&mut self, core: usize, address: usize, tags: &[Provenance]) {
+        let Some(end) = address.checked_add(tags.len()) else {
+            return;
+        };
+        self.ensure_local(core, end);
+        self.local[core][address..end].copy_from_slice(tags);
+    }
+
+    fn unique_versions(cells: &[GlobalCell]) -> Vec<u64> {
+        cells
+            .iter()
+            .filter_map(|cell| cell.writer.map(|writer| writer.version))
+            .collect::<BTreeSet<_>>()
+            .into_iter()
+            .collect()
+    }
+
+    fn unique_writers(cells: &[GlobalCell]) -> Vec<Writer> {
+        cells
+            .iter()
+            .filter_map(|cell| cell.writer)
+            .collect::<BTreeSet<_>>()
+            .into_iter()
+            .collect()
+    }
+
+    fn writer_json(writer: Writer) -> Value {
+        json!({
+            "version": writer.version,
+            "cycle": writer.cycle,
+            "core": writer.core,
+            "pc": writer.pc,
+            "core_iteration": writer.iteration,
+            "provenance": writer.provenance.samples(),
+            "provenance_state": writer.provenance.state(),
+        })
+    }
+
+    pub fn input_store(&mut self, address: usize, size: usize, sample: u32) {
+        let Some(end) = address.checked_add(size) else {
+            return;
+        };
+        self.ensure_global(end);
+        let provenance = Provenance::sample(sample);
+        self.next_version += 1;
+        let writer = Writer {
+            version: self.next_version,
+            cycle: self.context.cycle,
+            core: 0,
+            pc: 0,
+            iteration: sample,
+            provenance,
+        };
+        let overwritten_versions = Self::unique_versions(&self.global[address..end]);
+        for cell in &mut self.global[address..end] {
+            *cell = GlobalCell {
+                provenance,
+                writer: Some(writer),
+            };
+        }
+        let mut event = json!({
+            "event": "external_input_store",
+            "cycle": self.context.cycle,
+            "core": 0,
+            "pc": 0,
+            "core_iteration": sample,
+            "address": address,
+            "size": size,
+            "sample": sample,
+            "version": writer.version,
+            "overwritten_versions": overwritten_versions,
+        });
+        if let Some(object) = event.as_object_mut() {
+            object.extend(provenance.json().as_object().unwrap().clone());
+        }
+        self.write(event);
+    }
+
+    pub fn global_store_from_local(
+        &mut self,
+        core: usize,
+        global_address: usize,
+        local_address: usize,
+        size: usize,
+    ) {
+        let Some(end) = global_address.checked_add(size) else {
+            return;
+        };
+        let tags = self.local_tags(core, local_address, size);
+        self.ensure_global(end);
+        self.next_version += 1;
+        let provenance = Provenance::merge_all(tags.iter().copied());
+        let writer = Writer {
+            version: self.next_version,
+            cycle: self.context.cycle,
+            core,
+            pc: self.context.pc,
+            iteration: self.context.iteration,
+            provenance,
+        };
+        let overwritten_versions = Self::unique_versions(&self.global[global_address..end]);
+        let overwritten_writers = Self::unique_writers(&self.global[global_address..end]);
+        for (cell, tag) in self.global[global_address..end].iter_mut().zip(tags) {
+            *cell = GlobalCell {
+                provenance: tag,
+                writer: Some(writer),
+            };
+        }
+        let mut event = json!({
+            "event": "global_store",
+            "cycle": self.context.cycle,
+            "core": core,
+            "pc": self.context.pc,
+            "core_iteration": self.context.iteration,
+            "address": global_address,
+            "local_address": local_address,
+            "size": size,
+            "version": writer.version,
+            "overwritten_versions": overwritten_versions,
+            "overwritten_writers": overwritten_writers.into_iter().map(Self::writer_json).collect::<Vec<_>>(),
+        });
+        if let Some(object) = event.as_object_mut() {
+            object.extend(provenance.json().as_object().unwrap().clone());
+        }
+        self.write(event);
+    }
+
+    pub fn global_load_to_local(
+        &mut self,
+        core: usize,
+        global_address: usize,
+        local_address: usize,
+        size: usize,
+    ) {
+        let Some(end) = global_address.checked_add(size) else {
+            return;
+        };
+        self.ensure_global(end);
+        let cells = self.global[global_address..end].to_vec();
+        let tags: Vec<_> = cells.iter().map(|cell| cell.provenance).collect();
+        let provenance = Provenance::merge_all(tags.iter().copied());
+        let writers = Self::unique_writers(&cells);
+        let versions = Self::unique_versions(&cells);
+        self.store_local_tags(core, local_address, &tags);
+        let mut event = json!({
+            "event": "global_load",
+            "cycle": self.context.cycle,
+            "core": core,
+            "pc": self.context.pc,
+            "core_iteration": self.context.iteration,
+            "address": global_address,
+            "local_address": local_address,
+            "size": size,
+            "versions": versions,
+            "last_writers": writers.into_iter().map(Self::writer_json).collect::<Vec<_>>(),
+        });
+        if let Some(object) = event.as_object_mut() {
+            object.extend(provenance.json().as_object().unwrap().clone());
+        }
+        self.write(event);
+    }
+
+    pub fn local_copy(
+        &mut self,
+        core: usize,
+        destination: usize,
+        source: usize,
+        size: usize,
+        operation: &'static str,
+    ) {
+        let tags = self.local_tags(core, source, size);
+        let provenance = Provenance::merge_all(tags.iter().copied());
+        self.store_local_tags(core, destination, &tags);
+        self.local_event(
+            operation,
+            core,
+            destination,
+            size,
+            provenance,
+            &[provenance],
+        );
+    }
+
+    pub fn local_strided_copy(
+        &mut self,
+        core: usize,
+        destination: usize,
+        source: usize,
+        element_size: usize,
+        stride: usize,
+        element_count: usize,
+        operation: &'static str,
+    ) {
+        let mut tags = Vec::with_capacity(element_size.saturating_mul(element_count));
+        for index in 0..element_count {
+            let address =
+                source.saturating_add(index.saturating_mul(stride).saturating_mul(element_size));
+            tags.extend(self.local_tags(core, address, element_size));
+        }
+        let provenance = Provenance::merge_all(tags.iter().copied());
+        self.store_local_tags(core, destination, &tags);
+        self.local_event(
+            operation,
+            core,
+            destination,
+            tags.len(),
+            provenance,
+            &[provenance],
+        );
+    }
+
+    pub fn local_transform(
+        &mut self,
+        core: usize,
+        destination: usize,
+        sources: &[(usize, usize)],
+        output_size: usize,
+        operation: &'static str,
+    ) {
+        let source_tags: Vec<Vec<_>> = sources
+            .iter()
+            .map(|&(address, size)| self.local_tags(core, address, size))
+            .collect();
+        let mut output = Vec::with_capacity(output_size);
+        for index in 0..output_size {
+            let provenance = Provenance::merge_all(
+                source_tags
+                    .iter()
+                    .filter_map(|tags| tags.get(index).copied()),
+            );
+            output.push(provenance);
+        }
+        let provenance = Provenance::merge_all(output.iter().copied());
+        self.store_local_tags(core, destination, &output);
+        let operands: Vec<_> = source_tags
+            .iter()
+            .map(|tags| Provenance::merge_all(tags.iter().copied()))
+            .collect();
+        self.local_event(
+            operation,
+            core,
+            destination,
+            output_size,
+            provenance,
+            &operands,
+        );
+    }
+
+    pub fn local_broadcast_transform(
+        &mut self,
+        core: usize,
+        destination: usize,
+        source: usize,
+        source_size: usize,
+        output_size: usize,
+        operation: &'static str,
+    ) {
+        let tags = self.local_tags(core, source, source_size);
+        let provenance = Provenance::merge_all(tags.iter().copied());
+        self.store_local_tags(core, destination, &vec![provenance; output_size]);
+        self.local_event(
+            operation,
+            core,
+            destination,
+            output_size,
+            provenance,
+            &[provenance],
+        );
+    }
+
+    pub fn local_mvm_transform(
+        &mut self,
+        core: usize,
+        destination: usize,
+        source: usize,
+        element_size: usize,
+        output_size: usize,
+        used_rows: &[bool],
+        operation: &'static str,
+    ) {
+        let mut provenance = Provenance::Uninitialized;
+        for (row, used) in used_rows.iter().copied().enumerate() {
+            if used {
+                provenance = provenance.merge(Provenance::merge_all(self.local_tags(
+                    core,
+                    source + row * element_size,
+                    element_size,
+                )));
+            }
+        }
+        self.store_local_tags(core, destination, &vec![provenance; output_size]);
+        self.local_event(
+            operation,
+            core,
+            destination,
+            output_size,
+            provenance,
+            &[provenance],
+        );
+    }
+
+    pub fn send_transfer(
+        &mut self,
+        sender: usize,
+        receiver: usize,
+        source: usize,
+        destination: usize,
+        size: usize,
+    ) {
+        let tags = self.local_tags(sender, source, size);
+        let provenance = Provenance::merge_all(tags.iter().copied());
+        self.store_local_tags(receiver, destination, &tags);
+        let mut event = json!({
+            "event": "send_recv_transfer",
+            "cycle": self.context.cycle,
+            "pc": self.context.pc,
+            "core_iteration": self.context.iteration,
+            "sender_core": sender,
+            "receiver_core": receiver,
+            "source_address": source,
+            "destination_address": destination,
+            "size": size,
+        });
+        if let Some(object) = event.as_object_mut() {
+            object.extend(provenance.json().as_object().unwrap().clone());
+        }
+        self.write(event);
+    }
+
+    fn local_event(
+        &self,
+        operation: &'static str,
+        core: usize,
+        destination: usize,
+        size: usize,
+        provenance: Provenance,
+        operands: &[Provenance],
+    ) {
+        let mut event = json!({
+            "event": "local_compute",
+            "operation": operation,
+            "cycle": self.context.cycle,
+            "core": core,
+            "pc": self.context.pc,
+            "core_iteration": self.context.iteration,
+            "destination_address": destination,
+            "size": size,
+            "operand_provenance": operands.iter().map(|tag| tag.json()).collect::<Vec<_>>(),
+        });
+        if let Some(object) = event.as_object_mut() {
+            object.extend(provenance.json().as_object().unwrap().clone());
+        }
+        self.write(event);
+        if provenance.is_mixed() {
+            self.write(json!({
+                "event": "cross_sample_data_mix",
+                "cycle": self.context.cycle,
+                "core": core,
+                "pc": self.context.pc,
+                "core_iteration": self.context.iteration,
+                "operation": operation,
+                "destination_address": destination,
+                "size": size,
+                "operand_provenance": operands.iter().map(|tag| tag.json()).collect::<Vec<_>>(),
+                "provenance": provenance.samples(),
+            }));
+        }
+    }
+}

backend-simulators/pim/pim-simulator/src/lib/send_recv.rs

@@ -73,18 +73,27 @@ where
                 let data = inst.data;
                 TRACER.lock().unwrap().pre_recv(cpu, data);
             }
-            let [sender_core, receiver_core] =
-                cpu.get_multiple_cores([sender.internal_core, receiver.internal_core]);
-            let memory = sender_core
-                .load::<u8>(sender.address, sender.size)
-                .with_context(|| {
-                    format!(
-                        "Sender crashed while transferring memory from {} with size {}",
-                        sender.address, sender.size
-                    )
-                })
-                .unwrap();
-            receiver_core.execute_store(receiver.address, memory[0]);
+            {
+                let [sender_core, receiver_core] =
+                    cpu.get_multiple_cores([sender.internal_core, receiver.internal_core]);
+                let memory = sender_core
+                    .load::<u8>(sender.address, sender.size)
+                    .with_context(|| {
+                        format!(
+                            "Sender crashed while transferring memory from {} with size {}",
+                            sender.address, sender.size
+                        )
+                    })
+                    .unwrap();
+                receiver_core.execute_store(receiver.address, memory[0]);
+            }
+            cpu.provenance_send_transfer(
+                sender.internal_core,
+                receiver.internal_core,
+                sender.address,
+                receiver.address,
+                sender.size,
+            );
             {
                 let sender = &mut core_instructions[sender.internal_core];
                 let pc = sender.program_counter;

validation/tools/pim/pimcomp/compare/pimcomp_architecture_sync_evidence.json

@@ -0,0 +1,24 @@
+{
+  "schema": 1,
+  "status": "documentary_evidence_plus_repository_checks",
+  "architectures": {
+    "arch-a": {
+      "pimcomp_identity": "ISAAC-like static/deterministic timing model",
+      "primary_classification": "STATIC_TIMING_CONTRACT_MAPPING_UNPROVEN",
+      "config": "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json",
+      "hardware_reference": "ISAAC (HPCA 2016), documentary mapping requires review of the cited paper/configuration."
+    },
+    "arch-b": {
+      "pimcomp_identity": "PUMA-like architecture",
+      "primary_classification": "HARDWARE_SYNC_EXISTS_BUT_NOT_MODELED_BY_PIMSIM_NN",
+      "config": "validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json",
+      "hardware_reference": "PUMA, documentary valid/count synchronization is not encoded in ordinary PIMCOMP LD/ST."
+    },
+    "arch-c": {
+      "pimcomp_identity": "ISSCC 2023 ReRAM architecture row",
+      "primary_classification": "MAPPING_NOT_ESTABLISHED",
+      "config": "validation/pimsim_configs/pimcomp/arch-c/throughput_config_1000ms.json",
+      "hardware_reference": "ISSCC 2023 ReRAM reference; cross-system mapping is unresolved."
+    }
+  }
+}

validation/tools/pim/pimcomp/compare/test_PIMCOMP_architecture_sync_contract.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+"""Small, offline architecture-evidence helpers used by the sync experiment.
+
+The adversarial experiment owns the dynamic investigation.  This module keeps
+the stable repository/configuration facts it needs in one place and exposes a
+deliberately small API so the experiment can also be imported as a library.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import subprocess
+from pathlib import Path
+from typing import Any
+
+
+REPO = Path(__file__).resolve().parents[5]
+VALIDATION = REPO / "validation"
+PIMCOMP_ROOT = REPO / "third_party/PIMCOMP-NN"
+PIMSIM_ROOT = REPO / "backend-simulators/pim/pimsim-nn"
+RUST_ROOT = REPO / "backend-simulators/pim/pim-simulator"
+EVIDENCE_PATH = Path(__file__).with_name("pimcomp_architecture_sync_evidence.json")
+
+
+def _default_evidence() -> dict[str, Any]:
+    return {
+        "schema": 1,
+        "status": "documentary_evidence_plus_repository_checks",
+        "architectures": {
+            "arch-a": {
+                "pimcomp_identity": "ISAAC-like static/deterministic timing model",
+                "primary_classification": "STATIC_TIMING_CONTRACT_MAPPING_UNPROVEN",
+                "config": "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json",
+                "hardware_reference": "ISAAC (HPCA 2016), documentary mapping requires review of the cited paper/configuration.",
+            },
+            "arch-b": {
+                "pimcomp_identity": "PUMA-like architecture",
+                "primary_classification": "HARDWARE_SYNC_EXISTS_BUT_NOT_MODELED_BY_PIMSIM_NN",
+                "config": "validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json",
+                "hardware_reference": "PUMA, documentary valid/count synchronization is not encoded in ordinary PIMCOMP LD/ST.",
+            },
+            "arch-c": {
+                "pimcomp_identity": "ISSCC 2023 ReRAM architecture row",
+                "primary_classification": "MAPPING_NOT_ESTABLISHED",
+                "config": "validation/pimsim_configs/pimcomp/arch-c/throughput_config_1000ms.json",
+                "hardware_reference": "ISSCC 2023 ReRAM reference; cross-system mapping is unresolved.",
+            },
+        },
+    }
+
+
+def load_evidence() -> dict[str, Any]:
+    if EVIDENCE_PATH.is_file():
+        return json.loads(EVIDENCE_PATH.read_text(encoding="utf-8"))
+    return _default_evidence()
+
+
+def source_contract() -> dict[str, Any]:
+    """Return the contract claims used for report labeling.
+
+    These labels are intentionally conservative: they are not a substitute
+    for a paper citation and never turn an unordered relation into a safe one.
+    """
+
+    return {
+        "ld_st": "ordinary timed global-memory accesses in the Rust model",
+        "send_recv": "explicit point-to-point synchronization modeled by the simulator",
+        "wait_sync": "instruction-level synchronization when emitted",
+        "valid_count": "not encoded by PIMCOMP exported LD/ST instructions",
+        "static_timing": "not proven as a PIMCOMP-to-ISAAC contract",
+    }
+
+
+def classify_contract(
+    architecture: str, _contract: dict[str, Any], manifest: dict[str, Any]
+) -> str:
+    return manifest["architectures"][architecture]["primary_classification"]
+
+
+def architecture_source_evidence(architecture: str, manifest: dict[str, Any]) -> dict[str, Any]:
+    return dict(manifest["architectures"][architecture])
+
+
+def git_identity(path: Path) -> dict[str, Any]:
+    result: dict[str, Any] = {"path": str(path), "commit": None, "worktree_status": []}
+    if not path.exists():
+        result["error"] = "missing"
+        return result
+    try:
+        result["commit"] = subprocess.run(
+            ["git", "-C", str(path), "rev-parse", "HEAD"],
+            check=True,
+            capture_output=True,
+            text=True,
+        ).stdout.strip()
+        status = subprocess.run(
+            ["git", "-C", str(path), "status", "--short"],
+            check=True,
+            capture_output=True,
+            text=True,
+        ).stdout.splitlines()
+        result["worktree_status"] = status
+    except (OSError, subprocess.CalledProcessError) as exc:
+        result["error"] = f"{type(exc).__name__}: {exc}"
+    return result
+
+
+def tree_hash(root: Path, pattern: str) -> dict[str, str]:
+    values: dict[str, str] = {}
+    for path in sorted(root.glob(pattern)):
+        if not path.is_file():
+            continue
+        digest = hashlib.sha256(path.read_bytes()).hexdigest()
+        values[str(path.relative_to(root))] = digest
+    return values
+
+
+def instruction_graph(artifact: Path) -> tuple[dict[tuple[int, int], list[tuple[int, int]]], dict[str, Any]]:
+    """Build same-core program-order edges from the exported JSON streams.
+
+    Cross-core synchronization is added only when an artifact explicitly
+    carries matching SEND/RECV metadata.  Ordinary global LD/ST creates no
+    edge by construction; that is the relation this experiment is testing.
+    """
+
+    graph: dict[tuple[int, int], list[tuple[int, int]]] = {}
+    streams: dict[int, list[dict[str, Any]]] = {}
+    for path in sorted(artifact.glob("core_*.json"), key=lambda item: int(item.stem.split("_")[1])):
+        try:
+            instructions = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            continue
+        core = int(path.stem.split("_")[1]) + 1
+        streams[core] = instructions if isinstance(instructions, list) else []
+        for index in range(len(streams[core]) - 1):
+            graph.setdefault((core, index), []).append((core, index + 1))
+
+    sends: dict[tuple[int, int], tuple[int, int]] = {}
+    recvs: dict[tuple[int, int], tuple[int, int]] = {}
+    for core, instructions in streams.items():
+        for index, instruction in enumerate(instructions):
+            op = str(instruction.get("op", instruction.get("operation", ""))).lower()
+            peer = instruction.get("core")
+            if peer is None:
+                continue
+            key = (core, int(peer) + 1)
+            if op == "send":
+                sends.setdefault(key, (core, index))
+            elif op == "recv":
+                recvs.setdefault(key, (core, index))
+    for key, send in sends.items():
+        recv = recvs.get((key[1], key[0]))
+        if recv is not None:
+            graph.setdefault(send, []).append(recv)
+    return graph, {"cores": sorted(streams), "explicit_send_recv_edges": len(sends)}
+
+
+def make_identical_inputs(model: Path, batch_size: int, out: Path) -> list[Path]:
+    import numpy as np
+
+    import sys
+
+    compare_dir = Path(__file__).resolve().parent
+    if str(compare_dir) not in sys.path:
+        sys.path.insert(0, str(compare_dir))
+    import compare_raptor_pimcomp_model as compare  # noqa: PLC0415
+
+    inputs, _ = compare.onnx_io(model)
+    arrays = []
+    for _index, _name, element_type, shape in inputs:
+        dtype = compare._ONNX_TO_NP[element_type]
+        arrays.append(np.full(shape, 1.0, dtype=dtype))
+    flattened = np.concatenate(
+        [compare.flatten_pimcomp_input(array) for array in arrays]
+    ) if arrays else np.empty(0, dtype=np.float32)
+    samples = [[flattened.copy()] for _ in range(batch_size)]
+    return [
+        Path(path)
+        for path in compare.write_input_batch_binaries(
+            samples, out / "inputs/pimcomp_isolated"
+        )
+    ]
+
+
+if __name__ == "__main__":
+    print(json.dumps(load_evidence(), indent=2, sort_keys=True))

validation/tools/pim/pimcomp/compare/test_PIMCOMP_global_memory_sync.py

@@ -0,0 +1,479 @@
+#!/usr/bin/env python3
+"""Reusable build, execution, and provenance helpers for PIMCOMP audits."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+from bisect import bisect_left
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+
+REPO = Path(__file__).resolve().parents[5]
+VALIDATION = REPO / "validation"
+CONFIG_ROOT = VALIDATION / "pimsim_configs/pimcomp"
+PIMCOMP_ROOT = REPO / "third_party/PIMCOMP-NN"
+PIMSIM_NN_ROOT = REPO / "backend-simulators/pim/pimsim-nn"
+RUST_ROOT = REPO / "backend-simulators/pim/pim-simulator"
+RUST_BINARY = RUST_ROOT / "target/release/pim-simulator"
+COMPARE_SCRIPT = Path(__file__).with_name("compare_raptor_pimcomp_model.py")
+PYTHON = REPO / ".venv/bin/python"
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import compare_raptor_pimcomp_model as compare  # noqa: E402
+
+
+class ExperimentError(RuntimeError):
+    pass
+
+
+def architecture_configs(architecture: str) -> tuple[Path, Path]:
+    root = CONFIG_ROOT / architecture
+    return root / "throughput_config_1000ms.json", root / "latency_config.json"
+
+
+def check_prerequisites(throughput: Path, latency: Path) -> None:
+    required = {
+        "PIMCOMP backend": PIMCOMP_ROOT / "build/PIMCOMP-NN",
+        "PIMCOMP frontend": PIMCOMP_ROOT / "frontend/frontend.py",
+        "Raptor compiler": REPO / "build_release/Release/bin/onnx-mlir",
+        "Rust simulator source": RUST_ROOT,
+        "pimsim-nn build": PIMSIM_NN_ROOT / "build",
+        "throughput config": throughput,
+        "latency config": latency,
+    }
+    missing = [f"{label}: {path}" for label, path in required.items() if not path.exists()]
+    if missing:
+        raise ExperimentError("missing prerequisites:\n" + "\n".join(missing))
+
+
+def _run(cmd: list[str], cwd: Path, log: Path, timeout: float = 0.0) -> subprocess.CompletedProcess[str]:
+    log.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        result = subprocess.run(
+            [str(value) for value in cmd],
+            cwd=cwd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            timeout=None if timeout <= 0 else timeout,
+        )
+    except subprocess.TimeoutExpired as exc:
+        log.write_text((exc.stdout or "") + "\nTIMEOUT\n", encoding="utf-8")
+        raise ExperimentError(f"command timed out: {' '.join(map(str, cmd))}") from exc
+    log.write_text(result.stdout, encoding="utf-8")
+    if result.returncode:
+        raise ExperimentError(
+            f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n"
+            f"see {log}\n{result.stdout[-3000:]}"
+        )
+    return result
+
+
+def build_simulator(args: Any, out: Path) -> None:
+    _run(
+        [
+            "cargo", "build", "--release", "--no-default-features",
+            "--package", "pim-simulator", "--bin", "pim-simulator",
+        ],
+        RUST_ROOT,
+        out / "cargo_build.log",
+        float(getattr(args, "timeout", 0.0)),
+    )
+    if not RUST_BINARY.is_file():
+        raise ExperimentError(f"Rust simulator binary was not produced: {RUST_BINARY}")
+
+
+def make_model(path: Path) -> None:
+    import onnx
+    from onnx import TensorProto, helper, numpy_helper
+
+    shape = [1, 64, 8, 8]
+    weights = []
+    for name in ("w0", "w1"):
+        weight = np.zeros((64, 64, 3, 3), dtype=np.float32)
+        for channel in range(64):
+            weight[channel, channel, 1, 1] = 1.0
+        weights.append(numpy_helper.from_array(weight, name=name))
+    input_value = helper.make_tensor_value_info("input", TensorProto.FLOAT, shape)
+    output_value = helper.make_tensor_value_info("output", TensorProto.FLOAT, shape)
+    nodes = [
+        helper.make_node(
+            "Conv", ["input", "w0"], ["hidden"], name="conv0",
+            kernel_shape=[3, 3], strides=[1, 1], pads=[1, 1, 1, 1],
+            dilations=[1, 1], group=1,
+        ),
+        helper.make_node(
+            "Conv", ["hidden", "w1"], ["output"], name="conv1",
+            kernel_shape=[3, 3], strides=[1, 1], pads=[1, 1, 1, 1],
+            dilations=[1, 1], group=1,
+        ),
+    ]
+    graph = helper.make_graph(nodes, "pimcomp_sync_two_conv", [input_value], [output_value], weights)
+    model = helper.make_model(graph, opset_imports=[helper.make_operatorsetid("", 13)])
+    model.ir_version = min(model.ir_version, 8)
+    onnx.checker.check_model(model)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    onnx.save(model, path)
+
+
+def make_inputs(
+    model: Path, batch_size: int, seed: int, out: Path
+) -> tuple[list[list[np.ndarray]], list[Path], list[Path], list[np.ndarray]]:
+    del seed  # deterministic values are intentional; the seed remains in the report.
+    inputs_desc, _ = compare.onnx_io(model)
+    if len(inputs_desc) != 1:
+        raise ExperimentError("the synchronization model must have exactly one input")
+    _index, _name, element_type, shape = inputs_desc[0]
+    dtype = compare._ONNX_TO_NP[element_type]
+    arrays: list[np.ndarray] = [
+        np.full(shape, float(10 ** index), dtype=dtype) for index in range(max(8, batch_size))
+    ]
+    input_batch = [[array] for array in arrays]
+    raptor_paths = [
+        Path(path)
+        for path in compare.write_input_batch_binaries(input_batch, out / "inputs/raptor")
+    ]
+    pimcomp_batch = [[compare.flatten_pimcomp_input(array)] for array in arrays]
+    pimcomp_paths = [
+        Path(path)
+        for path in compare.write_input_batch_binaries(pimcomp_batch, out / "inputs/pimcomp")
+    ]
+    return input_batch, raptor_paths, pimcomp_paths, arrays
+
+
+def make_references(
+    model: Path, input_batch: list[list[np.ndarray]], architecture_out: Path, _args: Any
+) -> list[Path]:
+    import onnxruntime as ort
+
+    input_desc, output_desc = compare.onnx_io(model)
+    session = ort.InferenceSession(str(model), providers=["CPUExecutionProvider"])
+    references: list[Path] = []
+    for index, sample in enumerate(input_batch):
+        values = {input_desc[item][1]: sample[item] for item in range(len(input_desc))}
+        outputs = session.run(None, values)
+        directory = architecture_out / "reference" / f"iteration_{index:06d}"
+        directory.mkdir(parents=True, exist_ok=True)
+        for output, descriptor in zip(outputs, output_desc):
+            output_index, name, _dtype, _shape = descriptor
+            filename = f"output{output_index}_{compare.sanitize_output_name(name)}.csv"
+            np.savetxt(directory / filename, np.asarray(output).reshape(-1), delimiter=",")
+        references.append(directory)
+    return references
+
+
+def compile_artifact(args: Any, model: Path, architecture_out: Path, throughput_config: Path) -> dict[str, Any]:
+    comparison = architecture_out / "comparison"
+    command = [
+        str(PYTHON), str(COMPARE_SCRIPT),
+        "--model", str(model),
+        "--out-dir", str(comparison),
+        "--common-dir", str(architecture_out / "common"),
+        "--pimcomp-config", str(throughput_config),
+        "--pimsim-mode", "throughput",
+        "--pimsim-time-ms", "1000",
+        "--batch-size", str(args.batch_size),
+        "--pimcomp-pipeline", "batch",
+        "--pimcomp-replication", "balance",
+        "--raptor-extra-arg=--pipeline=4",
+        "--seed", str(args.seed),
+        "--timeout-seconds", str(args.timeout),
+    ]
+    if args.no_fast:
+        command.append("--no-fast")
+    log = architecture_out / "comparison_compile.log"
+    try:
+        _run(command, REPO, log, args.timeout)
+    except ExperimentError as exc:
+        report = comparison / "pimcomp/comparison_report.json"
+        if not report.is_file():
+            raise
+        report_data = json.loads(report.read_text(encoding="utf-8"))
+        raptor_error = "; ".join(
+            str(item.get("error", ""))
+            for item in report_data.get("failures", [])
+            if "RAPTOR" in str(item.get("stage", "")).upper()
+        ) or str(exc)
+    else:
+        report = comparison / "pimcomp/comparison_report.json"
+        report_data = json.loads(report.read_text(encoding="utf-8"))
+        raptor_error = "; ".join(
+            str(item.get("error", ""))
+            for item in report_data.get("failures", [])
+            if "RAPTOR" in str(item.get("stage", "")).upper()
+        ) or None
+
+    paths = report_data.get("paths", {})
+    pimcomp = Path(paths["pimcomp_exported_pim"]) if paths.get("pimcomp_exported_pim") else comparison / "pimcomp/exported"
+    pimsim = Path(paths["pimcomp_pimsim_nn"]) if paths.get("pimcomp_pimsim_nn") else comparison / "pimcomp/pimsim_nn"
+    raptor = Path(paths["raptor_pim"]) if paths.get("raptor_pim") else comparison / "raptor/pim.missing"
+    if not pimcomp.is_dir():
+        raise ExperimentError(f"PIMCOMP Rust artifact missing; see {report}")
+    return {
+        "artifact": pimcomp,
+        "pimsim_artifact": pimsim if pimsim.is_dir() else None,
+        "raptor_artifact": raptor if raptor.is_dir() else Path(),
+        "raptor_error": raptor_error,
+        "comparison_report": report,
+    }
+
+
+def _instruction_op(instruction: dict[str, Any]) -> str:
+    return str(instruction.get("op", instruction.get("operation", ""))).lower()
+
+
+def _address(instruction: dict[str, Any], registers: dict[int, int], register: str) -> int | None:
+    try:
+        base = int(registers[int(instruction[register])])
+    except (KeyError, TypeError, ValueError):
+        return None
+    offset = instruction.get("offset") or {}
+    select = int(offset.get("offset_select", 0))
+    value = int(offset.get("offset_value", 0))
+    # LD's global operand is r1 (selector bit 2); ST's global operand is rd
+    # (selector bit 1).  The local simulator uses the same asymmetric ISA.
+    selector_bit = 1 if register == "rd" else 2
+    return base + value if select & selector_bit else base
+
+
+def _static_instruction(
+    core_file_index: int,
+    artifact_format: str,
+    instruction_index: int,
+    instruction: dict[str, Any],
+    address: int,
+    size: int,
+    artifact: Path,
+) -> dict[str, Any]:
+    core = core_file_index + 1
+    core_file = f"core_{core_file_index}.json"
+    binary_file = f"core_{core_file_index}.pim"
+    return {
+        "core": core,
+        # JSON PIMCOMP streams enter the Rust executor after an initial
+        # synthetic slot; Raptor's emitted binary/JSON streams do not.
+        "pc": instruction_index if artifact_format == "binary+json" else instruction_index - 1,
+        "artifact_pc": instruction_index,
+        "address": address,
+        "size": size,
+        "instruction_file": core_file,
+        "execution_file": binary_file if (artifact / binary_file).is_file() else core_file,
+        "artifact_format": artifact_format,
+    }
+
+
+def analyze_artifact(artifact: Path) -> dict[str, Any]:
+    core_paths = sorted(artifact.glob("core_*.json"), key=lambda path: int(path.stem.split("_")[1]))
+    if not core_paths:
+        raise ExperimentError(f"artifact has no core_*.json files: {artifact}")
+    artifact_format = "binary+json" if any(artifact.glob("core_*.pim")) else "json"
+    stores: list[dict[str, Any]] = []
+    loads: list[dict[str, Any]] = []
+    counts: dict[str, int] = defaultdict(int)
+    instruction_files: dict[str, str] = {}
+    participating = 0
+    for path in core_paths:
+        core_file_index = int(path.stem.split("_")[1])
+        instructions = json.loads(path.read_text(encoding="utf-8"))
+        if instructions:
+            participating += 1
+        instruction_files[f"core_{core_file_index}"] = str(path)
+        registers: dict[int, int] = {}
+        for index, instruction in enumerate(instructions):
+            op = _instruction_op(instruction)
+            counts[op] += 1
+            if op in {"sldi", "lldi"} and "rd" in instruction and "imm" in instruction:
+                registers[int(instruction["rd"])] = int(instruction["imm"])
+                continue
+            if op not in {"ld", "st"}:
+                continue
+            address = _address(instruction, registers, "rd" if op == "st" else "rs1")
+            if address is None:
+                continue
+            size = int(instruction.get("size", instruction.get("len", 0)))
+            if size <= 0:
+                continue
+            item = _static_instruction(
+                core_file_index, artifact_format, index, instruction, address, size, artifact
+            )
+            (stores if op == "st" else loads).append(item)
+
+    loads_by_address = sorted(loads, key=lambda item: int(item["address"]))
+    starts = [int(item["address"]) for item in loads_by_address]
+    dependencies: list[dict[str, Any]] = []
+    for store in stores:
+        begin = int(store["address"])
+        end = begin + int(store["size"])
+        first = bisect_left(starts, end)
+        for load in loads_by_address[:first]:
+            if load["core"] == store["core"]:
+                continue
+            load_begin = int(load["address"])
+            load_end = load_begin + int(load["size"])
+            if load_end <= begin:
+                continue
+            dependencies.append({
+                "overlap": {
+                    "address_begin": max(begin, load_begin),
+                    "address_end": min(end, load_end),
+                },
+                "writer": dict(store),
+                "reader": dict(load),
+                "explicit_sync_ordering_evidence": False,
+            })
+    dependencies.sort(key=lambda item: (
+        item["overlap"]["address_begin"], item["writer"]["core"],
+        item["writer"]["pc"], item["reader"]["core"], item["reader"]["pc"],
+    ))
+    return {
+        "artifact_format": artifact_format,
+        "instruction_files": instruction_files,
+        "stores": stores,
+        "loads": loads,
+        "cross_core_dependencies": dependencies,
+        "cross_core_dependency_count": len(dependencies),
+        "participating_core_count": participating,
+        "instruction_counts": dict(sorted(counts.items())),
+        "representative_dependency": dependencies[0] if dependencies else None,
+    }
+
+
+def trace_events(path: Path) -> list[dict[str, Any]]:
+    if not path.is_file():
+        return []
+    events = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            events.append(json.loads(line))
+    return events
+
+
+def _prov(event: dict[str, Any]) -> tuple[int, ...]:
+    return tuple(int(value) for value in event.get("provenance", []))
+
+
+def _overlap(a: dict[str, Any], b: dict[str, Any]) -> bool:
+    return max(int(a["address"]), int(b["address"])) < min(
+        int(a["address"]) + int(a["size"]), int(b["address"]) + int(b["size"])
+    )
+
+
+def dynamic_analysis(events: list[dict[str, Any]], static: dict[str, Any], config: Path) -> dict[str, Any]:
+    stores = [event for event in events if event.get("event") == "global_store"]
+    loads = [event for event in events if event.get("event") == "global_load"]
+    input_stores = [event for event in events if event.get("event") == "external_input_store"]
+    event_keys = {
+        (int(event.get("core", -1)), int(event.get("pc", -1)), int(event.get("address", -1)), int(event.get("size", -1)))
+        for event in stores + loads
+    }
+    executed = sum(
+        1 for dependency in static["cross_core_dependencies"]
+        if (
+            int(dependency["writer"]["core"]), int(dependency["writer"]["pc"]),
+            int(dependency["writer"]["address"]), int(dependency["writer"]["size"]),
+        ) in event_keys
+        and (
+            int(dependency["reader"]["core"]), int(dependency["reader"]["pc"]),
+            int(dependency["reader"]["address"]), int(dependency["reader"]["size"]),
+        ) in event_keys
+    )
+    cross_links = []
+    for load in loads:
+        for writer in load.get("last_writers", []):
+            if int(writer.get("core", -1)) == 0 or not writer.get("provenance"):
+                continue
+            cross_links.append((load, writer))
+    host_races = []
+    for load in loads:
+        expected = (int(load.get("core_iteration", -1)),)
+        if any(_overlap(load, source) and _prov(load) and _prov(load) != expected for source in input_stores):
+            host_races.append(load)
+    reused_versions = {
+        int(version)
+        for store in stores
+        for version in store.get("overwritten_versions", [])
+    }
+    mixed = [event for event in events if event.get("event") == "cross_sample_data_mix"]
+    return {
+        "executed_cross_core_dependencies": executed,
+        "sample_dependent_cross_core_links": len(cross_links),
+        "host_input_lifetime_races": len(host_races),
+        "send_recv_generation_races": 0,
+        "mixed_sample_operations": len(mixed),
+        "global_memory_versions_reused": len(reused_versions),
+        "global_memory_generation_races": 0,
+        "host_input_race_contaminates_test": bool(host_races),
+        "config": str(config),
+    }
+
+
+def run_rust(
+    artifact: Path,
+    inputs: list[Path],
+    references: list[Path],
+    outputs_desc: list[tuple[int, str, int, list[int]]],
+    out: Path,
+    args: Any,
+    *,
+    schedule_policy: str = "greedy",
+    schedule_seed: int = 0,
+    schedule_target: str | None = None,
+    schedule_deferral_budget: int | None = None,
+    target_stall: str | None = None,
+    channel_last: bool = False,
+) -> dict[str, Any]:
+    if not inputs:
+        raise ExperimentError("Rust run requires at least one input")
+    out.mkdir(parents=True, exist_ok=True)
+    output = out / "output.bin"
+    batch_outputs = out / "iterations"
+    shutil.rmtree(batch_outputs, ignore_errors=True)
+    dump = compare.build_dump_ranges(artifact / "config.json", outputs_desc)
+    mode = "latency" if len(inputs) == 1 else "throughput"
+    command = [
+        str(RUST_BINARY), "--folder", str(artifact), "--output", str(output),
+        "--dump", dump, "--mode", mode, "--batch-size", str(len(inputs)),
+        "--input-dir", str(inputs[0].parent), "--batch-output-dir", str(batch_outputs),
+        "--provenance-trace", str(out / "provenance.jsonl"),
+        "--diagnostic-schedule-policy", schedule_policy,
+        "--diagnostic-schedule-seed", str(schedule_seed),
+    ]
+    if schedule_target:
+        command += ["--diagnostic-schedule-target", schedule_target]
+    if schedule_deferral_budget is not None:
+        command += ["--diagnostic-schedule-deferral-budget", str(schedule_deferral_budget)]
+    if target_stall:
+        command += ["--diagnostic-target-stall", target_stall]
+    log = out / "simulator.log"
+    try:
+        _run(command, RUST_ROOT, log, float(args.timeout))
+    except ExperimentError as exc:
+        return {
+            "passed": False, "max_diffs": {}, "error": str(exc), "completed": False,
+            "command": [str(value) for value in command],
+        }
+    failed: list[int] = []
+    max_diffs: dict[str, float] = {}
+    for index, reference in enumerate(references[: len(inputs)]):
+        result = compare.compare_simulator_outputs(
+            batch_outputs / f"output_{index:06d}.bin", outputs_desc, reference,
+            threshold=args.threshold, rtol=args.rtol, channel_last=channel_last,
+        )
+        if not result.passed:
+            failed.append(index)
+        for name, value in result.max_diffs.items():
+            max_diffs[name] = max(max_diffs.get(name, 0.0), value)
+    return {
+        "passed": not failed, "max_diffs": max_diffs,
+        "failed_iterations": failed, "completed": True,
+        "command": [str(value) for value in command],
+        "trace": str(out / "provenance.jsonl"),
+    }