From 5a322b6cb18417f563cf656a0a0a9949189195de Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 14:20:48 +0530
Subject: [PATCH 1/7] Let crashed hosts restart as fresh incarnations

---
 src/simloop/_net.py   |  24 +++++++
 tests/test_restart.py | 149 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 173 insertions(+)
 create mode 100644 tests/test_restart.py

diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index 99d8aa5..35d9a44 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -229,6 +229,9 @@ def create_task(self, coro: Any, *, name: str | None = None) -> asyncio.Task[Any
     def crash(self) -> None:
         self._net.crash(self._name)
 
+    def restart(self) -> None:
+        self._net.restart(self._name)
+
 
 class SimNetwork:
     """Registry of hosts and the traffic between them."""
@@ -756,6 +759,27 @@ def crash(self, name: str) -> None:
             else:
                 kept.append(packet)
         self._held = kept
+        self._loop._recorder.record(
+            "net", self._loop.time(), self._new_uid(), f"crash {name}"
+        )
+
+    def restart(self, name: str) -> None:
+        """Bring a crashed host back as a fresh incarnation.
+
+        Restart revives liveness and nothing else: the old incarnation's
+        tasks are already cancelled, its listeners and binds are gone, and
+        its stream connections are dead — packets addressed to them vanish,
+        so peers still learn about the outage only from their own timeouts.
+        The caller boots whatever should run on the revived machine, the
+        same way it booted the machine the first time.
+        """
+        self._require_host(name)
+        if self._alive[name]:
+            raise ValueError(f"host {name!r} is not crashed")
+        self._alive[name] = True
+        self._loop._recorder.record(
+            "net", self._loop.time(), self._new_uid(), f"restart {name}"
+        )
 
     def _register_task(self, task: asyncio.Task[Any]) -> None:
         owner = self._tasks[_current_host.get()]
diff --git a/tests/test_restart.py b/tests/test_restart.py
new file mode 100644
index 0000000..2094973
--- /dev/null
+++ b/tests/test_restart.py
@@ -0,0 +1,149 @@
+"""A crashed host coming back: fresh incarnation, same address, dead past."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+from simloop import SimLoop
+
+
+def _network(seed: int = 0) -> SimLoop:
+    loop = SimLoop(seed=seed)
+    loop.net.host("server")
+    loop.net.host("client")
+    loop.net.set_defaults(latency=(0.001, 0.001))
+    return loop
+
+
+async def _echo_once(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+    writer.write(await reader.readline())
+    writer.close()
+
+
+async def _ask(text: bytes) -> bytes:
+    reader, writer = await asyncio.open_connection("server", 9000)
+    writer.write(text)
+    reply = await reader.readline()
+    writer.close()
+    return reply
+
+
+def test_restarted_host_serves_again_on_the_same_port() -> None:
+    loop = _network()
+
+    async def main() -> tuple[bytes, bytes]:
+        async def boot() -> None:
+            await asyncio.start_server(_echo_once, "0.0.0.0", 9000)
+
+        await loop.net.host("server").create_task(boot())
+        await asyncio.sleep(0.01)
+        first = await loop.net.host("client").create_task(_ask(b"one\n"))
+        loop.net.crash("server")
+        loop.net.restart("server")
+        await loop.net.host("server").create_task(boot())
+        await asyncio.sleep(0.01)
+        second = await loop.net.host("client").create_task(_ask(b"two\n"))
+        return first, second
+
+    try:
+        first, second = loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert (first, second) == (b"one\n", b"two\n")
+
+
+def test_restart_requires_a_crash_first() -> None:
+    loop = _network()
+    with pytest.raises(ValueError, match="not crashed"):
+        loop.net.restart("server")
+    loop.close()
+
+
+def test_restart_requires_a_known_host() -> None:
+    loop = _network()
+    with pytest.raises(OSError, match="ghost"):
+        loop.net.restart("ghost")
+    loop.close()
+
+
+def test_dead_window_traffic_is_lost_but_new_traffic_flows() -> None:
+    loop = _network()
+    heard: list[bytes] = []
+
+    class Listener(asyncio.DatagramProtocol):
+        def datagram_received(self, data: bytes, addr: Any) -> None:
+            heard.append(data)
+
+    async def main() -> None:
+        async def bind() -> None:
+            await loop.create_datagram_endpoint(Listener, local_addr=("0.0.0.0", 5000))
+
+        async def send(payload: bytes) -> None:
+            transport, _ = await loop.create_datagram_endpoint(
+                asyncio.DatagramProtocol, remote_addr=("server", 5000)
+            )
+            transport.sendto(payload)
+            transport.close()
+
+        await loop.net.host("server").create_task(bind())
+        loop.net.crash("server")
+        await loop.net.host("client").create_task(send(b"into the void"))
+        await asyncio.sleep(0.01)  # delivery attempt lands while dead
+        loop.net.restart("server")
+        await loop.net.host("server").create_task(bind())
+        await loop.net.host("client").create_task(send(b"welcome back"))
+        await asyncio.sleep(0.01)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert heard == [b"welcome back"]
+
+
+def test_crash_and_restart_appear_in_the_trace() -> None:
+    loop = _network()
+
+    async def main() -> None:
+        loop.net.crash("server")
+        loop.net.restart("server")
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    events = [
+        event
+        for event in loop.trace
+        if "crash server" in event.label or "restart server" in event.label
+    ]
+    assert len(events) == 2
+
+
+def test_same_seed_replays_a_restart_schedule_identically() -> None:
+    def run() -> str:
+        loop = _network(seed=7)
+
+        async def main() -> None:
+            async def boot() -> None:
+                await asyncio.start_server(_echo_once, "0.0.0.0", 9000)
+
+            await loop.net.host("server").create_task(boot())
+            await asyncio.sleep(0.01)
+            await loop.net.host("client").create_task(_ask(b"a\n"))
+            loop.net.crash("server")
+            loop.net.restart("server")
+            await loop.net.host("server").create_task(boot())
+            await asyncio.sleep(0.01)
+            await loop.net.host("client").create_task(_ask(b"b\n"))
+
+        try:
+            loop.run_until_complete(main())
+            return loop.trace_hash()
+        finally:
+            loop.close()
+
+    assert run() == run()

From 8ed5e99370680b6946996dd47c4821f5e014efcd Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 14:30:45 +0530
Subject: [PATCH 2/7] Give every host a disk that survives its crashes

---
 src/simloop/__init__.py |  3 ++-
 src/simloop/_net.py     | 42 +++++++++++++++++++++++++++++---
 tests/test_disk.py      | 54 +++++++++++++++++++++++++++++++++++++++++
 3 files changed, 95 insertions(+), 4 deletions(-)
 create mode 100644 tests/test_disk.py

diff --git a/src/simloop/__init__.py b/src/simloop/__init__.py
index cb001ba..7dfb91d 100644
--- a/src/simloop/__init__.py
+++ b/src/simloop/__init__.py
@@ -2,7 +2,7 @@
 
 from simloop._explore import SeedReport, explore, sim_test
 from simloop._loop import SimLoop, SimulationDeadlockError, SimulationFenceError
-from simloop._net import Host, SimNetwork
+from simloop._net import Host, SimDisk, SimNetwork
 from simloop._sim import Sim, sim
 from simloop._trace import TraceEvent
 
@@ -12,6 +12,7 @@
     "Host",
     "SeedReport",
     "Sim",
+    "SimDisk",
     "SimLoop",
     "SimNetwork",
     "SimulationDeadlockError",
diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index 35d9a44..7ab2a0f 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -11,7 +11,7 @@
 
 import random
 import socket
-from collections.abc import Iterable
+from collections.abc import Iterable, Iterator, MutableMapping
 from contextvars import ContextVar
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
@@ -208,6 +208,36 @@ def _check_latency(value: tuple[float, float]) -> tuple[float, float]:
     return (lo, hi)
 
 
+class SimDisk(MutableMapping[str, object]):
+    """A host's storage that survives crashes and restarts.
+
+    A crash loses everything volatile — tasks, connections, binds — but not
+    what was written here, which is the whole point: this is where state
+    that a real process would fsync belongs. Writes are atomic at
+    assignment; there is no partial-write model. Values are stored as
+    given, so mutating a stored object later is the caller's own aliasing,
+    exactly as it would be with a cache in front of a real disk.
+    """
+
+    def __init__(self) -> None:
+        self._data: dict[str, object] = {}
+
+    def __getitem__(self, key: str) -> object:
+        return self._data[key]
+
+    def __setitem__(self, key: str, value: object) -> None:
+        self._data[key] = value
+
+    def __delitem__(self, key: str) -> None:
+        del self._data[key]
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self._data)
+
+    def __len__(self) -> int:
+        return len(self._data)
+
+
 class Host:
     """Handle for one simulated machine; tasks started here are pinned to it."""
 
@@ -219,6 +249,10 @@ def __init__(self, net: SimNetwork, name: str) -> None:
     def name(self) -> str:
         return self._name
 
+    @property
+    def disk(self) -> SimDisk:
+        return self._net._disks.setdefault(self._name, SimDisk())
+
     def create_task(self, coro: Any, *, name: str | None = None) -> asyncio.Task[Any]:
         token = _current_host.set(self._name)
         try:
@@ -245,6 +279,7 @@ def __init__(self, loop: SimLoop) -> None:
         self._addresses: dict[str, str] = {}
         self._names: dict[str, str] = {}
         self._alive: dict[str, bool] = {}
+        self._disks: dict[str, SimDisk] = {}
         self._tasks: dict[str, list[asyncio.Task[Any]]] = {}
         self._default_latency: tuple[float, float] = (0.0, 0.0)
         self._default_drop = 0.0
@@ -770,8 +805,9 @@ def restart(self, name: str) -> None:
         tasks are already cancelled, its listeners and binds are gone, and
         its stream connections are dead — packets addressed to them vanish,
         so peers still learn about the outage only from their own timeouts.
-        The caller boots whatever should run on the revived machine, the
-        same way it booted the machine the first time.
+        State meant to survive the reboot belongs on ``Host.disk``. The
+        caller boots whatever should run on the revived machine, the same
+        way it booted the machine the first time.
         """
         self._require_host(name)
         if self._alive[name]:
diff --git a/tests/test_disk.py b/tests/test_disk.py
new file mode 100644
index 0000000..6ce1f03
--- /dev/null
+++ b/tests/test_disk.py
@@ -0,0 +1,54 @@
+"""Per-host storage that survives the machine crashing."""
+
+from __future__ import annotations
+
+import pytest
+
+from simloop import SimLoop
+
+
+def _network() -> SimLoop:
+    loop = SimLoop(seed=0)
+    loop.net.host("server")
+    loop.net.host("client")
+    return loop
+
+
+def test_disk_survives_crash_and_restart() -> None:
+    loop = _network()
+    loop.net.host("server").disk["term"] = 7
+    loop.net.crash("server")
+    loop.net.restart("server")
+    assert loop.net.host("server").disk["term"] == 7
+    loop.close()
+
+
+def test_disks_are_per_host() -> None:
+    loop = _network()
+    loop.net.host("server").disk["k"] = "s"
+    loop.net.host("client").disk["k"] = "c"
+    assert loop.net.host("server").disk["k"] == "s"
+    assert loop.net.host("client").disk["k"] == "c"
+    loop.close()
+
+
+def test_disk_is_a_real_mapping() -> None:
+    loop = _network()
+    disk = loop.net.host("server").disk
+    disk["a"] = 1
+    disk["b"] = 2
+    assert len(disk) == 2
+    assert sorted(disk) == ["a", "b"]
+    del disk["a"]
+    with pytest.raises(KeyError):
+        disk["a"]
+    disk.clear()
+    assert len(disk) == 0
+    loop.close()
+
+
+def test_the_driver_has_a_disk_too() -> None:
+    loop = _network()
+    loop.net.host("driver").disk["x"] = 1
+    assert loop.net.host("driver").disk["x"] == 1
+    loop.close()

From b4e98cc531110a9280250d290d3c9fdc9b2d09ec Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 14:45:08 +0530
Subject: [PATCH 3/7] Let each host read its own version of the clock

---
 src/simloop/_loop.py     |  49 ++++++++-
 src/simloop/_net.py      |  35 +++++-
 tests/test_clock_skew.py | 228 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 307 insertions(+), 5 deletions(-)
 create mode 100644 tests/test_clock_skew.py

diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py
index f2ba3e2..a48a303 100644
--- a/src/simloop/_loop.py
+++ b/src/simloop/_loop.py
@@ -185,6 +185,18 @@ def net(self) -> SimNetwork:
     # ------------------------------------------------------------------
 
     def time(self) -> float:
+        # What the calling task's machine believes the time is. Everything
+        # internal — timer ordering, deadline advance, trace timestamps —
+        # uses the true clock (self._now) directly, so skew never perturbs
+        # scheduling and traces from skewed runs stay comparable.
+        return self._now + self._net._offset_now()
+
+    def _true_time(self) -> float:
+        """The shared virtual clock, the same reading for whoever asks.
+
+        The clock the simulation itself runs on, for the internals that must
+        not see a host's skew — the network's trace timestamps above all.
+        """
         return self._now
 
     def call_soon(
@@ -209,7 +221,14 @@ def call_later(
         *args: Unpack[_Ts],
         context: Context | None = None,
     ) -> asyncio.TimerHandle:
-        return self.call_at(self._now + delay, callback, *args, context=context)
+        # A delay is a duration, so it is measured on the true clock and no
+        # caller's offset touches it: a wrong wall clock does not make a
+        # second take longer. Going through call_at instead would subtract
+        # the caller's offset from a deadline that is already true.
+        when = self._now + delay
+        return self._call_at_true(
+            when, when + self._net._offset_now(), callback, args, context
+        )
 
     def call_at(
         self,
@@ -218,8 +237,34 @@ def call_at(
         *args: Unpack[_Ts],
         context: Context | None = None,
     ) -> asyncio.TimerHandle:
+        # ``when`` is a reading of the *scheduling* task's clock, so it is
+        # converted to true time once, here, with that task's offset. The
+        # callback runs later under the context asyncio copied into the
+        # handle — the scheduling task's — so it reads the same clock the
+        # scheduler did, which is coherent for the cases that matter:
+        # asyncio.timeout, sleep and protocol timers all schedule from the
+        # task they serve. Nothing re-converts per callback.
+        return self._call_at_true(
+            when - self._net._offset_now(), when, callback, args, context
+        )
+
+    def _call_at_true(
+        self,
+        when: float,
+        shown: float,
+        callback: Callable[[Unpack[_Ts]], object],
+        args: tuple[Unpack[_Ts]],
+        context: Context | None,
+    ) -> asyncio.TimerHandle:
+        """Push a timer whose ``when`` is already true time.
+
+        The heap orders on true time — the one clock the loop itself runs on
+        — while the handle advertises ``shown``, the same deadline on the
+        clock the caller reads, so ``timer.when()`` stays comparable with
+        that task's ``loop.time()``.
+        """
         self._check_closed()
-        timer = asyncio.TimerHandle(when, callback, args, self, context)
+        timer = asyncio.TimerHandle(shown, callback, args, self, context)
         seq = self._next_seq
         self._next_seq += 1
         label = _label(callback)
diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index 7ab2a0f..00b1f5d 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -280,6 +280,7 @@ def __init__(self, loop: SimLoop) -> None:
         self._names: dict[str, str] = {}
         self._alive: dict[str, bool] = {}
         self._disks: dict[str, SimDisk] = {}
+        self._clock_offsets: dict[str, float] = {}
         self._tasks: dict[str, list[asyncio.Task[Any]]] = {}
         self._default_latency: tuple[float, float] = (0.0, 0.0)
         self._default_drop = 0.0
@@ -436,6 +437,26 @@ def set_link(
         if duplicate is not None:
             link.duplicate = _check_probability("duplicate", duplicate)
 
+    def set_clock(self, name: str, *, offset: float) -> None:
+        """Skew what a host's tasks read from the clock, in seconds.
+
+        Offset changes what ``loop.time()`` *reads* on that host — never how
+        long a duration takes: ``asyncio.sleep(1.0)`` still costs one true
+        second everywhere, which is what a wrong wall clock does on a real
+        machine. Deadlines passed to ``call_at`` are interpreted on the
+        calling task's clock. The driver and unconfigured hosts read true
+        time.
+        """
+        self._require_host(name)
+        self._clock_offsets[name] = float(offset)
+
+    def clock_offset(self, name: str) -> float:
+        self._require_host(name)
+        return self._clock_offsets.get(name, 0.0)
+
+    def _offset_now(self) -> float:
+        return self._clock_offsets.get(_current_host.get(), 0.0)
+
     def partition(self, group_a: Iterable[str], group_b: Iterable[str]) -> None:
         side_a = [self._require_host(name) for name in group_a]
         side_b = [self._require_host(name) for name in group_b]
@@ -497,8 +518,16 @@ def _ephemeral(self) -> int:
         return port
 
     def _trace(self, verb: str, packet: _Packet) -> None:
+        # Trace timestamps are always the true clock, never the calling
+        # host's: a packet event happens once, at one shared instant, and
+        # traces from differently-skewed runs have to stay comparable.
+        # _deliver and the transports run under a host context, so
+        # loop.time() here would read that host's skewed clock.
         self._loop._recorder.record(
-            "net", self._loop.time(), packet.uid, f"{verb} {packet.src}>{packet.dst}"
+            "net",
+            self._loop._true_time(),
+            packet.uid,
+            f"{verb} {packet.src}>{packet.dst}",
         )
 
     def _transmit(self, packet: _Packet) -> None:
@@ -795,7 +824,7 @@ def crash(self, name: str) -> None:
                 kept.append(packet)
         self._held = kept
         self._loop._recorder.record(
-            "net", self._loop.time(), self._new_uid(), f"crash {name}"
+            "net", self._loop._true_time(), self._new_uid(), f"crash {name}"
         )
 
     def restart(self, name: str) -> None:
@@ -814,7 +843,7 @@ def restart(self, name: str) -> None:
             raise ValueError(f"host {name!r} is not crashed")
         self._alive[name] = True
         self._loop._recorder.record(
-            "net", self._loop.time(), self._new_uid(), f"restart {name}"
+            "net", self._loop._true_time(), self._new_uid(), f"restart {name}"
         )
 
     def _register_task(self, task: asyncio.Task[Any]) -> None:
diff --git a/tests/test_clock_skew.py b/tests/test_clock_skew.py
new file mode 100644
index 0000000..f553610
--- /dev/null
+++ b/tests/test_clock_skew.py
@@ -0,0 +1,228 @@
+"""Per-host clocks that read differently while true time stays shared."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from simloop import SimLoop
+
+
+def _network() -> SimLoop:
+    loop = SimLoop(seed=0)
+    loop.net.host("broker")
+    loop.net.host("worker")
+    return loop
+
+
+def test_each_host_reads_its_own_clock() -> None:
+    loop = _network()
+    loop.net.set_clock("worker", offset=2.5)
+
+    async def read_on(host: str) -> float:
+        reading: asyncio.Task[float] = loop.net.host(host).create_task(_read())
+        return await reading
+
+    async def _read() -> float:
+        return loop.time()
+
+    async def main() -> tuple[float, float, float]:
+        return (
+            await read_on("broker"),
+            await read_on("worker"),
+            loop.time(),  # driver, unskewed
+        )
+
+    broker, worker, driver = loop.run_until_complete(main())
+    loop.close()
+    assert worker == pytest.approx(broker + 2.5)
+    assert driver == pytest.approx(broker)
+
+
+def test_clock_offset_reads_back_what_was_set() -> None:
+    loop = _network()
+    assert loop.net.clock_offset("worker") == 0.0
+    loop.net.set_clock("worker", offset=-1.5)
+    assert loop.net.clock_offset("worker") == -1.5
+    assert loop.net.clock_offset("broker") == 0.0
+    loop.close()
+
+
+def test_durations_are_immune_to_offset() -> None:
+    loop = _network()
+    loop.net.set_clock("worker", offset=100.0)
+
+    async def timed_sleep() -> float:
+        before = loop.time()
+        await asyncio.sleep(1.0)
+        return loop.time() - before
+
+    async def timed_timeout() -> float:
+        before = loop.time()
+        with pytest.raises(TimeoutError):
+            async with asyncio.timeout(1.0):
+                await asyncio.sleep(10.0)
+        return loop.time() - before
+
+    async def timed_wait_for() -> float:
+        before = loop.time()
+        with pytest.raises(TimeoutError):
+            await asyncio.wait_for(asyncio.sleep(10.0), 1.0)
+        return loop.time() - before
+
+    async def main() -> tuple[float, float, float]:
+        host = loop.net.host("worker")
+        slept = await host.create_task(timed_sleep())
+        timed = await host.create_task(timed_timeout())
+        waited = await host.create_task(timed_wait_for())
+        return slept, timed, waited
+
+    slept, timed, waited = loop.run_until_complete(main())
+    loop.close()
+    assert slept == pytest.approx(1.0)
+    assert timed == pytest.approx(1.0)
+    assert waited == pytest.approx(1.0)
+
+
+def test_call_at_means_the_callers_clock() -> None:
+    loop = _network()
+    loop.net.set_clock("worker", offset=10.0)
+    fired_at: list[float] = []
+
+    async def schedule() -> None:
+        # "at 11 on my clock" = at 1 true
+        loop.call_at(loop.time() + 1.0, lambda: fired_at.append(loop.time()))
+        await asyncio.sleep(2.0)
+
+    loop.run_until_complete(loop.net.host("worker").create_task(schedule()))
+    loop.close()
+    assert len(fired_at) == 1
+    # The callback carries the scheduling task's context (asyncio copies the
+    # current context into the handle), so it reads the worker's clock as
+    # well: it sees exactly the deadline it was given.
+    assert fired_at[0] == pytest.approx(11.0)
+    # And that deadline is one true second after scheduling, not eleven.
+    assert fired_at[0] - loop.net.clock_offset("worker") == pytest.approx(1.0)
+
+
+def test_call_later_ignores_the_offset() -> None:
+    loop = _network()
+    loop.net.set_clock("worker", offset=10.0)
+    fired_at: list[float] = []
+
+    async def schedule() -> None:
+        loop.call_later(1.0, lambda: fired_at.append(loop.time()))
+        await asyncio.sleep(2.0)
+
+    loop.run_until_complete(loop.net.host("worker").create_task(schedule()))
+    loop.close()
+    # A delay is a duration: one true second later, which the worker reads
+    # as 11.0 because its clock runs ten seconds fast.
+    assert fired_at == [pytest.approx(11.0)]
+
+
+def test_a_deadline_from_the_driver_stays_true() -> None:
+    loop = _network()
+    loop.net.set_clock("worker", offset=10.0)
+    fired_at: list[float] = []
+
+    async def main() -> None:
+        # Scheduled from the driver, so 1.0 means 1.0 true.
+        loop.call_at(1.0, lambda: fired_at.append(loop.time()))
+        await asyncio.sleep(2.0)
+
+    loop.run_until_complete(main())
+    loop.close()
+    assert fired_at == [pytest.approx(1.0)]
+
+
+def test_hosts_disagree_about_lease_expiry() -> None:
+    loop = _network()
+    loop.net.set_clock("worker", offset=2.0)
+
+    async def broker_grants() -> float:
+        return loop.time() + 1.0  # lease valid one second, broker clock
+
+    async def worker_checks(expiry: float) -> bool:
+        return loop.time() < expiry
+
+    async def main() -> bool:
+        granting: asyncio.Task[float] = loop.net.host("broker").create_task(
+            broker_grants()
+        )
+        expiry = await granting
+        checking: asyncio.Task[bool] = loop.net.host("worker").create_task(
+            worker_checks(expiry)
+        )
+        return await checking
+
+    holds = loop.run_until_complete(main())
+    loop.close()
+    assert not holds  # the worker's fast clock already sees it expired
+
+
+def test_set_clock_validates() -> None:
+    loop = _network()
+    with pytest.raises(OSError):
+        loop.net.set_clock("ghost", offset=1.0)
+    with pytest.raises(OSError):
+        loop.net.clock_offset("ghost")
+    loop.close()
+
+
+def test_zero_skew_traces_match_a_loop_that_never_heard_of_skew() -> None:
+    def run(configure: bool) -> str:
+        loop = _network()
+        if configure:
+            loop.net.set_clock("worker", offset=0.0)
+
+        async def main() -> None:
+            async def _echo(r: asyncio.StreamReader, w: asyncio.StreamWriter) -> None:
+                w.write(await r.readline())
+                w.close()
+
+            async def serve() -> None:
+                server = await asyncio.start_server(_echo, "0.0.0.0", 9000)
+                async with server:
+                    await asyncio.sleep(0.5)
+
+            async def ask() -> None:
+                reader, writer = await asyncio.open_connection("broker", 9000)
+                writer.write(b"x\n")
+                await reader.readline()
+                writer.close()
+
+            serving = loop.net.host("broker").create_task(serve())
+            await asyncio.sleep(0.01)
+            await loop.net.host("worker").create_task(ask())
+            await serving
+
+        try:
+            loop.run_until_complete(main())
+            return loop.trace_hash()
+        finally:
+            loop.close()
+
+    assert run(False) == run(True)
+
+
+def test_skew_leaves_the_trace_on_the_true_clock() -> None:
+    def run(offset: float) -> str:
+        loop = _network()
+        loop.net.set_clock("worker", offset=offset)
+
+        async def work() -> None:
+            await asyncio.sleep(1.0)
+            async with asyncio.timeout(2.0):
+                await asyncio.sleep(0.5)
+
+        loop.run_until_complete(loop.net.host("worker").create_task(work()))
+        try:
+            return loop.trace_hash()
+        finally:
+            loop.close()
+
+    # Durations and trace timestamps are both true-clock, so a skewed run
+    # traces byte-identically to an unskewed one.
+    assert run(0.0) == run(3600.0) == run(-3600.0)

From e7fa8aa759685544b7ebdc80aa908f26c97e1d7a Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 15:02:08 +0530
Subject: [PATCH 4/7] Show lease safeguards absorbing a fast worker clock

---
 examples/jobqueue/tests/test_lease_skew.py | 119 +++++++++++++++++++++
 1 file changed, 119 insertions(+)
 create mode 100644 examples/jobqueue/tests/test_lease_skew.py

diff --git a/examples/jobqueue/tests/test_lease_skew.py b/examples/jobqueue/tests/test_lease_skew.py
new file mode 100644
index 0000000..dbc7756
--- /dev/null
+++ b/examples/jobqueue/tests/test_lease_skew.py
@@ -0,0 +1,119 @@
+"""Leases held by a worker whose clock runs two leases fast.
+
+The broker is the only clock that decides expiry, and everything the
+protocol puts on the wire is a duration, never a timestamp. These runs
+put a badly skewed worker in the middle of the lease machinery and check
+that the exactly-once story is unchanged.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from simloop import SimLoop, sim_test
+
+import helpers
+
+LEASE_S = 2.0  # the broker's default lease
+SKEW_S = 2 * LEASE_S  # a clock two full leases ahead of the rest of the cluster
+
+
+async def _clock_reading() -> float:
+    return helpers.sim_loop().time()
+
+
+@sim_test(seeds=25)
+async def test_a_fast_clock_does_not_cost_the_worker_its_lease() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=1, clients=1)
+    loop.net.set_clock("w1", offset=SKEW_S)
+    # The skew is real from inside the worker: its tasks read a clock two
+    # leases ahead of the broker's.
+    worker_now = await loop.net.host("w1").create_task(_clock_reading())
+    assert worker_now - loop.time() == pytest.approx(SKEW_S)
+    job_id = await loop.net.host("c1").create_task(
+        cluster.clients[0].submit("fast-clock", duration=1.5 * LEASE_S)
+    )
+    assert job_id is not None
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    # Renewals are heartbeats on a timer, not deadline arithmetic, so the
+    # skewed worker keeps the lease it was granted: one attempt, one commit,
+    # no re-run to fence.
+    assert cluster.broker.snapshot()[job_id] == ("done", 1)
+    assert [(c.token, c.value) for c in cluster.store.commits] == [(1, "fast-clock")]
+    assert cluster.store.rejected == []
+
+
+@sim_test(seeds=50)
+async def test_a_fast_worker_takes_over_a_lapsed_lease() -> None:
+    loop = helpers.sim_loop()
+    loop.net.host("w2")  # register it early so it can be skewed before it runs
+    loop.net.set_clock("w2", offset=SKEW_S)
+    # w1 is partitioned mid-job and turns into a zombie; the skewed w2 is the
+    # worker that picks the job back up once the broker expires the lease.
+    cluster = await helpers.zombie_run(helpers.EffectStore())
+    helpers.verify(cluster)
+    assert any(reason == "stale" for _, _, reason in cluster.store.rejected)
+    assert len(cluster.store.commits) == 1
+    assert cluster.store.commits[0].token >= 2  # the re-run's lease, not the zombie's
+
+
+@sim_test(seeds=25)
+async def test_skewed_and_honest_workers_share_a_queue() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=2, clients=1)
+    loop.net.set_clock("w2", offset=SKEW_S)
+    client = cluster.clients[0]
+    host = loop.net.host("c1")
+    job_ids = [await host.create_task(client.submit(f"m{i}")) for i in range(4)]
+    assert all(job_id is not None for job_id in job_ids)
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    assert len(cluster.store.commits) == 4
+
+
+async def _skew_workload() -> str:
+    """Run a small cluster to quiesce; return the trace hash of that run."""
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=2, clients=1)
+    client = cluster.clients[0]
+    host = loop.net.host("c1")
+    job_ids = [await host.create_task(client.submit(f"t{i}")) for i in range(2)]
+    assert all(job_id is not None for job_id in job_ids)
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    trace = loop.trace_hash()  # taken at quiesce: the run, not the teardown
+    # Then shut everything down — servers, workers, open connection handlers —
+    # so the loop closes with nothing suspended mid-write.
+    here = asyncio.current_task()
+    running = [task for task in asyncio.all_tasks() if task is not here]
+    for task in running:
+        task.cancel()
+    await asyncio.gather(*running, return_exceptions=True)
+    await asyncio.sleep(0.1)  # let cancelled renew heartbeats finish unwinding
+    return trace
+
+
+def _skew_hash(offset: float) -> str:
+    loop = SimLoop(seed=3)
+    # Both runs register the hosts in the same order; only the offset differs.
+    for name in ("broker", "w1", "w2", "c1"):
+        loop.net.host(name)
+    loop.net.set_clock("w2", offset=offset)
+    try:
+        trace: str = loop.run_until_complete(_skew_workload())
+    finally:
+        loop.close()
+    return trace
+
+
+def test_skew_alone_does_not_change_what_happens() -> None:
+    # Why the skew is absorbed rather than merely survived: no timestamp ever
+    # crosses a host boundary here. The broker sends `lease_s`, a duration;
+    # the worker sleeps and heartbeats on durations; only the broker compares
+    # readings, all of them its own. A worker's wrong clock is therefore
+    # unobservable — the packet trace is byte-identical either way, which is
+    # also why no ablation of *this* app can turn skew into a violation.
+    assert _skew_hash(0.0) == _skew_hash(SKEW_S) == _skew_hash(-SKEW_S)

From b37dc40d27918762e3d4692f2a5d08c267031e55 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 15:13:11 +0530
Subject: [PATCH 5/7] Document restart, disks and skewed clocks

---
 CHANGELOG.md          | 15 +++++++++++++++
 README.md             |  8 ++++++++
 docs/supported-api.md | 16 +++++++++++++++-
 src/simloop/_sim.py   |  5 +++--
 4 files changed, 41 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d17e820..5624913 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,21 @@
 
 ## 0.2.0 (unreleased)
 
+- Crashed hosts can come back: `loop.net.restart(name)` (or
+  `host.restart()`) revives a machine as a fresh incarnation. It restores
+  liveness and nothing else — the old tasks stay cancelled and the
+  listeners are gone, so the caller boots the machine again the way it
+  booted it the first time — and traffic due while the host was dead is
+  lost, leaving peers to notice the outage from their own timeouts.
+- Every host now has `host.disk`, a mapping that survives its crashes:
+  where state a real process would fsync belongs. Writes are atomic at
+  assignment; there is no partial-write model.
+- Clocks can lie per host: `loop.net.set_clock(name, offset=...)` skews
+  what that host's tasks read from `loop.time()`, and the deadlines they
+  hand to `call_at` with it, while durations (`sleep`, `timeout`,
+  `wait_for`, `call_later`) cost the same everywhere — which is what a
+  wrong wall clock does to a real machine. Traces stay on the true clock,
+  and runs that configure no offset are byte-identical to before.
 - A second flagship demo: `examples/raft/` is a teaching-sized Raft (leader
   election + log replication, plain asyncio on streams) tested only under
   simulation — four safety invariants checked over 50,000 chaos seeds, five
diff --git a/README.md b/README.md
index b458766..a40ffb4 100644
--- a/README.md
+++ b/README.md
@@ -171,6 +171,14 @@ loop.call_later(5.0, net.heal)                  # heals in virtual time
 net.crash("node2")                              # no reset, just silence
 ```
 
+- **Crashes with a way back** — `net.crash` kills a host's tasks and
+  binds, `net.restart` brings it back as a fresh incarnation, and
+  `host.disk` is a mapping that outlives both: machines die, reboot, and
+  remember what they wrote down.
+- **Clocks that lie** — `net.set_clock(name, offset=...)` skews what one
+  host reads from the clock without changing how long anything takes, for
+  testing lease and timeout code against machines that disagree about the
+  time.
 - **Replayable traces** — every scheduling and fault decision lands in
   an append-only trace whose hash proves a replay is exact.
 - **Seeded stand-ins** — `sim.random`, `sim.uuid4()` and `sim.time()`
diff --git a/docs/supported-api.md b/docs/supported-api.md
index 5583a04..4f56423 100644
--- a/docs/supported-api.md
+++ b/docs/supported-api.md
@@ -11,7 +11,7 @@ determinism. Fenced APIs raise `SimulationFenceError` (a subclass of
 | API | Behavior under simulation |
 |---|---|
 | `loop.call_soon` / `call_later` / `call_at` | Seeded ready-queue ordering; `(deadline, seq)` timer tie-break |
-| `loop.time()` / `asyncio.sleep` | Virtual clock starting at 0.0; never waits on wall time |
+| `loop.time()` / `asyncio.sleep` | Virtual clock starting at 0.0, shifted by whatever offset the calling task's host is configured with; never waits on wall time |
 | `loop.create_task` / `asyncio.create_task` | Real stdlib `Task`s, including custom task factories |
 | `loop.create_future` | Real stdlib `Future`s |
 | `run_until_complete` / `run_forever` / `stop` / `close` | Deadlock detection: raises `SimulationDeadlockError` when nothing can run |
@@ -40,6 +40,9 @@ a host belong to an implicit `driver` host.
 | `loop.net.set_defaults` / `set_link` | Per-direction latency ranges, drop and duplication probabilities, drawn from a seed-derived stream |
 | `loop.net.partition` / `heal` | Silent blackhole: datagrams are lost, stream traffic is held and resumes intact after healing; nothing errors — only your own timeouts fire |
 | `loop.net.crash` | A host's tasks are cancelled and it goes silent; no reset is sent — peers cannot tell a crash from a partition |
+| `loop.net.restart` / `host.restart()` | The counterpart to a crash: the host comes back as a fresh incarnation. Liveness is all that is revived — the old tasks stay cancelled, its listeners and binds are gone, and the caller boots whatever should run on the machine again, the same way it booted it the first time. A packet is checked against liveness when it arrives, so traffic due during the dead window is lost; a packet that was already in flight and lands after the machine is back is delivered, and finds a host that no longer holds the old incarnation's connections. Peers still learn about the outage only from their own timeouts |
+| `host.disk` | Storage that survives the crash: a `MutableMapping` per host, where state a real process would fsync belongs. Writes are atomic at assignment; there is no partial-write model. Values are stored as given, so mutating a stored object afterwards is the caller's own aliasing, exactly as with a cache in front of a real disk |
+| `loop.net.set_clock` / `clock_offset` | Per-host clock skew, in seconds. The offset changes what that host's tasks *read*: `loop.time()` (and `sim.time()` with it) returns true time plus the offset, and a deadline handed to `call_at` is interpreted on the calling task's clock. Durations are immune — `asyncio.sleep`, `asyncio.timeout`, `wait_for` and `call_later` cost the same everywhere, which is exactly what a wrong wall clock does to a real machine. By default the driver and unconfigured hosts read true time; the driver can be given an offset too. Trace timestamps stay on the true clock, so skew never perturbs scheduling and traces from skewed runs stay comparable |
 | `transport.abort()` | Peer gets `connection_lost(ConnectionResetError)` |
 | `loop.getaddrinfo` | Resolves against the host table, never DNS: a registered host name, its synthetic address, or a loopback-shaped name (`None`, `""`, `localhost`, `127.0.0.1`, `0.0.0.0`) meaning the calling task's own host. Returns stdlib-shaped rows — `(AF_INET, SOCK_STREAM, IPPROTO_TCP, "", (address, port))` and the `SOCK_DGRAM` / `IPPROTO_UDP` row — filtered by `family`, `type` and `proto`. Ports are numeric (`int`, a digit string, or `None` for 0); resolver `flags` have nothing to vary |
 | `loop.getnameinfo` | Reverse lookup: a synthetic address maps back to its host name, and a host name (what `get_extra_info("peername")` reports) maps to itself. `NI_NUMERICHOST` returns the address instead; services are always numeric |
@@ -54,6 +57,17 @@ address, an `AF_INET6` or `SOCK_RAW` request, a service name — raises
 `socket.gaierror(EAI_NONAME)`. Resolution is a pure lookup, not a scheduling
 decision: it never blocks and records no trace event.
 
+Partitions, crashes and reboots are all silent, so time is the only
+failure detector the code under test has — and `set_clock` lets that
+detector be wrong. A lease holder whose clock runs fast and an issuer
+whose clock runs slow disagree about when the lease expired, which is the
+disagreement leases exist to survive, and a test can produce it on
+purpose. The converse is worth knowing before reaching for it: a protocol
+that puts durations on the wire rather than timestamps is immune to skew
+by construction — in `examples/jobqueue/` only the broker reads a clock,
+so skewing a worker changes nothing the cluster decides. Clock faults
+reach only code that compares timestamps taken on different machines.
+
 Limitations, stated honestly: write-side flow control is not simulated
 (`drain()` never blocks, write buffers are unbounded, the peer cannot pause
 your writes); there is no retransmission or congestion model — streams are
diff --git a/src/simloop/_sim.py b/src/simloop/_sim.py
index 42d717d..197cc17 100644
--- a/src/simloop/_sim.py
+++ b/src/simloop/_sim.py
@@ -45,8 +45,9 @@ def uuid4(self) -> uuid.UUID:
         return uuid.UUID(int=loop._uuid_random.getrandbits(128), version=4)
 
     def time(self) -> float:
-        """Seconds: virtual loop time inside a run (starting at 0.0, not the
-        epoch), wall-clock ``time.time()`` outside."""
+        """Seconds: virtual loop time inside a run, wall-clock ``time.time()``
+        outside. Virtual time is not the epoch — it follows the calling
+        host's clock, which starts at 0.0 unless that host is skewed."""
         loop = _running_sim_loop()
         if loop is None:
             return time.time()

From e16cab5c06fd757d7aa334f2f0d5f5cd8277d06b Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 15:35:42 +0530
Subject: [PATCH 6/7] Say where skew stops and when cancellation lands

The clock row read as if durations were skewed too, and the restart
notes promised finished cancellations where crash only requests them.
Also record that crashes now leave a trace event, which moves the hash
of any workload that crashes a host.
---
 CHANGELOG.md          | 9 ++++++---
 docs/supported-api.md | 4 ++--
 src/simloop/_net.py   | 9 +++++++--
 3 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5624913..96f6dc3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,8 +6,11 @@
   `host.restart()`) revives a machine as a fresh incarnation. It restores
   liveness and nothing else — the old tasks stay cancelled and the
   listeners are gone, so the caller boots the machine again the way it
-  booted it the first time — and traffic due while the host was dead is
-  lost, leaving peers to notice the outage from their own timeouts.
+  booted it the first time. Traffic due while the host was dead is lost,
+  leaving peers to notice the outage from their own timeouts. Crashes are
+  now recorded too: `crash()` writes a trace event and consumes a uid, so
+  a workload that crashes a host hashes differently than it did in 0.1.0
+  (workloads that never crash stay byte-identical to 0.1.0).
 - Every host now has `host.disk`, a mapping that survives its crashes:
   where state a real process would fsync belongs. Writes are atomic at
   assignment; there is no partial-write model.
@@ -16,7 +19,7 @@
   hand to `call_at` with it, while durations (`sleep`, `timeout`,
   `wait_for`, `call_later`) cost the same everywhere — which is what a
   wrong wall clock does to a real machine. Traces stay on the true clock,
-  and runs that configure no offset are byte-identical to before.
+  and runs that configure no offset are byte-identical to 0.1.0.
 - A second flagship demo: `examples/raft/` is a teaching-sized Raft (leader
   election + log replication, plain asyncio on streams) tested only under
   simulation — four safety invariants checked over 50,000 chaos seeds, five
diff --git a/docs/supported-api.md b/docs/supported-api.md
index 4f56423..9deb34c 100644
--- a/docs/supported-api.md
+++ b/docs/supported-api.md
@@ -11,7 +11,7 @@ determinism. Fenced APIs raise `SimulationFenceError` (a subclass of
 | API | Behavior under simulation |
 |---|---|
 | `loop.call_soon` / `call_later` / `call_at` | Seeded ready-queue ordering; `(deadline, seq)` timer tie-break |
-| `loop.time()` / `asyncio.sleep` | Virtual clock starting at 0.0, shifted by whatever offset the calling task's host is configured with; never waits on wall time |
+| `loop.time()` / `asyncio.sleep` | Virtual clock starting at 0.0; never waits on wall time. What `loop.time()` *reads* is shifted by whatever offset the calling task's host is configured with; durations such as `asyncio.sleep` cost the same everywhere |
 | `loop.create_task` / `asyncio.create_task` | Real stdlib `Task`s, including custom task factories |
 | `loop.create_future` | Real stdlib `Future`s |
 | `run_until_complete` / `run_forever` / `stop` / `close` | Deadlock detection: raises `SimulationDeadlockError` when nothing can run |
@@ -40,7 +40,7 @@ a host belong to an implicit `driver` host.
 | `loop.net.set_defaults` / `set_link` | Per-direction latency ranges, drop and duplication probabilities, drawn from a seed-derived stream |
 | `loop.net.partition` / `heal` | Silent blackhole: datagrams are lost, stream traffic is held and resumes intact after healing; nothing errors — only your own timeouts fire |
 | `loop.net.crash` | A host's tasks are cancelled and it goes silent; no reset is sent — peers cannot tell a crash from a partition |
-| `loop.net.restart` / `host.restart()` | The counterpart to a crash: the host comes back as a fresh incarnation. Liveness is all that is revived — the old tasks stay cancelled, its listeners and binds are gone, and the caller boots whatever should run on the machine again, the same way it booted it the first time. A packet is checked against liveness when it arrives, so traffic due during the dead window is lost; a packet that was already in flight and lands after the machine is back is delivered, and finds a host that no longer holds the old incarnation's connections. Peers still learn about the outage only from their own timeouts |
+| `loop.net.restart` / `host.restart()` | The counterpart to a crash: the host comes back as a fresh incarnation. Liveness is all that is revived — the old tasks stay cancelled, its listeners and binds are gone, and the caller boots whatever should run on the machine again, the same way it booted it the first time. Cancellation is requested at crash and lands on the next scheduler step, so a restart in the same step can briefly coexist with a dying task that swallows `CancelledError`. A packet is checked against liveness when it arrives, so traffic due during the dead window is lost; a packet that was already in flight and lands after the machine is back is delivered, and finds a host that no longer holds the old incarnation's connections. Peers still learn about the outage only from their own timeouts |
 | `host.disk` | Storage that survives the crash: a `MutableMapping` per host, where state a real process would fsync belongs. Writes are atomic at assignment; there is no partial-write model. Values are stored as given, so mutating a stored object afterwards is the caller's own aliasing, exactly as with a cache in front of a real disk |
 | `loop.net.set_clock` / `clock_offset` | Per-host clock skew, in seconds. The offset changes what that host's tasks *read*: `loop.time()` (and `sim.time()` with it) returns true time plus the offset, and a deadline handed to `call_at` is interpreted on the calling task's clock. Durations are immune — `asyncio.sleep`, `asyncio.timeout`, `wait_for` and `call_later` cost the same everywhere, which is exactly what a wrong wall clock does to a real machine. By default the driver and unconfigured hosts read true time; the driver can be given an offset too. Trace timestamps stay on the true clock, so skew never perturbs scheduling and traces from skewed runs stay comparable |
 | `transport.abort()` | Peer gets `connection_lost(ConnectionResetError)` |
diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index 00b1f5d..9dd4299 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -444,8 +444,8 @@ def set_clock(self, name: str, *, offset: float) -> None:
         long a duration takes: ``asyncio.sleep(1.0)`` still costs one true
         second everywhere, which is what a wrong wall clock does on a real
         machine. Deadlines passed to ``call_at`` are interpreted on the
-        calling task's clock. The driver and unconfigured hosts read true
-        time.
+        calling task's clock. By default the driver and unconfigured hosts
+        read true time.
         """
         self._require_host(name)
         self._clock_offsets[name] = float(offset)
@@ -837,6 +837,11 @@ def restart(self, name: str) -> None:
         State meant to survive the reboot belongs on ``Host.disk``. The
         caller boots whatever should run on the revived machine, the same
         way it booted the machine the first time.
+
+        "Already cancelled" means requested, not finished: ``crash`` asks
+        each task to cancel and the cancellation lands on the next
+        scheduler step, so a restart in the same step can briefly coexist
+        with a dying task that swallows ``CancelledError``.
         """
         self._require_host(name)
         if self._alive[name]:

From 66da47d38209728c18f52aebd8294f8134721737 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 15:35:42 +0530
Subject: [PATCH 7/7] Pin the skew teardown and the call_at rule

Cancel in name order so the teardown cannot depend on set iteration,
and check the skew immunity hash on a second seed. The call_at test
kept an assertion its own comment claimed was independent.
---
 examples/jobqueue/tests/test_lease_skew.py | 19 +++++++++++++------
 tests/test_clock_skew.py                   |  5 +++--
 2 files changed, 16 insertions(+), 8 deletions(-)

diff --git a/examples/jobqueue/tests/test_lease_skew.py b/examples/jobqueue/tests/test_lease_skew.py
index dbc7756..6c5b12b 100644
--- a/examples/jobqueue/tests/test_lease_skew.py
+++ b/examples/jobqueue/tests/test_lease_skew.py
@@ -86,9 +86,14 @@ async def _skew_workload() -> str:
     helpers.verify(cluster)
     trace = loop.trace_hash()  # taken at quiesce: the run, not the teardown
     # Then shut everything down — servers, workers, open connection handlers —
-    # so the loop closes with nothing suspended mid-write.
+    # so the loop closes with nothing suspended mid-write. `all_tasks()` is a
+    # set, so sort it by name: the teardown cannot smuggle in id ordering.
+    # (The hash above is taken before any of this runs, deliberately.)
     here = asyncio.current_task()
-    running = [task for task in asyncio.all_tasks() if task is not here]
+    running = sorted(
+        (task for task in asyncio.all_tasks() if task is not here),
+        key=lambda task: task.get_name(),
+    )
     for task in running:
         task.cancel()
     await asyncio.gather(*running, return_exceptions=True)
@@ -96,9 +101,9 @@ async def _skew_workload() -> str:
     return trace
 
 
-def _skew_hash(offset: float) -> str:
-    loop = SimLoop(seed=3)
-    # Both runs register the hosts in the same order; only the offset differs.
+def _skew_hash(seed: int, offset: float) -> str:
+    loop = SimLoop(seed=seed)
+    # Every run registers the hosts in the same order; only the offset differs.
     for name in ("broker", "w1", "w2", "c1"):
         loop.net.host(name)
     loop.net.set_clock("w2", offset=offset)
@@ -116,4 +121,6 @@ def test_skew_alone_does_not_change_what_happens() -> None:
     # readings, all of them its own. A worker's wrong clock is therefore
     # unobservable — the packet trace is byte-identical either way, which is
     # also why no ablation of *this* app can turn skew into a violation.
-    assert _skew_hash(0.0) == _skew_hash(SKEW_S) == _skew_hash(-SKEW_S)
+    for seed in (0, 7):
+        hashes = {_skew_hash(seed, offset) for offset in (0.0, SKEW_S, -SKEW_S)}
+        assert len(hashes) == 1, f"seed {seed}: the skew moved the trace"
diff --git a/tests/test_clock_skew.py b/tests/test_clock_skew.py
index f553610..d99d7ed 100644
--- a/tests/test_clock_skew.py
+++ b/tests/test_clock_skew.py
@@ -97,13 +97,14 @@ async def schedule() -> None:
 
     loop.run_until_complete(loop.net.host("worker").create_task(schedule()))
     loop.close()
+    # The run lasts two true seconds, so firing at all is what pins the
+    # deadline to one true second away: read as eleven true seconds, the
+    # callback would never have run.
     assert len(fired_at) == 1
     # The callback carries the scheduling task's context (asyncio copies the
     # current context into the handle), so it reads the worker's clock as
     # well: it sees exactly the deadline it was given.
     assert fired_at[0] == pytest.approx(11.0)
-    # And that deadline is one true second after scheduling, not eleven.
-    assert fired_at[0] - loop.net.clock_offset("worker") == pytest.approx(1.0)
 
 
 def test_call_later_ignores_the_offset() -> None: