feat(host): a recency hot set for every post-copy source from the kernel's idle-page bits (ADR 0045 C2)

.github/workflows/ci.yml

@@ -1208,6 +1208,12 @@ jobs:
       # pushes past its window (it timed out under parallelism in CI). The
       # contention can't be slot-partitioned away, so this batch stays
       # serial; the dedup below is the win that doesn't risk a flake.
+      - name: Idle-page tracking and advisory drain tests (root)
+        run: |
+          sudo -E env "PATH=$PATH" cargo nextest run \
+            -p engram-host-agent -p engram-uffd-handler --lib \
+            -E 'test(page_idle::tests::anonymous_mapping_accesses_clear_idle_bits) | test(fake_capture_and_destination_poller_deliver_final_hot_trace) | test(peer_mode_drains_sealed_chunks_and_sweep_never_overwrites_them)'
+
       - name: NBD chunked-disk + netlink tests (Linux + KVM + nbd, root)
         if: steps.nbd_setup.outputs.available == 'true'
         env:

crates/engram-coordinator/src/placement.rs

@@ -1417,6 +1417,7 @@ mod tests {
                 base_shm_tmpfs: base_shm,
                 uffd_minor_shmem: uffd_minor,
                 nbd,
+                page_idle: CapStatus::Unknown,
                 bundle_stamp: bundle,
                 fc_snapshot_version: fc_snapshot_version.map(str::to_string),
                 wire_version: 7,
@@ -1450,6 +1451,19 @@ mod tests {
             assert!(host_meets_capabilities(&h, &req).is_ok());
         }
 
+        #[test]
+        fn page_idle_failure_never_gates_placement() {
+            let mut h = host(1);
+            h.capabilities = fully_ok();
+            h.capabilities.page_idle = CapStatus::Failed("unavailable".into());
+            let req = CapabilityRequirements {
+                needs_uffd_substrate: true,
+                needs_live_substrate: true,
+                fc_snapshot_version: None,
+            };
+            assert!(host_meets_capabilities(&h, &req).is_ok());
+        }
+
         /// ADR 0123: a live post-copy DESTINATION needs the substrate to
         /// be present, not merely "not applicable" — the migration
         /// restore forces UFFD. A File-backend host that honestly reports
@@ -2349,6 +2363,7 @@ mod tests {
                 base_shm_tmpfs: engram_core::types::host::CapStatus::Ok(None),
                 uffd_minor_shmem: engram_core::types::host::CapStatus::Ok(None),
                 nbd: engram_core::types::host::CapStatus::Ok(None),
+                page_idle: engram_core::types::host::CapStatus::Unknown,
                 bundle_stamp: engram_core::types::host::CapStatus::Ok(None),
                 fc_snapshot_version: Some("v9.0.0".into()),
                 wire_version: 7,

crates/engram-coordinator/tests/claim_cold_base_live_pg.rs

@@ -129,6 +129,7 @@ async fn seed_fc_host(meta: &Arc<dyn MetadataStore>, fc_version: &str) -> HostId
             base_shm_tmpfs: CapStatus::Ok(None),
             uffd_minor_shmem: CapStatus::Ok(None),
             nbd: CapStatus::Ok(None),
+            page_idle: CapStatus::Unknown,
             bundle_stamp: CapStatus::Ok(None),
             fc_snapshot_version: Some(fc_version.to_string()),
             wire_version: engram_protocol::WIRE_VERSION,

crates/engram-core/src/types/host.rs

@@ -271,6 +271,9 @@ pub struct HostCapabilities {
     /// dev fallback — this only makes a prod misconfiguration visible.
     #[serde(default)]
     pub nbd: CapStatus,
+    /// Idle-page bitmap access. Informational; never a placement requirement.
+    #[serde(default)]
+    pub page_idle: CapStatus,
     /// `bundles::read_stamp(bundle_dir)` returned a non-empty stamp —
     /// the host has *some* current RO-bundle generation staged.
     #[serde(default)]
@@ -317,6 +320,9 @@ impl HostCapabilities {
         if is_failing(&self.nbd) {
             out.push("nbd");
         }
+        if matches!(self.page_idle, CapStatus::Failed(_)) {
+            out.push("page_idle");
+        }
         if is_failing(&self.bundle_stamp) {
             out.push("bundle_stamp");
         }
@@ -772,13 +778,22 @@ mod tests {
             base_shm_tmpfs: CapStatus::NotApplicable,
             uffd_minor_shmem: CapStatus::NotApplicable,
             nbd: CapStatus::NotApplicable,
+            page_idle: CapStatus::NotApplicable,
             bundle_stamp: CapStatus::Failed("no stamp".to_string()),
             fc_snapshot_version: None,
             wire_version: 7,
         };
         assert_eq!(vz_like.failing_capabilities(), vec!["bundle_stamp"]);
     }
 
+    #[test]
+    fn page_idle_failure_is_visible_but_old_payloads_still_decode() {
+        let mut caps: HostCapabilities = serde_json::from_str("{}").unwrap();
+        assert_eq!(caps.page_idle, CapStatus::Unknown);
+        caps.page_idle = CapStatus::Failed("not configured".into());
+        assert_eq!(caps.failing_capabilities(), vec!["page_idle"]);
+    }
+
     #[test]
     fn failing_capabilities_sorted_deterministic_order() {
         let broken = HostCapabilities {
@@ -788,6 +803,7 @@ mod tests {
             base_shm_tmpfs: CapStatus::Failed("x".into()),
             uffd_minor_shmem: CapStatus::Ok(None),
             nbd: CapStatus::Unknown,
+            page_idle: CapStatus::Unknown,
             bundle_stamp: CapStatus::Ok(None),
             fc_snapshot_version: None,
             wire_version: 7,

crates/engram-core/src/types/snapshot.rs

@@ -215,10 +215,9 @@ pub struct MigrationSourceInfo {
     pub source_addr: String,
     pub memory_manifest_json: Vec<u8>,
     pub memory_manifest_ref: super::manifest::ManifestRef,
-    /// ADR 0045 C2 (E2B fold): the source guest's hot set in fault
-    /// order — the destination pulls these FIRST. Best-effort rider
-    /// (empty when the source had no trace); serde-default keeps
-    /// mixed rolls safe.
+    /// The presetup handler trace in fault order. The source retains this
+    /// first-tier order and adds idle-derived hashes to the capture-time
+    /// root seal update. Empty when the source had no handler trace.
     #[serde(default)]
     pub hot_chunks: Vec<[u8; 32]>,
     /// ADR 0045 C2: this is a POST-COPY move. The destination restores
@@ -365,6 +364,8 @@ pub enum MigrationItem {
     /// by chunk index — demand-fetched (and drained) by the dest's
     /// NBD backend straight out of the frozen source's RAM.
     DiskChunkAt(DiskRole, u64),
+    /// Optional advisory hot-set hint, fetched after the required VM state.
+    HotHint,
 }
 
 /// A frame of `migration_fetch`'s stream.

crates/engram-host-agent/src/capabilities.rs

@@ -324,6 +324,24 @@ pub async fn fc_snapshot_version(firecracker_bin: Option<&Path>) -> Option<Strin
     version
 }
 
+/// Informational on every FC host, including File-mode fresh creates.
+pub fn probe_page_idle(is_fc: bool) -> CapStatus {
+    if !is_fc {
+        return CapStatus::NotApplicable;
+    }
+    #[cfg(target_os = "linux")]
+    {
+        match crate::page_idle::open_bitmap() {
+            Ok(_) => CapStatus::Ok(None),
+            Err(error) => CapStatus::Failed(error.to_string()),
+        }
+    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        CapStatus::NotApplicable
+    }
+}
+
 /// Run every probe and assemble the vector this tick. `schema: 1`
 /// unconditionally — a `HostCapabilities::default()` (`schema: 0`)
 /// only ever comes from an old host-agent's absent field, never from
@@ -355,6 +373,7 @@ pub async fn probe_all(inputs: &ProbeInputs) -> HostCapabilities {
         base_shm_tmpfs,
         uffd_minor_shmem,
         nbd,
+        page_idle: probe_page_idle(is_fc),
         bundle_stamp,
         fc_snapshot_version,
         wire_version: engram_protocol::WIRE_VERSION,

crates/engram-host-agent/src/dirty_map.rs

@@ -31,8 +31,7 @@
 //!   server hashes the live bytes on request and demotes matches to the
 //!   class-2 fetch.
 //!
-//! Everything here is blackout-critical (the scan sits between the pause
-//! and the SEAL push) — reads are batched (4096 entries = 32 KiB per
+//! The dirty scan is blackout-critical (between pause and the SEAL push) — reads are batched (4096 entries = 32 KiB per
 //! `pread`) and the caller records `scan_ms` (R6: measure, never quote).
 
 use std::io::{self, BufRead, BufReader};
@@ -149,45 +148,60 @@ pub fn scan_dirty_chunks(
     chunk_size: u64,
     total_bytes: u64,
 ) -> io::Result<SealBitmap> {
-    use std::os::unix::fs::FileExt;
+    let mut seal = SealBitmap::new(chunk_size, total_bytes.div_ceil(chunk_size));
+    walk_pagemap_batches(pid, vmas, |base, bytes| {
+        let mut i = 0;
+        while i < bytes.len() / 8 {
+            let entry = u64::from_ne_bytes(bytes[i * 8..i * 8 + 8].try_into().unwrap());
+            let offset = base + i as u64 * PAGE_SIZE;
+            if offset < total_bytes && pagemap_entry_is_dirty(entry) {
+                let chunk = offset / chunk_size;
+                seal.set(chunk);
+                i += (((chunk + 1) * chunk_size - offset).div_ceil(PAGE_SIZE) as usize).max(1);
+            } else {
+                i += 1;
+            }
+        }
+        Ok(())
+    })?;
+    Ok(seal)
+}
+
+/// Visit each page once, in snapshot-offset order, with batched pagemap reads.
+#[cfg(target_os = "linux")]
+pub(crate) fn walk_pagemap(
+    pid: u32,
+    vmas: &[GuestVma],
+    mut visit: impl FnMut(u64, u64) -> io::Result<()>,
+) -> io::Result<()> {
+    walk_pagemap_batches(pid, vmas, |base, bytes| {
+        let (entries, _) = bytes.as_chunks::<8>();
+        for (i, entry) in entries.iter().enumerate() {
+            visit(base + i as u64 * PAGE_SIZE, u64::from_ne_bytes(*entry))?;
+        }
+        Ok(())
+    })
+}
 
-    let chunk_count = total_bytes.div_ceil(chunk_size);
-    let mut seal = SealBitmap::new(chunk_size, chunk_count);
+#[cfg(target_os = "linux")]
+fn walk_pagemap_batches(
+    pid: u32,
+    vmas: &[GuestVma],
+    mut visit: impl FnMut(u64, &[u8]) -> io::Result<()>,
+) -> io::Result<()> {
+    use std::os::unix::fs::FileExt;
     let pagemap = std::fs::File::open(format!("/proc/{pid}/pagemap"))?;
     let mut buf = vec![0u8; PAGEMAP_BATCH_ENTRIES * 8];
-
     for vma in vmas {
         let mut va = vma.start;
         while va < vma.end {
-            let pages_left = ((vma.end - va) / PAGE_SIZE) as usize;
-            let n = pages_left.min(PAGEMAP_BATCH_ENTRIES);
-            let byte_len = n * 8;
-            pagemap.read_exact_at(&mut buf[..byte_len], (va / PAGE_SIZE) * 8)?;
-
-            let mut i = 0usize;
-            while i < n {
-                let entry = u64::from_le_bytes(buf[i * 8..i * 8 + 8].try_into().unwrap());
-                if pagemap_entry_is_dirty(entry) {
-                    let offset = vma.file_offset + (va - vma.start) + (i as u64) * PAGE_SIZE;
-                    if offset < total_bytes {
-                        let chunk = offset / chunk_size;
-                        seal.set(chunk);
-                        // Any-page rule satisfied: skip the rest of this
-                        // chunk (within the batch; the next batch
-                        // re-checks cheaply via the bitmap).
-                        let chunk_end_off = (chunk + 1) * chunk_size;
-                        let pages_to_skip =
-                            (chunk_end_off.saturating_sub(offset)).div_ceil(PAGE_SIZE) as usize;
-                        i += pages_to_skip.max(1);
-                        continue;
-                    }
-                }
-                i += 1;
-            }
-            va += (n as u64) * PAGE_SIZE;
+            let n = (((vma.end - va) / PAGE_SIZE) as usize).min(PAGEMAP_BATCH_ENTRIES);
+            pagemap.read_exact_at(&mut buf[..n * 8], (va / PAGE_SIZE) * 8)?;
+            visit(vma.file_offset + va - vma.start, &buf[..n * 8])?;
+            va += n as u64 * PAGE_SIZE;
         }
     }
-    Ok(seal)
+    Ok(())
 }
 
 #[cfg(test)]

crates/engram-host-agent/src/grpc_server.rs

@@ -434,6 +434,7 @@ impl HostService for HostServiceImpl {
                     Err(_) => return Err(Status::invalid_argument("unknown disk role")),
                 };
                 match Kind::try_from(item.kind) {
+                    Ok(Kind::HotHint) => Ok(engram_core::types::snapshot::MigrationItem::HotHint),
                     Ok(Kind::StateBin) => Ok(engram_core::types::snapshot::MigrationItem::StateBin),
                     Ok(Kind::DiskSealInfo) => Ok(
                         engram_core::types::snapshot::MigrationItem::DiskSealInfo(role),

crates/engram-host-agent/src/lib.rs

@@ -43,6 +43,7 @@ pub mod harness;
 pub mod host_client;
 pub mod migrate_peer;
 pub mod migration;
+pub mod page_idle;
 pub mod peer_fill;
 pub mod session_epochs;
 pub mod substrate_server;

crates/engram-host-agent/src/migrate_peer.rs

@@ -701,10 +701,11 @@ impl PeerServer {
 /// as the proto's sync `read_frame` (4-byte big-endian length + bincode
 /// body, bounded by `MAX_FRAME_BYTES`). The page channel's codec is sync
 /// by design (the dest client must stay tokio-free), but the SOURCE
-/// reads only the opening `Hello` async so an unknown-export park can
+/// reads the opening `Hello` async so an unknown-export park can
 /// run on a tokio task instead of pinning a blocking-pool thread (issue
-/// #226 (b)); everything after the export resolves uses the sync codec.
-async fn read_frame_async<R, T>(r: &mut R) -> std::io::Result<T>
+/// #226 (b)); the destination also uses this for cancellable control reports.
+/// Page serving after the export resolves uses the sync codec.
+pub(crate) async fn read_frame_async<R, T>(r: &mut R) -> std::io::Result<T>
 where
     R: tokio::io::AsyncRead + Unpin,
     T: serde::de::DeserializeOwned,

crates/engram-host-agent/src/migration.rs

@@ -80,7 +80,7 @@ pub struct MigrationExport {
     /// session (`ttl_verdict` returns StayPaused, converging via the
     /// scanner within one cycle).
     pub state_served: Arc<AtomicBool>,
-    /// Last page/artifact-serving activity (the export TTL clock), as a
+    /// Last required page/artifact-serving activity (the export TTL clock), as a
     /// `now_mono` reading.
     pub last_activity: Arc<std::sync::Mutex<Duration>>,
     /// The sandbox's capture lock, held for the export's lifetime —
@@ -90,7 +90,7 @@ pub struct MigrationExport {
 }
 
 impl MigrationExport {
-    /// Refresh the activity clock (every artifact/page serve) off the
+    /// Refresh the activity clock (required artifact/page serves only) off the
     /// injected monotonic clock.
     pub fn touch(&self) {
         *self.last_activity.lock().expect("last_activity poisoned") = self.clock.now_mono();