cultnet-py: an ended session fails its flush; a repeated Connect owes one Accept

flush_reliable raised nothing for writes the ended session had forgotten; it
now raises ConnectionError with the reason. Server mode answered every
repeated Connect by queueing a fresh reliable Accept; a Connect from the peer
already accepted is now answered with the Accept still owed and queues
nothing, as the Rust hub does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/cultnet-py/src/cultnet_py/transport.py

@@ -505,6 +505,25 @@ class CultNetRudpSession:
         self._track_reliable(response, now_ms)
         return response
 
+    def answer_repeated_connect(self, packet: CultNetRudpPacket, now_ms: int = 0) -> CultNetRudpPacket:
+        """Answers a Connect from a peer this session already accepted.
+
+        The packet's sequence is remembered and nothing is queued, so a Connect
+        storm cannot grow the reliable queue. The reply is the Accept still awaiting
+        acknowledgement, or an Ack once it was acknowledged.
+        """
+        self._require_connection(packet)
+        if packet.packet_type != CultNetRudpPacketType.CONNECT:
+            raise ValueError(f"Expected RUDP connect packet, got {packet.packet_type.value}")
+        self._apply_acknowledgements(packet)
+        self._remember_received(packet.sequence)
+        self._last_received_at_ms = now_ms
+        for pending in self._pending_reliable.values():
+            if pending.packet.packet_type == CultNetRudpPacketType.ACCEPT:
+                pending.last_sent_at_ms = now_ms
+                return pending.packet
+        return self.create_ack()
+
     def send(
         self,
         channel_id: str,
@@ -975,6 +994,7 @@ class CultNetRudpSocketTransportConnection:
     def connect(self, payload: bytes = b"") -> None:
         if self.mode != CultNetRudpSocketMode.CLIENT:
             raise ValueError("Only a client RUDP socket transport can initiate connect")
+        self.disconnect_reason = None
         self._send_packet(self.session.create_connect(_now_ms(), payload))
 
     def send(self, channel_id: str, payload: bytes) -> None:
@@ -1024,12 +1044,18 @@ class CultNetRudpSocketTransportConnection:
             return None
 
         if self.mode == CultNetRudpSocketMode.SERVER and packet.packet_type == CultNetRudpPacketType.CONNECT:
+            # A Connect from the peer this transport already accepted repeats:
+            # answer it with the Accept already owed and queue nothing.
             try:
-                accept = self.session.accept_connect(packet, _now_ms())
+                if self.session.connected:
+                    reply = self.session.answer_repeated_connect(packet, _now_ms())
+                else:
+                    reply = self.session.accept_connect(packet, _now_ms())
+                    self.disconnect_reason = None
             except ValueError:
                 self._packets_dropped += 1
                 return None
-            self._send_packet(accept)
+            self._send_packet(reply)
             return None
 
         try:
@@ -1080,7 +1106,16 @@ class CultNetRudpSocketTransportConnection:
         self._delivered_frames.clear()
         self.socket.settimeout(poll_timeout)
         try:
-            while self.session.outstanding_reliable_packet_count > 0:
+            while True:
+                # An ended session forgot its unacknowledged writes; reporting
+                # them flushed would be a lie.
+                if self.disconnect_reason is not None:
+                    raise ConnectionError(
+                        "RUDP session ended before its reliable writes were acknowledged: "
+                        + self.disconnect_reason.decode("utf-8", errors="replace")
+                    )
+                if self.session.outstanding_reliable_packet_count == 0:
+                    return
                 if time.monotonic() >= deadline:
                     raise TimeoutError(
                         "RUDP reliable flush timed out with "

packages/cultnet-py/tests/test_cultnet.py

@@ -1079,6 +1079,63 @@ class CultNetTests(unittest.TestCase):
             peer_socket.close()
             server.close()
 
+    def _server_mode_pair(self, connection_id: int):
+        server_socket = bind_udp_socket()
+        peer_socket = bind_udp_socket()
+        server = CultNetRudpSocketTransportConnection(
+            CultNetRudpSocketTransportOptions(
+                runtime_id="python-rudp-server",
+                socket=server_socket,
+                mode=CultNetRudpSocketMode.SERVER,
+                connection_id=connection_id,
+            )
+        )
+        peer = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id))
+
+        def to_server(packet) -> None:
+            peer_socket.sendto(encode_rudp_packet(packet), server_socket.getsockname())
+
+        def drain_peer() -> list:
+            packets = []
+            while True:
+                try:
+                    wire, _ = peer_socket.recvfrom(65535)
+                except TimeoutError:
+                    return packets
+                packets.append(decode_rudp_packet(wire))
+
+        return server, server_socket, peer_socket, peer, to_server, drain_peer
+
+    def test_cultnet_rudp_flush_fails_for_a_write_the_ended_session_forgot(self) -> None:
+        server, _, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x10203054)
+        try:
+            to_server(peer.create_connect(0, b"join"))
+            server.receive_once()
+            peer.receive([p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0], 0)
+            server.send("schema", b"never acknowledged")
+            (frame,) = peer.send_many("schema", b"poison", CultNetRudpSendOptions(reliable=True, ordered=True))
+            to_server(replace(frame, fragment_count=2, fragment_id=0))
+            self.assertIsNone(server.receive_once())
+            with self.assertRaises(ConnectionError):
+                server.flush_reliable(0.2)
+        finally:
+            peer_socket.close()
+            server.close()
+
+    def test_cultnet_rudp_server_mode_owes_one_reliable_accept_however_many_connects_repeat(self) -> None:
+        server, _, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x10203055)
+        try:
+            connect = peer.create_connect(0, b"join")
+            for _ in range(20):
+                to_server(connect)
+                server.receive_once()
+            accepts = {p.sequence for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT}
+            self.assertEqual(len(accepts), 1)
+            self.assertLessEqual(server.outstanding_reliable_packet_count, 1)
+        finally:
+            peer_socket.close()
+            server.close()
+
     def test_cultnet_rudp_session_rejects_limits_that_cannot_admit_a_connect(self) -> None:
         with self.assertRaises(ValueError):
             CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1, initial_sequence=0xFFFFFFFF))