merge: absorb samwhosung/main on top of the upstream merge

Same fork policy: upstream structure wins, local-unique re-instated.

Cargo.toml unions both measured dev-profile philosophies (codegen-units=1 +
lto=off from samwhosung, debug=line-tables-only + split-debuginfo from the
fork's disk measurements).

glue_booth.rs takes samwhosung's typed GlueScene (token/of_race/from_model_path)
and re-adds the fork's TurtleWoW races (Goblin=9, BloodElf=10) as variants;
their glue fog stays on Human's row, exactly as the fork's scene_fog fallback did.

ui script takes samwhosung's mouseover/hyperlink rework (world_mouseover,
minimap_blip, hyperlink.rs) over the fork's inline hover_link/span_at/pointer
path — the rework covers the same OnHyperlinkEnter/Leave behaviour.

game_plugins.rs keeps the fork's re-measured ceilings (466 / 5,509); mail.rs
keeps the explicit 13-tuple with a float daysLeft; MAP.md takes the superset.

crates/benilla-protocol/src/auth.rs

@@ -43,8 +43,25 @@ const LOCALE_EN_US: u32 = 0x656e_5553; // "enUS"
 pub struct AuthReject {
     /// The grunt result byte (`WOW_FAIL_*`).
     pub code: u8,
+    /// Which reply carried it: the reference handles a byte by its reply, swallowing a proof's
+    /// `0x0A` (`0x5bada2`) where a challenge's reaches the patch state.
+    pub reply: LogonReply,
 }
 
+/// The two realmd replies a logon reads a result byte from.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum LogonReply {
+    /// `CMD_AUTH_LOGON_CHALLENGE` (`Grunt::OnLogonChallenge`, `0x5baa20`).
+    Challenge,
+    /// `CMD_AUTH_LOGON_PROOF` (`Grunt::OnLogonProof`, `0x5bacb0`).
+    Proof,
+}
+
+/// `WOW_SUCCESS_SURVEY`: a success. Both reply handlers take it down their failure branch
+/// (`0x5baa77`, `0x5bace5` test the byte for zero), reading nothing after it, and
+/// `Logon::OnAuthResult` sends it down 0x00's success arm (`0x5b2cd4` `cmp eax,0xe; je 0x5b2e16`).
+pub const SUCCESS_SURVEY: u8 = 0x0e;
+
 impl std::fmt::Display for AuthReject {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "server rejected logon: result {:#04x}", self.code)
@@ -93,16 +110,24 @@ pub fn write_logon_challenge(
     w.write_all(&packet)
 }
 
-/// Read the challenge reply; a non-success result is an [`AuthReject`].
-pub fn read_challenge_reply(r: &mut impl Read) -> Result<ChallengeReply> {
+/// Read the challenge reply: the SRP6 inputs, `None` for [`SUCCESS_SURVEY`] (a success that
+/// carries none), or an [`AuthReject`].
+pub fn read_challenge_reply(r: &mut impl Read) -> Result<Option<ChallengeReply>> {
     let opcode = read_u8(r)?;
     if opcode != CMD_AUTH_LOGON_CHALLENGE {
         bail!("expected CMD_AUTH_LOGON_CHALLENGE (0x00), got {opcode:#x}");
     }
     let _protocol_version = read_u8(r)?;
     let result = read_u8(r)?;
+    if result == SUCCESS_SURVEY {
+        return Ok(None);
+    }
     if result != 0 {
-        return Err(AuthReject { code: result }.into());
+        return Err(AuthReject {
+            code: result,
+            reply: LogonReply::Challenge,
+        }
+        .into());
     }
     let server_public_key = read_array::<32>(r)?;
     let generator_len = read_u8(r)?;
@@ -129,13 +154,13 @@ pub fn read_challenge_reply(r: &mut impl Read) -> Result<ChallengeReply> {
         let _pin_grid_seed = read_u32_le(r)?;
         let _pin_salt = read_array::<16>(r)?;
     }
-    Ok(ChallengeReply {
+    Ok(Some(ChallengeReply {
         server_public_key,
         generator,
         large_safe_prime,
         salt,
         crc_salt,
-    })
+    }))
 }
 
 // --- the version (client-integrity) proof --------------------------------------------------------
@@ -209,8 +234,9 @@ pub fn write_logon_proof(
     w.write_all(&packet)
 }
 
-/// Read the proof reply: the server's `M2`, or an [`AuthReject`].
-pub fn read_proof_reply(r: &mut impl Read) -> Result<[u8; 20]> {
+/// Read the proof reply: the server's `M2`, `None` for [`SUCCESS_SURVEY`] (a success with no `M2`
+/// read), or an [`AuthReject`].
+pub fn read_proof_reply(r: &mut impl Read) -> Result<Option<[u8; 20]>> {
     let opcode = read_u8(r)?;
     if opcode != CMD_AUTH_LOGON_PROOF {
         // mangos-family quirk: a build realmd will not accept is rejected at the proof stage with a
@@ -225,13 +251,20 @@ pub fn read_proof_reply(r: &mut impl Read) -> Result<[u8; 20]> {
         bail!("expected CMD_AUTH_LOGON_PROOF (0x01), got {opcode:#x}");
     }
     let result = read_u8(r)?;
+    if result == SUCCESS_SURVEY {
+        return Ok(None);
+    }
     if result != 0 {
         // A wrong password fails here, where `M1` does not verify: vmangos answers 0x04.
-        return Err(AuthReject { code: result }.into());
+        return Err(AuthReject {
+            code: result,
+            reply: LogonReply::Proof,
+        }
+        .into());
     }
     let server_proof = read_array::<20>(r)?;
     let _hardware_survey_id = read_u32_le(r)?;
-    Ok(server_proof)
+    Ok(Some(server_proof))
 }
 
 /// Send `CMD_REALM_LIST_Client` (opcode + a `u32` padding of 0).

crates/benilla-protocol/src/events.rs

@@ -71,13 +71,30 @@ pub enum LoginStage {
     Authenticating,
     /// Realm picked; the world-socket handshake and char enum (`LOGIN_STATE_HANDSHAKING`).
     Handshaking,
+    /// The world socket connected, before its authentication (`RESPONSE_CONNECTED`, `0x5aaa90`).
+    WorldConnected,
+}
+
+/// What a roster was fetched for. The screen takes a reply as its own only when it asked.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RosterCause {
+    /// The IO thread's own enumeration on reaching the character park, which the native screens
+    /// wait for; the stock screens ask for theirs instead.
+    Park,
+    /// The reply to a roster request from the screen (`CMSG_CHAR_ENUM` for
+    /// `GetCharacterListUpdate`, `0x5aabb0`).
+    Asked,
+    /// A create or delete changed the roster; re-sent before the action's result.
+    Changed,
 }
 
 /// How a world session ended. On a loss the reference fires `DISCONNECTED_FROM_SERVER` and
 /// `GlueParent.lua` returns to the login screen; a logout returns to character select.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum SessionEnd {
-    /// A logout the server confirmed (`SMSG_LOGOUT_COMPLETE`); a fresh roster follows.
+    /// A logout the server confirmed (`SMSG_LOGOUT_COMPLETE`), or a refused character login: the
+    /// world link is kept and the session is back at character select on it, as the reference's
+    /// `0x5aaeb0` and `0x5aad70` close nothing.
     LoggedOut,
     /// The stream died: a socket error, an EOF or a post-roster handshake failure. A displacement
     /// kick is this too: vmangos `WorldSession::KickPlayer` only closes the socket, no packet.
@@ -88,8 +105,11 @@ pub enum SessionEnd {
 /// `AUTH_LOGON_FAILED_SUSPENDED` to realmd but `AUTH_OK` to the world server.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum LoginRefusal {
-    /// realmd's logon-proof `AuthLogonResult` (`crate::AuthReject`).
+    /// realmd's `AuthLogonResult` in its logon-challenge reply (`crate::AuthReject`).
     Logon(u8),
+    /// realmd's `AuthLogonResult` in its logon-proof reply, where the reference swallows `0x0A`
+    /// (`0x5bada2`).
+    LogonProof(u8),
     /// The world server's `SMSG_AUTH_RESPONSE` code (`crate::WorldAuthReject`).
     World(u8),
 }
@@ -98,7 +118,7 @@ impl LoginRefusal {
     /// The raw byte, for logging only: without its server it is ambiguous.
     pub fn byte(self) -> u8 {
         match self {
-            LoginRefusal::Logon(b) | LoginRefusal::World(b) => b,
+            LoginRefusal::Logon(b) | LoginRefusal::LogonProof(b) | LoginRefusal::World(b) => b,
         }
     }
 }
@@ -112,9 +132,28 @@ impl LoginRefusal {
 pub enum SessionEvent {
     /// A login attempt reached `stage`; emitted by the IO thread, never wire-decoded.
     LoginStage { stage: LoginStage },
-    /// The account's realms (`CMD_REALM_LIST`); the IO thread blocks for the app's pick. Re-sent on
-    /// each refresh (the reference re-requests every 5 s) and on a realm change.
-    RealmList { realms: Vec<crate::RealmInfo> },
+    /// The account's realms (`CMD_REALM_LIST`); emitted by the IO thread after the logon and for
+    /// each query (the reference re-requests every 5 s). `failed` is a query that got no answer
+    /// (`REALM_LIST_FAILED`); `realms` is then the list last held, empty with no logon.
+    RealmList {
+        realms: Vec<crate::RealmInfo>,
+        failed: bool,
+    },
+    /// The world server accepted the authentication (`SMSG_AUTH_RESPONSE` `AUTH_OK`, `0x5ab2b8`
+    /// → `0x5aab80`): the world connect is done and the IO thread is at the character park, with
+    /// no roster fetched unless the native policy's enumeration follows. Emitted by the IO thread,
+    /// never wire-decoded. `epoch` numbers the world session, per process; `generation` is the
+    /// abandon generation of the attempt that dialed it, which a Cancel or a disconnect since has
+    /// moved past (the reference drops the socket there, `0x46befd`, `0x46d360`).
+    WorldAuthenticated {
+        realm: crate::RealmInfo,
+        epoch: u64,
+        generation: u64,
+    },
+    /// A roster ask on the parked session failed on the stream (`0x5ab300`'s failure leg,
+    /// `SetStatus(0, 0x2c)`, `CHAR_LIST_FAILED`); the session ends after it. Emitted by the IO
+    /// thread, never wire-decoded.
+    RosterFailed { epoch: u64 },
     /// Queued for a full realm (`AUTH_WAIT_QUEUE`), per queue packet: a wait, not an outcome.
     LoginQueued {
         position: Option<u32>,
@@ -134,6 +173,11 @@ pub enum SessionEvent {
     CharacterList {
         characters: Vec<Character>,
         realm: Option<crate::RealmInfo>,
+        cause: RosterCause,
+        /// The world session's number ([`Self::WorldAuthenticated`]'s), 0 on the raw decode path.
+        epoch: u64,
+        /// The abandon generation of the attempt that dialed that session.
+        generation: u64,
     },
     /// `SMSG_CHAR_CREATE`/`_DELETE`'s raw `WorldResult`; on success a fresh roster comes first.
     CharActionResult { action: CharAction, code: u8 },
@@ -152,14 +196,16 @@ pub enum SessionEvent {
     /// Our pick was refused (`SMSG_CHARACTER_LOGIN_FAILED`) after [`Self::Connected`] was sent;
     /// `result` is a 1-based index into the reference's six `CHAR_LOGIN_*` strings.
     CharacterLoginFailed { result: u8 },
-    /// Logout confirmed (`SMSG_LOGOUT_COMPLETE`); a fresh [`Self::CharacterList`] follows.
+    /// Logout confirmed (`SMSG_LOGOUT_COMPLETE`); the session returns to character select on the
+    /// same link, where the roster is asked for again.
     LoggedOut,
     /// `SMSG_LOGOUT_RESPONSE`. A non-zero `reason` refuses: 1 in combat, 2 GM-frozen, 3 falling
     /// (vmangos). Zero starts the 20 s camp timer, or with `instant` (resting, taxi, GM) logs out.
     LogoutResponse { reason: u32, instant: bool },
     /// The server dropped a pending logout at our `CMSG_LOGOUT_CANCEL` (`SMSG_LOGOUT_CANCEL_ACK`).
     LogoutCancelled,
-    /// The session ended; `reason` is human-readable.
+    /// The world session ended, and with [`SessionEnd::Lost`] the link with it; `reason` is
+    /// human-readable.
     Disconnected { reason: String, end: SessionEnd },
     /// A placed object entered view, its spawn identity decoded per object type.
     ObjectCreate {
@@ -1111,14 +1157,21 @@ impl SessionEventKind {
 /// The result of polling the reader for the next packet's events.
 pub enum Poll {
     /// A decoded packet; `events` is empty for an opcode parsed but not modelled. `tail` counts
-    /// body bytes left unread, which shows a decoder shorter than the server's layout.
+    /// body bytes left unread, which shows a decoder shorter than the server's layout. `body` is
+    /// the packet as it came off the wire, after the header, for the packet trace.
     Events {
         opcode: u16,
         events: Vec<SessionEvent>,
         tail: usize,
+        body: Vec<u8>,
+    },
+    /// An unparseable packet, skipped to keep the stream aligned; `reason` has a hex preview and
+    /// `body` the whole packet.
+    Skipped {
+        opcode: u16,
+        reason: String,
+        body: Vec<u8>,
     },
-    /// An unparseable packet, skipped to keep the stream aligned; `reason` has a hex preview.
-    Skipped { opcode: u16, reason: String },
 }
 
 mod decode;

crates/benilla-protocol/src/events/decode.rs

@@ -31,6 +31,9 @@ pub fn decode(packet: ServerPacket) -> Vec<SessionEvent> {
         ServerPacket::CharEnum { characters } => vec![SessionEvent::CharacterList {
             characters,
             realm: None,
+            cause: super::RosterCause::Park,
+            epoch: 0,
+            generation: 0,
         }],
         ServerPacket::CharacterLoginFailed { result } => {
             vec![SessionEvent::CharacterLoginFailed { result }]

crates/benilla-protocol/src/lib.rs

@@ -7,10 +7,10 @@ pub mod guid;
 pub mod messages;
 pub mod wire;
 pub mod world;
-pub use auth::AuthReject;
+pub use auth::{AuthReject, LogonReply};
 pub use events::{
-    decode, CharAction, EntityKind, LoginRefusal, LoginStage, MoveSpeeds, Poll, SessionEnd,
-    SessionEvent, SessionEventKind,
+    decode, CharAction, EntityKind, LoginRefusal, LoginStage, MoveSpeeds, Poll, RosterCause,
+    SessionEnd, SessionEvent, SessionEventKind,
 };
 pub use messages::field;
 pub use messages::{
@@ -20,8 +20,8 @@ pub use messages::{
     CHARACTER_FLAG_HIDE_CLOAK, CHARACTER_FLAG_HIDE_HELM, CHARACTER_FLAG_RENAME,
 };
 pub use world::{
-    observe_sends, SendObserver, WardenRequired, WorldAuthReject, WorldReader, WorldSession,
-    WorldWriter, WORLD_PORT,
+    observe_sends, SendObserver, Unparsed, WardenRequired, WorldAuthReject, WorldConnectEvent,
+    WorldReader, WorldSession, WorldWriter, WORLD_PORT,
 };
 
 use std::net::TcpStream;
@@ -173,6 +173,12 @@ impl Logon {
     pub fn realmd_live(&self) -> bool {
         self.stream.is_some()
     }
+
+    /// `CLoginMgr::Cancel` (`0x5b3320`): the realmd socket closes; the session key and the last
+    /// list are kept.
+    pub fn close(&mut self) {
+        self.stream = None;
+    }
 }
 
 /// The full SRP6 logon against a vanilla `realmd`, then the realm list.
@@ -193,8 +199,13 @@ pub fn logon(host: &str, username: &str, password: &str) -> Result<Logon> {
             let mut stream = dial(host, port)?;
             auth::write_logon_challenge(&mut stream, &username.to_uppercase(), REALMD_CLIENT_BUILD)
                 .context("sending logon challenge")?;
-            let reply =
-                auth::read_challenge_reply(&mut stream).context("reading logon challenge reply")?;
+            let Some(reply) =
+                auth::read_challenge_reply(&mut stream).context("reading logon challenge reply")?
+            else {
+                // `SUCCESS_SURVEY` on the challenge authenticates with no SRP exchange; the
+                // reference's success arm copies no key (length 0 at `0x5b2e16`).
+                return finish_logon(stream, [0u8; SESSION_KEY_LENGTH]);
+            };
             let server_public_key = PublicKey::from_le_bytes(reply.server_public_key)
                 .map_err(|e| anyhow!("invalid server public key: {e}"))?;
             let stable = server_public_key.is_width_stable();
@@ -224,12 +235,21 @@ pub fn logon(host: &str, username: &str, password: &str) -> Result<Logon> {
     )
     .context("sending logon proof")?;
 
-    let server_proof = auth::read_proof_reply(&mut stream).context("reading logon proof reply")?;
-    let client = challenge
-        .verify_server_proof(server_proof)
-        .map_err(|e| anyhow!("server proof mismatch (wrong password?): {e}"))?;
-    let session_key = *client.session_key();
+    let session_key =
+        match auth::read_proof_reply(&mut stream).context("reading logon proof reply")? {
+            Some(server_proof) => *challenge
+                .verify_server_proof(server_proof)
+                .map_err(|e| anyhow!("server proof mismatch (wrong password?): {e}"))?
+                .session_key(),
+            // `SUCCESS_SURVEY` on the proof authenticates with no `M2` to check; the key is the one
+            // this exchange derived.
+            None => challenge.into_unverified_session_key(),
+        };
+    finish_logon(stream, session_key)
+}
 
+/// An authenticated logon's tail: the realm list over the same socket.
+fn finish_logon(mut stream: TcpStream, session_key: [u8; SESSION_KEY_LENGTH]) -> Result<Logon> {
     auth::write_realm_list_request(&mut stream).context("requesting realm list")?;
     let realms = auth::read_realm_list(&mut stream).context("reading realm list")?;
 

crates/benilla-protocol/src/world/mod.rs

@@ -19,7 +19,8 @@ mod session;
 mod writer;
 
 pub use reader::WorldReader;
-pub use session::{WardenRequired, WorldAuthReject, WorldSession};
+pub use session::{WardenRequired, WorldAuthReject, WorldConnectEvent, WorldSession};
+
 pub use writer::WorldWriter;
 
 /// The stock `mangosd` port, for probes that dial the world server without a realm list.
@@ -45,11 +46,34 @@ pub(super) fn recv_packet(
     stream
         .read_exact(&mut body)
         .map_err(|e| anyhow!("reading world body (opcode {opcode:#x}, {body_len} bytes): {e}"))?;
-    messages::parse_server(opcode, &body).map_err(|e| anyhow!("parsing opcode {opcode:#x}: {e}"))
+    messages::parse_server(opcode, &body).map_err(|e| {
+        Unparsed {
+            opcode,
+            reason: e.to_string(),
+        }
+        .into()
+    })
 }
 
-/// What [`observe_sends`] takes: called with a sent packet's opcode and body length.
-pub type SendObserver = fn(opcode: u16, body_len: usize);
+/// A packet whose body did not parse. The whole body was read, so the stream is still aligned and
+/// a reader may skip it and go on, as [`WorldReader::poll`] does.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Unparsed {
+    pub opcode: u16,
+    pub reason: String,
+}
+
+impl std::fmt::Display for Unparsed {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "parsing opcode {:#x}: {}", self.opcode, self.reason)
+    }
+}
+
+impl std::error::Error for Unparsed {}
+
+/// What [`observe_sends`] takes: called with a sent packet's opcode and its body, the bytes after
+/// the 6-byte header and as plain as the caller built them (encryption covers the header alone).
+pub type SendObserver = fn(opcode: u16, body: &[u8]);
 
 static SEND_OBSERVER: OnceLock<SendObserver> = OnceLock::new();
 
@@ -87,7 +111,7 @@ pub(super) fn send_packet(
         .write_all(&packet)
         .map_err(|e| anyhow!("sending opcode {opcode:#x}: {e}"))?;
     if let Some(observer) = SEND_OBSERVER.get() {
-        observer(opcode, body.len());
+        observer(opcode, body);
     }
     Ok(())
 }

crates/benilla-protocol/src/world/reader.rs

@@ -8,10 +8,13 @@ use crate::messages::{self, ServerPacket};
 
 use super::recv_packet;
 
-/// The read half of a split [`WorldSession`](super::WorldSession): cloned socket and decrypter.
+/// The read half of a split [`WorldSession`](super::WorldSession): cloned socket and decrypter,
+/// and what the connection learned before the split, for
+/// [`WorldSession::rejoin`](super::WorldSession::rejoin).
 pub struct WorldReader {
     pub(super) stream: TcpStream,
     pub(super) decrypter: DecrypterHalf,
+    pub(super) link: super::session::LinkFacts,
 }
 
 impl WorldReader {
@@ -40,6 +43,7 @@ impl WorldReader {
                 opcode,
                 events: crate::decode(packet),
                 tail,
+                body,
             }),
             Err(e) => Ok(crate::Poll::Skipped {
                 opcode,
@@ -48,6 +52,7 @@ impl WorldReader {
                     messages::opcode_name(opcode).unwrap_or("?"),
                     hex_preview(&body, 64)
                 ),
+                body,
             }),
         }
     }

crates/benilla-protocol/src/world/session.rs

@@ -58,21 +58,41 @@ impl std::error::Error for WardenRequired {}
 pub struct WorldSession {
     stream: TcpStream,
     crypto: HeaderCrypto,
-    /// Roster guid to race, so [`Self::player_login`] can pick the chat language.
-    roster_races: std::collections::HashMap<u64, u8>,
+    link: LinkFacts,
     /// The language chat sends speak, the character's faction tongue: vmangos drops chat, even
     /// dot-commands, in a language the character does not know.
     chat_language: u32,
+    /// `SMSG_TUTORIAL_FLAGS` when it lands during the handshake rather than in the world stream.
+    tutorial_flags: Option<Vec<u8>>,
+}
+
+/// What the connection learned that outlives one character's world session, carried by the read
+/// half across a split so [`WorldSession::rejoin`] has it back.
+#[derive(Default)]
+pub(super) struct LinkFacts {
+    /// Roster guid to race, so [`WorldSession::player_login`] can pick the chat language.
+    roster_races: std::collections::HashMap<u64, u8>,
     /// The rested billing minutes the admitting `SMSG_AUTH_RESPONSE` carried. Deviation: `0` when
     /// the body is too short, where the reference leaves its global as it was; this field lives
     /// per connection, with nothing earlier to keep.
     billing_time_rested: u32,
-    /// `SMSG_TUTORIAL_FLAGS` when it lands during the handshake rather than in the world stream.
-    tutorial_flags: Option<Vec<u8>>,
     /// `SMSG_ADDON_INFO`'s per-record status bytes in order; `None` until the server answers.
+    /// vmangos sends it once per connection, after the authentication
+    /// (`WorldSocket.cpp:446-448`), and the reference applies it to add-on records that outlive
+    /// the world session (`0x5b4260` → `0x51da70`), so every login on the link reads it.
     addon_info: Option<Vec<u8>>,
 }
 
+/// What a world connect reports on its way to `SMSG_AUTH_RESPONSE`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum WorldConnectEvent {
+    /// The socket connected: the reference's connect operation completes `RESPONSE_CONNECTED`
+    /// here (`0x5aaa90`) and its authentication begins.
+    Connected,
+    /// A queue position (`AUTH_WAIT_QUEUE`), or `None` when the body carried none.
+    Queued(Option<u32>),
+}
+
 impl WorldSession {
     /// Connect and complete the auth handshake with the logon's account and session key.
     pub fn connect(
@@ -83,17 +103,21 @@ impl WorldSession {
         Self::connect_queued(addr, username, session_key, &mut |_| true)
     }
 
-    /// [`Self::connect`], calling `on_queue` with each queue position; `false` abandons the queue.
+    /// [`Self::connect`], calling `on_event` once the socket connects and with each queue position;
+    /// `false` abandons the connect.
     /// Deviation: checked per queue packet where the reference checks each tick, so a cancel can
     /// lag one server update.
     pub fn connect_queued(
         addr: impl ToSocketAddrs,
         username: &str,
         session_key: [u8; SESSION_KEY_LENGTH],
-        on_queue: &mut dyn FnMut(Option<u32>) -> bool,
+        on_event: &mut dyn FnMut(WorldConnectEvent) -> bool,
     ) -> Result<Self> {
         let mut queued = false;
         let mut stream = TcpStream::connect(addr).context("connecting to world server")?;
+        if !on_event(WorldConnectEvent::Connected) {
+            bail!("world connect abandoned");
+        }
         // Nagle off: the reference sets `TCP_NODELAY` on its game socket (`0x5bca60`).
         stream
             .set_nodelay(true)
@@ -130,11 +154,9 @@ impl WorldSession {
         let mut session = WorldSession {
             stream,
             crypto,
-            roster_races: Default::default(),
+            link: LinkFacts::default(),
             chat_language: messages::LANGUAGE_COMMON,
-            billing_time_rested: 0,
             tutorial_flags: None,
-            addon_info: None,
         };
 
         // AUTH_RESPONSE is not always first, so others are skipped; Warden data ends the connect.
@@ -146,7 +168,7 @@ impl WorldSession {
                     ..
                 } if result == messages::AUTH_OK => {
                     // The reference keeps the last AUTH_RESPONSE's billing group.
-                    session.billing_time_rested = billing_time_rested.unwrap_or(0);
+                    session.link.billing_time_rested = billing_time_rested.unwrap_or(0);
                     break;
                 }
                 // Queued, not refused: the server re-sends as we move up and ends with `AUTH_OK`.
@@ -161,7 +183,7 @@ impl WorldSession {
                             .set_read_timeout(Some(QUEUE_READ_TIMEOUT))
                             .context("relaxing the read timeout for the login queue")?;
                     }
-                    if !on_queue(queue_position) {
+                    if !on_event(WorldConnectEvent::Queued(queue_position)) {
                         bail!("login queue abandoned");
                     }
                 }
@@ -186,7 +208,7 @@ impl WorldSession {
 
     /// Rested billing minutes (`GetBillingTimeRested`); always 0 from vmangos (`World.cpp:331`).
     pub fn billing_time_rested(&self) -> u32 {
-        self.billing_time_rested
+        self.link.billing_time_rested
     }
 
     /// The tutorial bank captured during the login handshake, if any.
@@ -206,15 +228,27 @@ impl WorldSession {
         let packet = recv_packet(&mut self.stream, Some(self.crypto.decrypter()))?;
         // `SMSG_ADDON_INFO` can reach any of the handshake's read loops, so it is caught here.
         if let ServerPacket::AddonInfo { statuses } = &packet {
-            self.addon_info = Some(statuses.clone());
+            self.link.addon_info = Some(statuses.clone());
         }
         Ok(packet)
     }
 
-    /// The `SMSG_ADDON_INFO` statuses, taken once; `None` is real, as vmangos stays silent when it
-    /// rejects the addon block (`WorldSocket.cpp:447`).
-    pub fn take_addon_info(&mut self) -> Option<Vec<u8>> {
-        self.addon_info.take()
+    /// The next packet a request at character select reads past, or `None` for one whose body did
+    /// not parse: skipped, as the world stream skips it ([`WorldReader::poll`]), since back from
+    /// the world the link still carries what the world session left in flight (a pong, a reply
+    /// vmangos serves to an authenticated session).
+    fn recv_parsed(&mut self) -> Result<Option<ServerPacket>> {
+        match self.recv() {
+            Ok(packet) => Ok(Some(packet)),
+            Err(e) if e.downcast_ref::<super::Unparsed>().is_some() => Ok(None),
+            Err(e) => Err(e),
+        }
+    }
+
+    /// The `SMSG_ADDON_INFO` statuses this connection received, for every login on it; `None` is
+    /// real, as vmangos stays silent when it rejects the addon block (`WorldSocket.cpp:447`).
+    pub fn addon_info(&self) -> Option<&[u8]> {
+        self.link.addon_info.as_deref()
     }
 
     /// Send a client packet (encrypted header + plaintext body).
@@ -231,9 +265,12 @@ impl WorldSession {
     pub fn char_enum(&mut self) -> Result<Vec<Character>> {
         self.send(opcode::CMSG_CHAR_ENUM, &[])?;
         loop {
-            match self.recv()? {
+            let Some(packet) = self.recv_parsed()? else {
+                continue;
+            };
+            match packet {
                 ServerPacket::CharEnum { characters } => {
-                    self.roster_races = characters.iter().map(|c| (c.guid, c.race)).collect();
+                    self.link.roster_races = characters.iter().map(|c| (c.guid, c.race)).collect();
                     return Ok(characters);
                 }
                 // Warden can arm on either side of SMSG_AUTH_RESPONSE, so this step refuses it too.
@@ -255,9 +292,8 @@ impl WorldSession {
     pub fn create_character(&mut self, req: &messages::CharCreateReq) -> Result<u8> {
         self.send(opcode::CMSG_CHAR_CREATE, &messages::char_create(req))?;
         loop {
-            match self.recv()? {
-                ServerPacket::CharCreate { result } => return Ok(result),
-                _ => continue,
+            if let Some(ServerPacket::CharCreate { result }) = self.recv_parsed()? {
+                return Ok(result);
             }
         }
     }
@@ -266,9 +302,8 @@ impl WorldSession {
     pub fn delete_character(&mut self, guid: u64) -> Result<u8> {
         self.send(opcode::CMSG_CHAR_DELETE, &messages::full_guid(guid))?;
         loop {
-            match self.recv()? {
-                ServerPacket::CharDelete { result } => return Ok(result),
-                _ => continue,
+            if let Some(ServerPacket::CharDelete { result }) = self.recv_parsed()? {
+                return Ok(result);
             }
         }
     }
@@ -276,6 +311,7 @@ impl WorldSession {
     /// Enter the world as `guid`, adopting its faction tongue for chat (Common if not enumerated).
     pub fn player_login(&mut self, guid: u64) -> Result<()> {
         self.chat_language = self
+            .link
             .roster_races
             .get(&guid)
             .map_or(messages::LANGUAGE_COMMON, |&race| {
@@ -773,6 +809,7 @@ impl WorldSession {
             WorldReader {
                 stream: read_stream,
                 decrypter,
+                link: self.link,
             },
             WorldWriter {
                 stream: self.stream,
@@ -782,6 +819,29 @@ impl WorldSession {
         ))
     }
 
+    /// The halves [`Self::into_split`] made, one session again on the same connection: a logout
+    /// or a refused character login returns to character select on the link it entered by,
+    /// where the reference re-enumerates (`0x5aaeb0` → `0x401ee0(1, 1, 0)` → `0x46b880`), with
+    /// no new authentication. The cipher keeps the state each direction reached, and the reads
+    /// wait with the handshake's bound again, as every park step awaits one reply.
+    pub fn rejoin(reader: WorldReader, writer: WorldWriter) -> Result<Self> {
+        // The read half is a clone of the write half's socket; dropping it closes nothing.
+        let WorldReader {
+            decrypter, link, ..
+        } = reader;
+        let session = WorldSession {
+            stream: writer.stream,
+            crypto: HeaderCrypto::join(writer.encrypter, decrypter),
+            link,
+            chat_language: writer.chat_language,
+            tutorial_flags: None,
+        };
+        session
+            .set_read_timeout(Some(HANDSHAKE_READ_TIMEOUT))
+            .context("restoring the park's read timeout")?;
+        Ok(session)
+    }
+
     /// Log out to character select, waiting up to `timeout`; needs a socket read timeout.
     pub fn logout(&mut self, timeout: std::time::Duration) -> Result<()> {
         self.send(opcode::CMSG_LOGOUT_REQUEST, &[])?;