From c6aceb6524672b3f59a1a151a7124f17ae865ca2 Mon Sep 17 00:00:00 2001 From: matthewevans Date: Wed, 12 Aug 2026 18:36:55 -0700 Subject: [PATCH 1/8] fix(native): settle resolve all and reconnect identity --- .../src/adapter/__tests__/ws-adapter.test.ts | 150 +++++++++-- client/src/adapter/ws-adapter.ts | 59 +++-- crates/phase-server/src/main.rs | 249 +++++++++++++----- 3 files changed, 342 insertions(+), 116 deletions(-) diff --git a/client/src/adapter/__tests__/ws-adapter.test.ts b/client/src/adapter/__tests__/ws-adapter.test.ts index c1dc50c9a1..cbc84a082c 100644 --- a/client/src/adapter/__tests__/ws-adapter.test.ts +++ b/client/src/adapter/__tests__/ws-adapter.test.ts @@ -599,6 +599,26 @@ describe("WebSocketAdapter", () => { }); describe("native P2P pregame transport", () => { + const nativeReconnectAdapter = () => new WebSocketAdapter( + "native-engine", + "join", + { main_deck: [], sideboard: [] }, + undefined, + undefined, + undefined, + "Guest", + { + nativePregame: { + kind: "reconnect", + socketFactory: () => new MockWebSocket("native-engine") as unknown as PhaseSocketTransport, + gameCode: "NATIVE", + playerId: 1, + playerToken: "guest-token", + fullKey: { game_code: "NATIVE", generation: 1 }, + }, + }, + ); + it("rejects a native seat attachment without a Full session key", async () => { const nativeAdapter = new WebSocketAdapter( "native-engine", @@ -683,25 +703,7 @@ describe("WebSocketAdapter", () => { }); it("reconnects a persisted native viewer with its expected seat", async () => { - const nativeAdapter = new WebSocketAdapter( - "native-engine", - "join", - { main_deck: [], sideboard: [] }, - undefined, - undefined, - undefined, - "Guest", - { - nativePregame: { - kind: "reconnect", - socketFactory: () => new MockWebSocket("native-engine") as unknown as PhaseSocketTransport, - gameCode: "NATIVE", - playerId: 1, - playerToken: "guest-token", - fullKey: { game_code: "NATIVE", generation: 1 }, - }, - }, - ); + const nativeAdapter = nativeReconnectAdapter(); const attached = nativeAdapter.initializePregame(); const nativeSocket = await completeHandshake(nativeAdapter); @@ -719,7 +721,12 @@ describe("WebSocketAdapter", () => { "message", JSON.stringify({ type: "GameStarted", - data: { state_revision: 7, state: createMockState(), your_player: 1 }, + data: { + state_revision: 7, + state: createMockState(), + your_player: 1, + full_key: { game_code: "NATIVE", generation: 1 }, + }, }), ); await expect(attached).resolves.toEqual({ @@ -730,6 +737,109 @@ describe("WebSocketAdapter", () => { }); }); + it("seeds a native reconnect before direct initialize attaches the socket", async () => { + const nativeAdapter = nativeReconnectAdapter(); + const initialized = nativeAdapter.initialize(); + const nativeSocket = await completeHandshake(nativeAdapter); + + nativeSocket.dispatchSynthetic( + "message", + JSON.stringify({ + type: "GameStarted", + data: { + state_revision: 7, + state: createMockState(), + your_player: 1, + full_key: { game_code: "NATIVE", generation: 1 }, + }, + }), + ); + + await expect(initialized).resolves.toBeUndefined(); + expect(nativeAdapter.nativeSession).toEqual({ + gameCode: "NATIVE", + playerId: 1, + playerToken: "guest-token", + fullKey: { game_code: "NATIVE", generation: 1 }, + }); + }); + + it("rejects a native reconnect GameStarted for a different player before caching it", async () => { + const nativeAdapter = nativeReconnectAdapter(); + const listener = vi.fn(); + nativeAdapter.onEvent(listener); + const attached = nativeAdapter.initializePregame(); + const nativeSocket = await completeHandshake(nativeAdapter); + listener.mockClear(); + + nativeSocket.dispatchSynthetic( + "message", + JSON.stringify({ + type: "GameStarted", + data: { + state_revision: 7, + state: createMockState(), + your_player: 0, + full_key: { game_code: "NATIVE", generation: 1 }, + }, + }), + ); + + await expect(attached).rejects.toThrow("Native reconnect attached player 0, expected 1"); + await expect(nativeAdapter.getSnapshot()).rejects.toThrow("No game state available"); + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: "error" })); + }); + + it("rejects a native reconnect GameStarted with a changed Full session key before caching it", async () => { + const nativeAdapter = nativeReconnectAdapter(); + const listener = vi.fn(); + nativeAdapter.onEvent(listener); + const attached = nativeAdapter.initializePregame(); + const nativeSocket = await completeHandshake(nativeAdapter); + listener.mockClear(); + + nativeSocket.dispatchSynthetic( + "message", + JSON.stringify({ + type: "GameStarted", + data: { + state_revision: 7, + state: createMockState(), + your_player: 1, + full_key: { game_code: "NATIVE", generation: 2 }, + }, + }), + ); + + await expect(attached).rejects.toThrow("Server changed the Full session identity"); + await expect(nativeAdapter.getSnapshot()).rejects.toThrow("No game state available"); + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: "error" })); + }); + + it("rejects a native reconnect GameStarted without a Full session key before caching it", async () => { + const nativeAdapter = nativeReconnectAdapter(); + const listener = vi.fn(); + nativeAdapter.onEvent(listener); + const attached = nativeAdapter.initializePregame(); + const nativeSocket = await completeHandshake(nativeAdapter); + listener.mockClear(); + + nativeSocket.dispatchSynthetic( + "message", + JSON.stringify({ + type: "GameStarted", + data: { state_revision: 7, state: createMockState(), your_player: 1 }, + }), + ); + + await expect(attached).rejects.toThrow("Server omitted a valid Full session identity"); + await expect(nativeAdapter.getSnapshot()).rejects.toThrow("No game state available"); + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: "error" })); + }); + it("rejects native pregame attachment when the server returns an error", async () => { const nativeAdapter = new WebSocketAdapter( "native-engine", diff --git a/client/src/adapter/ws-adapter.ts b/client/src/adapter/ws-adapter.ts index b74a1f155d..b2000c837a 100644 --- a/client/src/adapter/ws-adapter.ts +++ b/client/src/adapter/ws-adapter.ts @@ -529,6 +529,7 @@ export class WebSocketAdapter implements EngineAdapter { return; } + this.seedNativeReconnectSession(); const setupFrame = this.options.nativeAi ? this.nativeAiSetupFrame(this.options.nativeAi) @@ -564,12 +565,7 @@ export class WebSocketAdapter implements EngineAdapter { if (!options) { throw new AdapterError("WS_ERROR", "Pregame initialization requires a native socket", false); } - if (options.kind === "reconnect") { - this._gameCode = options.gameCode; - this._playerId = options.playerId; - this.playerToken = options.playerToken; - this.fullSessionKey = options.fullKey; - } + this.seedNativeReconnectSession(); return new Promise((resolve, reject) => { this.pregameResolve = resolve; this.pregameReject = reject; @@ -1118,6 +1114,18 @@ export class WebSocketAdapter implements EngineAdapter { }; } + /** Seeds persisted native credentials before either initialization path + * attaches the socket, so its first reconnect response can be authenticated. */ + private seedNativeReconnectSession(): void { + const options = this.options.nativePregame; + if (options?.kind !== "reconnect") return; + + this._gameCode = options.gameCode; + this._playerId = options.playerId; + this.playerToken = options.playerToken; + this.fullSessionKey = options.fullKey; + } + private nativeSocketOptions(): NativeSocketAdapterOptions | null { return this.options.nativeAi ?? this.options.nativePregame ?? null; } @@ -1324,6 +1332,22 @@ export class WebSocketAdapter implements EngineAdapter { case "GameStarted": { const data = msg.data as { state_revision: number; state: GameState; your_player: PlayerId; opponent_name?: string; player_names?: string[]; legal_actions?: GameAction[]; auto_pass_recommended?: boolean; end_continuous_effect_offers?: LegalActionsResult["endContinuousEffectOffers"]; mana_payment_shortcut_actions?: GameAction[]; spell_costs?: Record; legal_actions_by_object?: Record; viewer_interaction?: LegalActionsResult["viewerInteraction"]; derived?: GameState["derived"]; player_token?: string; full_key?: FullSessionKey; events?: GameEvent[]; rewind_targets?: RewindOption[] }; + const nativeReconnect = this.options.nativePregame?.kind === "reconnect" + ? this.options.nativePregame + : null; + if (nativeReconnect) { + if (data.your_player !== nativeReconnect.playerId) { + const error = new AdapterError( + "WS_ERROR", + `Native reconnect attached player ${data.your_player}, expected ${nativeReconnect.playerId}`, + false, + ); + this.rejectInitialization(error); + this.emit({ type: "error", message: error.message }); + break; + } + if (!this.acceptFullSessionKey(data.full_key)) break; + } if (this.reconnectInFlight) { this.reconnectInFlight = false; this.reconnectAttempt = 0; @@ -1348,23 +1372,12 @@ export class WebSocketAdapter implements EngineAdapter { }, ); this._playerId = data.your_player; - if (this.options.nativePregame?.kind === "reconnect") { - const expected = this.options.nativePregame; - if (data.your_player !== expected.playerId) { - const error = new AdapterError( - "WS_ERROR", - `Native reconnect attached player ${data.your_player}, expected ${expected.playerId}`, - false, - ); - this.rejectInitialization(error); - this.emit({ type: "error", message: error.message }); - break; - } + if (nativeReconnect) { const attachment: NativeSessionAttachment = { - gameCode: expected.gameCode, - playerId: expected.playerId, - playerToken: expected.playerToken, - fullKey: expected.fullKey, + gameCode: nativeReconnect.gameCode, + playerId: nativeReconnect.playerId, + playerToken: nativeReconnect.playerToken, + fullKey: nativeReconnect.fullKey, }; this.emit({ type: "sessionChanged", session: this.currentSession() }); this.emit({ type: "sessionAttached", attachment }); @@ -1378,7 +1391,7 @@ export class WebSocketAdapter implements EngineAdapter { if (!this._gameCode && this.joinGameCode) { this._gameCode = this.joinGameCode; } - if (data.full_key && !this.acceptFullSessionKey(data.full_key)) break; + if (!nativeReconnect && data.full_key && !this.acceptFullSessionKey(data.full_key)) break; if (data.player_token) { this.playerToken = data.player_token; this.emit({ type: "sessionChanged", session: this.currentSession() }); diff --git a/crates/phase-server/src/main.rs b/crates/phase-server/src/main.rs index e2cb2b3737..eb71732d6e 100644 --- a/crates/phase-server/src/main.rs +++ b/crates/phase-server/src/main.rs @@ -502,6 +502,38 @@ fn build_state_update_message( /// state; the requester acknowledgement carries progress metadata separately. const MAX_RESOLVE_ALL_LOG_ENTRIES: usize = 128; +/// Resolving the batch and then resuming normal AI play are one authoritative +/// transition. Retain their engine-authored logs in that order while keeping +/// the compact final snapshot bounded. +fn resolve_all_log_tail( + batch_log_entries: &[GameLogEntry], + ai_results: &[RevisionedActionResult], +) -> Vec { + fn append_tail(tail: &mut Vec, entries: &[GameLogEntry]) { + if entries.len() >= MAX_RESOLVE_ALL_LOG_ENTRIES { + tail.clear(); + tail.extend_from_slice(&entries[entries.len() - MAX_RESOLVE_ALL_LOG_ENTRIES..]); + return; + } + + let overflow = tail + .len() + .saturating_add(entries.len()) + .saturating_sub(MAX_RESOLVE_ALL_LOG_ENTRIES); + if overflow > 0 { + tail.drain(..overflow); + } + tail.extend_from_slice(entries); + } + + let mut tail = Vec::with_capacity(MAX_RESOLVE_ALL_LOG_ENTRIES); + append_tail(&mut tail, batch_log_entries); + for (_, (_, _, _, log_entries, _, _, _)) in ai_results { + append_tail(&mut tail, log_entries); + } + tail +} + #[allow(clippy::too_many_arguments)] fn build_resolve_all_state_update_message( raw_state: &GameState, @@ -4206,82 +4238,110 @@ async fn handle_resolve_all( let processed = { let mut mgr = state.lock().await; match mgr.resolve_all_for_player(&game_code, &player_token, max_resolutions) { - Ok((transition, summary)) => { - let session = mgr - .sessions - .get(&game_code) - .expect("Resolve All retains its session"); - let eliminated = session.state.eliminated_players.clone(); - let rewind_targets = session.rewind_options(); - let player_count = session.player_count; - let game_over_winner = match &session.state.waiting_for { - engine::types::game_state::WaitingFor::GameOver { winner } => Some(*winner), - _ => None, - }; - let terminal = if let Some(winner) = game_over_winner { - let ranked_result = ranked_duel_players(session).and_then(|players| { - ranked_result_for_duel(game_db, &game_code, &players, winner) - }); - terminal_artifact(session, winner, "Game ended".to_string(), ranked_result) - .map(Some) - } else if transition.is_some() { - persist_full_session_async(game_db, session); - Ok(None) - } else { - Ok(None) - }; - terminal.map(|terminal| { - ( - transition, - summary, - eliminated, - rewind_targets, - player_count, - game_over_winner, - terminal, - ) - }) - } + Ok((transition, summary)) => match transition { + None => Ok((summary, None)), + Some((_, (_, _, _, batch_log_entries, _, _, _))) => { + let session = mgr + .sessions + .get_mut(&game_code) + .expect("Resolve All retains its session"); + // Resolve All is a shortcut through a human-authorized batch, + // not a replacement for the session's ordinary AI hand-off. + // Keep that hand-off under the same lock, then derive the one + // final payload from the current session rather than the batch + // transition it has already moved past. + let ai_results = session.run_ai(); + let (raw_state, legal_actions, _auto_pass, spell_costs, by_object) = + session.current_broadcast_snapshot(); + let revision = session.state_revision; + let log_entries = resolve_all_log_tail(&batch_log_entries, &ai_results); + let eliminated = session.state.eliminated_players.clone(); + let rewind_targets = session.rewind_options(); + let player_count = session.player_count; + let game_over_winner = match &session.state.waiting_for { + engine::types::game_state::WaitingFor::GameOver { winner } => Some(*winner), + _ => None, + }; + let terminal = if let Some(winner) = game_over_winner { + let ranked_result = ranked_duel_players(session).and_then(|players| { + ranked_result_for_duel(game_db, &game_code, &players, winner) + }); + terminal_artifact(session, winner, "Game ended".to_string(), ranked_result) + .map(Some) + } else { + persist_full_session_async(game_db, session); + Ok(None) + }; + terminal.map(|terminal| { + ( + summary, + Some(( + revision, + raw_state, + legal_actions, + log_entries, + spell_costs, + by_object, + eliminated, + rewind_targets, + player_count, + game_over_winner, + terminal, + )), + ) + }) + } + }, Err(error) => Err(error), } }; - let (transition, summary, eliminated, rewind_targets, player_count, game_over_winner, terminal) = - match processed { - Ok(processed) => processed, - Err(reason) => { - let _ = tx.send(ServerMessage::ResolveAllRejected { request_id, reason }); - return; - } - }; + let (summary, payload) = match processed { + Ok(processed) => processed, + Err(reason) => { + let _ = tx.send(ServerMessage::ResolveAllRejected { request_id, reason }); + return; + } + }; let acknowledgement = ServerMessage::ResolveAllResult { request_id, items_resolved: summary.items_resolved, total: summary.total, }; - let Some((revision, result)) = transition else { + let Some(( + revision, + raw_state, + legal_actions, + log_entries, + spell_costs, + by_object, + eliminated, + rewind_targets, + player_count, + game_over_winner, + terminal, + )) = payload + else { let _ = tx.send(acknowledgement); return; }; - let (raw_state, _events, legal_actions, logs, _auto_pass, spell_costs, by_object) = &result; - let log_entries = &logs[logs.len().saturating_sub(MAX_RESOLVE_ALL_LOG_ENTRIES)..]; if let Err(reason) = guard_state_snapshot_broadcast(StateSnapshotParts { - state: raw_state, + state: &raw_state, events: &[], - log_entries, - legal_actions, - legal_actions_by_object: by_object, - spell_costs, + log_entries: &log_entries, + legal_actions: &legal_actions, + legal_actions_by_object: &by_object, + spell_costs: &spell_costs, }) { warn!(game = %game_code, %reason, "Resolve All snapshot exceeds broadcast bounds after commit"); let _ = tx.send(build_resolve_all_state_update_message( - raw_state, - log_entries, - legal_actions, - spell_costs, - by_object, + &raw_state, + &log_entries, + &legal_actions, + &spell_costs, + &by_object, revision, requester, eliminated.clone(), @@ -4297,11 +4357,11 @@ async fn handle_resolve_all( Err(error) => { error!(game = %game_code, %error, "Resolve All terminal preparation failed after commit"); let _ = tx.send(build_resolve_all_state_update_message( - raw_state, - log_entries, - legal_actions, - spell_costs, - by_object, + &raw_state, + &log_entries, + &legal_actions, + &spell_costs, + &by_object, revision, requester, eliminated.clone(), @@ -4317,11 +4377,11 @@ async fn handle_resolve_all( // Queue the requester's final state and acknowledgement through its direct // sender in order; the adapter resolves only after this cached snapshot. let requester_update = build_resolve_all_state_update_message( - raw_state, - log_entries, - legal_actions, - spell_costs, - by_object, + &raw_state, + &log_entries, + &legal_actions, + &spell_costs, + &by_object, revision, requester, eliminated.clone(), @@ -4339,11 +4399,11 @@ async fn handle_resolve_all( } if let Some(sender) = players.get(&player) { let _ = sender.send(build_resolve_all_state_update_message( - raw_state, - log_entries, - legal_actions, - spell_costs, - by_object, + &raw_state, + &log_entries, + &legal_actions, + &spell_costs, + &by_object, revision, player, eliminated.clone(), @@ -4354,7 +4414,7 @@ async fn handle_resolve_all( } } if let Ok(spectator_update) = - build_spectator_state_update_message(raw_state, &[], log_entries, revision) + build_spectator_state_update_message(&raw_state, &[], &log_entries, revision) { let mut spectators = game_spectators.lock().await; if let Some(senders) = spectators.get_mut(&game_code) { @@ -7780,6 +7840,49 @@ mod state_transport_derived_tests { } } + #[test] + fn resolve_all_final_log_tail_orders_batch_before_ai_follow_up_logs() { + let state = GameState::new_two_player(42); + let batch_logs: Vec<_> = (0..=MAX_RESOLVE_ALL_LOG_ENTRIES) + .map(|seq| GameLogEntry { + seq: seq as u32, + turn: 1, + phase: Phase::PreCombatMain, + category: LogCategory::Game, + segments: vec![LogSegment::Text(format!("batch {seq}"))], + presentation: Default::default(), + }) + .collect(); + let ai_logs: Vec<_> = (0..2) + .map(|seq| GameLogEntry { + seq: (100 + seq) as u32, + turn: 1, + phase: Phase::PreCombatMain, + category: LogCategory::Game, + segments: vec![LogSegment::Text(format!("ai {seq}"))], + presentation: Default::default(), + }) + .collect(); + let ai_results = vec![( + 2, + ( + state, + Vec::new(), + Vec::new(), + ai_logs.clone(), + false, + HashMap::new(), + HashMap::new(), + ), + )]; + + let tail = resolve_all_log_tail(&batch_logs, &ai_results); + + assert_eq!(tail.len(), MAX_RESOLVE_ALL_LOG_ENTRIES); + assert_eq!(tail.first(), batch_logs.get(3)); + assert_eq!(&tail[tail.len() - ai_logs.len()..], ai_logs.as_slice()); + } + #[test] fn turn_controller_receives_low_use_window_recommendation_instead_of_controlled_seat() { let controlled = PlayerId(0); From 928547ce96f7cec815a770a4591c6093ead0b6c4 Mon Sep 17 00:00:00 2001 From: matthewevans Date: Wed, 12 Aug 2026 18:48:28 -0700 Subject: [PATCH 2/8] fix(native): validate reconnect transition frames --- .../src/adapter/__tests__/ws-adapter.test.ts | 100 +++++++++++++ client/src/adapter/ws-adapter.ts | 60 ++++++-- crates/phase-server/src/main.rs | 135 +++++++++++++++++- 3 files changed, 282 insertions(+), 13 deletions(-) diff --git a/client/src/adapter/__tests__/ws-adapter.test.ts b/client/src/adapter/__tests__/ws-adapter.test.ts index cbc84a082c..71bfcc2b9f 100644 --- a/client/src/adapter/__tests__/ws-adapter.test.ts +++ b/client/src/adapter/__tests__/ws-adapter.test.ts @@ -725,6 +725,7 @@ describe("WebSocketAdapter", () => { state_revision: 7, state: createMockState(), your_player: 1, + player_token: "guest-token", full_key: { game_code: "NATIVE", generation: 1 }, }, }), @@ -737,6 +738,71 @@ describe("WebSocketAdapter", () => { }); }); + it("rejects a hostile GameCreated before it can replace native reconnect credentials", async () => { + const nativeAdapter = nativeReconnectAdapter(); + const listener = vi.fn(); + nativeAdapter.onEvent(listener); + const attached = nativeAdapter.initializePregame(); + const nativeSocket = await completeHandshake(nativeAdapter); + listener.mockClear(); + + nativeSocket.dispatchSynthetic( + "message", + JSON.stringify({ + type: "GameCreated", + data: { + game_code: "ATTACK", + player_token: "attacker-token", + full_key: { game_code: "ATTACK", generation: 9 }, + }, + }), + ); + + await expect(attached).rejects.toThrow("Native reconnect attached game ATTACK, expected NATIVE"); + await expect(nativeAdapter.getSnapshot()).rejects.toThrow("No game state available"); + expect(nativeAdapter.nativeSession).toEqual({ + gameCode: "NATIVE", + playerId: 1, + playerToken: "guest-token", + fullKey: { game_code: "NATIVE", generation: 1 }, + }); + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: "error" })); + }); + + it("rejects a hostile SessionAttached before it can settle native reconnect", async () => { + const nativeAdapter = nativeReconnectAdapter(); + const listener = vi.fn(); + nativeAdapter.onEvent(listener); + const attached = nativeAdapter.initializePregame(); + const nativeSocket = await completeHandshake(nativeAdapter); + listener.mockClear(); + + nativeSocket.dispatchSynthetic( + "message", + JSON.stringify({ + type: "SessionAttached", + data: { + game_code: "NATIVE", + player_id: 1, + player_token: "attacker-token", + full_key: { game_code: "NATIVE", generation: 1 }, + }, + }), + ); + + await expect(attached).rejects.toThrow("Native reconnect changed the player token"); + await expect(nativeAdapter.getSnapshot()).rejects.toThrow("No game state available"); + expect(nativeAdapter.nativeSession).toEqual({ + gameCode: "NATIVE", + playerId: 1, + playerToken: "guest-token", + fullKey: { game_code: "NATIVE", generation: 1 }, + }); + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: "error" })); + }); + it("seeds a native reconnect before direct initialize attaches the socket", async () => { const nativeAdapter = nativeReconnectAdapter(); const initialized = nativeAdapter.initialize(); @@ -840,6 +906,40 @@ describe("WebSocketAdapter", () => { expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: "error" })); }); + it("rejects a native reconnect GameStarted token before caching or attaching", async () => { + const nativeAdapter = nativeReconnectAdapter(); + const listener = vi.fn(); + nativeAdapter.onEvent(listener); + const attached = nativeAdapter.initializePregame(); + const nativeSocket = await completeHandshake(nativeAdapter); + listener.mockClear(); + + nativeSocket.dispatchSynthetic( + "message", + JSON.stringify({ + type: "GameStarted", + data: { + state_revision: 7, + state: createMockState(), + your_player: 1, + player_token: "attacker-token", + full_key: { game_code: "NATIVE", generation: 1 }, + }, + }), + ); + + await expect(attached).rejects.toThrow("Native reconnect changed the player token"); + await expect(nativeAdapter.getSnapshot()).rejects.toThrow("No game state available"); + expect(nativeAdapter.nativeSession).toEqual({ + gameCode: "NATIVE", + playerId: 1, + playerToken: "guest-token", + fullKey: { game_code: "NATIVE", generation: 1 }, + }); + expect(listener).toHaveBeenCalledTimes(1); + expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: "error" })); + }); + it("rejects native pregame attachment when the server returns an error", async () => { const nativeAdapter = new WebSocketAdapter( "native-engine", diff --git a/client/src/adapter/ws-adapter.ts b/client/src/adapter/ws-adapter.ts index b2000c837a..9a8ec1c259 100644 --- a/client/src/adapter/ws-adapter.ts +++ b/client/src/adapter/ws-adapter.ts @@ -1223,6 +1223,11 @@ export class WebSocketAdapter implements EngineAdapter { player_token: string; full_key?: FullSessionKey; }; + if (!this.acceptNativeReconnectIdentity({ + gameCode: data.game_code, + playerToken: data.player_token, + fullKey: data.full_key, + })) break; this._gameCode = data.game_code; this.playerToken = data.player_token; if (data.full_key && !this.acceptFullSessionKey(data.full_key)) break; @@ -1240,6 +1245,12 @@ export class WebSocketAdapter implements EngineAdapter { player_token: string; full_key?: FullSessionKey; }; + if (!this.acceptNativeReconnectIdentity({ + gameCode: data.game_code, + playerId: data.player_id, + playerToken: data.player_token, + fullKey: data.full_key, + })) break; this._gameCode = data.game_code; const fullKey = data.full_key; if (!fullKey) { @@ -1335,17 +1346,12 @@ export class WebSocketAdapter implements EngineAdapter { const nativeReconnect = this.options.nativePregame?.kind === "reconnect" ? this.options.nativePregame : null; + if (!this.acceptNativeReconnectIdentity({ + playerId: data.your_player, + playerToken: data.player_token, + fullKey: data.full_key, + })) break; if (nativeReconnect) { - if (data.your_player !== nativeReconnect.playerId) { - const error = new AdapterError( - "WS_ERROR", - `Native reconnect attached player ${data.your_player}, expected ${nativeReconnect.playerId}`, - false, - ); - this.rejectInitialization(error); - this.emit({ type: "error", message: error.message }); - break; - } if (!this.acceptFullSessionKey(data.full_key)) break; } if (this.reconnectInFlight) { @@ -1392,7 +1398,7 @@ export class WebSocketAdapter implements EngineAdapter { this._gameCode = this.joinGameCode; } if (!nativeReconnect && data.full_key && !this.acceptFullSessionKey(data.full_key)) break; - if (data.player_token) { + if (!nativeReconnect && data.player_token) { this.playerToken = data.player_token; this.emit({ type: "sessionChanged", session: this.currentSession() }); } @@ -1782,4 +1788,36 @@ export class WebSocketAdapter implements EngineAdapter { this.fullSessionKey = key; return true; } + + /** Reject identity-bearing reconnect frames before they can update session state. */ + private acceptNativeReconnectIdentity(frame: { + gameCode?: string; + playerId?: PlayerId; + playerToken?: string; + fullKey?: FullSessionKey; + }): boolean { + const expected = this.options.nativePregame?.kind === "reconnect" + ? this.options.nativePregame + : null; + if (!expected) return true; + + const errorMessage = frame.gameCode !== undefined && frame.gameCode !== expected.gameCode + ? `Native reconnect attached game ${frame.gameCode}, expected ${expected.gameCode}` + : frame.playerId !== undefined && frame.playerId !== expected.playerId + ? `Native reconnect attached player ${frame.playerId}, expected ${expected.playerId}` + : frame.playerToken !== undefined && frame.playerToken !== expected.playerToken + ? "Native reconnect changed the player token" + : !frame.fullKey + ? "Server omitted a valid Full session identity" + : frame.fullKey.game_code !== expected.fullKey.game_code + || frame.fullKey.generation !== expected.fullKey.generation + ? "Server changed the Full session identity" + : null; + if (!errorMessage) return true; + + const error = new AdapterError("WS_ERROR", errorMessage, false); + this.rejectInitialization(error); + this.emit({ type: "error", message: error.message }); + return false; + } } diff --git a/crates/phase-server/src/main.rs b/crates/phase-server/src/main.rs index eb71732d6e..4082b3d2ba 100644 --- a/crates/phase-server/src/main.rs +++ b/crates/phase-server/src/main.rs @@ -7742,10 +7742,12 @@ async fn handle_client_message( #[cfg(test)] mod state_transport_derived_tests { use super::*; - use engine::types::ability::SearchSelectionConstraint; + use engine::game::deck_loading::PlayerDeckPayload; + use engine::types::ability::{Effect, ResolvedAbility, SearchSelectionConstraint}; use engine::types::actions::GameAction; use engine::types::game_state::{ - ActiveSearchDecisionAuthority, ActiveSearchDecisionControl, PriorityPassingMode, WaitingFor, + ActiveSearchDecisionAuthority, ActiveSearchDecisionControl, PriorityPassingMode, + StackEntry, StackEntryKind, WaitingFor, }; use engine::types::identifiers::ObjectId; use engine::types::log::{GameLogEntry, LogCategory, LogSegment}; @@ -7883,6 +7885,135 @@ mod state_transport_derived_tests { assert_eq!(&tail[tail.len() - ai_logs.len()..], ai_logs.as_slice()); } + #[tokio::test] + async fn resolve_all_handler_sends_the_post_ai_snapshot_before_its_acknowledgement() { + let mut manager = SessionManager::new(); + let (game_code, player_token) = manager.create_game(PlayerDeckPayload::default()); + let ai_player = PlayerId(1); + let session = manager + .sessions + .get_mut(&game_code) + .expect("new game retains its session"); + session.ai_seats.insert(ai_player); + session.ai_configs.insert( + ai_player, + phase_ai::config::create_config_for_players( + phase_ai::config::AiDifficulty::Easy, + phase_ai::config::Platform::Native, + 2, + ), + ); + let stack_object = ObjectId(1); + session.state.active_player = ai_player; + session.state.priority_player = PlayerId(0); + session.state.waiting_for = WaitingFor::Priority { + player: PlayerId(0), + }; + session.state.stack.push(StackEntry { + id: stack_object, + source_id: stack_object, + controller: PlayerId(0), + kind: StackEntryKind::ActivatedAbility { + source_id: stack_object, + ability: Box::new(ResolvedAbility::new( + Effect::NoOp, + Vec::new(), + stack_object, + PlayerId(0), + )), + }, + }); + let revision_before = session.state_revision; + + let state: SharedState = Arc::new(Mutex::new(manager)); + let draft_state: SharedDraftState = Arc::new(Mutex::new(DraftSessionManager::new())); + let connections: SharedConnections = Arc::new(Mutex::new(HashMap::new())); + let game_spectators: SharedGameSpectators = Arc::new(Mutex::new(HashMap::new())); + let db_file = tempfile::NamedTempFile::new().expect("temporary game database"); + let game_db = Arc::new( + persistence::GameDb::open(db_file.path(), persistence::SessionRetention::Multiplayer) + .expect("open temporary game database"), + ); + let (requester_tx, mut requester_rx) = mpsc::unbounded_channel(); + let (ai_tx, mut ai_rx) = mpsc::unbounded_channel(); + connections + .lock() + .await + .insert(game_code.clone(), HashMap::from([(ai_player, ai_tx)])); + let identity = SocketIdentity { + game_code: Some(game_code.clone()), + player_id: Some(PlayerId(0)), + player_token: Some(player_token), + lobby_subscribed: false, + session_span: None, + client_hello: None, + lobby_host_game: None, + seat_reservations: Vec::new(), + lobby_reservations: Vec::new(), + draft_code: None, + draft_seat: None, + draft_token: None, + spectator_draft_code: None, + spectator_visibility: None, + spectator_game_code: None, + }; + + handle_resolve_all( + 41, + 1, + &state, + &draft_state, + &connections, + &requester_tx, + &game_db, + &game_spectators, + &identity, + ) + .await; + + let (expected_revision, expected_waiting_for) = { + let manager = state.lock().await; + let session = manager + .sessions + .get(&game_code) + .expect("Resolve All retains its session"); + assert!( + session.state_revision >= revision_before + 2, + "one batch resolution and the AI priority action must both advance the revision" + ); + (session.state_revision, session.state.waiting_for.clone()) + }; + + match requester_rx.recv().await.expect("requester state update") { + ServerMessage::StateUpdate { + state_revision, + state, + .. + } => { + assert_eq!(state_revision, expected_revision); + assert_eq!(state.waiting_for, expected_waiting_for); + } + other => panic!("expected requester StateUpdate, got {other:?}"), + } + assert!(matches!( + requester_rx + .recv() + .await + .expect("Resolve All acknowledgement"), + ServerMessage::ResolveAllResult { + request_id: 41, + items_resolved: 1, + total: 1, + } + )); + match ai_rx.recv().await.expect("AI recipient state update") { + ServerMessage::StateUpdate { state_revision, .. } => { + assert_eq!(state_revision, expected_revision); + } + other => panic!("expected AI recipient StateUpdate, got {other:?}"), + } + } + #[test] fn turn_controller_receives_low_use_window_recommendation_instead_of_controlled_seat() { let controlled = PlayerId(0); From 5bc530eb42a800d759b27aca50ece45bf1902fde Mon Sep 17 00:00:00 2001 From: matthewevans Date: Wed, 12 Aug 2026 19:04:08 -0700 Subject: [PATCH 3/8] fix(native): terminate rejected reconnect identities --- .../src/adapter/__tests__/ws-adapter.test.ts | 13 +++++++++ client/src/adapter/ws-adapter.ts | 27 ++++++++++++++----- 2 files changed, 33 insertions(+), 7 deletions(-) diff --git a/client/src/adapter/__tests__/ws-adapter.test.ts b/client/src/adapter/__tests__/ws-adapter.test.ts index 71bfcc2b9f..d554a1dba0 100644 --- a/client/src/adapter/__tests__/ws-adapter.test.ts +++ b/client/src/adapter/__tests__/ws-adapter.test.ts @@ -760,6 +760,18 @@ describe("WebSocketAdapter", () => { await expect(attached).rejects.toThrow("Native reconnect attached game ATTACK, expected NATIVE"); await expect(nativeAdapter.getSnapshot()).rejects.toThrow("No game state available"); + nativeSocket.dispatchSynthetic( + "message", + JSON.stringify({ + type: "StateUpdate", + data: { + state_revision: 8, + state: createMockState(), + events: [], + }, + }), + ); + await expect(nativeAdapter.getSnapshot()).rejects.toThrow("No game state available"); expect(nativeAdapter.nativeSession).toEqual({ gameCode: "NATIVE", playerId: 1, @@ -768,6 +780,7 @@ describe("WebSocketAdapter", () => { }); expect(listener).toHaveBeenCalledTimes(1); expect(listener).toHaveBeenCalledWith(expect.objectContaining({ type: "error" })); + expect(nativeSocket.close).toHaveBeenCalledTimes(1); }); it("rejects a hostile SessionAttached before it can settle native reconnect", async () => { diff --git a/client/src/adapter/ws-adapter.ts b/client/src/adapter/ws-adapter.ts index 9a8ec1c259..99557c07ba 100644 --- a/client/src/adapter/ws-adapter.ts +++ b/client/src/adapter/ws-adapter.ts @@ -381,6 +381,8 @@ export class WebSocketAdapter implements EngineAdapter { private reconnectTimer: ReturnType | null = null; private pingInterval: ReturnType | null = null; private disposed = false; + /** A rejected Full identity is terminal for this socket. */ + private sessionIdentityRejected = false; private gameEnded = false; /** * Populated once the server's `ServerHello` arrives. `null` between the @@ -693,6 +695,7 @@ export class WebSocketAdapter implements EngineAdapter { clearInterval(this.pingInterval); this.pingInterval = null; } + if (this.sessionIdentityRejected) return; // Clear the "host waiting for opponent" latch on socket close — // otherwise a host who received GameCreated, disconnected before // GameStarted, and then reconnected through a different path would @@ -1211,6 +1214,8 @@ export class WebSocketAdapter implements EngineAdapter { } private handleMessage(msg: { type: string; data?: unknown }): void { + if (this.sessionIdentityRejected) return; + switch (msg.type) { // ServerHello is no longer observed here — the shared // `openPhaseSocket` helper consumes it during `attachSocket`, and @@ -1771,9 +1776,7 @@ export class WebSocketAdapter implements EngineAdapter { private acceptFullSessionKey(key: FullSessionKey | undefined): boolean { if (!key || key.game_code !== this._gameCode || key.generation < 1) { const error = new AdapterError("WS_ERROR", "Server omitted a valid Full session identity", false); - this.rejectInitialization(error); - this.emit({ type: "error", message: error.message }); - return false; + return this.rejectSessionIdentity(error); } if ( this.fullSessionKey @@ -1781,9 +1784,7 @@ export class WebSocketAdapter implements EngineAdapter { || this.fullSessionKey.generation !== key.generation) ) { const error = new AdapterError("WS_ERROR", "Server changed the Full session identity", false); - this.rejectInitialization(error); - this.emit({ type: "error", message: error.message }); - return false; + return this.rejectSessionIdentity(error); } this.fullSessionKey = key; return true; @@ -1815,9 +1816,21 @@ export class WebSocketAdapter implements EngineAdapter { : null; if (!errorMessage) return true; - const error = new AdapterError("WS_ERROR", errorMessage, false); + return this.rejectSessionIdentity(new AdapterError("WS_ERROR", errorMessage, false)); + } + + /** Latches an invalid Full identity before later frames can mutate session state. */ + private rejectSessionIdentity(error: AdapterError): false { + if (this.sessionIdentityRejected) return false; + + this.sessionIdentityRejected = true; + if (this.pingInterval) { + clearInterval(this.pingInterval); + this.pingInterval = null; + } this.rejectInitialization(error); this.emit({ type: "error", message: error.message }); + this.ws?.close(); return false; } } From 95b53ffdc0faaa0bb34a0e8787abf946254da22a Mon Sep 17 00:00:00 2001 From: matthewevans Date: Wed, 12 Aug 2026 19:08:26 -0700 Subject: [PATCH 4/8] fix(test): use persistent stack insertion --- crates/phase-server/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/phase-server/src/main.rs b/crates/phase-server/src/main.rs index 4082b3d2ba..3d978be599 100644 --- a/crates/phase-server/src/main.rs +++ b/crates/phase-server/src/main.rs @@ -7909,7 +7909,7 @@ mod state_transport_derived_tests { session.state.waiting_for = WaitingFor::Priority { player: PlayerId(0), }; - session.state.stack.push(StackEntry { + session.state.stack.push_back(StackEntry { id: stack_object, source_id: stack_object, controller: PlayerId(0), From de56e5f99b73e98079ae975299062292ca095150 Mon Sep 17 00:00:00 2001 From: matthewevans Date: Wed, 12 Aug 2026 19:34:37 -0700 Subject: [PATCH 5/8] test(server): bound resolve all delivery waits --- crates/phase-server/src/main.rs | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/crates/phase-server/src/main.rs b/crates/phase-server/src/main.rs index 3d978be599..cb9e403f08 100644 --- a/crates/phase-server/src/main.rs +++ b/crates/phase-server/src/main.rs @@ -7984,7 +7984,11 @@ mod state_transport_derived_tests { (session.state_revision, session.state.waiting_for.clone()) }; - match requester_rx.recv().await.expect("requester state update") { + match tokio::time::timeout(std::time::Duration::from_secs(1), requester_rx.recv()) + .await + .expect("Resolve All must send the requester state update") + .expect("requester state update channel remains open") + { ServerMessage::StateUpdate { state_revision, state, @@ -7996,17 +8000,21 @@ mod state_transport_derived_tests { other => panic!("expected requester StateUpdate, got {other:?}"), } assert!(matches!( - requester_rx - .recv() + tokio::time::timeout(std::time::Duration::from_secs(1), requester_rx.recv()) .await - .expect("Resolve All acknowledgement"), + .expect("Resolve All must acknowledge after its state update") + .expect("requester acknowledgement channel remains open"), ServerMessage::ResolveAllResult { request_id: 41, items_resolved: 1, total: 1, } )); - match ai_rx.recv().await.expect("AI recipient state update") { + match tokio::time::timeout(std::time::Duration::from_secs(1), ai_rx.recv()) + .await + .expect("Resolve All must fan out the final state to the AI seat") + .expect("AI recipient channel remains open") + { ServerMessage::StateUpdate { state_revision, .. } => { assert_eq!(state_revision, expected_revision); } From e328235915ba15eee1fdf71b11001d1aa2115813 Mon Sep 17 00:00:00 2001 From: matthewevans Date: Wed, 12 Aug 2026 19:58:22 -0700 Subject: [PATCH 6/8] test(server): make resolve-all snapshot assertion deterministic --- crates/phase-server/src/main.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/phase-server/src/main.rs b/crates/phase-server/src/main.rs index cb9e403f08..7342c1c24b 100644 --- a/crates/phase-server/src/main.rs +++ b/crates/phase-server/src/main.rs @@ -7886,7 +7886,7 @@ mod state_transport_derived_tests { } #[tokio::test] - async fn resolve_all_handler_sends_the_post_ai_snapshot_before_its_acknowledgement() { + async fn resolve_all_handler_sends_the_final_snapshot_before_its_acknowledgement() { let mut manager = SessionManager::new(); let (game_code, player_token) = manager.create_game(PlayerDeckPayload::default()); let ai_player = PlayerId(1); @@ -7978,8 +7978,8 @@ mod state_transport_derived_tests { .get(&game_code) .expect("Resolve All retains its session"); assert!( - session.state_revision >= revision_before + 2, - "one batch resolution and the AI priority action must both advance the revision" + session.state_revision >= revision_before + 1, + "the resolved batch must advance the authoritative revision" ); (session.state_revision, session.state.waiting_for.clone()) }; From 361b0c722a8bee0fd8ffb0091313796cc73442eb Mon Sep 17 00:00:00 2001 From: matthewevans Date: Wed, 12 Aug 2026 20:16:18 -0700 Subject: [PATCH 7/8] test(server): satisfy resolve all revision lint --- crates/phase-server/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/phase-server/src/main.rs b/crates/phase-server/src/main.rs index 7342c1c24b..e53c4bbcea 100644 --- a/crates/phase-server/src/main.rs +++ b/crates/phase-server/src/main.rs @@ -7978,7 +7978,7 @@ mod state_transport_derived_tests { .get(&game_code) .expect("Resolve All retains its session"); assert!( - session.state_revision >= revision_before + 1, + session.state_revision > revision_before, "the resolved batch must advance the authoritative revision" ); (session.state_revision, session.state.waiting_for.clone()) From 3c869d2c2116ee3cd592efdb481e964d199c3ada Mon Sep 17 00:00:00 2001 From: matthewevans Date: Wed, 12 Aug 2026 20:52:06 -0700 Subject: [PATCH 8/8] test(server): make resolve all transport fixture deterministic --- crates/phase-server/src/main.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/phase-server/src/main.rs b/crates/phase-server/src/main.rs index e53c4bbcea..40a85ea7af 100644 --- a/crates/phase-server/src/main.rs +++ b/crates/phase-server/src/main.rs @@ -7909,6 +7909,9 @@ mod state_transport_derived_tests { session.state.waiting_for = WaitingFor::Priority { player: PlayerId(0), }; + // The AI has already passed in this priority cycle, so the requesting + // human's pass deterministically resolves the stack entry. + session.state.priority_passes.insert(ai_player); session.state.stack.push_back(StackEntry { id: stack_object, source_id: stack_object,