diff --git a/README.md b/README.md
index bf4e1be..2703c2a 100644
--- a/README.md
+++ b/README.md
@@ -93,6 +93,16 @@ net.crash("node2")                              # no reset, just silence
 - **Replayable traces** — every scheduling and fault decision lands in
   an append-only trace whose hash proves a replay is exact.
 
+## Proving it: the jobqueue demo
+
+`examples/jobqueue/` is a complete distributed system — an exactly-once
+job scheduler (leases, fencing tokens, idempotency keys, backoff, and
+dead-lettering) written in plain asyncio — tested end to end with
+simloop. Its suite runs hundreds of seeds of partitions, crashes, and
+poison jobs, and shows that removing any load-bearing safeguard produces
+a violation the explorer finds and replays from a seed. The bug table
+lives in [examples/jobqueue/README.md](examples/jobqueue/README.md).
+
 ## Honest limits
 
 Code that goes through the event-loop API is supported; code that
diff --git a/examples/jobqueue/README.md b/examples/jobqueue/README.md
new file mode 100644
index 0000000..50b2492
--- /dev/null
+++ b/examples/jobqueue/README.md
@@ -0,0 +1,62 @@
+# jobqueue — an exactly-once job scheduler, proven by simloop
+
+A demo distributed system written in plain asyncio (stdlib only, no simloop
+imports): one broker, stateless workers, submitting clients. Its test suite
+runs entirely under [simloop](../../README.md) — seeded scheduling, virtual
+time, simulated partitions and crashes — and every failure it can produce
+replays exactly from a seed.
+
+## The claim, stated honestly
+
+Attempts are at-least-once; **effects commit exactly once**. Workers may
+re-run a job after a crash or an expired lease, but the fenced, idempotent
+effect store accepts one commit per job. No acknowledged job is lost: it
+ends `done` (effect committed) or `dead` (dead-lettered after
+`max_attempts`, visible, never silent).
+
+Mechanisms: time-based leases with heartbeat renewal, per-job monotonic
+fencing tokens checked at both broker and store, idempotency-key submit
+dedupe, exponential-backoff requeue, dead-letter state. Time is the only
+failure detector — under simloop a crashed peer sends no reset and a
+partition stalls silently, exactly like production.
+
+## Invariants
+
+Checked after every simulated run (`tests/invariants.py`):
+
+1. **no-loss** — every acknowledged submit ends done or dead
+2. **exactly-once** — at most one accepted commit per job and per logical
+   submit; every done job has exactly one
+3. **no-zombie-writes** — no commit accepted from a superseded lease
+4. **convergence** — nothing left queued or leased at quiesce
+
+## The numbers
+
+- Scenario suite: 7 scenarios × 25–50 seeds each, all green.
+- Campaign: **300 seeds** of randomized partitions, a worker crash, and
+  poison jobs per seed — invariants held on every seed.
+- Ablations: remove any load-bearing safeguard and the explorer finds a
+  violating schedule. Found-during-development bugs are listed too.
+
+| # | Safeguard removed / bug | Invariant violated | Found at seed | Seeds searched | Reproduce |
+|---|---|---|---|---|---|
+| 1 | Store fencing off (`EffectStore(fenced=False)`) | no-zombie-writes | 0 | 1 | `uv run pytest examples/jobqueue/tests/test_mutations.py::test_unfenced_store_admits_a_zombie_write` |
+| 2 | Store idempotency off (`EffectStore(idempotent=False)`) | exactly-once | 0 | 1 | `... ::test_unidempotent_store_double_commits_after_a_crash` |
+| 3 | Broker fencing off + store ablated | exactly-once | 0 | 1 | `... ::test_broker_fencing_off_lets_a_zombie_finish_the_job` |
+| 4 | Client idempotency keys off | exactly-once | 0 | 1 | `... ::test_no_idempotency_key_duplicates_a_lost_ack_submit` |
+| 5 | Unbounded retries (`max_attempts=None`) | convergence | 0 | 1 | `... ::test_unbounded_attempts_never_converge_on_poison` |
+| 6 | Worker renewals off + store idempotency off | exactly-once | 0 | 1 | `... ::test_renew_off_paired_with_unidempotent_store_double_commits` |
+
+Rows 1–6 are labeled ablations — detection demonstrations, not bugs that
+were ever shipped. Renewals off *alone* stays safe across 75 seeds
+(defense in depth); so does broker fencing alone — the store is the
+last line, and the suite proves both lines independently.
+
+## Run it
+
+    uv run pytest examples/jobqueue/tests -q            # fast suite
+    uv run pytest examples/jobqueue/tests -q -m slow    # campaign + safety proofs
+
+Replay any sim-test failure exactly:
+
+    uv run pytest 'examples/jobqueue/tests/test_campaign.py::test_campaign_holds_invariants_under_chaos' -m slow --simloop-replay=0
diff --git a/examples/jobqueue/conftest.py b/examples/jobqueue/conftest.py
new file mode 100644
index 0000000..ecf1d61
--- /dev/null
+++ b/examples/jobqueue/conftest.py
@@ -0,0 +1,8 @@
+"""Make ``import jobqueue`` work when pytest runs from the repository root."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
diff --git a/examples/jobqueue/jobqueue/__init__.py b/examples/jobqueue/jobqueue/__init__.py
new file mode 100644
index 0000000..68c2df9
--- /dev/null
+++ b/examples/jobqueue/jobqueue/__init__.py
@@ -0,0 +1,9 @@
+"""An exactly-once job scheduler in plain asyncio, tested with simloop.
+
+Plain-stdlib demo application: a single broker leases jobs to stateless
+workers under time-based leases with fencing tokens; clients submit with
+idempotency keys. Nothing in this package imports simloop — the harness
+lives entirely in the test suite.
+"""
+
+from __future__ import annotations
diff --git a/examples/jobqueue/jobqueue/broker.py b/examples/jobqueue/jobqueue/broker.py
new file mode 100644
index 0000000..0dd79a4
--- /dev/null
+++ b/examples/jobqueue/jobqueue/broker.py
@@ -0,0 +1,182 @@
+"""The broker: single authority for the job state machine.
+
+States per job: queued -> leased -> done, with leased -> queued on lease
+expiry and queued -> dead once the attempt budget is spent. Time is the
+only failure detector — a worker that goes silent is indistinguishable
+from a partitioned one, so the lease deadline decides, never the
+connection.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass, field
+from typing import Any
+
+from jobqueue import wire
+
+
+@dataclass
+class _Job:
+    job_id: str
+    key: str
+    payload: dict[str, Any]
+    state: str = "queued"
+    attempts: int = 0
+    token: int = 0
+    ready_at: float = 0.0
+    expiry: asyncio.TimerHandle | None = field(default=None, repr=False)
+
+
+class Broker:
+    def __init__(
+        self,
+        *,
+        lease_s: float = 2.0,
+        max_attempts: int | None = 3,
+        backoff_base_s: float = 0.5,
+        fencing: bool = True,
+    ) -> None:
+        self._lease_s = lease_s
+        self._max_attempts = max_attempts
+        self._backoff_base_s = backoff_base_s
+        self._fencing = fencing
+        self._jobs: dict[str, _Job] = {}
+        self._by_key: dict[str, str] = {}
+        self._order: list[str] = []
+        self._ready = asyncio.Event()
+        self._next_id = 0
+
+    async def serve(self, port: int = 7000) -> None:
+        server = await asyncio.start_server(self._connection, "0.0.0.0", port)
+        async with server:
+            await server.serve_forever()
+
+    async def _connection(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        try:
+            while True:
+                message = await wire.read_message(reader)
+                response = await self.handle(message)
+                wire.write_message(writer, response)
+                await writer.drain()
+        except (asyncio.IncompleteReadError, ConnectionResetError, wire.FrameError):
+            pass
+        finally:
+            writer.close()
+
+    async def handle(self, message: dict[str, Any]) -> dict[str, Any]:
+        op = message["op"]
+        if op == "submit":
+            return self._submit(message["key"], message["payload"])
+        if op == "acquire":
+            return await self._acquire(message["worker_id"], message["wait_s"])
+        if op == "renew":
+            return self._renew(message["job_id"], message["token"])
+        if op == "complete":
+            return self._complete(message["job_id"], message["token"])
+        if op == "status":
+            return self._status(message["job_id"])
+        return {"error": f"unknown op {op!r}"}
+
+    def snapshot(self) -> dict[str, tuple[str, int]]:
+        return {j.job_id: (j.state, j.attempts) for j in self._jobs.values()}
+
+    def _submit(self, key: str, payload: dict[str, Any]) -> dict[str, Any]:
+        job_id = self._by_key.get(key)
+        if job_id is None:
+            self._next_id += 1
+            job_id = f"j{self._next_id}"
+            self._jobs[job_id] = _Job(job_id=job_id, key=key, payload=payload)
+            self._by_key[key] = job_id
+            self._order.append(job_id)
+            self._ready.set()
+        return {"job_id": job_id}
+
+    async def _acquire(self, worker_id: str, wait_s: float) -> dict[str, Any]:
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + wait_s
+        while True:
+            job = self._next_ready(loop.time())
+            if job is not None:
+                return self._lease(job)
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                return {"job_id": None}
+            self._ready.clear()
+            try:
+                async with asyncio.timeout(remaining):
+                    await self._ready.wait()
+            except TimeoutError:
+                return {"job_id": None}
+
+    def _next_ready(self, now: float) -> _Job | None:
+        for job_id in self._order:
+            job = self._jobs[job_id]
+            if job.state == "queued" and job.ready_at <= now:
+                return job
+        return None
+
+    def _lease(self, job: _Job) -> dict[str, Any]:
+        loop = asyncio.get_running_loop()
+        job.state = "leased"
+        job.token += 1
+        job.attempts += 1
+        self._order.remove(job.job_id)
+        job.expiry = loop.call_later(
+            self._lease_s, self._expire, job.job_id, job.token
+        )
+        return {
+            "job_id": job.job_id,
+            "token": job.token,
+            "payload": job.payload,
+            "lease_s": self._lease_s,
+        }
+
+    def _expire(self, job_id: str, token: int) -> None:
+        job = self._jobs.get(job_id)
+        if job is None or job.state != "leased" or job.token != token:
+            return
+        if self._max_attempts is not None and job.attempts >= self._max_attempts:
+            job.state = "dead"
+            return
+        loop = asyncio.get_running_loop()
+        delay = self._backoff_base_s * 2 ** (job.attempts - 1)
+        job.state = "queued"
+        job.ready_at = loop.time() + delay
+        self._order.append(job_id)
+        loop.call_later(delay, self._ready.set)
+
+    def _renew(self, job_id: str, token: int) -> dict[str, Any]:
+        job = self._jobs.get(job_id)
+        if job is None or job.state != "leased":
+            return {"ok": False}
+        if self._fencing and token != job.token:
+            return {"ok": False}
+        if job.expiry is not None:
+            job.expiry.cancel()
+        loop = asyncio.get_running_loop()
+        job.expiry = loop.call_later(
+            self._lease_s, self._expire, job.job_id, job.token
+        )
+        return {"ok": True}
+
+    def _complete(self, job_id: str, token: int) -> dict[str, Any]:
+        job = self._jobs.get(job_id)
+        if job is None or job.state == "dead":
+            return {"ok": False}
+        if job.state == "done":
+            return {"ok": True}
+        if self._fencing and (job.state != "leased" or token != job.token):
+            return {"ok": False}
+        if job.expiry is not None:
+            job.expiry.cancel()
+        if job.state == "queued":
+            self._order.remove(job.job_id)
+        job.state = "done"
+        return {"ok": True}
+
+    def _status(self, job_id: str) -> dict[str, Any]:
+        job = self._jobs.get(job_id)
+        return {"state": None if job is None else job.state}
diff --git a/examples/jobqueue/jobqueue/client.py b/examples/jobqueue/jobqueue/client.py
new file mode 100644
index 0000000..017065f
--- /dev/null
+++ b/examples/jobqueue/jobqueue/client.py
@@ -0,0 +1,74 @@
+"""A submitting client: idempotency keys, timeouts, retries with backoff.
+
+A timed-out submit is ambiguous — the broker may or may not have created
+the job. The client retries with the *same* idempotency key, so the broker
+can answer "that one again" instead of minting a duplicate.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from jobqueue import wire
+
+
+class Client:
+    def __init__(
+        self,
+        client_id: str,
+        *,
+        broker_host: str = "broker",
+        broker_port: int = 7000,
+        idempotency: bool = True,
+        rpc_timeout_s: float = 1.0,
+        retries: int = 6,
+        retry_base_s: float = 0.25,
+        poll_s: float = 0.5,
+    ) -> None:
+        self._client_id = client_id
+        self._broker_host = broker_host
+        self._broker_port = broker_port
+        self._idempotency = idempotency
+        self._rpc_timeout_s = rpc_timeout_s
+        self._retries = retries
+        self._retry_base_s = retry_base_s
+        self._poll_s = poll_s
+        self._counter = 0
+        self.acknowledged: dict[str, str] = {}
+
+    async def submit(
+        self, value: str, *, duration: float = 0.1, poison: bool = False
+    ) -> str | None:
+        """Submit one logical job; None means never acknowledged."""
+        self._counter += 1
+        key = f"{self._client_id}.{self._counter}"
+        payload = {"value": value, "duration": duration, "poison": poison}
+        for attempt in range(self._retries):
+            wire_key = key if self._idempotency else f"{key}.try{attempt}"
+            response = await wire.call(
+                self._broker_host,
+                self._broker_port,
+                {"op": "submit", "key": wire_key, "payload": payload},
+                timeout_s=self._rpc_timeout_s,
+            )
+            if response is not None:
+                job_id: str = response["job_id"]
+                self.acknowledged[key] = job_id
+                return job_id
+            await asyncio.sleep(self._retry_base_s * 2**attempt)
+        return None
+
+    async def wait(self, job_id: str) -> str:
+        """Poll until the job reaches a terminal state; returns it."""
+        while True:
+            response = await wire.call(
+                self._broker_host,
+                self._broker_port,
+                {"op": "status", "job_id": job_id},
+                timeout_s=self._rpc_timeout_s,
+            )
+            if response is not None and response["state"] in ("done", "dead"):
+                state: str = response["state"]
+                return state
+            await asyncio.sleep(self._poll_s)
diff --git a/examples/jobqueue/jobqueue/store.py b/examples/jobqueue/jobqueue/store.py
new file mode 100644
index 0000000..53705f3
--- /dev/null
+++ b/examples/jobqueue/jobqueue/store.py
@@ -0,0 +1,58 @@
+"""The demo's stand-in for fenced external storage.
+
+Real systems make exactly-once true at the storage boundary: writes carry a
+fencing token and the store rejects tokens older than the newest it has
+seen, and applying the same logical effect twice is a no-op. This class
+plays that role for the whole cluster. Both checks can be switched off so
+the test suite can demonstrate what each one prevents.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Commit:
+    """One accepted effect. ``stale`` marks a write from a superseded lease."""
+
+    job_id: str
+    token: int
+    value: str
+    stale: bool
+
+
+class EffectStore:
+    def __init__(self, *, fenced: bool = True, idempotent: bool = True) -> None:
+        self.fenced = fenced
+        self.idempotent = idempotent
+        self.begun: list[tuple[str, int]] = []
+        self.commits: list[Commit] = []
+        self.rejected: list[tuple[str, int, str]] = []
+        self._newest: dict[str, int] = {}
+        self._committed: set[str] = set()
+
+    def begin(self, job_id: str, token: int) -> bool:
+        """Announce a token before working; False means the lease is stale."""
+        newest = self._newest.get(job_id, 0)
+        if self.fenced and token < newest:
+            self.rejected.append((job_id, token, "stale-begin"))
+            return False
+        self._newest[job_id] = max(newest, token)
+        self.begun.append((job_id, token))
+        return True
+
+    def commit(self, job_id: str, token: int, value: str) -> str:
+        newest = self._newest.get(job_id, 0)
+        if self.fenced and token < newest:
+            self.rejected.append((job_id, token, "stale"))
+            return "stale"
+        if self.idempotent and job_id in self._committed:
+            self.rejected.append((job_id, token, "duplicate"))
+            return "duplicate"
+        self._newest[job_id] = max(newest, token)
+        self._committed.add(job_id)
+        self.commits.append(
+            Commit(job_id=job_id, token=token, value=value, stale=token < newest)
+        )
+        return "ok"
diff --git a/examples/jobqueue/jobqueue/wire.py b/examples/jobqueue/jobqueue/wire.py
new file mode 100644
index 0000000..6332a08
--- /dev/null
+++ b/examples/jobqueue/jobqueue/wire.py
@@ -0,0 +1,64 @@
+"""Length-prefixed JSON framing and a one-shot request/response helper.
+
+Every RPC in the demo opens a fresh connection, sends one JSON object,
+reads one JSON object back, and closes. One connection per call keeps
+request/response correlation trivial when a timeout abandons a call whose
+reply is still somewhere in the network.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import struct
+from typing import Any
+
+_HEADER = struct.Struct(">I")
+MAX_FRAME = 1 << 20
+
+
+class FrameError(Exception):
+    """The peer sent something that is not a framed JSON object."""
+
+
+async def read_message(reader: asyncio.StreamReader) -> dict[str, Any]:
+    header = await reader.readexactly(_HEADER.size)
+    (length,) = _HEADER.unpack(header)
+    if length > MAX_FRAME:
+        raise FrameError(f"frame of {length} bytes exceeds MAX_FRAME")
+    body = await reader.readexactly(length)
+    try:
+        message = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise FrameError(f"undecodable frame: {exc}") from exc
+    if not isinstance(message, dict):
+        raise FrameError(f"expected a JSON object, got {type(message).__name__}")
+    return message
+
+
+def write_message(writer: asyncio.StreamWriter, message: dict[str, Any]) -> None:
+    body = json.dumps(message, separators=(",", ":")).encode("utf-8")
+    writer.write(_HEADER.pack(len(body)) + body)
+
+
+async def call(
+    host: str, port: int, message: dict[str, Any], *, timeout_s: float
+) -> dict[str, Any] | None:
+    """One request/response on a fresh connection; ``None`` on any failure.
+
+    Timeouts, refused connections, and torn streams all collapse to ``None``
+    because the caller's only recourse is the same either way: back off and
+    retry. Under simulation a partition stalls silently and a crashed peer
+    sends no reset, so the timeout is the only failure detector there is.
+    """
+    try:
+        async with asyncio.timeout(timeout_s):
+            reader, writer = await asyncio.open_connection(host, port)
+            try:
+                write_message(writer, message)
+                await writer.drain()
+                return await read_message(reader)
+            finally:
+                writer.close()
+    except (TimeoutError, OSError, asyncio.IncompleteReadError, FrameError):
+        return None
diff --git a/examples/jobqueue/jobqueue/worker.py b/examples/jobqueue/jobqueue/worker.py
new file mode 100644
index 0000000..307e6e2
--- /dev/null
+++ b/examples/jobqueue/jobqueue/worker.py
@@ -0,0 +1,101 @@
+"""A stateless worker: lease, execute, heartbeat, commit, complete.
+
+The worker holds no durable state, so crash recovery is simply "start
+another worker". Correctness rests on the fencing token it carries: the
+store refuses stale tokens, and a stale commit tells the worker its lease
+was superseded, at which point it walks away without completing.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from jobqueue import wire
+from jobqueue.store import EffectStore
+
+
+class Worker:
+    def __init__(
+        self,
+        worker_id: str,
+        store: EffectStore,
+        *,
+        broker_host: str = "broker",
+        broker_port: int = 7000,
+        renew: bool = True,
+        wait_s: float = 1.0,
+        rpc_timeout_s: float = 1.0,
+        renew_every_s: float = 0.5,
+        retries: int = 4,
+        retry_base_s: float = 0.25,
+    ) -> None:
+        self._worker_id = worker_id
+        self._store = store
+        self._broker_host = broker_host
+        self._broker_port = broker_port
+        self._renew = renew
+        self._wait_s = wait_s
+        self._rpc_timeout_s = rpc_timeout_s
+        self._renew_every_s = renew_every_s
+        self._retries = retries
+        self._retry_base_s = retry_base_s
+
+    async def run(self) -> None:
+        while True:
+            lease = await self._rpc(
+                {
+                    "op": "acquire",
+                    "worker_id": self._worker_id,
+                    "wait_s": self._wait_s,
+                }
+            )
+            if lease is None or lease.get("job_id") is None:
+                continue
+            await self._work(lease)
+
+    async def _work(self, lease: dict[str, Any]) -> None:
+        job_id: str = lease["job_id"]
+        token: int = lease["token"]
+        payload: dict[str, Any] = lease["payload"]
+        if not self._store.begin(job_id, token):
+            return  # a newer lease exists; don't even start
+        renew_task: asyncio.Task[None] | None = None
+        if self._renew:
+            renew_task = asyncio.get_running_loop().create_task(
+                self._renew_forever(job_id, token)
+            )
+        try:
+            try:
+                await self._execute(payload)
+            except Exception:
+                return  # poison: abandon and let the lease expire
+            if self._store.commit(job_id, token, payload["value"]) == "stale":
+                return  # superseded mid-run; the new holder owns completion
+            await self._rpc({"op": "complete", "job_id": job_id, "token": token})
+        finally:
+            if renew_task is not None:
+                renew_task.cancel()
+
+    async def _execute(self, payload: dict[str, Any]) -> None:
+        await asyncio.sleep(payload["duration"])
+        if payload["poison"]:
+            raise RuntimeError(f"poison job {payload['value']!r}")
+
+    async def _renew_forever(self, job_id: str, token: int) -> None:
+        while True:
+            await asyncio.sleep(self._renew_every_s)
+            await self._rpc({"op": "renew", "job_id": job_id, "token": token})
+
+    async def _rpc(self, message: dict[str, Any]) -> dict[str, Any] | None:
+        for attempt in range(self._retries):
+            response = await wire.call(
+                self._broker_host,
+                self._broker_port,
+                message,
+                timeout_s=self._rpc_timeout_s,
+            )
+            if response is not None:
+                return response
+            await asyncio.sleep(self._retry_base_s * 2**attempt)
+        return None
diff --git a/examples/jobqueue/tests/helpers.py b/examples/jobqueue/tests/helpers.py
new file mode 100644
index 0000000..317fea9
--- /dev/null
+++ b/examples/jobqueue/tests/helpers.py
@@ -0,0 +1,186 @@
+"""Cluster assembly and fault choreography shared across the sim suites."""
+
+from __future__ import annotations
+
+import asyncio
+import random
+from dataclasses import dataclass
+from typing import Any
+
+from simloop import SimLoop
+
+from invariants import check_invariants
+from jobqueue.broker import Broker as Broker
+from jobqueue.client import Client
+from jobqueue.store import EffectStore as EffectStore
+from jobqueue.worker import Worker
+
+
+def sim_loop() -> SimLoop:
+    loop = asyncio.get_running_loop()
+    assert isinstance(loop, SimLoop)
+    return loop
+
+
+@dataclass
+class Cluster:
+    broker: Broker
+    store: EffectStore
+    clients: list[Client]
+    worker_hosts: list[str]
+    tasks: list[asyncio.Task[Any]]
+
+
+def add_worker(cluster: Cluster, host_name: str, **kwargs: Any) -> None:
+    loop = sim_loop()
+    worker = Worker(host_name, cluster.store, **kwargs)
+    task = loop.net.host(host_name).create_task(worker.run(), name=host_name)
+    cluster.worker_hosts.append(host_name)
+    cluster.tasks.append(task)
+
+
+async def start_cluster(
+    *,
+    workers: int = 2,
+    clients: int = 1,
+    store: EffectStore | None = None,
+    broker: Broker | None = None,
+    worker_kwargs: dict[str, Any] | None = None,
+    client_kwargs: dict[str, Any] | None = None,
+) -> Cluster:
+    loop = sim_loop()
+    loop.net.set_defaults(latency=(0.01, 0.05))
+    cluster = Cluster(
+        broker=broker if broker is not None else Broker(),
+        store=store if store is not None else EffectStore(),
+        clients=[],
+        worker_hosts=[],
+        tasks=[],
+    )
+    cluster.tasks.append(
+        loop.net.host("broker").create_task(cluster.broker.serve(), name="broker")
+    )
+    for i in range(workers):
+        add_worker(cluster, f"w{i + 1}", **(worker_kwargs or {}))
+    for i in range(clients):
+        name = f"c{i + 1}"
+        loop.net.host(name)
+        cluster.clients.append(Client(name, **(client_kwargs or {})))
+    await asyncio.sleep(0.05)  # let the broker start listening
+    return cluster
+
+
+def acknowledged(cluster: Cluster) -> dict[str, str]:
+    acked: dict[str, str] = {}
+    for client in cluster.clients:
+        acked.update(client.acknowledged)
+    return acked
+
+
+async def settle(cluster: Cluster, *, timeout_s: float = 120.0) -> None:
+    """Wait (in virtual time) until every acknowledged job is done or dead."""
+    async with asyncio.timeout(timeout_s):
+        while True:
+            jobs = cluster.broker.snapshot()
+            acked = acknowledged(cluster).values()
+            if all(jobs[job_id][0] in ("done", "dead") for job_id in acked):
+                return
+            await asyncio.sleep(0.25)
+
+
+def verify(cluster: Cluster) -> None:
+    check_invariants(cluster.store, cluster.broker, acknowledged(cluster))
+
+
+class CrashOnFirstCommit(EffectStore):
+    """Crashes ``host`` the instant its first commit is accepted.
+
+    Models a worker dying between applying its effect and telling the
+    broker: the job re-runs, and only the store's idempotent apply keeps
+    the effect single.
+    """
+
+    def __init__(self, host: str, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._crash_host = host
+        self._fired = False
+
+    def commit(self, job_id: str, token: int, value: str) -> str:
+        result = super().commit(job_id, token, value)
+        if not self._fired and result == "ok":
+            self._fired = True
+            sim_loop().net.host(self._crash_host).crash()
+        return result
+
+
+async def zombie_run(store: EffectStore, *, broker: Broker | None = None) -> Cluster:
+    """One slow job; its first worker is partitioned into a zombie mid-run."""
+    loop = sim_loop()
+    cluster = await start_cluster(workers=1, clients=1, store=store, broker=broker)
+    job_id = await loop.net.host("c1").create_task(
+        cluster.clients[0].submit("z", duration=3.0)
+    )
+    assert job_id is not None
+    # Wait for w1 to actually start executing (not merely for the broker to
+    # grant the lease — the grant response could still be in flight, and
+    # partitioning then would strand it, leaving no zombie at all).
+    async with asyncio.timeout(5.0):
+        while not store.begun:
+            await asyncio.sleep(0.05)
+    loop.net.partition(["w1"], ["broker"])  # renewals now go nowhere
+    add_worker(cluster, "w2")
+    await asyncio.sleep(4.0)  # the lease lapses; w2 re-runs while w1 is a zombie
+    loop.net.heal()
+    await settle(cluster, timeout_s=60.0)
+    return cluster
+
+
+async def crash_after_commit_run(store: EffectStore) -> Cluster:
+    """The first worker dies right after committing, before completing."""
+    loop = sim_loop()
+    cluster = await start_cluster(workers=1, clients=1, store=store)
+    job_id = await loop.net.host("c1").create_task(
+        cluster.clients[0].submit("c", duration=0.2)
+    )
+    assert job_id is not None
+    async with asyncio.timeout(10.0):
+        while not store.commits:
+            await asyncio.sleep(0.05)
+    add_worker(cluster, "w2")  # replacement capacity; w1 is gone
+    await settle(cluster, timeout_s=60.0)
+    return cluster
+
+
+async def slow_job_run(store: EffectStore) -> Cluster:
+    """No renewals and a job slower than its lease: it must re-run."""
+    loop = sim_loop()
+    broker = Broker(lease_s=1.0, backoff_base_s=0.25)
+    cluster = await start_cluster(
+        workers=1,
+        clients=1,
+        store=store,
+        broker=broker,
+        worker_kwargs={"renew": False},
+    )
+    job_id = await loop.net.host("c1").create_task(
+        cluster.clients[0].submit("s", duration=1.5)
+    )
+    assert job_id is not None
+    await settle(cluster, timeout_s=60.0)
+    return cluster
+
+
+async def chaos(cluster: Cluster, rng: random.Random) -> None:
+    """A seed-derived fault schedule: partition windows, then one crash."""
+    loop = sim_loop()
+    for _ in range(3):
+        await asyncio.sleep(rng.uniform(0.5, 3.0))
+        victim = rng.choice(cluster.worker_hosts)
+        loop.net.partition([victim], ["broker"])
+        await asyncio.sleep(rng.uniform(0.5, 4.0))
+        loop.net.heal()
+    await asyncio.sleep(rng.uniform(0.5, 2.0))
+    victim = rng.choice(cluster.worker_hosts)
+    loop.net.host(victim).crash()
+    cluster.worker_hosts.remove(victim)
+    add_worker(cluster, f"{victim}r")
diff --git a/examples/jobqueue/tests/invariants.py b/examples/jobqueue/tests/invariants.py
new file mode 100644
index 0000000..31e2e92
--- /dev/null
+++ b/examples/jobqueue/tests/invariants.py
@@ -0,0 +1,56 @@
+"""The four safety/liveness claims every intact run must satisfy."""
+
+from __future__ import annotations
+
+from collections import Counter
+
+from jobqueue.broker import Broker
+from jobqueue.store import EffectStore
+
+
+class InvariantViolation(AssertionError):
+    def __init__(self, invariant: str, detail: str) -> None:
+        super().__init__(f"{invariant}: {detail}")
+        self.invariant = invariant
+
+
+def check_invariants(
+    store: EffectStore, broker: Broker, acknowledged: dict[str, str]
+) -> None:
+    jobs = broker.snapshot()
+
+    for key, job_id in acknowledged.items():
+        state = jobs.get(job_id, ("missing", 0))[0]
+        if state not in ("done", "dead"):
+            raise InvariantViolation(
+                "no-loss", f"acknowledged {key} -> {job_id} ended {state!r}"
+            )
+
+    per_job = Counter(commit.job_id for commit in store.commits)
+    for job_id, count in per_job.items():
+        if count > 1:
+            raise InvariantViolation(
+                "exactly-once", f"{job_id} committed {count} times"
+            )
+    per_value = Counter(commit.value for commit in store.commits)
+    for value, count in per_value.items():
+        if count > 1:
+            raise InvariantViolation(
+                "exactly-once", f"value {value!r} committed {count} times"
+            )
+    for job_id, (state, _) in jobs.items():
+        if state == "done" and per_job[job_id] != 1:
+            raise InvariantViolation(
+                "exactly-once", f"{job_id} done with {per_job[job_id]} commits"
+            )
+
+    for commit in store.commits:
+        if commit.stale:
+            raise InvariantViolation(
+                "no-zombie-writes",
+                f"{commit.job_id} accepted token {commit.token} after a newer lease",
+            )
+
+    for job_id, (state, _) in jobs.items():
+        if state in ("queued", "leased"):
+            raise InvariantViolation("convergence", f"{job_id} still {state}")
diff --git a/examples/jobqueue/tests/test_broker.py b/examples/jobqueue/tests/test_broker.py
new file mode 100644
index 0000000..38d92f3
--- /dev/null
+++ b/examples/jobqueue/tests/test_broker.py
@@ -0,0 +1,177 @@
+"""Broker state machine, driven directly through handle() on a bare SimLoop."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from simloop import SimLoop
+
+from jobqueue.broker import Broker
+
+
+def _run(coro: Any) -> Any:
+    loop = SimLoop(seed=0)
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+def test_submit_dedupes_on_idempotency_key() -> None:
+    async def main() -> tuple[dict[str, Any], dict[str, Any], int]:
+        broker = Broker()
+        first = await broker.handle(
+            {"op": "submit", "key": "k1", "payload": {"value": "v"}}
+        )
+        second = await broker.handle(
+            {"op": "submit", "key": "k1", "payload": {"value": "v"}}
+        )
+        return first, second, len(broker.snapshot())
+
+    first, second, jobs = _run(main())
+    assert first["job_id"] == second["job_id"]
+    assert jobs == 1
+
+
+def test_lease_expiry_requeues_after_backoff() -> None:
+    async def main() -> list[Any]:
+        broker = Broker(lease_s=1.0, backoff_base_s=0.5, max_attempts=3)
+        submit = await broker.handle(
+            {"op": "submit", "key": "k", "payload": {"value": "v"}}
+        )
+        job_id = submit["job_id"]
+        lease = await broker.handle(
+            {"op": "acquire", "worker_id": "w", "wait_s": 0.1}
+        )
+        seen = [lease["token"]]
+        await asyncio.sleep(1.2)  # expiry at 1.0; requeued, ready at 1.5
+        seen.append(broker.snapshot()[job_id])
+        early = await broker.handle(
+            {"op": "acquire", "worker_id": "w", "wait_s": 0.1}
+        )
+        seen.append(early["job_id"])  # still backing off: nothing to lease
+        await asyncio.sleep(0.5)
+        again = await broker.handle(
+            {"op": "acquire", "worker_id": "w", "wait_s": 0.1}
+        )
+        seen.append(again["token"])
+        return seen
+
+    token1, state, early_job, token2 = _run(main())
+    assert token1 == 1
+    assert state == ("queued", 1)
+    assert early_job is None
+    assert token2 == 2
+
+
+def test_stale_tokens_are_rejected_and_valid_complete_lands() -> None:
+    async def main() -> list[Any]:
+        broker = Broker(lease_s=1.0, backoff_base_s=0.1)
+        submit = await broker.handle(
+            {"op": "submit", "key": "k", "payload": {"value": "v"}}
+        )
+        job_id = submit["job_id"]
+        await broker.handle({"op": "acquire", "worker_id": "w1", "wait_s": 0.1})
+        await asyncio.sleep(1.3)  # token-1 lease expires; requeued, ready at 1.1
+        lease2 = await broker.handle(
+            {"op": "acquire", "worker_id": "w2", "wait_s": 0.1}
+        )
+        results = [lease2["token"]]
+        results.append(
+            (await broker.handle({"op": "renew", "job_id": job_id, "token": 1}))["ok"]
+        )
+        results.append(
+            (await broker.handle({"op": "complete", "job_id": job_id, "token": 1}))["ok"]
+        )
+        results.append(
+            (await broker.handle({"op": "complete", "job_id": job_id, "token": 2}))["ok"]
+        )
+        results.append(
+            (await broker.handle({"op": "complete", "job_id": job_id, "token": 2}))["ok"]
+        )
+        results.append(broker.snapshot()[job_id][0])
+        return results
+
+    token2, stale_renew, stale_complete, complete, again, state = _run(main())
+    assert token2 == 2
+    assert stale_renew is False
+    assert stale_complete is False
+    assert complete is True
+    assert again is True  # completing a done job is idempotent
+    assert state == "done"
+
+
+def test_renew_extends_the_lease() -> None:
+    async def main() -> tuple[str, str]:
+        broker = Broker(lease_s=1.0, backoff_base_s=0.1)
+        submit = await broker.handle(
+            {"op": "submit", "key": "k", "payload": {"value": "v"}}
+        )
+        job_id = submit["job_id"]
+        await broker.handle({"op": "acquire", "worker_id": "w", "wait_s": 0.1})
+        await asyncio.sleep(0.8)
+        renewed = await broker.handle(
+            {"op": "renew", "job_id": job_id, "token": 1}
+        )
+        assert renewed["ok"] is True
+        await asyncio.sleep(0.4)  # t=1.2: past the original expiry, inside renewal
+        alive = broker.snapshot()[job_id][0]
+        await asyncio.sleep(0.8)  # t=2.0: renewed lease (0.8+1.0) has expired
+        return alive, broker.snapshot()[job_id][0]
+
+    alive, expired = _run(main())
+    assert alive == "leased"
+    assert expired == "queued"
+
+
+def test_attempts_exhaustion_dead_letters_the_job() -> None:
+    async def main() -> tuple[str, int]:
+        broker = Broker(lease_s=0.5, backoff_base_s=0.1, max_attempts=2)
+        submit = await broker.handle(
+            {"op": "submit", "key": "k", "payload": {"value": "v"}}
+        )
+        job_id = submit["job_id"]
+        await broker.handle({"op": "acquire", "worker_id": "w", "wait_s": 0.1})
+        await asyncio.sleep(0.7)  # first lease expires; ready again at 0.6
+        await broker.handle({"op": "acquire", "worker_id": "w", "wait_s": 0.1})
+        await asyncio.sleep(0.7)  # second lease expires; attempts == max
+        return broker.snapshot()[job_id][0], broker.snapshot()[job_id][1]
+
+    state, attempts = _run(main())
+    assert state == "dead"
+    assert attempts == 2
+
+
+def test_long_poll_wakes_on_submit() -> None:
+    async def main() -> tuple[Any, float]:
+        loop = asyncio.get_running_loop()
+        broker = Broker()
+        waiter = loop.create_task(
+            broker.handle({"op": "acquire", "worker_id": "w", "wait_s": 5.0})
+        )
+        await asyncio.sleep(0.5)
+        await broker.handle({"op": "submit", "key": "k", "payload": {"value": "v"}})
+        lease = await waiter
+        return lease, loop.time()
+
+    lease, elapsed = _run(main())
+    assert lease["job_id"] is not None
+    assert elapsed < 1.0  # woke on submit, not at the 5s wait cap
+
+
+def test_status_reports_states() -> None:
+    async def main() -> tuple[str | None, str | None]:
+        broker = Broker()
+        submit = await broker.handle(
+            {"op": "submit", "key": "k", "payload": {"value": "v"}}
+        )
+        known = (await broker.handle({"op": "status", "job_id": submit["job_id"]}))[
+            "state"
+        ]
+        unknown = (await broker.handle({"op": "status", "job_id": "nope"}))["state"]
+        return known, unknown
+
+    known, unknown = _run(main())
+    assert known == "queued"
+    assert unknown is None
diff --git a/examples/jobqueue/tests/test_campaign.py b/examples/jobqueue/tests/test_campaign.py
new file mode 100644
index 0000000..c179e63
--- /dev/null
+++ b/examples/jobqueue/tests/test_campaign.py
@@ -0,0 +1,36 @@
+"""300 seeds of chaos: random faults, every invariant must hold."""
+
+from __future__ import annotations
+
+import pytest
+
+from simloop import sim, sim_test
+
+import helpers
+
+
+@pytest.mark.slow
+@sim_test(seeds=300)
+async def test_campaign_holds_invariants_under_chaos() -> None:
+    loop = helpers.sim_loop()
+    rng = sim.random
+    cluster = await helpers.start_cluster(workers=3, clients=2)
+    submits = []
+    for i, client in enumerate(cluster.clients):
+        host = loop.net.host(f"c{i + 1}")
+        for j in range(4):
+            submits.append(
+                host.create_task(
+                    client.submit(
+                        f"c{i + 1}.{j}",
+                        duration=rng.uniform(0.05, 1.0),
+                        poison=rng.random() < 0.15,
+                    )
+                )
+            )
+    chaos_task = loop.create_task(helpers.chaos(cluster, rng))
+    job_ids = [await task for task in submits]
+    assert all(job_id is not None for job_id in job_ids)
+    await helpers.settle(cluster, timeout_s=600.0)
+    chaos_task.cancel()
+    helpers.verify(cluster)
diff --git a/examples/jobqueue/tests/test_cluster_sim.py b/examples/jobqueue/tests/test_cluster_sim.py
new file mode 100644
index 0000000..9f3b240
--- /dev/null
+++ b/examples/jobqueue/tests/test_cluster_sim.py
@@ -0,0 +1,138 @@
+"""Whole-cluster scenarios under seed exploration."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from simloop import SimLoop, sim, sim_test
+
+import helpers
+
+
+@sim_test(seeds=25)
+async def test_one_job_end_to_end() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=1, clients=1)
+    client = cluster.clients[0]
+    job_id = await loop.net.host("c1").create_task(client.submit("v1"))
+    assert job_id is not None
+    state = await loop.net.host("c1").create_task(client.wait(job_id))
+    assert state == "done"
+    assert [commit.value for commit in cluster.store.commits] == ["v1"]
+    assert cluster.broker.snapshot()[job_id] == ("done", 1)
+
+
+@sim_test(seeds=25)
+async def test_happy_path_many_jobs_commit_once() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=2, clients=2)
+    submits = []
+    for i, client in enumerate(cluster.clients):
+        host = loop.net.host(f"c{i + 1}")
+        for j in range(3):
+            submits.append(host.create_task(client.submit(f"v{i}.{j}")))
+    job_ids = [await task for task in submits]
+    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) == 6
+
+
+@sim_test(seeds=50)
+async def test_worker_crash_mid_job_still_commits_once() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=2, clients=1)
+    job_id = await loop.net.host("c1").create_task(
+        cluster.clients[0].submit("crashy", duration=1.0)
+    )
+    assert job_id is not None
+    await asyncio.sleep(0.2 + sim.random.uniform(0.0, 1.2))
+    loop.net.host("w1").crash()
+    helpers.add_worker(cluster, "w3")
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    assert cluster.broker.snapshot()[job_id][0] == "done"
+
+
+@sim_test(seeds=50)
+async def test_partitioned_zombie_is_fenced() -> None:
+    cluster = await helpers.zombie_run(helpers.EffectStore())
+    helpers.verify(cluster)
+    # the zombie's late commit was refused; the re-run's commit won
+    assert any(reason == "stale" for _, _, reason in cluster.store.rejected)
+    assert len(cluster.store.commits) == 1
+    assert cluster.store.commits[0].token >= 2
+
+
+@sim_test(seeds=50)
+async def test_lost_submit_ack_does_not_duplicate_the_job() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=1, clients=1)
+    submit_task = loop.net.host("c1").create_task(
+        cluster.clients[0].submit("d", duration=0.1)
+    )
+    await asyncio.sleep(0.12)  # the request (or its ack) is now in flight
+    loop.net.partition(["c1"], ["broker"])
+    await asyncio.sleep(1.5)  # the client times out and starts retrying
+    loop.net.heal()
+    job_id = await submit_task
+    assert job_id is not None
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    assert len(cluster.broker.snapshot()) == 1  # dedupe: one job, however many tries
+
+
+@sim_test(seeds=50)
+async def test_crash_between_commit_and_complete_is_absorbed() -> None:
+    store = helpers.CrashOnFirstCommit("w1")
+    cluster = await helpers.crash_after_commit_run(store)
+    helpers.verify(cluster)
+    assert len(cluster.store.commits) == 1
+    assert any(reason == "duplicate" for _, _, reason in cluster.store.rejected)
+
+
+@sim_test(seeds=25)
+async def test_poison_job_dead_letters_without_collateral() -> None:
+    loop = helpers.sim_loop()
+    cluster = await helpers.start_cluster(workers=2, clients=1)
+    client = cluster.clients[0]
+    good = await loop.net.host("c1").create_task(client.submit("good"))
+    bad = await loop.net.host("c1").create_task(client.submit("bad", poison=True))
+    assert good is not None and bad is not None
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    assert cluster.broker.snapshot()[good] == ("done", 1)
+    assert cluster.broker.snapshot()[bad] == ("dead", 3)
+    assert [commit.value for commit in cluster.store.commits] == ["good"]
+
+
+async def _replay_workload() -> None:
+    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"r{i}")) for i in range(3)]
+    assert all(job_id is not None for job_id in job_ids)
+    await helpers.settle(cluster)
+    helpers.verify(cluster)
+    for task in cluster.tasks:
+        task.cancel()
+    await asyncio.gather(*cluster.tasks, return_exceptions=True)
+    await asyncio.sleep(0.1)  # let cancelled renew heartbeats finish unwinding
+
+
+def _replay_hash(seed: int) -> str:
+    loop = SimLoop(seed=seed)
+    try:
+        loop.run_until_complete(_replay_workload())
+    finally:
+        loop.close()
+    return loop.trace_hash()
+
+
+@pytest.mark.slow
+def test_cluster_replay_is_stable() -> None:
+    for seed in (0, 7):
+        hashes = {_replay_hash(seed) for _ in range(5)}
+        assert len(hashes) == 1, f"seed {seed} produced diverging traces"
diff --git a/examples/jobqueue/tests/test_mutations.py b/examples/jobqueue/tests/test_mutations.py
new file mode 100644
index 0000000..5c3f6af
--- /dev/null
+++ b/examples/jobqueue/tests/test_mutations.py
@@ -0,0 +1,136 @@
+"""Ablations: switch one safeguard off and prove the explorer catches it."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable, Coroutine
+from typing import Any
+
+import pytest
+
+from simloop import SeedReport, explore
+
+import helpers
+from invariants import InvariantViolation
+
+BUDGET = 200
+
+
+def _find(scenario: Callable[[], Coroutine[Any, Any, object]]) -> SeedReport:
+    report = explore(scenario, range(BUDGET))
+    assert report is not None, "ablation went undetected across the seed budget"
+    return report
+
+
+def test_unfenced_store_admits_a_zombie_write() -> None:
+    async def scenario() -> None:
+        cluster = await helpers.zombie_run(helpers.EffectStore(fenced=False))
+        helpers.verify(cluster)
+
+    report = _find(scenario)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant == "no-zombie-writes"
+
+
+def test_unidempotent_store_double_commits_after_a_crash() -> None:
+    async def scenario() -> None:
+        store = helpers.CrashOnFirstCommit("w1", idempotent=False)
+        cluster = await helpers.crash_after_commit_run(store)
+        helpers.verify(cluster)
+
+    report = _find(scenario)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant == "exactly-once"
+
+
+def test_broker_fencing_off_lets_a_zombie_finish_the_job() -> None:
+    # Broker fencing is second-line defense: with the store intact nothing
+    # stale ever reaches the broker, so its ablation is shown paired with a
+    # fully ablated store — the zombie both writes and completes.
+    async def scenario() -> None:
+        cluster = await helpers.zombie_run(
+            helpers.EffectStore(fenced=False, idempotent=False),
+            broker=helpers.Broker(fencing=False),
+        )
+        helpers.verify(cluster)
+
+    report = _find(scenario)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant in ("exactly-once", "no-zombie-writes")
+
+
+def test_no_idempotency_key_duplicates_a_lost_ack_submit() -> None:
+    async def scenario() -> None:
+        loop = helpers.sim_loop()
+        cluster = await helpers.start_cluster(
+            workers=1, clients=1, client_kwargs={"idempotency": False}
+        )
+        submit_task = loop.net.host("c1").create_task(
+            cluster.clients[0].submit("d", duration=0.1)
+        )
+        await asyncio.sleep(0.12)
+        loop.net.partition(["c1"], ["broker"])
+        await asyncio.sleep(1.5)
+        loop.net.heal()
+        job_id = await submit_task
+        assert job_id is not None
+        await helpers.settle(cluster)
+        helpers.verify(cluster)
+
+    report = _find(scenario)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant == "exactly-once"
+
+
+def test_unbounded_attempts_never_converge_on_poison() -> None:
+    async def scenario() -> None:
+        loop = helpers.sim_loop()
+        broker = helpers.Broker(max_attempts=None, lease_s=0.5, backoff_base_s=0.1)
+        cluster = await helpers.start_cluster(workers=1, clients=1, broker=broker)
+        job_id = await loop.net.host("c1").create_task(
+            cluster.clients[0].submit("p", duration=0.05, poison=True)
+        )
+        assert job_id is not None
+        await helpers.settle(cluster, timeout_s=30.0)
+
+    report = _find(scenario)
+    assert report.seed == 0  # every seed spins forever; the first one shows it
+    assert isinstance(report.exception, TimeoutError)
+
+
+def test_renew_off_paired_with_unidempotent_store_double_commits() -> None:
+    async def scenario() -> None:
+        store = helpers.EffectStore(idempotent=False)
+        cluster = await helpers.slow_job_run(store)
+        helpers.verify(cluster)
+
+    report = _find(scenario)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant == "exactly-once"
+
+
+@pytest.mark.slow
+def test_renew_off_alone_stays_safe() -> None:
+    # Defense in depth: without renewals jobs re-run more, but the intact
+    # store still commits each effect exactly once across every seed.
+    async def scenario() -> None:
+        cluster = await helpers.slow_job_run(helpers.EffectStore())
+        helpers.verify(cluster)
+
+    assert explore(scenario, range(75)) is None
+
+
+@pytest.mark.slow
+def test_broker_fencing_off_alone_stays_safe() -> None:
+    # Defense in depth: broker-side fencing is second-line. With the effect
+    # store intact, a zombie's stale commit is refused at the store no matter
+    # what the broker does, so turning broker fencing off alone stays safe
+    # across every seed.
+    async def scenario() -> None:
+        cluster = await helpers.zombie_run(
+            helpers.EffectStore(),
+            broker=helpers.Broker(fencing=False),
+        )
+        helpers.verify(cluster)
+
+    assert explore(scenario, range(75)) is None
diff --git a/examples/jobqueue/tests/test_store.py b/examples/jobqueue/tests/test_store.py
new file mode 100644
index 0000000..d5174c2
--- /dev/null
+++ b/examples/jobqueue/tests/test_store.py
@@ -0,0 +1,60 @@
+"""The fenced, idempotent effect sink and its ablatable checks."""
+
+from __future__ import annotations
+
+from jobqueue.store import EffectStore
+
+
+def test_first_commit_is_accepted() -> None:
+    store = EffectStore()
+    assert store.begin("j1", 1) is True
+    assert store.begun == [("j1", 1)]
+    assert store.commit("j1", 1, "v") == "ok"
+    assert len(store.commits) == 1
+    assert store.commits[0].job_id == "j1"
+    assert store.commits[0].stale is False
+
+
+def test_second_commit_for_a_job_is_duplicate() -> None:
+    store = EffectStore()
+    store.begin("j1", 1)
+    store.commit("j1", 1, "v")
+    store.begin("j1", 2)
+    assert store.commit("j1", 2, "v") == "duplicate"
+    assert len(store.commits) == 1
+    assert ("j1", 2, "duplicate") in store.rejected
+
+
+def test_stale_commit_is_fenced_out() -> None:
+    store = EffectStore()
+    store.begin("j1", 1)
+    store.begin("j1", 2)
+    assert store.commit("j1", 1, "v") == "stale"
+    assert store.commits == []
+    assert ("j1", 1, "stale") in store.rejected
+
+
+def test_stale_begin_is_fenced_out() -> None:
+    store = EffectStore()
+    store.begin("j1", 2)
+    assert store.begin("j1", 1) is False
+    assert ("j1", 1, "stale-begin") in store.rejected
+    assert store.begun == [("j1", 2)]
+
+
+def test_unfenced_store_accepts_but_flags_zombie_writes() -> None:
+    store = EffectStore(fenced=False)
+    store.begin("j1", 1)
+    store.begin("j1", 2)
+    assert store.commit("j1", 1, "v") == "ok"
+    assert store.commits[0].stale is True
+
+
+def test_unidempotent_store_double_commits() -> None:
+    store = EffectStore(idempotent=False)
+    store.begin("j1", 1)
+    store.commit("j1", 1, "v")
+    store.begin("j1", 2)
+    assert store.commit("j1", 2, "v") == "ok"
+    assert len(store.commits) == 2
+    assert store.commits[1].stale is False
diff --git a/examples/jobqueue/tests/test_wire.py b/examples/jobqueue/tests/test_wire.py
new file mode 100644
index 0000000..5653454
--- /dev/null
+++ b/examples/jobqueue/tests/test_wire.py
@@ -0,0 +1,108 @@
+"""Framing round-trips, frame limits, and the one-shot call helper."""
+
+from __future__ import annotations
+
+import asyncio
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from simloop import SimLoop, sim_test
+
+from jobqueue import wire
+
+PACKAGE_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _run(coro: Any) -> Any:
+    loop = SimLoop(seed=0)
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+def test_oversized_frame_is_rejected() -> None:
+    async def main() -> None:
+        reader = asyncio.StreamReader()
+        reader.feed_data(wire._HEADER.pack(wire.MAX_FRAME + 1))
+        with pytest.raises(wire.FrameError):
+            await wire.read_message(reader)
+
+    _run(main())
+
+
+def test_non_object_payload_is_rejected() -> None:
+    async def main() -> None:
+        reader = asyncio.StreamReader()
+        body = b"[1, 2]"
+        reader.feed_data(wire._HEADER.pack(len(body)) + body)
+        reader.feed_eof()
+        with pytest.raises(wire.FrameError):
+            await wire.read_message(reader)
+
+    _run(main())
+
+
+@sim_test(seeds=5)
+async def test_call_round_trips_a_message() -> None:
+    loop = asyncio.get_running_loop()
+    assert isinstance(loop, SimLoop)
+    loop.net.host("server")
+    loop.net.host("client")
+
+    async def handle(
+        reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        message = await wire.read_message(reader)
+        wire.write_message(writer, {"echo": message})
+        await writer.drain()
+        writer.close()
+
+    async def serve() -> None:
+        server = await asyncio.start_server(handle, "0.0.0.0", 7000)
+        async with server:
+            await server.serve_forever()
+
+    loop.net.host("server").create_task(serve(), name="server")
+    await asyncio.sleep(0.05)
+    reply = await loop.net.host("client").create_task(
+        wire.call("server", 7000, {"op": "ping"}, timeout_s=1.0)
+    )
+    assert reply == {"echo": {"op": "ping"}}
+
+
+@sim_test(seeds=5)
+async def test_call_returns_none_when_nobody_listens() -> None:
+    loop = asyncio.get_running_loop()
+    assert isinstance(loop, SimLoop)
+    loop.net.host("a")
+    loop.net.host("b")
+    reply = await loop.net.host("a").create_task(
+        wire.call("b", 7000, {"op": "ping"}, timeout_s=0.5)
+    )
+    assert reply is None
+
+
+def test_jobqueue_stays_stdlib_only() -> None:
+    code = (
+        "import sys\n"
+        "import jobqueue.wire\n"
+        "import jobqueue.store\n"
+        "import jobqueue.broker\n"
+        "import jobqueue.client\n"
+        "import jobqueue.worker\n"
+        "bad = sorted(m for m in sys.modules"
+        " if m.split('.')[0] in ('simloop', 'pytest'))\n"
+        "assert not bad, bad\n"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        cwd=str(PACKAGE_ROOT),
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 0, result.stderr
diff --git a/pyproject.toml b/pyproject.toml
index 6ebc287..c029d11 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -30,10 +30,11 @@ requires = ["hatchling>=1.27"]
 build-backend = "hatchling.build"
 
 [tool.pytest.ini_options]
-testpaths = ["tests"]
+testpaths = ["tests", "examples/jobqueue/tests"]
 addopts = "-m 'not slow'"
-markers = ["slow: long-running replay-stability checks"]
+markers = ["slow: long-running seed sweeps and replay-stability checks"]
 
 [tool.mypy]
 strict = true
-files = ["src", "tests", "benchmarks"]
+files = ["src", "tests", "benchmarks", "examples/jobqueue"]
+mypy_path = "examples/jobqueue"