detcore_model/

time.rs

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 * All rights reserved.
 *
 * This source code is licensed under the BSD-style license found in the
 * LICENSE file in the root directory of this source tree.
 */

use std::collections::HashMap;
use std::fmt;
use std::ops::Add;
use std::ops::Sub;
use std::time::Duration;

use chrono::DateTime;
use chrono::Utc;
use reverie_syscalls::Timespec;
use reverie_syscalls::Timeval;
use serde::Deserialize;
use serde::Serialize;
use tracing::trace;

use crate::config::Config;
use crate::pid::DetTid;

// Time conversion constants from https://doc.rust-lang.org/stable/src/core/time.rs.html#26-30
const NANOS_PER_SEC: u64 = 1_000_000_000;
const NANOS_PER_MILLI: u64 = 1_000_000;
const NANOS_PER_MICRO: u64 = 1_000;
const MILLIS_PER_SEC: u64 = 1_000;
const MICROS_PER_SEC: u64 = 1_000_000;

// TODO: make all of these integral types to rule out fractional values.

/// Default virtual nanoseconds elapsed for callers that do not provide a syscall-specific cost.
pub const NANOS_PER_SYSCALL: f64 = 10000.0;

/// Virtual nanoseconds elapsed per Retired Conditional Branch.
pub const NANOS_PER_RCB: f64 = 10.0;

// AUTONOMOUS-BOT-IMPLEMENTED
// TODO-HUMAN-REVIEW(PR-1151)
/// Fixed-point scale used for deterministic per-thread RCB time multipliers.
/// Q32 keeps accumulation independent of how a backend batches RCB updates.
const RCB_TIME_MULTIPLIER_SCALE: u64 = 1_u64 << 32;

// AUTONOMOUS-BOT-IMPLEMENTED
// TODO-HUMAN-REVIEW(PR-1151)
/// A positive Q32 multiplier for converting RCB progress into virtual time.
#[derive(
    Debug,
    Clone,
    Copy,
    Serialize,
    Deserialize,
    Ord,
    PartialOrd,
    Eq,
    PartialEq,
    Hash
)]
pub struct RcbTimeMultiplier(u64);

impl RcbTimeMultiplier {
    /// The identity multiplier.
    pub const ONE: Self = Self(RCB_TIME_MULTIPLIER_SCALE);

    /// Largest representable multiplier.
    pub const MAX: f64 = u64::MAX as f64 / RCB_TIME_MULTIPLIER_SCALE as f64;

    /// Quantize a finite positive multiplier to deterministic Q32 units.
    pub fn from_f64(value: f64) -> Self {
        assert!(value.is_finite() && value > 0.0 && value <= Self::MAX);
        let scaled = (value * RCB_TIME_MULTIPLIER_SCALE as f64).round() as u64;
        Self(scaled.max(1))
    }

    /// Convert the fixed-point value back to a floating-point multiplier.
    pub fn as_f64(self) -> f64 {
        self.0 as f64 / RCB_TIME_MULTIPLIER_SCALE as f64
    }

    fn units(self) -> u64 {
        self.0
    }
}

impl Default for RcbTimeMultiplier {
    fn default() -> Self {
        Self::ONE
    }
}

/// Virtual nanoseconds elapsed per nondeterministic instruction other than system calls.
pub const NANOS_PER_NONDET_INSTR: f64 = 25.0;

/// Virtual nanoseconds elapsed per step of the scheduler.
pub const NANOS_PER_SCHED: f64 = 500_000.0;

// TODO: should map addresses to physical addresses.

// Deterministic Time:
//--------------------------------------------------------------------------------

/// Represents an absolute point in time in nanoseconds.
/// Parts of this API are largely inspired by `std::time::Duration`.
/// This could go to 128 bits if we need more than ~585 years of nanosecond precision.
#[derive(
    Default,
    Debug,
    Clone,
    Copy,
    Serialize,
    Deserialize,
    Ord,
    PartialOrd,
    Eq,
    PartialEq,
    Hash
)]
pub struct LogicalTime(u64);

// TODO: replace this with a wrapper around Duration, and change places that currently use Duration
// but should use LogicalDuration.
pub type LogicalDuration = LogicalTime;

impl LogicalTime {
    /// 0 integer nanoseconds.
    pub const ZERO: LogicalTime = LogicalTime(0);
    /// The maximum representable integer nanoseconds.
    pub const MAX: LogicalTime = LogicalTime(u64::MAX);

    /// The sentinel deadline for a wait that has *no* deadline.
    ///
    /// A `pause(2)` blocks until a signal arrives and can never time out, so it
    /// registers this value rather than a real deadline. The saturating `Add`
    /// impls below also land here when a guest arms an absurdly far-future timer
    /// (issue #219), which is the same situation: a deadline that can never be
    /// reached while virtual time remains representable.
    ///
    /// Callers that fast-forward virtual time to a pending deadline must check
    /// [`LogicalTime::is_indefinite`] first. Jumping the global clock onto this
    /// value would both destroy the continuity of virtual time (a ~584-year
    /// step) and wake a waiter that Linux would have left blocked.
    pub const INDEFINITE: LogicalTime = LogicalTime::MAX;

    /// Is this the [`LogicalTime::INDEFINITE`] sentinel, i.e. "no deadline"?
    pub fn is_indefinite(&self) -> bool {
        *self == LogicalTime::INDEFINITE
    }

    /// Returns the total number of whole microseconds contained by this `LogicalTime`.
    pub fn as_micros(&self) -> u64 {
        self.0 / NANOS_PER_MICRO
    }

    /// Returns the total number of whole milliseconds contained by this `LogicalTime`.
    pub fn as_millis(&self) -> u64 {
        self.0 / NANOS_PER_MILLI
    }

    /// Returns the total number of nanoseconds contained by this `LogicalTime`.
    pub fn as_nanos(&self) -> u64 {
        self.0
    }

    /// Returns the total number of *whole* seconds contained by this `LogicalTime`.
    /// The returned value does not include fractional (nanosecond) part of the duration,
    /// which can be obtained using `subsec_nanos`.
    pub fn as_secs(&self) -> u64 {
        self.0 / NANOS_PER_SEC
    }

    /// Creates a new `LogicalTime` from the specified number of microseconds.
    pub fn from_micros(micros: u64) -> Self {
        LogicalTime(micros * NANOS_PER_MICRO)
    }

    /// Creates a new `LogicalTime` from the specified number of milliseconds.
    pub fn from_millis(millis: u64) -> Self {
        LogicalTime(millis * NANOS_PER_MILLI)
    }

    /// Creates a new `LogicalTime` from the specified number of nanoseconds.
    pub fn from_nanos(nanos: u64) -> Self {
        LogicalTime(nanos)
    }

    /// Creates a new `LogicalTime` from the specified number of nanoseconds.
    pub fn from_big_nanos(nanos: u128) -> Self {
        // No good solution for this until we change the internal rep to 128 bit:
        LogicalTime(nanos as u64)
    }

    /// Creates a new `LogicalTime` from the specified number of seconds.
    pub fn from_secs(secs: u64) -> Self {
        LogicalTime(secs * NANOS_PER_SEC)
    }

    /// Returns the fractional part of this `LogicalTime`, in microseconds.
    /// This method does not return the length of the duration when represented by microseconds. The returned number always represents
    /// a fractional portion of a second (i.e., it is less than one million).
    pub fn subsec_micros(&self) -> u32 {
        ((self.0 / NANOS_PER_MICRO) % MICROS_PER_SEC) as u32
    }

    /// Returns the fractional part of this `LogicalTime`, in milliseconds.
    /// This method does not return the length of the duration when represented by milliseconds. The returned number always represents
    /// a fractional portion of a second (i.e., it is less than one thousand).
    pub fn subsec_millis(&self) -> u32 {
        ((self.0 / NANOS_PER_MILLI) % MILLIS_PER_SEC) as u32
    }

    /// Returns the fractional part of this `LogicalTime`, in nanoseconds.
    /// This method does not return the length of the duration when represented by nanoseconds. The returned number always represents
    /// a fractional portion of a second (i.e., it is less than one billion).
    pub fn subsec_nanos(&self) -> u32 {
        (self.0 % NANOS_PER_SEC) as u32
    }

    /// Convert a number of Retired Conditional Branches (RCBs) to Nanoseconds
    pub fn from_rcbs(n: u64) -> Self {
        LogicalTime((n as f64 * NANOS_PER_RCB) as u64)
    }

    /// Inverse of from_rcbs.  Non-injective, as it loses information, truncating to a
    /// coarser grained unit of time.
    pub fn into_rcbs(self) -> u64 {
        (self.0 as f64 / NANOS_PER_RCB) as u64
    }

    /// Convert a virtual duration to RCBs after applying a logical clock multiplier.
    pub fn into_rcbs_with_multiplier(self, multiplier: f64) -> u64 {
        debug_assert!(multiplier > 0.0);
        (self.0 as f64 / (NANOS_PER_RCB * multiplier)).floor() as u64
    }

    /// Test if the quantity is zero nanoseconds.
    pub fn is_zero(&self) -> bool {
        self.0 == 0
    }

    /// Measure the duration of the time interval since a previous time.
    pub fn duration_since(&self, from: LogicalTime) -> Duration {
        if from.0 > self.0 {
            panic!(
                "LogicalTime::duration_since cannot take duration since a time in the *future* ({}), relative to {}",
                from, self
            );
        }
        Duration::from_nanos(self.0 - from.0)
    }
}

impl std::fmt::Display for LogicalTime {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        // Start with the raw characters for printed u64:
        let chars = format!("{}", self.0);
        let mut remain = chars.len();
        let mut first_char = true;
        for ch in chars.chars() {
            if !first_char && remain % 3 == 0 {
                if remain == 9 {
                    write!(f, ".")?;
                } else {
                    // Could also consider \u{2009} thin space here, but it prints fixed
                    // width in most terminals anyway:
                    write!(f, "_")?;
                }
            }
            first_char = false;
            remain -= 1;
            write!(f, "{}", ch)?;
        }
        if chars.len() <= 9 {
            write!(f, "ns")
        } else {
            write!(f, "s")
        }
    }
}

impl Add for LogicalTime {
    type Output = Self;
    fn add(self, rhs: Self) -> Self {
        // Saturate at LogicalTime::MAX rather than panicking on overflow: a
        // far-future logical time is clamped to "the end of time".
        LogicalTime(self.0.saturating_add(rhs.0))
    }
}

impl Add<Duration> for LogicalTime {
    type Output = Self;
    fn add(self, rhs: Duration) -> Self {
        // A `Duration` can hold more nanoseconds than fit in u64, and the sum
        // can exceed u64::MAX (e.g. Java arms a far-future timer, issue #219).
        // Saturate the u128->u64 conversion and the addition instead of
        // overflowing/truncating.
        let nanos = u64::try_from(rhs.as_nanos()).unwrap_or(u64::MAX);
        LogicalTime(self.0.saturating_add(nanos))
    }
}

impl Add<u128> for LogicalTime {
    type Output = Self;
    fn add(self, rhs: u128) -> Self {
        // Saturate both the u128->u64 clamp and the addition.
        LogicalTime(self.0.saturating_add(rhs.min(u64::MAX as u128) as u64))
    }
}

impl Sub for LogicalTime {
    type Output = Self;
    fn sub(self, rhs: Self) -> Self {
        LogicalTime(self.0 - rhs.0)
    }
}

impl From<LogicalTime> for Timespec {
    fn from(logical_time: LogicalTime) -> Timespec {
        Timespec {
            tv_sec: logical_time.as_secs() as i64,
            tv_nsec: logical_time.subsec_nanos() as i64,
        }
    }
}

impl From<LogicalTime> for Timeval {
    fn from(logical_time: LogicalTime) -> Timeval {
        Timeval {
            tv_sec: logical_time.as_secs() as i64,
            tv_usec: logical_time.subsec_micros() as i64,
        }
    }
}

#[test]
fn print_nanoseconds() {
    let ns1 = LogicalTime(946_684_799_000_000_000);
    let ns2 = LogicalTime(729_860_000);
    assert_eq!(format!("{}", ns1), "946_684_799.000_000_000s");
    assert_eq!(format!("{}", ns1 + ns2), "946_684_799.729_860_000s");
    assert_eq!(format!("{}", ns2), "729_860_000ns");
}

#[test]
fn subsecond_units_are_converted_from_nanoseconds() {
    let time = LogicalTime::from_secs(2) + LogicalTime::from_nanos(345_678_901);

    assert_eq!(time.subsec_millis(), 345);
    assert_eq!(time.subsec_micros(), 345_678);
    assert_eq!(time.subsec_nanos(), 345_678_901);

    let timeval: Timeval = time.into();
    assert_eq!(timeval.tv_sec, 2);
    assert_eq!(timeval.tv_usec, 345_678);
}

#[test]
fn indefinite_is_recognized_and_nothing_else_is() {
    assert!(LogicalTime::INDEFINITE.is_indefinite());
    assert!(!LogicalTime::ZERO.is_indefinite());
    assert!(!LogicalTime::from_secs(1_767_225_600).is_indefinite());
    assert!(!LogicalTime(u64::MAX - 1).is_indefinite());

    // A saturated far-future deadline (issue #219) lands on the same sentinel,
    // and is equally unreachable, so the scheduler must treat it the same way.
    assert!((LogicalTime(u64::MAX - 10) + LogicalTime::from_secs(1)).is_indefinite());
}

#[test]
fn add_saturates_instead_of_overflowing() {
    // Regression for issue #219: Java arms a far-future timer whose deadline
    // overflows u64. All three `Add` impls must saturate at LogicalTime::MAX
    // rather than panic.
    let near_max = LogicalTime(u64::MAX - 10);

    // Add<LogicalTime>
    assert_eq!(near_max + LogicalTime::from_secs(1), LogicalTime::MAX);

    // Add<Duration>, including a Duration whose nanos exceed u64::MAX.
    assert_eq!(near_max + Duration::from_secs(1), LogicalTime::MAX);
    assert_eq!(
        LogicalTime::ZERO + Duration::from_secs(u64::MAX / 1_000_000_000 + 1),
        LogicalTime::MAX
    );

    // Add<u128>, including an rhs larger than u64::MAX.
    assert_eq!(near_max + 100u128, LogicalTime::MAX);
    assert_eq!(
        LogicalTime::ZERO + (u128::from(u64::MAX) + 5),
        LogicalTime::MAX
    );

    // A non-overflowing add still produces the exact sum.
    assert_eq!(
        LogicalTime::from_secs(2) + Duration::from_secs(3),
        LogicalTime::from_secs(5)
    );
}

/// The same basic type alias as nanoseconds. Just for clarity/readability.
pub type Microseconds = u64;

/// A determinstic notion of time, measuring progress of one thread.
///
/// It is based on counting syscalls and conditionals.  Ideally it would count
/// instructions, but that is not possible deterministically.
///
/// `DetTime` is a measure of the LOCAL progress of a thread. The notion of
/// global time is defined by aggretion of multiple local times (vector
/// clocks, as in the Kendo algorithm).
///
/// Here are a few relevant definitions for deterministic time:
///
/// **Granularity**
///
/// How rapidly and consistently does the clock tick with thread progress?
/// A finer notion of deterministic time counts *more* events (ideally instructions).
/// A coarser notion of deterministic time counts fewer events (like syscalls).
///
/// **Productivity**
///
/// In corecursion, or coinductive datatypes like streams, productivity means you can get
/// the next result with finite work. Deterministic time can be viewed a stream of "tick
/// events". We don't want a guest thread to be able to do an unbounded amount of work,
/// without a tick occurring. For example, retired branches are a safe bet (any
/// non-trivial amount of work will execute a branch), but system calls are not: a
/// spinning thread can burn cycles forever without executing a syscall.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DetTime {
    /// The syscalls issued by this thread.
    syscalls: u64,

    /// Accumulated syscall cost before applying `multiplier`.
    ///
    /// `None` preserves the uniform-cost interpretation of serialized `DetTime` values created
    /// before syscall-specific costs were introduced.
    #[serde(default)]
    syscall_nanos: Option<u64>,

    /// Retired conditional branches, as given "opaquely" by the reverie clock.
    /// Technically, that these are RCBs is an implementation detail of reverie.
    rcbs: u64,

    // AUTONOMOUS-BOT-IMPLEMENTED
    // TODO-HUMAN-REVIEW(PR-1151)
    /// RCB progress weighted by per-thread virtual-time multipliers, in Q32 RCB units.
    /// `None` preserves the uniform interpretation of older serialized values.
    // This field participates in Reverie's non-self-describing bincode RPC.
    // Never skip it during serialization: doing so would shift the following
    // `GlobalRequest` bytes and corrupt the tuple on decode.
    #[serde(default)]
    weighted_rcbs: Option<u128>,

    /// Number of nondeterministic instructions (rdtsc, cpuid)
    nondet_instrs: u64,

    /// Explicit virtual-time advances, such as a PMU maximum when RCB accounting is disabled.
    #[serde(default)]
    extra_nanos: u64,

    /// Baseline amount of time to add.
    starting_micros: Microseconds,

    /// Multiplier for all time advances.
    multiplier: f64,

    /// Elapsed local time inherited when this thread was created. It remains
    /// part of the absolute clock, but is work already charged to its ancestors.
    /// Like the other clock fields, this must always be present in bincode RPCs.
    #[serde(default)]
    inherited_nanos: LogicalDuration,
}

// Don't derive Default because it would give us a 0.0 multiplier:
impl Default for DetTime {
    fn default() -> Self {
        DetTime {
            syscalls: 0,
            syscall_nanos: Some(0),
            rcbs: 0,
            weighted_rcbs: None,
            nondet_instrs: 0,
            extra_nanos: 0,
            starting_micros: 0,
            multiplier: 1.0,
            inherited_nanos: LogicalTime::ZERO,
        }
    }
}

impl Eq for DetTime {}

/// `DetTime` behaves as a totally ordered scalar.
impl Ord for DetTime {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        let nanos = other.as_nanos();
        self.as_nanos().cmp(&nanos)
    }
}

#[allow(clippy::non_canonical_partial_ord_impl)]
impl PartialOrd for DetTime {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        let nanos = other.as_nanos();
        self.as_nanos().partial_cmp(&nanos)
    }
}

impl PartialEq for DetTime {
    fn eq(&self, other: &Self) -> bool {
        self.as_nanos() == other.as_nanos()
    }
}

impl From<&DateTime<Utc>> for DetTime {
    fn from(dt: &DateTime<Utc>) -> Self {
        DetTime {
            syscalls: 0,
            syscall_nanos: Some(0),
            rcbs: 0,
            weighted_rcbs: None,
            nondet_instrs: 0,
            extra_nanos: 0,
            starting_micros: micros_from_utc(dt),
            multiplier: 1.0,
            inherited_nanos: LogicalTime::ZERO,
        }
    }
}

fn micros_from_utc(dt: &DateTime<Utc>) -> Microseconds {
    dt.timestamp() as Microseconds * 1_000_000 + dt.timestamp_subsec_micros() as Microseconds
}

// implementing From<DetTime> for Timespec is not possible due to dependency graph
#[allow(clippy::from_over_into)]
impl Into<Timespec> for DetTime {
    fn into(self) -> Timespec {
        self.as_nanos().into()
    }
}

impl DetTime {
    /// Create an initial `DetTime` respecting a `Config`
    pub fn new(cfg: &Config) -> Self {
        // We inflate the amount of time everything consumes to compensate for the fact that we
        // are ONLY counting certain events sparsely. In theory this should be based on some kind
        // of expected value for compute-between-syscalls on average applications. (But is that
        // even a normal distribution?)
        let additional_multiplier = if cfg.sequentialize_threads && !cfg.use_rcb_time() {
            500.0
        } else {
            // Otherwise, virtual time isn't really used for scheduling, just for
            // metadata, so it doesn't really matter what the rate of ticking is.
            1.0
        };
        match cfg.clock_multiplier {
            Some(m) => DetTime::from(&cfg.epoch).with_multiplier(m * additional_multiplier),
            None => DetTime::from(&cfg.epoch).with_multiplier(additional_multiplier),
        }
    }

    /// Create a new `Dettime` which is the earliest possible.
    pub fn zero() -> Self {
        DetTime {
            syscalls: 0,
            syscall_nanos: Some(0),
            rcbs: 0,
            weighted_rcbs: None,
            nondet_instrs: 0,
            extra_nanos: 0,
            starting_micros: 0,
            multiplier: 1.0,
            inherited_nanos: LogicalTime::ZERO,
        }
    }

    /// Inherit an absolute clock for a newly created thread without attributing
    /// its ancestors' work to that thread. Ordinary `Clone` still copies an
    /// existing clock, including its original inheritance, for RPC snapshots.
    pub fn clone_for_child(&self) -> Self {
        let mut child = self.clone();
        child.inherited_nanos = self.without_starting();
        child
    }

    /// Local elapsed time already accounted for by this thread's ancestors.
    pub fn inherited_nanos(&self) -> LogicalDuration {
        self.inherited_nanos
    }

    /// Register that another syscall has executed.
    pub fn add_syscall(&mut self) {
        self.add_syscall_with_cost(NANOS_PER_SYSCALL as u64);
    }

    /// Register a syscall with its unscaled virtual-time cost in nanoseconds.
    pub fn add_syscall_with_cost(&mut self, nanos: u64) {
        let previous_uniform_nanos = (self.syscalls as f64 * NANOS_PER_SYSCALL) as u64;
        self.syscalls += 1;
        match &mut self.syscall_nanos {
            Some(syscall_nanos) => *syscall_nanos += nanos,
            None => self.syscall_nanos = Some(previous_uniform_nanos + nanos),
        }
        trace!(
            "[detcore] added syscall cost of {}ns to logical time, yielding: {:?}",
            nanos, self
        );
    }

    /// Register that an `rdtsc` intsruction has executed.
    pub fn add_rdtsc(&mut self) {
        self.nondet_instrs += 1;
    }

    /// Register that an `cpuid` intsruction has executed.
    pub fn add_cpuid(&mut self) {
        self.nondet_instrs += 1;
    }

    /// Update internal counts using the reverie clock value.
    pub fn add_rcbs(&mut self, count: u64) {
        if let Some(weighted_rcbs) = &mut self.weighted_rcbs {
            *weighted_rcbs += u128::from(count) * u128::from(RCB_TIME_MULTIPLIER_SCALE);
        }
        self.rcbs += count;
    }

    // AUTONOMOUS-BOT-IMPLEMENTED
    // TODO-HUMAN-REVIEW(PR-1151)
    /// Add RCB progress using a deterministic per-thread virtual-time multiplier.
    ///
    /// The Q32 accumulator makes the result independent of whether a backend reports
    /// the same RCB total in one update or several smaller updates.
    pub fn add_rcbs_with_multiplier(&mut self, count: u64, factor: RcbTimeMultiplier) {
        let weighted_rcbs = self
            .weighted_rcbs
            .get_or_insert_with(|| u128::from(self.rcbs) * u128::from(RCB_TIME_MULTIPLIER_SCALE));
        *weighted_rcbs += u128::from(count) * u128::from(factor.units());
        self.rcbs += count;
    }

    /// Advance this local clock to an explicit virtual deadline.
    pub fn advance_to(&mut self, deadline: LogicalTime) {
        let current = self.as_nanos();
        assert!(deadline >= current);
        self.extra_nanos += (deadline - current).as_nanos();
    }

    /// Return current rcbs
    pub fn rcbs(&self) -> u64 {
        self.rcbs
    }

    /// Project deterministic logical time into a rough number of nanoseconds.
    pub fn as_nanos(&self) -> LogicalTime {
        // Note: these counts could be pre-collapsed into scalar within the DetTime
        // representation.  But currently we leave them separate for debuggability.
        let syscall_nanos = self
            .syscall_nanos
            .unwrap_or((self.syscalls as f64 * NANOS_PER_SYSCALL) as u64);
        let rcb_nanos = self.weighted_rcbs.map_or_else(
            || self.rcbs as f64 * NANOS_PER_RCB,
            |weighted| weighted as f64 * NANOS_PER_RCB / RCB_TIME_MULTIPLIER_SCALE as f64,
        );
        LogicalTime(
            (self.starting_micros * 1000)
                + self.extra_nanos
                + ((syscall_nanos as f64 * self.multiplier) as u64)
                + ((rcb_nanos * self.multiplier) as u64)
                + ((self.nondet_instrs as f64 * NANOS_PER_NONDET_INSTR * self.multiplier) as u64),
        )
    }

    /// Same as as_nanos but without the starting time.
    pub fn without_starting(&self) -> LogicalDuration {
        let LogicalTime(t1) = self.as_nanos();
        LogicalTime(t1 - (self.starting_micros * 1000))
    }

    // TODO-HUMAN-REVIEW(#797): Review logical user/system CPU-time projections.
    /// Guest-execution time that corresponds to user-space instructions.
    pub fn user_cpu_time(&self) -> LogicalDuration {
        let rcb_nanos = self.weighted_rcbs.map_or_else(
            || self.rcbs as f64 * NANOS_PER_RCB,
            |weighted| weighted as f64 * NANOS_PER_RCB / RCB_TIME_MULTIPLIER_SCALE as f64,
        );
        LogicalTime(
            ((rcb_nanos + (self.nondet_instrs as f64 * NANOS_PER_NONDET_INSTR)) * self.multiplier)
                as u64,
        )
    }

    // TODO-HUMAN-REVIEW(#797): Review logical user/system CPU-time projections.
    /// Synthetic time charged for intercepted syscall execution.
    pub fn system_cpu_time(&self) -> LogicalDuration {
        let syscall_nanos = self
            .syscall_nanos
            .unwrap_or((self.syscalls as f64 * NANOS_PER_SYSCALL) as u64);
        LogicalTime((syscall_nanos as f64 * self.multiplier) as u64)
    }

    /// Project deterministic logical time into a rough number of microseconds.
    pub fn as_micros(&self) -> Microseconds {
        self.as_nanos().0 / 1000
    }

    /// Set the clock multiplier
    pub fn with_multiplier(mut self, m: f64) -> Self {
        self.multiplier = m;
        self
    }

    /// Project deterministic time duration from imaginary starting point of deterministic time creation
    pub fn as_duration(&self) -> std::time::Duration {
        std::time::Duration::from_nanos(self.as_nanos().0 - self.starting_micros * 1000)
    }
}

#[cfg(test)]
mod rcb_multiplier_tests {
    use super::*;

    // AUTONOMOUS-BOT-IMPLEMENTED
    // TODO-HUMAN-REVIEW(PR-1151)
    #[test]
    fn weighted_rcb_time_is_batching_independent() {
        let factor = RcbTimeMultiplier::from_f64(2.5);
        let mut one_batch = DetTime::zero();
        one_batch.add_rcbs_with_multiplier(10, factor);

        let mut split_batches = DetTime::zero();
        split_batches.add_rcbs_with_multiplier(4, factor);
        split_batches.add_rcbs_with_multiplier(6, factor);

        assert_eq!(one_batch.rcbs(), 10);
        assert_eq!(split_batches.rcbs(), 10);
        assert_eq!(one_batch.as_nanos(), LogicalTime::from_nanos(250));
        assert_eq!(one_batch.as_nanos(), split_batches.as_nanos());
        assert_eq!(one_batch.user_cpu_time(), split_batches.user_cpu_time());
    }

    // AUTONOMOUS-BOT-IMPLEMENTED
    // TODO-HUMAN-REVIEW(PR-1151)
    #[test]
    fn uniform_rcbs_after_weighted_rcbs_keep_continuity() {
        let mut time = DetTime::zero();
        time.add_rcbs_with_multiplier(10, RcbTimeMultiplier::from_f64(0.5));
        assert_eq!(time.as_nanos(), LogicalTime::from_nanos(50));

        time.add_rcbs(5);
        assert_eq!(time.rcbs(), 15);
        assert_eq!(time.as_nanos(), LogicalTime::from_nanos(100));
    }
}

/// Deterministic global time, combining local times.
#[derive(Default, Debug, Clone, Serialize, Deserialize)]
pub struct GlobalTime {
    /// The time when the container began execution.
    starting_nanos: LogicalTime,

    /// Latest absolute local duration for each thread, excluding the epoch.
    /// This includes inherited history for scheduler and replay consumers.
    time_vector: HashMap<DetTid, LogicalTime>,

    /// The inherited part of each local duration. It contributes no new work
    /// to aggregate time. Missing entries represent zero for older snapshots.
    #[serde(default)]
    inherited_time: HashMap<DetTid, LogicalDuration>,

    /// A source of central, logically external, time passage generated by the scheduler.
    extra_time: LogicalTime,

    /// This won't always be possible, but for now, since local clocks don't tick
    /// asynchronously, it is quite straightforward to keep a count of cumulative progress.
    total: LogicalTime,

    /// Immutable. Simply copied from the Config.
    multiplier: f64,

    /// Immutable. The virtual nanoseconds each scheduler turn adds when
    /// `--scheduler-turn-cost` is configured ([`Config::scheduler_turn_charge`]).
    /// `None`, without the option or in a snapshot from before the field
    /// existed, means the default charge for `multiplier`.
    #[serde(default)]
    scheduler_turn_charge: Option<u64>,
}

/// A scheduler turn whose configured charge would pass the end of
/// representable virtual time ([`GlobalTime::add_scheduler_time`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ClockExhausted {
    /// The configured charge of the refused turn, in nanoseconds.
    pub charge: u64,
    /// The global time the turn would have advanced.
    pub total: LogicalTime,
}

impl std::fmt::Display for ClockExhausted {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "HERMIT_VIRTUAL_CLOCK_EXHAUSTED: virtual clock exhausted: a scheduler turn charging {} \
             ns would pass the end of representable time at {} ns; lower --scheduler-turn-cost or \
             --clock-multiplier",
            self.charge, self.total.0
        )
    }
}

impl GlobalTime {
    /// Create a fresh global time, respecting the `Config`.
    pub fn new(cfg: &Config) -> Self {
        let base = DetTime::new(cfg);
        GlobalTime {
            starting_nanos: LogicalTime::from_micros(micros_from_utc(&cfg.epoch)),
            time_vector: HashMap::new(),
            inherited_time: HashMap::new(),
            extra_time: LogicalTime::from_nanos(0),
            total: base.as_nanos(),
            multiplier: cfg.clock_multiplier.unwrap_or(1.0),
            // Only a configured cost is stored: without one, the default
            // charge stays the old expression with its old saturation.
            // `hermit run` refuses a cost this rejects, so reaching the panic
            // means the configuration bypassed it.
            scheduler_turn_charge: cfg.scheduler_turn_cost.map(|_| {
                cfg.scheduler_turn_charge()
                    .unwrap_or_else(|error| panic!("invalid scheduler turn cost: {error}"))
            }),
        }
    }

    /// Tick the time of a particular thread.
    pub fn update_global_time(
        &mut self,
        tid: DetTid,
        newtime: LogicalTime,
        inherited: LogicalDuration,
    ) {
        if newtime < self.starting_nanos {
            panic!(
                "update_global_time: Cannot set thread {} time to {}, which is before start of container execution {}",
                tid, newtime, self.starting_nanos
            );
        }

        // TODO(T136359599): change to duration_since, and store durations in time_vector:
        let newtime = newtime - self.starting_nanos;
        trace!(
            "[tid {}] ticked its global time component to {}",
            tid, newtime,
        );
        if let Some(old) = self.time_vector.get_mut(&tid) {
            if *old > newtime {
                panic!(
                    "Attempted to update tid {} time to {}, but was already {}",
                    tid, newtime, old
                );
            }
            // Update the cached total for efficiency:
            let LogicalTime(diff) = newtime - *old;
            *old = newtime;
            // Exec may reload fresh local state. Its first response restores
            // the absolute clock, while this existing component retains the
            // original inherited baseline rather than charging that work again.
            self.bump_total(Duration::from_nanos(diff));
        } else {
            assert!(
                inherited <= newtime,
                "thread {tid} inherited time {inherited} beyond its local duration {newtime}"
            );
            self.time_vector.insert(tid, newtime);
            self.inherited_time.insert(tid, inherited);
            // A child's first startup RPC reports its inherited clock before
            // it has executed guest work. Its arrival must contribute zero,
            // whether it precedes or follows the scheduler's time snapshot.
            self.bump_total(Duration::from_nanos((newtime - inherited).0));
        }
    }

    fn sanity(&self) {
        debug_assert_eq!(self.sum_up(), self.total);
    }

    fn bump_total(&mut self, delta: Duration) {
        self.total = self.total + delta;
        self.sanity();
    }

    // The expensive way to get the total (internal)
    fn sum_up(&self) -> LogicalTime {
        let mut sum = self.starting_nanos;
        for (tid, tm) in &self.time_vector {
            sum = sum + (*tm - self.inherited_duration(*tid));
        }
        sum + self.extra_time
    }

    /// Add time that passage is not driven by the internal events within guest threads.
    /// This is effectively used to account for "time" consumed by the scheduler, and to
    /// ensure monotonic increase of global time while scheduling.
    ///
    /// With a configured `--scheduler-turn-cost`, a turn whose charge would
    /// pass the end of representable time is refused with
    /// [`ClockExhausted`] and adds nothing: a saturated clock would let later
    /// turns advance nothing while the run went on. The scheduler turns the
    /// error into a fatal exit. Without the option the default charge
    /// saturates, as it always did, and this never fails.
    pub fn add_scheduler_time(&mut self) -> Result<LogicalTime, ClockExhausted> {
        self.add_capped_scheduler_time(u64::MAX)
    }

    /// [`Self::add_scheduler_time`] for a turn charged at most `cap`
    /// nanoseconds: the scheduler's charge for a turn that ends a timeslice
    /// under `--cap-preemption-charge`, whose `cap` is the virtual time the
    /// thread ran since its previous turn. The capped charge is at least one
    /// nanosecond, so the turn still advances the clock, as a configured
    /// `--scheduler-turn-cost` must (`Config::scheduler_turn_charge`); a
    /// default charge that the clock multiplier truncates to zero stays zero,
    /// as it always was. A `cap` at or above the turn's charge changes
    /// nothing.
    pub fn add_capped_scheduler_time(&mut self, cap: u64) -> Result<LogicalTime, ClockExhausted> {
        let full = self
            .scheduler_turn_charge
            .unwrap_or((NANOS_PER_SCHED * self.multiplier) as u64);
        let charge = full.min(cap.max(1));
        if self.scheduler_turn_charge.is_some() && self.total.0.checked_add(charge).is_none() {
            return Err(ClockExhausted {
                charge,
                total: self.total,
            });
        }
        Ok(self.add_extra_time(Duration::from_nanos(charge)))
    }

    /// Update the global clock to account for time not driven by internal
    /// within guest threads.  This is a central or external expenditure of
    /// time, rather than a thread-internal one.
    ///
    /// The argument is in nanosecods and should have had any clock multiplier
    /// applied alreday.
    pub fn add_extra_time(&mut self, delta: Duration) -> LogicalTime {
        self.extra_time = self.extra_time + delta;
        // Update the cached total for efficiency:
        self.bump_total(delta);
        self.as_nanos()
    }

    /// Project a thread's absolute local clock, including its inherited history
    /// and the epoch. Exec recovery and scheduler replay use this projection.
    pub fn threads_time(&self, dtid: DetTid) -> LogicalTime {
        self.starting_nanos + self.threads_duration(dtid)
    }

    // AUTONOMOUS-BOT-IMPLEMENTED
    // TODO-HUMAN-REVIEW(PR-845): Review thread-presence detection for backend reconnects.
    /// Returns whether this clock has observed work from a thread.
    pub fn contains_thread(&self, dtid: DetTid) -> bool {
        self.time_vector.contains_key(&dtid)
    }

    // AUTONOMOUS-BOT-IMPLEMENTED
    // TODO-HUMAN-REVIEW(PR-1173): Review SaBRe exec clock reassignment.
    /// Move a surviving thread's clock component to the process-leader TID
    /// installed by Linux after a non-leader thread successfully execs.
    ///
    /// Work previously performed by the old leader remains part of aggregate
    /// time, but no longer belongs to the new image's local thread clock.
    pub fn reassign_thread(&mut self, from: DetTid, to: DetTid) {
        if from == to {
            return;
        }

        // Work the old leader did stays in aggregate time.
        self.retire_thread(to);
        let survivor_time = self
            .time_vector
            .remove(&from)
            .unwrap_or_else(|| panic!("cannot reassign missing thread clock {from}"));
        let survivor_inherited = self.inherited_time.remove(&from).unwrap_or_default();
        self.time_vector.insert(to, survivor_time);
        self.inherited_time.insert(to, survivor_inherited);
        self.sanity();
    }

    /// Drop the clock component of a thread that is gone, keeping its own
    /// work in aggregate time. A later thread that Linux gives the same ID
    /// then starts a component of its own, with its own inherited baseline,
    /// instead of continuing the old thread's clock.
    pub fn retire_thread(&mut self, dtid: DetTid) {
        let inherited = self.inherited_time.remove(&dtid).unwrap_or_default();
        if let Some(time) = self.time_vector.remove(&dtid) {
            self.extra_time = self.extra_time + (time - inherited);
        }
        self.sanity();
    }

    fn inherited_duration(&self, dtid: DetTid) -> LogicalDuration {
        self.inherited_time.get(&dtid).copied().unwrap_or_default()
    }

    /// The absolute clock inherited when this thread started. Its first
    /// report can arrive after scheduler registration; work since startup is
    /// the current thread clock minus this baseline, not minus the epoch.
    pub fn threads_inherited_time(&self, dtid: DetTid) -> LogicalTime {
        self.starting_nanos + self.inherited_duration(dtid)
    }

    /// Project a thread's local duration, including inherited history but
    /// excluding the epoch. This is the duration used by scheduler replay;
    /// aggregate time separately excludes the inherited part.
    pub fn threads_duration(&self, dtid: DetTid) -> LogicalDuration {
        *self.time_vector.get(&dtid).unwrap_or_else(|| {
            panic!(
                "Trying to extract time for thread {}, but no entry found!",
                dtid
            )
        })
    }

    /// Deterministic lower bound on the amount of work that has happened across all
    /// threads, starting with the same epoch time as individual thread clocks.
    ///
    /// This roughly models something like real time if all threads were running on one core.
    pub fn as_nanos(&self) -> LogicalTime {
        self.total
    }

    /// Elapsed virtual time in the same precision domain as this clock's
    /// baseline. `DetTime` stores the epoch at microsecond precision, so
    /// subtracting the original nanosecond `DateTime` can underflow for an
    /// otherwise empty run whose epoch has a sub-microsecond component.
    ///
    /// Global time never falls behind its own baseline; `None` reports that
    /// broken invariant in every build profile instead of wrapping in release.
    pub fn elapsed_nanos(&self) -> Option<LogicalDuration> {
        self.total
            .as_nanos()
            .checked_sub(self.starting_nanos.as_nanos())
            .map(LogicalTime::from_nanos)
    }
}

#[cfg(test)]
mod global_time_tests {
    use super::*;

    fn publish(time: &mut GlobalTime, tid: DetTid, clock: &DetTime) {
        time.update_global_time(tid, clock.as_nanos(), clock.inherited_nanos());
    }

    #[test]
    fn submicrosecond_epoch_starts_with_zero_elapsed_time() {
        let config = Config {
            epoch: "2026-09-24T23:52:07.760605385Z".parse().unwrap(),
            ..Config::default()
        };
        let mut time = GlobalTime::new(&config);
        assert_eq!(time.elapsed_nanos(), Some(LogicalTime::ZERO));
        time.add_extra_time(Duration::from_nanos(1));
        assert_eq!(time.elapsed_nanos(), Some(LogicalTime::from_nanos(1)));
    }

    #[test]
    fn global_time_behind_its_baseline_is_refused_not_wrapped() {
        let mut time = GlobalTime::new(&Config::default());
        time.total = LogicalTime::from_nanos(time.starting_nanos.as_nanos() - 1);
        assert_eq!(time.elapsed_nanos(), None);
    }

    #[test]
    fn descendants_preserve_absolute_clocks_and_charge_only_their_own_work() {
        let config = Config {
            clock_multiplier: Some(2.0),
            ..Config::default()
        };
        let root = DetTid::from_raw(3);
        let child = DetTid::from_raw(4);
        let grandchild = DetTid::from_raw(5);
        let mut time = GlobalTime::new(&config);
        let start = time.as_nanos();
        let mut root_clock = DetTime::new(&config);
        root_clock.add_syscall_with_cost(11);
        root_clock.add_rcbs(3);
        root_clock.add_rdtsc();
        root_clock.advance_to(root_clock.as_nanos() + LogicalTime::from_nanos(1));
        assert_eq!(root_clock.without_starting(), LogicalTime::from_nanos(133));
        publish(&mut time, root, &root_clock);

        let mut child_clock = root_clock.clone_for_child();
        assert_eq!(child_clock.as_nanos(), root_clock.as_nanos());
        publish(&mut time, child, &child_clock);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(133));
        assert_eq!(time.threads_time(child), child_clock.as_nanos());
        child_clock.add_syscall_with_cost(7);
        let mut split_rcbs = child_clock.clone();
        split_rcbs.add_rcbs_with_multiplier(1, RcbTimeMultiplier::from_f64(0.5));
        split_rcbs.add_rcbs_with_multiplier(3, RcbTimeMultiplier::from_f64(0.5));
        child_clock.add_rcbs_with_multiplier(4, RcbTimeMultiplier::from_f64(0.5));
        assert_eq!(split_rcbs.as_nanos(), child_clock.as_nanos());
        assert_eq!(split_rcbs.inherited_nanos(), child_clock.inherited_nanos());
        child_clock.add_cpuid();
        child_clock.advance_to(child_clock.as_nanos() + LogicalTime::from_nanos(1));
        assert_eq!(child_clock.without_starting(), LogicalTime::from_nanos(238));
        let snapshot = child_clock.clone();
        assert_eq!(snapshot.inherited_nanos(), LogicalTime::from_nanos(133));
        publish(&mut time, child, &snapshot);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(238));
        publish(&mut time, child, &snapshot);
        publish(&mut time, child, &snapshot);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(238));

        let mut grandchild_clock = child_clock.clone_for_child();
        assert_eq!(
            grandchild_clock.inherited_nanos(),
            LogicalTime::from_nanos(238)
        );
        publish(&mut time, grandchild, &grandchild_clock);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(238));
        grandchild_clock.add_syscall_with_cost(3);
        grandchild_clock.advance_to(grandchild_clock.as_nanos() + LogicalTime::from_nanos(1));
        publish(&mut time, grandchild, &grandchild_clock);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(245));
        assert_eq!(time.threads_time(child), child_clock.as_nanos());
        assert_eq!(time.threads_time(grandchild), grandchild_clock.as_nanos());
        assert_eq!(
            time.threads_duration(grandchild),
            LogicalTime::from_nanos(245)
        );

        time.add_scheduler_time().unwrap();
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_000_245));
    }

    /// `--scheduler-turn-cost` replaces the 500,000 ns charge per scheduler
    /// turn, and `--clock-multiplier` scales it like the default.
    #[test]
    fn scheduler_turn_cost_sets_the_charge_per_turn() {
        let mut default = GlobalTime::new(&Config::default());
        let start = default.as_nanos();
        default.add_scheduler_time().unwrap();
        assert_eq!(default.as_nanos(), start + LogicalTime::from_nanos(500_000));

        let config = Config {
            scheduler_turn_cost: std::num::NonZeroU64::new(10_000),
            clock_multiplier: Some(0.1),
            ..Config::default()
        };
        let mut time = GlobalTime::new(&config);
        let start = time.as_nanos();
        time.add_scheduler_time().unwrap();
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_000));
        time.add_scheduler_time().unwrap();
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(2_000));
    }

    /// A sanity check: at the largest accepted charge every turn of a run
    /// that starts at a current epoch advances the clock by exactly that
    /// charge.
    #[test]
    fn the_largest_scheduler_turn_charge_advances_every_turn() {
        let config = Config {
            scheduler_turn_cost: std::num::NonZeroU64::new(crate::config::MAX_SCHEDULER_TURN_COST),
            ..Config::default()
        };
        let mut time = GlobalTime::new(&config);
        let mut previous = time.as_nanos();
        for _ in 0..1_000 {
            let now = time.add_scheduler_time().unwrap();
            assert_eq!(
                now,
                previous + LogicalTime::from_nanos(crate::config::MAX_SCHEDULER_TURN_COST)
            );
            previous = now;
        }
    }

    /// With a configured charge, the turn that would pass the end of
    /// representable time is refused and adds nothing; it does not saturate.
    #[test]
    fn a_configured_turn_charge_refuses_to_saturate_the_clock() {
        // Three seconds before u64::MAX nanoseconds after the Unix epoch.
        let config = Config {
            scheduler_turn_cost: std::num::NonZeroU64::new(crate::config::MAX_SCHEDULER_TURN_COST),
            epoch: "2554-07-21T23:34:30Z".parse().unwrap(),
            ..Config::default()
        };
        let mut time = GlobalTime::new(&config);
        let mut previous = time.as_nanos();
        for _ in 0..3 {
            let now = time.add_scheduler_time().unwrap();
            assert_eq!(
                now,
                previous + LogicalTime::from_nanos(crate::config::MAX_SCHEDULER_TURN_COST)
            );
            previous = now;
        }
        assert_eq!(
            time.add_scheduler_time(),
            Err(ClockExhausted {
                charge: crate::config::MAX_SCHEDULER_TURN_COST,
                total: previous,
            })
        );
        assert_eq!(time.as_nanos(), previous, "a refused turn adds nothing");
    }

    /// A capped turn charges the smaller of its cap and the full charge, and
    /// at least one nanosecond, with the default and a configured cost alike.
    #[test]
    fn a_capped_turn_charges_its_cap_bounded_by_the_full_charge_and_one_nanosecond() {
        let advance = |config: &Config, cap: u64| {
            let mut time = GlobalTime::new(config);
            let start = time.as_nanos();
            time.add_capped_scheduler_time(cap).unwrap() - start
        };
        let default = Config::default();
        let configured = Config {
            scheduler_turn_cost: std::num::NonZeroU64::new(10_000),
            ..Config::default()
        };
        for (config, full) in [(&default, NANOS_PER_SCHED as u64), (&configured, 10_000)] {
            assert_eq!(advance(config, 1_000), LogicalTime::from_nanos(1_000));
            assert_eq!(advance(config, full - 1), LogicalTime::from_nanos(full - 1));
            assert_eq!(advance(config, full), LogicalTime::from_nanos(full));
            assert_eq!(advance(config, full + 1), LogicalTime::from_nanos(full));
            assert_eq!(advance(config, u64::MAX), LogicalTime::from_nanos(full));
            assert_eq!(advance(config, 0), LogicalTime::from_nanos(1));
        }
        let mut uncapped = GlobalTime::new(&default);
        let start = uncapped.as_nanos();
        assert_eq!(
            uncapped.add_scheduler_time().unwrap() - start,
            LogicalTime::from_nanos(NANOS_PER_SCHED as u64)
        );
    }

    /// A capped turn that would pass the end of representable time under a
    /// configured cost is refused with its capped charge, and adds nothing.
    #[test]
    fn a_capped_turn_refuses_to_saturate_the_clock_with_its_capped_charge() {
        let config = Config {
            scheduler_turn_cost: std::num::NonZeroU64::new(crate::config::MAX_SCHEDULER_TURN_COST),
            epoch: "2554-07-21T23:34:33Z".parse().unwrap(),
            ..Config::default()
        };
        let mut time = GlobalTime::new(&config);
        let headroom = u64::MAX - time.as_nanos().as_nanos();
        time.add_capped_scheduler_time(headroom).unwrap();
        let end = time.as_nanos();
        assert_eq!(end, LogicalTime::MAX);
        assert_eq!(
            time.add_capped_scheduler_time(5),
            Err(ClockExhausted {
                charge: 5,
                total: end,
            })
        );
        assert_eq!(time.as_nanos(), end, "a refused turn adds nothing");
    }

    /// Without `--scheduler-turn-cost` the default charge saturates at the
    /// end of representable time, as it always did, rather than fail.
    #[test]
    fn the_default_turn_charge_still_saturates() {
        let config = Config {
            epoch: "2554-07-21T23:34:33Z".parse().unwrap(),
            ..Config::default()
        };
        let mut time = GlobalTime::new(&config);
        for _ in 0..2_000 {
            time.add_scheduler_time().unwrap();
        }
        assert_eq!(time.as_nanos(), LogicalTime::MAX);
    }

    #[test]
    fn exec_preserves_inherited_baselines_and_each_retired_threads_work() {
        let config = Config::default();
        let ancestor = DetTid::from_raw(3);
        let leader = DetTid::from_raw(4);
        let worker = DetTid::from_raw(5);
        let child_after_exec = DetTid::from_raw(6);
        let mut time = GlobalTime::new(&config);
        let start = time.as_nanos();
        let mut ancestor_clock = DetTime::new(&config);
        ancestor_clock.advance_to(start + LogicalTime::from_nanos(1_000));
        publish(&mut time, ancestor, &ancestor_clock);
        let mut leader_clock = ancestor_clock.clone_for_child();
        leader_clock.advance_to(start + LogicalTime::from_nanos(1_100));
        publish(&mut time, leader, &leader_clock);
        let mut worker_clock = leader_clock.clone_for_child();
        worker_clock.advance_to(start + LogicalTime::from_nanos(1_350));
        publish(&mut time, worker, &worker_clock);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_350));

        time.reassign_thread(worker, leader);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_350));
        assert_eq!(time.threads_time(leader), worker_clock.as_nanos());
        assert!(!time.contains_thread(worker));

        // An in-process backend reloads DetTime after exec and restores its
        // absolute clock from the RPC response. The global component must keep
        // the pre-exec baseline even though this fresh local field is zero.
        let mut reloaded = DetTime::new(&config);
        reloaded.advance_to(time.threads_time(leader));
        assert_eq!(reloaded.inherited_nanos(), LogicalTime::ZERO);
        publish(&mut time, leader, &reloaded);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_350));
        reloaded.advance_to(reloaded.as_nanos() + LogicalTime::from_nanos(1));
        publish(&mut time, leader, &reloaded);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_351));
        // A subsequent leader exec retains the same component and baseline.
        time.reassign_thread(leader, leader);
        let mut leader_reload = DetTime::new(&config);
        leader_reload.advance_to(time.threads_time(leader));
        publish(&mut time, leader, &leader_reload);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_351));
        let mut after_exec = reloaded.clone_for_child();
        publish(&mut time, child_after_exec, &after_exec);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_351));
        after_exec.advance_to(after_exec.as_nanos() + LogicalTime::from_nanos(1));
        publish(&mut time, child_after_exec, &after_exec);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_352));
    }

    #[test]
    fn inherited_clock_survives_rpc_serialization_and_legacy_json_defaults() {
        let mut parent = DetTime::zero();
        parent.add_syscall_with_cost(13);
        let child = parent.clone_for_child();
        // The following tuple field detects a skipped positional clock field,
        // which would otherwise consume bytes from the RPC request.
        let wire = bincode::serde::encode_to_vec(
            (child.clone(), 0x1234_5678_u64),
            bincode::config::legacy(),
        )
        .unwrap();
        let ((restored, following), consumed): ((DetTime, u64), usize) =
            bincode::serde::decode_from_slice(&wire, bincode::config::legacy()).unwrap();
        assert_eq!(consumed, wire.len());
        assert_eq!(following, 0x1234_5678);
        assert_eq!(restored.as_nanos(), child.as_nanos());
        assert_eq!(restored.inherited_nanos(), LogicalTime::from_nanos(13));

        let mut legacy = serde_json::to_value(parent.clone()).unwrap();
        legacy.as_object_mut().unwrap().remove("inherited_nanos");
        let restored: DetTime = serde_json::from_value(legacy).unwrap();
        assert_eq!(restored.as_nanos(), parent.as_nanos());
        assert_eq!(restored.inherited_nanos(), LogicalTime::ZERO);
    }

    #[test]
    #[should_panic(expected = "beyond its local duration")]
    fn inherited_baseline_cannot_exceed_the_first_local_duration() {
        let config = Config::default();
        let mut time = GlobalTime::new(&config);
        time.update_global_time(
            DetTid::from_raw(3),
            time.as_nanos(),
            LogicalTime::from_nanos(1),
        );
    }

    #[test]
    fn exec_reassigns_survivor_clock_without_losing_aggregate_time() {
        let config = Config::default();
        let mut time = GlobalTime::new(&config);
        let leader = DetTid::from_raw(17);
        let worker = DetTid::from_raw(18);
        let start = DetTime::new(&config).as_nanos();
        let leader_time = start + LogicalTime::from_nanos(100);
        let worker_time = start + LogicalTime::from_nanos(250);
        time.update_global_time(leader, leader_time, LogicalTime::ZERO);
        time.update_global_time(worker, worker_time, LogicalTime::ZERO);
        let total_before = time.as_nanos();

        time.reassign_thread(worker, leader);

        assert_eq!(time.as_nanos(), total_before);
        assert_eq!(time.threads_time(leader), worker_time);
        assert!(time.contains_thread(leader));
        assert!(!time.contains_thread(worker));
    }

    /// Linux gives an exited thread's ID to a later thread once IDs wrap at
    /// `pid_max`. The later thread starts at its creator's time with a
    /// component of its own: none of its inherited history is charged, and
    /// the first holder's own work stays in aggregate time.
    #[test]
    fn a_reused_thread_id_starts_its_own_clock_and_keeps_the_old_work() {
        let config = Config::default();
        let mut time = GlobalTime::new(&config);
        let start = time.as_nanos();
        let parent = DetTid::from_raw(17);
        let reused = DetTid::from_raw(18);
        let mut parent_clock = DetTime::new(&config);
        parent_clock.add_syscall_with_cost(100);
        publish(&mut time, parent, &parent_clock);
        let mut first_holder = parent_clock.clone_for_child();
        first_holder.add_syscall_with_cost(50);
        publish(&mut time, reused, &first_holder);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(150));

        time.retire_thread(reused);
        assert!(!time.contains_thread(reused));
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(150));

        parent_clock.add_syscall_with_cost(1_000);
        publish(&mut time, parent, &parent_clock);
        let mut second_holder = parent_clock.clone_for_child();
        publish(&mut time, reused, &second_holder);
        assert_eq!(time.threads_time(reused), second_holder.as_nanos());
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_150));
        second_holder.add_syscall_with_cost(7);
        publish(&mut time, reused, &second_holder);
        assert_eq!(time.as_nanos(), start + LogicalTime::from_nanos(1_157));
    }
}