From c09d53b9937107d3a185f5aee25c7670e2442097 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sun, 12 Jul 2026 19:51:33 +0530
Subject: [PATCH 1/9] Add network host registry with task pinning

---
 src/simloop/__init__.py |  3 ++
 src/simloop/_loop.py    | 32 ++++++++++-----
 src/simloop/_net.py     | 90 +++++++++++++++++++++++++++++++++++++++++
 tests/test_net.py       | 81 +++++++++++++++++++++++++++++++++++++
 4 files changed, 195 insertions(+), 11 deletions(-)
 create mode 100644 src/simloop/_net.py
 create mode 100644 tests/test_net.py

diff --git a/src/simloop/__init__.py b/src/simloop/__init__.py
index 6a3fd3c..53e2e45 100644
--- a/src/simloop/__init__.py
+++ b/src/simloop/__init__.py
@@ -1,14 +1,17 @@
 """simloop — deterministic simulation testing for Python asyncio."""
 
 from simloop._loop import SimLoop, SimulationDeadlockError, SimulationFenceError
+from simloop._net import Host, SimNetwork
 from simloop._sim import Sim, sim
 from simloop._trace import TraceEvent
 
 __version__ = "0.0.1.dev0"
 
 __all__ = [
+    "Host",
     "Sim",
     "SimLoop",
+    "SimNetwork",
     "SimulationDeadlockError",
     "SimulationFenceError",
     "TraceEvent",
diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py
index 34dfe9a..73cecd3 100644
--- a/src/simloop/_loop.py
+++ b/src/simloop/_loop.py
@@ -15,6 +15,7 @@
 if TYPE_CHECKING:
     from asyncio.events import _TaskFactory
 
+from simloop._net import SimNetwork
 from simloop._trace import TraceEvent, TraceRecorder
 
 _Ts = TypeVarTuple("_Ts")
@@ -97,6 +98,7 @@ def __init__(self, seed: int = 0) -> None:
         self._unhandled: list[BaseException] = []
         self._exception_handler: _ExceptionHandler | None = None
         self._task_factory: _TaskFactory | None = None
+        self._net = SimNetwork(self)
 
     # ------------------------------------------------------------------
     # Introspection
@@ -113,6 +115,10 @@ def trace(self) -> tuple[TraceEvent, ...]:
     def trace_hash(self) -> str:
         return self._recorder.hash()
 
+    @property
+    def net(self) -> SimNetwork:
+        return self._net
+
     # ------------------------------------------------------------------
     # Clock and scheduling
     # ------------------------------------------------------------------
@@ -285,17 +291,21 @@ def create_task(
     ) -> asyncio.Task[Any]:
         self._check_closed()
         if self._task_factory is None:
-            return asyncio.Task(coro, loop=self, name=name, context=context)
-        factory: Any = self._task_factory
-        task: asyncio.Task[Any] = (
-            factory(self, coro)
-            if context is None
-            else factory(self, coro, context=context)
-        )
-        if name is not None:
-            set_name = getattr(task, "set_name", None)
-            if set_name is not None:
-                set_name(name)
+            task: asyncio.Task[Any] = asyncio.Task(
+                coro, loop=self, name=name, context=context
+            )
+        else:
+            factory: Any = self._task_factory
+            task = (
+                factory(self, coro)
+                if context is None
+                else factory(self, coro, context=context)
+            )
+            if name is not None:
+                set_name = getattr(task, "set_name", None)
+                if set_name is not None:
+                    set_name(name)
+        self._net._register_task(task)
         return task
 
     def set_task_factory(self, factory: _TaskFactory | None) -> None:
diff --git a/src/simloop/_net.py b/src/simloop/_net.py
new file mode 100644
index 0000000..946f691
--- /dev/null
+++ b/src/simloop/_net.py
@@ -0,0 +1,90 @@
+"""In-memory network of named hosts for code running under a SimLoop.
+
+Tasks are pinned to hosts through a context variable: a task started via
+``Host.create_task`` — and every task it spawns — carries that host's name,
+which is how the network attributes traffic to a source machine and how a
+crash knows which tasks to kill. Tasks created outside any host belong to
+an implicit ``driver`` host, so test glue needs no ceremony.
+"""
+
+from __future__ import annotations
+
+import random
+from contextvars import ContextVar
+from typing import TYPE_CHECKING, Any
+
+import asyncio
+
+if TYPE_CHECKING:
+    from simloop._loop import SimLoop
+
+DRIVER = "driver"
+
+_current_host: ContextVar[str] = ContextVar("simloop_current_host", default=DRIVER)
+
+# Host names appear inside trace labels, whose hash serialization relies on
+# "|" and newline never occurring in a label; ">" is the separator inside
+# network labels themselves.
+_FORBIDDEN_NAME_CHARS = ("|", "\n", ">")
+
+
+class Host:
+    """Handle for one simulated machine; tasks started here are pinned to it."""
+
+    def __init__(self, net: SimNetwork, name: str) -> None:
+        self._net = net
+        self._name = name
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def create_task(self, coro: Any, *, name: str | None = None) -> asyncio.Task[Any]:
+        token = _current_host.set(self._name)
+        try:
+            return self._net._loop.create_task(coro, name=name)
+        finally:
+            _current_host.reset(token)
+
+    def crash(self) -> None:
+        self._net.crash(self._name)
+
+
+class SimNetwork:
+    """Registry of hosts and the traffic between them."""
+
+    def __init__(self, loop: SimLoop) -> None:
+        self._loop = loop
+        # Fault decisions draw from their own seed-derived stream so they can
+        # never perturb the scheduler's draws or the sim.* user streams.
+        self._rng = random.Random(f"{loop.seed}:net")
+        self._hosts: dict[str, Host] = {}
+        self._alive: dict[str, bool] = {}
+        self._tasks: dict[str, list[asyncio.Task[Any]]] = {}
+        self.host(DRIVER)
+
+    def host(self, name: str) -> Host:
+        if not name:
+            raise ValueError("host name must be a non-empty string")
+        if any(ch in name for ch in _FORBIDDEN_NAME_CHARS):
+            raise ValueError(f"host name {name!r} may not contain '|', '>' or newline")
+        if name in self._hosts:
+            raise ValueError(f"host {name!r} already exists")
+        host = Host(self, name)
+        self._hosts[name] = host
+        self._alive[name] = True
+        self._tasks[name] = []
+        return host
+
+    def crash(self, name: str) -> None:
+        raise NotImplementedError
+
+    def _register_task(self, task: asyncio.Task[Any]) -> None:
+        owner = self._tasks[_current_host.get()]
+        owner.append(task)
+        task.add_done_callback(owner.remove)
+
+    def _require_host(self, name: str) -> str:
+        if name not in self._hosts:
+            raise OSError(f"unknown host {name!r}")
+        return name
diff --git a/tests/test_net.py b/tests/test_net.py
new file mode 100644
index 0000000..a3183ee
--- /dev/null
+++ b/tests/test_net.py
@@ -0,0 +1,81 @@
+"""Host registry, task pinning, and the simulated packet network."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from simloop import Host, SimLoop, SimNetwork
+
+
+def test_loop_exposes_a_network() -> None:
+    loop = SimLoop(seed=0)
+    try:
+        assert isinstance(loop.net, SimNetwork)
+    finally:
+        loop.close()
+
+
+def test_host_registration_and_validation() -> None:
+    loop = SimLoop(seed=0)
+    try:
+        node = loop.net.host("node1")
+        assert isinstance(node, Host)
+        assert node.name == "node1"
+        with pytest.raises(ValueError, match="already"):
+            loop.net.host("node1")
+        with pytest.raises(ValueError, match="already"):
+            loop.net.host("driver")  # implicit driver host is pre-registered
+        for bad in ("", "a|b", "a>b", "a\nb"):
+            with pytest.raises(ValueError):
+                loop.net.host(bad)
+    finally:
+        loop.close()
+
+
+def test_tasks_are_pinned_to_their_host() -> None:
+    loop = SimLoop(seed=0)
+    seen: list[str] = []
+
+    async def whoami() -> None:
+        from simloop._net import _current_host
+
+        seen.append(_current_host.get())
+
+    async def parent() -> None:
+        # A child task created inside a pinned task inherits the pin.
+        await asyncio.create_task(whoami())
+
+    async def main() -> None:
+        node = loop.net.host("node1")
+        await node.create_task(parent())
+        await asyncio.create_task(whoami())  # unpinned: belongs to the driver
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert seen == ["node1", "driver"]
+
+
+def test_task_registry_tracks_creation_and_completion() -> None:
+    loop = SimLoop(seed=0)
+
+    async def nap() -> None:
+        await asyncio.sleep(0.01)
+
+    async def main() -> None:
+        node = loop.net.host("node1")
+        task = node.create_task(nap())
+        assert task in loop.net._tasks["node1"]
+        await task
+        # The clock only advances once the ready queue is drained, so after
+        # this sleep the removal callback has certainly run.
+        await asyncio.sleep(0.01)
+        assert loop.net._tasks["node1"] == []
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()

From f3f4b71778e9d19a2ebe6b62d5fc807a29ac49a6 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sun, 12 Jul 2026 20:22:14 +0530
Subject: [PATCH 2/9] Simulate datagram endpoints with seeded link faults

---
 src/simloop/_loop.py       |  23 ++++-
 src/simloop/_net.py        | 202 +++++++++++++++++++++++++++++++++++++
 src/simloop/_trace.py      |  11 +-
 src/simloop/_transports.py |  89 ++++++++++++++++
 tests/test_net.py          | 162 +++++++++++++++++++++++++++++
 5 files changed, 479 insertions(+), 8 deletions(-)
 create mode 100644 src/simloop/_transports.py

diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py
index 73cecd3..e630e87 100644
--- a/src/simloop/_loop.py
+++ b/src/simloop/_loop.py
@@ -46,6 +46,14 @@ def _fence(api: str) -> NoReturn:
     )
 
 
+def _reject_kwargs(api: str, kwargs: dict[str, Any]) -> None:
+    # Optional stdlib arguments (ssl, sock, interface selectors, ...) reach
+    # outside the simulation; anything actually requested must fail loudly.
+    for name, value in kwargs.items():
+        if value:
+            _fence(f"{api}({name}=...)")
+
+
 def _label(callback: Callable[..., object]) -> str:
     # Labels feed the trace hash, so they must be stable across processes:
     # qualified names only, never repr() (which can embed memory addresses).
@@ -308,6 +316,18 @@ def create_task(
         self._net._register_task(task)
         return task
 
+    async def create_datagram_endpoint(
+        self,
+        protocol_factory: Any,
+        local_addr: Any = None,
+        remote_addr: Any = None,
+        **kwargs: Any,
+    ) -> Any:
+        _reject_kwargs("create_datagram_endpoint", kwargs)
+        return await self._net._open_datagram_endpoint(
+            protocol_factory, local_addr, remote_addr
+        )
+
     def set_task_factory(self, factory: _TaskFactory | None) -> None:
         if factory is not None and not callable(factory):
             raise TypeError("task factory must be a callable or None")
@@ -454,9 +474,6 @@ def sendfile(self, *args: Any, **kwargs: Any) -> Any:
     def sock_sendfile(self, *args: Any, **kwargs: Any) -> Any:
         _fence("sock_sendfile")
 
-    def create_datagram_endpoint(self, *args: Any, **kwargs: Any) -> Any:
-        _fence("create_datagram_endpoint")
-
     def connect_read_pipe(self, *args: Any, **kwargs: Any) -> Any:
         _fence("connect_read_pipe")
 
diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index 946f691..b1d4af4 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -11,10 +11,13 @@
 
 import random
 from contextvars import ContextVar
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
 import asyncio
 
+from simloop._transports import _SimDatagramTransport
+
 if TYPE_CHECKING:
     from simloop._loop import SimLoop
 
@@ -28,6 +31,39 @@
 _FORBIDDEN_NAME_CHARS = ("|", "\n", ">")
 
 
+@dataclass(slots=True)
+class _Packet:
+    kind: str  # "dgram", "syn", "accept", "refuse", "data", "fin", "rst"
+    src: str
+    dst: str
+    src_port: int
+    dst_port: int
+    conn: int  # connection id; -1 for datagrams
+    seq: int  # per-direction stream sequence; -1 for datagrams
+    payload: bytes
+    uid: int
+
+
+@dataclass(slots=True)
+class _Link:
+    latency: tuple[float, float] | None = None
+    drop: float | None = None
+    duplicate: float | None = None
+
+
+def _check_probability(name: str, value: float) -> float:
+    if not 0.0 <= value <= 1.0:
+        raise ValueError(f"{name} must be within [0.0, 1.0], got {value!r}")
+    return value
+
+
+def _check_latency(value: tuple[float, float]) -> tuple[float, float]:
+    lo, hi = value
+    if lo < 0.0 or hi < lo:
+        raise ValueError(f"latency must satisfy 0 <= lo <= hi, got {value!r}")
+    return (lo, hi)
+
+
 class Host:
     """Handle for one simulated machine; tasks started here are pinned to it."""
 
@@ -61,6 +97,13 @@ def __init__(self, loop: SimLoop) -> None:
         self._hosts: dict[str, Host] = {}
         self._alive: dict[str, bool] = {}
         self._tasks: dict[str, list[asyncio.Task[Any]]] = {}
+        self._default_latency: tuple[float, float] = (0.0, 0.0)
+        self._default_drop = 0.0
+        self._default_duplicate = 0.0
+        self._links: dict[tuple[str, str], _Link] = {}
+        self._datagrams: dict[tuple[str, int], _SimDatagramTransport] = {}
+        self._next_uid = 0
+        self._next_port = 49152
         self.host(DRIVER)
 
     def host(self, name: str) -> Host:
@@ -76,6 +119,165 @@ def host(self, name: str) -> Host:
         self._tasks[name] = []
         return host
 
+    def set_defaults(
+        self,
+        *,
+        latency: tuple[float, float] | None = None,
+        drop: float | None = None,
+        duplicate: float | None = None,
+    ) -> None:
+        if latency is not None:
+            self._default_latency = _check_latency(latency)
+        if drop is not None:
+            self._default_drop = _check_probability("drop", drop)
+        if duplicate is not None:
+            self._default_duplicate = _check_probability("duplicate", duplicate)
+
+    def set_link(
+        self,
+        src: str,
+        dst: str,
+        *,
+        latency: tuple[float, float] | None = None,
+        drop: float | None = None,
+        duplicate: float | None = None,
+    ) -> None:
+        self._require_host(src)
+        self._require_host(dst)
+        link = self._links.setdefault((src, dst), _Link())
+        if latency is not None:
+            link.latency = _check_latency(latency)
+        if drop is not None:
+            link.drop = _check_probability("drop", drop)
+        if duplicate is not None:
+            link.duplicate = _check_probability("duplicate", duplicate)
+
+    def _resolved(self, src: str, dst: str) -> tuple[tuple[float, float], float, float]:
+        link = self._links.get((src, dst))
+        if link is None:
+            return (self._default_latency, self._default_drop, self._default_duplicate)
+        return (
+            link.latency if link.latency is not None else self._default_latency,
+            link.drop if link.drop is not None else self._default_drop,
+            link.duplicate if link.duplicate is not None else self._default_duplicate,
+        )
+
+    # ------------------------------------------------------------------
+    # Packet pipeline
+    # ------------------------------------------------------------------
+
+    def _new_uid(self) -> int:
+        uid = self._next_uid
+        self._next_uid += 1
+        return uid
+
+    def _ephemeral(self) -> int:
+        port = self._next_port
+        self._next_port += 1
+        return port
+
+    def _trace(self, verb: str, packet: _Packet) -> None:
+        self._loop._recorder.record(
+            "net", self._loop.time(), packet.uid, f"{verb} {packet.src}>{packet.dst}"
+        )
+
+    def _transmit(self, packet: _Packet) -> None:
+        latency, drop, duplicate = self._resolved(packet.src, packet.dst)
+        if packet.kind == "dgram":
+            # Only datagrams are lossy: a reliable stream that loses bytes
+            # would be lying about being a stream.
+            if self._rng.random() < drop:
+                self._trace("drop", packet)
+                return
+            if self._rng.random() < duplicate:
+                self._trace("dup", packet)
+                self._schedule(packet, latency)
+        self._schedule(packet, latency)
+
+    def _schedule(self, packet: _Packet, latency: tuple[float, float]) -> None:
+        self._trace("send", packet)
+        delay = self._rng.uniform(latency[0], latency[1])
+        self._loop.call_later(delay, self._deliver, packet)
+
+    def _deliver(self, packet: _Packet) -> None:
+        if not (self._alive[packet.src] and self._alive[packet.dst]):
+            self._trace("lost", packet)
+            return
+        # Anything the receiving protocol schedules (including tasks spawned
+        # from connection_made or datagram_received) must be pinned to the
+        # receiving host, not to whichever context sent the packet.
+        token = _current_host.set(packet.dst)
+        try:
+            if packet.kind == "dgram":
+                transport = self._datagrams.get((packet.dst, packet.dst_port))
+                if transport is None:
+                    self._trace("lost", packet)
+                    return
+                transport._datagram_arrived(packet.payload, (packet.src, packet.src_port))
+            else:
+                self._dispatch_stream(packet)
+        finally:
+            _current_host.reset(token)
+
+    def _dispatch_stream(self, packet: _Packet) -> None:
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Datagram endpoints
+    # ------------------------------------------------------------------
+
+    def _bind_address(self, host: str | None, port: int) -> tuple[str, int]:
+        owner = _current_host.get()
+        if host in (None, "", "0.0.0.0", "localhost", "127.0.0.1"):
+            # Production-shaped bind addresses mean "this machine": the host
+            # the calling task is pinned to.
+            return (owner, port)
+        if host != owner:
+            raise OSError(f"cannot bind to {host!r} from host {owner!r}")
+        return (owner, port)
+
+    async def _open_datagram_endpoint(
+        self,
+        protocol_factory: Any,
+        local_addr: tuple[str, int] | None,
+        remote_addr: tuple[str, int] | None,
+    ) -> tuple[_SimDatagramTransport, Any]:
+        if local_addr is None:
+            bind = (_current_host.get(), self._ephemeral())
+        else:
+            bind = self._bind_address(local_addr[0], local_addr[1])
+        if bind in self._datagrams:
+            raise OSError(f"address {bind[0]!r}:{bind[1]} already in use")
+        remote: tuple[str, int] | None = None
+        if remote_addr is not None:
+            remote = (self._require_host(remote_addr[0]), remote_addr[1])
+        transport = _SimDatagramTransport(self, bind, remote)
+        self._datagrams[bind] = transport
+        protocol = protocol_factory()
+        transport._begin(protocol)
+        return transport, protocol
+
+    def _send_datagram(
+        self, src: tuple[str, int], dst: tuple[str, int], payload: bytes
+    ) -> None:
+        self._require_host(dst[0])
+        self._transmit(
+            _Packet(
+                kind="dgram",
+                src=src[0],
+                dst=dst[0],
+                src_port=src[1],
+                dst_port=dst[1],
+                conn=-1,
+                seq=-1,
+                payload=payload,
+                uid=self._new_uid(),
+            )
+        )
+
+    def _unbind_datagram(self, addr: tuple[str, int]) -> None:
+        self._datagrams.pop(addr, None)
+
     def crash(self, name: str) -> None:
         raise NotImplementedError
 
diff --git a/src/simloop/_trace.py b/src/simloop/_trace.py
index c62af99..e8ab2c9 100644
--- a/src/simloop/_trace.py
+++ b/src/simloop/_trace.py
@@ -10,7 +10,7 @@
 from dataclasses import dataclass
 from typing import Literal
 
-EventKind = Literal["schedule", "run", "advance", "cancel"]
+EventKind = Literal["schedule", "run", "advance", "cancel", "net"]
 
 
 @dataclass(frozen=True, slots=True)
@@ -35,10 +35,11 @@ def events(self) -> tuple[TraceEvent, ...]:
     def hash(self) -> str:
         digest = hashlib.sha256()
         for event in self._events:
-            # Labels are always qualified callback names, never free-form text,
-            # so they cannot contain the "|" field separator or a newline. That
-            # keeps this delimiter-based serialization injective: two distinct
-            # event streams can never collide onto the same byte sequence.
+            # Labels are qualified callback names or network labels built from
+            # validated host names, so they cannot contain the "|" field
+            # separator or a newline. That keeps this delimiter-based
+            # serialization injective: two distinct event streams can never
+            # collide onto the same byte sequence.
             line = f"{event.kind}|{event.when!r}|{event.seq}|{event.label}\n"
             digest.update(line.encode("utf-8"))
         return digest.hexdigest()
diff --git a/src/simloop/_transports.py b/src/simloop/_transports.py
new file mode 100644
index 0000000..8211136
--- /dev/null
+++ b/src/simloop/_transports.py
@@ -0,0 +1,89 @@
+"""Transport implementations over the simulated packet network.
+
+These are genuine ``asyncio`` transports driving user-supplied protocols, so
+code written against the standard transport/protocol contract — including
+``asyncio.open_connection`` and ``start_server`` — runs unchanged. All actual
+packet movement is delegated to the owning ``SimNetwork``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from simloop._net import SimNetwork
+
+_Addr = tuple[str, int]
+
+
+def _check_bytes(data: object) -> bytes:
+    if not isinstance(data, (bytes, bytearray, memoryview)):
+        raise TypeError(
+            f"data argument must be a bytes-like object, not {type(data).__name__!r}"
+        )
+    return bytes(data)
+
+
+class _SimDatagramTransport(asyncio.DatagramTransport):
+    def __init__(self, net: SimNetwork, local: _Addr, remote: _Addr | None) -> None:
+        super().__init__()
+        self._net = net
+        self._local = local
+        self._remote = remote
+        self._protocol: Any = None
+        self._closing = False
+
+    def _begin(self, protocol: Any) -> None:
+        self._protocol = protocol
+        protocol.connection_made(self)
+
+    def _datagram_arrived(self, data: bytes, addr: _Addr) -> None:
+        if self._closing:
+            return
+        # A connected endpoint only hears its configured peer host and port.
+        if self._remote is not None and addr != self._remote:
+            return
+        self._protocol.datagram_received(data, addr)
+
+    def sendto(self, data: Any, addr: Any = None) -> None:
+        payload = _check_bytes(data)
+        if self._closing:
+            raise RuntimeError("Cannot send on closing transport")
+        target = addr if addr is not None else self._remote
+        if target is None:
+            raise ValueError("no address is set")
+        if self._remote is not None and tuple(target) != self._remote:
+            raise ValueError(f"Invalid address: must be None or {self._remote}")
+        self._net._send_datagram(self._local, (target[0], target[1]), payload)
+
+    def close(self) -> None:
+        if self._closing:
+            return
+        self._closing = True
+        self._net._unbind_datagram(self._local)
+        self._net._loop.call_soon(self._lost)
+
+    def abort(self) -> None:
+        self.close()
+
+    def _lost(self) -> None:
+        protocol, self._protocol = self._protocol, None
+        if protocol is not None:
+            protocol.connection_lost(None)
+
+    def is_closing(self) -> bool:
+        return self._closing
+
+    def get_extra_info(self, name: str, default: Any = None) -> Any:
+        if name == "sockname":
+            return self._local
+        if name == "peername":
+            return self._remote
+        return default
+
+    def set_protocol(self, protocol: Any) -> None:
+        self._protocol = protocol
+
+    def get_protocol(self) -> Any:
+        return self._protocol
diff --git a/tests/test_net.py b/tests/test_net.py
index a3183ee..f3baca1 100644
--- a/tests/test_net.py
+++ b/tests/test_net.py
@@ -79,3 +79,165 @@ async def main() -> None:
         loop.run_until_complete(main())
     finally:
         loop.close()
+
+
+class _Collector(asyncio.DatagramProtocol):
+    def __init__(self) -> None:
+        self.received: list[tuple[bytes, tuple[str, int]]] = []
+
+    def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None:
+        self.received.append((data, addr))
+
+
+async def _bound_endpoint(
+    host: Host, port: int
+) -> tuple[asyncio.DatagramTransport, _Collector]:
+    loop = asyncio.get_running_loop()
+
+    async def bind() -> tuple[asyncio.DatagramTransport, _Collector]:
+        result: tuple[asyncio.DatagramTransport, _Collector] = (
+            await loop.create_datagram_endpoint(
+                _Collector, local_addr=("0.0.0.0", port)
+            )
+        )
+        return result
+
+    task: asyncio.Task[tuple[asyncio.DatagramTransport, _Collector]] = host.create_task(
+        bind()
+    )
+    return await task
+
+
+def _run_datagram_exchange(
+    seed: int, *, drop: float = 0.0, duplicate: float = 0.0, count: int = 1
+) -> tuple[list[tuple[bytes, tuple[str, int]]], SimLoop]:
+    loop = SimLoop(seed=seed)
+    alpha = loop.net.host("alpha")
+    beta = loop.net.host("beta")
+    loop.net.set_link("beta", "alpha", drop=drop, duplicate=duplicate)
+
+    async def main() -> list[tuple[bytes, tuple[str, int]]]:
+        transport_a, collector = await _bound_endpoint(alpha, 7000)
+        transport_b, _ = await _bound_endpoint(beta, 7001)
+
+        async def send_all() -> None:
+            for number in range(count):
+                transport_b.sendto(f"ping{number}".encode(), ("alpha", 7000))
+
+        await beta.create_task(send_all())
+        await asyncio.sleep(1.0)
+        transport_a.close()
+        transport_b.close()
+        await asyncio.sleep(0.01)
+        return collector.received
+
+    try:
+        received = loop.run_until_complete(main())
+    finally:
+        loop.close()
+    return received, loop
+
+
+def test_datagram_reaches_its_destination_with_source_address() -> None:
+    received, _ = _run_datagram_exchange(seed=0)
+    assert received == [(b"ping0", ("beta", 7001))]
+
+
+def test_datagram_drop_probability_one_loses_everything() -> None:
+    received, loop = _run_datagram_exchange(seed=0, drop=1.0, count=5)
+    assert received == []
+    labels = [e.label for e in loop.trace if e.kind == "net"]
+    assert labels.count("drop beta>alpha") == 5
+
+
+def test_datagram_duplicate_probability_one_doubles_everything() -> None:
+    received, loop = _run_datagram_exchange(seed=0, duplicate=1.0, count=3)
+    assert len(received) == 6
+    labels = [e.label for e in loop.trace if e.kind == "net"]
+    assert labels.count("dup beta>alpha") == 3
+
+
+def test_latency_reorders_but_replays_identically() -> None:
+    def run(seed: int) -> tuple[list[tuple[bytes, tuple[str, int]]], str]:
+        loop = SimLoop(seed=seed)
+        alpha = loop.net.host("alpha")
+        beta = loop.net.host("beta")
+        loop.net.set_defaults(latency=(0.001, 0.1))
+
+        async def main() -> list[tuple[bytes, tuple[str, int]]]:
+            transport_a, collector = await _bound_endpoint(alpha, 7000)
+            transport_b, _ = await _bound_endpoint(beta, 7001)
+            for number in range(20):
+                transport_b.sendto(f"m{number:02d}".encode(), ("alpha", 7000))
+            await asyncio.sleep(2.0)
+            transport_a.close()
+            transport_b.close()
+            await asyncio.sleep(0.01)
+            return collector.received
+
+        try:
+            result = loop.run_until_complete(main())
+        finally:
+            loop.close()
+        return result, loop.trace_hash()
+
+    first, first_hash = run(3)
+    again, again_hash = run(3)
+    assert first == again
+    assert first_hash == again_hash
+    in_order = sorted(first, key=lambda item: item[0])
+    assert first != in_order  # at least one pair actually arrived reordered
+
+
+def test_unknown_destination_host_raises() -> None:
+    loop = SimLoop(seed=0)
+    alpha = loop.net.host("alpha")
+
+    async def main() -> None:
+        transport, _ = await _bound_endpoint(alpha, 7000)
+        with pytest.raises(OSError, match="unknown host"):
+            transport.sendto(b"x", ("ghost", 1))
+        transport.close()
+        await asyncio.sleep(0.01)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+
+
+def test_fault_config_is_validated() -> None:
+    loop = SimLoop(seed=0)
+    try:
+        net = loop.net
+        net.host("alpha")
+        net.host("beta")
+        with pytest.raises(ValueError):
+            net.set_defaults(drop=1.5)
+        with pytest.raises(ValueError):
+            net.set_defaults(duplicate=-0.1)
+        with pytest.raises(ValueError):
+            net.set_defaults(latency=(0.2, 0.1))
+        with pytest.raises(ValueError):
+            net.set_defaults(latency=(-0.1, 0.1))
+        with pytest.raises(OSError, match="unknown host"):
+            net.set_link("alpha", "ghost", drop=0.5)
+    finally:
+        loop.close()
+
+
+def test_duplicate_datagram_bind_is_rejected() -> None:
+    loop = SimLoop(seed=0)
+    alpha = loop.net.host("alpha")
+
+    async def main() -> None:
+        transport, _ = await _bound_endpoint(alpha, 7000)
+        with pytest.raises(OSError, match="in use"):
+            await _bound_endpoint(alpha, 7000)
+        transport.close()
+        await asyncio.sleep(0.01)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()

From e10a30bf03950b3d780ad5da96d7010613a19ca1 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Mon, 13 Jul 2026 00:13:15 +0530
Subject: [PATCH 3/9] Simulate stream connections behind the asyncio streams
 API

---
 src/simloop/_loop.py       |  27 +++-
 src/simloop/_net.py        | 232 +++++++++++++++++++++++++++-
 src/simloop/_transports.py | 183 ++++++++++++++++++++++
 tests/test_net.py          |   6 +-
 tests/test_streams.py      | 307 +++++++++++++++++++++++++++++++++++++
 5 files changed, 741 insertions(+), 14 deletions(-)
 create mode 100644 tests/test_streams.py

diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py
index e630e87..8be76cc 100644
--- a/src/simloop/_loop.py
+++ b/src/simloop/_loop.py
@@ -328,6 +328,27 @@ async def create_datagram_endpoint(
             protocol_factory, local_addr, remote_addr
         )
 
+    async def create_connection(
+        self,
+        protocol_factory: Any,
+        host: Any = None,
+        port: Any = None,
+        **kwargs: Any,
+    ) -> Any:
+        _reject_kwargs("create_connection", kwargs)
+        return await self._net._open_connection(protocol_factory, host, port)
+
+    async def create_server(
+        self,
+        protocol_factory: Any,
+        host: Any = None,
+        port: Any = None,
+        **kwargs: Any,
+    ) -> Any:
+        kwargs.pop("backlog", None)  # accepted and irrelevant: no accept queue
+        _reject_kwargs("create_server", kwargs)
+        return await self._net._start_server(protocol_factory, host, port)
+
     def set_task_factory(self, factory: _TaskFactory | None) -> None:
         if factory is not None and not callable(factory):
             raise TypeError("task factory must be a callable or None")
@@ -459,12 +480,6 @@ def getaddrinfo(self, *args: Any, **kwargs: Any) -> Any:
     def getnameinfo(self, *args: Any, **kwargs: Any) -> Any:
         _fence("getnameinfo")
 
-    def create_connection(self, *args: Any, **kwargs: Any) -> Any:
-        _fence("create_connection")
-
-    def create_server(self, *args: Any, **kwargs: Any) -> Any:
-        _fence("create_server")
-
     def start_tls(self, *args: Any, **kwargs: Any) -> Any:
         _fence("start_tls")
 
diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index b1d4af4..e9bcebc 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -16,7 +16,7 @@
 
 import asyncio
 
-from simloop._transports import _SimDatagramTransport
+from simloop._transports import _SimDatagramTransport, _SimStreamTransport
 
 if TYPE_CHECKING:
     from simloop._loop import SimLoop
@@ -51,6 +51,69 @@ class _Link:
     duplicate: float | None = None
 
 
+@dataclass(slots=True)
+class _Listener:
+    factory: Any
+    server: SimServer
+
+
+@dataclass(slots=True)
+class _Connect:
+    # An in-flight outbound connection. The client transport is built when the
+    # accept lands (not when the connector resumes), so a peer that speaks
+    # first cannot outrun the transport's registration.
+    fut: asyncio.Future[tuple[_SimStreamTransport, Any]]
+    factory: Any
+    local: tuple[str, int]
+    remote: tuple[str, int]
+
+
+class _InOrder:
+    """Reassembles one direction of a stream connection into seq order."""
+
+    def __init__(self, net: SimNetwork) -> None:
+        self._net = net
+        self._next = 0
+        self._early: dict[int, _Packet] = {}
+
+    def push(self, packet: _Packet) -> None:
+        self._early[packet.seq] = packet
+        while self._next in self._early:
+            ready = self._early.pop(self._next)
+            self._next += 1
+            self._net._dispatch_ready(ready)
+
+
+class SimServer(asyncio.AbstractServer):
+    """A listening endpoint; serving from creation, like the stdlib default."""
+
+    def __init__(self, net: SimNetwork, host: str, port: int) -> None:
+        self._net = net
+        self._host = host
+        self._port = port
+        self._closed_fut: asyncio.Future[None] = net._loop.create_future()
+
+    def close(self) -> None:
+        if not self._closed_fut.done():
+            self._net._listeners.pop((self._host, self._port), None)
+            self._closed_fut.set_result(None)
+
+    def is_serving(self) -> bool:
+        return not self._closed_fut.done()
+
+    async def wait_closed(self) -> None:
+        await asyncio.shield(self._closed_fut)
+
+    async def start_serving(self) -> None:
+        return None
+
+    async def serve_forever(self) -> None:
+        await asyncio.shield(self._closed_fut)
+
+    def get_loop(self) -> Any:
+        return self._net._loop
+
+
 def _check_probability(name: str, value: float) -> float:
     if not 0.0 <= value <= 1.0:
         raise ValueError(f"{name} must be within [0.0, 1.0], got {value!r}")
@@ -102,17 +165,23 @@ def __init__(self, loop: SimLoop) -> None:
         self._default_duplicate = 0.0
         self._links: dict[tuple[str, str], _Link] = {}
         self._datagrams: dict[tuple[str, int], _SimDatagramTransport] = {}
+        self._listeners: dict[tuple[str, int], _Listener] = {}
+        self._streams: dict[tuple[int, str], _SimStreamTransport] = {}
+        self._inbound: dict[tuple[int, str], _InOrder] = {}
+        self._pending: dict[int, _Connect] = {}
+        self._next_conn = 0
         self._next_uid = 0
         self._next_port = 49152
         self.host(DRIVER)
 
     def host(self, name: str) -> Host:
+        existing = self._hosts.get(name)
+        if existing is not None:
+            return existing
         if not name:
             raise ValueError("host name must be a non-empty string")
         if any(ch in name for ch in _FORBIDDEN_NAME_CHARS):
             raise ValueError(f"host name {name!r} may not contain '|', '>' or newline")
-        if name in self._hosts:
-            raise ValueError(f"host {name!r} already exists")
         host = Host(self, name)
         self._hosts[name] = host
         self._alive[name] = True
@@ -220,7 +289,162 @@ def _deliver(self, packet: _Packet) -> None:
             _current_host.reset(token)
 
     def _dispatch_stream(self, packet: _Packet) -> None:
-        raise NotImplementedError
+        key = (packet.conn, packet.dst)
+        queue = self._inbound.get(key)
+        if queue is None:
+            queue = self._inbound[key] = _InOrder(self)
+        queue.push(packet)
+
+    def _dispatch_ready(self, packet: _Packet) -> None:
+        if packet.kind == "syn":
+            self._handle_syn(packet)
+            return
+        if packet.kind in ("accept", "refuse"):
+            connect = self._pending.pop(packet.conn, None)
+            if connect is None or connect.fut.done():
+                # The connector gave up (cancelled) before the answer landed.
+                return
+            if packet.kind == "accept":
+                # Stand the client transport up now, in the same in-order step
+                # that processes the accept (seq 0). Data the peer sent from
+                # connection_made is seq 1+, so it is dispatched strictly after
+                # this and always finds a registered transport.
+                client = _SimStreamTransport(
+                    self, packet.conn, local=connect.local, remote=connect.remote
+                )
+                self._streams[(packet.conn, connect.local[0])] = client
+                protocol = connect.factory()
+                client._begin(protocol)
+                connect.fut.set_result((client, protocol))
+            else:
+                connect.fut.set_exception(
+                    ConnectionRefusedError(
+                        f"connect to ({packet.src!r}, {packet.dst_port}) refused"
+                    )
+                )
+            return
+        transport = self._streams.get((packet.conn, packet.dst))
+        if transport is None:
+            return  # connection already torn down locally
+        if packet.kind == "data":
+            transport._data_arrived(packet.payload)
+        elif packet.kind == "fin":
+            transport._eof_arrived()
+        elif packet.kind == "rst":
+            transport._reset_arrived()
+
+    def _handle_syn(self, packet: _Packet) -> None:
+        listener = self._listeners.get((packet.dst, packet.dst_port))
+        if listener is None:
+            self._send_stream(
+                kind="refuse",
+                src=packet.dst,
+                dst=packet.src,
+                conn=packet.conn,
+                seq=0,
+                dst_port=packet.dst_port,
+            )
+            return
+        transport = _SimStreamTransport(
+            self,
+            packet.conn,
+            local=(packet.dst, packet.dst_port),
+            remote=(packet.src, packet.src_port),
+        )
+        self._streams[(packet.conn, packet.dst)] = transport
+        # The accept is seq 0 of the server-to-client direction, so any data
+        # the protocol writes from connection_made (seq 1+) can never arrive
+        # ahead of the accept, whatever the latency draws say.
+        self._send_stream(
+            kind="accept", src=packet.dst, dst=packet.src, conn=packet.conn, seq=0
+        )
+        protocol = listener.factory()
+        transport._begin(protocol)
+
+    def _send_stream(
+        self,
+        *,
+        kind: str,
+        src: str,
+        dst: str,
+        conn: int,
+        seq: int,
+        payload: bytes = b"",
+        src_port: int = 0,
+        dst_port: int = 0,
+    ) -> None:
+        self._transmit(
+            _Packet(
+                kind=kind,
+                src=src,
+                dst=dst,
+                src_port=src_port,
+                dst_port=dst_port,
+                conn=conn,
+                seq=seq,
+                payload=payload,
+                uid=self._new_uid(),
+            )
+        )
+
+    def _drop_stream(self, conn: int, host: str) -> None:
+        self._streams.pop((conn, host), None)
+
+    async def _open_connection(
+        self, protocol_factory: Any, host: Any, port: Any
+    ) -> tuple[_SimStreamTransport, Any]:
+        if not isinstance(host, str) or not isinstance(port, int):
+            raise ValueError("host and port are required")
+        self._require_host(host)
+        src = _current_host.get()
+        conn = self._next_conn
+        self._next_conn += 1
+        src_port = self._ephemeral()
+        fut: asyncio.Future[tuple[_SimStreamTransport, Any]] = (
+            self._loop.create_future()
+        )
+        self._pending[conn] = _Connect(
+            fut=fut,
+            factory=protocol_factory,
+            local=(src, src_port),
+            remote=(host, port),
+        )
+        self._send_stream(
+            kind="syn",
+            src=src,
+            dst=host,
+            conn=conn,
+            seq=0,
+            src_port=src_port,
+            dst_port=port,
+        )
+        try:
+            # The accept handler builds the transport, calls connection_made,
+            # and resolves this future with the ready-made pair.
+            return await fut
+        except asyncio.CancelledError:
+            # If the accept resolved this future in the same step the connector
+            # was cancelled, the transport is already built and connection_made
+            # has already run; abort it so a cancelled connect leaves nothing
+            # connected behind it.
+            if fut.done() and not fut.cancelled() and fut.exception() is None:
+                established, _ = fut.result()
+                established.abort()
+            raise
+        finally:
+            self._pending.pop(conn, None)
+
+    async def _start_server(
+        self, protocol_factory: Any, host: Any, port: Any
+    ) -> SimServer:
+        if not isinstance(port, int):
+            raise ValueError("port is required")
+        bind = self._bind_address(host, port)
+        if bind in self._listeners:
+            raise OSError(f"address {bind[0]!r}:{bind[1]} already in use")
+        server = SimServer(self, bind[0], bind[1])
+        self._listeners[bind] = _Listener(factory=protocol_factory, server=server)
+        return server
 
     # ------------------------------------------------------------------
     # Datagram endpoints
diff --git a/src/simloop/_transports.py b/src/simloop/_transports.py
index 8211136..0845114 100644
--- a/src/simloop/_transports.py
+++ b/src/simloop/_transports.py
@@ -87,3 +87,186 @@ def set_protocol(self, protocol: Any) -> None:
 
     def get_protocol(self) -> Any:
         return self._protocol
+
+
+class _SimStreamTransport(asyncio.Transport):
+    """One end of a reliable, ordered byte-stream connection.
+
+    Reliability comes from per-direction sequence numbers dispatched in
+    order by the network, not from retransmission: stream packets are never
+    dropped, only delayed or held. Flow control is not simulated — writes
+    leave immediately, so the reported write-buffer size is always zero and
+    the peer can never pause this side.
+    """
+
+    def __init__(
+        self, net: SimNetwork, conn: int, local: _Addr, remote: _Addr
+    ) -> None:
+        super().__init__()
+        self._net = net
+        self._conn = conn
+        self._local = local
+        self._remote = remote
+        self._protocol: Any = None
+        self._out_seq = 1  # seq 0 was this direction's handshake packet
+        self._closing = False
+        self._closed = False
+        self._eof_sent = False
+        self._read_paused = False
+        self._backlog: list[bytes] = []
+        self._eof_pending = False
+        self._limits = (16 * 1024, 64 * 1024)  # (low, high): recorded, inert
+
+    def _begin(self, protocol: Any) -> None:
+        self._protocol = protocol
+        protocol.connection_made(self)
+
+    # ------------------------------------------------------------------
+    # Outbound
+    # ------------------------------------------------------------------
+
+    def _send(self, kind: str, payload: bytes = b"") -> None:
+        seq = self._out_seq
+        self._out_seq += 1
+        self._net._send_stream(
+            kind=kind,
+            src=self._local[0],
+            dst=self._remote[0],
+            conn=self._conn,
+            seq=seq,
+            payload=payload,
+        )
+
+    def write(self, data: Any) -> None:
+        payload = _check_bytes(data)
+        if self._eof_sent or self._closing or self._closed:
+            raise RuntimeError("Cannot write to closing transport")
+        if payload:
+            self._send("data", payload)
+
+    def writelines(self, list_of_data: Any) -> None:
+        for data in list_of_data:
+            self.write(data)
+
+    def write_eof(self) -> None:
+        if self._eof_sent or self._closed:
+            return
+        self._eof_sent = True
+        self._send("fin")
+
+    def can_write_eof(self) -> bool:
+        return True
+
+    def close(self) -> None:
+        if self._closing or self._closed:
+            return
+        self._closing = True
+        if not self._eof_sent:
+            self._eof_sent = True
+            self._send("fin")
+        self._net._loop.call_soon(self._finish, None)
+
+    def abort(self) -> None:
+        if self._closed:
+            return
+        self._closing = True
+        self._send("rst")
+        self._net._loop.call_soon(self._finish, None)
+
+    def _finish(self, exc: Exception | None) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        self._closing = True
+        self._net._drop_stream(self._conn, self._local[0])
+        protocol, self._protocol = self._protocol, None
+        if protocol is not None:
+            protocol.connection_lost(exc)
+
+    # ------------------------------------------------------------------
+    # Inbound (called by the network, already in seq order)
+    # ------------------------------------------------------------------
+
+    def _data_arrived(self, data: bytes) -> None:
+        if self._closed:
+            return
+        if self._read_paused:
+            self._backlog.append(data)
+            return
+        self._protocol.data_received(data)
+
+    def _eof_arrived(self) -> None:
+        if self._closed:
+            return
+        if self._read_paused:
+            self._eof_pending = True
+            return
+        keep_open = self._protocol.eof_received()
+        if not keep_open:
+            self.close()
+
+    def _reset_arrived(self) -> None:
+        if self._closed:
+            return
+        self._finish(ConnectionResetError("Connection reset by peer"))
+
+    # ------------------------------------------------------------------
+    # Read flow control (honored locally; never propagates to the peer)
+    # ------------------------------------------------------------------
+
+    def pause_reading(self) -> None:
+        if self._closed or self._read_paused:
+            return
+        self._read_paused = True
+
+    def resume_reading(self) -> None:
+        if self._closed or not self._read_paused:
+            return
+        self._read_paused = False
+        while self._backlog and not self._read_paused and not self._closed:
+            self._protocol.data_received(self._backlog.pop(0))
+        if self._eof_pending and not self._read_paused and not self._closed:
+            self._eof_pending = False
+            self._eof_arrived()
+
+    def is_reading(self) -> bool:
+        return not self._read_paused and not self._closed
+
+    # ------------------------------------------------------------------
+    # Introspection
+    # ------------------------------------------------------------------
+
+    def is_closing(self) -> bool:
+        return self._closing or self._closed
+
+    def get_extra_info(self, name: str, default: Any = None) -> Any:
+        if name == "sockname":
+            return self._local
+        if name == "peername":
+            return self._remote
+        return default
+
+    def set_protocol(self, protocol: Any) -> None:
+        self._protocol = protocol
+
+    def get_protocol(self) -> Any:
+        return self._protocol
+
+    def set_write_buffer_limits(
+        self, high: int | None = None, low: int | None = None
+    ) -> None:
+        if high is None:
+            high = 64 * 1024 if low is None else 4 * low
+        if low is None:
+            low = high // 4
+        if not high >= low >= 0:
+            raise ValueError(
+                f"high ({high!r}) must be >= low ({low!r}) must be >= 0"
+            )
+        self._limits = (low, high)
+
+    def get_write_buffer_limits(self) -> tuple[int, int]:
+        return self._limits
+
+    def get_write_buffer_size(self) -> int:
+        return 0
diff --git a/tests/test_net.py b/tests/test_net.py
index f3baca1..b4f2838 100644
--- a/tests/test_net.py
+++ b/tests/test_net.py
@@ -23,10 +23,8 @@ def test_host_registration_and_validation() -> None:
         node = loop.net.host("node1")
         assert isinstance(node, Host)
         assert node.name == "node1"
-        with pytest.raises(ValueError, match="already"):
-            loop.net.host("node1")
-        with pytest.raises(ValueError, match="already"):
-            loop.net.host("driver")  # implicit driver host is pre-registered
+        assert loop.net.host("node1") is node  # repeat lookup returns the same host
+        assert loop.net.host("driver").name == "driver"  # implicit driver host
         for bad in ("", "a|b", "a>b", "a\nb"):
             with pytest.raises(ValueError):
                 loop.net.host(bad)
diff --git a/tests/test_streams.py b/tests/test_streams.py
new file mode 100644
index 0000000..0e40a07
--- /dev/null
+++ b/tests/test_streams.py
@@ -0,0 +1,307 @@
+"""Simulated stream connections: handshake, transfer, teardown."""
+
+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")
+    return loop
+
+
+async def _echo_lines(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+    while line := await reader.readline():
+        writer.write(line.upper())
+        await writer.drain()
+    writer.close()
+    await writer.wait_closed()
+
+
+async def _reap(task: "asyncio.Task[None]") -> None:
+    """Cancel a long-lived server task so nothing is left pending at stop."""
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+
+
+def test_unmodified_streams_echo() -> None:
+    loop = _network()
+
+    async def serve() -> None:
+        server = await asyncio.start_server(_echo_lines, "0.0.0.0", 9000)
+        async with server:
+            await asyncio.sleep(10.0)
+
+    async def request() -> list[bytes]:
+        reader, writer = await asyncio.open_connection("server", 9000)
+        replies = []
+        for word in (b"one\n", b"two\n", b"three\n"):
+            writer.write(word)
+            await writer.drain()
+            replies.append(await reader.readline())
+        writer.close()
+        await writer.wait_closed()
+        return replies
+
+    async def main() -> list[bytes]:
+        serve_task = loop.net.host("server").create_task(serve())
+        await asyncio.sleep(0.01)
+        replies: list[bytes] = await loop.net.host("client").create_task(request())
+        await _reap(serve_task)
+        return replies
+
+    try:
+        replies = loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert replies == [b"ONE\n", b"TWO\n", b"THREE\n"]
+
+
+def test_user_protocol_classes_run_unchanged() -> None:
+    loop = _network()
+    events: list[str] = []
+
+    class Greeter(asyncio.Protocol):
+        def connection_made(self, transport: Any) -> None:
+            events.append(f"server saw {transport.get_extra_info('peername')[0]}")
+            transport.write(b"hello")
+            transport.close()
+
+    class Listener(asyncio.Protocol):
+        def __init__(self) -> None:
+            self.done = asyncio.get_running_loop().create_future()
+
+        def data_received(self, data: bytes) -> None:
+            events.append(f"client got {data.decode()}")
+
+        def connection_lost(self, exc: Exception | None) -> None:
+            events.append(f"client lost {exc!r}")
+            self.done.set_result(None)
+
+    async def main() -> None:
+        running = asyncio.get_running_loop()
+
+        async def serve() -> None:
+            await running.create_server(Greeter, "0.0.0.0", 9000)
+            await asyncio.sleep(10.0)
+
+        async def connect() -> None:
+            _, protocol = await running.create_connection(Listener, "server", 9000)
+            await protocol.done
+
+        serve_task = loop.net.host("server").create_task(serve())
+        await asyncio.sleep(0.01)
+        await loop.net.host("client").create_task(connect())
+        await _reap(serve_task)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert events == ["server saw client", "client got hello", "client lost None"]
+
+
+def test_connect_to_nothing_is_refused_after_a_round_trip() -> None:
+    loop = _network()
+    loop.net.set_defaults(latency=(0.05, 0.05))
+
+    async def main() -> float:
+        with pytest.raises(ConnectionRefusedError):
+            await asyncio.open_connection("server", 9999)
+        return asyncio.get_running_loop().time()
+
+    try:
+        elapsed = loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert elapsed == pytest.approx(0.1)  # syn there + refusal back
+
+
+def test_bytes_arrive_complete_and_in_order_under_latency_chaos() -> None:
+    loop = _network(seed=5)
+    loop.net.set_defaults(latency=(0.001, 0.2))
+
+    async def collect(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        data = await reader.read()
+        chunks.append(data)
+        writer.close()
+
+    chunks: list[bytes] = []
+    payload = b"".join(f"chunk-{i:03d};".encode() for i in range(50))
+
+    async def main() -> None:
+        server = await loop.net.host("server").create_task(
+            asyncio.start_server(collect, "0.0.0.0", 9000)
+        )
+
+        async def send() -> None:
+            _, writer = await asyncio.open_connection("server", 9000)
+            for i in range(50):
+                writer.write(f"chunk-{i:03d};".encode())
+            writer.close()
+            await writer.wait_closed()
+
+        await loop.net.host("client").create_task(send())
+        await asyncio.sleep(2.0)
+        server.close()
+        await server.wait_closed()
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert chunks == [payload]
+
+
+def test_abort_resets_the_peer() -> None:
+    loop = _network()
+    lost: list[BaseException | None] = []
+
+    class Victim(asyncio.Protocol):
+        def connection_lost(self, exc: Exception | None) -> None:
+            lost.append(exc)
+
+    async def main() -> None:
+        running = asyncio.get_running_loop()
+
+        async def serve() -> None:
+            await running.create_server(Victim, "0.0.0.0", 9000)
+            await asyncio.sleep(10.0)
+
+        async def connect_and_abort() -> None:
+            transport, _ = await running.create_connection(
+                asyncio.Protocol, "server", 9000
+            )
+            transport.abort()
+
+        serve_task = loop.net.host("server").create_task(serve())
+        await asyncio.sleep(0.01)
+        await loop.net.host("client").create_task(connect_and_abort())
+        await asyncio.sleep(0.5)
+        await _reap(serve_task)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert len(lost) == 1
+    assert isinstance(lost[0], ConnectionResetError)
+
+
+def test_duplicate_bind_and_foreign_bind_are_rejected() -> None:
+    loop = _network()
+
+    async def main() -> None:
+        running = asyncio.get_running_loop()
+
+        async def serve_twice() -> None:
+            await running.create_server(asyncio.Protocol, "0.0.0.0", 9000)
+            with pytest.raises(OSError, match="in use"):
+                await running.create_server(asyncio.Protocol, "0.0.0.0", 9000)
+            with pytest.raises(OSError, match="cannot bind"):
+                await running.create_server(asyncio.Protocol, "client", 9001)
+
+        await loop.net.host("server").create_task(serve_twice())
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+
+
+def test_ssl_arguments_are_fenced() -> None:
+    from simloop import SimulationFenceError
+
+    loop = _network()
+
+    async def main() -> None:
+        running: Any = asyncio.get_running_loop()
+        with pytest.raises(SimulationFenceError, match="create_connection"):
+            await running.create_connection(
+                asyncio.Protocol, "server", 9000, ssl=object()
+            )
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+
+
+def test_server_close_stops_accepting() -> None:
+    loop = _network()
+
+    async def main() -> None:
+        running = asyncio.get_running_loop()
+
+        async def serve() -> asyncio.AbstractServer:
+            return await running.create_server(asyncio.Protocol, "0.0.0.0", 9000)
+
+        server = await loop.net.host("server").create_task(serve())
+        assert server.is_serving()
+        server.close()
+        await server.wait_closed()
+        assert not server.is_serving()
+        with pytest.raises(ConnectionRefusedError):
+            await asyncio.open_connection("server", 9000)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+
+
+def test_connect_cancelled_in_accept_window_leaves_nothing_connected() -> None:
+    # A timeout tuned to the accept's arrival can land in the very step that
+    # builds the client transport: connection_made runs, but the connector is
+    # cancelled before it is handed the transport. That half-open connection
+    # must be torn down, not orphaned. Seed 2 with a round-trip-length timeout
+    # deterministically lands in that window.
+    loop = _network(seed=2)
+    loop.net.set_defaults(latency=(0.05, 0.05))
+    server_lost: list[BaseException | None] = []
+    made: list[str] = []
+
+    class Server(asyncio.Protocol):
+        def connection_lost(self, exc: Exception | None) -> None:
+            server_lost.append(exc)
+
+    class Client(asyncio.Protocol):
+        def connection_made(self, transport: Any) -> None:
+            made.append("made")
+
+    async def main() -> None:
+        running = asyncio.get_running_loop()
+
+        async def serve() -> None:
+            server = await running.create_server(Server, "0.0.0.0", 9000)
+            async with server:
+                await asyncio.sleep(5.0)
+
+        async def connect() -> None:
+            with pytest.raises(TimeoutError):
+                async with asyncio.timeout(0.1):
+                    await running.create_connection(Client, "server", 9000)
+
+        serve_task = loop.net.host("server").create_task(serve())
+        await asyncio.sleep(0.001)
+        await loop.net.host("client").create_task(connect())
+        await asyncio.sleep(1.0)  # let the reset reach the server
+        await _reap(serve_task)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert made == ["made"]  # the accept window really was reached
+    assert not any(key[1] == "client" for key in loop.net._streams)
+    assert len(server_lost) == 1 and isinstance(server_lost[0], ConnectionResetError)

From 1f26e77d02269c440af656088d0e000e2cda5aeb Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Mon, 13 Jul 2026 00:39:20 +0530
Subject: [PATCH 4/9] Add network partitions that hold traffic until healed

---
 src/simloop/_net.py   | 46 ++++++++++++++++++++++++
 tests/test_net.py     | 69 ++++++++++++++++++++++++++++++++++++
 tests/test_streams.py | 81 ++++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 195 insertions(+), 1 deletion(-)

diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index e9bcebc..6a8389a 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import random
+from collections.abc import Iterable
 from contextvars import ContextVar
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
@@ -164,6 +165,8 @@ def __init__(self, loop: SimLoop) -> None:
         self._default_drop = 0.0
         self._default_duplicate = 0.0
         self._links: dict[tuple[str, str], _Link] = {}
+        self._cuts: set[frozenset[str]] = set()
+        self._held: list[_Packet] = []
         self._datagrams: dict[tuple[str, int], _SimDatagramTransport] = {}
         self._listeners: dict[tuple[str, int], _Listener] = {}
         self._streams: dict[tuple[int, str], _SimStreamTransport] = {}
@@ -221,6 +224,42 @@ def set_link(
         if duplicate is not None:
             link.duplicate = _check_probability("duplicate", duplicate)
 
+    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]
+        if not side_a or not side_b:
+            raise ValueError("both partition groups must be non-empty")
+        overlap = set(side_a) & set(side_b)
+        if overlap:
+            raise ValueError(
+                f"hosts cannot be on both sides of a partition: {sorted(overlap)}"
+            )
+        for a in side_a:
+            for b in side_b:
+                self._cuts.add(frozenset((a, b)))
+
+    def heal(self) -> None:
+        self._cuts.clear()
+        held, self._held = self._held, []
+        for packet in held:
+            self._trace("release", packet)
+            self._transmit(packet)
+
+    def _is_cut(self, a: str, b: str) -> bool:
+        return frozenset((a, b)) in self._cuts
+
+    def _blackhole(self, packet: _Packet) -> None:
+        # Datagrams crossing a cut are simply gone. Stream packets are held
+        # and released on heal: with no retransmission model, permanently
+        # dropping a mid-stream packet would leave the receiver waiting on a
+        # sequence gap forever, so held-then-released is what "the bytes stop
+        # flowing, then the connection resumes intact" has to mean here.
+        if packet.kind == "dgram":
+            self._trace("drop", packet)
+        else:
+            self._held.append(packet)
+            self._trace("hold", packet)
+
     def _resolved(self, src: str, dst: str) -> tuple[tuple[float, float], float, float]:
         link = self._links.get((src, dst))
         if link is None:
@@ -251,6 +290,9 @@ def _trace(self, verb: str, packet: _Packet) -> None:
         )
 
     def _transmit(self, packet: _Packet) -> None:
+        if self._is_cut(packet.src, packet.dst):
+            self._blackhole(packet)
+            return
         latency, drop, duplicate = self._resolved(packet.src, packet.dst)
         if packet.kind == "dgram":
             # Only datagrams are lossy: a reliable stream that loses bytes
@@ -272,6 +314,10 @@ def _deliver(self, packet: _Packet) -> None:
         if not (self._alive[packet.src] and self._alive[packet.dst]):
             self._trace("lost", packet)
             return
+        if self._is_cut(packet.src, packet.dst):
+            # The cut appeared while this packet was in flight.
+            self._blackhole(packet)
+            return
         # Anything the receiving protocol schedules (including tasks spawned
         # from connection_made or datagram_received) must be pinned to the
         # receiving host, not to whichever context sent the packet.
diff --git a/tests/test_net.py b/tests/test_net.py
index b4f2838..3576415 100644
--- a/tests/test_net.py
+++ b/tests/test_net.py
@@ -239,3 +239,72 @@ async def main() -> None:
         loop.run_until_complete(main())
     finally:
         loop.close()
+
+
+def test_partition_blackholes_datagrams_and_heal_restores() -> None:
+    loop = SimLoop(seed=0)
+    alpha = loop.net.host("alpha")
+    beta = loop.net.host("beta")
+
+    async def main() -> list[tuple[bytes, tuple[str, int]]]:
+        transport_a, collector = await _bound_endpoint(alpha, 7000)
+        transport_b, _ = await _bound_endpoint(beta, 7001)
+        loop.net.partition({"alpha"}, {"beta"})
+        transport_b.sendto(b"during", ("alpha", 7000))
+        await asyncio.sleep(0.5)
+        loop.net.heal()
+        transport_b.sendto(b"after", ("alpha", 7000))
+        await asyncio.sleep(0.5)
+        transport_a.close()
+        transport_b.close()
+        await asyncio.sleep(0.01)
+        return collector.received
+
+    try:
+        received = loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert received == [(b"after", ("beta", 7001))]
+    labels = [e.label for e in loop.trace if e.kind == "net"]
+    assert "drop beta>alpha" in labels
+
+
+def test_partition_validation() -> None:
+    loop = SimLoop(seed=0)
+    try:
+        net = loop.net
+        net.host("alpha")
+        net.host("beta")
+        with pytest.raises(OSError, match="unknown host"):
+            net.partition({"alpha"}, {"ghost"})
+        with pytest.raises(ValueError, match="both sides"):
+            net.partition({"alpha"}, {"alpha", "beta"})
+        with pytest.raises(ValueError, match="non-empty"):
+            net.partition(set(), {"alpha"})
+    finally:
+        loop.close()
+
+
+def test_driver_is_unaffected_by_partitions_it_is_not_named_in() -> None:
+    loop = SimLoop(seed=0)
+    alpha = loop.net.host("alpha")
+    loop.net.host("beta")
+    loop.net.partition({"alpha"}, {"beta"})
+
+    async def main() -> list[tuple[bytes, tuple[str, int]]]:
+        transport_a, collector = await _bound_endpoint(alpha, 7000)
+        driver_transport, _ = await asyncio.get_running_loop().create_datagram_endpoint(
+            _Collector, local_addr=("0.0.0.0", 7002)
+        )
+        driver_transport.sendto(b"hello", ("alpha", 7000))
+        await asyncio.sleep(0.5)
+        transport_a.close()
+        driver_transport.close()
+        await asyncio.sleep(0.01)
+        return collector.received
+
+    try:
+        received = loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert received == [(b"hello", ("driver", 7002))]
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 0e40a07..e200d09 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -7,7 +7,7 @@
 
 import pytest
 
-from simloop import SimLoop
+from simloop import SimLoop, SimulationDeadlockError
 
 
 def _network(seed: int = 0) -> SimLoop:
@@ -305,3 +305,82 @@ async def connect() -> None:
     assert made == ["made"]  # the accept window really was reached
     assert not any(key[1] == "client" for key in loop.net._streams)
     assert len(server_lost) == 1 and isinstance(server_lost[0], ConnectionResetError)
+
+
+def test_connect_across_partition_hangs_until_timeout() -> None:
+    # The syn is held before any listener lookup happens, so no server is
+    # needed to observe the hang: only the connector's own timeout fires.
+    loop = _network()
+    loop.net.partition({"server"}, {"client"})
+
+    async def connect() -> None:
+        with pytest.raises(TimeoutError):
+            async with asyncio.timeout(1.0):
+                await asyncio.open_connection("server", 9000)
+
+    async def main() -> None:
+        await loop.net.host("client").create_task(connect())
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+
+
+def test_connect_across_partition_without_timeout_is_a_deadlock() -> None:
+    # No timer is pending while the syn is held, so the run can never make
+    # progress — exactly the missing-timeout bug this tool exists to expose.
+    # Single top-level task on purpose: run_until_complete cancels and reaps
+    # only its own future on the stall path, so this stays stderr-clean.
+    loop = _network()
+    loop.net.partition({"server"}, {"driver"})
+
+    async def connect() -> None:
+        await asyncio.open_connection("server", 9000)
+
+    try:
+        with pytest.raises(SimulationDeadlockError):
+            loop.run_until_complete(connect())
+    finally:
+        loop.close()
+
+
+def test_stream_goes_silent_under_partition_and_resumes_after_heal() -> None:
+    loop = _network()
+
+    async def main() -> bytes:
+        running = asyncio.get_running_loop()
+
+        async def serve() -> None:
+            server = await asyncio.start_server(_echo_lines, "0.0.0.0", 9000)
+            async with server:
+                await asyncio.sleep(10.0)
+
+        async def request() -> bytes:
+            reader, writer = await asyncio.open_connection("server", 9000)
+            loop.net.partition({"server"}, {"client"})
+            writer.write(b"during\n")
+            with pytest.raises(TimeoutError):
+                async with asyncio.timeout(1.0):
+                    await reader.readline()
+            running.call_later(1.0, loop.net.heal)
+            line = await reader.readline()  # released bytes arrive after heal
+            writer.close()
+            await writer.wait_closed()
+            return line
+
+        serve_task = loop.net.host("server").create_task(serve())
+        await asyncio.sleep(0.01)
+        line: bytes = await loop.net.host("client").create_task(request())
+        await _reap(serve_task)
+        await asyncio.sleep(0.01)  # let the echoed connection's own fin land
+        return line
+
+    try:
+        line = loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert line == b"DURING\n"
+    labels = [e.label for e in loop.trace if e.kind == "net"]
+    assert "hold client>server" in labels
+    assert "release client>server" in labels

From 832670dd126d3f26f00e9786df966f17ca3f7796 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Mon, 13 Jul 2026 01:07:14 +0530
Subject: [PATCH 5/9] Add host crash injection

---
 src/simloop/_net.py |  26 +++++++-
 tests/test_net.py   | 140 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 165 insertions(+), 1 deletion(-)

diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index 6a8389a..6d42ad1 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -549,7 +549,31 @@ def _unbind_datagram(self, addr: tuple[str, int]) -> None:
         self._datagrams.pop(addr, None)
 
     def crash(self, name: str) -> None:
-        raise NotImplementedError
+        """Kill a host mid-run: its tasks are cancelled and it goes silent.
+
+        A crashed machine sends no reset — peers see nothing at all, which is
+        what makes crashes indistinguishable from partitions to the code
+        under test until a timeout says otherwise.
+        """
+        self._require_host(name)
+        if name == DRIVER:
+            raise ValueError("the driver host cannot crash")
+        if not self._alive[name]:
+            raise ValueError(f"host {name!r} already crashed")
+        self._alive[name] = False
+        for task in list(self._tasks[name]):
+            task.cancel()
+        for key in [key for key in self._listeners if key[0] == name]:
+            self._listeners[key].server.close()
+        for key in [key for key in self._datagrams if key[0] == name]:
+            del self._datagrams[key]
+        kept: list[_Packet] = []
+        for packet in self._held:
+            if name in (packet.src, packet.dst):
+                self._trace("lost", packet)
+            else:
+                kept.append(packet)
+        self._held = kept
 
     def _register_task(self, task: asyncio.Task[Any]) -> None:
         owner = self._tasks[_current_host.get()]
diff --git a/tests/test_net.py b/tests/test_net.py
index 3576415..ba41790 100644
--- a/tests/test_net.py
+++ b/tests/test_net.py
@@ -308,3 +308,143 @@ async def main() -> list[tuple[bytes, tuple[str, int]]]:
     finally:
         loop.close()
     assert received == [(b"hello", ("driver", 7002))]
+
+
+def test_crash_cancels_pinned_tasks_and_silences_traffic() -> None:
+    loop = SimLoop(seed=0)
+    alpha = loop.net.host("alpha")
+    beta = loop.net.host("beta")
+    cancelled: list[str] = []
+
+    async def chatter(transport: asyncio.DatagramTransport) -> None:
+        try:
+            while True:
+                transport.sendto(b"tick", ("alpha", 7000))
+                await asyncio.sleep(0.1)
+        except asyncio.CancelledError:
+            cancelled.append("chatter")
+            raise
+
+    async def main() -> int:
+        transport_a, collector = await _bound_endpoint(alpha, 7000)
+        transport_b, _ = await _bound_endpoint(beta, 7001)
+        beta.create_task(chatter(transport_b))
+        loop.call_later(0.35, beta.crash)
+        await asyncio.sleep(2.0)
+        transport_a.close()
+        await asyncio.sleep(0.01)
+        return len(collector.received)
+
+    try:
+        count = loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert cancelled == ["chatter"]
+    assert count == 4  # ticks at t=0, 0.1, 0.2, 0.3 — nothing after the crash
+    assert loop.net._tasks["beta"] == []
+
+
+def test_crash_discards_held_packets() -> None:
+    loop = SimLoop(seed=0)
+    alpha = loop.net.host("alpha")
+    beta = loop.net.host("beta")
+
+    async def swallow(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        try:
+            async with asyncio.timeout(5.0):
+                await reader.read()
+        except TimeoutError:
+            pass
+        finally:
+            writer.close()
+
+    async def serve() -> None:
+        server = await asyncio.start_server(swallow, "0.0.0.0", 9000)
+        async with server:
+            await asyncio.sleep(10.0)
+
+    async def open_stream() -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
+        return await asyncio.open_connection("beta", 9000)
+
+    async def main() -> None:
+        beta.create_task(serve())  # dies with the crash; no reaping needed
+        await asyncio.sleep(0.01)
+        _, writer = await alpha.create_task(open_stream())
+        loop.net.partition({"alpha"}, {"beta"})
+        writer.write(b"held\n")
+        await asyncio.sleep(0.1)
+        loop.net.crash("beta")  # the held write is discarded here
+        loop.net.heal()
+        await asyncio.sleep(0.5)
+        writer.close()
+        await asyncio.sleep(0.01)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    labels = [e.label for e in loop.trace if e.kind == "net"]
+    assert "hold alpha>beta" in labels
+    assert "lost alpha>beta" in labels
+    assert "release alpha>beta" not in labels
+
+
+def test_crash_guards() -> None:
+    loop = SimLoop(seed=0)
+    try:
+        net = loop.net
+        net.host("alpha")
+        with pytest.raises(OSError, match="unknown host"):
+            net.crash("ghost")
+        with pytest.raises(ValueError, match="driver"):
+            net.crash("driver")
+        net.crash("alpha")
+        with pytest.raises(ValueError, match="already crashed"):
+            net.crash("alpha")
+    finally:
+        loop.close()
+
+
+def test_server_tasks_are_pinned_to_the_server_not_the_dialing_client() -> None:
+    # Delivery pins the receiving context: a handler task spawned when the
+    # server accepts a connection must die with the server, and must survive
+    # the client.
+    loop = SimLoop(seed=0)
+    server_host = loop.net.host("api")
+    client_host = loop.net.host("cli")
+    outcome: list[str] = []
+
+    async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        try:
+            async with asyncio.timeout(5.0):
+                await reader.read()
+            outcome.append("finished")
+        except asyncio.CancelledError:
+            outcome.append("cancelled")
+            raise
+        finally:
+            writer.close()
+
+    async def main() -> None:
+        async def serve() -> None:
+            server = await asyncio.start_server(handler, "0.0.0.0", 9000)
+            async with server:
+                await asyncio.sleep(10.0)
+
+        async def connect() -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
+            return await asyncio.open_connection("api", 9000)
+
+        server_host.create_task(serve())
+        await asyncio.sleep(0.01)
+        _, writer = await client_host.create_task(connect())
+        loop.net.crash("cli")   # handler survives: it belongs to "api"
+        await asyncio.sleep(0.1)
+        loop.net.crash("api")   # now the handler dies
+        await asyncio.sleep(0.1)
+        writer.close()
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert outcome == ["cancelled"]

From c94d02edc39c93b376d1423f5b7aeeebae397450 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Mon, 13 Jul 2026 09:47:57 +0530
Subject: [PATCH 6/9] Tear down a crashed host's open connections

---
 src/simloop/_net.py |  8 ++++++++
 tests/test_net.py   | 49 +++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 57 insertions(+)

diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index 6d42ad1..e789ffc 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -567,6 +567,14 @@ def crash(self, name: str) -> None:
             self._listeners[key].server.close()
         for key in [key for key in self._datagrams if key[0] == name]:
             del self._datagrams[key]
+        # Tear down this host's own stream transports in-band: _finish sends no
+        # packet (a crashed host stays silent), pops the transport from
+        # _streams, and delivers connection_lost(None), so cleanup never falls
+        # to garbage collection. The materialized key list keeps _drop_stream's
+        # mutation from disturbing the iteration. The peer's transport for the
+        # same connection is left alone: it is still alive and times out itself.
+        for stream_key in [sk for sk in self._streams if sk[1] == name]:
+            self._streams[stream_key]._finish(None)
         kept: list[_Packet] = []
         for packet in self._held:
             if name in (packet.src, packet.dst):
diff --git a/tests/test_net.py b/tests/test_net.py
index ba41790..26b3925 100644
--- a/tests/test_net.py
+++ b/tests/test_net.py
@@ -389,6 +389,55 @@ async def main() -> None:
     assert "release alpha>beta" not in labels
 
 
+def test_crash_tears_down_the_hosts_open_connections() -> None:
+    # A crashed host's own stream transports must be dropped the moment it
+    # crashes, not left for the garbage collector to reap later: a transport
+    # that lingers in _streams is only reachable through the cancelled task's
+    # orphaned writer, whose close() then fires at GC time, so anything that
+    # depended on it would vary with GC timing rather than the seed.
+    loop = SimLoop(seed=0)
+    hub = loop.net.host("hub")
+    node = loop.net.host("node")
+
+    async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        try:
+            async with asyncio.timeout(5.0):
+                await reader.read()
+        except (TimeoutError, asyncio.CancelledError):
+            pass
+        finally:
+            writer.close()
+
+    async def serve() -> None:
+        server = await asyncio.start_server(handler, "0.0.0.0", 9000)
+        async with server:
+            await asyncio.sleep(10.0)
+
+    async def dial() -> None:
+        _, writer = await asyncio.open_connection("hub", 9000)
+        try:
+            while True:
+                await asyncio.sleep(0.05)
+        finally:
+            writer.close()
+
+    async def main() -> None:
+        hub.create_task(serve())
+        await asyncio.sleep(0.01)
+        node.create_task(dial())
+        await asyncio.sleep(0.05)
+        assert any(key[1] == "node" for key in loop.net._streams)
+        loop.net.crash("node")
+        assert not any(key[1] == "node" for key in loop.net._streams)
+        loop.net.crash("hub")  # reap the server side so no task lingers
+        await asyncio.sleep(0.05)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+
+
 def test_crash_guards() -> None:
     loop = SimLoop(seed=0)
     try:

From bac4a8085dae2d696fe77034ef514902de633fd2 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Mon, 13 Jul 2026 09:48:32 +0530
Subject: [PATCH 7/9] Add network replay stability tests

---
 tests/replay_net_workload.py | 144 +++++++++++++++++++++++++++++++++++
 tests/test_net_hardening.py  |  67 ++++++++++++++++
 2 files changed, 211 insertions(+)
 create mode 100644 tests/replay_net_workload.py
 create mode 100644 tests/test_net_hardening.py

diff --git a/tests/replay_net_workload.py b/tests/replay_net_workload.py
new file mode 100644
index 0000000..c4cbaf0
--- /dev/null
+++ b/tests/replay_net_workload.py
@@ -0,0 +1,144 @@
+"""Reference multi-host network workload for replay-stability checks.
+
+Importable for in-process runs; also runnable as a script —
+``python tests/replay_net_workload.py <seed>`` prints one line:
+``<trace_hash> <result_digest>``. It exercises streams, datagrams, link
+faults, a partition that heals, and a host crash, so every fault-injection
+draw is part of the replay proof.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import sys
+from typing import Any
+
+from simloop import Host, SimLoop, sim
+
+_PINGS = 12
+
+
+class _PingCounter(asyncio.DatagramProtocol):
+    def __init__(self) -> None:
+        self.log: list[str] = []
+
+    def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None:
+        self.log.append(f"{data.decode()}@{addr[0]}")
+
+
+async def _echo(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+    try:
+        async with asyncio.timeout(3.0):
+            while line := await reader.readline():
+                await asyncio.sleep(sim.random.uniform(0.001, 0.01))
+                writer.write(line.upper())
+                await writer.drain()
+    except TimeoutError:
+        pass
+    finally:
+        writer.close()
+
+
+async def _serve() -> None:
+    server = await asyncio.start_server(_echo, "0.0.0.0", 9000)
+    async with server:
+        await asyncio.sleep(6.0)
+
+
+async def _client(name: str, lines: int) -> list[str]:
+    reader, writer = await asyncio.open_connection("hub", 9000)
+    replies: list[str] = []
+    for number in range(lines):
+        writer.write(f"{name}-{number}-{sim.uuid4()}\n".encode())
+        await writer.drain()
+        replies.append((await reader.readline()).decode())
+        await asyncio.sleep(sim.random.uniform(0.001, 0.02))
+    writer.close()
+    await writer.wait_closed()
+    return replies
+
+
+async def _chatter() -> None:
+    reader, writer = await asyncio.open_connection("hub", 9000)
+    try:
+        while True:
+            writer.write(b"noise\n")
+            await asyncio.sleep(0.05)
+    except asyncio.CancelledError:
+        raise
+
+
+async def _ping(transport: asyncio.DatagramTransport) -> None:
+    for number in range(_PINGS):
+        transport.sendto(f"ping{number}".encode(), ("alpha", 7000))
+        await asyncio.sleep(0.02)
+
+
+async def _main(loop: SimLoop, hosts: dict[str, Host]) -> str:
+    net = loop.net
+    hub, alpha, beta, gamma = (hosts[n] for n in ("hub", "alpha", "beta", "gamma"))
+
+    serve_task = hub.create_task(_serve())
+    await asyncio.sleep(0.01)
+
+    counter = _PingCounter()
+
+    async def bind_counter() -> asyncio.DatagramTransport:
+        transport, _ = await asyncio.get_running_loop().create_datagram_endpoint(
+            lambda: counter, local_addr=("0.0.0.0", 7000)
+        )
+        return transport
+
+    async def bind_pinger() -> asyncio.DatagramTransport:
+        transport, _ = await asyncio.get_running_loop().create_datagram_endpoint(
+            asyncio.DatagramProtocol, local_addr=("0.0.0.0", 7001)
+        )
+        return transport
+
+    counter_transport = await alpha.create_task(bind_counter())
+    pinger_transport = await beta.create_task(bind_pinger())
+
+    tasks = [
+        alpha.create_task(_client("alpha", 4)),
+        beta.create_task(_client("beta", 4)),
+        beta.create_task(_ping(pinger_transport)),
+    ]
+    gamma.create_task(_chatter())
+
+    loop.call_later(0.08, net.partition, {"beta"}, {"hub", "alpha"})
+    loop.call_later(0.20, net.heal)
+    loop.call_later(0.12, net.crash, "gamma")
+
+    results: list[Any] = []
+    for task in tasks:
+        results.append(await task)
+    await asyncio.sleep(4.0)  # let echo handlers time out and finish cleanly
+    serve_task.cancel()
+    try:
+        await serve_task
+    except asyncio.CancelledError:
+        pass
+    counter_transport.close()
+    pinger_transport.close()
+    await asyncio.sleep(0.01)
+    results.append(sorted(counter.log))
+    return repr(results)
+
+
+def run(seed: int) -> str:
+    loop = SimLoop(seed)
+    net = loop.net
+    hosts = {name: net.host(name) for name in ("hub", "alpha", "beta", "gamma")}
+    net.set_defaults(latency=(0.001, 0.03))
+    net.set_link("beta", "alpha", drop=0.25, duplicate=0.25)
+    try:
+        outcome = loop.run_until_complete(_main(loop, hosts))
+    finally:
+        loop.close()
+    digest = hashlib.sha256(outcome.encode()).hexdigest()
+    return f"{loop.trace_hash()} {digest}"
+
+
+if __name__ == "__main__":
+    print(run(int(sys.argv[1])))
diff --git a/tests/test_net_hardening.py b/tests/test_net_hardening.py
new file mode 100644
index 0000000..4d504d2
--- /dev/null
+++ b/tests/test_net_hardening.py
@@ -0,0 +1,67 @@
+"""Replay-stability checks for the simulated network (slow ones run in CI)."""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+_SCRIPT = Path(__file__).with_name("replay_net_workload.py")
+
+
+def _load_workload() -> Any:
+    spec = importlib.util.spec_from_file_location("replay_net_workload", _SCRIPT)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _run_child(seed: int, hashseed: str | None) -> str:
+    env = os.environ.copy()
+    env.pop("PYTHONHASHSEED", None)
+    if hashseed is not None:
+        env["PYTHONHASHSEED"] = hashseed
+    result = subprocess.run(
+        [sys.executable, str(_SCRIPT), str(seed)],
+        capture_output=True,
+        text=True,
+        env=env,
+        check=True,
+        timeout=60,
+    )
+    assert result.stderr == ""
+    return result.stdout.strip()
+
+
+def test_same_seed_replays_identically() -> None:
+    workload = _load_workload()
+    for seed in range(3):
+        assert workload.run(seed) == workload.run(seed)
+
+
+def test_different_seeds_diverge() -> None:
+    workload = _load_workload()
+    assert len({workload.run(seed) for seed in range(3)}) == 3
+
+
+@pytest.mark.slow
+def test_hundred_network_reruns_per_seed_are_stable() -> None:
+    workload = _load_workload()
+    for seed in range(5):
+        results = {workload.run(seed) for _ in range(100)}
+        assert len(results) == 1, f"seed {seed} produced diverging runs"
+
+
+@pytest.mark.slow
+def test_network_replay_is_stable_across_processes_and_hash_seeds() -> None:
+    workload = _load_workload()
+    for seed in (0, 7):
+        results = {_run_child(seed, hs) for hs in (None, "0", "1", "random")}
+        results.add(workload.run(seed))
+        assert len(results) == 1, f"seed {seed} diverged across processes"

From 8722f0c0d85fb804d9643e79f97d1954c77f11ac Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Mon, 13 Jul 2026 09:53:59 +0530
Subject: [PATCH 8/9] Document the simulated network

---
 README.md             |  9 ++++-----
 docs/supported-api.md | 24 ++++++++++++++++++++----
 2 files changed, 24 insertions(+), 9 deletions(-)

diff --git a/README.md b/README.md
index 3cccf7d..3f1531f 100644
--- a/README.md
+++ b/README.md
@@ -1,7 +1,7 @@
 # simloop
 
 Deterministic simulation testing for Python asyncio — seeded task scheduling,
-virtual time, and (planned) a simulated network with fault injection. Any failure
+virtual time, and a simulated network with fault injection. Any failure
 it finds replays exactly from a seed.
 
 Rust has [madsim](https://github.com/madsim-rs/madsim) and
@@ -17,16 +17,15 @@ a test that deadlocks *every* time at seed 4217, in milliseconds, under a debugg
 virtual time, replay-proving trace hashes, deadlock detection, and seeded
 `sim.random` / `sim.uuid4` / `sim.time` shims, with the stdlib coordination
 primitives (`Queue`, `Event`, `Lock`, `gather`, `TaskGroup`, `timeout`)
-running unchanged on top. No simulated network yet. See
+running unchanged on top. The simulated network is in: `asyncio.open_connection`,
+`start_server`, and datagram endpoints run unchanged over in-memory transports
+with seeded latency, drops, duplication, partitions, and host crashes. See
 [the supported subset](docs/supported-api.md) for the exact contract.
 
 ## Planned
 
 - **SimLoop** — a deterministic `asyncio` event loop: virtual clock, seeded
   ready-queue ordering, append-only scheduling trace with replay-proving hashes.
-- **SimNetwork** — in-memory transports behind the standard asyncio streams and
-  protocol APIs, with seeded fault injection: latency, drops, duplication,
-  reordering, and network partitions.
 - **pytest plugin** — `@sim_test(seeds=1000)` to explore schedules, and an exact
   replay flag for any failing seed.
 
diff --git a/docs/supported-api.md b/docs/supported-api.md
index 073b5d4..9428db6 100644
--- a/docs/supported-api.md
+++ b/docs/supported-api.md
@@ -26,11 +26,27 @@ determinism. Fenced APIs raise `SimulationFenceError` (a subclass of
 — everything built purely on futures, tasks and timers. Timeouts fire in
 virtual time. Each of these claims is exercised by the test suite.
 
-## Planned: simulated network
+## Simulated network
 
-`create_connection`, `create_server`, `create_datagram_endpoint` and the
-streams API, backed by in-memory transports with seeded fault injection
-(latency, drops, duplication, reordering, partitions).
+Networking runs over an in-memory packet layer with seeded fault injection.
+Hosts are named machines; a task started via `loop.net.host("name").create_task(...)`
+— and every task it spawns — belongs to that host. Tasks never started under
+a host belong to an implicit `driver` host.
+
+| API | Behavior under simulation |
+|---|---|
+| `loop.create_connection` / `create_server`, `asyncio.open_connection` / `start_server` | Real transports and protocols over reliable, ordered in-memory streams; connecting costs one round trip of virtual latency; connecting to a closed port raises `ConnectionRefusedError` |
+| `loop.create_datagram_endpoint` | Unreliable messaging: per-link drop, duplication, and latency apply per datagram |
+| `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 |
+| `transport.abort()` | Peer gets `connection_lost(ConnectionResetError)` |
+
+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
+reliable by construction; addresses are host names, not IPs, and
+`getaddrinfo` stays fenced.
 
 ## Fenced
 

From b9d0c570ec7f7bb91f162e9699c89d25ccd1c266 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Mon, 13 Jul 2026 10:15:26 +0530
Subject: [PATCH 9/9] Support client-closing on simulated servers

---
 src/simloop/_net.py   | 12 ++++++++
 tests/test_streams.py | 71 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 83 insertions(+)

diff --git a/src/simloop/_net.py b/src/simloop/_net.py
index e789ffc..e43eeef 100644
--- a/src/simloop/_net.py
+++ b/src/simloop/_net.py
@@ -102,6 +102,18 @@ def close(self) -> None:
     def is_serving(self) -> bool:
         return not self._closed_fut.done()
 
+    def close_clients(self) -> None:
+        """Close every connection this server accepted."""
+        for transport in list(self._net._streams.values()):
+            if transport._local == (self._host, self._port):
+                transport.close()
+
+    def abort_clients(self) -> None:
+        """Reset every connection this server accepted."""
+        for transport in list(self._net._streams.values()):
+            if transport._local == (self._host, self._port):
+                transport.abort()
+
     async def wait_closed(self) -> None:
         await asyncio.shield(self._closed_fut)
 
diff --git a/tests/test_streams.py b/tests/test_streams.py
index e200d09..4edf464 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -198,6 +198,77 @@ async def connect_and_abort() -> None:
     assert isinstance(lost[0], ConnectionResetError)
 
 
+def test_server_close_clients_disconnects_the_peer() -> None:
+    loop = _network()
+    lost: list[BaseException | None] = []
+    holder: dict[str, Any] = {}
+
+    class Peer(asyncio.Protocol):
+        def connection_lost(self, exc: Exception | None) -> None:
+            lost.append(exc)
+
+    async def main() -> None:
+        running = asyncio.get_running_loop()
+
+        async def serve() -> None:
+            holder["server"] = await running.create_server(
+                asyncio.Protocol, "0.0.0.0", 9000
+            )
+            await asyncio.sleep(10.0)
+
+        async def connect() -> None:
+            await running.create_connection(Peer, "server", 9000)
+
+        serve_task = loop.net.host("server").create_task(serve())
+        await asyncio.sleep(0.01)
+        await loop.net.host("client").create_task(connect())
+        holder["server"].close_clients()
+        await asyncio.sleep(0.5)
+        await _reap(serve_task)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert lost == [None]
+
+
+def test_server_abort_clients_resets_the_peer() -> None:
+    loop = _network()
+    lost: list[BaseException | None] = []
+    holder: dict[str, Any] = {}
+
+    class Peer(asyncio.Protocol):
+        def connection_lost(self, exc: Exception | None) -> None:
+            lost.append(exc)
+
+    async def main() -> None:
+        running = asyncio.get_running_loop()
+
+        async def serve() -> None:
+            holder["server"] = await running.create_server(
+                asyncio.Protocol, "0.0.0.0", 9000
+            )
+            await asyncio.sleep(10.0)
+
+        async def connect() -> None:
+            await running.create_connection(Peer, "server", 9000)
+
+        serve_task = loop.net.host("server").create_task(serve())
+        await asyncio.sleep(0.01)
+        await loop.net.host("client").create_task(connect())
+        holder["server"].abort_clients()
+        await asyncio.sleep(0.5)
+        await _reap(serve_task)
+
+    try:
+        loop.run_until_complete(main())
+    finally:
+        loop.close()
+    assert len(lost) == 1
+    assert isinstance(lost[0], ConnectionResetError)
+
+
 def test_duplicate_bind_and_foreign_bind_are_rejected() -> None:
     loop = _network()