Logout returns to character select on the same world connection (#1022)

In 1.12.1, logging out or a refused character login returns to character select on the same world connection and re-lists there (`0x5aaeb0` → `0x401ee0`, then `0x46b880` → `CMSG_CHAR_ENUM`). There's no realmd logon and no new `CMSG_AUTH_SESSION`. Enter World closes the realmd link (`0x46b70c` → `0x5b3320`), and only the glue's own re-dials reopen it.

- benilla used to drop the socket and log on again from cached credentials. Now the world session splits and rejoins on one connection, and the header cipher carries its state across.
- The realmd link closes at the pick.
- A leftover packet at character select is skipped rather than fatal.
- The write thread applies a pending writer swap before any command.
- A kept-link teardown ends the character without a "server down".

Fake-server tests pin one logon, one auth and one more enumerate. The logout smoke fails without a same-link relist.

Fixes #984
Fixes #983

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

crates/benilla-app/src/char_select/mod.rs

@@ -85,20 +85,7 @@ impl Plugin for CharSelectPlugin {
             .add_systems(
                 Update,
                 (
-                    // Runs in every state: the reconnect auto-answer and the logout edge arrive
-                    // `InWorld`. `back_on_disconnect` runs after the entry: a session that died
-                    // during entry queues both edges into one frame, and the dead one must win.
-                    (
-                        apply_roster_policy,
-                        enter_on_connected,
-                        back_on_logout,
-                        back_on_login_refused,
-                        back_on_disconnect,
-                        // Last and ungated: it mirrors the pick the frame ended with, and world
-                        // entry is reached from the create screen too.
-                        persist_last_character,
-                    )
-                        .chain(),
+                    session_edges(),
                     (
                         screen::materialize_screen,
                         input::select_input.run_if(crate::ui_script::glue::native_glue),
@@ -138,6 +125,26 @@ impl Plugin for CharSelectPlugin {
     }
 }
 
+/// The session's edges, in every state: the reconnect auto-answer and the logout edge arrive
+/// `InWorld`. A logout or a refused login clears the pick before the roster policy reads it: the
+/// roster comes right behind them on the kept link, often in the same drain, and must be shown,
+/// not answered with the pick just left. `back_on_disconnect` runs after the entry: a session that
+/// died during entry queues both edges into one frame, and the dead one must win.
+fn session_edges() -> bevy::ecs::schedule::ScheduleConfigs<bevy::ecs::system::ScheduleSystem> {
+    (
+        back_on_logout,
+        back_on_login_refused,
+        apply_roster_policy,
+        enter_on_connected,
+        back_on_disconnect,
+        // Last and ungated: it mirrors the pick the frame ended with, and world entry is reached
+        // from the create screen too.
+        persist_last_character,
+    )
+        .chain()
+        .into_configs()
+}
+
 // ── The roster and pick policy ──────────────────────────────────────────────────────────────────
 
 /// The account roster and the pick policy's memory.
@@ -442,7 +449,7 @@ fn back_on_disconnect(
 
 /// `SMSG_CHARACTER_LOGIN_FAILED`: back to select with the refusal in the glue dialog. The IO
 /// thread announces the entry with the pick, so this undoes a half-built one. The pick must be
-/// cleared, or the relist behind the refusal is auto-answered and the refused login loops.
+/// cleared, or the roster behind the refusal is auto-answered and the refused login loops.
 fn back_on_login_refused(
     mut msgs: MessageReader<CharacterLoginFailedMessage>,
     mut roster: ResMut<Roster>,
@@ -589,7 +596,8 @@ fn debug_glue_roundtrip(
 }
 
 /// The logout smoke (`WOW_LOGOUT_SMOKE=1`, with `WOW_CHAR`): once in the world, `/logout`, return
-/// to CharSelect, enter the world again and close the window. The second entry streams a map
+/// to CharSelect, wait for the roster the logout brings on the link the world was entered by,
+/// enter the world again and close the window. The second entry streams a map
 /// `terrain_stream::release_world` tore down. It ends by a window close, not an `AppExit`, since
 /// only the close reaches the shutdown tail that saves variables and addon files.
 fn debug_logout_smoke(
@@ -603,12 +611,35 @@ fn debug_logout_smoke(
     mut exit: MessageWriter<AppExit>,
     mut close: MessageWriter<bevy::window::WindowCloseRequested>,
     windows: Query<Entity, With<bevy::window::PrimaryWindow>>,
+    mut lists: MessageReader<CharListMessage>,
+    // The world session's number at the entry, and whether a roster has come since the logout.
+    mut link: Local<Option<u64>>,
+    mut relisted: Local<bool>,
     mut phase: Local<u8>,
     mut mark: Local<f32>,
 ) {
     if std::env::var("WOW_LOGOUT_SMOKE").is_err() {
         return;
     }
+    for list in lists.read() {
+        if *phase < 2 {
+            *link = Some(list.epoch);
+        } else if !*relisted {
+            *relisted = true;
+            if Some(list.epoch) == *link {
+                info!(
+                    "logout-smoke: roster after the logout on the same link — {} char(s)",
+                    list.characters.len()
+                );
+            } else {
+                error!(
+                    "logout-smoke: FAILED — the roster after the logout came on world session {} \
+                     where the world was entered on {:?}",
+                    list.epoch, *link
+                );
+            }
+        }
+    }
     let now = time.elapsed_secs();
     match *phase {
         0 if *state.get() == ClientState::InWorld && player.active => {
@@ -626,21 +657,24 @@ fn debug_logout_smoke(
         }
         // Residency is read here, not on arrival: `release_world` runs about 150 ms after the
         // state edge, on the world-live falling edge.
-        3 if now - *mark > 4.0 => match roster.chars.first().map(|c| (c.guid, c.name.clone())) {
-            Some((guid, name)) => {
-                info!(
-                    "logout-smoke: {} tiles resident after the release (must be 0)",
-                    streamer.residency().1
-                );
-                info!("logout-smoke: re-entering the world as {name}");
-                send_pick(&mut roster, &pick, guid);
-                (*phase, *mark) = (4, now);
+        // The roster the logout brings, not the one that survived it, decides the re-entry.
+        3 if now - *mark > 4.0 && *relisted => {
+            match roster.chars.first().map(|c| (c.guid, c.name.clone())) {
+                Some((guid, name)) => {
+                    info!(
+                        "logout-smoke: {} tiles resident after the release (must be 0)",
+                        streamer.residency().1
+                    );
+                    info!("logout-smoke: re-entering the world as {name}");
+                    send_pick(&mut roster, &pick, guid);
+                    (*phase, *mark) = (4, now);
+                }
+                None => {
+                    warn!("logout-smoke: empty roster — cannot test re-entry");
+                    *phase = 5;
+                }
             }
-            None => {
-                warn!("logout-smoke: empty roster — cannot test re-entry");
-                *phase = 5;
-            }
-        },
+        }
         4 if *state.get() == ClientState::InWorld && player.active && now - *mark > 3.0 => {
             // The suppressor reading catches a character that re-entered unable to move;
             // `scripts/smoke.sh` fails on anything but `none`. A GM account's logout is instant,
@@ -1056,6 +1090,56 @@ mod tests {
         );
     }
 
+    /// A logout, or a refused login, and the roster the kept link sends right behind it can land
+    /// in one drain: the pick the world was entered with goes first, so the roster is shown and
+    /// not answered with that pick again.
+    #[test]
+    fn a_return_and_its_roster_in_one_drain_show_the_roster() {
+        for refused in [false, true] {
+            let (tx, rx) = crossbeam_channel::unbounded();
+            let mut app = App::new();
+            app.add_plugins((MinimalPlugins, bevy::state::app::StatesPlugin))
+                .insert_state(ClientState::InWorld)
+                .insert_resource(Roster::with_pending_pick(vec![character(7, "Seven")], 7))
+                .insert_resource(CharPick(tx.into()))
+                .init_resource::<crate::cvars::Cvars>()
+                .init_resource::<crate::glue::dialog::GlueDialog>()
+                .add_message::<CharListMessage>()
+                .add_message::<EnteredWorldMessage>()
+                .add_message::<LoggedOutMessage>()
+                .add_message::<CharacterLoginFailedMessage>()
+                .add_message::<crate::net::DisconnectedMessage>()
+                .add_systems(Update, session_edges());
+            app.world_mut().resource_mut::<Roster>().env_read = true;
+            if refused {
+                app.world_mut()
+                    .write_message(CharacterLoginFailedMessage { result: 0x01 });
+            } else {
+                app.world_mut().write_message(LoggedOutMessage);
+            }
+            app.world_mut().write_message(CharListMessage {
+                characters: vec![character(7, "Seven")],
+                realm: None,
+                cause: benilla_protocol::RosterCause::Park,
+                epoch: 1,
+                generation: 0,
+            });
+            app.update();
+            app.update();
+
+            assert_eq!(
+                *app.world().resource::<State<ClientState>>().get(),
+                ClientState::CharSelect,
+                "refused: {refused}"
+            );
+            assert!(
+                rx.try_recv().is_err(),
+                "refused: {refused} — the roster was answered with the pick just left"
+            );
+            assert_eq!(app.world().resource::<Roster>().pending_pick, None);
+        }
+    }
+
     /// Every refusal byte resolves to the sentence 1.12 ships, off the player's chain. `4` is
     /// `CHAR_LOGIN_DISABLED`: the reference's table skips `CHAR_LOGIN_FAILED` (0x41).
     #[test]

crates/benilla-app/src/login/mod.rs

@@ -124,8 +124,8 @@ enum IoPark {
 /// the resubmit timer.
 #[derive(Resource, Default)]
 pub(crate) struct LoginIntent {
-    /// Kept in-world so the logout relist and an unattended reconnect re-authenticate silently;
-    /// cleared by select's Back, a refusal, a Cancel and a lost session.
+    /// Kept in-world so an unattended reconnect re-authenticates silently; cleared by select's
+    /// Back, a refusal, a Cancel and a lost session.
     creds: Option<(String, String)>,
     /// Between our send and its LoginFailed or CharacterList answer.
     in_flight: bool,
@@ -602,8 +602,13 @@ fn drive_policy(
         }
     }
     for msg in disconnects.read() {
-        attempt.intent.park = IoPark::AtLogin;
         attempt.intent.in_flight = false;
+        // A logout or a refused character login keeps the world link: the IO thread is back at
+        // character select on it (`0x5aaeb0`, `0x5aad70`), with nothing to log on again.
+        if msg.end == benilla_protocol::SessionEnd::LoggedOut {
+            continue;
+        }
+        attempt.intent.park = IoPark::AtLogin;
         if msg.session_over {
             // `DISCONNECTED_FROM_SERVER`: `GlueParent.lua` shows the login screen and the
             // `DISCONNECTED` dialog. The credentials go, so nothing takes the account back.
@@ -615,14 +620,9 @@ fn drive_policy(
             dialog.open_error(strings.text("DISCONNECTED", "Disconnected from server"));
             continue;
         }
-        // Otherwise a silent re-auth: at once after a logout, paced after a stream death.
+        // Otherwise an unattended run's silent re-auth, paced after the stream death.
         if attempt.intent.creds.is_some() {
-            let delay = if msg.end == benilla_protocol::SessionEnd::LoggedOut {
-                0.0
-            } else {
-                RETRY_DELAY_SECS
-            };
-            attempt.intent.retry_at = Some(now + delay);
+            attempt.intent.retry_at = Some(now + RETRY_DELAY_SECS);
         }
     }
 
@@ -638,11 +638,11 @@ fn drive_policy(
     }
 }
 
-/// The login's success edge: go to CharSelect, but only from `Login`, since a reconnect's roster
-/// lands while `InWorld`. The native screens take the roster for it, which the IO thread fetches on
-/// reaching the park; the stock screens take the world server's authentication
-/// (`SMSG_AUTH_RESPONSE`, `0x5ab2b8`), since the roster there is the screen's own ask and comes
-/// only once the screen is up.
+/// The login's success edge: go to CharSelect, but only from `Login`, since an unattended
+/// reconnect's roster lands while `InWorld`, and a logout's at character select. The native
+/// screens take the roster for it, which the IO thread fetches on reaching the park; the stock
+/// screens take the world server's authentication (`SMSG_AUTH_RESPONSE`, `0x5ab2b8`), since the
+/// roster there is the screen's own ask and comes only once the screen is up.
 fn to_select_on_roster(
     mut msgs: MessageReader<CharListMessage>,
     mut auths: MessageReader<WorldAuthMessage>,
@@ -684,8 +684,8 @@ fn to_select_on_roster(
     intent.in_flight = false;
     intent.park = IoPark::Active;
     intent.retry_at = None;
-    // Only the dialogs the roster answers: a refused character login's relist brings a roster,
-    // and its `Error` is dismissed by its Okay, never by a packet.
+    // Only the dialogs the roster answers: a refused character login's roster follows it on the
+    // kept link, and its `Error` is dismissed by its Okay, never by a packet.
     if matches!(dialog.kind, Some(DialogKind::Status | DialogKind::Queued)) {
         dialog.close();
     }
@@ -1335,19 +1335,23 @@ mod tests {
         assert_eq!(dialog.text, "Disconnected from server");
     }
 
-    /// A logout rides the same message and keeps its silent relist; its roster is the character
-    /// select the player asked for.
+    /// A logout rides the same message but keeps the world link, so nothing logs on again: the
+    /// IO thread is at character select on that link, and the roster comes on it.
     #[test]
-    fn a_logout_teardown_still_relists_at_once() {
-        let (mut app, _requests) = policy_app();
-        app.world_mut().resource_mut::<LoginIntent>().creds =
-            Some(("player".into(), "secret".into()));
+    fn a_logout_teardown_logs_on_no_more() {
+        let (mut app, requests) = policy_app();
+        {
+            let mut intent = app.world_mut().resource_mut::<LoginIntent>();
+            intent.creds = Some(("player".into(), "secret".into()));
+            intent.park = IoPark::Active;
+        }
         app.world_mut().write_message(DisconnectedMessage {
             reason: "logged out".into(),
             end: benilla_protocol::SessionEnd::LoggedOut,
             session_over: false,
         });
         app.update();
+        app.update();
 
         let intent = app.world().resource::<LoginIntent>();
         assert!(
@@ -1355,9 +1359,12 @@ mod tests {
             "a logout keeps the account signed in"
         );
         assert!(
-            intent.in_flight,
-            "and the same tick resubmits — the delay for a logout is 0, so the roster comes \
-             straight back",
+            intent.park == IoPark::Active && !intent.in_flight && intent.retry_at.is_none(),
+            "and submits nothing: the link it entered by is still up",
+        );
+        assert!(
+            requests.try_recv().is_err(),
+            "no login request reached the IO thread"
         );
         assert!(
             app.world().resource::<GlueDialog>().kind.is_none(),
@@ -1365,8 +1372,8 @@ mod tests {
         );
     }
 
-    /// A refused character login's relist brings a roster a second later; it must not close
-    /// the refusal's `Error`.
+    /// A refused character login's roster comes on the kept link right behind the refusal; it must
+    /// not close the refusal's `Error`.
     #[test]
     fn an_arriving_roster_closes_the_connecting_dialog_but_not_an_error() {
         let mut app = App::new();
@@ -1378,7 +1385,7 @@ mod tests {
             .add_message::<WorldAuthMessage>()
             .add_systems(Update, to_select_on_roster);
 
-        // The refusal's dialog survives its own relist.
+        // The refusal's dialog survives the roster behind it.
         app.world_mut()
             .resource_mut::<GlueDialog>()
             .open_error("World server is down");

crates/benilla-app/src/net.rs

@@ -515,7 +515,8 @@ pub(crate) enum RealmRequest {
     /// credentials. At character select, where `RealmListDialogCancelled` is a no-op, nothing.
     Close,
     /// `ConnectToRealm` with the world up (`0x46b22b` → `0x5ab1a0`): the session at character
-    /// select drops and the thread returns to the logon park, realmd still up. Elsewhere, nothing.
+    /// select drops and the thread returns to the logon park, the realmd link as it was (closed
+    /// since a world entry). Elsewhere, nothing.
     Leave,
 }
 
@@ -2163,7 +2164,7 @@ pub(crate) struct WorldEnterCascadeMessage;
 
 /// `SMSG_ADDON_INFO`: the addons the server hid from the Lua index space, or `None` if it did
 /// not answer. A resource, since it must exist before the world-entry UI load runs any addon;
-/// rewritten on every login.
+/// rewritten on every login from the reply its link got.
 #[derive(Resource, Default)]
 pub(crate) struct AddonInfoReply(pub(crate) Option<Vec<String>>);
 
@@ -2197,7 +2198,9 @@ pub(crate) struct LoginQueuedMessage {
     pub(crate) realm: Option<String>,
 }
 
-/// The session ended; the IO thread is back at its pre-logon park.
+/// The world session ended. After a logout or a refused character login ([`SessionEnd::LoggedOut`](
+/// benilla_protocol::SessionEnd::LoggedOut)) the IO thread is back at character select on the
+/// same link; after a loss, at its pre-logon park.
 #[derive(Message, Clone)]
 pub(crate) struct DisconnectedMessage {
     pub(crate) reason: String,
@@ -2209,8 +2212,8 @@ pub(crate) struct DisconnectedMessage {
 }
 
 impl DisconnectedMessage {
-    /// `session_over` is false after a logout (the relist is the select the player asked for) and
-    /// in a run declared unattended ([`crate::run_mode::unattended`]), which reconnects. Not the
+    /// `session_over` is false after a logout (the link is kept and character select is what the
+    /// player asked for) and in a run declared unattended ([`crate::run_mode::unattended`]), which reconnects. Not the
     /// login env vars: a person's launch line sets them too, and a reconnect would take the
     /// account back from whoever displaced us.
     pub(crate) fn new(reason: String, end: benilla_protocol::SessionEnd) -> Self {

crates/benilla-app/src/net/io.rs

@@ -3,16 +3,19 @@
 //! [`spawn_net`]'s read thread parks pre-logon for a [`LoginRequest`], logs on to realmd, then
 //! holds the realmd link at the logon park, serving realm list queries, until the app names a realm
 //! or closes the logon; a world attempt that fails returns there. Past the world handshake it
-//! parks at character select until the app picks a guid; a new login supersedes either park. The pick sends
-//! `CMSG_PLAYER_LOGIN` and announces the connection without waiting for the verdict, as the
-//! reference's world is already loading when it sends the login (`0x46c272`), so the destination's
-//! tiles stream a round trip early; a later `SMSG_CHARACTER_LOGIN_FAILED` ends the cycle like a
-//! logout ([`Cycle::LoginRefused`]). A stream failure emits
-//! [`SessionEvent::Disconnected`] with [`SessionEnd::Lost`] and re-parks. All policy lives in
-//! [`crate::login`] and [`crate::char_select`]; this thread only sequences and never sleeps.
+//! parks at character select until the app picks a guid; a new login supersedes either park. The
+//! pick closes the realmd link (`0x46b70c`), sends `CMSG_PLAYER_LOGIN` and announces the connection
+//! without waiting for the verdict, as the reference's world is already loading when it sends the
+//! login (`0x46c272`), so the destination's tiles stream a round trip early. A logout or a later
+//! `SMSG_CHARACTER_LOGIN_FAILED` ends the world session, announced as a
+//! [`SessionEvent::Disconnected`] with [`SessionEnd::LoggedOut`], and parks at character select
+//! again on the same link. A stream failure emits it with [`SessionEnd::Lost`] and re-parks for
+//! credentials. All policy lives in [`crate::login`] and [`crate::char_select`]; this thread only
+//! sequences and never sleeps.
 //!
-//! One long-lived write thread drains [`ClientCommand`](super::ClientCommand)s; each connection
-//! hands it the fresh [`WorldWriter`] over a swap channel, so there is only ever one writer.
+//! One long-lived write thread drains [`ClientCommand`](super::ClientCommand)s; each world entry
+//! hands it the [`WorldWriter`] over a swap channel, and a logout takes it back, so there is only
+//! ever one writer.
 
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
@@ -23,7 +26,7 @@ use anyhow::{anyhow, Result};
 use benilla_assets::LockRecover;
 use benilla_protocol::{
     messages, AuthReject, CharAction, LoginStage, Poll, RosterCause, SessionEnd, SessionEvent,
-    WardenRequired, WorldSession, WorldWriter,
+    WardenRequired, WorldReader, WorldSession, WorldWriter,
 };
 use crossbeam_channel::{Receiver, Sender};
 
@@ -181,20 +184,28 @@ enum Parked {
     Exit,
 }
 
-/// How one connection cycle ended; a stream failure is the `Err` of `Result`.
+/// How one connection cycle ended; a stream failure is the `Err` of `Result`. A logout and a
+/// refused character login end no cycle: the link goes back to character select.
 enum Cycle {
     /// The app dropped a channel end: end the read thread.
     Exit,
     /// Back to the pre-logon park with nothing to announce: a pre-roster failure (already
     /// reported), a canceled attempt, or select's Back ([`CharRequest::Abandon`]).
     Repark,
-    /// A clean logout: emit the teardown `Disconnected`, then park; the app's pending credentials
-    /// fetch the roster again.
-    LoggedOut,
-    /// `SMSG_CHARACTER_LOGIN_FAILED`: the announced entry is taken back and the cycle ends like a
-    /// logout. Deviation: the reference keeps its connection, but here the session is already
-    /// split and the writer handed away, so the logout's reconnect and relist are reused.
-    LoginRefused,
+}
+
+/// What the read thread hands the write thread on the swap channel, in order.
+pub(super) enum WriterSwap {
+    /// A world entry's writer. `fresh_link` is the connection's first entry, whose ping block
+    /// starts empty; a later entry on the same link keeps it, as the reference's lives on the
+    /// connection (`conn+0x1a6c`).
+    Install {
+        writer: WorldWriter,
+        fresh_link: bool,
+    },
+    /// The world session is over and the link kept: the writer goes back to the read thread on
+    /// this reply, and commands drop until the next install.
+    Reclaim(Sender<WorldWriter>),
 }
 
 /// Everything [`spawn_net`] hands the app.
@@ -224,8 +235,9 @@ pub(super) fn spawn_net(connect: bool) -> NetHandles {
     let login_abandon = Arc::new(AtomicU64::new(0));
     let ping_clock = Arc::new(Mutex::new(PingClock::default()));
     if connect {
-        // The writer thread outlives connections; the read thread hands it each new WorldWriter.
-        let (writer_tx, writer_rx) = crossbeam_channel::unbounded::<WorldWriter>();
+        // The writer thread outlives connections; the read thread hands it each world entry's
+        // WorldWriter and takes it back after a logout.
+        let (writer_tx, writer_rx) = crossbeam_channel::unbounded::<WriterSwap>();
         let clock = Arc::clone(&ping_clock);
         // `SMSG_PONG` is timed on the read thread, where it lands ([`PingClock`]).
         let read_clock = Arc::clone(&ping_clock);
@@ -267,22 +279,6 @@ pub(super) fn spawn_net(connect: bool) -> NetHandles {
                     ) {
                         Ok(Cycle::Exit) => return,
                         Ok(Cycle::Repark) => {}
-                        Ok(end @ (Cycle::LoggedOut | Cycle::LoginRefused)) => {
-                            // A refused login tears down what the entry had started to build.
-                            let reason = match end {
-                                Cycle::LoggedOut => "logged out",
-                                _ => "character login refused",
-                            };
-                            if events_tx
-                                .send(SessionEvent::Disconnected {
-                                    reason: reason.into(),
-                                    end: SessionEnd::LoggedOut,
-                                })
-                                .is_err()
-                            {
-                                return;
-                            }
-                        }
                         Err(e) => {
                             // A stream failure; a displacement kick arrives as a bare EOF. No
                             // sleep: what follows is the app's policy.
@@ -336,9 +332,10 @@ enum WorldEnd {
     /// The cycle is over.
     Cycle(Cycle),
     /// It failed before the roster, reported, or was cancelled, or Change Realm left the session at
-    /// character select: the logon park resumes with the realmd link up, as the reference's
-    /// state-2 failure arm (`0x46bea7`-`0x46bf96`) and `ConnectToRealm`'s disconnect (`0x46b22b`)
-    /// drop only the world socket.
+    /// character select: the logon park resumes with the realmd link as it stands, as the
+    /// reference's state-2 failure arm (`0x46bea7`-`0x46bf96`) and `ConnectToRealm`'s disconnect
+    /// (`0x46b22b`) drop only the world socket. Closed since a world entry, it is dialled again by
+    /// the next query ([`query`]).
     Failed,
     /// A new login superseded the session at character select.
     Login(LoginRequest),
@@ -416,9 +413,9 @@ fn login_park(parks: &Parks, events_tx: &Sender<SessionEvent>) -> Option<LoginRe
     }
 }
 
-/// The logon park: the realmd link is up and the list published. Waits for the app's realm,
-/// answering list queries in place on the link (`RequestRealmList`, `0x5ab840`); a new login
-/// supersedes it and a close ends it.
+/// The logon park: the realmd link is up, or closed since a world entry, and the list published.
+/// Waits for the app's realm, answering list queries on the link (`RequestRealmList`, `0x5ab840`);
+/// a new login supersedes it and a close ends it.
 fn logon_park(
     logon: &mut benilla_protocol::Logon,
     req: &LoginRequest,
@@ -461,7 +458,7 @@ fn logon_park(
 /// and returns to the logon park; any retry is the app's.
 fn run(
     events_tx: &Sender<SessionEvent>,
-    writer_tx: &Sender<WorldWriter>,
+    writer_tx: &Sender<WriterSwap>,
     parks: &Parks,
     abandon: &AtomicU64,
     ping_clock: &Mutex<PingClock>,
@@ -571,7 +568,7 @@ fn world(
     logon: &mut benilla_protocol::Logon,
     pick: (benilla_protocol::RealmInfo, u64),
     events_tx: &Sender<SessionEvent>,
-    writer_tx: &Sender<WorldWriter>,
+    writer_tx: &Sender<WriterSwap>,
     parks: &Parks,
     abandon: &AtomicU64,
     ping_clock: &Mutex<PingClock>,
@@ -681,7 +678,7 @@ fn world(
         // (`GetCharacterListUpdate`, [`CharRequest::Enumerate`], `0x46c1a7`), as the reference
         // does: the world connect completes with the authentication response (`0x5ab2b8`), and
         // the `CMSG_CHAR_ENUM` is the screen's.
-        let asks = screen_asks_roster.load(Ordering::SeqCst);
+        let mut asks = screen_asks_roster.load(Ordering::SeqCst);
         let mut characters = if asks {
             Vec::new()
         } else {
@@ -702,7 +699,7 @@ fn world(
             return Ok(WorldEnd::Failed);
         }
         // A request or realm answer queued during a dead cycle must not answer this roster: a
-        // logout must land on the list, not bounce back in off a stale pick. The app re-sends
+        // reconnect must land on the list, not bounce back in off a stale pick. The app re-sends
         // its wants.
         while park_rx.try_recv().is_ok() {}
         let epoch = WORLD_EPOCH.fetch_add(1, Ordering::SeqCst) + 1;
@@ -730,90 +727,120 @@ fn world(
             return Ok(WorldEnd::Cycle(Cycle::Exit));
         }
 
-        // Park at character select until the app answers. Create and delete are serviced in place.
-        // If the server kicked the parked socket meanwhile, the login below fails, the cycle
-        // restarts and the app re-sends its pick, so the park needs no keepalive.
-        bevy::log::info!("net: parked at character select");
-        let guid = loop {
-            // The realm list's requests and the character screen's share one channel, in the
-            // order they were made: `RealmList.xml` is a `frameStrata="DIALOG"` frame, so Change
-            // Realm raises it over character select and both must be served while it is open,
-            // and a roster ask made before a realm change goes out on this link before the
-            // change drops it (the reference sends at the verb, `0x46b880`, and `ConnectToRealm`
-            // only then closes the link, `0x46b22b`).
-            //
-            // No `break` or `continue` inside the `select!` arms: the macro expands into a loop of
-            // its own, which they would bind to. The arms answer with a [`Parked`].
-            let answer = crossbeam_channel::select! {
-                recv(park_rx) -> msg => match msg {
-                    Err(_) => Parked::Exit,
-                    // `RequestRealmList`: a query on the realmd link this cycle holds, then the
-                    // list; also Change Realm's first list.
-                    Ok(ParkRequest::Realm(RealmRequest::Refresh)) => {
-                        if query(logon, req, events_tx) {
-                            Parked::StayPut
-                        } else {
-                            Parked::Exit
-                        }
-                    }
-                    // `RealmListDialogCancelled` over character select is a no-op (`0x46b810`).
-                    Ok(ParkRequest::Realm(RealmRequest::Close)) => Parked::StayPut,
-                    // `ChangeRealm(category, index)`: dial the chosen realm with the session key
-                    // we hold.
-                    Ok(ParkRequest::Realm(RealmRequest::Enter { realm, generation })) => {
-                        Parked::Realm(realm, generation)
-                    }
-                    // `ConnectToRealm` with the world up drops it first (`0x46b22b`).
-                    Ok(ParkRequest::Realm(RealmRequest::Leave)) => Parked::Leave,
-                    Ok(ParkRequest::Char(CharRequest::Enter(guid))) => Parked::Play(guid),
-                    // Select's Back: drop the parked session, return to the login park.
-                    Ok(ParkRequest::Char(CharRequest::Abandon)) => Parked::Repark,
-                    Ok(ParkRequest::Char(CharRequest::Create(create))) => {
-                        Parked::Acted(CharAction::Create, session.create_character(&create)?)
-                    }
-                    Ok(ParkRequest::Char(CharRequest::Delete(target))) => {
-                        Parked::Acted(CharAction::Delete, session.delete_character(target)?)
-                    }
-                    // `GetCharacterListUpdate` (`0x4723d0` → `0x46b880` → `0x5aabb0` → `0x5b4690`):
-                    // the reference asks the world server again whenever the glue screen wants
-                    // the list. A Warden refusal is the server's answer, as at the park entry.
-                    Ok(ParkRequest::Char(CharRequest::Enumerate)) => match session.char_enum() {
-                        Ok(c) => {
-                            characters = c;
-                            Parked::Listed
-                        }
-                        Err(e) => match e.downcast_ref::<WardenRequired>() {
-                            Some(w) => Parked::Refused(w.to_string()),
-                            // The stream failed under the ask: the screen's query fails
-                            // (`0x5ab300`), and the session ends as a dropped socket does.
-                            None => {
-                                let _ = events_tx.send(SessionEvent::RosterFailed { epoch });
-                                return Err(e);
+        // The link: the character park, the world, and back to the park on the same link after a
+        // logout or a refused character login (`0x5aaeb0`, `0x5aad70`: `0x401ee0` tears the world
+        // down and closes nothing). Its first world entry starts the connection's ping block.
+        let mut fresh_link = true;
+        loop {
+            // Park at character select until the app answers. Create and delete are serviced in place.
+            // If the server kicked the parked socket meanwhile, the login below fails, the cycle
+            // restarts and the app re-sends its pick, so the park needs no keepalive.
+            bevy::log::info!("net: parked at character select");
+            let guid = loop {
+                // The realm list's requests and the character screen's share one channel, in the
+                // order they were made: `RealmList.xml` is a `frameStrata="DIALOG"` frame, so Change
+                // Realm raises it over character select and both must be served while it is open,
+                // and a roster ask made before a realm change goes out on this link before the
+                // change drops it (the reference sends at the verb, `0x46b880`, and `ConnectToRealm`
+                // only then closes the link, `0x46b22b`).
+                //
+                // No `break` or `continue` inside the `select!` arms: the macro expands into a loop of
+                // its own, which they would bind to. The arms answer with a [`Parked`].
+                let answer = crossbeam_channel::select! {
+                    recv(park_rx) -> msg => match msg {
+                        Err(_) => Parked::Exit,
+                        // `RequestRealmList`: a query on the realmd link, dialled again once Enter
+                        // World has closed it (`0x5b2200`), then the list; also Change Realm's
+                        // first list.
+                        Ok(ParkRequest::Realm(RealmRequest::Refresh)) => {
+                            if query(logon, req, events_tx) {
+                                Parked::StayPut
+                            } else {
+                                Parked::Exit
                             }
+                        }
+                        // `RealmListDialogCancelled` over character select is a no-op (`0x46b810`).
+                        Ok(ParkRequest::Realm(RealmRequest::Close)) => Parked::StayPut,
+                        // `ChangeRealm(category, index)`: dial the chosen realm with the session key
+                        // we hold.
+                        Ok(ParkRequest::Realm(RealmRequest::Enter { realm, generation })) => {
+                            Parked::Realm(realm, generation)
+                        }
+                        // `ConnectToRealm` with the world up drops it first (`0x46b22b`).
+                        Ok(ParkRequest::Realm(RealmRequest::Leave)) => Parked::Leave,
+                        Ok(ParkRequest::Char(CharRequest::Enter(guid))) => Parked::Play(guid),
+                        // Select's Back: drop the parked session, return to the login park.
+                        Ok(ParkRequest::Char(CharRequest::Abandon)) => Parked::Repark,
+                        Ok(ParkRequest::Char(CharRequest::Create(create))) => {
+                            Parked::Acted(CharAction::Create, session.create_character(&create)?)
+                        }
+                        Ok(ParkRequest::Char(CharRequest::Delete(target))) => {
+                            Parked::Acted(CharAction::Delete, session.delete_character(target)?)
+                        }
+                        // `GetCharacterListUpdate` (`0x4723d0` → `0x46b880` → `0x5aabb0` → `0x5b4690`):
+                        // the reference asks the world server again whenever the glue screen wants
+                        // the list. A Warden refusal is the server's answer, as at the park entry.
+                        Ok(ParkRequest::Char(CharRequest::Enumerate)) => match session.char_enum() {
+                            Ok(c) => {
+                                characters = c;
+                                Parked::Listed
+                            }
+                            Err(e) => match e.downcast_ref::<WardenRequired>() {
+                                Some(w) => Parked::Refused(w.to_string()),
+                                // The stream failed under the ask: the screen's query fails
+                                // (`0x5ab300`), and the session ends as a dropped socket does.
+                                None => {
+                                    let _ = events_tx.send(SessionEvent::RosterFailed { epoch });
+                                    return Err(e);
+                                }
+                            },
                         },
                     },
-                },
-                recv(login_rx) -> msg => msg.map_or(Parked::Exit, Parked::Login),
-            };
-            let (action, code) = match answer {
-                Parked::Exit => return Ok(WorldEnd::Cycle(Cycle::Exit)),
-                Parked::Repark => return Ok(WorldEnd::Cycle(Cycle::Repark)),
-                Parked::Login(next) => return Ok(WorldEnd::Login(next)),
-                Parked::StayPut => continue,
-                Parked::Play(guid) => break guid,
-                Parked::Realm(chosen, asked) => {
-                    (realm, generation) = (chosen, asked);
-                    continue 'realm;
-                }
-                Parked::Leave => return Ok(WorldEnd::Failed),
-                Parked::Refused(reason) => return fail_terminal(reason),
-                Parked::Acted(action, code) => (action, code),
-                Parked::Listed => {
+                    recv(login_rx) -> msg => msg.map_or(Parked::Exit, Parked::Login),
+                };
+                let (action, code) = match answer {
+                    Parked::Exit => return Ok(WorldEnd::Cycle(Cycle::Exit)),
+                    Parked::Repark => return Ok(WorldEnd::Cycle(Cycle::Repark)),
+                    Parked::Login(next) => return Ok(WorldEnd::Login(next)),
+                    Parked::StayPut => continue,
+                    Parked::Play(guid) => break guid,
+                    Parked::Realm(chosen, asked) => {
+                        (realm, generation) = (chosen, asked);
+                        continue 'realm;
+                    }
+                    Parked::Leave => return Ok(WorldEnd::Failed),
+                    Parked::Refused(reason) => return fail_terminal(reason),
+                    Parked::Acted(action, code) => (action, code),
+                    Parked::Listed => {
+                        if events_tx
+                            .send(SessionEvent::CharacterList {
+                                characters: characters.clone(),
+                                realm: Some(realm.clone()),
+                                cause: RosterCause::Asked,
+                                epoch,
+                                generation,
+                            })
+                            .is_err()
+                        {
+                            return Ok(WorldEnd::Cycle(Cycle::Exit));
+                        }
+                        continue;
+                    }
+                };
+                // A success changed the roster: under the native policy re-emit it before the result,
+                // so the screen has the fresh list when it reacts. The stock screen asks again itself
+                // (`0x46c1a7`).
+                let changed = match action {
+                    CharAction::Create => code == messages::CHAR_CREATE_SUCCESS,
+                    CharAction::Delete => code == messages::CHAR_DELETE_SUCCESS,
+                };
+                if changed && !asks {
+                    characters = session.char_enum()?;
                     if events_tx
                         .send(SessionEvent::CharacterList {
                             characters: characters.clone(),
                             realm: Some(realm.clone()),
-                            cause: RosterCause::Asked,
+                            cause: RosterCause::Changed,
                             epoch,
                             generation,
                         })
@@ -821,23 +848,110 @@ fn world(
                     {
                         return Ok(WorldEnd::Cycle(Cycle::Exit));
                     }
-                    continue;
+                }
+                if events_tx
+                    .send(SessionEvent::CharActionResult { action, code })
+                    .is_err()
+                {
+                    return Ok(WorldEnd::Cycle(Cycle::Exit));
                 }
             };
-            // A success changed the roster: under the native policy re-emit it before the result,
-            // so the screen has the fresh list when it reacts. The stock screen asks again itself
-            // (`0x46c1a7`).
-            let changed = match action {
-                CharAction::Create => code == messages::CHAR_CREATE_SUCCESS,
-                CharAction::Delete => code == messages::CHAR_DELETE_SUCCESS,
+            let name = characters
+                .iter()
+                .find(|c| c.guid == guid)
+                .map(|c| c.name.clone())
+                .unwrap_or_default();
+            // `EnterWorld` closes the realmd link (`CLoginMgr::Cancel`, `0x46b70c` → `0x5b3320`)
+            // and nothing in the world reopens it; back at the park, a realm list query dials it
+            // again ([`query`]).
+            logon.close();
+            // No `CMSG_SET_ACTIVE_MOVER` here: it waits for our own player's create, as the
+            // reference's does (`super::enter_world_on_self_create`); a server may drop it before then.
+            session.player_login(refuse_once(guid))?;
+
+            let billing_time_rested = session.billing_time_rested();
+            let tutorial_flags = session.take_tutorial_flags();
+            // `SMSG_ADDON_INFO` carries no names, so it is paired back against the block we sent. No
+            // reply means no AddOn index space at all, carried as `None`. The link's one reply
+            // answers every login on it.
+            let statuses = session.addon_info().map(<[u8]>::to_vec);
+            let addon_info = statuses.as_deref().map(|statuses| {
+                benilla_protocol::messages::hidden_from_reply(
+                    statuses,
+                    &benilla_protocol::messages::STOCK_SECURE_ADDONS,
+                )
+            });
+            match (&statuses, &addon_info) {
+                (Some(statuses), Some(hidden)) => bevy::log::info!(
+                    "net: SMSG_ADDON_INFO — {} record(s), {} hidden from the AddOn index space",
+                    statuses.len(),
+                    hidden.len()
+                ),
+                _ => bevy::log::warn!(
+                    "net: no SMSG_ADDON_INFO — GetNumAddOns() stays 0, as the reference's would"
+                ),
+            }
+            let (mut reader, writer) = session.into_split()?;
+            // Installed before the entry is announced, so the app's first command finds it.
+            let install = WriterSwap::Install {
+                writer,
+                fresh_link: std::mem::take(&mut fresh_link),
             };
-            if changed && !asks {
-                characters = session.char_enum()?;
+            if writer_tx.send(install).is_err() {
+                // The writer thread ends only on app exit.
+                return Ok(WorldEnd::Cycle(Cycle::Exit));
+            }
+            if events_tx
+                .send(SessionEvent::Connected {
+                    self_guid: guid,
+                    name,
+                    billing_time_rested,
+                    tutorial_flags,
+                    addon_info,
+                })
+                .is_err()
+            {
+                return Ok(WorldEnd::Cycle(Cycle::Exit));
+            }
+            bevy::log::info!("net: connected to {world_addr}");
+
+            let reason = match stream(&mut reader, park_rx, events_tx, ping_clock, tails_announced)?
+            {
+                StreamEnd::Exit => return Ok(WorldEnd::Cycle(Cycle::Exit)),
+                StreamEnd::Back(reason) => reason,
+            };
+            // The world is torn down (`0x401ee0`'s shutdown) and the link kept: the app's teardown
+            // edge, then the session joined again at the character park.
+            if events_tx
+                .send(SessionEvent::Disconnected {
+                    reason: reason.into(),
+                    end: SessionEnd::LoggedOut,
+                })
+                .is_err()
+            {
+                return Ok(WorldEnd::Cycle(Cycle::Exit));
+            }
+            let Some(writer) = reclaim(writer_tx)? else {
+                return Ok(WorldEnd::Cycle(Cycle::Exit));
+            };
+            session = WorldSession::rejoin(reader, writer)?;
+            bevy::log::info!("net: {reason} — back at character select on the same link");
+            // The native screens wait for the park's own enumeration, as on the dial; the stock
+            // screen's `CharacterSelect_OnShow` asks for its roster itself (`0x46b880`).
+            asks = screen_asks_roster.load(Ordering::SeqCst);
+            if !asks {
+                characters = match session.char_enum() {
+                    Ok(c) => c,
+                    Err(e) => match e.downcast_ref::<WardenRequired>() {
+                        Some(w) => return fail_terminal(w.to_string()),
+                        None => return Err(e),
+                    },
+                };
                 if events_tx
                     .send(SessionEvent::CharacterList {
                         characters: characters.clone(),
                         realm: Some(realm.clone()),
-                        cause: RosterCause::Changed,
+                        cause: RosterCause::Park,
                         epoch,
                         generation,
                     })
@@ -846,156 +960,143 @@ fn world(
                     return Ok(WorldEnd::Cycle(Cycle::Exit));
                 }
             }
-            if events_tx
-                .send(SessionEvent::CharActionResult { action, code })
-                .is_err()
-            {
-                return Ok(WorldEnd::Cycle(Cycle::Exit));
-            }
-        };
-        let name = characters
-            .iter()
-            .find(|c| c.guid == guid)
-            .map(|c| c.name.clone())
-            .unwrap_or_default();
-        // No `CMSG_SET_ACTIVE_MOVER` here: it waits for our own player's create, as the
-        // reference's does (`super::enter_world_on_self_create`); a server may drop it before then.
-        session.player_login(refuse_once(guid))?;
-
-        let billing_time_rested = session.billing_time_rested();
-        let tutorial_flags = session.take_tutorial_flags();
-        // `SMSG_ADDON_INFO` carries no names, so it is paired back against the block we sent. No
-        // reply means no AddOn index space at all, carried as `None`.
-        let statuses = session.take_addon_info();
-        let addon_info = statuses.as_deref().map(|statuses| {
-            benilla_protocol::messages::hidden_from_reply(
-                statuses,
-                &benilla_protocol::messages::STOCK_SECURE_ADDONS,
-            )
-        });
-        match (&statuses, &addon_info) {
-            (Some(statuses), Some(hidden)) => bevy::log::info!(
-                "net: SMSG_ADDON_INFO — {} record(s), {} hidden from the AddOn index space",
-                statuses.len(),
-                hidden.len()
-            ),
-            _ => bevy::log::warn!(
-                "net: no SMSG_ADDON_INFO — GetNumAddOns() stays 0, as the reference's would"
-            ),
-        }
-        let (mut reader, writer) = session.into_split()?;
-        if events_tx
-            .send(SessionEvent::Connected {
-                self_guid: guid,
-                name,
-                billing_time_rested,
-                tutorial_flags,
-                addon_info,
-            })
-            .is_err()
-        {
-            return Ok(WorldEnd::Cycle(Cycle::Exit));
-        }
-        if writer_tx.send(writer).is_err() {
-            // The writer thread ends only on app exit.
-            return Ok(WorldEnd::Cycle(Cycle::Exit));
-        }
-        bevy::log::info!("net: connected to {world_addr}");
-
-        // `poll` skips packets it cannot parse; a long run of skips means the stream desynced.
-        let (mut skip_run, mut skip_logged) = (0u32, 0u32);
-        loop {
-            let polled = reader.poll()?;
-            note_inbound(); // parsed or not, the census counts liveness
-            match polled {
-                Poll::Events {
-                    opcode,
-                    events,
-                    tail,
-                    body,
-                } => {
-                    skip_run = 0;
-                    // A body is length-framed, so a decoder shorter than the server's layout
-                    // succeeds silently; report it once per opcode, and never skip the packet.
-                    if tail > 0 && tails_announced.insert(opcode) {
-                        bevy::log::info!(
-                            "net: opcode {} ({opcode:#06x}) left {tail} trailing byte(s) after \
-                             decode (first occurrence; announced once per opcode)",
-                            benilla_protocol::messages::opcode_name(opcode).unwrap_or("?"),
-                        );
-                    }
-                    // Every inbound opcode by name, length and bytes (tag `in`), including one that
-                    // parsed into no event.
-                    if benilla_assets::trace::enabled_for("in") {
-                        benilla_assets::trace::line(
-                            "in",
-                            &wire_trace::in_line(opcode, events.len(), &body),
-                        );
-                    }
-                    for ev in events {
-                        // The pong bypass: the reference's `OnData` (`0x537b10`) hands `SMSG_PONG`
-                        // to `HandlePong` (`0x537d60`) inline, not onto the game thread's queue, so
-                        // it is timed and consumed here.
-                        if let SessionEvent::Pong { sequence } = ev {
-                            if let Some(rtt) = ping_clock.lock_recover().record_pong(sequence) {
-                                bevy::log::debug!("net: pong seq={sequence} rtt={rtt}ms");
-                            }
-                            continue;
-                        }
-                        // A logout or a refused login ends the cycle after the app hears it, so
-                        // the screen has the reason before the teardown.
-                        let ends = match ev {
-                            SessionEvent::LoggedOut => Some(Cycle::LoggedOut),
-                            SessionEvent::CharacterLoginFailed { .. } => Some(Cycle::LoginRefused),
-                            _ => None,
-                        };
-                        // Receiver dropped: the app exited.
-                        if events_tx.send(ev).is_err() {
-                            return Ok(WorldEnd::Cycle(Cycle::Exit));
-                        }
-                        if let Some(cycle) = ends {
-                            return Ok(WorldEnd::Cycle(cycle));
-                        }
-                    }
-                }
-                Poll::Skipped {
-                    opcode,
-                    reason,
-                    body,
-                } => {
-                    skip_run += 1;
-                    // Every skip, uncapped, into the trace (tag `skip`) with its whole body:
-                    // otherwise a packet that failed to parse looks like one that never arrived.
-                    if benilla_assets::trace::enabled_for("skip") {
-                        benilla_assets::trace::line(
-                            "skip",
-                            &wire_trace::skip_line(opcode, &reason, &body),
-                        );
-                    }
-                    // Capped, but enough for a post-teleport burst of unparseable object updates.
-                    if skip_logged < 40 {
-                        bevy::log::warn!("net: skipping unparseable packet — {reason}");
-                        skip_logged += 1;
-                    }
-                    // A parse error feeds the dropped-packet tally like an unknown opcode.
-                    if events_tx
-                        .send(SessionEvent::PacketDropped {
-                            opcode,
-                            unparseable: true,
-                        })
-                        .is_err()
-                    {
-                        return Ok(WorldEnd::Cycle(Cycle::Exit));
-                    }
-                    if skip_run > 1024 {
-                        return Err(anyhow!("world stream desynced after 1024 skipped packets"));
-                    }
-                }
-            }
         }
     } // 'realm
 }
 
+/// How the world stream ended, short of a socket failure.
+enum StreamEnd {
+    /// The app dropped a channel end.
+    Exit,
+    /// `SMSG_LOGOUT_COMPLETE` or `SMSG_CHARACTER_LOGIN_FAILED`, already announced: the world session
+    /// is over and the link goes back to character select. The reason names which.
+    Back(&'static str),
+}
+
+/// The world stream: every packet off the socket into [`SessionEvent`]s until a logout or a refused
+/// character login ends the world session, the socket fails or the app exits.
+fn stream(
+    reader: &mut WorldReader,
+    park_rx: &Receiver<ParkRequest>,
+    events_tx: &Sender<SessionEvent>,
+    ping_clock: &Mutex<PingClock>,
+    tails_announced: &mut std::collections::HashSet<u16>,
+) -> Result<StreamEnd> {
+    // `poll` skips packets it cannot parse; a long run of skips means the stream desynced.
+    let (mut skip_run, mut skip_logged) = (0u32, 0u32);
+    loop {
+        let polled = reader.poll()?;
+        note_inbound(); // parsed or not, the census counts liveness
+        match polled {
+            Poll::Events {
+                opcode,
+                events,
+                tail,
+                body,
+            } => {
+                skip_run = 0;
+                // A body is length-framed, so a decoder shorter than the server's layout succeeds
+                // silently; report it once per opcode, and never skip the packet.
+                if tail > 0 && tails_announced.insert(opcode) {
+                    bevy::log::info!(
+                        "net: opcode {} ({opcode:#06x}) left {tail} trailing byte(s) after \
+                         decode (first occurrence; announced once per opcode)",
+                        benilla_protocol::messages::opcode_name(opcode).unwrap_or("?"),
+                    );
+                }
+                // Every inbound opcode by name, length and bytes (tag `in`), including one that
+                // parsed into no event.
+                if benilla_assets::trace::enabled_for("in") {
+                    benilla_assets::trace::line(
+                        "in",
+                        &wire_trace::in_line(opcode, events.len(), &body),
+                    );
+                }
+                for ev in events {
+                    // The pong bypass: the reference's `OnData` (`0x537b10`) hands `SMSG_PONG` to
+                    // `HandlePong` (`0x537d60`) inline, not onto the game thread's queue, so it is
+                    // timed and consumed here.
+                    if let SessionEvent::Pong { sequence } = ev {
+                        if let Some(rtt) = ping_clock.lock_recover().record_pong(sequence) {
+                            bevy::log::debug!("net: pong seq={sequence} rtt={rtt}ms");
+                        }
+                        continue;
+                    }
+                    // A logout or a refused login ends the world session after the app hears it,
+                    // so the screen has the reason before the teardown.
+                    let back = match ev {
+                        SessionEvent::LoggedOut => Some("logged out"),
+                        SessionEvent::CharacterLoginFailed { .. } => {
+                            Some("character login refused")
+                        }
+                        _ => None,
+                    };
+                    // An ask queued in the world must not answer the park this returns to: a
+                    // logout must land on the list, not bounce back in off a stale pick. Dropped
+                    // before the app hears of the return, so every ask it makes after is served.
+                    if back.is_some() {
+                        while park_rx.try_recv().is_ok() {}
+                    }
+                    // Receiver dropped: the app exited.
+                    if events_tx.send(ev).is_err() {
+                        return Ok(StreamEnd::Exit);
+                    }
+                    if let Some(reason) = back {
+                        return Ok(StreamEnd::Back(reason));
+                    }
+                }
+            }
+            Poll::Skipped {
+                opcode,
+                reason,
+                body,
+            } => {
+                skip_run += 1;
+                // Every skip, uncapped, into the trace (tag `skip`) with its whole body: otherwise
+                // a packet that failed to parse looks like one that never arrived.
+                if benilla_assets::trace::enabled_for("skip") {
+                    benilla_assets::trace::line(
+                        "skip",
+                        &wire_trace::skip_line(opcode, &reason, &body),
+                    );
+                }
+                // Capped, but enough for a post-teleport burst of unparseable object updates.
+                if skip_logged < 40 {
+                    bevy::log::warn!("net: skipping unparseable packet — {reason}");
+                    skip_logged += 1;
+                }
+                // A parse error feeds the dropped-packet tally like an unknown opcode.
+                if events_tx
+                    .send(SessionEvent::PacketDropped {
+                        opcode,
+                        unparseable: true,
+                    })
+                    .is_err()
+                {
+                    return Ok(StreamEnd::Exit);
+                }
+                if skip_run > 1024 {
+                    return Err(anyhow!("world stream desynced after 1024 skipped packets"));
+                }
+            }
+        }
+    }
+}
+
+/// Takes the world writer back from the write thread, the world session over and the link kept.
+/// `None` once the write thread is gone (app exit).
+fn reclaim(writer_tx: &Sender<WriterSwap>) -> Result<Option<WorldWriter>> {
+    let (reply, returned) = crossbeam_channel::bounded(1);
+    if writer_tx.send(WriterSwap::Reclaim(reply)).is_err() {
+        return Ok(None);
+    }
+    // The install went ahead of this on the same channel, so the writer is there to return.
+    returned
+        .recv()
+        .map(Some)
+        .map_err(|_| anyhow!("the write thread held no world writer to return"))
+}
+
 /// `WOW_REFUSE_LOGIN=1` makes the server refuse the run's first character login, which cannot be
 /// staged by hand. Only the wire guid changes, to one vmangos's `!packet.guid.IsPlayer()` guard
 /// refuses; the app still sees the real character. Once only, so the run reaches the recovery
@@ -1015,11 +1116,42 @@ fn refuse_once(guid: u64) -> u64 {
     }
 }
 
+/// One [`WriterSwap`] applied to the write thread's writer, its warning count and its keepalive.
+fn apply_swap(
+    swap: WriterSwap,
+    writer: &mut Option<WorldWriter>,
+    warned: &mut u32,
+    ping_tick: &mut Receiver<Instant>,
+    ping_clock: &Mutex<PingClock>,
+) {
+    match swap {
+        WriterSwap::Install {
+            writer: w,
+            fresh_link,
+        } => {
+            *writer = Some(w);
+            *warned = 0;
+            // Sequence 1 is the new socket's first ping, so an old socket's pong cannot match.
+            if fresh_link {
+                ping_clock.lock_recover().clear();
+            }
+            // The first keepalive is a full interval after connect (`0x537bcf`).
+            *ping_tick = crossbeam_channel::after(PING_INTERVAL);
+        }
+        WriterSwap::Reclaim(reply) => {
+            *ping_tick = crossbeam_channel::never();
+            if let Some(w) = writer.take() {
+                let _ = reply.send(w);
+            }
+        }
+    }
+}
+
 /// The write thread: app commands, writer swaps and the [`PING_INTERVAL`] keepalive. With no live
 /// writer, commands drop with a capped warning. Ends when the app drops every command sender.
-fn writer_loop(
+pub(super) fn writer_loop(
     cmd_rx: &Receiver<ClientCommand>,
-    mut writer_rx: Receiver<WorldWriter>,
+    mut writer_rx: Receiver<WriterSwap>,
     ping_clock: &Mutex<PingClock>,
 ) {
     let mut writer: Option<WorldWriter> = None;
@@ -1030,15 +1162,7 @@ fn writer_loop(
     loop {
         crossbeam_channel::select! {
             recv(writer_rx) -> w => match w {
-                Ok(w) => {
-                    writer = Some(w);
-                    warned = 0;
-                    // Sequence 1 is the new socket's first ping, so an old socket's pong cannot
-                    // match.
-                    ping_clock.lock_recover().clear();
-                    // The first keepalive is a full interval after connect (`0x537bcf`).
-                    ping_tick = crossbeam_channel::after(PING_INTERVAL);
-                }
+                Ok(swap) => apply_swap(swap, &mut writer, &mut warned, &mut ping_tick, ping_clock),
                 // The read thread ended (app exit). A disconnected receiver is always ready and
                 // would spin the select, so stop selecting it.
                 Err(_) => writer_rx = crossbeam_channel::never(),
@@ -1071,6 +1195,12 @@ fn writer_loop(
             },
             recv(cmd_rx) -> cmd => {
                 let Ok(cmd) = cmd else { return }; // all app senders dropped → app exit
+                // The read thread installs a writer before it announces the entry, so a command
+                // the app sends on that announcement finds it here even when this wake-up took
+                // the command first, whether no writer is held or a lost link's dead one is.
+                while let Ok(swap) = writer_rx.try_recv() {
+                    apply_swap(swap, &mut writer, &mut warned, &mut ping_tick, ping_clock);
+                }
                 let Some(w) = writer.as_mut() else {
                     // No live writer: the command is dropped, and both lines name it.
                     if benilla_assets::trace::enabled() {

crates/benilla-app/src/net/io/cycle_tests.rs

@@ -7,10 +7,10 @@ use std::sync::Mutex;
 use std::thread;
 
 use benilla_protocol::messages::opcode;
-use benilla_protocol::{RealmInfo, RosterCause, SessionEvent, WorldWriter};
+use benilla_protocol::{RealmInfo, RosterCause, SessionEvent};
 
 use super::fake::{self, FakeRealm, WAIT};
-use super::{run, Cycle, LoginRequest, Parks, PingClock};
+use super::{run, Cycle, LoginRequest, Parks, PingClock, WriterSwap};
 use crate::net::{CharRequest, ParkRequest, RealmRequest};
 
 const REALM: &str = "Fake";
@@ -61,7 +61,7 @@ fn an_empty_account_parks_at_select_without_creating_a_character() {
     let (realmd, world) = fake::servers(|world| vec![FakeRealm::new(REALM, world)]);
 
     let (events_tx, events) = crossbeam_channel::unbounded();
-    let (writer_tx, _writer_rx) = crossbeam_channel::unbounded::<WorldWriter>();
+    let (writer_tx, _writer_rx) = crossbeam_channel::unbounded::<WriterSwap>();
     let (login_tx, login_rx) = crossbeam_channel::unbounded();
     let (park_tx, park_rx) = crossbeam_channel::unbounded();
     let cycle = thread::spawn(move || {
@@ -529,3 +529,341 @@ fn only_the_native_policy_re_enumerates_after_a_create() {
     ));
     assert_eq!(world.enumerations(), 0);
 }
+
+/// Logs in, dials the one realm and picks `GUID` at character select: the session's epoch once
+/// the entry is announced. Under the native policy the park's roster comes first.
+fn enter_world(cycles: &fake::Cycles, realmd: &fake::Realmd, asking: bool) -> u64 {
+    fake::login(cycles, &realmd.addr);
+    let (realms, _) = next_list(&cycles.events);
+    cycles.realm.send(enter(realms[0].clone())).unwrap();
+    let epoch = match next_event(&cycles.events) {
+        SessionEvent::WorldAuthenticated { epoch, .. } => epoch,
+        other => panic!("expected the authentication, got {other:?}"),
+    };
+    if !asking {
+        assert!(matches!(
+            next_event(&cycles.events),
+            SessionEvent::CharacterList {
+                cause: RosterCause::Park,
+                ..
+            }
+        ));
+    }
+    cycles.pick.send(CharRequest::Enter(GUID)).unwrap();
+    match next_event(&cycles.events) {
+        SessionEvent::Connected { self_guid, .. } => assert_eq!(self_guid, GUID),
+        other => panic!("expected the world entry, got {other:?}"),
+    }
+    epoch
+}
+
+/// The character every entry picks.
+const GUID: u64 = 0x2a;
+
+/// The world session's end after `ended`: the event itself, then the teardown edge with the link
+/// kept, a `Disconnected` that ends in a logout.
+fn expect_back(
+    events: &crossbeam_channel::Receiver<SessionEvent>,
+    ended: fn(&SessionEvent) -> bool,
+) {
+    let ev = next_event(events);
+    assert!(ended(&ev), "expected the world session's end, got {ev:?}");
+    match next_event(events) {
+        SessionEvent::Disconnected { end, .. } => {
+            assert_eq!(end, benilla_protocol::SessionEnd::LoggedOut)
+        }
+        other => panic!("expected the teardown, got {other:?}"),
+    }
+}
+
+/// A logout returns to character select on the world link it entered by (`SMSG_LOGOUT_COMPLETE`
+/// → `0x5aaeb0` → `0x401ee0(1, 1, 0)`, then `CharacterSelect_OnShow` → `0x46b880`): one more
+/// `CMSG_CHAR_ENUM` on the same session, with no realmd logon and no second `CMSG_AUTH_SESSION`.
+/// The native screens get the park's enumeration; the stock screen's comes when it asks.
+#[test]
+fn a_logout_returns_to_character_select_on_the_same_link() {
+    for asking in [false, true] {
+        let (realmd, world) = fake::servers(|world| vec![FakeRealm::new(REALM, world)]);
+        let cycles = if asking {
+            fake::spawn_cycles_asking()
+        } else {
+            fake::spawn_cycles()
+        };
+        let epoch = enter_world(&cycles, &realmd, asking);
+        cycles
+            .commands
+            .send(crate::net::ClientCommand::Logout)
+            .unwrap();
+        expect_back(&cycles.events, |ev| matches!(ev, SessionEvent::LoggedOut));
+        if asking {
+            assert!(
+                cycles
+                    .events
+                    .recv_timeout(std::time::Duration::from_millis(400))
+                    .is_err(),
+                "the park sent a roster the stock screen did not ask for"
+            );
+            cycles.pick.send(CharRequest::Enumerate).unwrap();
+        }
+        match next_event(&cycles.events) {
+            SessionEvent::CharacterList {
+                cause, epoch: on, ..
+            } => {
+                let want = if asking {
+                    RosterCause::Asked
+                } else {
+                    RosterCause::Park
+                };
+                assert_eq!(cause, want, "asking: {asking}");
+                assert_eq!(
+                    on, epoch,
+                    "the roster comes on the link the world was entered by"
+                );
+            }
+            other => panic!("expected the roster at select, got {other:?}"),
+        }
+        let before = usize::from(!asking);
+        assert_eq!(
+            world.enumerations_in(0),
+            before + 1,
+            "one CMSG_CHAR_ENUM after the logout, on the first session"
+        );
+        assert_eq!(world.authentications(), 1, "no second CMSG_AUTH_SESSION");
+        assert_eq!(world.open(), 1, "the world link is kept");
+        assert_eq!(realmd.logons(), 1, "no realmd logon");
+    }
+}
+
+/// A refused character login returns the same way (`SMSG_CHARACTER_LOGIN_FAILED` → `0x5aad70` →
+/// `0x401ee0(1, 1, 1)`, then the dialog's `StatusDialogClick` → `0x46b880`): the session is kept,
+/// and the native policy enumerates once on it.
+#[test]
+fn a_refused_login_returns_to_character_select_on_the_same_link() {
+    let (realmd, world) = fake::servers(|world| vec![FakeRealm::new(REALM, world)]);
+    world.refuse_logins();
+    let cycles = fake::spawn_cycles();
+    let epoch = enter_world(&cycles, &realmd, false);
+    expect_back(&cycles.events, |ev| {
+        matches!(ev, SessionEvent::CharacterLoginFailed { result: 1 })
+    });
+    match next_event(&cycles.events) {
+        SessionEvent::CharacterList {
+            cause, epoch: on, ..
+        } => {
+            assert_eq!(cause, RosterCause::Park);
+            assert_eq!(on, epoch);
+        }
+        other => panic!("expected the roster at select, got {other:?}"),
+    }
+    assert_eq!(world.enumerations_in(0), 2);
+    assert_eq!(world.authentications(), 1, "no second CMSG_AUTH_SESSION");
+    assert_eq!(realmd.logons(), 1, "no realmd logon");
+    // Still parked on the link: the next pick goes out on it.
+    cycles.pick.send(CharRequest::Enter(GUID)).unwrap();
+    assert!(matches!(
+        next_event(&cycles.events),
+        SessionEvent::Connected { .. }
+    ));
+    assert_eq!(world.authentications(), 1);
+}
+
+/// `EnterWorld` closes the realmd link (`CLoginMgr::Cancel`, `0x46b70c` → `0x5b3320`); it stays up
+/// at character select until then.
+#[test]
+fn entering_the_world_closes_the_realmd_link() {
+    let (realmd, _world) = fake::servers(|world| vec![FakeRealm::new(REALM, world)]);
+    let cycles = fake::spawn_cycles();
+    fake::login(&cycles, &realmd.addr);
+    let (realms, _) = next_list(&cycles.events);
+    cycles.realm.send(enter(realms[0].clone())).unwrap();
+    assert!(matches!(
+        next_roster(&cycles.events),
+        SessionEvent::CharacterList { .. }
+    ));
+    assert_eq!(
+        realmd.open(),
+        1,
+        "the realmd link is up at character select"
+    );
+    cycles.pick.send(CharRequest::Enter(GUID)).unwrap();
+    assert!(matches!(
+        next_event(&cycles.events),
+        SessionEvent::Connected { .. }
+    ));
+    wait_until(|| realmd.open() == 0, "the entry closes the realmd link");
+}
+
+/// Change Realm after a logout finds the realmd link closed since Enter World and dials it again
+/// (`0x46b8d0` → `0x5ab840` → `0x5b2200`), then dials the realm with the new logon's key: a second
+/// world session.
+#[test]
+fn change_realm_after_a_logout_dials_realmd_again() {
+    let (realmd, world) = fake::servers(|world| vec![FakeRealm::new(REALM, world)]);
+    let cycles = fake::spawn_cycles();
+    let epoch = enter_world(&cycles, &realmd, false);
+    cycles
+        .commands
+        .send(crate::net::ClientCommand::Logout)
+        .unwrap();
+    expect_back(&cycles.events, |ev| matches!(ev, SessionEvent::LoggedOut));
+    assert!(matches!(
+        next_event(&cycles.events),
+        SessionEvent::CharacterList { .. }
+    ));
+    assert_eq!(realmd.logons(), 1);
+    cycles.realm.send(RealmRequest::Refresh).unwrap();
+    let (realms, failed) = next_list(&cycles.events);
+    assert!(!failed, "the re-dial answers");
+    assert_eq!(realmd.logons(), 2, "the query dialled realmd again");
+    cycles.realm.send(enter(realms[0].clone())).unwrap();
+    match next_event(&cycles.events) {
+        SessionEvent::WorldAuthenticated { epoch: next, .. } => assert_ne!(next, epoch),
+        other => panic!("expected the new session's authentication, got {other:?}"),
+    }
+    assert_eq!(
+        world.authentications(),
+        2,
+        "the change dials a new world session"
+    );
+}
+
+/// Back from the world the link still carries what the world session left in flight: a pong for
+/// a ping sent before the writer came back, a reply vmangos serves an authenticated session. A
+/// packet of a known opcode whose body does not parse is skipped at the park, as the world stream
+/// skips it, and the roster behind it is read; the link is not lost over it.
+#[test]
+fn a_packet_left_in_flight_by_the_world_does_not_cost_the_link() {
+    let (realmd, world) = fake::servers(|world| vec![FakeRealm::new(REALM, world)]);
+    world.trail_logouts_with(vec![
+        // An `SMSG_PONG` with no sequence: a known opcode that does not parse.
+        (opcode::SMSG_PONG, Vec::new()),
+        (opcode::SMSG_PONG, 7u32.to_le_bytes().to_vec()),
+    ]);
+    let cycles = fake::spawn_cycles();
+    let epoch = enter_world(&cycles, &realmd, false);
+    cycles
+        .commands
+        .send(crate::net::ClientCommand::Logout)
+        .unwrap();
+    expect_back(&cycles.events, |ev| matches!(ev, SessionEvent::LoggedOut));
+    match next_event(&cycles.events) {
+        SessionEvent::CharacterList {
+            cause, epoch: on, ..
+        } => {
+            assert_eq!(cause, RosterCause::Park);
+            assert_eq!(on, epoch);
+        }
+        other => panic!("expected the roster past the leftovers, got {other:?}"),
+    }
+    assert_eq!(world.open(), 1, "the world link is kept");
+}
+
+/// The write thread holds a lost link's dead writer until the next entry installs a new one, and
+/// a command can be ready in the same wake-up as that install. Every pending swap is applied
+/// before a command is judged, so the command goes out on the new link. The write thread is held
+/// inside the first install (on the ping clock's lock) while the second install and the command
+/// queue, so both are ready together; without the drain the pick between them is a coin toss,
+/// and twelve rounds catch it.
+#[test]
+fn a_command_ready_with_an_install_goes_out_on_the_new_writer() {
+    use super::{writer_loop, WriterSwap};
+    use crate::net::ClientCommand;
+    const KEY: [u8; benilla_srp::SESSION_KEY_LENGTH] = [9; benilla_srp::SESSION_KEY_LENGTH];
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+    let (keys_tx, keys_rx) = crossbeam_channel::unbounded();
+    keys_tx.send(KEY).unwrap();
+    let world = fake::world(listener, keys_rx);
+    let writer = || {
+        benilla_protocol::WorldSession::connect(&addr, fake::USER, KEY)
+            .unwrap()
+            .into_split()
+            .unwrap()
+    };
+
+    let (cmd_tx, cmd_rx) = crossbeam_channel::unbounded::<ClientCommand>();
+    let (swap_tx, swap_rx) = crossbeam_channel::unbounded::<WriterSwap>();
+    let clock = std::sync::Arc::new(Mutex::new(PingClock::default()));
+    let thread_clock = std::sync::Arc::clone(&clock);
+    thread::spawn(move || writer_loop(&cmd_rx, swap_rx, &thread_clock));
+
+    let played_in = |session: usize| {
+        world
+            .session_packets
+            .lock()
+            .unwrap()
+            .get(session)
+            .is_some_and(|packets| {
+                packets
+                    .iter()
+                    .any(|p| p[2..4] == opcode::CMSG_PLAYED_TIME.to_le_bytes())
+            })
+    };
+    let mut readers = Vec::new();
+    for round in 0..12 {
+        let (old_reader, old) = writer();
+        let (new_reader, new) = writer();
+        readers.extend([old_reader, new_reader]);
+        let held = clock.lock().unwrap();
+        swap_tx
+            .send(WriterSwap::Install {
+                writer: old,
+                fresh_link: true,
+            })
+            .unwrap();
+        // The write thread has taken the old writer and waits on the clock inside the install.
+        thread::sleep(std::time::Duration::from_millis(5));
+        swap_tx
+            .send(WriterSwap::Install {
+                writer: new,
+                fresh_link: true,
+            })
+            .unwrap();
+        cmd_tx.send(ClientCommand::PlayedTime).unwrap();
+        drop(held);
+        let (old_index, new_index) = (2 * round, 2 * round + 1);
+        wait_until(
+            || played_in(old_index) || played_in(new_index),
+            "the command went out",
+        );
+        assert!(
+            played_in(new_index) && !played_in(old_index),
+            "round {round}: the command went out on the writer being replaced"
+        );
+    }
+}
+
+/// Under the stock screens a refused login returns the same way, and the roster comes only when
+/// the screen asks (in the reference from the refusal dialog's `StatusDialogClick`, `0x46b767` →
+/// `0x46b880`), on the same link.
+#[test]
+fn a_refused_login_under_the_stock_screens_waits_for_the_screens_ask() {
+    let (realmd, world) = fake::servers(|world| vec![FakeRealm::new(REALM, world)]);
+    world.refuse_logins();
+    let cycles = fake::spawn_cycles_asking();
+    let epoch = enter_world(&cycles, &realmd, true);
+    expect_back(&cycles.events, |ev| {
+        matches!(ev, SessionEvent::CharacterLoginFailed { result: 1 })
+    });
+    assert!(
+        cycles
+            .events
+            .recv_timeout(std::time::Duration::from_millis(400))
+            .is_err(),
+        "the park sent a roster the stock screen did not ask for"
+    );
+    assert_eq!(world.enumerations_in(0), 0);
+    cycles.pick.send(CharRequest::Enumerate).unwrap();
+    match next_event(&cycles.events) {
+        SessionEvent::CharacterList {
+            cause, epoch: on, ..
+        } => {
+            assert_eq!(cause, RosterCause::Asked);
+            assert_eq!(on, epoch);
+        }
+        other => panic!("expected the asked-for roster, got {other:?}"),
+    }
+    assert_eq!(world.enumerations_in(0), 1);
+    assert_eq!(world.authentications(), 1, "no second CMSG_AUTH_SESSION");
+    assert_eq!(realmd.logons(), 1, "no realmd logon");
+}

crates/benilla-app/src/net/io/fake.rs

@@ -10,7 +10,7 @@ use std::thread;
 use std::time::Duration;
 
 use benilla_protocol::messages::{self, opcode};
-use benilla_protocol::{SessionEnd, SessionEvent, WorldWriter};
+use benilla_protocol::{SessionEnd, SessionEvent};
 use benilla_srp::vanilla_header::HeaderCrypto;
 use benilla_srp::{
     password_verifier, NormalizedString, GENERATOR, LARGE_SAFE_PRIME_LITTLE_ENDIAN,
@@ -20,8 +20,8 @@ use crossbeam_channel::{Receiver, Sender};
 use num_bigint::BigUint;
 use sha1::{Digest, Sha1};
 
-use super::{run, Cycle, LoginRequest, Parks, PingClock};
-use crate::net::{CharRequest, ParkRequest, ParkSender, RealmRequest};
+use super::{run, writer_loop, Cycle, LoginRequest, Parks, PingClock, WriterSwap};
+use crate::net::{CharRequest, ClientCommand, ParkRequest, ParkSender, RealmRequest};
 
 pub(crate) const USER: &str = "empty";
 pub(crate) const PASS: &str = "secret";
@@ -318,6 +318,9 @@ pub(crate) fn char_enum_body(characters: &[benilla_protocol::Character]) -> Vec<
     b
 }
 
+/// Packets, by opcode and body, the fake world sends behind `SMSG_LOGOUT_COMPLETE`.
+type Trail = Arc<Mutex<Vec<(u16, Vec<u8>)>>>;
+
 /// The fake world server's record: every opcode the client sent, and the sessions still open.
 pub(crate) struct World {
     pub(crate) sent: Arc<Mutex<Vec<u16>>>,
@@ -334,6 +337,10 @@ pub(crate) struct World {
     warden: Arc<AtomicBool>,
     /// Close the socket on `CMSG_CHAR_ENUM` instead of answering.
     cut: Arc<AtomicBool>,
+    /// Answer `CMSG_PLAYER_LOGIN` with `SMSG_CHARACTER_LOGIN_FAILED`.
+    refuse: Arc<AtomicBool>,
+    /// Packets sent right behind `SMSG_LOGOUT_COMPLETE`, as the world session leaves them in flight.
+    trail: Trail,
     open: Arc<AtomicUsize>,
 }
 
@@ -358,6 +365,22 @@ impl World {
         self.cut.store(true, Ordering::SeqCst);
     }
 
+    /// Have the server refuse every character login from now on (`SMSG_CHARACTER_LOGIN_FAILED`,
+    /// vmangos's result 1).
+    pub(crate) fn refuse_logins(&self) {
+        self.refuse.store(true, Ordering::SeqCst);
+    }
+
+    /// Send these packets right behind every `SMSG_LOGOUT_COMPLETE`.
+    pub(crate) fn trail_logouts_with(&self, packets: Vec<(u16, Vec<u8>)>) {
+        *self.trail.lock().unwrap() = packets;
+    }
+
+    /// The `CMSG_AUTH_SESSION`s the server admitted: one per world session.
+    pub(crate) fn authentications(&self) -> usize {
+        self.session_packets.lock().unwrap().len()
+    }
+
     /// Hold every roster answer back this long.
     pub(crate) fn delay_rosters(&self, by: Duration) {
         *self.enum_delay.lock().unwrap() = by;
@@ -394,8 +417,10 @@ impl World {
 }
 
 /// A world server whose account has no characters: it admits each session under the latest key
-/// realmd made, answers every `CMSG_CHAR_ENUM` with its roster (empty until [`World::set_roster`])
-/// and a `CMSG_CHAR_CREATE` with success, and logs every opcode the client sends.
+/// realmd made, answers every `CMSG_CHAR_ENUM` with its roster (empty until [`World::set_roster`]),
+/// a `CMSG_CHAR_CREATE` with success, a `CMSG_LOGOUT_REQUEST` with `SMSG_LOGOUT_COMPLETE` and a
+/// `CMSG_PLAYER_LOGIN` with nothing (or a refusal, [`World::refuse_logins`]), and logs every opcode
+/// the client sends.
 pub(crate) fn world(listener: TcpListener, keys: Receiver<Key>) -> World {
     let sent = Arc::new(Mutex::new(Vec::new()));
     let log = Arc::clone(&sent);
@@ -411,6 +436,10 @@ pub(crate) fn world(listener: TcpListener, keys: Receiver<Key>) -> World {
     let warden_flag = Arc::clone(&warden);
     let cut = Arc::new(AtomicBool::new(false));
     let cut_flag = Arc::clone(&cut);
+    let refuse = Arc::new(AtomicBool::new(false));
+    let refuse_flag = Arc::clone(&refuse);
+    let trail: Trail = Arc::default();
+    let trailing = Arc::clone(&trail);
     let open = Arc::new(AtomicUsize::new(0));
     let sessions = Arc::clone(&open);
     thread::spawn(move || {
@@ -451,6 +480,8 @@ pub(crate) fn world(listener: TcpListener, keys: Receiver<Key>) -> World {
             let enum_wait = Arc::clone(&enum_wait);
             let warden_flag = Arc::clone(&warden_flag);
             let cut_flag = Arc::clone(&cut_flag);
+            let refuse_flag = Arc::clone(&refuse_flag);
+            let trailing = Arc::clone(&trailing);
             let index = {
                 let mut all = session_log.lock().unwrap();
                 all.push(Vec::new());
@@ -487,6 +518,18 @@ pub(crate) fn world(listener: TcpListener, keys: Receiver<Key>) -> World {
                             opcode::SMSG_CHAR_CREATE,
                             &[messages::CHAR_CREATE_SUCCESS],
                         ),
+                        opcode::CMSG_PLAYER_LOGIN if refuse_flag.load(Ordering::SeqCst) => send(
+                            &mut s,
+                            Some(&mut crypto),
+                            opcode::SMSG_CHARACTER_LOGIN_FAILED,
+                            &[1],
+                        ),
+                        opcode::CMSG_LOGOUT_REQUEST => {
+                            send(&mut s, Some(&mut crypto), opcode::SMSG_LOGOUT_COMPLETE, &[]);
+                            for (op, body) in trailing.lock().unwrap().clone() {
+                                send(&mut s, Some(&mut crypto), op, &body);
+                            }
+                        }
                         _ => {}
                     }
                 }
@@ -502,6 +545,8 @@ pub(crate) fn world(listener: TcpListener, keys: Receiver<Key>) -> World {
         enum_delay,
         warden,
         cut,
+        refuse,
+        trail,
         open,
     }
 }
@@ -525,14 +570,16 @@ pub(crate) fn dead_address() -> String {
 /// The app's ends of the IO thread's channels.
 pub(crate) struct Cycles {
     pub(crate) events: Receiver<SessionEvent>,
+    /// The app's commands, to the write thread and on to the world socket while in the world.
+    pub(crate) commands: Sender<ClientCommand>,
     pub(crate) login: Sender<LoginRequest>,
     pub(crate) realm: ParkSender<RealmRequest>,
     pub(crate) pick: ParkSender<CharRequest>,
     pub(crate) abandon: Arc<AtomicU64>,
 }
 
-/// The read thread's cycle loop, as [`super::spawn_net`] runs it, with the world writer dropped:
-/// the native screens' policy, the IO thread enumerating on reaching the character park.
+/// The read thread's cycle loop and the write thread, as [`super::spawn_net`] runs them: the native
+/// screens' policy, the IO thread enumerating on reaching the character park.
 pub(crate) fn spawn_cycles() -> Cycles {
     spawn_cycles_for(false)
 }
@@ -544,12 +591,14 @@ pub(crate) fn spawn_cycles_asking() -> Cycles {
 
 fn spawn_cycles_for(screen_asks_roster: bool) -> Cycles {
     let (events_tx, events) = crossbeam_channel::unbounded();
+    let (commands, cmd_rx) = crossbeam_channel::unbounded::<ClientCommand>();
+    let (writer_tx, writer_rx) = crossbeam_channel::unbounded::<WriterSwap>();
+    thread::spawn(move || writer_loop(&cmd_rx, writer_rx, &Mutex::new(PingClock::default())));
     let (login, login_rx) = crossbeam_channel::unbounded();
     let (park_tx, park_rx) = crossbeam_channel::unbounded::<ParkRequest>();
     let abandon = Arc::new(AtomicU64::new(0));
     let thread_abandon = Arc::clone(&abandon);
     thread::spawn(move || {
-        let (writer_tx, _writer_rx) = crossbeam_channel::unbounded::<WorldWriter>();
         let parks = Parks {
             login_rx,
             park_rx,
@@ -568,12 +617,6 @@ fn spawn_cycles_for(screen_asks_roster: bool) -> Cycles {
             ) {
                 Ok(Cycle::Exit) => return,
                 Ok(Cycle::Repark) => {}
-                Ok(Cycle::LoggedOut | Cycle::LoginRefused) => {
-                    let _ = events_tx.send(SessionEvent::Disconnected {
-                        reason: "logged out".into(),
-                        end: SessionEnd::LoggedOut,
-                    });
-                }
                 Err(e) => {
                     if events_tx
                         .send(SessionEvent::Disconnected {
@@ -590,6 +633,7 @@ fn spawn_cycles_for(screen_asks_roster: bool) -> Cycles {
     });
     Cycles {
         events,
+        commands,
         login,
         realm: ParkSender::park(park_tx.clone()),
         pick: ParkSender::park(park_tx),

crates/benilla-app/src/net/session.rs

@@ -464,8 +464,8 @@ fn character_list(
         "net: character select — {} character(s) on the account",
         characters.len()
     );
-    // A roster means a live link: clear the failure the logout's synthesized `Disconnected`
-    // left, or the select screen keeps its "Server down" note until the next world entry.
+    // A roster means a live link: clear the reason a lost link left, or the select screen keeps
+    // its "Server down" note until the next world entry.
     status.last_reason = None;
     char_lists.write(CharListMessage {
         characters,
@@ -542,7 +542,8 @@ fn logged_out(
     logged_out.write(LoggedOutMessage);
 }
 
-/// The session ended: tears down the streamed world and clears every session-scoped cache.
+/// The world session ended, the link kept or not: tears down the streamed world and clears every
+/// session-scoped cache.
 fn disconnected(
     reason: String,
     end: benilla_protocol::SessionEnd,
@@ -564,12 +565,14 @@ fn disconnected(
     // An unfinished far teleport dies with the socket.
     pending_transfer.0 = None;
     status.connected = false;
+    // A logout or a refused login ends the character, the link kept: nothing reconnects it.
+    let character_over = over || end == benilla_protocol::SessionEnd::LoggedOut;
     // The RTT ring is not cleared here: the read thread wipes it as it re-enters its cycle loop.
     // Every streamed entity despawns at once, not faded, so none races the reconnect's creates.
-    // The self avatar stays as the local puppet for the reconnect to refresh, unless the session
-    // is over, and then it goes as on logout.
-    let keep = if over { None } else { self_guid.0 };
-    if over {
+    // The self avatar stays as the local puppet for the reconnect to refresh, unless the
+    // character is over, and then it goes as on logout.
+    let keep = if character_over { None } else { self_guid.0 };
+    if character_over {
         self_guid.0 = None;
     }
     index.0.retain(|guid, e| {
@@ -579,7 +582,10 @@ fn disconnected(
         commands.entity(*e).despawn();
         false
     });
-    status.last_reason = Some(reason);
+    // A failure to show at character select; a kept link has none.
+    if end != benilla_protocol::SessionEnd::LoggedOut {
+        status.last_reason = Some(reason);
+    }
     // In-flight name queries died with the socket.
     names.clear_pending();
     items.clear_session();
@@ -799,3 +805,83 @@ fn packet_dropped(opcode: u16, unparseable: bool, dropped: &mut DroppedOpcodes)
 
 #[cfg(test)]
 mod reputation_tests;
+
+#[cfg(test)]
+mod teardown_tests {
+    use super::*;
+    use bevy::ecs::system::RunSystemOnce;
+
+    /// One teardown of `end` over a session in the world as guid 7, whose avatar is spawned:
+    /// whether the self guid survived and the reason the select screen would show.
+    fn teardown(end: benilla_protocol::SessionEnd) -> (Option<u64>, Option<String>, bool) {
+        let mut app = App::new();
+        app.init_resource::<GuidIndex>()
+            .insert_resource(SelfGuid(Some(7)))
+            .init_resource::<NetStatus>()
+            .init_resource::<NameCache>()
+            .init_resource::<Items>()
+            .init_resource::<crate::spell::Cooldowns>()
+            .init_resource::<PendingTransfer>()
+            .add_message::<DisconnectedMessage>();
+        let avatar = app.world_mut().spawn_empty().id();
+        app.world_mut()
+            .resource_mut::<GuidIndex>()
+            .0
+            .insert(7, avatar);
+        app.world_mut()
+            .run_system_once(
+                move |mut commands: Commands,
+                      mut index: ResMut<GuidIndex>,
+                      mut self_guid: ResMut<SelfGuid>,
+                      mut status: ResMut<NetStatus>,
+                      mut names: ResMut<NameCache>,
+                      mut items: ResMut<Items>,
+                      mut cooldowns: ResMut<crate::spell::Cooldowns>,
+                      mut transfer: ResMut<PendingTransfer>,
+                      mut out: MessageWriter<DisconnectedMessage>| {
+                    disconnected(
+                        "character login refused".into(),
+                        end,
+                        &mut commands,
+                        &mut index,
+                        &mut self_guid,
+                        &mut status,
+                        &mut names,
+                        &mut items,
+                        &mut cooldowns,
+                        &mut transfer,
+                        &mut out,
+                    );
+                },
+            )
+            .unwrap();
+        (
+            app.world().resource::<SelfGuid>().0,
+            app.world().resource::<NetStatus>().last_reason.clone(),
+            app.world().get_entity(avatar).is_ok(),
+        )
+    }
+
+    /// A refused login (or a logout) ends the character with the link kept: the self guid goes as
+    /// on a logout, and no failure is left for character select to show as "Server down".
+    #[test]
+    fn a_teardown_on_a_kept_link_ends_the_character_and_reports_no_failure() {
+        let (guid, reason, avatar) = teardown(benilla_protocol::SessionEnd::LoggedOut);
+        assert_eq!(guid, None, "the refused character's guid outlived it");
+        assert_eq!(reason, None, "a kept link is no failure");
+        assert!(!avatar);
+    }
+
+    /// An unattended run's lost link keeps the avatar for the reconnect, and its reason.
+    #[test]
+    fn a_reconnectable_loss_keeps_the_avatar_and_its_reason() {
+        let _lock = crate::local_state::test_env::ENV_LOCK
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let _decl = crate::local_state::test_env::EnvGuard::set("WOW_UNATTENDED", "1");
+        let (guid, reason, avatar) = teardown(benilla_protocol::SessionEnd::Lost);
+        assert_eq!(guid, Some(7));
+        assert!(reason.is_some());
+        assert!(avatar);
+    }
+}

crates/benilla-app/src/realm_select/mod.rs

@@ -9,14 +9,14 @@
 //! character park, and takes one answer. The policy, in order:
 //!
 //! 1. `WOW_REALM`, matched case-insensitively; a name not on the list falls through.
-//! 2. The realm this session is on, so a logout or a dropped connection returns to it.
+//! 2. The realm this session is on, so an unattended reconnect returns to it.
 //! 3. The remembered realm, the persisted `realmName` CVar.
 //! 4. An unattended run (`crate::run_mode::unattended`) takes the first realm that is up.
 //! 5. Otherwise the dialog.
 //!
 //! Change Realm costs a world dial, not a login. `RequestRealmList` (`0x46ecf0` -> `0x46b8d0`)
-//! begins a realm query on the realmd socket, which stays open until world entry (`0x46b70c`), and
-//! names no screen; only `ConnectToRealm` (`0x46b210`) drops the world connection, once a realm is
+//! begins a realm query on the realmd socket, which stays open until world entry (`0x46b70c`) and
+//! is dialled again for a query after it (`0x5b2200`), and names no screen; only `ConnectToRealm` (`0x46b210`) drops the world connection, once a realm is
 //! chosen. So the character park serves the list in place and Cancel there does nothing.
 //!
 //! Under `WOW_GLUE_STOCK=1` the stock `RealmList` and its verbs take the list instead

crates/benilla-app/src/ui_script/glue_mgr.rs

@@ -857,9 +857,11 @@ impl GlueMgr {
         self.op.complete(true, STATUS_AUTH_OK);
     }
 
-    /// The session ended. Over (`session_over`), the disconnect handler (`0x46c540`): state 0,
-    /// and `DISCONNECTED_FROM_SERVER` (`0x46c62e`), which the stock screens answer with the login
-    /// screen and the `DISCONNECTED` dialog. A logout's relist or an unattended reconnect is no
+    /// The session ended. A logout or a refused character login ends only the world: the link
+    /// is kept (`0x5aaeb0` and `0x5aad70` reach no close), so the screen's roster ask goes out on
+    /// it (`IsConnectedToServer`, `0x46d380`). Over (`session_over`), the disconnect handler
+    /// (`0x46c540`): state 0, and `DISCONNECTED_FROM_SERVER` (`0x46c62e`), which the stock screens
+    /// answer with the login screen and the `DISCONNECTED` dialog. An unattended reconnect is no
     /// disconnect to the screens.
     ///
     /// Out of state 2 it fires the event (`0x46c618`); in state 2 a completed, failed operation
@@ -867,7 +869,10 @@ impl GlueMgr {
     /// the dropped connection is `ClientServices`' current one (`0x5abd40`, else "Invalid
     /// disconnect"), has no native counterpart: the IO thread holds one session and reports only
     /// its own end.
-    fn on_disconnected(&mut self, session_over: bool) {
+    fn on_disconnected(&mut self, end: benilla_protocol::SessionEnd, session_over: bool) {
+        if end == benilla_protocol::SessionEnd::LoggedOut {
+            return;
+        }
         self.connected = false;
         if !session_over {
             return;
@@ -1250,7 +1255,7 @@ pub(crate) fn drive_glue_mgr(
         mgr.on_roster(msg.characters.clone(), msg.cause, msg.epoch);
     }
     for msg in news.disconnects.read() {
-        mgr.on_disconnected(msg.session_over);
+        mgr.on_disconnected(msg.end, msg.session_over);
     }
     mgr.tick();
     send_asks(&mut mgr, logins, &mut asks);

crates/benilla-app/src/ui_script/glue_mgr_tests.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 use super::*;
 use crate::char_select::ClientState;
 use crate::net::{LoginAbandon, LoginRequest, LoginSubmit, RosterFailedMessage, WorldAuthMessage};
-use benilla_protocol::RealmInfo;
+use benilla_protocol::{RealmInfo, SessionEnd};
 
 /// The manager with both agreements accepted, the way a returning player's CVars seed it.
 fn accepted() -> GlueMgr {
@@ -694,31 +694,50 @@ fn status_dialog_click_cancels_by_state() {
     assert!(m.outbox.is_empty(), "a cancelled query reports nothing");
 }
 
-/// A session that is over fires `DISCONNECTED_FROM_SERVER`; a logout's relist does not; in the
-/// world connect, a completed failed operation opens `OKAY` with its text instead
+/// A session that is over fires `DISCONNECTED_FROM_SERVER`; an unattended reconnect does not; in
+/// the world connect, a completed failed operation opens `OKAY` with its text instead
 /// (`0x46c671`-`0x46c68f`), and a running one still fires the event.
 #[test]
 fn a_lost_session_fires_disconnected() {
     let mut m = accepted();
-    m.on_disconnected(false);
+    m.on_disconnected(SessionEnd::Lost, false);
     assert!(m.outbox.is_empty());
-    m.on_disconnected(true);
+    m.on_disconnected(SessionEnd::Lost, true);
     assert_eq!(m.outbox, [Outgoing::Disconnected]);
 
     let mut m = accepted();
     m.state = ScreenState::Connect;
     m.op = Operation::begin(OpKind::Connect, STATUS_CONNECTING);
-    m.on_disconnected(true);
+    m.on_disconnected(SessionEnd::Lost, true);
     assert_eq!(m.outbox, [Outgoing::Disconnected]);
 
     let mut m = accepted();
     m.state = ScreenState::Connect;
     m.op.complete(false, STATUS_FAILED_TO_CONNECT);
-    m.on_disconnected(true);
+    m.on_disconnected(SessionEnd::Lost, true);
     assert_eq!(m.outbox, [open("OKAY", Some("RESPONSE_FAILED_TO_CONNECT"))]);
     assert_eq!(m.state, ScreenState::Idle);
 }
 
+/// A logout or a refused login keeps the world link (`0x5aaeb0`, `0x5aad70`), so the character
+/// screen that comes back up finds it connected (`0x46d380`) and its `GetCharacterListUpdate`
+/// sends the roster ask on it (`0x46b880` → `0x5aabb0`), with no dialog of a disconnect.
+#[test]
+fn a_logout_keeps_the_link_for_the_screens_roster_ask() {
+    let mut m = accepted();
+    m.connected = true;
+    m.on_disconnected(SessionEnd::LoggedOut, false);
+    assert!(m.connected, "the link is kept");
+    assert!(m.outbox.is_empty());
+    m.get_character_list();
+    let (_, park) = m.take_asks();
+    assert!(
+        matches!(park[..], [ParkRequest::Char(CharRequest::Enumerate)]),
+        "the ask goes out on the link"
+    );
+    assert!(!m.op.done, "and waits for its reply");
+}
+
 // ── The stock screens over the native login, end to end ─────────────────────────────────────────
 
 /// An app with the manager, the native login's glue asks and the stock glue VM on the login

crates/benilla-protocol/src/events.rs

@@ -92,7 +92,9 @@ pub enum RosterCause {
 /// `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.
@@ -194,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 {

crates/benilla-protocol/src/lib.rs

@@ -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, WorldConnectEvent, WorldReader,
-    WorldSession, WorldWriter, WORLD_PORT,
+    observe_sends, SendObserver, Unparsed, WardenRequired, WorldAuthReject, WorldConnectEvent,
+    WorldReader, WorldSession, WorldWriter, WORLD_PORT,
 };
 
 use std::net::TcpStream;
@@ -161,6 +161,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.

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

@@ -20,6 +20,7 @@ mod writer;
 
 pub use reader::WorldReader;
 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,9 +46,31 @@ 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()
+    })
 }
 
+/// 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]);

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 {

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

@@ -58,18 +58,28 @@ 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>>,
 }
 
@@ -144,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.
@@ -160,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`.
@@ -200,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.
@@ -220,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).
@@ -245,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.
@@ -269,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);
             }
         }
     }
@@ -280,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);
             }
         }
     }
@@ -290,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| {
@@ -787,6 +809,7 @@ impl WorldSession {
             WorldReader {
                 stream: read_stream,
                 decrypter,
+                link: self.link,
             },
             WorldWriter {
                 stream: self.stream,
@@ -796,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, &[])?;

crates/benilla-protocol/tests/handshake.rs

@@ -147,3 +147,97 @@ fn sends_before_the_split_are_observed_in_order() {
     assert_eq!(mine[1].1, messages::full_guid(0x2a));
     assert!(mine[2].1.is_empty());
 }
+
+/// Read one client packet as the server sees it, header decrypted: its opcode.
+fn read_client(stream: &mut TcpStream, crypto: &mut HeaderCrypto) -> Option<u16> {
+    let mut header = [0u8; 6];
+    stream.read_exact(&mut header).ok()?;
+    crypto.decrypter().decrypt(&mut header);
+    let size = u16::from_be_bytes([header[0], header[1]]) as usize;
+    let mut body = vec![0u8; size - 4];
+    stream.read_exact(&mut body).ok()?;
+    Some(u16::from_le_bytes([header[2], header[3]]))
+}
+
+/// A session split for the world and joined again after the logout is the same connection: the
+/// header cipher goes on from where each direction stood, so the next `CMSG_CHAR_ENUM` is read and
+/// answered on it, and the `SMSG_ADDON_INFO` the connection got once still stands for the next
+/// login (vmangos sends it only after the authentication, `WorldSocket.cpp:446-448`).
+#[test]
+fn a_rejoined_session_goes_on_on_the_same_connection() {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+    let server = thread::spawn(move || {
+        let (mut stream, _) = listener.accept().unwrap();
+        send(
+            &mut stream,
+            None,
+            opcode::SMSG_AUTH_CHALLENGE,
+            &SERVER_SEED.to_le_bytes(),
+        );
+        read_auth_session(&mut stream);
+        let mut crypto = HeaderCrypto::from_session_key(SESSION_KEY);
+        send(
+            &mut stream,
+            Some(&mut crypto),
+            opcode::SMSG_AUTH_RESPONSE,
+            &[messages::AUTH_OK],
+        );
+        // One record, status 2, nothing provided.
+        send(
+            &mut stream,
+            Some(&mut crypto),
+            opcode::SMSG_ADDON_INFO,
+            &[2, 0, 0],
+        );
+        let mut seen = Vec::new();
+        while let Some(op) = read_client(&mut stream, &mut crypto) {
+            seen.push(op);
+            match op {
+                opcode::CMSG_CHAR_ENUM => {
+                    send(&mut stream, Some(&mut crypto), opcode::SMSG_CHAR_ENUM, &[0])
+                }
+                opcode::CMSG_LOGOUT_REQUEST => send(
+                    &mut stream,
+                    Some(&mut crypto),
+                    opcode::SMSG_LOGOUT_COMPLETE,
+                    &[],
+                ),
+                _ => {}
+            }
+        }
+        seen
+    });
+
+    let mut session = WorldSession::connect(&addr, "one", SESSION_KEY).unwrap();
+    assert!(session.char_enum().unwrap().is_empty());
+    assert_eq!(session.addon_info(), Some(&[2u8][..]));
+    session.player_login(0x2a).unwrap();
+    let (mut reader, mut writer) = session.into_split().unwrap();
+    writer.logout_request().unwrap();
+    assert!(matches!(
+        reader.recv().unwrap(),
+        messages::ServerPacket::LogoutComplete
+    ));
+
+    let mut session = WorldSession::rejoin(reader, writer).unwrap();
+    assert!(
+        session.char_enum().unwrap().is_empty(),
+        "the roster is asked and answered on the joined session"
+    );
+    assert_eq!(
+        session.addon_info(),
+        Some(&[2u8][..]),
+        "the connection's add-on statuses stand for the next login"
+    );
+    drop(session);
+    assert_eq!(
+        server.join().unwrap(),
+        [
+            opcode::CMSG_CHAR_ENUM,
+            opcode::CMSG_PLAYER_LOGIN,
+            opcode::CMSG_LOGOUT_REQUEST,
+            opcode::CMSG_CHAR_ENUM
+        ]
+    );
+}

crates/benilla-srp/src/vanilla_header.rs

@@ -149,6 +149,11 @@ impl HeaderCrypto {
         (self.encrypt, self.decrypt)
     }
 
+    /// The halves [`Self::split`] made, joined again with the state each direction reached.
+    pub fn join(encrypt: EncrypterHalf, decrypt: DecrypterHalf) -> Self {
+        Self { encrypt, decrypt }
+    }
+
     /// The cipher from a session key alone, skipping the proof: a server's [`EncrypterHalf`] is
     /// what a client's [`DecrypterHalf`] inverts, so a fake world server speaks real headers.
     pub fn from_session_key(session_key: [u8; SESSION_KEY_LENGTH]) -> Self {

docs/MAP.md

@@ -496,7 +496,7 @@
 - `WOW_UI_SCALE` — benilla-app/src/cvars/mod.rs, benilla-app/src/ui_script/mod.rs
 - `WOW_UI_SPLICE_VERIFY` — benilla-app/src/ui_script/extract/mod.rs
 - `WOW_UISCRIPT_DUMP` — benilla-app/tests/ui_script_wall.rs
-- `WOW_UNATTENDED` — benilla-app/src/local_state.rs, benilla-app/src/login/mod.rs, benilla-app/src/net.rs, benilla-app/src/run_mode.rs, benilla-world/src/bgwin.rs
+- `WOW_UNATTENDED` — benilla-app/src/local_state.rs, benilla-app/src/login/mod.rs, benilla-app/src/net.rs, benilla-app/src/net/session.rs, benilla-app/src/run_mode.rs, benilla-world/src/bgwin.rs
 - `WOW_UNIT_VISUALS` — benilla-app/src/capture/probes/visual_census.rs, benilla-app/src/dev.rs
 - `WOW_UNIT_VISUALS_RADIUS` — benilla-app/src/capture/probes/visual_census.rs
 - `WOW_UPLOAD_BUDGET` — benilla-world/src/world_plugins.rs

scripts/smoke.sh

@@ -101,6 +101,15 @@ printf '%s' "$plain" | grep -q "logout-smoke: empty roster" &&
     fail "$user has no characters — the re-entry leg cannot run (make one, or rig it: WOW_RIG=…)"
 printf '%s' "$plain" | grep -q "logout-smoke: done" ||
     fail "the round trip never completed (no 'logout-smoke: done')"
+# The logout returns to character select on the link the world was entered by and re-lists there
+# (`0x5aaeb0` → `0x46b880`): a roster on that world session, and one realmd logon for the walk.
+printf '%s' "$plain" | grep -q "logout-smoke: FAILED" &&
+    fail "$(printf '%s\n' "$plain" | sed -n 's/.*logout-smoke: FAILED — //p' | tail -1)"
+printf '%s' "$plain" | grep -q "logout-smoke: roster after the logout on the same link" ||
+    fail "no roster came on the same link after the logout"
+logons="$(printf '%s\n' "$plain" | grep -cF "net: parked at the realm list")"
+[ "$logons" -eq 1 ] ||
+    fail "$logons realmd logons in the round trip, expected 1 — the logout logged on again"
 
 # The re-entered character must be drivable: the leg lists what would block movement (a /logout
 # root must end with its session), and anything but `none` fails.