portable: Phase 5 - session protocol, owner/host store split, host driver, simulated-realm matrix

OwnerService/HostService exchange only SessionOffer, PortableSessionStarted, PortableCheckpoint and
OwnerAck. Every checkpoint is merge3(C0, B0, current B1); sequences are monotonic and idempotent;
stale/superseded/malformed/oversized messages change nothing. Design and login-path audit in
docs/PORTABLE_SESSIONS.md.

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

crates/coa-core/src/portable/ids.rs

@@ -96,6 +96,10 @@ uuid7_id!(
     /// One import of a portable character into a realm (the key of the import journal).
     ImportId
 );
+uuid7_id!(
+    /// One runtime session of a portable character on a realm: from its automatic baseline to its final checkpoint.
+    SessionId
+);
 
 /// A reference to a piece of game content that cannot be mistaken for another namespace's numbering:
 /// `core:wotlk:item:19019`, `coa:wardrobe:10581`, `mod:<module>:item:7`.

crates/coa-core/src/portable/migrations/005_sessions.sql

@@ -0,0 +1,69 @@
+-- Schema 5: runtime portable sessions (Phase 5). The OWNER side and the HOST side keep their own tables; one Manager may
+-- play both roles on one file in a prototype, but a deployment uses two files and the roles never read each other's.
+--
+--  owner_session / owner_checkpoint : what the Owner has offered, started and applied for one session of one character on one
+--                                     realm. C0 and B0 are kept as sealed snapshots (the anchors of every checkpoint merge) and
+--                                     every applied sequence is logged (idempotency, ordering, lost acknowledgements).
+--  host_session / host_outbox       : what the Host has armed on its realm, the sequence it is at, the state it has already
+--                                     read from the realm, and the messages not yet acknowledged.
+
+CREATE TABLE owner_session (
+    session_id          TEXT PRIMARY KEY,
+    character_id        TEXT NOT NULL REFERENCES character(character_id) ON DELETE CASCADE,
+    server_id           TEXT NOT NULL,
+    state               TEXT NOT NULL CHECK (state IN ('offered', 'open', 'closed', 'superseded')),
+    c0_revision         INTEGER NOT NULL,
+    c0_hash             BLOB NOT NULL CHECK (length(c0_hash) = 32),
+    c0_payload          BLOB NOT NULL,
+    baseline_generation INTEGER,
+    b0_hash             BLOB CHECK (b0_hash IS NULL OR length(b0_hash) = 32),
+    b0_payload          BLOB,
+    -- the canonical revision this session produced last (starts at c0_revision); the next checkpoint must still find it
+    head_revision       INTEGER NOT NULL,
+    last_sequence       INTEGER NOT NULL DEFAULT 0,
+    created_at          TEXT NOT NULL,
+    updated_at          TEXT NOT NULL,
+    CHECK (state NOT IN ('open', 'closed') OR b0_hash IS NOT NULL)
+) STRICT;
+CREATE UNIQUE INDEX owner_session_one_live ON owner_session(character_id, server_id) WHERE state IN ('offered', 'open');
+
+CREATE TABLE owner_checkpoint (
+    session_id         TEXT NOT NULL REFERENCES owner_session(session_id) ON DELETE CASCADE,
+    sequence           INTEGER NOT NULL,
+    content_hash       BLOB NOT NULL CHECK (length(content_hash) = 32),
+    resulting_revision INTEGER NOT NULL,
+    final              INTEGER NOT NULL CHECK (final IN (0, 1)),
+    next_session_id    TEXT,
+    applied_at         TEXT NOT NULL,
+    PRIMARY KEY (session_id, sequence)
+) STRICT;
+
+CREATE TABLE host_session (
+    session_id        TEXT PRIMARY KEY,
+    character_id      TEXT NOT NULL REFERENCES character(character_id) ON DELETE CASCADE,
+    server_id         TEXT NOT NULL,
+    local_guid        INTEGER,
+    base_revision     INTEGER NOT NULL,
+    generation        INTEGER NOT NULL DEFAULT 1,
+    -- armed: the realm holds the marker, B0 not read yet; open: B0 sent, checkpoints running; closed
+    state             TEXT NOT NULL CHECK (state IN ('armed', 'open', 'closed')),
+    -- the sequence the next checkpoint will use; persisted BEFORE the realm is asked to save
+    next_sequence     INTEGER NOT NULL DEFAULT 1,
+    pending_sequence  INTEGER,
+    acked_sequence    INTEGER NOT NULL DEFAULT 0,
+    owned_items       TEXT NOT NULL DEFAULT '[]',
+    owned_pets        TEXT NOT NULL DEFAULT '[]',
+    created_at        TEXT NOT NULL,
+    updated_at        TEXT NOT NULL
+) STRICT;
+CREATE UNIQUE INDEX host_session_one_live ON host_session(character_id, server_id) WHERE state IN ('armed', 'open');
+
+CREATE TABLE host_outbox (
+    session_id   TEXT NOT NULL REFERENCES host_session(session_id) ON DELETE CASCADE,
+    sequence     INTEGER NOT NULL,
+    kind         TEXT NOT NULL CHECK (kind IN ('started', 'checkpoint')),
+    message      BLOB NOT NULL,
+    state        TEXT NOT NULL CHECK (state IN ('pending', 'acked')),
+    created_at   TEXT NOT NULL,
+    PRIMARY KEY (session_id, sequence)
+) STRICT;

crates/coa-core/src/portable/mod.rs

@@ -13,6 +13,7 @@ pub mod ids;
 pub mod merge;
 pub mod model;
 pub mod realm;
+pub mod session;
 pub mod snapshot;
 pub mod store;
 pub mod versions;
@@ -22,6 +23,6 @@ pub(crate) mod fixtures;
 
 pub use collection::IdSet;
 pub use error::{PortableError, Result};
-pub use ids::{CharacterId, ContentId, ImportId, PortableItemId, PortablePetId, ProfileId};
+pub use ids::{CharacterId, ContentId, ImportId, PortableItemId, PortablePetId, ProfileId, SessionId};
 pub use model::{PortableCharacter, Ruleset};
 pub use store::Store;

crates/coa-core/src/portable/session/bridge.rs

@@ -0,0 +1,78 @@
+//! What the Host needs from a realm to run a session: the core's session marker, the checkpoint/release operations and a
+//! read-only export. `LiveBridge` (see `live.rs`) implements it with the Manager's `Db` and RA; tests implement it with a
+//! simulated realm.
+
+use std::collections::HashMap;
+
+use super::super::error::Result;
+use super::super::ids::{CharacterId, PortableItemId, PortablePetId, SessionId};
+use super::super::realm::Exported;
+
+/// The core's `coa_portable_session.state`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RowState {
+    /// Armed by the importer/updater: the character's first load will take the baseline.
+    WaitingBaseline,
+    /// The baseline transaction was written; the player is held until the Host releases the session.
+    BaselineReady,
+    /// Released: the session runs.
+    Active,
+    /// The logout save was written.
+    Ended,
+}
+
+impl RowState {
+    pub fn from_code(code: u64) -> Option<RowState> {
+        Some(match code {
+            0 => RowState::WaitingBaseline,
+            1 => RowState::BaselineReady,
+            2 => RowState::Active,
+            3 => RowState::Ended,
+            _ => return None,
+        })
+    }
+}
+
+/// One row of the core's session table.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SessionRow {
+    pub guid: u32,
+    pub session_id: SessionId,
+    pub character_id: CharacterId,
+    pub imported_revision: u64,
+    pub generation: u32,
+    pub state: RowState,
+    pub checkpoint_seq: u64,
+    pub save_seq: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CheckpointReply {
+    /// The save was queued. This is **not** persistence: the marker row has to be seen.
+    Queued,
+    /// The character is being teleported or otherwise cannot be saved right now: try again.
+    Busy,
+    NotOnline,
+    /// The core refused (wrong session, stale sequence, not a portable character).
+    Refused(String),
+}
+
+/// A read of the realm: the character as committed, and the session row **of the same consistent snapshot**.
+pub struct RealmRead {
+    pub exported: Exported,
+    pub session: Option<SessionRow>,
+    pub online: bool,
+}
+
+pub trait RealmBridge {
+    /// Cheap poll of one character's session row.
+    fn session_row(&mut self, local_guid: u32) -> Result<Option<SessionRow>>;
+    /// Read-only export of the committed character. Allowed while the character is online **if** it has a session row.
+    fn read(&mut self, local_guid: u32, prior_items: &HashMap<u32, (PortableItemId, String)>, prior_pets: &HashMap<u32, (PortablePetId, String)>) -> Result<RealmRead>;
+    /// `portable checkpoint <local_guid> <session_id> <sequence>`
+    fn request_checkpoint(&mut self, local_guid: u32, session: SessionId, sequence: u64) -> Result<CheckpointReply>;
+    /// `portable release <session_id>`
+    fn release(&mut self, session: SessionId) -> Result<()>;
+    /// Put the character's marker back to "waiting for baseline" with a new session (the character is offline).
+    fn arm(&mut self, local_guid: u32, session: SessionId, character: CharacterId, revision: u64, generation: u32) -> Result<()>;
+}

crates/coa-core/src/portable/session/fake.rs

@@ -0,0 +1,259 @@
+//! A simulated realm for the session tests: the core's marker, gate, checkpoint and logout behaviour, and a realm that normalises
+//! characters on its own. It is deliberately small; the real worldserver is exercised by `live` tests.
+
+use std::collections::HashMap;
+
+use super::super::error::{PortableError, Result};
+use super::super::identity::item_identity;
+use super::super::ids::{CharacterId, ContentId, PortableItemId, PortablePetId, SessionId};
+use super::super::model::*;
+use super::super::realm::Exported;
+use super::super::store::{pet_identity, ImportAllocation, ItemObservation, PetObservation, PlannedItem, PlannedPet, Store};
+use super::bridge::*;
+
+pub struct FakeRealm {
+    pub guid: u32,
+    /// What the player is playing right now.
+    pub memory: PortableCharacter,
+    /// What the realm's tables hold (the last save).
+    pub db: PortableCharacter,
+    item_guid: HashMap<PortableItemId, u32>,
+    pet_number: HashMap<PortablePetId, u32>,
+    next_item_guid: u32,
+    next_pet_number: u32,
+    pub row: Option<SessionRow>,
+    pub online: bool,
+    /// The core holds the player until the Host releases the session.
+    pub gated: bool,
+    /// Checkpoint requests that answer `Busy` before one succeeds.
+    pub busy: u32,
+    pub saves: u32,
+    pub released: Vec<SessionId>,
+    pub normalize: fn(&mut FakeRealm),
+}
+
+pub fn new_internal_item(n: u64, slot: u8, entry: u64) -> PortableItem {
+    PortableItem {
+        id: PortableItemId::from_uuid(crate::portable::fixtures::id7(7_000_000 + n)).unwrap(),
+        container: None,
+        slot,
+        entry: ContentId::new("coa", "item", entry).unwrap(),
+        count: 1,
+        duration: 0,
+        charges: vec![],
+        flags: 0,
+        enchantments: vec![],
+        random_property_id: 0,
+        durability: 30,
+        played_time: 0,
+        text: None,
+        creator_name: None,
+        gift: None,
+    }
+}
+
+impl FakeRealm {
+    pub fn new(guid: u32) -> Self {
+        let empty = crate::portable::fixtures::naked_level_one();
+        Self {
+            guid,
+            memory: empty.clone(),
+            db: empty,
+            item_guid: HashMap::new(),
+            pet_number: HashMap::new(),
+            next_item_guid: 50_000,
+            next_pet_number: 900,
+            row: None,
+            online: false,
+            gated: false,
+            busy: 0,
+            saves: 0,
+            released: vec![],
+            normalize: |_| {},
+        }
+    }
+
+    /// The character arrives (an import): items and pets get local ids in plan order, like the offline importer does.
+    pub fn import(&mut self, store: &mut Store, character: CharacterId, server_id: &str, revision: u64, session: SessionId) -> Result<()> {
+        let model = store.load_snapshot(character, revision)?;
+        let items: Vec<PlannedItem> = model.items.iter().map(|i| PlannedItem { id: i.id, entry: i.entry.clone(), identity: item_identity(&i.entry, i.random_property_id) }).collect();
+        let pets: Vec<PlannedPet> = model.pets.iter().map(|p| PlannedPet { id: p.id, entry: p.entry.clone(), identity: pet_identity(&p.entry, p.pet_type, p.created_by_spell) }).collect();
+        let ticket = store.begin_import(character, server_id, revision, &items, &pets)?;
+        let (item_base, pet_base) = (self.next_item_guid, self.next_pet_number);
+        self.next_item_guid += items.len() as u32;
+        self.next_pet_number += pets.len() as u32;
+        for (i, item) in items.iter().enumerate() {
+            self.item_guid.insert(item.id, item_base + i as u32);
+        }
+        for (i, pet) in pets.iter().enumerate() {
+            self.pet_number.insert(pet.id, pet_base + i as u32);
+        }
+        self.db = model.clone();
+        self.memory = model;
+        store.finish_import(ticket.import_id, ImportAllocation { local_guid: self.guid, item_base, pet_base })?;
+        self.row = Some(SessionRow { guid: self.guid, session_id: session, character_id: character, imported_revision: revision, generation: 1, state: RowState::WaitingBaseline, checkpoint_seq: 0, save_seq: 0 });
+        Ok(())
+    }
+
+    fn free_slot(&self) -> u8 {
+        (23u8..=38).chain(39..=66).find(|s| !self.memory.items.iter().any(|i| i.container.is_none() && i.slot == *s)).expect("a free slot")
+    }
+
+    /// The realm gives the character an item; it goes to the first free place.
+    pub fn give_item(&mut self, mut item: PortableItem) -> u32 {
+        item.slot = self.free_slot();
+        let guid = self.next_item_guid;
+        self.next_item_guid += 1;
+        self.item_guid.insert(item.id, guid);
+        self.memory.items.push(item);
+        guid
+    }
+
+    fn save(&mut self) {
+        self.db = self.memory.clone().normalized();
+        self.saves += 1;
+        if let Some(r) = &mut self.row {
+            r.save_seq += 1;
+        }
+    }
+
+    /// The player logs in: the realm normalises, and a portable character in `WaitingBaseline` is saved and held.
+    pub fn login(&mut self) {
+        assert!(!self.online);
+        self.online = true;
+        self.memory = self.db.clone();
+        (self.normalize)(self);
+        let waiting = matches!(&self.row, Some(r) if matches!(r.state, RowState::WaitingBaseline | RowState::BaselineReady));
+        if waiting {
+            self.save();
+            let row = self.row.as_mut().unwrap();
+            row.state = RowState::BaselineReady;
+            self.gated = true;
+        } else if let Some(r) = &mut self.row {
+            if r.state == RowState::Ended {
+                r.state = RowState::Active;
+            }
+        }
+    }
+
+    /// Gameplay. It is a test failure to play before the baseline was released: that is the property the gate guarantees.
+    pub fn play(&mut self, f: impl FnOnce(&mut PortableCharacter)) {
+        assert!(self.online, "the player is not online");
+        assert!(!self.gated, "gameplay happened before B0 was taken");
+        f(&mut self.memory);
+    }
+
+    pub fn autosave(&mut self) {
+        assert!(self.online);
+        self.save();
+    }
+
+    pub fn logout(&mut self) {
+        assert!(self.online);
+        self.save();
+        self.online = false;
+        self.gated = false;
+        if let Some(r) = &mut self.row {
+            r.state = RowState::Ended;
+        }
+    }
+
+    /// The worldserver or the client dies: the last save stays, nothing else.
+    pub fn crash(&mut self) {
+        self.online = false;
+        self.gated = false;
+        self.memory = self.db.clone();
+    }
+
+    /// The gate's timeout: the player is disconnected, never released.
+    pub fn gate_timeout(&mut self) {
+        assert!(self.gated);
+        self.online = false;
+        self.gated = false;
+    }
+}
+
+impl RealmBridge for FakeRealm {
+    fn session_row(&mut self, local_guid: u32) -> Result<Option<SessionRow>> {
+        Ok(if local_guid == self.guid { self.row.clone() } else { None })
+    }
+
+    fn read(&mut self, local_guid: u32, prior_items: &HashMap<u32, (PortableItemId, String)>, prior_pets: &HashMap<u32, (PortablePetId, String)>) -> Result<RealmRead> {
+        assert_eq!(local_guid, self.guid);
+        let mut model = self.db.clone();
+        let mut ids: HashMap<PortableItemId, PortableItemId> = HashMap::new();
+        let mut observations = Vec::new();
+        for item in &mut model.items {
+            let internal = item.id;
+            let guid = *self.item_guid.get(&internal).ok_or_else(|| PortableError::Invalid("the simulated realm lost track of an item".into()))?;
+            let identity = item_identity(&item.entry, item.random_property_id);
+            let id = match prior_items.get(&guid) {
+                Some((id, prior)) if *prior == identity => *id,
+                _ => PortableItemId::new(),
+            };
+            ids.insert(internal, id);
+            item.id = id;
+            observations.push(ItemObservation { portable_item_id: id, local_item_guid: guid, entry: item.entry.clone(), identity });
+        }
+        for item in &mut model.items {
+            item.container = item.container.and_then(|c| ids.get(&c).copied());
+        }
+        let mut pet_observations = Vec::new();
+        for pet in &mut model.pets {
+            let internal = pet.id;
+            let number = *self.pet_number.entry(internal).or_insert_with(|| {
+                self.next_pet_number += 1;
+                self.next_pet_number
+            });
+            let identity = pet_identity(&pet.entry, pet.pet_type, pet.created_by_spell);
+            let id = match prior_pets.get(&number) {
+                Some((id, prior)) if *prior == identity => *id,
+                _ => PortablePetId::new(),
+            };
+            pet.id = id;
+            pet_observations.push(PetObservation { portable_pet_id: id, local_pet_number: number, identity });
+        }
+        // the realm's tables are keyed by guid: a later read finds the same rows under the same internal ids
+        let model = model.normalized();
+        Ok(RealmRead { exported: Exported { model, local_guid: self.guid, account: 1, observations, pet_observations, warnings: vec![] }, session: self.row.clone(), online: self.online })
+    }
+
+    fn request_checkpoint(&mut self, local_guid: u32, session: SessionId, sequence: u64) -> Result<CheckpointReply> {
+        if local_guid != self.guid || !self.online {
+            return Ok(CheckpointReply::NotOnline);
+        }
+        let Some(row) = &self.row else { return Ok(CheckpointReply::Refused("not a portable character".into())) };
+        if row.session_id != session || row.state != RowState::Active {
+            return Ok(CheckpointReply::Refused("wrong session".into()));
+        }
+        if sequence <= row.checkpoint_seq {
+            return Ok(CheckpointReply::Refused("stale sequence".into()));
+        }
+        if self.busy > 0 {
+            self.busy -= 1;
+            return Ok(CheckpointReply::Busy);
+        }
+        self.save();
+        self.row.as_mut().unwrap().checkpoint_seq = sequence;
+        Ok(CheckpointReply::Queued)
+    }
+
+    fn release(&mut self, session: SessionId) -> Result<()> {
+        if let Some(r) = &mut self.row {
+            if r.session_id == session && r.state == RowState::BaselineReady {
+                r.state = RowState::Active;
+                self.gated = false;
+                self.released.push(session);
+            }
+        }
+        Ok(())
+    }
+
+    fn arm(&mut self, local_guid: u32, session: SessionId, character: CharacterId, revision: u64, generation: u32) -> Result<()> {
+        assert_eq!(local_guid, self.guid);
+        assert!(!self.online, "a character is re-armed while it is offline");
+        let save_seq = self.row.as_ref().map_or(0, |r| r.save_seq);
+        self.row = Some(SessionRow { guid: local_guid, session_id: session, character_id: character, imported_revision: revision, generation, state: RowState::WaitingBaseline, checkpoint_seq: 0, save_seq });
+        Ok(())
+    }
+}

crates/coa-core/src/portable/session/host.rs

@@ -0,0 +1,199 @@
+//! The Host's session driver: arms sessions, takes the automatic baseline, runs checkpoints and queues the messages the
+//! Owner must receive. It never decides what the canonical character becomes and never opens the Owner's database.
+
+use std::collections::HashMap;
+
+use super::super::error::{PortableError, Result};
+use super::super::ids::{CharacterId, PortableItemId, PortablePetId, ProfileId, SessionId};
+use super::super::store::{AckEffect, HostSession, HostState, OutboxMessage, Store};
+use super::bridge::{CheckpointReply, RealmBridge, RealmRead, RowState};
+use super::protocol::*;
+
+#[derive(Debug, Clone)]
+pub struct HostConfig {
+    /// Seconds between checkpoints of a running session.
+    pub checkpoint_interval_secs: u64,
+}
+
+impl Default for HostConfig {
+    fn default() -> Self {
+        Self { checkpoint_interval_secs: 60 }
+    }
+}
+
+/// What one tick did, for logs and tests.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum HostEvent {
+    /// `B0` was read and `PortableSessionStarted` queued; the player is released.
+    BaselineTaken { session: SessionId },
+    /// The realm held the player but the baseline could not be taken yet.
+    BaselineRetry { session: SessionId, why: String },
+    /// The realm released the session without the Host (nothing can be done: the baseline is lost).
+    BaselineMissed { session: SessionId },
+    CheckpointRequested { session: SessionId, sequence: u64 },
+    CheckpointBusy { session: SessionId, sequence: u64 },
+    CheckpointQueued { session: SessionId, sequence: u64, final_checkpoint: bool },
+    /// The realm shows another session than the Host armed: the Host's session is closed.
+    SessionReplaced { session: SessionId },
+    Refused { session: SessionId, why: String },
+}
+
+pub struct HostService<'a> {
+    store: &'a mut Store,
+    server_id: String,
+    config: HostConfig,
+    /// When the last checkpoint of each session was requested (not persisted: after a restart the first tick checkpoints).
+    last_checkpoint: HashMap<SessionId, u64>,
+}
+
+impl<'a> HostService<'a> {
+    pub fn new(store: &'a mut Store, server_id: &str, config: HostConfig) -> Self {
+        Self { store, server_id: server_id.to_string(), config, last_checkpoint: HashMap::new() }
+    }
+
+    pub fn store(&self) -> &Store {
+        self.store
+    }
+
+    /// Hand the Host an offer: it keeps a copy of the character and prepares the session. The caller then imports the
+    /// character into the realm (offline importer or the core's import service) with this session id armed, and calls
+    /// [`HostService::bind`] with the realm's local guid.
+    pub fn accept_offer(&mut self, profile: ProfileId, offer: &SessionOffer) -> Result<HostSession> {
+        if offer.protocol_version != PROTOCOL_VERSION {
+            return Err(PortableError::UnsupportedFormat { found: offer.protocol_version, supported: PROTOCOL_VERSION });
+        }
+        if offer.server_id != self.server_id {
+            return Err(PortableError::Invalid("the offer is for another realm".into()));
+        }
+        let model = offer.snapshot.open()?;
+        if model.character_id != offer.character_id {
+            return Err(PortableError::WrongCharacter { expected: offer.character_id, found: model.character_id });
+        }
+        self.store.host_install_copy(profile, &model, offer.canonical_revision, "owner")?;
+        self.store.host_prepare_session(offer, &model)
+    }
+
+    /// The character exists on the realm with this guid.
+    pub fn bind(&mut self, session: SessionId, local_guid: u32) -> Result<()> {
+        self.store.host_bind_session(session, local_guid)
+    }
+
+    /// One pass over every live session of this realm. Safe to call as often as wanted; every step is idempotent and resumable.
+    pub fn tick(&mut self, bridge: &mut dyn RealmBridge, now_secs: u64) -> Result<Vec<HostEvent>> {
+        let mut events = Vec::new();
+        for s in self.store.host_live_sessions(&self.server_id)? {
+            let Some(guid) = s.local_guid else { continue };
+            let Some(row) = bridge.session_row(guid)? else { continue };
+            if row.session_id != s.session_id {
+                if s.state == HostState::Open {
+                    events.push(HostEvent::SessionReplaced { session: s.session_id });
+                }
+                continue;
+            }
+            match (s.state, row.state) {
+                (HostState::Armed, RowState::BaselineReady) => self.take_baseline(bridge, &s, guid, &mut events)?,
+                (HostState::Armed, RowState::Active | RowState::Ended) => events.push(HostEvent::BaselineMissed { session: s.session_id }),
+                (HostState::Open, RowState::Active) => self.run_checkpoint(bridge, &s, guid, row.checkpoint_seq, false, now_secs, &mut events)?,
+                (HostState::Open, RowState::Ended) => self.run_checkpoint(bridge, &s, guid, row.checkpoint_seq, true, now_secs, &mut events)?,
+                _ => {}
+            }
+        }
+        Ok(events)
+    }
+
+    fn priors(&self, s: &HostSession) -> Result<(HashMap<u32, (PortableItemId, String)>, HashMap<u32, (PortablePetId, String)>)> {
+        Ok((self.store.active_item_lookup(s.character_id, &s.server_id)?, self.store.active_pet_lookup(s.character_id, &s.server_id)?))
+    }
+
+    fn take_baseline(&mut self, bridge: &mut dyn RealmBridge, s: &HostSession, guid: u32, events: &mut Vec<HostEvent>) -> Result<()> {
+        let (items, pets) = self.priors(s)?;
+        let RealmRead { exported, session, .. } = bridge.read(guid, &items, &pets)?;
+        // the baseline counts only if the very snapshot it came from still shows the baseline marker of this session
+        let Some(row) = session.filter(|r| r.session_id == s.session_id && r.state == RowState::BaselineReady) else {
+            events.push(HostEvent::BaselineRetry { session: s.session_id, why: "the marker moved while the baseline was being read".into() });
+            return Ok(());
+        };
+        let msg = PortableSessionStarted::new(s.session_id, s.character_id, &s.server_id, s.base_revision, row.generation, &exported.model)?;
+        self.store.host_queue_started(s.session_id, &msg, &exported.model, &exported.observations, &exported.pet_observations)?;
+        // the baseline is persisted (and queued): only now may the player move on
+        bridge.release(s.session_id)?;
+        events.push(HostEvent::BaselineTaken { session: s.session_id });
+        Ok(())
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn run_checkpoint(&mut self, bridge: &mut dyn RealmBridge, s: &HostSession, guid: u32, marker_seq: u64, final_checkpoint: bool, now_secs: u64, events: &mut Vec<HostEvent>) -> Result<()> {
+        // one checkpoint at a time: the last one has to be queued before another is reserved
+        let due = s.pending_sequence.is_some() || final_checkpoint || self.last_checkpoint.get(&s.session_id).is_none_or(|t| now_secs >= *t + self.config.checkpoint_interval_secs);
+        if !due {
+            return Ok(());
+        }
+        let sequence = self.store.host_begin_checkpoint(s.session_id)?;
+        if !final_checkpoint && marker_seq < sequence {
+            match bridge.request_checkpoint(guid, s.session_id, sequence)? {
+                CheckpointReply::Queued => {
+                    self.last_checkpoint.insert(s.session_id, now_secs);
+                    events.push(HostEvent::CheckpointRequested { session: s.session_id, sequence });
+                }
+                CheckpointReply::Busy => events.push(HostEvent::CheckpointBusy { session: s.session_id, sequence }),
+                CheckpointReply::NotOnline => {}
+                CheckpointReply::Refused(why) => events.push(HostEvent::Refused { session: s.session_id, why }),
+            }
+            // the marker is read from the database, never assumed from the reply
+            return Ok(());
+        }
+        let (items, pets) = self.priors(s)?;
+        let RealmRead { exported, session, .. } = bridge.read(guid, &items, &pets)?;
+        let Some(row) = session.filter(|r| r.session_id == s.session_id) else { return Ok(()) };
+        if !final_checkpoint && row.checkpoint_seq < sequence {
+            return Ok(());
+        }
+        if final_checkpoint && row.state != RowState::Ended {
+            return Ok(());
+        }
+        let msg = PortableCheckpoint::new(s.session_id, s.character_id, &s.server_id, s.base_revision, sequence, final_checkpoint, &exported.model)?;
+        self.store.host_queue_checkpoint(s.session_id, &msg, &exported.observations, &exported.pet_observations)?;
+        self.last_checkpoint.insert(s.session_id, now_secs);
+        events.push(HostEvent::CheckpointQueued { session: s.session_id, sequence, final_checkpoint });
+        Ok(())
+    }
+
+    /// Messages waiting for delivery, in order.
+    pub fn outbox(&self) -> Result<Vec<OutboxMessage>> {
+        self.store.host_outbox_pending(&self.server_id)
+    }
+
+    /// Apply an acknowledgement. When it ends a session, the realm's marker is re-armed for the session that follows.
+    pub fn receive_ack(&mut self, bridge: &mut dyn RealmBridge, ack: &OwnerAck) -> Result<AckEffect> {
+        let session = self.store.host_session(ack.session_id)?;
+        let effect = self.store.host_receive_ack(ack)?;
+        if let (AckEffect::Finished { next_session, next_revision }, Some(old)) = (&effect, session) {
+            self.rearm(bridge, old.character_id, *next_session, *next_revision)?;
+        }
+        Ok(effect)
+    }
+
+    /// Re-arm the realm for every armed session whose marker is not on the realm yet (after a crash between the local commit
+    /// of a final acknowledgement and the realm write).
+    pub fn rearm_pending(&mut self, bridge: &mut dyn RealmBridge) -> Result<usize> {
+        let mut n = 0;
+        for s in self.store.host_live_sessions(&self.server_id)? {
+            let (Some(guid), HostState::Armed) = (s.local_guid, s.state) else { continue };
+            let current = bridge.session_row(guid)?;
+            if current.as_ref().is_some_and(|r| r.session_id == s.session_id) {
+                continue;
+            }
+            if s.generation > 1 {
+                bridge.arm(guid, s.session_id, s.character_id, s.base_revision, s.generation)?;
+                n += 1;
+            }
+        }
+        Ok(n)
+    }
+
+    fn rearm(&mut self, bridge: &mut dyn RealmBridge, character: CharacterId, session: SessionId, revision: u64) -> Result<()> {
+        let s = self.store.host_session(session)?.ok_or_else(|| PortableError::Invalid("the next session was not recorded".into()))?;
+        let guid = s.local_guid.ok_or_else(|| PortableError::Invalid("the next session has no realm character".into()))?;
+        bridge.arm(guid, session, character, revision, s.generation)
+    }
+}

crates/coa-core/src/portable/session/mod.rs

@@ -0,0 +1,12 @@
+pub mod protocol;
+pub mod bridge;
+pub mod host;
+pub mod owner;
+
+pub use host::{HostConfig, HostEvent, HostService};
+pub use owner::OwnerService;
+
+#[cfg(test)]
+mod fake;
+#[cfg(test)]
+mod tests;

crates/coa-core/src/portable/session/owner.rs

@@ -0,0 +1,41 @@
+//! The Owner's entry points: the only place a message from a realm enters the canonical store.
+
+use super::super::error::Result;
+use super::super::ids::{CharacterId, SessionId};
+use super::super::store::Store;
+use super::protocol::*;
+
+pub struct OwnerService<'a> {
+    store: &'a mut Store,
+}
+
+impl<'a> OwnerService<'a> {
+    pub fn new(store: &'a mut Store) -> Self {
+        Self { store }
+    }
+
+    pub fn store(&self) -> &Store {
+        self.store
+    }
+
+    /// Offer the character's current canonical revision to a realm and open the session that follows it.
+    pub fn offer(&mut self, character: CharacterId, server_id: &str) -> Result<SessionOffer> {
+        self.store.owner_offer_session(character, server_id)
+    }
+
+    /// `PortableSessionStarted` as bytes from a transport. A message that cannot be parsed is an error and changes nothing.
+    pub fn handle_started(&mut self, bytes: &[u8]) -> Result<OwnerAck> {
+        let msg: PortableSessionStarted = from_json(bytes)?;
+        self.store.owner_session_started(&msg)
+    }
+
+    /// `PortableCheckpoint` as bytes from a transport.
+    pub fn handle_checkpoint(&mut self, bytes: &[u8]) -> Result<OwnerAck> {
+        let msg: PortableCheckpoint = from_json(bytes)?;
+        self.store.owner_checkpoint(&msg)
+    }
+
+    pub fn session(&self, session: SessionId) -> Result<Option<(String, u64, u64)>> {
+        self.store.owner_session_info(session)
+    }
+}

crates/coa-core/src/portable/session/protocol.rs

@@ -0,0 +1,181 @@
+//! The messages of a runtime portable session. Transport-neutral: a Registry/Relay later only has to carry these bytes.
+//!
+//! Every message is canonical JSON with unknown fields refused; snapshots travel in the existing snapshot envelope
+//! (zstd of canonical JSON, SHA-256 of the canonical JSON). Size limits are checked before anything is parsed.
+
+use base64::Engine;
+use serde::de::DeserializeOwned;
+use serde::{Deserialize, Serialize};
+
+use super::super::error::{PortableError, Result};
+use super::super::ids::{CharacterId, PortableItemId, PortablePetId, SessionId};
+use super::super::model::{limits::MAX_COMPRESSED_BYTES, PortableCharacter};
+use super::super::snapshot;
+
+pub const PROTOCOL_VERSION: u32 = 1;
+/// A message carries at most one compressed snapshot (base64: 4/3) plus a little structure.
+pub const MAX_MESSAGE_BYTES: usize = MAX_COMPRESSED_BYTES * 4 / 3 + 64 * 1024;
+/// The most item/pet ids an acknowledgement lists.
+pub const MAX_ACK_IDS: usize = 20_000;
+
+/// A sealed snapshot: the hash of the canonical JSON and the compressed bytes.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Envelope {
+    pub content_hash: String,
+    pub payload: String,
+}
+
+impl Envelope {
+    pub fn seal(model: &PortableCharacter) -> Result<Envelope> {
+        let encoded = snapshot::encode(model)?;
+        Ok(Envelope { content_hash: hex::encode(encoded.content_hash), payload: base64::engine::general_purpose::STANDARD.encode(&encoded.payload) })
+    }
+
+    pub fn from_encoded(encoded: &snapshot::EncodedSnapshot) -> Envelope {
+        Envelope { content_hash: hex::encode(encoded.content_hash), payload: base64::engine::general_purpose::STANDARD.encode(&encoded.payload) }
+    }
+
+    pub fn hash(&self) -> Result<[u8; 32]> {
+        let bytes = hex::decode(&self.content_hash).map_err(|_| PortableError::Invalid("the content hash is not hex".into()))?;
+        <[u8; 32]>::try_from(bytes.as_slice()).map_err(|_| PortableError::Invalid("the content hash is not 32 bytes".into()))
+    }
+
+    /// Decode and fully verify (size caps while inflating, hash, structure).
+    pub fn open(&self) -> Result<PortableCharacter> {
+        if self.payload.len() > MAX_COMPRESSED_BYTES * 4 / 3 + 8 {
+            return Err(PortableError::LimitExceeded("the snapshot payload is over the limit".into()));
+        }
+        let bytes = base64::engine::general_purpose::STANDARD.decode(&self.payload).map_err(|_| PortableError::Invalid("the snapshot payload is not base64".into()))?;
+        snapshot::decode(&bytes, Some(&self.hash()?))
+    }
+
+    /// The raw compressed bytes, for storage next to their hash.
+    pub fn bytes(&self) -> Result<Vec<u8>> {
+        base64::engine::general_purpose::STANDARD.decode(&self.payload).map_err(|_| PortableError::Invalid("the snapshot payload is not base64".into()))
+    }
+}
+
+/// Owner -> Host: a character to put on the realm, and the session that will follow it.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct SessionOffer {
+    pub protocol_version: u32,
+    pub session_id: SessionId,
+    pub character_id: CharacterId,
+    pub server_id: String,
+    pub canonical_revision: u64,
+    pub snapshot: Envelope,
+}
+
+/// Host -> Owner: the realm's own first load/save of the character was captured, before any gameplay.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct PortableSessionStarted {
+    pub protocol_version: u32,
+    pub session_id: SessionId,
+    pub character_id: CharacterId,
+    pub server_id: String,
+    pub base_canonical_revision: u64,
+    pub baseline_generation: u32,
+    pub b0: Envelope,
+    pub content_hash: String,
+}
+
+/// Host -> Owner: the realm's state now. `sequence` is monotonic per session; `final_checkpoint` ends it.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct PortableCheckpoint {
+    pub protocol_version: u32,
+    pub session_id: SessionId,
+    pub character_id: CharacterId,
+    pub server_id: String,
+    pub base_canonical_revision: u64,
+    pub sequence: u64,
+    pub final_checkpoint: bool,
+    pub realm_snapshot: Envelope,
+    pub content_hash: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, tag = "status", content = "reason", rename_all = "snake_case")]
+pub enum AckOutcome {
+    Applied,
+    /// The same sequence with the same content was applied before; the original result is repeated.
+    Duplicate,
+    /// An older sequence than the Owner already holds: ignored.
+    StaleSequence,
+    /// The session was superseded (another session or revision moved the character): nothing is accepted.
+    StaleSession,
+    Rejected(String),
+}
+
+impl AckOutcome {
+    /// The Owner holds this state (applied now or before).
+    pub fn accepted(&self) -> bool {
+        matches!(self, AckOutcome::Applied | AckOutcome::Duplicate)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct NextSession {
+    pub session_id: SessionId,
+    pub canonical_revision: u64,
+}
+
+/// Owner -> Host.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct OwnerAck {
+    pub protocol_version: u32,
+    pub session_id: SessionId,
+    /// 0 acknowledges `PortableSessionStarted`.
+    pub sequence: u64,
+    pub outcome: AckOutcome,
+    pub canonical_revision: u64,
+    pub canonical_hash: String,
+    /// The items and pets the canonical character owns after this checkpoint (what the Host maps as `present`).
+    pub owned_items: Vec<PortableItemId>,
+    pub owned_pets: Vec<PortablePetId>,
+    /// Only on the acknowledgement of a final checkpoint: the canonical character, which the realm is now synchronised with.
+    pub canonical: Option<Envelope>,
+    /// Only on the acknowledgement of a final checkpoint: the session that follows.
+    pub next_session: Option<NextSession>,
+}
+
+pub fn to_json<T: Serialize>(message: &T) -> Result<Vec<u8>> {
+    let bytes = serde_json::to_vec(message)?;
+    if bytes.len() > MAX_MESSAGE_BYTES {
+        return Err(PortableError::LimitExceeded(format!("a message is limited to {MAX_MESSAGE_BYTES} bytes")));
+    }
+    Ok(bytes)
+}
+
+/// Parse one message. The size is checked before the bytes are looked at, unknown fields are refused, the protocol version
+/// is checked before anything else is trusted.
+pub fn from_json<T: DeserializeOwned>(bytes: &[u8]) -> Result<T> {
+    if bytes.len() > MAX_MESSAGE_BYTES {
+        return Err(PortableError::LimitExceeded(format!("a message is limited to {MAX_MESSAGE_BYTES} bytes")));
+    }
+    let value: serde_json::Value = serde_json::from_slice(bytes)?;
+    let version = value.get("protocol_version").and_then(|v| v.as_u64()).ok_or_else(|| PortableError::Invalid("the message has no protocol version".into()))?;
+    if version != PROTOCOL_VERSION as u64 {
+        return Err(PortableError::UnsupportedFormat { found: version.min(u32::MAX as u64) as u32, supported: PROTOCOL_VERSION });
+    }
+    Ok(serde_json::from_value(value)?)
+}
+
+impl PortableSessionStarted {
+    pub fn new(session_id: SessionId, character_id: CharacterId, server_id: &str, base_canonical_revision: u64, baseline_generation: u32, b0: &PortableCharacter) -> Result<Self> {
+        let b0 = Envelope::seal(b0)?;
+        Ok(Self { protocol_version: PROTOCOL_VERSION, session_id, character_id, server_id: server_id.to_string(), base_canonical_revision, baseline_generation, content_hash: b0.content_hash.clone(), b0 })
+    }
+}
+
+impl PortableCheckpoint {
+    pub fn new(session_id: SessionId, character_id: CharacterId, server_id: &str, base_canonical_revision: u64, sequence: u64, final_checkpoint: bool, b1: &PortableCharacter) -> Result<Self> {
+        let realm_snapshot = Envelope::seal(b1)?;
+        Ok(Self { protocol_version: PROTOCOL_VERSION, session_id, character_id, server_id: server_id.to_string(), base_canonical_revision, sequence, final_checkpoint, content_hash: realm_snapshot.content_hash.clone(), realm_snapshot })
+    }
+}

crates/coa-core/src/portable/session/tests.rs

@@ -0,0 +1,550 @@
+//! The Owner and the Host as two separate stores, a simulated realm between them, and the messages carried by hand.
+
+use super::bridge::*;
+use super::fake::*;
+use super::protocol::*;
+use super::*;
+use crate::portable::fixtures::geared_level_eighty;
+use crate::portable::ids::{CharacterId, PortableItemId, SessionId};
+use crate::portable::model::*;
+use crate::portable::store::{HostState, Store};
+
+const SERVER: &str = "realm-b";
+const GUID: u32 = 4242;
+
+struct Setup {
+    owner: Store,
+    host: Store,
+    realm: FakeRealm,
+    id: CharacterId,
+    session: SessionId,
+    c0: PortableCharacter,
+}
+
+/// What a realm does by itself when it loads a character: default spells, resets, items it holds back, items of its own.
+fn normalizes(r: &mut FakeRealm) {
+    r.memory.build.spells.extend([(6654, 255), (23250, 255)]);
+    r.memory.progression.honor.today_honor = 0;
+    r.memory.progression.chosen_title = 0;
+    r.memory.progression.known_currencies |= 1 << 40;
+    let held: Vec<PortableItemId> = plain(&r.memory).into_iter().take(4).collect();
+    r.memory.items.retain(|i| !held.contains(&i.id));
+    r.give_item(new_internal_item(1, 30, 66_001));
+    r.give_item(new_internal_item(2, 31, 66_002));
+}
+
+fn plain(m: &PortableCharacter) -> Vec<PortableItemId> {
+    let containers: std::collections::HashSet<PortableItemId> = m.items.iter().filter_map(|i| i.container).collect();
+    m.items.iter().filter(|i| i.container.is_none() && !containers.contains(&i.id) && (23..40).contains(&i.slot)).map(|i| i.id).collect()
+}
+
+/// Canonical revision 10 on the Owner; the realm has the character imported and armed, not yet logged in.
+fn setup() -> Setup {
+    setup_in(Store::open_in_memory().unwrap(), Store::open_in_memory().unwrap())
+}
+
+fn setup_in(mut owner: Store, mut host: Store) -> Setup {
+    let profile = owner.default_profile().unwrap();
+    let id = owner.create_character(profile, geared_level_eighty(), "realm-a").unwrap();
+    for _ in 2..=10 {
+        let mut m = owner.load_current(id).unwrap();
+        m.progression.xp += 1;
+        let rev = owner.character(id).unwrap().revision;
+        owner.commit_snapshot(id, rev, m, "realm-a", None).unwrap();
+    }
+    assert_eq!(owner.character(id).unwrap().revision, 10);
+    let c0 = owner.load_current(id).unwrap();
+    let offer = OwnerService::new(&mut owner).offer(id, SERVER).unwrap();
+    let host_profile = host.default_profile().unwrap();
+    let mut realm = FakeRealm::new(GUID);
+    realm.normalize = normalizes;
+    {
+        let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+        h.accept_offer(host_profile, &offer).unwrap();
+        realm.import(&mut host, id, SERVER, offer.canonical_revision, offer.session_id).unwrap();
+        HostService::new(&mut host, SERVER, HostConfig::default()).bind(offer.session_id, GUID).unwrap();
+    }
+    Setup { owner, host, realm, id, session: offer.session_id, c0 }
+}
+
+/// Carry every queued message to the Owner and its answer back.
+fn deliver(host: &mut HostService, owner: &mut OwnerService, realm: &mut FakeRealm) -> Vec<OwnerAck> {
+    let mut acks = Vec::new();
+    for m in host.outbox().unwrap() {
+        let ack = if m.started { owner.handle_started(&m.bytes) } else { owner.handle_checkpoint(&m.bytes) }.unwrap();
+        host.receive_ack(realm, &ack).unwrap();
+        acks.push(ack);
+    }
+    acks
+}
+
+fn revision(owner: &OwnerService, id: CharacterId) -> u64 {
+    owner.store().character(id).unwrap().revision
+}
+
+/// One checkpoint: request, then wait for the marker and read.
+fn checkpoint(host: &mut HostService, owner: &mut OwnerService, realm: &mut FakeRealm, now: u64) -> Vec<HostEvent> {
+    let mut events = host.tick(realm, now).unwrap();
+    events.extend(host.tick(realm, now + 1).unwrap());
+    deliver(host, owner, realm);
+    events
+}
+
+#[test]
+fn the_whole_session_from_an_automatic_baseline_to_the_final_checkpoint() {
+    let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+
+    // the player logs in; the realm normalises and holds the player; the Host takes B0 without anybody doing anything
+    realm.login();
+    assert!(realm.gated, "the core holds the player until B0 exists");
+    let events = h.tick(&mut realm, 0).unwrap();
+    assert_eq!(events, vec![HostEvent::BaselineTaken { session }]);
+    assert!(!realm.gated, "released only after B0 was persisted");
+    let acks = deliver(&mut h, &mut o, &mut realm);
+    assert_eq!((acks.len(), &acks[0].outcome), (1, &AckOutcome::Applied));
+    assert_eq!(o.session(session).unwrap().unwrap().0, "open");
+    assert_eq!(revision(&o, id), 10, "a baseline changes nothing canonical");
+
+    // play, checkpoint #1 -> revision 11
+    realm.play(|m| {
+        m.progression.money += 5_000;
+        m.progression.honor.total_kills += 3;
+        m.build.spells.push((1111, 255));
+        m.quests.rewarded.push(777);
+        m.pets[0].exp += 100;
+    });
+    let bonus = new_internal_item(3, 32, 77_001);
+    realm.give_item(bonus.clone());
+    realm.memory.items.retain(|i| i.entry.id() != 66_002);
+    let events = checkpoint(&mut h, &mut o, &mut realm, 100);
+    assert!(events.contains(&HostEvent::CheckpointRequested { session, sequence: 1 }) && events.contains(&HostEvent::CheckpointQueued { session, sequence: 1, final_checkpoint: false }), "{events:?}");
+    assert_eq!(revision(&o, id), 11);
+    let c1 = o.store().load_current(id).unwrap();
+    assert_eq!(c1.progression.money, c0.progression.money + 5_000);
+    assert_eq!(c1.progression.honor.total_kills, c0.progression.honor.total_kills + 3);
+    assert!(c1.build.spells.contains(&(1111, 255)) && c1.quests.rewarded.contains(&777));
+    assert_eq!(c1.pets[0].exp, c0.pets[0].exp + 100);
+    assert!(c1.items.iter().any(|i| i.entry.id() == 77_001), "the new item is a session gain");
+    assert!(c1.items.iter().all(|i| i.entry.id() != 66_001 && i.entry.id() != 66_002), "the realm's own items never become canonical");
+    // realm normalisation did not propagate, and what the realm held back is still canonical
+    assert!(!c1.build.spells.contains(&(6654, 255)) && !c1.build.spells.contains(&(23250, 255)));
+    assert_eq!(c1.progression.honor.today_honor, c0.progression.honor.today_honor);
+    assert_eq!(c1.progression.chosen_title, c0.progression.chosen_title);
+    assert_eq!(c1.progression.known_currencies, c0.progression.known_currencies);
+    assert_eq!(c1.items.len(), c0.items.len() + 1, "four items the realm held back stay, one session gain");
+    let first = h.outbox().unwrap();
+    assert!(first.is_empty(), "the acknowledgement emptied the outbox");
+
+    // the same checkpoint delivered again: nothing happens
+    let sent = host_outbox_history(&h, session, 1);
+    let again = o.handle_checkpoint(&sent).unwrap();
+    assert_eq!(again.outcome, AckOutcome::Duplicate);
+    assert_eq!(revision(&o, id), 11);
+    assert_eq!(again.canonical_revision, 11);
+
+    // more play, checkpoint #2 -> revision 12; money counted once
+    realm.play(|m| m.progression.money += 250);
+    checkpoint(&mut h, &mut o, &mut realm, 200);
+    assert_eq!(revision(&o, id), 12);
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 5_250);
+
+    // the delayed #1 arrives after #2: ignored, the original answer is repeated
+    let late = o.handle_checkpoint(&sent).unwrap();
+    assert!(matches!(late.outcome, AckOutcome::Duplicate) && late.canonical_revision == 11);
+    assert_eq!(revision(&o, id), 12);
+
+    // logout -> final checkpoint -> the owner's final revision, the session closes, the next one is armed
+    realm.logout();
+    let events = h.tick(&mut realm, 300).unwrap();
+    assert!(events.contains(&HostEvent::CheckpointQueued { session, sequence: 3, final_checkpoint: true }), "{events:?}");
+    let acks = deliver(&mut h, &mut o, &mut realm);
+    assert_eq!(acks.len(), 1);
+    let next = acks[0].next_session.clone().expect("the final acknowledgement names the next session");
+    assert_eq!(o.session(session).unwrap().unwrap().0, "closed");
+    assert!(revision(&o, id) >= 12);
+    let row = realm.row.clone().unwrap();
+    assert_eq!((row.session_id, row.state, row.generation), (next.session_id, RowState::WaitingBaseline, 2), "the realm is armed for the next login");
+    let hs = h.store().host_session(session).unwrap().unwrap();
+    assert_eq!(hs.state, HostState::Closed);
+    let synced = h.store().synced_model(id, SERVER).unwrap().unwrap();
+    assert_eq!(synced, o.store().load_current(id).unwrap(), "the Host's realm is synchronised with what the Owner now holds");
+}
+
+/// The bytes of a message the Host queued (kept in the outbox table after acknowledgement).
+fn host_outbox_history(h: &HostService, session: SessionId, sequence: u64) -> Vec<u8> {
+    h.store().host_message(session, sequence).unwrap().expect("the message is kept")
+}
+
+// ---- the matrix ----------------------------------------------------------------------------------------------------------------------
+
+fn start(h: &mut HostService, o: &mut OwnerService, realm: &mut FakeRealm) {
+    realm.login();
+    if realm.gated {
+        h.tick(realm, 0).unwrap();
+        deliver(h, o, realm);
+    }
+}
+
+fn tamper(bytes: &[u8], f: impl FnOnce(&mut serde_json::Value)) -> Vec<u8> {
+    let mut v: serde_json::Value = serde_json::from_slice(bytes).unwrap();
+    f(&mut v);
+    serde_json::to_vec(&v).unwrap()
+}
+
+fn sealed_checkpoint(host: &Store, session: SessionId, sequence: u64) -> Vec<u8> {
+    host.host_message(session, sequence).unwrap().expect("queued")
+}
+
+#[test]
+fn a_host_that_never_answers_never_lets_the_player_play_and_the_baseline_is_retaken_at_the_next_login() {
+    let Setup { mut owner, mut host, mut realm, id, session, .. } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    realm.login();
+    assert!(realm.gated);
+    realm.gate_timeout();
+    assert!(!realm.online && realm.row.as_ref().unwrap().state == RowState::BaselineReady && realm.released.is_empty());
+    realm.login();
+    assert!(realm.gated);
+    assert_eq!(h.tick(&mut realm, 0).unwrap(), vec![HostEvent::BaselineTaken { session }]);
+    deliver(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 1);
+    assert_eq!(revision(&o, id), 10);
+}
+
+#[test]
+fn a_failed_intermediate_checkpoint_changes_nothing_canonical_and_is_retried() {
+    let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 100);
+    realm.busy = 2;
+    let e1 = h.tick(&mut realm, 100).unwrap();
+    let e2 = h.tick(&mut realm, 101).unwrap();
+    assert!(e1.contains(&HostEvent::CheckpointBusy { session, sequence: 1 }) && e2.contains(&HostEvent::CheckpointBusy { session, sequence: 1 }), "{e1:?} {e2:?}");
+    assert_eq!(revision(&o, id), 10, "a failed checkpoint changes nothing");
+    assert!(h.outbox().unwrap().is_empty());
+    h.tick(&mut realm, 102).unwrap();
+    h.tick(&mut realm, 103).unwrap();
+    deliver(&mut h, &mut o, &mut realm);
+    assert_eq!(revision(&o, id), 11);
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 100);
+    assert_eq!(h.store().host_session(session).unwrap().unwrap().acked_sequence, 1, "the retry used the same sequence number");
+}
+
+#[test]
+fn a_lost_acknowledgement_is_healed_by_redelivery_and_nothing_is_counted_twice() {
+    let Setup { mut owner, mut host, mut realm, id, c0, .. } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 700);
+    h.tick(&mut realm, 100).unwrap();
+    h.tick(&mut realm, 101).unwrap();
+    let msg = h.outbox().unwrap().remove(0);
+    let lost = o.handle_checkpoint(&msg.bytes).unwrap();
+    assert_eq!((lost.outcome.clone(), revision(&o, id)), (AckOutcome::Applied, 11));
+    assert_eq!(h.outbox().unwrap().len(), 1, "the Host still holds the message");
+    let acks = deliver(&mut h, &mut o, &mut realm);
+    assert_eq!(acks[0].outcome, AckOutcome::Duplicate);
+    assert_eq!(revision(&o, id), 11);
+    assert!(h.outbox().unwrap().is_empty());
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 700);
+}
+
+#[test]
+fn the_host_manager_can_restart_between_the_request_and_the_read_and_between_checkpoints() {
+    let dir = tempfile::tempdir().unwrap();
+    let host_file = dir.path().join("host.db");
+    let Setup { mut owner, host, mut realm, id, session, c0 } = setup_in(Store::open_in_memory().unwrap(), Store::open_file(&host_file).unwrap());
+    drop(host);
+    let mut o = OwnerService::new(&mut owner);
+    {
+        let mut hs = Store::open_file(&host_file).unwrap();
+        let mut h = HostService::new(&mut hs, SERVER, HostConfig::default());
+        start(&mut h, &mut o, &mut realm);
+        realm.play(|m| m.progression.money += 40);
+        h.tick(&mut realm, 100).unwrap();
+        assert_eq!(realm.row.as_ref().unwrap().checkpoint_seq, 1, "the realm saved at the request");
+    }
+    let mut hs = Store::open_file(&host_file).unwrap();
+    assert_eq!(hs.host_session(session).unwrap().unwrap().pending_sequence, Some(1), "the reserved sequence survived");
+    {
+        let mut h = HostService::new(&mut hs, SERVER, HostConfig::default());
+        h.tick(&mut realm, 101).unwrap();
+        deliver(&mut h, &mut o, &mut realm);
+        assert_eq!(revision(&o, id), 11);
+        realm.play(|m| m.progression.money += 60);
+        h.tick(&mut realm, 200).unwrap();
+        h.tick(&mut realm, 201).unwrap();
+        assert_eq!(h.outbox().unwrap().len(), 1);
+    }
+    drop(hs);
+    let mut hs = Store::open_file(&host_file).unwrap();
+    let mut h = HostService::new(&mut hs, SERVER, HostConfig::default());
+    assert_eq!(h.outbox().unwrap().len(), 1, "the outbox is persistent");
+    deliver(&mut h, &mut o, &mut realm);
+    assert_eq!(revision(&o, id), 12);
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 100);
+}
+
+#[test]
+fn the_owner_manager_can_restart_at_any_point() {
+    let dir = tempfile::tempdir().unwrap();
+    let owner_file = dir.path().join("owner.db");
+    let Setup { owner, mut host, mut realm, id, session, c0 } = setup_in(Store::open_file(&owner_file).unwrap(), Store::open_in_memory().unwrap());
+    drop(owner);
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    {
+        let mut os = Store::open_file(&owner_file).unwrap();
+        let mut o = OwnerService::new(&mut os);
+        start(&mut h, &mut o, &mut realm);
+    }
+    realm.play(|m| m.progression.money += 5);
+    h.tick(&mut realm, 100).unwrap();
+    h.tick(&mut realm, 101).unwrap();
+    {
+        let mut os = Store::open_file(&owner_file).unwrap();
+        let mut o = OwnerService::new(&mut os);
+        assert_eq!(o.session(session).unwrap().unwrap().0, "open", "the baseline survived the restart");
+        deliver(&mut h, &mut o, &mut realm);
+        assert_eq!(revision(&o, id), 11);
+    }
+    realm.play(|m| m.progression.money += 6);
+    h.tick(&mut realm, 200).unwrap();
+    h.tick(&mut realm, 201).unwrap();
+    let mut os = Store::open_file(&owner_file).unwrap();
+    let mut o = OwnerService::new(&mut os);
+    deliver(&mut h, &mut o, &mut realm);
+    assert_eq!(revision(&o, id), 12);
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 11);
+}
+
+#[test]
+fn a_worldserver_restart_continues_the_same_session_without_a_new_baseline() {
+    let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 10);
+    checkpoint(&mut h, &mut o, &mut realm, 100);
+    assert_eq!(revision(&o, id), 11);
+    realm.play(|m| m.progression.money += 99);
+    realm.crash();
+    assert!(h.tick(&mut realm, 150).unwrap().is_empty(), "an offline character with a running session needs nothing");
+    assert_eq!(revision(&o, id), 11, "the last acknowledged checkpoint wins");
+    realm.login();
+    assert!(!realm.gated);
+    assert_eq!(realm.row.as_ref().unwrap().session_id, session);
+    realm.play(|m| m.progression.money += 5);
+    checkpoint(&mut h, &mut o, &mut realm, 200);
+    assert_eq!(revision(&o, id), 12);
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 15, "the lost 99 stayed lost, the rest was counted once");
+}
+
+#[test]
+fn a_superseded_session_is_refused_and_closed() {
+    let Setup { mut owner, mut host, mut realm, id, session, .. } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    {
+        let mut o = OwnerService::new(&mut owner);
+        start(&mut h, &mut o, &mut realm);
+    }
+    let mut m = owner.load_current(id).unwrap();
+    m.progression.money += 1;
+    owner.commit_snapshot(id, 10, m, "realm-c", None).unwrap();
+    let mut o = OwnerService::new(&mut owner);
+    realm.play(|m| m.progression.money += 3);
+    h.tick(&mut realm, 100).unwrap();
+    h.tick(&mut realm, 101).unwrap();
+    let acks = deliver(&mut h, &mut o, &mut realm);
+    assert_eq!(acks[0].outcome, AckOutcome::StaleSession);
+    assert_eq!(revision(&o, id), 11, "nothing was merged over the newer state");
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, geared_level_eighty().progression.money + 1);
+    assert_eq!(h.store().host_session(session).unwrap().unwrap().state, HostState::Closed);
+    let msg = sealed_checkpoint(h.store(), session, 1);
+    assert_eq!(o.handle_checkpoint(&msg).unwrap().outcome, AckOutcome::StaleSession);
+}
+
+#[test]
+fn messages_for_the_wrong_character_session_or_realm_and_malformed_ones_change_nothing() {
+    let Setup { mut owner, mut host, mut realm, id, session, .. } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 9);
+    h.tick(&mut realm, 100).unwrap();
+    h.tick(&mut realm, 101).unwrap();
+    let good = h.outbox().unwrap().remove(0).bytes;
+    let before = revision(&o, id);
+    let other_character = CharacterId::new();
+    let other_session = SessionId::new();
+
+    let cases: Vec<(&str, Vec<u8>)> = vec![
+        ("another character", tamper(&good, |v| v["character_id"] = other_character.to_string().into())),
+        ("an unknown session", tamper(&good, |v| v["session_id"] = other_session.to_string().into())),
+        ("another realm", tamper(&good, |v| v["server_id"] = "realm-z".into())),
+        ("another base revision", tamper(&good, |v| v["base_canonical_revision"] = 3.into())),
+        ("a wrong content hash", tamper(&good, |v| v["content_hash"] = "00".repeat(32).into())),
+        ("sequence zero", tamper(&good, |v| v["sequence"] = 0.into())),
+        ("a snapshot of another character", {
+            let mut other = geared_level_eighty();
+            other.character_id = other_character;
+            to_json(&PortableCheckpoint::new(session, id, SERVER, 10, 1, false, &other).unwrap()).unwrap()
+        }),
+        ("a snapshot that does not match its hash", tamper(&good, |v| {
+            let payload = v["realm_snapshot"]["payload"].as_str().unwrap().to_string();
+            v["realm_snapshot"]["payload"] = format!("{}AA", &payload[..payload.len() - 4]).into();
+        })),
+    ];
+    for (name, bytes) in cases {
+        let ack = o.handle_checkpoint(&bytes).unwrap_or_else(|e| panic!("{name}: {e}"));
+        assert!(matches!(ack.outcome, AckOutcome::Rejected(_)), "{name}: {:?}", ack.outcome);
+        assert_eq!(revision(&o, id), before, "{name} must not change the canonical character");
+    }
+    for (name, bytes) in [
+        ("garbage", b"\x00\x01 not json".to_vec()),
+        ("an unknown field", tamper(&good, |v| v["extra"] = 1.into())),
+        ("a newer protocol", tamper(&good, |v| v["protocol_version"] = 99.into())),
+        ("no protocol version", tamper(&good, |v| {
+            v.as_object_mut().unwrap().remove("protocol_version");
+        })),
+        ("an over-limit message", vec![b' '; MAX_MESSAGE_BYTES + 1]),
+    ] {
+        assert!(o.handle_checkpoint(&bytes).is_err(), "{name}");
+        assert_eq!(revision(&o, id), before, "{name}");
+    }
+    let bomb = zstd::stream::encode_all(vec![b' '; crate::portable::model::limits::MAX_SNAPSHOT_BYTES + 4096].as_slice(), 19).unwrap();
+    let evil = tamper(&good, |v| {
+        v["realm_snapshot"]["payload"] = base64::Engine::encode(&base64::engine::general_purpose::STANDARD, &bomb).into();
+        v["realm_snapshot"]["content_hash"] = "11".repeat(32).into();
+        v["content_hash"] = "11".repeat(32).into();
+    });
+    assert!(matches!(o.handle_checkpoint(&evil).unwrap().outcome, AckOutcome::Rejected(_)));
+    assert_eq!(revision(&o, id), before);
+    assert_eq!(o.handle_checkpoint(&good).unwrap().outcome, AckOutcome::Applied);
+    assert_eq!(revision(&o, id), before + 1);
+}
+
+#[test]
+fn checkpoints_out_of_order_or_with_reused_sequences_are_refused_safely() {
+    let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 10);
+    h.tick(&mut realm, 100).unwrap();
+    h.tick(&mut realm, 101).unwrap();
+    realm.play(|m| m.progression.money += 20);
+    h.tick(&mut realm, 200).unwrap();
+    h.tick(&mut realm, 201).unwrap();
+    let (one, two) = (sealed_checkpoint(h.store(), session, 1), sealed_checkpoint(h.store(), session, 2));
+    assert_eq!(o.handle_checkpoint(&two).unwrap().outcome, AckOutcome::Applied);
+    assert_eq!(revision(&o, id), 11);
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 30);
+    assert_eq!(o.handle_checkpoint(&one).unwrap().outcome, AckOutcome::StaleSequence);
+    assert_eq!(revision(&o, id), 11);
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 30);
+    let reused = tamper(&two, |v| {
+        let mut m = geared_level_eighty();
+        m.progression.money += 123;
+        let other = PortableCheckpoint::new(session, id, SERVER, 10, 2, false, &m).unwrap();
+        v["realm_snapshot"] = serde_json::to_value(&other.realm_snapshot).unwrap();
+        v["content_hash"] = other.content_hash.into();
+    });
+    assert!(matches!(o.handle_checkpoint(&reused).unwrap().outcome, AckOutcome::Rejected(_)));
+    assert_eq!(revision(&o, id), 11);
+}
+
+#[test]
+fn a_checkpoint_that_changes_nothing_makes_no_revision_and_new_items_keep_their_ids() {
+    let Setup { mut owner, mut host, mut realm, id, c0, .. } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.give_item(new_internal_item(9, 40, 88_001));
+    realm.play(|m| m.progression.money += 1);
+    checkpoint(&mut h, &mut o, &mut realm, 100);
+    assert_eq!(revision(&o, id), 11);
+    let gained: Vec<PortableItemId> = o.store().load_current(id).unwrap().items.iter().filter(|i| i.entry.id() == 88_001).map(|i| i.id).collect();
+    assert_eq!(gained.len(), 1);
+    checkpoint(&mut h, &mut o, &mut realm, 200);
+    assert_eq!(revision(&o, id), 11, "no new revision for an unchanged character");
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 1, "nothing was counted again");
+    realm.play(|m| m.progression.money += 1);
+    checkpoint(&mut h, &mut o, &mut realm, 300);
+    let again: Vec<PortableItemId> = o.store().load_current(id).unwrap().items.iter().filter(|i| i.entry.id() == 88_001).map(|i| i.id).collect();
+    assert_eq!(again, gained, "an item first seen after B0 keeps one identity");
+    assert_eq!(revision(&o, id), 12);
+}
+
+#[test]
+fn a_logout_while_a_checkpoint_is_pending_turns_it_into_the_final_one() {
+    let Setup { mut owner, mut host, mut realm, id, session, c0 } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 8);
+    h.tick(&mut realm, 100).unwrap();
+    realm.play(|m| m.progression.money += 2);
+    realm.logout();
+    let events = h.tick(&mut realm, 101).unwrap();
+    assert!(events.contains(&HostEvent::CheckpointQueued { session, sequence: 1, final_checkpoint: true }), "{events:?}");
+    let acks = deliver(&mut h, &mut o, &mut realm);
+    assert!(acks[0].next_session.is_some());
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 10, "the final state includes the logout save");
+}
+
+#[test]
+fn a_lost_final_acknowledgement_is_healed_and_the_realm_is_armed_again_afterwards() {
+    let Setup { mut owner, mut host, mut realm, id, session, .. } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 4);
+    realm.logout();
+    h.tick(&mut realm, 100).unwrap();
+    let msg = h.outbox().unwrap().remove(0);
+    let lost = o.handle_checkpoint(&msg.bytes).unwrap();
+    let next = lost.next_session.clone().unwrap();
+    assert_eq!(realm.row.as_ref().unwrap().session_id, session, "the Host never heard the answer: the realm is not re-armed yet");
+    let acks = deliver(&mut h, &mut o, &mut realm);
+    assert_eq!(acks[0].outcome, AckOutcome::Duplicate);
+    assert_eq!(acks[0].next_session.as_ref().unwrap().session_id, next.session_id, "the same next session is named again");
+    assert_eq!(realm.row.as_ref().unwrap().session_id, next.session_id);
+    assert_eq!(o.session(next.session_id).unwrap().unwrap().0, "offered");
+    realm.login();
+    assert!(realm.gated);
+    assert_eq!(h.tick(&mut realm, 200).unwrap(), vec![HostEvent::BaselineTaken { session: next.session_id }]);
+    deliver(&mut h, &mut o, &mut realm);
+    assert!(matches!(h.store().host_session(next.session_id).unwrap().unwrap().state, HostState::Open));
+    assert_eq!(o.session(next.session_id).unwrap().unwrap().0, "open");
+    assert_eq!(revision(&o, id), o.store().character(id).unwrap().revision);
+}
+
+#[test]
+fn a_realm_that_was_never_given_a_session_row_is_left_alone() {
+    let Setup { mut host, mut realm, .. } = setup();
+    realm.row = None;
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    realm.login();
+    assert!(h.tick(&mut realm, 0).unwrap().is_empty());
+    assert!(!realm.gated);
+}
+
+#[test]
+fn an_autosave_between_checkpoints_is_harmless() {
+    let Setup { mut owner, mut host, mut realm, id, c0, .. } = setup();
+    let mut h = HostService::new(&mut host, SERVER, HostConfig::default());
+    let mut o = OwnerService::new(&mut owner);
+    start(&mut h, &mut o, &mut realm);
+    realm.play(|m| m.progression.money += 7);
+    realm.autosave();
+    realm.play(|m| m.progression.money += 3);
+    checkpoint(&mut h, &mut o, &mut realm, 100);
+    assert_eq!(revision(&o, id), 11);
+    assert_eq!(o.store().load_current(id).unwrap().progression.money, c0.progression.money + 10);
+}

crates/coa-core/src/portable/store.rs

@@ -18,7 +18,7 @@ use super::versions::{PORTABLE_COLLECTION_FORMAT_VERSION, SNAPSHOT_FORMAT_VERSIO
 pub const DATABASE_FILE: &str = "portable.db";
 /// `PRAGMA application_id`: "COAP". Refuses to adopt an unrelated SQLite file.
 const APPLICATION_ID: i64 = 0x434F_4150;
-const MIGRATIONS: &[&str] = &[include_str!("migrations/001_init.sql"), include_str!("migrations/002_item_mapping_lifecycle.sql"), include_str!("migrations/003_import_journal.sql"), include_str!("migrations/004_reconciliation.sql")];
+const MIGRATIONS: &[&str] = &[include_str!("migrations/001_init.sql"), include_str!("migrations/002_item_mapping_lifecycle.sql"), include_str!("migrations/003_import_journal.sql"), include_str!("migrations/004_reconciliation.sql"), include_str!("migrations/005_sessions.sql")];
 
 pub const DEFAULT_HISTORY_KEEP: u32 = 20;
 pub const MAX_HISTORY_KEEP: u32 = 1_000;
@@ -1018,8 +1018,11 @@ fn validate_observations(server_id: &str, observations: &[ItemObservation]) -> R
     Ok(())
 }
 
+mod host;
 mod journal;
+mod owner;
 mod sync;
+pub use host::*;
 pub use journal::*;
 pub use sync::*;
 

crates/coa-core/src/portable/store/host.rs

@@ -0,0 +1,332 @@
+//! The HOST side of a runtime portable session: what the realm's Manager remembers about the sessions it runs.
+//!
+//! The Host keeps a **copy** of the canonical character at the revision the Owner handed over (it needs it to import, to map
+//! items and to know what the realm was synchronised with); it never produces a new canonical revision. It keeps its own
+//! sequence numbers and an outbox of messages the Owner has not acknowledged.
+
+use rusqlite::{params, OptionalExtension};
+
+use super::*;
+use crate::portable::ids::{PortablePetId, SessionId};
+use crate::portable::session::protocol::*;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum HostState {
+    /// The realm holds the marker; `B0` has not been read yet.
+    Armed,
+    /// `B0` was read and sent; checkpoints run.
+    Open,
+    Closed,
+}
+
+impl HostState {
+    fn parse(s: &str) -> Result<Self> {
+        Ok(match s {
+            "armed" => HostState::Armed,
+            "open" => HostState::Open,
+            "closed" => HostState::Closed,
+            other => return Err(PortableError::Invalid(format!("unknown host session state {other:?}"))),
+        })
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HostSession {
+    pub session_id: SessionId,
+    pub character_id: CharacterId,
+    pub server_id: String,
+    pub local_guid: Option<u32>,
+    pub base_revision: u64,
+    pub generation: u32,
+    pub state: HostState,
+    pub next_sequence: u64,
+    pub pending_sequence: Option<u64>,
+    pub acked_sequence: u64,
+    pub owned_items: Vec<PortableItemId>,
+    pub owned_pets: Vec<PortablePetId>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OutboxMessage {
+    pub session_id: SessionId,
+    pub sequence: u64,
+    pub started: bool,
+    pub bytes: Vec<u8>,
+}
+
+/// What an acknowledgement changed on the Host.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AckEffect {
+    /// Nothing was waiting for it (an acknowledgement delivered twice).
+    Ignored,
+    Applied,
+    /// The final checkpoint was applied: the realm is now synchronised with the canonical character, and this session follows.
+    Finished { next_session: SessionId, next_revision: u64 },
+    /// The Owner will never accept this message (stale or rejected); it was dropped from the outbox.
+    Dropped(String),
+}
+
+fn read_host_session(conn: &Connection, session: SessionId) -> Result<Option<HostSession>> {
+    type Raw = (String, String, Option<i64>, i64, i64, String, i64, Option<i64>, i64, String, String);
+    let raw: Option<Raw> = conn
+        .query_row(
+            "SELECT character_id, server_id, local_guid, base_revision, generation, state, next_sequence, pending_sequence, acked_sequence, owned_items, owned_pets FROM host_session WHERE session_id = ?1",
+            [session.to_string()],
+            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?, r.get(7)?, r.get(8)?, r.get(9)?, r.get(10)?)),
+        )
+        .optional()?;
+    let Some((character, server_id, guid, base, generation, state, next, pending, acked, items, pets)) = raw else { return Ok(None) };
+    Ok(Some(HostSession {
+        session_id: session,
+        character_id: character.parse()?,
+        server_id,
+        local_guid: guid.map(|g| g as u32),
+        base_revision: base as u64,
+        generation: generation as u32,
+        state: HostState::parse(&state)?,
+        next_sequence: next as u64,
+        pending_sequence: pending.map(|p| p as u64),
+        acked_sequence: acked as u64,
+        owned_items: serde_json::from_str(&items)?,
+        owned_pets: serde_json::from_str(&pets)?,
+    }))
+}
+
+fn insert_host_session(tx: &Transaction<'_>, session: SessionId, id: CharacterId, server_id: &str, guid: Option<u32>, base: u64, generation: u32, model: &PortableCharacter) -> Result<()> {
+    let items: Vec<PortableItemId> = model.items.iter().map(|i| i.id).collect();
+    let pets: Vec<PortablePetId> = model.pets.iter().map(|p| p.id).collect();
+    let at = now();
+    tx.execute(
+        "INSERT INTO host_session(session_id, character_id, server_id, local_guid, base_revision, generation, state, next_sequence, acked_sequence, owned_items, owned_pets, created_at, updated_at)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'armed', 1, 0, ?7, ?8, ?9, ?9)",
+        params![session.to_string(), id.to_string(), server_id, guid, base as i64, generation, serde_json::to_string(&items)?, serde_json::to_string(&pets)?, at],
+    )?;
+    Ok(())
+}
+
+/// Install (or advance to) the canonical character the Owner handed over, as the Host's copy. The revision is the Owner's.
+fn install_copy_in_tx(tx: &Transaction<'_>, profile: ProfileId, model: &PortableCharacter, revision: u64, source_server_id: &str) -> Result<()> {
+    let encoded = snapshot::encode(model)?;
+    let id = model.character_id;
+    let at = now();
+    let exists: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM character WHERE character_id = ?1)", [id.to_string()], |r| r.get(0))?;
+    if !exists {
+        tx.execute(
+            "INSERT INTO character(character_id, profile_id, ruleset, name, race, class, gender, level, revision, created_at, updated_at, archived)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?10, 0)",
+            params![id.to_string(), profile.to_string(), model.ruleset.as_str(), model.identity.name, model.identity.race.to_string(), model.identity.class.to_string(), model.identity.gender, model.progression.level, revision as i64, at],
+        )?;
+        insert_snapshot(tx, id, revision, &encoded, source_server_id, Some("received"), &at)?;
+        return Ok(());
+    }
+    let record = read_character(tx, id)?;
+    if record.ruleset != model.ruleset {
+        return Err(PortableError::RulesetChange { from: record.ruleset.to_string(), to: model.ruleset.to_string() });
+    }
+    if revision < record.revision {
+        return Err(PortableError::StaleRevision { expected: revision, current: record.revision });
+    }
+    if revision == record.revision {
+        let (_, have) = read_snapshot_row(tx, id, revision)?;
+        if have.content_hash != encoded.content_hash {
+            return Err(PortableError::Invalid("the Host already holds a different canonical state at this revision".into()));
+        }
+        return Ok(());
+    }
+    insert_snapshot(tx, id, revision, &encoded, source_server_id, Some("received"), &at)?;
+    update_character(tx, id, model, revision, &at)?;
+    Ok(())
+}
+
+/// Presence of every active mapping after the Owner said what the canonical character owns: owned and shown = present, owned and
+/// not shown = filtered, shown and not owned = realm-local, neither = retired.
+fn reclassify_in_tx(tx: &Transaction<'_>, id: CharacterId, server_id: &str, revision: u64, items: &[PortableItemId], pets: &[PortablePetId]) -> Result<()> {
+    let at = now();
+    for (table, column, owned) in [("item_mapping", "portable_item_id", items.iter().map(|i| i.to_string()).collect::<std::collections::HashSet<_>>()), ("pet_mapping", "portable_pet_id", pets.iter().map(|p| p.to_string()).collect())] {
+        let mut stmt = tx.prepare(&format!("SELECT mapping_id, {column}, presence FROM {table} WHERE character_id = ?1 AND server_id = ?2 AND state = 'active'"))?;
+        let rows: Vec<(i64, String, String)> = stmt.query_map(params![id.to_string(), server_id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?.collect::<std::result::Result<_, _>>()?;
+        drop(stmt);
+        for (mapping, portable, presence) in rows {
+            let observed = presence != "filtered";
+            let is_owned = owned.contains(&portable);
+            match (is_owned, observed) {
+                (true, true) => tx.execute(&format!("UPDATE {table} SET presence = 'present', updated_at = ?2 WHERE mapping_id = ?1"), params![mapping, at])?,
+                (true, false) => 0,
+                (false, true) => tx.execute(&format!("UPDATE {table} SET presence = 'realm_local', updated_at = ?2 WHERE mapping_id = ?1"), params![mapping, at])?,
+                (false, false) => tx.execute(&format!("UPDATE {table} SET state = 'retired', retired_revision = ?3, retired_reason = 'absent', updated_at = ?2 WHERE mapping_id = ?1"), params![mapping, at, revision as i64])?,
+            };
+        }
+    }
+    Ok(())
+}
+
+impl Store {
+    /// The Host's copy of the character the Owner handed over (creates it, or advances it to a newer revision).
+    pub fn host_install_copy(&mut self, profile: ProfileId, model: &PortableCharacter, revision: u64, owner_server_id: &str) -> Result<()> {
+        check_server_id(owner_server_id)?;
+        let tx = self.write_tx()?;
+        install_copy_in_tx(&tx, profile, model, revision, owner_server_id)?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    /// Record that a session is about to be armed on the realm (before the import runs, so a crash in between is recoverable).
+    pub fn host_prepare_session(&mut self, offer: &SessionOffer, model: &PortableCharacter) -> Result<HostSession> {
+        let tx = self.write_tx()?;
+        if let Some(existing) = read_host_session(&tx, offer.session_id)? {
+            return Ok(existing);
+        }
+        insert_host_session(&tx, offer.session_id, offer.character_id, &offer.server_id, None, offer.canonical_revision, 1, model)?;
+        tx.commit()?;
+        Ok(read_host_session(&self.conn, offer.session_id)?.expect("just inserted"))
+    }
+
+    /// The import committed: the session now belongs to this local character.
+    pub fn host_bind_session(&mut self, session: SessionId, local_guid: u32) -> Result<()> {
+        let changed = self.conn.execute("UPDATE host_session SET local_guid = ?2, updated_at = ?3 WHERE session_id = ?1 AND (local_guid IS NULL OR local_guid = ?2)", params![session.to_string(), local_guid, now()])?;
+        if changed == 0 {
+            return Err(PortableError::Invalid(format!("session {session} does not exist or is bound to another character")));
+        }
+        Ok(())
+    }
+
+    pub fn host_session(&self, session: SessionId) -> Result<Option<HostSession>> {
+        read_host_session(&self.conn, session)
+    }
+
+    /// Armed and open sessions on one realm, oldest first.
+    pub fn host_live_sessions(&self, server_id: &str) -> Result<Vec<HostSession>> {
+        let mut stmt = self.conn.prepare("SELECT session_id FROM host_session WHERE server_id = ?1 AND state IN ('armed', 'open') ORDER BY session_id")?;
+        let ids: Vec<String> = stmt.query_map([server_id], |r| r.get(0))?.collect::<std::result::Result<_, _>>()?;
+        ids.iter().map(|s| Ok(read_host_session(&self.conn, s.parse()?)?.expect("it was just listed"))).collect()
+    }
+
+    /// Reserve the sequence of the next checkpoint **before** the realm is asked to save. Asking again returns the same number
+    /// until the checkpoint is queued, so a restart resumes the same checkpoint.
+    pub fn host_begin_checkpoint(&mut self, session: SessionId) -> Result<u64> {
+        let tx = self.write_tx()?;
+        let s = read_host_session(&tx, session)?.ok_or_else(|| PortableError::Invalid(format!("unknown session {session}")))?;
+        if s.state != HostState::Open {
+            return Err(PortableError::Invalid("a checkpoint needs an open session".into()));
+        }
+        let sequence = s.pending_sequence.unwrap_or(s.next_sequence);
+        tx.execute("UPDATE host_session SET pending_sequence = ?2, updated_at = ?3 WHERE session_id = ?1", params![session.to_string(), sequence as i64, now()])?;
+        tx.commit()?;
+        Ok(sequence)
+    }
+
+    /// `B0` was read from the realm: persist the baseline (presence of items and pets), queue `PortableSessionStarted`, open.
+    pub fn host_queue_started(&mut self, session: SessionId, msg: &PortableSessionStarted, b0: &PortableCharacter, items: &[ItemObservation], pets: &[PetObservation]) -> Result<()> {
+        let tx = self.write_tx()?;
+        let s = read_host_session(&tx, session)?.ok_or_else(|| PortableError::Invalid(format!("unknown session {session}")))?;
+        match s.state {
+            HostState::Armed => {}
+            _ => return Err(PortableError::Invalid("this session already has its baseline".into())),
+        }
+        capture_baseline_in_tx(&tx, s.character_id, &s.server_id, &BaselineInput { b0, items, pets })?;
+        let bytes = to_json(msg)?;
+        tx.execute("INSERT INTO host_outbox(session_id, sequence, kind, message, state, created_at) VALUES (?1, 0, 'started', ?2, 'pending', ?3)", params![session.to_string(), bytes, now()])?;
+        tx.execute("UPDATE host_session SET state = 'open', generation = ?2, updated_at = ?3 WHERE session_id = ?1", params![session.to_string(), msg.baseline_generation, now()])?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    /// `B1` was read after the realm's checkpoint marker appeared: register what the realm shows (so new items keep their ids in
+    /// the next checkpoint) and queue the message.
+    pub fn host_queue_checkpoint(&mut self, session: SessionId, msg: &PortableCheckpoint, items: &[ItemObservation], pets: &[PetObservation]) -> Result<()> {
+        let tx = self.write_tx()?;
+        let s = read_host_session(&tx, session)?.ok_or_else(|| PortableError::Invalid(format!("unknown session {session}")))?;
+        if s.state != HostState::Open || s.pending_sequence != Some(msg.sequence) {
+            return Err(PortableError::Invalid("this checkpoint was not reserved".into()));
+        }
+        let baseline = read_open_baseline(&tx, s.character_id, &s.server_id)?.ok_or(PortableError::NoBaseline)?;
+        let (c0_items, b0_items): (std::collections::HashSet<_>, std::collections::HashSet<_>) = (baseline.c0.items.iter().map(|i| i.id).collect(), baseline.b0.items.iter().map(|i| i.id).collect());
+        let (c0_pets, b0_pets): (std::collections::HashSet<_>, std::collections::HashSet<_>) = (baseline.c0.pets.iter().map(|p| p.id).collect(), baseline.b0.pets.iter().map(|p| p.id).collect());
+        // what the Owner is expected to own: what it owned at the last acknowledgement plus what is new since B0
+        let mut canonical_items: std::collections::HashSet<_> = s.owned_items.iter().copied().collect();
+        canonical_items.extend(items.iter().map(|o| o.portable_item_id).filter(|i| !b0_items.contains(i) && !c0_items.contains(i)));
+        let mut canonical_pets: std::collections::HashSet<_> = s.owned_pets.iter().copied().collect();
+        canonical_pets.extend(pets.iter().map(|o| o.portable_pet_id).filter(|p| !b0_pets.contains(p) && !c0_pets.contains(p)));
+        let mapping_revision: i64 = tx.query_row("SELECT last_revision FROM character_server_mapping WHERE character_id = ?1 AND server_id = ?2", params![s.character_id.to_string(), s.server_id], |r| r.get(0))?;
+        let protection = Protection { realm_local_items: items.iter().map(|o| o.portable_item_id).filter(|i| !canonical_items.contains(i)).collect(), canonical_items };
+        reconcile_with(&tx, s.character_id, &s.server_id, mapping_revision as u64, items, &protection)?;
+        let pet_protection = PetProtection { realm_local_pets: pets.iter().map(|o| o.portable_pet_id).filter(|p| !canonical_pets.contains(p)).collect(), canonical_pets };
+        sync_pets_in_tx(&tx, s.character_id, &s.server_id, mapping_revision as u64, pets, &pet_protection)?;
+        let bytes = to_json(msg)?;
+        tx.execute("INSERT INTO host_outbox(session_id, sequence, kind, message, state, created_at) VALUES (?1, ?2, 'checkpoint', ?3, 'pending', ?4)", params![session.to_string(), msg.sequence as i64, bytes, now()])?;
+        tx.execute("UPDATE host_session SET next_sequence = ?2, pending_sequence = NULL, updated_at = ?3 WHERE session_id = ?1", params![session.to_string(), msg.sequence as i64 + 1, now()])?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    /// The bytes of one queued message, acknowledged or not.
+    pub fn host_message(&self, session: SessionId, sequence: u64) -> Result<Option<Vec<u8>>> {
+        Ok(self.conn.query_row("SELECT message FROM host_outbox WHERE session_id = ?1 AND sequence = ?2", params![session.to_string(), sequence as i64], |r| r.get(0)).optional()?)
+    }
+
+    /// Messages the Owner has not acknowledged yet, in the order they must be delivered.
+    pub fn host_outbox_pending(&self, server_id: &str) -> Result<Vec<OutboxMessage>> {
+        let mut stmt = self.conn.prepare(
+            "SELECT o.session_id, o.sequence, o.kind, o.message FROM host_outbox o JOIN host_session s ON s.session_id = o.session_id
+             WHERE s.server_id = ?1 AND o.state = 'pending' ORDER BY o.created_at, o.sequence",
+        )?;
+        let rows = stmt.query_map([server_id], |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?, r.get::<_, String>(2)?, r.get::<_, Vec<u8>>(3)?)))?;
+        let mut out = Vec::new();
+        for row in rows {
+            let (session, sequence, kind, bytes) = row?;
+            out.push(OutboxMessage { session_id: session.parse()?, sequence: sequence as u64, started: kind == "started", bytes });
+        }
+        Ok(out)
+    }
+
+    /// Apply the Owner's acknowledgement.
+    pub fn host_receive_ack(&mut self, ack: &OwnerAck) -> Result<AckEffect> {
+        let tx = self.write_tx()?;
+        let Some(s) = read_host_session(&tx, ack.session_id)? else { return Ok(AckEffect::Ignored) };
+        let waiting: Option<String> = tx
+            .query_row("SELECT state FROM host_outbox WHERE session_id = ?1 AND sequence = ?2", params![ack.session_id.to_string(), ack.sequence as i64], |r| r.get(0))
+            .optional()?;
+        match waiting.as_deref() {
+            None => return Ok(AckEffect::Ignored),
+            Some("acked") => return Ok(AckEffect::Ignored),
+            Some(_) => {}
+        }
+        let at = now();
+        if !ack.outcome.accepted() {
+            tx.execute("UPDATE host_outbox SET state = 'acked' WHERE session_id = ?1 AND sequence = ?2", params![ack.session_id.to_string(), ack.sequence as i64])?;
+            if matches!(ack.outcome, AckOutcome::StaleSession) {
+                tx.execute("UPDATE host_session SET state = 'closed', updated_at = ?2 WHERE session_id = ?1", params![ack.session_id.to_string(), at])?;
+            }
+            tx.commit()?;
+            return Ok(AckEffect::Dropped(format!("{:?}", ack.outcome)));
+        }
+        tx.execute("UPDATE host_outbox SET state = 'acked' WHERE session_id = ?1 AND sequence = ?2", params![ack.session_id.to_string(), ack.sequence as i64])?;
+        if ack.sequence > 0 {
+            tx.execute(
+                "UPDATE host_session SET acked_sequence = max(acked_sequence, ?2), owned_items = ?3, owned_pets = ?4, updated_at = ?5 WHERE session_id = ?1",
+                params![ack.session_id.to_string(), ack.sequence as i64, serde_json::to_string(&ack.owned_items)?, serde_json::to_string(&ack.owned_pets)?, at],
+            )?;
+            reclassify_in_tx(&tx, s.character_id, &s.server_id, ack.canonical_revision, &ack.owned_items, &ack.owned_pets)?;
+        }
+        let (Some(canonical), Some(next)) = (&ack.canonical, &ack.next_session) else {
+            tx.commit()?;
+            return Ok(AckEffect::Applied);
+        };
+
+        // the final checkpoint: the realm is now synchronised with the canonical character, and the next session follows
+        let model = canonical.open()?;
+        if hex::encode(canonical.hash()?) != ack.canonical_hash || model.character_id != s.character_id {
+            return Err(PortableError::Invalid("the final acknowledgement carries another character than its session".into()));
+        }
+        let profile: String = tx.query_row("SELECT profile_id FROM character WHERE character_id = ?1", [s.character_id.to_string()], |r| r.get(0))?;
+        install_copy_in_tx(&tx, profile.parse()?, &model, ack.canonical_revision, "owner")?;
+        let encoded = snapshot::encode(&model)?;
+        tx.execute("UPDATE character_server_mapping SET last_revision = ?3, state = 'synced', updated_at = ?4 WHERE character_id = ?1 AND server_id = ?2", params![s.character_id.to_string(), s.server_id, ack.canonical_revision as i64, at])?;
+        set_synced_in_tx(&tx, s.character_id, &s.server_id, &encoded)?;
+        tx.execute("UPDATE realm_baseline SET state = 'closed', updated_at = ?3 WHERE character_id = ?1 AND server_id = ?2 AND state = 'open'", params![s.character_id.to_string(), s.server_id, at])?;
+        tx.execute("UPDATE host_session SET state = 'closed', updated_at = ?2 WHERE session_id = ?1", params![ack.session_id.to_string(), at])?;
+        insert_host_session(&tx, next.session_id, s.character_id, &s.server_id, s.local_guid, next.canonical_revision, s.generation + 1, &model)?;
+        tx.commit()?;
+        Ok(AckEffect::Finished { next_session: next.session_id, next_revision: next.canonical_revision })
+    }
+}

crates/coa-core/src/portable/store/owner.rs

@@ -0,0 +1,278 @@
+//! The OWNER side of a runtime portable session: offer a character to a realm, accept the realm's baseline and apply its
+//! checkpoints to the canonical character. Nothing here reads a realm, and nothing here knows how the messages travel.
+//!
+//! Every checkpoint is `merge3(C0, B0, current B1)` with the session's own `C0` and `B0` as anchors, so a checkpoint that
+//! is repeated, re-sent or superseded by a later one can never count progress twice.
+
+use rusqlite::{params, OptionalExtension};
+
+use super::*;
+use crate::portable::ids::SessionId;
+use crate::portable::merge::{merge3, Mode};
+use crate::portable::session::protocol::*;
+
+struct OwnerRow {
+    session_id: SessionId,
+    character_id: CharacterId,
+    server_id: String,
+    state: String,
+    c0_revision: u64,
+    c0_hash: [u8; 32],
+    c0_payload: Vec<u8>,
+    generation: Option<u32>,
+    b0_hash: Option<[u8; 32]>,
+    b0_payload: Option<Vec<u8>>,
+    head_revision: u64,
+    last_sequence: u64,
+}
+
+fn read_owner_row(conn: &Connection, session: SessionId) -> Result<Option<OwnerRow>> {
+    type Raw = (String, String, String, i64, Vec<u8>, Vec<u8>, Option<i64>, Option<Vec<u8>>, Option<Vec<u8>>, i64, i64);
+    let raw: Option<Raw> = conn
+        .query_row(
+            "SELECT character_id, server_id, state, c0_revision, c0_hash, c0_payload, baseline_generation, b0_hash, b0_payload, head_revision, last_sequence FROM owner_session WHERE session_id = ?1",
+            [session.to_string()],
+            |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?, r.get(7)?, r.get(8)?, r.get(9)?, r.get(10)?)),
+        )
+        .optional()?;
+    let Some((character, server_id, state, c0_revision, c0_hash, c0_payload, generation, b0_hash, b0_payload, head, last)) = raw else { return Ok(None) };
+    Ok(Some(OwnerRow {
+        session_id: session,
+        character_id: character.parse()?,
+        server_id,
+        state,
+        c0_revision: c0_revision as u64,
+        c0_hash: hash_from_blob(c0_hash)?,
+        c0_payload,
+        generation: generation.map(|g| g as u32),
+        b0_hash: b0_hash.map(hash_from_blob).transpose()?,
+        b0_payload,
+        head_revision: head as u64,
+        last_sequence: last as u64,
+    }))
+}
+
+fn head_hash(conn: &Connection, id: CharacterId, revision: u64) -> Result<[u8; 32]> {
+    let blob: Vec<u8> = conn.query_row("SELECT content_hash FROM snapshot WHERE character_id = ?1 AND revision = ?2", params![id.to_string(), revision as i64], |r| r.get(0)).map_err(|_| PortableError::UnknownRevision { character: id, revision })?;
+    hash_from_blob(blob)
+}
+
+fn ack(session: SessionId, sequence: u64, outcome: AckOutcome, revision: u64, hash: [u8; 32]) -> OwnerAck {
+    OwnerAck { protocol_version: PROTOCOL_VERSION, session_id: session, sequence, outcome, canonical_revision: revision, canonical_hash: hex::encode(hash), owned_items: vec![], owned_pets: vec![], canonical: None, next_session: None }
+}
+
+fn owned_ids(model: &PortableCharacter) -> (Vec<PortableItemId>, Vec<crate::portable::ids::PortablePetId>) {
+    let items: Vec<_> = model.items.iter().map(|i| i.id).take(MAX_ACK_IDS).collect();
+    let pets: Vec<_> = model.pets.iter().map(|p| p.id).take(MAX_ACK_IDS).collect();
+    (items, pets)
+}
+
+fn offer_from_row(row: &OwnerRow) -> SessionOffer {
+    SessionOffer {
+        protocol_version: PROTOCOL_VERSION,
+        session_id: row.session_id,
+        character_id: row.character_id,
+        server_id: row.server_id.clone(),
+        canonical_revision: row.c0_revision,
+        snapshot: Envelope::from_encoded(&EncodedSnapshot { content_hash: row.c0_hash, uncompressed_size: 0, payload: row.c0_payload.clone() }),
+    }
+}
+
+fn open_offer(tx: &Transaction<'_>, id: CharacterId, server_id: &str) -> Result<SessionOffer> {
+    let record = read_character(tx, id)?;
+    let (_, enc) = read_snapshot_row(tx, id, record.revision)?;
+    let session = SessionId::new();
+    let at = now();
+    tx.execute(
+        "INSERT INTO owner_session(session_id, character_id, server_id, state, c0_revision, c0_hash, c0_payload, head_revision, last_sequence, created_at, updated_at)
+         VALUES (?1, ?2, ?3, 'offered', ?4, ?5, ?6, ?4, 0, ?7, ?7)",
+        params![session.to_string(), id.to_string(), server_id, record.revision as i64, enc.content_hash.as_slice(), enc.payload, at],
+    )?;
+    Ok(offer_from_row(&read_owner_row(tx, session)?.expect("just inserted")))
+}
+
+impl Store {
+    /// Offer the character's current canonical revision to a realm and open the session that will follow it. An offer that was
+    /// made and not yet started is repeated as it was; one made against an older revision is superseded by a fresh one.
+    pub fn owner_offer_session(&mut self, id: CharacterId, server_id: &str) -> Result<SessionOffer> {
+        check_server_id(server_id)?;
+        let tx = self.write_tx()?;
+        let record = read_character(&tx, id)?;
+        let live: Option<String> = tx
+            .query_row("SELECT session_id FROM owner_session WHERE character_id = ?1 AND server_id = ?2 AND state IN ('offered', 'open')", params![id.to_string(), server_id], |r| r.get(0))
+            .optional()?;
+        if let Some(live) = live {
+            let row = read_owner_row(&tx, live.parse()?)?.expect("it was just found");
+            if row.state == "open" {
+                return Err(PortableError::SessionOpen);
+            }
+            if row.c0_revision == record.revision {
+                return Ok(offer_from_row(&row));
+            }
+            tx.execute("UPDATE owner_session SET state = 'superseded', updated_at = ?2 WHERE session_id = ?1", params![live, now()])?;
+        }
+        let offer = open_offer(&tx, id, server_id)?;
+        tx.commit()?;
+        Ok(offer)
+    }
+
+    /// The Host captured `B0`. Idempotent for the same baseline.
+    pub fn owner_session_started(&mut self, msg: &PortableSessionStarted) -> Result<OwnerAck> {
+        let tx = self.write_tx()?;
+        let Some(row) = read_owner_row(&tx, msg.session_id)? else {
+            return Ok(ack(msg.session_id, 0, AckOutcome::Rejected("unknown session".into()), 0, [0; 32]));
+        };
+        let current_hash = head_hash(&tx, row.character_id, row.head_revision)?;
+        let reject = |why: &str| ack(msg.session_id, 0, AckOutcome::Rejected(why.into()), row.head_revision, current_hash);
+        if msg.character_id != row.character_id || msg.server_id != row.server_id {
+            return Ok(reject("the message does not belong to this session"));
+        }
+        if msg.base_canonical_revision != row.c0_revision {
+            return Ok(reject("the baseline was taken against another canonical revision"));
+        }
+        let b0_hash = match msg.b0.hash() {
+            Ok(h) if hex::encode(h) == msg.content_hash => h,
+            _ => return Ok(reject("the content hash does not match the baseline")),
+        };
+        match row.state.as_str() {
+            "open" => {
+                return Ok(if row.b0_hash == Some(b0_hash) && row.generation == Some(msg.baseline_generation) {
+                    ack(msg.session_id, 0, AckOutcome::Duplicate, row.head_revision, current_hash)
+                } else {
+                    reject("this session already has a different baseline")
+                })
+            }
+            "offered" => {}
+            _ => return Ok(ack(msg.session_id, 0, AckOutcome::StaleSession, row.head_revision, current_hash)),
+        }
+        let b0 = match msg.b0.open() {
+            Ok(m) => m,
+            Err(e) => return Ok(reject(&format!("the baseline cannot be read: {e}"))),
+        };
+        let c0 = snapshot::decode(&row.c0_payload, Some(&row.c0_hash))?;
+        if b0.character_id != row.character_id || b0.ruleset != c0.ruleset || b0.content_namespace != c0.content_namespace {
+            return Ok(reject("the baseline is another character or ruleset"));
+        }
+        tx.execute(
+            "UPDATE owner_session SET state = 'open', baseline_generation = ?2, b0_hash = ?3, b0_payload = ?4, updated_at = ?5 WHERE session_id = ?1",
+            params![msg.session_id.to_string(), msg.baseline_generation, b0_hash.as_slice(), msg.b0.bytes()?, now()],
+        )?;
+        tx.commit()?;
+        Ok(ack(msg.session_id, 0, AckOutcome::Applied, row.head_revision, current_hash))
+    }
+
+    /// Apply one checkpoint of the realm: `merge3(C0, B0, B1)`.
+    pub fn owner_checkpoint(&mut self, msg: &PortableCheckpoint) -> Result<OwnerAck> {
+        let keep = self.history_keep()?;
+        let tx = self.write_tx()?;
+        let Some(row) = read_owner_row(&tx, msg.session_id)? else {
+            return Ok(ack(msg.session_id, msg.sequence, AckOutcome::Rejected("unknown session".into()), 0, [0; 32]));
+        };
+        let record = read_character(&tx, row.character_id)?;
+        let seen_hash = head_hash(&tx, row.character_id, record.revision)?;
+        let outcome = |o: AckOutcome| ack(msg.session_id, msg.sequence, o, record.revision, seen_hash);
+        let reject = |why: &str| outcome(AckOutcome::Rejected(why.into()));
+
+        if msg.character_id != row.character_id || msg.server_id != row.server_id {
+            return Ok(reject("the message does not belong to this session"));
+        }
+        if msg.base_canonical_revision != row.c0_revision {
+            return Ok(reject("the checkpoint was taken against another canonical revision"));
+        }
+        let b1_hash = match msg.realm_snapshot.hash() {
+            Ok(h) if hex::encode(h) == msg.content_hash => h,
+            _ => return Ok(reject("the content hash does not match the snapshot")),
+        };
+        if msg.sequence == 0 {
+            return Ok(reject("sequences start at 1"));
+        }
+
+        // a sequence the Owner has applied before is answered from its log, whatever the session's state is now
+        let logged: Option<(Vec<u8>, i64, i64, Option<String>)> = tx
+            .query_row("SELECT content_hash, resulting_revision, final, next_session_id FROM owner_checkpoint WHERE session_id = ?1 AND sequence = ?2", params![msg.session_id.to_string(), msg.sequence as i64], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))
+            .optional()?;
+        if let Some((hash, revision, was_final, next)) = logged {
+            if hash.as_slice() != b1_hash.as_slice() {
+                return Ok(reject("this sequence was applied with different content"));
+            }
+            let revision = revision as u64;
+            let mut a = ack(msg.session_id, msg.sequence, AckOutcome::Duplicate, revision, head_hash(&tx, row.character_id, revision)?);
+            if let Ok((model, _)) = read_snapshot_row(&tx, row.character_id, revision) {
+                (a.owned_items, a.owned_pets) = owned_ids(&model);
+                if was_final == 1 {
+                    a.canonical = Some(Envelope::seal(&model)?);
+                    if let Some(next) = next {
+                        a.next_session = Some(NextSession { session_id: next.parse()?, canonical_revision: revision });
+                    }
+                }
+            }
+            return Ok(a);
+        }
+        match row.state.as_str() {
+            "open" => {}
+            "offered" => return Ok(reject("the session has not started")),
+            _ => return Ok(outcome(AckOutcome::StaleSession)),
+        }
+        if msg.sequence < row.last_sequence {
+            return Ok(outcome(AckOutcome::StaleSequence));
+        }
+        if record.revision != row.head_revision {
+            tx.execute("UPDATE owner_session SET state = 'superseded', updated_at = ?2 WHERE session_id = ?1", params![msg.session_id.to_string(), now()])?;
+            tx.commit()?;
+            return Ok(ack(msg.session_id, msg.sequence, AckOutcome::StaleSession, record.revision, head_hash(&self.conn, row.character_id, record.revision)?));
+        }
+
+        let b1 = match msg.realm_snapshot.open() {
+            Ok(m) => m,
+            Err(e) => return Ok(reject(&format!("the snapshot cannot be read: {e}"))),
+        };
+        let (c0, b0) = (snapshot::decode(&row.c0_payload, Some(&row.c0_hash))?, snapshot::decode(row.b0_payload.as_deref().expect("an open session has B0"), row.b0_hash.as_ref())?);
+        if b1.character_id != row.character_id || b1.ruleset != c0.ruleset || b1.content_namespace != c0.content_namespace {
+            return Ok(reject("the snapshot is another character or ruleset"));
+        }
+        let merged = match merge3(&c0, &b0, &b1, Mode::Lenient) {
+            Ok(m) => m.model,
+            Err(e) => return Ok(reject(&format!("the checkpoint cannot be merged: {e}"))),
+        };
+        let encoded = match snapshot::encode(&merged) {
+            Ok(e) => e,
+            Err(e) => return Ok(reject(&format!("the merged character is not valid: {e}"))),
+        };
+
+        let at = now();
+        let head = head_hash(&tx, row.character_id, record.revision)?;
+        let revision = if head == encoded.content_hash {
+            record.revision
+        } else {
+            let revision = record.revision + 1;
+            insert_snapshot(&tx, row.character_id, revision, &encoded, &row.server_id, Some(&format!("session {} checkpoint {}", msg.session_id, msg.sequence)), &at)?;
+            update_character(&tx, row.character_id, &merged, revision, &at)?;
+            prune_tx(&tx, row.character_id, revision, keep)?;
+            revision
+        };
+
+        let mut a = ack(msg.session_id, msg.sequence, AckOutcome::Applied, revision, encoded.content_hash);
+        (a.owned_items, a.owned_pets) = owned_ids(&merged);
+        let mut next_id = None;
+        if msg.final_checkpoint {
+            tx.execute("UPDATE owner_session SET state = 'closed', last_sequence = ?2, head_revision = ?3, updated_at = ?4 WHERE session_id = ?1", params![msg.session_id.to_string(), msg.sequence as i64, revision as i64, at])?;
+            let offer = open_offer(&tx, row.character_id, &row.server_id)?;
+            a.canonical = Some(Envelope::from_encoded(&encoded));
+            a.next_session = Some(NextSession { session_id: offer.session_id, canonical_revision: revision });
+            next_id = Some(offer.session_id.to_string());
+        } else {
+            tx.execute("UPDATE owner_session SET last_sequence = ?2, head_revision = ?3, updated_at = ?4 WHERE session_id = ?1", params![msg.session_id.to_string(), msg.sequence as i64, revision as i64, at])?;
+        }
+        tx.execute(
+            "INSERT INTO owner_checkpoint(session_id, sequence, content_hash, resulting_revision, final, next_session_id, applied_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
+            params![msg.session_id.to_string(), msg.sequence as i64, b1_hash.as_slice(), revision as i64, msg.final_checkpoint as i64, next_id, at],
+        )?;
+        tx.commit()?;
+        Ok(a)
+    }
+
+    /// What the Owner knows of one session: (state, last applied sequence, head revision).
+    pub fn owner_session_info(&self, session: SessionId) -> Result<Option<(String, u64, u64)>> {
+        Ok(read_owner_row(&self.conn, session)?.map(|r| (r.state, r.last_sequence, r.head_revision)))
+    }
+}

crates/coa-core/src/portable/store/sync.rs

@@ -309,36 +309,10 @@ impl Store {
     /// = **filtered** (kept, mapping stays active); shown by the realm but not owned = **realm-local** (never merged).
     pub fn capture_baseline(&mut self, id: CharacterId, server_id: &str, input: BaselineInput<'_>) -> Result<Baseline> {
         check_server_id(server_id)?;
-        let c0 = self.synced_model(id, server_id)?.ok_or_else(|| PortableError::Invalid("the realm has no synchronised snapshot to use as the session base".into()))?;
-        if input.b0.character_id != id || input.b0.ruleset != c0.ruleset {
-            return Err(PortableError::WrongCharacter { expected: id, found: input.b0.character_id });
-        }
         let tx = self.write_tx()?;
-        let record = read_character(&tx, id)?;
-        let mapping_revision: i64 = tx.query_row("SELECT last_revision FROM character_server_mapping WHERE character_id = ?1 AND server_id = ?2", params![id.to_string(), server_id], |r| r.get(0))?;
-        if record.revision != mapping_revision as u64 {
-            return Err(PortableError::StaleRevision { expected: mapping_revision as u64, current: record.revision });
-        }
-        let open: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM realm_baseline WHERE character_id = ?1 AND server_id = ?2 AND state = 'open')", params![id.to_string(), server_id], |r| r.get(0))?;
-        if open {
-            return Err(PortableError::Invalid("this character already has an open session baseline on this realm".into()));
-        }
-        let c0_enc = snapshot::encode(&c0)?;
-        let b0_enc = snapshot::encode(input.b0)?;
-        let c0_items: HashSet<PortableItemId> = c0.items.iter().map(|i| i.id).collect();
-        let c0_pets: HashSet<PortablePetId> = c0.pets.iter().map(|p| p.id).collect();
-        let protection = Protection { realm_local_items: input.items.iter().map(|o| o.portable_item_id).filter(|i| !c0_items.contains(i)).collect(), canonical_items: c0_items };
-        reconcile_with(&tx, id, server_id, record.revision, input.items, &protection)?;
-        let pet_protection = PetProtection { realm_local_pets: input.pets.iter().map(|o| o.portable_pet_id).filter(|p| !c0_pets.contains(p)).collect(), canonical_pets: c0_pets };
-        sync_pets_in_tx(&tx, id, server_id, record.revision, input.pets, &pet_protection)?;
-        let at = now();
-        tx.execute(
-            "INSERT INTO realm_baseline(character_id, server_id, c0_revision, head_revision, state, c0_hash, c0_payload, b0_hash, b0_payload, created_at, updated_at)
-             VALUES (?1, ?2, ?3, ?3, 'open', ?4, ?5, ?6, ?7, ?8, ?8)",
-            params![id.to_string(), server_id, record.revision as i64, c0_enc.content_hash.as_slice(), c0_enc.payload, b0_enc.content_hash.as_slice(), b0_enc.payload, at],
-        )?;
+        let baseline = capture_baseline_in_tx(&tx, id, server_id, &input)?;
         tx.commit()?;
-        Ok(Baseline { character_id: id, server_id: server_id.to_string(), c0_revision: record.revision, head_revision: record.revision, c0, b0: input.b0.clone(), created_at: at })
+        Ok(baseline)
     }
 
     pub fn open_baseline(&self, id: CharacterId, server_id: &str) -> Result<Option<Baseline>> {
@@ -407,7 +381,52 @@ impl Store {
     }
 }
 
-fn read_open_baseline(conn: &Connection, id: CharacterId, server_id: &str) -> Result<Option<Baseline>> {
+pub(super) fn synced_model_in(conn: &Connection, id: CharacterId, server_id: &str) -> Result<Option<PortableCharacter>> {
+    let row: Option<(Option<Vec<u8>>, Option<Vec<u8>>)> = conn
+        .query_row("SELECT synced_hash, synced_payload FROM character_server_mapping WHERE character_id = ?1 AND server_id = ?2", params![id.to_string(), server_id], |r| Ok((r.get(0)?, r.get(1)?)))
+        .optional()?;
+    match row {
+        Some((Some(hash), Some(payload))) => Ok(Some(snapshot::decode(&payload, Some(&hash_from_blob(hash)?))?)),
+        Some(_) => Ok(None),
+        None => Err(PortableError::Invalid(format!("character {id} is not bound to server {server_id}"))),
+    }
+}
+
+/// Persist the baseline of a session inside a transaction: `C0` is the snapshot the realm was synchronised with, `B0` what the
+/// realm shows after its own normalisation. Items and pets the canonical character owns but the realm does not show are
+/// **filtered** (kept, mapping stays active); shown but not owned = **realm-local** (never merged).
+pub(super) fn capture_baseline_in_tx(tx: &Transaction<'_>, id: CharacterId, server_id: &str, input: &BaselineInput<'_>) -> Result<Baseline> {
+    let c0 = synced_model_in(tx, id, server_id)?.ok_or_else(|| PortableError::Invalid("the realm has no synchronised snapshot to use as the session base".into()))?;
+    if input.b0.character_id != id || input.b0.ruleset != c0.ruleset {
+        return Err(PortableError::WrongCharacter { expected: id, found: input.b0.character_id });
+    }
+    let record = read_character(tx, id)?;
+    let mapping_revision: i64 = tx.query_row("SELECT last_revision FROM character_server_mapping WHERE character_id = ?1 AND server_id = ?2", params![id.to_string(), server_id], |r| r.get(0))?;
+    if record.revision != mapping_revision as u64 {
+        return Err(PortableError::StaleRevision { expected: mapping_revision as u64, current: record.revision });
+    }
+    let open: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM realm_baseline WHERE character_id = ?1 AND server_id = ?2 AND state = 'open')", params![id.to_string(), server_id], |r| r.get(0))?;
+    if open {
+        return Err(PortableError::Invalid("this character already has an open session baseline on this realm".into()));
+    }
+    let c0_enc = snapshot::encode(&c0)?;
+    let b0_enc = snapshot::encode(input.b0)?;
+    let c0_items: HashSet<PortableItemId> = c0.items.iter().map(|i| i.id).collect();
+    let c0_pets: HashSet<PortablePetId> = c0.pets.iter().map(|p| p.id).collect();
+    let protection = Protection { realm_local_items: input.items.iter().map(|o| o.portable_item_id).filter(|i| !c0_items.contains(i)).collect(), canonical_items: c0_items };
+    reconcile_with(tx, id, server_id, record.revision, input.items, &protection)?;
+    let pet_protection = PetProtection { realm_local_pets: input.pets.iter().map(|o| o.portable_pet_id).filter(|p| !c0_pets.contains(p)).collect(), canonical_pets: c0_pets };
+    sync_pets_in_tx(tx, id, server_id, record.revision, input.pets, &pet_protection)?;
+    let at = now();
+    tx.execute(
+        "INSERT INTO realm_baseline(character_id, server_id, c0_revision, head_revision, state, c0_hash, c0_payload, b0_hash, b0_payload, created_at, updated_at)
+         VALUES (?1, ?2, ?3, ?3, 'open', ?4, ?5, ?6, ?7, ?8, ?8)",
+        params![id.to_string(), server_id, record.revision as i64, c0_enc.content_hash.as_slice(), c0_enc.payload, b0_enc.content_hash.as_slice(), b0_enc.payload, at],
+    )?;
+    Ok(Baseline { character_id: id, server_id: server_id.to_string(), c0_revision: record.revision, head_revision: record.revision, c0, b0: input.b0.clone(), created_at: at })
+}
+
+pub(super) fn read_open_baseline(conn: &Connection, id: CharacterId, server_id: &str) -> Result<Option<Baseline>> {
     type Row = (i64, i64, Vec<u8>, Vec<u8>, Vec<u8>, Vec<u8>, String);
     let row: Option<Row> = conn
         .query_row(

docs/PORTABLE_SESSIONS.md

@@ -0,0 +1,201 @@
+# Runtime portable sessions (Phase 5): design and audit
+
+Code: core `src/server/coa/CoAPortableSession*.{h,cpp}` and `CoAPortableImport*.{h,cpp}` (branch `feat/portable-session-bridge` of the
+core fork), Manager `crates/coa-core/src/portable/session/`. CoA only; Wildcard portable transfer stays disabled.
+
+Phase 4 proved the offline model with a real client. Its one fragile rule is that the baseline `B0` must be captured *after*
+the realm's own normalisation and *before* the character is played. Phase 5 makes that rule a property of the running
+server instead of a user procedure:
+
+```text
+running worldserver -> automatic B0 -> 60-second checkpoints -> Owner canonical revisions -> final logout checkpoint
+```
+
+## 1. Roles (the API reflects the real deployment)
+
+| | OWNER (player's Manager) | HOST (the realm's Manager) |
+|---|---|---|
+| owns | canonical `PortableCharacter`, revision history, `C0` per session, applied-checkpoint log | realm binding, item/pet mappings, synced snapshot of what the realm was given, outbox |
+| does | validates every message, runs `merge3(C0, B0, B1)`, writes revision N+1 | controls the local worldserver, captures `B0`/`B1` from the realm, emits messages |
+| never | touches a realm | opens or mutates the Owner's database; holds canonical history |
+
+Both roles run on `Store` (one schema) in this phase, but the APIs are `OwnerService` and `HostService`, they exchange only
+the messages of section 5, and every test runs them on two separate stores (two files, two processes' worth of state).
+A later Registry/Relay carries the same messages unchanged.
+
+## 2. Audit of the login path (what the core actually does)
+
+`WorldSession::HandlePlayerLoginFromDB` (`Handlers/CharacterHandler.cpp`) is one synchronous function on the world thread:
+
+1. `Player::LoadFromDB` (the realm's normalisation: unknown spells deleted, default spells/skills/reputation added, honor and title
+   resets, items that cannot be placed mailed or dropped);
+2. `SendInitialPacketsBeforeAddToMap`, `AddPlayerToMap`, `SendInitialPacketsAfterAddToMap`, social/guild/group announcements,
+   `LoadCorpse`, at-login flags (`resetSpells`, `resetTalents`, first-login spells), `LoadPet`;
+3. **`sScriptMgr->OnPlayerLogin`** and **`OnPlayerFirstLogin`** (module normalisation, e.g. the starter-kit restore) as the
+   *last* statements.
+
+`OnPlayerLogin` is therefore **not** early (it is the end of the function) but the end of the function **is** the right point:
+everything the realm does to a loading character has run, and nothing the player does can have run, because
+
+* client packets are handled in `WorldSession::Update`, which cannot run before this handler returns (same thread), and
+* the map's `Update` (mobs, auras, regeneration, `Player::Update`) runs later in the same world tick.
+
+So the character state at the end of the handler is exactly "normalised, not yet played". Bots enter through the same function
+(`BotMgr` calls `HandlePlayerLoginFromDB`), so the same code serves tests with bot-logged characters. The "login after
+disconnect" shortcut (`HandlePlayerLoginToCharInWorld`, a player that never left the world) does not reload the character and
+does not start a session.
+
+Other facts that shape the design:
+
+* `Player::SaveToDB(trans, create, logout)` is the single place that writes a character; manual save, autosave, logout save,
+  `.saveall` and checkpoints all go through it. It returns early (a delayed save) while `IsBeingTeleportedFar()`, and it saves the
+  pet through a **separate** transaction (`Pet::SavePetToDB`). With `CharacterDatabase.WorkerThreads = 1` (the default, and
+  enforced by the portable subsystem at startup) transactions commit in FIFO order, so the pet's commit precedes ours.
+* `CommitTransaction` is asynchronous: an RA reply is not persistence. The Host Manager reads the database and treats a save
+  as having happened only when it **sees the marker row**.
+* The save is a snapshot taken when `SaveToDB` runs, not when the commit lands. That is what makes `B0` exact.
+* RA/console commands run on the world thread (`QueueCliCommand`), so a command can safely touch a `Player*`.
+* The character list is stale on already connected clients after an import; `CharacterCache` and
+  `World::UpdateRealmCharCount` must be updated (the importer does, section 7).
+* `PlayerDumpReader::LoadDump` is the precedent for the importer's runtime invariants (live generators advanced only after the
+  commit, name check, cache entry, realm character count).
+
+## 3. Session marker (core table `coa_portable_session`)
+
+One row per portable character on the realm (pending SQL `data/sql/updates/pending_db_characters/`):
+
+| column | meaning |
+|---|---|
+| `guid` | local character guid (primary key) |
+| `session_id`, `character_id` | UUIDv7 strings: the session and the portable character |
+| `imported_revision` | canonical revision `C0` that the realm holds (`base_canonical_revision`) |
+| `baseline_generation` | increments each time the session is (re)armed |
+| `state` | 0 `waiting_baseline`, 1 `baseline_ready` (gated), 2 `active`, 3 `ended` |
+| `checkpoint_seq` | the last checkpoint sequence the core saved |
+| `save_seq` | incremented by **every** save of this character |
+| `updated_at` | unix time |
+
+The Host arms a session (state 0) in the same realm transaction that imports/updates the character (offline importer, in-place
+update) or the core's import service does (online). That is the "pending portable-session marker" the character carries until
+its first load.
+
+### Automatic B0
+
+At the end of `HandlePlayerLoginFromDB`, if the character's row is in state 0, the core, in **one** transaction:
+`SaveToDB(trans, false, false)` + `UPDATE coa_portable_session SET state=1, baseline_generation=..., save_seq=save_seq+1`,
+commits, and **gates** the session:
+
+* while gated the session's packet queue is not processed (the client stays connected; `CMSG_PING` is answered by the socket
+  layer), the character is made non-attackable and rooted, nothing the client sends is lost - it is processed after release;
+* the gate ends when the Host sends `portable release <session_id>` (the Host has seen `state=1`, exported `B0` in one read-only
+  consistent snapshot, verified inside that snapshot that the marker is still the baseline's, and sent `PortableSessionStarted`);
+* if no release arrives within `PortableSession.GateTimeoutSeconds` (default 30) the character is **kicked**, never released:
+  no gameplay may happen before `B0` exists. The row stays in state 1; the next login re-runs the baseline.
+
+Characters whose row is in state 2 (a worldserver restart or a relog in the middle of a session) get no baseline, only the
+in-memory registration; `B0` and `C0` are already held by the Owner.
+
+### Every save carries a marker
+
+`Player::SaveToDB(trans, ...)` appends, for a registered portable character, `UPDATE coa_portable_session SET
+save_seq=save_seq+1, updated_at=...` to the **same** transaction (plus `checkpoint_seq=N` when the save is a checkpoint and
+`state=3` when it is the logout save). So the rows of a character in a read snapshot always belong to exactly the save whose
+marker is in that snapshot; an autosave between two checkpoints is visible as a higher `save_seq`.
+
+### Checkpoint (typed core operation)
+
+`portable checkpoint <local_guid> <session_id> <sequence>` (console/RA only): the character is addressed by guid, never by name;
+the command verifies the player is online, a portable session is registered, `session_id` matches, the state is 2, the
+sequence is greater than `checkpoint_seq`, and the character is not mid-teleport (`BUSY`, retry). Then **one** `SaveToDB`
+transaction with the checkpoint marker, committed. The reply is `QUEUED <sequence>`; it is **not** an acknowledgement of
+persistence. `.saveall` is never used, so one checkpoint writes one character (verified, section 9).
+
+Logout: the logout save itself carries `state=3`; the Host exports and sends the final checkpoint.
+
+## 4. Host flow
+
+```text
+import / update of a portable character  ->  arm session (state 0), outbox: nothing yet
+poll realm every ~1 s (cheap indexed read of coa_portable_session):
+  state 1, no PortableSessionStarted sent  ->  read B0 (consistent snapshot, marker re-checked) -> persist -> release (RA)
+                                              -> emit PortableSessionStarted
+  state 2, online, every 60 s              ->  RA checkpoint(seq) -> wait for checkpoint_seq == seq -> read B1 ->
+                                              persist in outbox -> emit PortableCheckpoint(seq)
+  state 3 (logout) or character offline    ->  read final B1 -> emit PortableCheckpoint(final) -> on ack: re-arm (state 0)
+```
+
+A failed intermediate checkpoint changes nothing canonical and is retried at the next tick. The sequence number is persisted
+**before** the RA call, so a Host restart resumes with the same number (the marker tells whether the save happened).
+Messages wait in the outbox until acknowledged; redelivery is harmless (section 6).
+
+## 5. Messages (transport-neutral, versioned, JSON)
+
+`protocol_version = 1`; each message is canonical JSON with unknown fields refused. Snapshots travel in the existing
+snapshot envelope (zstd of canonical JSON, SHA-256 of the canonical JSON).
+
+```text
+PortableSessionStarted { protocol_version, session_id, character_id, server_id, base_canonical_revision,
+                         baseline_generation, b0: envelope, content_hash }
+PortableCheckpoint     { protocol_version, session_id, character_id, server_id, base_canonical_revision,
+                         sequence, final_checkpoint, realm_snapshot: envelope, content_hash }
+OwnerAck               { protocol_version, session_id, sequence, outcome, canonical_revision, canonical_hash,
+                         [final: canonical snapshot envelope] }
+```
+
+`outcome` is `applied`, `duplicate` (same sequence and hash, already applied), `stale_sequence`, `stale_session`
+(superseded: another session or revision), or `rejected(reason)`.
+
+The Owner validates, **before** `merge3`: protocol version; message size and decompression limits; `content_hash` against the
+payload; `character_id` exists and the ruleset is CoA; the session is known (opened by `PortableSessionStarted` whose
+`base_canonical_revision` equals the revision the Owner recorded as `C0` for that `server_id`); `server_id` and
+`character_id` match the session; the payload's own `character_id`/ruleset match; the sequence.
+
+## 6. Ordering, idempotency, no double counting
+
+* Per session the Owner keeps `last_sequence` and a log of `(sequence, content_hash, resulting_revision)`.
+* sequence = `last_sequence + 1`: applied. Same sequence with the same hash: `duplicate`, nothing happens, the original result
+  is returned (lost acknowledgements are healed by redelivery). Same sequence, different hash: `rejected`.
+  `sequence <= last_sequence` otherwise: `stale_sequence`, ignored. A gap (`sequence > last_sequence + 1`) is accepted only when
+  the Owner holds a later state (checkpoints are snapshots, not deltas, so skipping one loses nothing).
+* **Every checkpoint is `merge3(C0, B0, current B1)`**, anchored to the session's original `C0` and `B0` (Phase 4's
+  property), never `merge(previous canonical, previous B1, B1)`. A repeated or re-sent state therefore cannot add money, honor or xp
+  twice; the canonical revision can advance while the anchors stay.
+* A superseded session (the character's canonical revision no longer equals the session head) refuses checkpoints
+  (`stale_session`); no divergent-state merging is invented here.
+* Crash: the last acknowledged checkpoint is canonical. A worldserver crash leaves the row in state 2 and the character offline:
+  the Host marks nothing, the next login continues the same session (no new baseline). A final export after a crash is a
+  normal offline export of the last committed save.
+
+## 7. PortableImportService (production import without stopping the realm)
+
+Follows `PORTABLE_IMPORT_SERVICE.md` with one concrete decision: the **job file is canonical JSON, not zstd**, because the core
+has no zstd or JSON dependency and a hand-rolled bounded strict parser is easier to audit than adding either. The size cap
+(16 MiB, as the Manager's decompression cap) is enforced before parsing; the header carries the SHA-256 the file must match.
+
+```text
+Manager writes <JobDir>/<job_id>.job (temp name + rename)
+-> RA: portable import <job_id>        (job_id = UUIDv7, nothing else)
+-> world thread: validate -> allocate from live generators -> prepared statements -> ONE CharacterDatabaseTransaction
+   (character, items, spells, ..., session marker in state 0) -> commit
+-> after the commit only: CharacterCache entry, generators advanced, UpdateRealmCharCount
+-> <JobDir>/<job_id>.result (temp + rename); RA reply "OK <job_id>" or "ERR <code>"
+```
+
+No path, SQL or snapshot travels through RA. Idempotency: the marker row carries the job nonce; re-running a finished job
+rewrites the same result. Never touches a character that is online or already exists.
+
+## 8. What is not in this phase
+
+VPS Registry/Relay and any network transport, public server list, remote account provisioning, Wildcard, vanity/wardrobe
+transfer, personal Ascension bank, equipment sets, level-cap projection.
+
+## 9. Verification plan
+
+* Owner/Host on two stores: canonical revision 10 -> running realm -> automatic `B0` before gameplay -> play -> checkpoint 1
+  (rev 11) -> same checkpoint again (still 11) -> more play -> checkpoint 2 (rev 12) -> delayed checkpoint 1 refused ->
+  final checkpoint.
+* Money/item/spell/quest/pet progression; realm normalisation not propagated; retry; lost acknowledgement; Host, Owner and
+  worldserver restart; stale session; wrong character or session; malformed message; over-limit payload; logout during a
+  checkpoint; disconnect/crash before the final checkpoint.
+* Real worldserver with the character online and several checkpoints without logout; the number of `characters` rows written
+  during a checkpoint is counted on the database (general log) to show one character is saved, not all.