fix(host-agent): net prewarm pending charge, gate ledger gauges on measured, real statfs tmpfs-used, pin watch-channel invariant (findings 2, 3, 4, 5)

- finding 2: RamLedger::register_pending_base_shm now stores the target
  file path alongside the expected byte count. sample() nets each
  pending charge against that file's own st_blocks (file_allocated_bytes)
  instead of holding the full charge outstanding for the whole write
  window — the fix for the transient double-charge where a ~19 GiB
  dev-brain prewarm at 90% written was under-reporting allocatable_mib
  by ~17 GiB (MemAvailable already dropping AND the full charge still
  subtracted). New regression test
  pending_charge_nets_against_bytes_already_written proves the charge
  shrinks as bytes actually land.

- finding 3: the whole HOST_RAM_LEDGER_MIB / HOST_RAM_ALLOCATABLE_MIB /
  HOST_BASE_SHM_TMPFS_* gauge-emission block is now gated on
  ram_snapshot.measured, so VZ/Process/non-Linux hosts (and a genuine
  /proc/meminfo parse failure) stop reporting a permanently-zero gauge
  family that reads as "this host has 0 MiB of everything" instead of
  "unmeasured" — matches the PR's own acceptance criterion.

- finding 5: HOST_BASE_SHM_TMPFS_USED_MIB was byte-identical to the
  base_shm category gauge (a read_dir/st_blocks walk over known
  files). tmpfs_stat_mib now also does a real statfs read
  (f_blocks - f_bfree) and RamLedgerSnapshot carries a new
  base_shm_tmpfs_used_mib field for it, so an unlinked-but-open file or
  stray subdir the dir walk can't see is no longer invisible to the
  gauge that exists specifically to debug ENOSPC-class tmpfs incidents.

- finding 4: added
  ram_ledger_snapshot_round_trips_through_watch_channel, publishing a
  RamLedgerSnapshot on a tokio::sync::watch channel and asserting two
  independent readers observe byte-identical values to what was
  published — pins the "one source of truth" mechanism the heartbeat
  and pressure gate rely on.

Review threads:
#561 (comment) (finding 2)
#561 (comment) (finding 3)
#561 (comment) (finding 4)
#561 (comment) (finding 5)

Author: nikhilunni

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

@@ -660,7 +660,11 @@ async fn prefetch_one(
                          the handler's lazy path backstops",
                     );
                 } else {
-                    ram_ledger.register_pending_base_shm(memory_ref, pending_bytes);
+                    ram_ledger.register_pending_base_shm(
+                        memory_ref,
+                        base_path.clone(),
+                        pending_bytes,
+                    );
                     let prewarm_result =
                         prewarm_base_shm(&base_path, &memory_manifest, chunk_store, chunk_cache)
                             .await;

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

@@ -1173,26 +1173,46 @@ impl HostAgent {
                         &guest_mem.unwrap_or_default(),
                     );
                     ram_ledger_tx_for_heartbeat.send_replace(ram_snapshot);
-                    ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "running_vms")
-                        .set(ram_snapshot.running_vm_pss_mib as f64);
-                    ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "parked_paused")
-                        .set(ram_snapshot.parked_paused_pss_mib as f64);
-                    ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "base_shm")
-                        .set(ram_snapshot.base_shm_mib as f64);
-                    ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "base_shm_pending")
-                        .set(ram_snapshot.base_shm_pending_mib as f64);
-                    ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "parked_local_memfiles")
-                        .set(ram_snapshot.parked_local_memfile_mib as f64);
-                    ::metrics::gauge!(crate::metrics::HOST_RAM_ALLOCATABLE_MIB)
-                        .set(ram_snapshot.allocatable_mib() as f64);
-                    ::metrics::gauge!(crate::metrics::HOST_BASE_SHM_TMPFS_TOTAL_MIB)
-                        .set(ram_snapshot.base_shm_tmpfs_total_mib as f64);
-                    ::metrics::gauge!(crate::metrics::HOST_BASE_SHM_TMPFS_USED_MIB)
-                        .set(ram_snapshot.base_shm_mib as f64);
+                    // Issue #540 review finding 3: gate every ledger gauge
+                    // on `measured` — VZ/Process/non-Linux backends (and a
+                    // genuine `/proc/meminfo` parse failure) never took a
+                    // real sample, so `ram_snapshot` is the all-zero
+                    // default. Emitting that as a value would look like
+                    // "this host has 0 MiB of everything" on a dashboard
+                    // instead of "unmeasured" — matches the acceptance
+                    // criterion's "gauges not emitted" posture.
+                    if ram_snapshot.measured {
+                        ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "running_vms")
+                            .set(ram_snapshot.running_vm_pss_mib as f64);
+                        ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "parked_paused")
+                            .set(ram_snapshot.parked_paused_pss_mib as f64);
+                        ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "base_shm")
+                            .set(ram_snapshot.base_shm_mib as f64);
+                        ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "base_shm_pending")
+                            .set(ram_snapshot.base_shm_pending_mib as f64);
+                        ::metrics::gauge!(crate::metrics::HOST_RAM_LEDGER_MIB, "category" => "parked_local_memfiles")
+                            .set(ram_snapshot.parked_local_memfile_mib as f64);
+                        ::metrics::gauge!(crate::metrics::HOST_RAM_ALLOCATABLE_MIB)
+                            .set(ram_snapshot.allocatable_mib() as f64);
+                        ::metrics::gauge!(crate::metrics::HOST_BASE_SHM_TMPFS_TOTAL_MIB)
+                            .set(ram_snapshot.base_shm_tmpfs_total_mib as f64);
+                        // Issue #540 review finding 5: this is the tmpfs
+                        // mount's own `statfs` used figure (`f_blocks -
+                        // f_bfree`), NOT `base_shm_mib` (this ledger's
+                        // st_blocks walk over known files) — the two can
+                        // legitimately diverge (an unlinked-but-open file,
+                        // a stray subdir) and this gauge exists specifically
+                        // to catch that divergence during an ENOSPC-class
+                        // incident.
+                        ::metrics::gauge!(crate::metrics::HOST_BASE_SHM_TMPFS_USED_MIB)
+                            .set(ram_snapshot.base_shm_tmpfs_used_mib as f64);
+                    }
                     // Issue #540: single emission site for this gauge (was
                     // previously only set inside the idle-evictor's
                     // pressure-aware branch, so it read stale/unset when
-                    // that mode was off). Every tick now, unconditionally.
+                    // that mode was off). Every tick now, unconditionally
+                    // (still gated on `measured` via `free_pct()`'s own
+                    // `None` return).
                     if let Some(pct) = ram_snapshot.free_pct() {
                         ::metrics::gauge!(crate::metrics::HOST_MEM_FREE_PCT).set(f64::from(pct));
                     }