feat: expand progression and multiplayer systems

Add named save slots, progression import/export, and a unified play flow. Add live friend requests, presence, invitations, and relationship controls without durable discovery-server social storage. Advance the network protocol with isolated channels, movement reconciliation, late-join recovery, fishing replication, and animation synchronization. Preserve per-species catch totals, refine generated-world startup and water recovery, and complete the related input and interface improvements.

network/discovery_client.gd

@@ -9,6 +9,15 @@ signal host_state_changed(state: int)
 signal public_join_prepared(endpoint: String)
 signal public_join_status_changed(message: String, is_error: bool)
 signal public_join_state_changed(state: int)
+signal friend_presence_updated(friends: Array[Dictionary])
+signal presence_sharing_changed(enabled: bool)
+signal social_status_changed(message: String, is_error: bool)
+signal friend_invite_received(
+	fingerprint: String,
+	display_name: String,
+	room: Dictionary,
+)
+signal friend_invite_finished(success: bool, message: String)
 
 const BASE_URL_SETTING: String = "network/discovery/base_url"
 const BASE_URL_ENVIRONMENT: String = "NETFISHING_DISCOVERY_URL"
@@ -28,6 +37,7 @@ const TRAVERSAL_PACKET_PREFIX: String = "NETFISHING_TRAVERSAL_V1 "
 const UPNP_MAPPING_DURATION_SECONDS: int = 3600
 const UPNP_RENEW_INTERVAL_SECONDS: float = 2700.0
 const UPNP_RETRY_INTERVAL_SECONDS: float = 300.0
+const SOCIAL_POLL_INTERVAL_SECONDS: float = 5.0
 
 enum HostState {
 	UNAVAILABLE,
@@ -55,6 +65,7 @@ enum HostRequestKind {
 }
 
 var _session: NetworkSession
+var _relationships: PlayerRelationshipStore
 var _base_url: String = ""
 var _room_name: String = DEFAULT_ROOM_NAME
 var _room_name_uses_default: bool = true
@@ -92,6 +103,18 @@ var _upnp_operation_is_renewal: bool = false
 var _upnp_renew_timer: Timer
 var _upnp: UPNP
 var _upnp_mapped_port: int = 0
+var _presence_sharing: bool = false
+var _social_timer: Timer
+var _presence_request: HTTPRequest
+var _presence_query_request: HTTPRequest
+var _invitation_poll_request: HTTPRequest
+var _invitation_send_request: HTTPRequest
+var _friend_presence: Dictionary[String, Dictionary] = {}
+var _published_presence_tokens: PackedStringArray = PackedStringArray()
+var _pending_presence_online: bool = false
+var _pending_presence_tokens: PackedStringArray = PackedStringArray()
+var _pending_invite_fingerprint: String = ""
+var _joined_public_room_id: String = ""
 
 
 func _ready() -> void:
@@ -147,16 +170,48 @@ func _ready() -> void:
 	_join_probe_timer.one_shot = false
 	add_child(_join_probe_timer)
 	_join_probe_timer.timeout.connect(_send_pending_join_probe)
+
+	_presence_request = _make_social_request(
+		"FriendPresencePublish", _on_presence_request_completed
+	)
+	_presence_query_request = _make_social_request(
+		"FriendPresenceQuery", _on_presence_query_completed
+	)
+	_invitation_poll_request = _make_social_request(
+		"FriendInvitationPoll", _on_invitation_poll_completed
+	)
+	_invitation_send_request = _make_social_request(
+		"FriendInvitationSend", _on_invitation_send_completed
+	)
+	_social_timer = Timer.new()
+	_social_timer.name = "FriendPresencePoll"
+	_social_timer.wait_time = SOCIAL_POLL_INTERVAL_SECONDS
+	_social_timer.one_shot = false
+	add_child(_social_timer)
+	_social_timer.timeout.connect(_refresh_social_state)
 	_load_settings()
 	_base_url = _configured_base_url()
 
 
-func setup(session: NetworkSession) -> void:
+func setup(
+	session: NetworkSession,
+	relationships: PlayerRelationshipStore = null,
+) -> void:
 	_session = session
+	_relationships = relationships
 	if _room_name_uses_default:
 		_room_name = _default_room_name(_session.get_local_display_name())
 		_save_settings()
 	_session.set_session_display_name(_room_name)
+	if (
+		_relationships != null
+		and not _relationships.relationship_changed.is_connected(
+			_on_social_relationship_changed
+		)
+	):
+		_relationships.relationship_changed.connect(
+			_on_social_relationship_changed
+		)
 	if not _session.state_changed.is_connected(_on_session_state_changed):
 		_session.state_changed.connect(_on_session_state_changed)
 	if not _session.peer_count_changed.is_connected(_on_peer_count_changed):
@@ -173,6 +228,9 @@ func setup(session: NetworkSession) -> void:
 	else:
 		_set_host_state(HostState.CLOSED)
 		_set_host_status("Open the game before listing it publicly.", false)
+	if is_configured() and _relationships != null:
+		_social_timer.start()
+		call_deferred("_refresh_social_state")
 
 
 func is_configured() -> bool:
@@ -183,6 +241,112 @@ func get_base_url() -> String:
 	return _base_url
 
 
+func is_presence_sharing() -> bool:
+	return _presence_sharing
+
+
+func set_presence_sharing(enabled: bool) -> bool:
+	if enabled and (not is_configured() or _relationships == null):
+		social_status_changed.emit(
+			"Friend presence is not configured in this build.", true
+		)
+		return false
+	if _presence_sharing == enabled:
+		return true
+	_presence_sharing = enabled
+	_save_settings()
+	presence_sharing_changed.emit(enabled)
+	_refresh_social_state()
+	return true
+
+
+func get_friend_presence() -> Array[Dictionary]:
+	var result: Array[Dictionary] = []
+	if _relationships == null:
+		return result
+	for friend: Dictionary in _relationships.get_friends():
+		var fingerprint := str(friend.get("fingerprint", ""))
+		var live: Dictionary = _friend_presence.get(fingerprint, {})
+		result.append({
+			"fingerprint": fingerprint,
+			"display_name": str(
+				friend.get("last_known_display_name", "Player")
+			),
+			"online": bool(live.get("online", false)),
+			"room": (
+				live.get("room", {}).duplicate(true)
+				if typeof(live.get("room", {})) == TYPE_DICTIONARY
+				else {}
+			),
+		})
+	result.sort_custom(func(a: Dictionary, b: Dictionary) -> bool:
+		if bool(a["online"]) != bool(b["online"]):
+			return bool(a["online"])
+		return str(a["display_name"]).naturalnocasecmp_to(
+			str(b["display_name"])
+		) < 0
+	)
+	return result
+
+
+func request_friend_presence() -> bool:
+	if not is_configured() or _relationships == null:
+		friend_presence_updated.emit(get_friend_presence())
+		return false
+	_refresh_social_state()
+	return true
+
+
+func send_friend_invite(fingerprint: String) -> bool:
+	if (
+		_relationships == null
+		or not _relationships.is_friend(fingerprint)
+		or _relationships.is_blocked(fingerprint)
+		or not is_configured()
+	):
+		friend_invite_finished.emit(
+			false, "This person needs to be online to do this."
+		)
+		return false
+	if (
+		_session == null
+		or not _session.is_host()
+		or not _host_verified
+		or _lease_room_id.is_empty()
+	):
+		friend_invite_finished.emit(
+			false,
+			"List your open room in discovery before inviting friends.",
+		)
+		return false
+	if (
+		not _pending_invite_fingerprint.is_empty()
+		or _invitation_send_request.get_http_client_status()
+			!= HTTPClient.STATUS_DISCONNECTED
+	):
+		return false
+	var friend: Dictionary = _relationships.get_friend_record(fingerprint)
+	var inbox_token := str(friend.get("remote_invite_token", ""))
+	if inbox_token.is_empty():
+		friend_invite_finished.emit(false, "Friend invitation is unavailable.")
+		return false
+	var error := _post_social_json(
+		_invitation_send_request,
+		"/v1/invitations",
+		{
+			"inbox_token": inbox_token,
+			"room_id": _lease_room_id,
+		},
+	)
+	if error != OK:
+		friend_invite_finished.emit(
+			false, "Friend invitation could not be sent."
+		)
+		return false
+	_pending_invite_fingerprint = fingerprint
+	return true
+
+
 func set_base_url_override(value: String) -> bool:
 	var normalized: String = value.strip_edges()
 	while normalized.ends_with("/"):
@@ -345,7 +509,7 @@ func prepare_public_join(room: Dictionary) -> bool:
 
 func request_rooms() -> bool:
 	if not is_configured():
-		rooms_updated.emit([])
+		rooms_updated.emit(_empty_rooms())
 		browse_status_changed.emit(
 			"Public room discovery is not configured in this build.", true
 		)
@@ -362,7 +526,7 @@ func request_rooms() -> bool:
 	]
 	var error: Error = _browse_request.request(url)
 	if error != OK:
-		rooms_updated.emit([])
+		rooms_updated.emit(_empty_rooms())
 		browse_status_changed.emit("Could not request public rooms.", true)
 		return false
 	_browse_request_in_flight = true
@@ -813,12 +977,12 @@ func _on_browse_request_completed(
 	_browse_request_in_flight = false
 	var response: Dictionary = _parse_response_dictionary(body)
 	if result != HTTPRequest.RESULT_SUCCESS or response_code != HTTPClient.RESPONSE_OK:
-		rooms_updated.emit([])
+		rooms_updated.emit(_empty_rooms())
 		browse_status_changed.emit(_request_failure(response), true)
 		return
 	var raw_rooms: Variant = response.get("rooms", [])
 	if typeof(raw_rooms) != TYPE_ARRAY:
-		rooms_updated.emit([])
+		rooms_updated.emit(_empty_rooms())
 		browse_status_changed.emit("Discovery returned an invalid room list.", true)
 		return
 	var rooms: Array[Dictionary] = []
@@ -873,6 +1037,309 @@ func _valid_json_integer(value: Variant, minimum: int, maximum: int) -> bool:
 	)
 
 
+func _make_social_request(name: String, callback: Callable) -> HTTPRequest:
+	var request := HTTPRequest.new()
+	request.name = name
+	request.timeout = REQUEST_TIMEOUT_SECONDS
+	add_child(request)
+	request.request_completed.connect(callback)
+	return request
+
+
+func _social_payload(fields: Dictionary = {}) -> Dictionary:
+	var payload: Dictionary = fields.duplicate(true)
+	payload["game_version"] = NetworkProtocol.game_version()
+	payload["protocol_version"] = NetworkProtocol.PROTOCOL_VERSION
+	return payload
+
+
+func _post_social_json(
+	request: HTTPRequest,
+	path: String,
+	fields: Dictionary,
+) -> Error:
+	return request.request(
+		_base_url + path,
+		PackedStringArray(["Content-Type: application/json"]),
+		HTTPClient.METHOD_POST,
+		JSON.stringify(_social_payload(fields)),
+	)
+
+
+func _refresh_social_state() -> void:
+	if not is_configured() or _relationships == null:
+		return
+	_publish_friend_presence()
+	_query_friend_presence()
+	_poll_friend_invitations()
+
+
+func _publish_friend_presence() -> void:
+	if (
+		_presence_request.get_http_client_status()
+			!= HTTPClient.STATUS_DISCONNECTED
+	):
+		return
+	var current_tokens := _friend_social_values(
+		"local_presence_write_token"
+	)
+	var online := _presence_sharing and not current_tokens.is_empty()
+	var publish_online := online
+	var tokens := PackedStringArray()
+	if online:
+		# Revoke removed or blocked friendships before refreshing the remaining
+		# capabilities. This makes status disappear on the next request instead of
+		# waiting for the server's short presence TTL.
+		tokens = _presence_tokens_not_in(
+			_published_presence_tokens, current_tokens
+		)
+		if not tokens.is_empty():
+			publish_online = false
+		else:
+			tokens = current_tokens
+	else:
+		tokens = _published_presence_tokens
+	if tokens.is_empty():
+		return
+	var error := _post_social_json(
+		_presence_request,
+		"/v1/presence",
+		{
+			"display_name": _session.get_local_display_name(),
+			"room_id": _current_presence_room_id(),
+			"online": publish_online,
+			"write_tokens": Array(tokens),
+		},
+	)
+	if error == OK:
+		_pending_presence_online = publish_online
+		_pending_presence_tokens = tokens.duplicate()
+	else:
+		social_status_changed.emit("Could not update friend presence.", true)
+
+
+func _query_friend_presence() -> void:
+	if (
+		_presence_query_request.get_http_client_status()
+			!= HTTPClient.STATUS_DISCONNECTED
+	):
+		return
+	var channels := _friend_social_values("remote_presence_channel")
+	if channels.is_empty():
+		var had_presence := not _friend_presence.is_empty()
+		_friend_presence.clear()
+		if had_presence:
+			friend_presence_updated.emit(get_friend_presence())
+		return
+	var error := _post_social_json(
+		_presence_query_request,
+		"/v1/presence/query",
+		{"channels": Array(channels)},
+	)
+	if error != OK:
+		social_status_changed.emit("Could not refresh friend status.", true)
+
+
+func _poll_friend_invitations() -> void:
+	if (
+		_invitation_poll_request.get_http_client_status()
+			!= HTTPClient.STATUS_DISCONNECTED
+	):
+		return
+	var tokens := _friend_social_values("local_invite_token")
+	if tokens.is_empty():
+		return
+	var error := _post_social_json(
+		_invitation_poll_request,
+		"/v1/invitations/poll",
+		{"inbox_tokens": Array(tokens)},
+	)
+	if error != OK:
+		social_status_changed.emit("Could not check friend invitations.", true)
+
+
+func _friend_social_values(key: String) -> PackedStringArray:
+	var result := PackedStringArray()
+	if _relationships == null:
+		return result
+	for friend: Dictionary in _relationships.get_friends():
+		var value := str(friend.get(key, ""))
+		if not value.is_empty() and value not in result:
+			result.append(value)
+	return result
+
+
+func _current_presence_room_id() -> String:
+	if _session == null:
+		return ""
+	if _session.is_host() and _host_verified:
+		return _lease_room_id
+	if _session.is_joined_client():
+		return _joined_public_room_id
+	return ""
+
+
+func _on_presence_request_completed(
+	result: int,
+	response_code: int,
+	_headers: PackedStringArray,
+	body: PackedByteArray,
+) -> void:
+	var published_online := _pending_presence_online
+	var request_tokens := _pending_presence_tokens.duplicate()
+	_pending_presence_online = false
+	_pending_presence_tokens = PackedStringArray()
+	var response := _parse_response_dictionary(body)
+	if result != HTTPRequest.RESULT_SUCCESS or response_code != HTTPClient.RESPONSE_OK:
+		social_status_changed.emit(_request_failure(response), true)
+		return
+	if published_online:
+		for token: String in request_tokens:
+			if token not in _published_presence_tokens:
+				_published_presence_tokens.append(token)
+	else:
+		for token: String in request_tokens:
+			var index := _published_presence_tokens.find(token)
+			if index >= 0:
+				_published_presence_tokens.remove_at(index)
+	if _presence_publish_needs_follow_up():
+		call_deferred("_publish_friend_presence")
+
+
+func _presence_publish_needs_follow_up() -> bool:
+	var desired := (
+		_friend_social_values("local_presence_write_token")
+		if _presence_sharing
+		else PackedStringArray()
+	)
+	return (
+		not _presence_tokens_not_in(
+			_published_presence_tokens, desired
+		).is_empty()
+		or not _presence_tokens_not_in(
+			desired, _published_presence_tokens
+		).is_empty()
+	)
+
+
+func _presence_tokens_not_in(
+	first: PackedStringArray,
+	second: PackedStringArray,
+) -> PackedStringArray:
+	var result := PackedStringArray()
+	for token: String in first:
+		if token not in second:
+			result.append(token)
+	return result
+
+
+func _on_presence_query_completed(
+	result: int,
+	response_code: int,
+	_headers: PackedStringArray,
+	body: PackedByteArray,
+) -> void:
+	var response := _parse_response_dictionary(body)
+	if result != HTTPRequest.RESULT_SUCCESS or response_code != HTTPClient.RESPONSE_OK:
+		social_status_changed.emit(_request_failure(response), true)
+		return
+	var by_channel: Dictionary[String, String] = {}
+	for friend: Dictionary in _relationships.get_friends():
+		by_channel[str(friend.get("remote_presence_channel", ""))] = str(
+			friend.get("fingerprint", "")
+		)
+	var next_presence: Dictionary[String, Dictionary] = {}
+	var raw_presence: Variant = response.get("presence", [])
+	if typeof(raw_presence) == TYPE_ARRAY:
+		for value: Variant in raw_presence:
+			if typeof(value) != TYPE_DICTIONARY:
+				continue
+			var presence: Dictionary = value
+			var fingerprint := str(
+				by_channel.get(str(presence.get("channel", "")), "")
+			)
+			if fingerprint.is_empty() or _relationships.is_blocked(fingerprint):
+				continue
+			var room: Dictionary = {}
+			var raw_room: Variant = presence.get("room", {})
+			if typeof(raw_room) == TYPE_DICTIONARY and _valid_public_room(raw_room):
+				room = (raw_room as Dictionary).duplicate(true)
+			next_presence[fingerprint] = {
+				"online": true,
+				"room": room,
+			}
+	if next_presence == _friend_presence:
+		return
+	_friend_presence = next_presence
+	friend_presence_updated.emit(get_friend_presence())
+
+
+func _on_invitation_poll_completed(
+	result: int,
+	response_code: int,
+	_headers: PackedStringArray,
+	body: PackedByteArray,
+) -> void:
+	var response := _parse_response_dictionary(body)
+	if result != HTTPRequest.RESULT_SUCCESS or response_code != HTTPClient.RESPONSE_OK:
+		return
+	var by_inbox: Dictionary[String, Dictionary] = {}
+	for friend: Dictionary in _relationships.get_friends():
+		var token := str(friend.get("local_invite_token", ""))
+		if not token.is_empty():
+			by_inbox[PlayerRelationshipStore.invite_inbox_id(token)] = friend
+	var invitations: Variant = response.get("invitations", [])
+	if typeof(invitations) != TYPE_ARRAY:
+		return
+	for value: Variant in invitations:
+		if typeof(value) != TYPE_DICTIONARY:
+			continue
+		var invitation: Dictionary = value
+		var friend: Dictionary = by_inbox.get(
+			str(invitation.get("inbox_id", "")), {}
+		)
+		var raw_room: Variant = invitation.get("room", {})
+		if friend.is_empty() or typeof(raw_room) != TYPE_DICTIONARY:
+			continue
+		var room := raw_room as Dictionary
+		if not _valid_public_room(room):
+			continue
+		var fingerprint := str(friend.get("fingerprint", ""))
+		if _relationships.is_blocked(fingerprint):
+			continue
+		friend_invite_received.emit(
+			fingerprint,
+			str(friend.get("last_known_display_name", "Player")),
+			room.duplicate(true),
+		)
+
+
+func _on_invitation_send_completed(
+	result: int,
+	response_code: int,
+	_headers: PackedStringArray,
+	body: PackedByteArray,
+) -> void:
+	_pending_invite_fingerprint = ""
+	var response := _parse_response_dictionary(body)
+	var success := (
+		result == HTTPRequest.RESULT_SUCCESS
+		and response_code == HTTPClient.RESPONSE_CREATED
+	)
+	friend_invite_finished.emit(
+		success,
+		"Invitation sent." if success else _request_failure(response),
+	)
+
+
+func _on_social_relationship_changed(_fingerprint: String) -> void:
+	for fingerprint: String in _friend_presence.keys():
+		if not _relationships.is_friend(fingerprint):
+			_friend_presence.erase(fingerprint)
+	friend_presence_updated.emit(get_friend_presence())
+	call_deferred("_refresh_social_state")
+
+
 func _parse_response_dictionary(body: PackedByteArray) -> Dictionary:
 	if body.is_empty():
 		return {}
@@ -924,6 +1391,21 @@ func _discovery_version_mismatch_message(required_version: String) -> String:
 
 
 func _on_session_state_changed(state: NetworkSession.State) -> void:
+	if state == NetworkSession.State.JOINED_CLIENT:
+		if not _pending_join_room_id.is_empty():
+			_joined_public_room_id = _pending_join_room_id
+	elif state in [
+		NetworkSession.State.PRIVATE_HOST,
+		NetworkSession.State.OPEN_HOST,
+		NetworkSession.State.SERVER_LOST,
+		NetworkSession.State.CONNECTION_FAILED,
+	]:
+		_joined_public_room_id = ""
+	elif (
+		state == NetworkSession.State.INACTIVE
+		and not _preserve_pending_join_on_inactive
+	):
+		_joined_public_room_id = ""
 	if state == NetworkSession.State.CONNECTING and not _pending_join_token.is_empty():
 		_preserve_pending_join_on_inactive = false
 		_set_public_join_state(PublicJoinState.CONNECTING)
@@ -979,6 +1461,7 @@ func _on_session_state_changed(state: NetworkSession.State) -> void:
 			_set_host_status("Open the game before listing it publicly.", false)
 		else:
 			_set_host_state(HostState.CLOSED)
+	call_deferred("_refresh_social_state")
 
 
 func _on_host_openness_changed(is_open: bool) -> void:
@@ -1039,6 +1522,10 @@ func _set_public_join_state(state: PublicJoinState) -> void:
 	public_join_state_changed.emit(int(state))
 
 
+func _empty_rooms() -> Array[Dictionary]:
+	return []
+
+
 func _sanitize_room_name(value: String) -> String:
 	var cleaned: String = value.strip_edges().replace("\n", " ").replace("\r", " ")
 	cleaned = cleaned.replace("\t", " ")
@@ -1065,7 +1552,7 @@ func _load_settings() -> void:
 		)
 		if not saved_name.is_empty():
 			_room_name = saved_name
-		_room_name_uses_default = bool(config.get_value(
+			_room_name_uses_default = bool(config.get_value(
 			"host",
 			"room_name_uses_default",
 			saved_name in [
@@ -1073,6 +1560,9 @@ func _load_settings() -> void:
 				LEGACY_DEFAULT_ROOM_NAME,
 				LEGACY_DEDICATED_DEFAULT_ROOM_NAME,
 			],
+			))
+		_presence_sharing = bool(config.get_value(
+			"social", "share_presence", false
 		))
 
 
@@ -1084,6 +1574,7 @@ func _save_settings() -> void:
 	config.set_value(
 		"host", "room_name_uses_default", _room_name_uses_default
 	)
+	config.set_value("social", "share_presence", _presence_sharing)
 	var error: Error = config.save(SETTINGS_PATH)
 	if error != OK:
-		push_warning("Could not save the local NETfishing room name.")
+		push_warning("Could not save local NETfishing discovery settings.")

network/network_fish_showcase_protocol.gd

@@ -2,7 +2,7 @@ class_name NetworkFishShowcaseProtocol
 extends RefCounted
 
 const CAPABILITY: StringName = &"fish_showcase_v1"
-const RELIABLE_CHANNEL: int = NetworkProtocol.ITEM_RELIABLE_CHANNEL
+const RELIABLE_CHANNEL: int = NetworkProtocol.SHOWCASE_RELIABLE_CHANNEL
 const MAX_SESSION_ID_LENGTH: int = 96
 const MAX_FISH_ID_LENGTH: int = 96
 const MAX_WEIGHT_LB: float = 1000.0

network/network_fish_showcase_service.gd

@@ -7,6 +7,7 @@ const FishCatchType = preload("res://fish/fish_catch.gd")
 const FishQualityType = preload("res://fish/fish_quality.gd")
 const FishInventoryType = preload("res://inventory/fish_inventory.gd")
 const PlayerHotbarType = preload("res://inventory/player_hotbar.gd")
+const LOCAL_STATE_RETRY_SECONDS: float = 0.75
 
 var _session: NetworkSession
 var _spawn_service: PlayerSpawnService
@@ -15,6 +16,8 @@ var _local_inventory: FishInventoryType
 var _local_hotbar: PlayerHotbarType
 var _states: Dictionary[int, Dictionary] = {}
 var _local_revision: int = 0
+var _local_acknowledged_revision: int = -1
+var _local_state_retry_elapsed: float = 0.0
 var _local_visible: bool = false
 var _local_catch_id: StringName
 
@@ -54,6 +57,30 @@ func setup(
 	)
 
 
+func _process(delta: float) -> void:
+	if (
+		_session == null
+		or not _session.is_joined_client()
+		or _local_revision <= 0
+		or _local_revision <= _local_acknowledged_revision
+	):
+		_local_state_retry_elapsed = 0.0
+		return
+	_local_state_retry_elapsed += delta
+	if _local_state_retry_elapsed < LOCAL_STATE_RETRY_SECONDS:
+		return
+	var fish_catch: FishCatchType
+	if _local_visible:
+		fish_catch = _local_inventory.get_catch_by_id(_local_catch_id)
+		if fish_catch == null or not fish_catch.is_valid():
+			_submit_local_state(null, false)
+			return
+	# A client can finish a generated-world transition on the same frame that
+	# its local showcase changes. Republish until the authoritative echo arrives
+	# so that one early application-level miss cannot leave held fish out of sync.
+	_submit_local_state(fish_catch, _local_visible)
+
+
 func toggle_selected_fish() -> bool:
 	if _local_hotbar == null or _local_inventory == null:
 		return false
@@ -78,6 +105,7 @@ func get_local_showcase_catch_id() -> StringName:
 
 func _submit_local_state(fish_catch: FishCatchType, should_show: bool) -> void:
 	_local_revision += 1
+	_local_state_retry_elapsed = 0.0
 	var local_peer_id: int = (
 		_session.get_local_peer_id() if _session != null else 1
 	)
@@ -166,6 +194,11 @@ func receive_showcase_state(data: Dictionary) -> void:
 	):
 		return
 	var peer_id: int = int(data["owner_peer_id"])
+	if peer_id == _session.get_local_peer_id():
+		_local_acknowledged_revision = maxi(
+			_local_acknowledged_revision,
+			int(data["revision"]),
+		)
 	var previous: Dictionary = _states.get(peer_id, {})
 	if int(data["revision"]) <= int(previous.get("revision", -1)):
 		return
@@ -267,6 +300,8 @@ func _on_session_state_changed(state: NetworkSession.State) -> void:
 	_states.clear()
 	_local_visible = false
 	_local_catch_id = StringName()
+	_local_acknowledged_revision = -1
+	_local_state_retry_elapsed = 0.0
 	var local_avatar: Player = _spawn_service.get_avatar(
 		_session.get_local_peer_id()
 	)

network/network_fishing_protocol.gd

@@ -8,6 +8,8 @@ const MAX_REEL_SPEED: float = 10.0
 const MAX_BARRIER_DAMAGE: int = 128
 const INPUT_CHANNEL: int = 3
 const SNAPSHOT_CHANNEL: int = 4
+const OBSERVER_SNAPSHOT_CHANNEL: int = 10
+const RELIABLE_CHANNEL: int = NetworkProtocol.FISHING_RELIABLE_CHANNEL
 const INPUT_RATE: float = 1.0 / 30.0
 const SNAPSHOT_RATE: float = 1.0 / 20.0
 
@@ -35,6 +37,7 @@ static func validate_cast_request(data: Variant) -> String:
 		"rarity_multipliers",
 		"discovered_fish_ids",
 		"capacity_available",
+		"movement_sequence",
 	]:
 		if not payload.has(key):
 			return "Malformed fishing request."
@@ -51,11 +54,16 @@ static func validate_cast_request(data: Variant) -> String:
 		or typeof(payload["rarity_multipliers"]) != TYPE_ARRAY
 		or typeof(payload["discovered_fish_ids"]) != TYPE_ARRAY
 		or typeof(payload["capacity_available"]) != TYPE_BOOL
+		or typeof(payload["movement_sequence"]) != TYPE_INT
 	):
 		return "Malformed fishing request."
-	if payload.has("bait_id") and typeof(payload["bait_id"]) not in [TYPE_STRING, TYPE_STRING_NAME]:
+	if payload.has("bait_id") and typeof(payload["bait_id"]) not in [
+		TYPE_STRING, TYPE_STRING_NAME
+	]:
 		return "Malformed fishing request."
-	if payload.has("lure_id") and typeof(payload["lure_id"]) not in [TYPE_STRING, TYPE_STRING_NAME]:
+	if payload.has("lure_id") and typeof(payload["lure_id"]) not in [
+		TYPE_STRING, TYPE_STRING_NAME
+	]:
 		return "Malformed fishing request."
 	var request_id: String = payload["request_id"]
 	var session_id: String = payload["session_id"]
@@ -92,6 +100,8 @@ static func validate_cast_request(data: Variant) -> String:
 		or str(payload["rod_id"]).is_empty()
 		or str(payload["rod_id"]).length() > 96
 		or str(payload.get("lure_id", "")).length() > 96
+		or int(payload["movement_sequence"]) < 0
+		or int(payload["movement_sequence"]) > 2147483647
 	):
 		return "Fishing request values are outside allowed limits."
 	for value: Variant in rarity:

network/network_fishing_service.gd

@@ -21,6 +21,8 @@ const MAX_LEDGER_ENTRIES_PER_PEER: int = 64
 const CAST_ORIGIN_TOLERANCE: float = 2.5
 const CAPACITY_RESPONSE_TIMEOUT: float = 5.0
 const MIN_CAST_INTERVAL: float = 0.25
+const OBSERVER_SNAPSHOT_INTERVAL: float = 0.1
+const OBSERVER_RELEVANCE_DISTANCE: float = 96.0
 
 signal local_cast_accepted(attempt_id: String, target: Vector3)
 signal local_cast_rejected(message: String)
@@ -51,7 +53,13 @@ var _last_input_time: Dictionary[int, float] = {}
 var _remote_presentations: Dictionary[int, RemoteFishingPresentation] = {}
 var _pending_local_bait_by_request: Dictionary[String, StringName] = {}
 var _snapshot_accumulator: float = 0.0
+var _observer_snapshot_accumulator: float = 0.0
 var _local_input_sequence: int = 0
+var _presentation_enabled: bool = true
+var _owner_snapshot_packets_sent: int = 0
+var _owner_snapshot_states_sent: int = 0
+var _observer_snapshot_packets_sent: int = 0
+var _observer_snapshot_states_sent: int = 0
 
 
 func setup(
@@ -67,6 +75,7 @@ func setup(
 	item_catalog: ItemCatalog,
 	fish_catalog: FishPoolType,
 	item_use: NetworkItemUseService,
+	presentation_enabled: bool = true,
 ) -> void:
 	_session = session
 	_spawn_service = spawn_service
@@ -80,6 +89,7 @@ func setup(
 	_item_catalog = item_catalog
 	_fish_catalog = fish_catalog
 	_item_use = item_use
+	_presentation_enabled = presentation_enabled
 	if not _session.peer_removed.is_connected(_on_peer_removed):
 		_session.peer_removed.connect(_on_peer_removed)
 	if not _session.state_changed.is_connected(_on_session_state_changed):
@@ -130,6 +140,7 @@ func request_local_cast(
 		"capacity_available": bool(evidence.get("capacity_available", false)),
 		"bait_id": str(bait_id),
 		"lure_id": str(lure_id),
+		"movement_sequence": _session.get_latest_movement_input_sequence(),
 	}
 	if _session.is_host():
 		_handle_cast_request(_session.get_local_peer_id(), data)
@@ -213,15 +224,27 @@ func _process(delta: float) -> void:
 				if now >= attempt.capacity_deadline:
 					_cancel_attempt(peer_id, "Fishing attempt ended.")
 	_snapshot_accumulator += delta
+	_observer_snapshot_accumulator += delta
 	if _snapshot_accumulator >= NetworkFishingProtocol.SNAPSHOT_RATE:
 		_snapshot_accumulator = fmod(
 			_snapshot_accumulator,
 			NetworkFishingProtocol.SNAPSHOT_RATE
 		)
-		_broadcast_snapshots()
+		_broadcast_owner_snapshots()
+	if _observer_snapshot_accumulator >= OBSERVER_SNAPSHOT_INTERVAL:
+		_observer_snapshot_accumulator = fmod(
+			_observer_snapshot_accumulator,
+			OBSERVER_SNAPSHOT_INTERVAL,
+		)
+		_broadcast_observer_snapshots()
 
 
-@rpc("any_peer", "call_remote", "reliable", 0)
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func submit_cast_request(data: Dictionary) -> void:
 	var sender_id: int = multiplayer.get_remote_sender_id()
 	if not _session.is_host() or not _session.is_authenticated_peer(sender_id):
@@ -268,7 +291,13 @@ func _handle_cast_request(peer_id: int, data: Dictionary) -> void:
 		return
 	var origin: Vector3 = NetworkFishingProtocol.array_to_vector3(data["origin"])
 	var target: Vector3 = NetworkFishingProtocol.array_to_vector3(data["target"])
-	var authoritative_origin: Vector3 = avatar.get_cast_origin_position()
+	var movement_state: Dictionary = avatar.get_lag_compensated_movement_state(
+		int(data.get("movement_sequence", 0))
+	)
+	var authoritative_origin: Vector3 = movement_state.get(
+		"cast_origin",
+		avatar.get_cast_origin_position(),
+	)
 	if origin.distance_to(authoritative_origin) > CAST_ORIGIN_TOLERANCE:
 		_record_and_reject(peer_id, request_id, "Cannot fish here.")
 		return
@@ -279,7 +308,10 @@ func _handle_cast_request(peer_id: int, data: Dictionary) -> void:
 		_fishing_spot.maximum_cast_distance,
 		float(data["charge"])
 	)
-	var facing: Vector3 = avatar.get_facing_direction()
+	var facing: Vector3 = movement_state.get(
+		"facing",
+		avatar.get_facing_direction(),
+	)
 	facing.y = 0.0
 	if (
 		cast_offset.length() < _fishing_spot.minimum_cast_distance - 0.25
@@ -642,7 +674,12 @@ func _handle_fishing_input(peer_id: int, data: Dictionary) -> void:
 		attempt.controller.handle_primary_pressed()
 
 
-@rpc("any_peer", "call_remote", "reliable", 0)
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func submit_cancel_request(attempt_id: String) -> void:
 	var sender_id: int = multiplayer.get_remote_sender_id()
 	var attempt: NetworkFishingAttempt = _attempts.get(sender_id)
@@ -687,16 +724,59 @@ func _on_encounter_updated(
 	})
 
 
-func _broadcast_snapshots() -> void:
-	var snapshots: Array[Dictionary] = []
+func _broadcast_owner_snapshots() -> void:
 	for attempt: NetworkFishingAttempt in _attempts.values():
 		var snapshot: Dictionary = attempt.get_meta("snapshot", {})
-		if not snapshot.is_empty():
-			snapshots.append(snapshot)
-	if snapshots.is_empty():
+		if snapshot.is_empty():
+			continue
+		var owner_peer_id: int = attempt.owner_peer_id
+		if owner_peer_id == _session.get_local_peer_id():
+			_apply_snapshots([snapshot])
+		elif owner_peer_id > 1:
+			receive_fishing_snapshots.rpc_id(owner_peer_id, [snapshot])
+			_owner_snapshot_packets_sent += 1
+			_owner_snapshot_states_sent += 1
+
+
+func _broadcast_observer_snapshots() -> void:
+	if _attempts.is_empty():
 		return
-	_apply_snapshots(snapshots)
-	receive_fishing_snapshots.rpc(snapshots)
+	var observer_ids: Array[int] = _session.get_authenticated_peer_ids()
+	for observer_peer_id: int in observer_ids:
+		var observer_avatar: Player = _spawn_service.get_avatar(observer_peer_id)
+		if observer_avatar == null:
+			continue
+		var summaries: Array = []
+		for attempt: NetworkFishingAttempt in _attempts.values():
+			if attempt.owner_peer_id == observer_peer_id:
+				continue
+			var owner_avatar: Player = _spawn_service.get_avatar(
+				attempt.owner_peer_id
+			)
+			if (
+				owner_avatar == null
+				or observer_avatar.global_position.distance_squared_to(
+					owner_avatar.global_position
+				) > OBSERVER_RELEVANCE_DISTANCE * OBSERVER_RELEVANCE_DISTANCE
+			):
+				continue
+			summaries.append([
+				attempt.attempt_id,
+				attempt.owner_peer_id,
+				attempt.bobber_position,
+				int(attempt.phase),
+			])
+		if summaries.is_empty():
+			continue
+		if observer_peer_id == _session.get_local_peer_id():
+			_apply_observer_snapshots(summaries)
+		elif observer_peer_id > 1:
+			receive_fishing_observer_snapshots.rpc_id(
+				observer_peer_id,
+				summaries,
+			)
+			_observer_snapshot_packets_sent += 1
+			_observer_snapshot_states_sent += summaries.size()
 
 
 @rpc("authority", "call_remote", "unreliable_ordered", 4)
@@ -704,6 +784,55 @@ func receive_fishing_snapshots(snapshots: Array) -> void:
 	_apply_snapshots(snapshots)
 
 
+@rpc(
+	"authority",
+	"call_remote",
+	"unreliable_ordered",
+	NetworkFishingProtocol.OBSERVER_SNAPSHOT_CHANNEL,
+)
+func receive_fishing_observer_snapshots(summaries: Array) -> void:
+	_apply_observer_snapshots(summaries)
+
+
+func _apply_observer_snapshots(summaries: Array) -> void:
+	if not _presentation_enabled or summaries.size() > 128:
+		return
+	for value: Variant in summaries:
+		if typeof(value) != TYPE_ARRAY:
+			continue
+		var fields: Array = value
+		if (
+			fields.size() != 4
+			or typeof(fields[0]) != TYPE_STRING
+			or str(fields[0]).is_empty()
+			or str(fields[0]).length() > NetworkFishingProtocol.MAX_ID_LENGTH
+			or typeof(fields[1]) != TYPE_INT
+			or int(fields[1]) <= 0
+			or typeof(fields[2]) != TYPE_VECTOR3
+			or typeof(fields[3]) != TYPE_INT
+			or not (fields[2] as Vector3).is_finite()
+			or int(fields[3]) not in [
+				NetworkFishingAttempt.Phase.WAITING_FOR_BITE,
+				NetworkFishingAttempt.Phase.FIGHTING,
+				NetworkFishingAttempt.Phase.PENDING_CAPACITY,
+			]
+		):
+			continue
+		var owner_peer_id: int = int(fields[1])
+		if owner_peer_id == _session.get_local_peer_id():
+			continue
+		var presentation := _get_remote_presentation(owner_peer_id)
+		if presentation != null:
+			presentation.synchronize_active(
+				str(fields[0]),
+				fields[2],
+				int(fields[3]) in [
+					NetworkFishingAttempt.Phase.FIGHTING,
+					NetworkFishingAttempt.Phase.PENDING_CAPACITY,
+				],
+			)
+
+
 func _apply_snapshots(snapshots: Array) -> void:
 	var local_peer_id: int = _session.get_local_peer_id()
 	for value: Variant in snapshots:
@@ -810,7 +939,12 @@ func _on_attempt_caught(peer_id: int) -> void:
 		receive_capacity_probe.rpc_id(peer_id, probe)
 
 
-@rpc("authority", "call_remote", "reliable", 0)
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func receive_capacity_probe(data: Dictionary) -> void:
 	_handle_local_capacity_probe(data)
 
@@ -844,7 +978,12 @@ func _handle_local_capacity_probe(data: Dictionary) -> void:
 		)
 
 
-@rpc("any_peer", "call_remote", "reliable", 0)
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func submit_capacity_response(
 	attempt_id: String,
 	capacity_nonce: String,
@@ -902,7 +1041,12 @@ func _finalize_catch(attempt: NetworkFishingAttempt) -> void:
 	_dispose_attempt(attempt.owner_peer_id)
 
 
-@rpc("authority", "call_remote", "reliable", 0)
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func receive_target_outcome(data: Dictionary) -> void:
 	_apply_target_outcome(data)
 
@@ -934,7 +1078,7 @@ func _apply_target_outcome(data: Dictionary) -> void:
 			)
 		)
 		_local_inventory.add_catch(fish_catch)
-		_local_collection.mark_quality_discovered(
+		_local_collection.record_catch(
 			fish_id,
 			fish_catch.quality,
 		)
@@ -970,7 +1114,12 @@ func _acknowledge_result(result_id: String, catch_id: StringName) -> void:
 	acknowledge_fishing_result.rpc_id(1, result_id, str(catch_id))
 
 
-@rpc("any_peer", "call_remote", "reliable", 0)
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func acknowledge_fishing_result(result_id: String, catch_id: String) -> void:
 	var sender_id: int = multiplayer.get_remote_sender_id()
 	if (
@@ -1027,7 +1176,12 @@ func _broadcast_public_outcome(
 	receive_public_outcome.rpc(data)
 
 
-@rpc("authority", "call_remote", "reliable", 0)
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func receive_public_outcome(data: Dictionary) -> void:
 	_apply_public_outcome(data)
 
@@ -1079,7 +1233,12 @@ func _broadcast_cast_accepted(data: Dictionary) -> void:
 	receive_cast_accepted.rpc(data)
 
 
-@rpc("authority", "call_remote", "reliable", 0)
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func receive_cast_accepted(data: Dictionary) -> void:
 	_apply_cast_accepted(data)
 
@@ -1114,10 +1273,19 @@ func _apply_cast_accepted(data: Dictionary) -> void:
 	else:
 		var presentation := _get_remote_presentation(peer_id)
 		if presentation != null:
-			presentation.show_cast(origin, target)
+			presentation.show_cast(
+				origin,
+				target,
+				str(data["attempt_id"]),
+			)
 
 
-@rpc("authority", "call_remote", "reliable", 0)
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func receive_bite_pending(data: Dictionary) -> void:
 	_apply_bite_pending(data)
 
@@ -1132,7 +1300,12 @@ func _apply_bite_pending(data: Dictionary) -> void:
 		local_bite_pending.emit(str(data["attempt_id"]))
 
 
-@rpc("authority", "call_remote", "reliable", 0)
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func receive_bite_started(data: Dictionary) -> void:
 	_apply_bite_started(data)
 
@@ -1193,7 +1366,12 @@ func _send_cast_rejected(
 		receive_cast_rejected.rpc_id(peer_id, request_id, message.left(128))
 
 
-@rpc("authority", "call_remote", "reliable", 0)
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkFishingProtocol.RELIABLE_CHANNEL,
+)
 func receive_cast_rejected(request_id: String, message: String) -> void:
 	_pending_local_bait_by_request.erase(request_id)
 	local_cast_rejected.emit(message)
@@ -1228,6 +1406,8 @@ func _resend_request_response(peer_id: int, response: Dictionary) -> void:
 func _get_remote_presentation(
 	peer_id: int,
 ) -> RemoteFishingPresentation:
+	if not _presentation_enabled:
+		return null
 	var existing: RemoteFishingPresentation = _remote_presentations.get(peer_id)
 	if existing != null and is_instance_valid(existing):
 		return existing
@@ -1330,9 +1510,21 @@ func _clear_all() -> void:
 	_last_input_time.clear()
 	_pending_local_bait_by_request.clear()
 	_snapshot_accumulator = 0.0
+	_observer_snapshot_accumulator = 0.0
 	_local_input_sequence = 0
 
 
+func get_network_metrics() -> Dictionary:
+	return {
+		"active_attempts": _attempts.size(),
+		"remote_presentations": _remote_presentations.size(),
+		"owner_snapshot_packets_sent": _owner_snapshot_packets_sent,
+		"owner_snapshot_states_sent": _owner_snapshot_states_sent,
+		"observer_snapshot_packets_sent": _observer_snapshot_packets_sent,
+		"observer_snapshot_states_sent": _observer_snapshot_states_sent,
+	}
+
+
 func _bound_result_ledger() -> void:
 	while _result_ledgers.size() > MAX_LEDGER_ENTRIES_PER_PEER:
 		_result_ledgers.erase(_result_ledgers.keys().front())

network/network_item_use_service.gd

@@ -47,6 +47,7 @@ func setup(
 	_session.peer_removed.connect(_on_peer_removed)
 	_session.state_changed.connect(_on_session_state_changed)
 	_session.peer_authenticated.connect(_on_peer_authenticated)
+	_spawn_service.avatar_spawned.connect(_on_avatar_spawned)
 
 
 func request_use(item_id: StringName) -> String:
@@ -436,6 +437,12 @@ func _on_peer_authenticated(peer_id: int, _display_name: String) -> void:
 		receive_equipped_state.rpc_id(peer_id, state)
 
 
+func _on_avatar_spawned(peer_id: int, _avatar: Player) -> void:
+	var state: Dictionary = _equipped_states.get(peer_id, {})
+	if not state.is_empty():
+		_apply_equipped(state)
+
+
 func _on_peer_removed(peer_id: int) -> void:
 	_requests.erase(peer_id)
 	_pending_by_peer.erase(peer_id)

network/network_player_list_service.gd

@@ -3,6 +3,15 @@ extends Node
 
 signal entries_changed
 signal moderation_finished(success: bool, message: String)
+signal friend_request_received(
+	request_id: String,
+	fingerprint: String,
+	display_name: String,
+)
+signal friend_action_finished(success: bool, message: String)
+
+const MAX_PENDING_FRIEND_REQUESTS := 16
+const FRIEND_REQUEST_ID_LENGTH := 32
 
 var _session: NetworkSession
 var _relationships: PlayerRelationshipStore
@@ -18,6 +27,9 @@ var _peer_fingerprints: Dictionary[int, String] = {}
 var _remote_bans: Array[Dictionary] = []
 var _ban_snapshot_requested: bool = false
 var _ban_snapshot_loaded: bool = false
+var _pending_outgoing_friends: Dictionary[String, Dictionary] = {}
+var _pending_incoming_friends: Dictionary[String, Dictionary] = {}
+var _host_friend_routes: Dictionary[String, Dictionary] = {}
 
 
 func setup(
@@ -74,6 +86,18 @@ func get_entries() -> Array[PlayerListEntry]:
 		entry.ping_to_host_ms = _session.get_peer_rtt_ms(peer_id)
 		entry.muted = _relationships.is_muted(record.identity_fingerprint)
 		entry.blocked = blocked
+		entry.is_friend = _relationships.is_friend(record.identity_fingerprint)
+		entry.can_request_friend = (
+			not entry.is_local_player
+			and not entry.is_friend
+			and not entry.blocked
+			and _session.supports_server_capability(
+				NetworkProtocol.FRIENDS_CAPABILITY
+			)
+			and _session.peer_supports_capability(
+				peer_id, NetworkProtocol.FRIENDS_CAPABILITY
+			)
+		)
 		entry.can_kick = (
 			_session.can_local_moderate()
 			and peer_id != local_id
@@ -125,6 +149,10 @@ func get_relationships() -> Array[Dictionary]:
 	return _relationships.get_records()
 
 
+func get_friends() -> Array[Dictionary]:
+	return _relationships.get_friends()
+
+
 func get_bans() -> Array[Dictionary]:
 	if _session.is_host():
 		return _bans.get_bans(_session.get_host_identity_fingerprint())
@@ -180,6 +208,160 @@ func set_blocked(fingerprint: String, display_name: String, value: bool) -> bool
 	return _relationships.set_blocked(fingerprint, display_name, value)
 
 
+func remove_friend(fingerprint: String, display_name: String) -> bool:
+	if fingerprint == _session.get_local_identity_fingerprint():
+		return false
+	var ok := _relationships.remove_friend(fingerprint, display_name)
+	friend_action_finished.emit(
+		ok,
+		"Friend removed." if ok else "Friend could not be removed.",
+	)
+	return ok
+
+
+func send_friend_request(
+	peer_id: int,
+	fingerprint: String,
+	display_name: String,
+) -> bool:
+	if (
+		_pending_outgoing_friends.size() >= MAX_PENDING_FRIEND_REQUESTS
+		or fingerprint == _session.get_local_identity_fingerprint()
+		or _relationships.is_blocked(fingerprint)
+		or _relationships.is_friend(fingerprint)
+		or not _valid_friend_target(peer_id, fingerprint)
+	):
+		friend_action_finished.emit(
+			false, "This person needs to be online to do this."
+		)
+		return false
+	var local_capabilities: Dictionary = (
+		_relationships.create_friend_capabilities()
+	)
+	if local_capabilities.is_empty():
+		friend_action_finished.emit(false, "Friend request could not be created.")
+		return false
+	var request_id := NetworkIdentityCrypto.secure_id(16)
+	_pending_outgoing_friends[request_id] = {
+		"target_peer_id": peer_id,
+		"target_fingerprint": fingerprint,
+		"target_display_name": display_name,
+		"local_capabilities": local_capabilities,
+	}
+	var public_capabilities := _public_friend_capabilities(local_capabilities)
+	if _session.is_host():
+		_route_friend_request(
+			_session.get_local_peer_id(), peer_id, request_id, public_capabilities
+		)
+	else:
+		request_friendship.rpc_id(
+			1, peer_id, request_id, public_capabilities
+		)
+	friend_action_finished.emit(true, "Friend request sent.")
+	return true
+
+
+func respond_friend_request(request_id: String, accepted: bool) -> bool:
+	var pending: Dictionary = _pending_incoming_friends.get(request_id, {})
+	if pending.is_empty():
+		friend_action_finished.emit(
+			false, "This friend request is no longer available."
+		)
+		return false
+	_pending_incoming_friends.erase(request_id)
+	var local_capabilities: Dictionary = {}
+	var public_capabilities: Dictionary = {}
+	var response_accepted := accepted
+	if accepted:
+		local_capabilities = _relationships.create_friend_capabilities()
+		response_accepted = (
+			not local_capabilities.is_empty()
+			and _relationships.add_friend(
+				str(pending["requester_fingerprint"]),
+				str(pending["requester_display_name"]),
+				local_capabilities,
+				pending["remote_capabilities"],
+			)
+		)
+		if response_accepted:
+			public_capabilities = _public_friend_capabilities(local_capabilities)
+	if _session.is_host():
+		_route_friend_response(
+			_session.get_local_peer_id(),
+			request_id,
+			response_accepted,
+			public_capabilities,
+		)
+	else:
+		respond_friendship.rpc_id(
+			1, request_id, response_accepted, public_capabilities
+		)
+	friend_action_finished.emit(
+		response_accepted or not accepted,
+		"Friend added."
+		if response_accepted
+		else "Friend request declined."
+		if not accepted
+		else "Friend could not be saved.",
+	)
+	return response_accepted or not accepted
+
+
+@rpc("any_peer", "call_remote", "reliable", 0)
+func request_friendship(
+	target_peer_id: int,
+	request_id: String,
+	public_capabilities: Dictionary,
+) -> void:
+	_route_friend_request(
+		multiplayer.get_remote_sender_id(),
+		target_peer_id,
+		request_id,
+		public_capabilities,
+	)
+
+
+@rpc("authority", "call_remote", "reliable", 0)
+func receive_friend_request(
+	request_id: String,
+	requester_fingerprint: String,
+	requester_display_name: String,
+	public_capabilities: Dictionary,
+) -> void:
+	_receive_friend_request_local(
+		request_id,
+		requester_fingerprint,
+		requester_display_name,
+		public_capabilities,
+	)
+
+
+@rpc("any_peer", "call_remote", "reliable", 0)
+func respond_friendship(
+	request_id: String,
+	accepted: bool,
+	public_capabilities: Dictionary,
+) -> void:
+	_route_friend_response(
+		multiplayer.get_remote_sender_id(),
+		request_id,
+		accepted,
+		public_capabilities,
+	)
+
+
+@rpc("authority", "call_remote", "reliable", 0)
+func receive_friend_result(
+	request_id: String,
+	accepted: bool,
+	public_capabilities: Dictionary,
+	message: String,
+) -> void:
+	_receive_friend_result_local(
+		request_id, accepted, public_capabilities, message
+	)
+
+
 func kick(peer_id: int, fingerprint: String, revision: int) -> bool:
 	if _session.is_host():
 		return _kick_on_host(peer_id, fingerprint, revision, true)
@@ -454,6 +636,32 @@ func _on_peer_removed(peer_id: int) -> void:
 		for key: String in _host_block_pairs.keys():
 			if fingerprint in key.split(":"):
 				_host_block_pairs.erase(key)
+		for request_id: String in _host_friend_routes.keys():
+			var route: Dictionary = _host_friend_routes[request_id]
+			if int(route.get("target_peer_id", 0)) == peer_id:
+				_send_friend_result_to_peer(
+					int(route.get("requester_peer_id", 0)),
+					request_id,
+					false,
+					{},
+					"This person needs to be online to do this.",
+				)
+			if (
+				int(route.get("target_peer_id", 0)) == peer_id
+				or int(route.get("requester_peer_id", 0)) == peer_id
+			):
+				_host_friend_routes.erase(request_id)
+	for request_id: String in _pending_outgoing_friends.keys():
+		var pending: Dictionary = _pending_outgoing_friends[request_id]
+		if str(pending.get("target_fingerprint", "")) == fingerprint:
+			_pending_outgoing_friends.erase(request_id)
+			friend_action_finished.emit(
+				false, "This person needs to be online to do this."
+			)
+	for request_id: String in _pending_incoming_friends.keys():
+		var pending: Dictionary = _pending_incoming_friends[request_id]
+		if str(pending.get("requester_fingerprint", "")) == fingerprint:
+			_pending_incoming_friends.erase(request_id)
 	_changed()
 
 
@@ -469,6 +677,9 @@ func _on_session_state_changed(state: NetworkSession.State) -> void:
 		_remote_bans.clear()
 		_ban_snapshot_requested = false
 		_ban_snapshot_loaded = false
+		_pending_outgoing_friends.clear()
+		_pending_incoming_friends.clear()
+		_host_friend_routes.clear()
 	elif state == NetworkSession.State.JOINED_CLIENT:
 		_request_ban_snapshot()
 	_changed()
@@ -506,6 +717,245 @@ func _peer_for_fingerprint(fingerprint: String) -> int:
 	return 0
 
 
+func _valid_friend_target(peer_id: int, fingerprint: String) -> bool:
+	if (
+		not _session.is_gameplay_session_active()
+		or not _session.supports_server_capability(
+			NetworkProtocol.FRIENDS_CAPABILITY
+		)
+		or not _session.peer_supports_capability(
+			peer_id, NetworkProtocol.FRIENDS_CAPABILITY
+		)
+	):
+		return false
+	var record: PeerRegistry.PeerRecord = _session.get_peer_record(peer_id)
+	return (
+		record != null
+		and record.identity_authenticated
+		and record.identity_fingerprint == fingerprint
+	)
+
+
+func _route_friend_request(
+	requester_peer_id: int,
+	target_peer_id: int,
+	request_id: String,
+	public_capabilities: Dictionary,
+) -> void:
+	if (
+		not _session.is_host()
+		or not _valid_friend_request_id(request_id)
+		or not _valid_public_friend_capabilities(public_capabilities)
+		or not _session.is_authenticated_peer(requester_peer_id)
+	):
+		return
+	var requester: PeerRegistry.PeerRecord = (
+		_session.get_peer_record(requester_peer_id)
+	)
+	var target: PeerRegistry.PeerRecord = _session.get_peer_record(target_peer_id)
+	var can_route := (
+		requester != null
+		and target != null
+		and requester_peer_id != target_peer_id
+		and _valid_friend_target(
+			target_peer_id, target.identity_fingerprint
+		)
+		and _session.peer_supports_capability(
+			requester_peer_id, NetworkProtocol.FRIENDS_CAPABILITY
+		)
+		and not pair_is_blocked(
+			requester.identity_fingerprint, target.identity_fingerprint
+		)
+		and _host_friend_routes.size() < (
+			MAX_PENDING_FRIEND_REQUESTS * 8
+		)
+	)
+	if not can_route:
+		_send_friend_result_to_peer(
+			requester_peer_id,
+			request_id,
+			false,
+			{},
+			"This person needs to be online to do this.",
+		)
+		return
+	var requester_pending_count := 0
+	for route: Dictionary in _host_friend_routes.values():
+		if int(route.get("requester_peer_id", 0)) == requester_peer_id:
+			requester_pending_count += 1
+	if requester_pending_count >= MAX_PENDING_FRIEND_REQUESTS:
+		_send_friend_result_to_peer(
+			requester_peer_id,
+			request_id,
+			false,
+			{},
+			"Too many friend requests are pending.",
+		)
+		return
+	_host_friend_routes[request_id] = {
+		"requester_peer_id": requester_peer_id,
+		"target_peer_id": target_peer_id,
+	}
+	if target_peer_id == _session.get_local_peer_id():
+		_receive_friend_request_local(
+			request_id,
+			requester.identity_fingerprint,
+			requester.display_name,
+			public_capabilities,
+		)
+	else:
+		receive_friend_request.rpc_id(
+			target_peer_id,
+			request_id,
+			requester.identity_fingerprint,
+			requester.display_name,
+			public_capabilities,
+		)
+
+
+func _receive_friend_request_local(
+	request_id: String,
+	requester_fingerprint: String,
+	requester_display_name: String,
+	public_capabilities: Dictionary,
+) -> void:
+	if (
+		not _valid_friend_request_id(request_id)
+		or not _valid_public_friend_capabilities(public_capabilities)
+		or not NetworkIdentityCrypto.valid_fingerprint(requester_fingerprint)
+		or not NetworkProfilePreferences.is_valid_display_name(
+			requester_display_name
+		)
+		or _pending_incoming_friends.size()
+			>= MAX_PENDING_FRIEND_REQUESTS
+		or _relationships.is_blocked(requester_fingerprint)
+		or _relationships.is_friend(requester_fingerprint)
+	):
+		_decline_friend_request(request_id)
+		return
+	_pending_incoming_friends[request_id] = {
+		"requester_fingerprint": requester_fingerprint,
+		"requester_display_name": requester_display_name,
+		"remote_capabilities": public_capabilities.duplicate(true),
+	}
+	friend_request_received.emit(
+		request_id, requester_fingerprint, requester_display_name
+	)
+
+
+func _decline_friend_request(request_id: String) -> void:
+	if _session.is_host():
+		_route_friend_response(
+			_session.get_local_peer_id(), request_id, false, {}
+		)
+	elif _session.is_joined_client():
+		respond_friendship.rpc_id(1, request_id, false, {})
+
+
+func _route_friend_response(
+	responder_peer_id: int,
+	request_id: String,
+	accepted: bool,
+	public_capabilities: Dictionary,
+) -> void:
+	if not _session.is_host():
+		return
+	var route: Dictionary = _host_friend_routes.get(request_id, {})
+	if (
+		route.is_empty()
+		or int(route.get("target_peer_id", 0)) != responder_peer_id
+	):
+		return
+	_host_friend_routes.erase(request_id)
+	var response_accepted := (
+		accepted and _valid_public_friend_capabilities(public_capabilities)
+	)
+	_send_friend_result_to_peer(
+		int(route["requester_peer_id"]),
+		request_id,
+		response_accepted,
+		public_capabilities if response_accepted else {},
+		"Friend request accepted."
+		if response_accepted
+		else "Friend request declined."
+		if not accepted
+		else "Friend request could not be completed.",
+	)
+
+
+func _send_friend_result_to_peer(
+	peer_id: int,
+	request_id: String,
+	accepted: bool,
+	public_capabilities: Dictionary,
+	message: String,
+) -> void:
+	if peer_id == _session.get_local_peer_id():
+		_receive_friend_result_local(
+			request_id, accepted, public_capabilities, message
+		)
+	elif _session.is_authenticated_peer(peer_id):
+		receive_friend_result.rpc_id(
+			peer_id,
+			request_id,
+			accepted,
+			public_capabilities,
+			message.left(120),
+		)
+
+
+func _receive_friend_result_local(
+	request_id: String,
+	accepted: bool,
+	public_capabilities: Dictionary,
+	message: String,
+) -> void:
+	var pending: Dictionary = _pending_outgoing_friends.get(request_id, {})
+	if pending.is_empty():
+		return
+	_pending_outgoing_friends.erase(request_id)
+	var success := false
+	if accepted and _valid_public_friend_capabilities(public_capabilities):
+		success = _relationships.add_friend(
+			str(pending["target_fingerprint"]),
+			str(pending["target_display_name"]),
+			pending["local_capabilities"],
+			public_capabilities,
+		)
+		message = "Friend added." if success else "Friend could not be saved."
+	friend_action_finished.emit(success, message.left(120))
+
+
+func _public_friend_capabilities(capabilities: Dictionary) -> Dictionary:
+	return {
+		"presence_channel": str(capabilities.get("presence_channel", "")),
+		"invite_token": str(capabilities.get("invite_token", "")),
+	}
+
+
+func _valid_public_friend_capabilities(capabilities: Dictionary) -> bool:
+	return (
+		_valid_friend_hex(str(capabilities.get("presence_channel", "")))
+		and _valid_friend_hex(str(capabilities.get("invite_token", "")))
+	)
+
+
+func _valid_friend_request_id(request_id: String) -> bool:
+	return _valid_friend_hex(request_id, FRIEND_REQUEST_ID_LENGTH)
+
+
+func _valid_friend_hex(
+	value: String,
+	expected_length: int = PlayerRelationshipStore.SOCIAL_TOKEN_LENGTH,
+) -> bool:
+	if value.length() != expected_length:
+		return false
+	for character: String in value:
+		if character not in "0123456789abcdef":
+			return false
+	return true
+
+
 func _valid_moderation_target(
 	peer_id: int,
 	fingerprint: String,

network/network_protocol.gd

@@ -3,7 +3,7 @@ extends RefCounted
 
 const WorldLayoutType = preload("res://world/world_layout.gd")
 
-const PROTOCOL_VERSION: int = 9
+const PROTOCOL_VERSION: int = 10
 const GAME_BUILD: String = "prealpha"
 const MAX_GAME_VERSION_LENGTH: int = 64
 const MAX_DISPLAY_NAME_LENGTH: int = 24
@@ -13,14 +13,24 @@ const MAX_PUBLIC_KEY_LENGTH: int = 8192
 const MAX_SIGNATURE_LENGTH: int = 2048
 # ENet channels: 0 reliable lifecycle, 1 movement input, 2 movement
 # snapshots, 3 fishing input, 4 fishing snapshots, 5 reliable sales,
-# 6 reliable shop transactions, 7 reliable item/equipment/showcase/drawing,
-# 8 reliable ordered session chat, 9 reliable private session mail.
+# 6 reliable shop transactions, 7 reliable item/equipment,
+# 8 reliable ordered session chat, 9 reliable private session mail,
+# 10 observer-only fishing summaries, 11 reliable movement animation state,
+# 12 reliable fish showcase state, 13 reliable world-spawn events,
+# 14 reliable artwork state, 15 world-spawn snapshots, and 16 reliable
+# fishing lifecycle. Bulk world/art payloads must not head-of-line block held
+# items, fishing, or animation presentation.
 const SALE_RELIABLE_CHANNEL: int = 5
 const SHOP_RELIABLE_CHANNEL: int = 6
 const ITEM_RELIABLE_CHANNEL: int = 7
 const CHAT_RELIABLE_CHANNEL: int = 8
 const MAIL_RELIABLE_CHANNEL: int = 9
-const ENET_CHANNEL_COUNT: int = 10
+const MOVEMENT_ANIMATION_CHANNEL: int = 11
+const SHOWCASE_RELIABLE_CHANNEL: int = 12
+const WORLD_SPAWN_RELIABLE_CHANNEL: int = 13
+const DRAWING_RELIABLE_CHANNEL: int = 14
+const FISHING_RELIABLE_CHANNEL: int = 16
+const ENET_CHANNEL_COUNT: int = 17
 const SURFACE_DRAWING_CAPABILITY: String = "surface_drawing_v2"
 const ART_SHOP_CAPABILITY: String = "art_shop_v1"
 const BACKPACK_SHOP_CAPABILITY: String = "backpack_shop_v1"
@@ -32,6 +42,9 @@ const WORLD_SPAWN_CAPABILITY: String = "world_spawn_envelope_v1"
 const APPEARANCE_PREVIEW_CAPABILITY: String = "appearance_preview_v1"
 const WORLD_GENERATION_CAPABILITY: String = "world_generation_v1"
 const WORLD_LAYOUT_CAPABILITY: String = "world_layout_v1"
+const FRIENDS_CAPABILITY: String = "friends_v1"
+const MOVEMENT_RECONCILIATION_CAPABILITY: String = "movement_reconciliation_v2"
+const FISHING_REPLICATION_CAPABILITY: String = "fishing_replication_v2"
 const DEFAULT_WORLD_SEED: int = 13001
 const MAX_WORLD_SEED: int = 2147483646
 
@@ -186,6 +199,9 @@ static func make_client_hello(
 			BACKPACK_SHOP_CAPABILITY,
 			WORLD_GENERATION_CAPABILITY,
 			WORLD_LAYOUT_CAPABILITY,
+			FRIENDS_CAPABILITY,
+			MOVEMENT_RECONCILIATION_CAPABILITY,
+			FISHING_REPLICATION_CAPABILITY,
 		]),
 		"cosmetic_snapshot": cosmetic_snapshot,
 		"identity_fingerprint": identity_fingerprint,
@@ -323,6 +339,9 @@ static func make_server_hello(
 			APPEARANCE_PREVIEW_CAPABILITY,
 			WORLD_GENERATION_CAPABILITY,
 			WORLD_LAYOUT_CAPABILITY,
+			FRIENDS_CAPABILITY,
+			MOVEMENT_RECONCILIATION_CAPABILITY,
+			FISHING_REPLICATION_CAPABILITY,
 			"chat_v1",
 			"mail_v1",
 			"profile_v1",

network/network_sale_service.gd

@@ -14,6 +14,7 @@ const PlayerBagType = preload("res://inventory/player_bag.gd")
 const ItemResalePolicyType = preload("res://economy/item_resale_policy.gd")
 
 const MAX_LEDGER_ENTRIES_PER_PEER: int = 64
+const LOCAL_REQUEST_RETRY_SECONDS: float = 2.0
 const PELICAN_BUYER_ID: StringName = &"pelicans"
 const MAIN_SHOP_BUYER_ID: StringName = &"main_fishing_shop"
 
@@ -45,6 +46,8 @@ var _acknowledged_results: Dictionary[String, bool] = {}
 var _applied_results: Dictionary[String, bool] = {}
 var _received_results: Dictionary[String, bool] = {}
 var _pending_local_request_id: String = ""
+var _pending_local_request: Dictionary = {}
+var _local_request_retry_elapsed: float = 0.0
 var _pending_local_catch_ids: Array[StringName] = []
 var _pending_local_items: Array[Dictionary] = []
 var _pending_local_buyer_id: StringName
@@ -52,6 +55,25 @@ var _reservations: PlayerAssetReservationService
 var _inventory_layout: PlayerInventoryLayout
 
 
+func _process(delta: float) -> void:
+	if (
+		_pending_local_request.is_empty()
+		or _session == null
+		or not _session.is_joined_client()
+	):
+		_local_request_retry_elapsed = 0.0
+		return
+	_local_request_retry_elapsed += delta
+	if _local_request_retry_elapsed < LOCAL_REQUEST_RETRY_SECONDS:
+		return
+	_local_request_retry_elapsed = 0.0
+	# The request ID is stable and the host ledger is idempotent, so an
+	# application-level retry cannot apply a sale twice. This covers the narrow
+	# transition where a generated client has authenticated but is only just
+	# resuming regular multiplayer polling.
+	submit_sale_request.rpc_id(1, _pending_local_request)
+
+
 func setup(
 	session: NetworkSession,
 	spawn_service: PlayerSpawnService,
@@ -208,6 +230,8 @@ func request_local_mixed_sale(
 		"items": item_evidence,
 	}
 	_pending_local_request_id = request_id
+	_pending_local_request = request.duplicate(true)
+	_local_request_retry_elapsed = 0.0
 	_pending_local_catch_ids = catch_ids.duplicate()
 	_pending_local_items = item_evidence.duplicate(true)
 	_pending_local_buyer_id = buyer.id
@@ -595,6 +619,8 @@ func _finish_local_sale(
 	payout: int,
 ) -> void:
 	_pending_local_request_id = ""
+	_pending_local_request.clear()
+	_local_request_retry_elapsed = 0.0
 	_pending_local_catch_ids.clear()
 	_pending_local_items.clear()
 	_pending_local_buyer_id = StringName()
@@ -730,6 +756,8 @@ func _clear_session_state() -> void:
 	_applied_results.clear()
 	_received_results.clear()
 	_pending_local_request_id = ""
+	_pending_local_request.clear()
+	_local_request_retry_elapsed = 0.0
 	_pending_local_catch_ids.clear()
 	_pending_local_items.clear()
 	_pending_local_buyer_id = StringName()

network/network_session.gd

@@ -7,13 +7,50 @@ const DEFAULT_SESSION_MAX_PLAYERS: int = 8
 const DEFAULT_TRANSPORT_MAX_CLIENTS: int = 31
 const CONNECTION_TIMEOUT_SECONDS: float = 10.0
 const AUTHENTICATION_TIMEOUT_SECONDS: float = 60.0
+const ENET_TIMEOUT_LIMIT: int = 32
+const ENET_TIMEOUT_MINIMUM_MS: int = 10000
+const ENET_TIMEOUT_MAXIMUM_MS: int = 120000
 const INPUT_INTERVAL: float = 1.0 / 30.0
+const IDLE_INPUT_INTERVAL: float = 1.0 / 5.0
 const SNAPSHOT_INTERVAL: float = 1.0 / 30.0
+const NEAR_REMOTE_SNAPSHOT_DIVISOR: int = 2
+const FAR_REMOTE_SNAPSHOT_DIVISOR: int = 6
+const DISTANT_REMOTE_SNAPSHOT_DIVISOR: int = 8
+const NEAR_REMOTE_DISTANCE: float = 48.0
+const FAR_REMOTE_DISTANCE: float = 96.0
 const MOVEMENT_SNAPSHOT_BATCH_SIZE: int = 8
-const MOVEMENT_SNAPSHOT_FIELD_COUNT: int = 13
+const MOVEMENT_INPUT_FIELD_COUNT: int = 8
+const MOVEMENT_SNAPSHOT_FIELD_COUNT: int = 6
+const MOVEMENT_ANIMATION_FIELD_COUNT: int = 7
+const MOVEMENT_ANIMATION_ACTION_FIELD_COUNT: int = 5
+const MAX_PENDING_MOVEMENT_INPUTS: int = 96
+const ANIMATION_REFRESH_INTERVAL: float = 1.0
 const MAX_MOVEMENT_INPUT_SEQUENCE: int = 2147483647
 const MAX_MOVEMENT_ONE_WAY_TRANSIT_SECONDS: float = 0.25
 
+const MOVEMENT_FLAG_JUMP: int = 1 << 0
+const MOVEMENT_FLAG_SPRINT: int = 1 << 1
+const MOVEMENT_FLAG_SNEAK: int = 1 << 2
+const MOVEMENT_FLAG_SLOW_WALK: int = 1 << 3
+const MOVEMENT_FLAG_SITTING: int = 1 << 4
+const MOVEMENT_FLAG_CASTING: int = 1 << 5
+const MOVEMENT_ALLOWED_FLAGS: int = (
+	MOVEMENT_FLAG_JUMP
+	| MOVEMENT_FLAG_SPRINT
+	| MOVEMENT_FLAG_SNEAK
+	| MOVEMENT_FLAG_SLOW_WALK
+	| MOVEMENT_FLAG_SITTING
+	| MOVEMENT_FLAG_CASTING
+)
+const SNAPSHOT_FLAG_GROUNDED: int = 1 << 0
+const SNAPSHOT_FLAG_SITTING: int = 1 << 1
+const SNAPSHOT_FLAG_CASTING: int = 1 << 2
+const SNAPSHOT_ALLOWED_FLAGS: int = (
+	SNAPSHOT_FLAG_GROUNDED
+	| SNAPSHOT_FLAG_SITTING
+	| SNAPSHOT_FLAG_CASTING
+)
+
 signal state_changed(state: State)
 signal status_message_changed(message: String)
 signal connection_error(message: String)
@@ -80,7 +117,21 @@ var _client_nonce: String = ""
 var _current_route: ConnectionRoute
 var _input_sequence: int = 0
 var _input_accumulator: float = 0.0
+var _idle_input_accumulator: float = 0.0
+var _last_input_state_hash: int = 0
+var _pending_movement_inputs: Array[Dictionary] = []
 var _snapshot_accumulator: float = 0.0
+var _movement_snapshot_tick: int = 0
+var _animation_refresh_accumulator: float = 0.0
+var _last_animation_state_by_peer: Dictionary[int, Dictionary] = {}
+var _pending_animation_state_by_peer: Dictionary[int, Dictionary] = {}
+var _last_local_animation_action_signature: Array = []
+var _movement_inputs_sent: int = 0
+var _movement_inputs_received: int = 0
+var _movement_snapshot_packets_sent: int = 0
+var _movement_snapshot_states_sent: int = 0
+var _movement_animation_packets_sent: int = 0
+var _movement_animation_states_sent: int = 0
 var _last_server_max_players: int = DEFAULT_SESSION_MAX_PLAYERS
 var _last_server_player_count: int = 0
 var _last_server_display_name: String = ""
@@ -278,6 +329,9 @@ func _register_player_host() -> void:
 			NetworkProtocol.BACKPACK_SHOP_CAPABILITY,
 			NetworkProtocol.WORLD_GENERATION_CAPABILITY,
 			NetworkProtocol.WORLD_LAYOUT_CAPABILITY,
+			NetworkProtocol.FRIENDS_CAPABILITY,
+			NetworkProtocol.MOVEMENT_RECONCILIATION_CAPABILITY,
+			NetworkProtocol.FISHING_REPLICATION_CAPABILITY,
 		]),
 	)
 	_registry.update_appearance(1, _local_appearance_snapshot)
@@ -602,6 +656,9 @@ func supports_server_capability(capability: StringName) -> bool:
 			NetworkProtocol.APPEARANCE_PREVIEW_CAPABILITY,
 			NetworkProtocol.WORLD_GENERATION_CAPABILITY,
 			NetworkProtocol.WORLD_LAYOUT_CAPABILITY,
+			NetworkProtocol.FRIENDS_CAPABILITY,
+			NetworkProtocol.MOVEMENT_RECONCILIATION_CAPABILITY,
+			NetworkProtocol.FISHING_REPLICATION_CAPABILITY,
 			"chat_v1",
 			"mail_v1",
 			"profile_v1",
@@ -649,6 +706,18 @@ func get_peer_rtt_ms(peer_id: int) -> int:
 	return int(packet_peer.get_statistic(ENetPacketPeer.PEER_ROUND_TRIP_TIME))
 
 
+func get_movement_metrics() -> Dictionary:
+	return {
+		"inputs_sent": _movement_inputs_sent,
+		"inputs_received": _movement_inputs_received,
+		"snapshot_packets_sent": _movement_snapshot_packets_sent,
+		"snapshot_states_sent": _movement_snapshot_states_sent,
+		"animation_packets_sent": _movement_animation_packets_sent,
+		"animation_states_sent": _movement_animation_states_sent,
+		"pending_local_inputs": _pending_movement_inputs.size(),
+	}
+
+
 func kick_authenticated_peer(
 	peer_id: int,
 	fingerprint: String,
@@ -813,6 +882,10 @@ func get_local_peer_id() -> int:
 	return multiplayer.get_unique_id() if is_gameplay_session_active() else 0
 
 
+func get_latest_movement_input_sequence() -> int:
+	return _input_sequence if state == State.JOINED_CLIENT else 0
+
+
 func _process(delta: float) -> void:
 	var now: float = Time.get_ticks_msec() / 1000.0
 	if (
@@ -833,9 +906,13 @@ func _process(delta: float) -> void:
 		return
 	_input_accumulator += delta
 	_snapshot_accumulator += delta
+	_animation_refresh_accumulator += delta
 	if state == State.JOINED_CLIENT and _input_accumulator >= INPUT_INTERVAL:
+		var elapsed_input_time: float = _input_accumulator
 		_input_accumulator = fmod(_input_accumulator, INPUT_INTERVAL)
-		_send_local_input()
+		_idle_input_accumulator += elapsed_input_time
+		_maybe_send_local_animation_action()
+		_maybe_send_local_input()
 	if is_host() and _snapshot_accumulator >= SNAPSHOT_INTERVAL:
 		_snapshot_accumulator = fmod(_snapshot_accumulator, SNAPSHOT_INTERVAL)
 		_broadcast_movement_snapshots()
@@ -873,6 +950,7 @@ func _on_peer_connected(peer_id: int) -> void:
 	if state != State.OPEN_HOST:
 		multiplayer.multiplayer_peer.disconnect_peer(peer_id)
 		return
+	_configure_enet_peer_timeout(peer_id)
 	_pending_authentication[peer_id] = (
 		Time.get_ticks_msec() / 1000.0 + AUTHENTICATION_TIMEOUT_SECONDS
 	)
@@ -881,6 +959,7 @@ func _on_peer_connected(peer_id: int) -> void:
 func _on_connected_to_server() -> void:
 	if state != State.CONNECTING:
 		return
+	_configure_enet_peer_timeout(1)
 	_set_state(State.AUTHENTICATING, "Authenticating...")
 	_client_nonce = NetworkIdentityCrypto.secure_id(32)
 	_client_identity_attempt = NetworkProtocol.make_identity_hello(
@@ -892,6 +971,24 @@ func _on_connected_to_server() -> void:
 	submit_identity_hello.rpc_id(1, _client_identity_attempt)
 
 
+func _configure_enet_peer_timeout(peer_id: int) -> void:
+	var enet := multiplayer.multiplayer_peer as ENetMultiplayerPeer
+	if enet == null:
+		return
+	var packet_peer: ENetPacketPeer = enet.get_peer(peer_id)
+	if packet_peer == null:
+		return
+	# A deterministic generated world can temporarily occupy the client's main
+	# thread during the authenticated join transition. Keep that bounded load
+	# from looking like a dead connection while retaining normal ENet liveness
+	# checks once the client resumes polling.
+	packet_peer.set_timeout(
+		ENET_TIMEOUT_LIMIT,
+		ENET_TIMEOUT_MINIMUM_MS,
+		ENET_TIMEOUT_MAXIMUM_MS,
+	)
+
+
 func _on_connection_failed() -> void:
 	if state not in [
 		State.CONNECTING,
@@ -931,6 +1028,8 @@ func _on_peer_disconnected(peer_id: int) -> void:
 	_pending_identity_challenges.erase(peer_id)
 	_authenticated_identity_cache.erase(peer_id)
 	_operator_peer_ids.erase(peer_id)
+	_last_animation_state_by_peer.erase(peer_id)
+	_pending_animation_state_by_peer.erase(peer_id)
 	var recovery_attempt: String = _recovery_attempts.get(peer_id, "")
 	if not recovery_attempt.is_empty():
 		_recovery_attempts.erase(peer_id)
@@ -1209,6 +1308,16 @@ func submit_client_hello(data: Dictionary) -> void:
 			NetworkProtocol.RejectionCode.UNSUPPORTED_CLIENT,
 		)
 		return
+	for required_capability: String in [
+		NetworkProtocol.MOVEMENT_RECONCILIATION_CAPABILITY,
+		NetworkProtocol.FISHING_REPLICATION_CAPABILITY,
+	]:
+		if required_capability not in client_capabilities:
+			_reject_peer(
+				sender_id,
+				NetworkProtocol.RejectionCode.UNSUPPORTED_CLIENT,
+			)
+			return
 	if (
 		_host_world_layout == WorldLayout.STARTER_ISLAND
 		and NetworkProtocol.WORLD_LAYOUT_CAPABILITY not in client_capabilities
@@ -1432,6 +1541,14 @@ func receive_server_hello(data: Dictionary) -> void:
 		_teardown_peer()
 		_fail("This server does not support generated worlds.")
 		return
+	for required_capability: String in [
+		NetworkProtocol.MOVEMENT_RECONCILIATION_CAPABILITY,
+		NetworkProtocol.FISHING_REPLICATION_CAPABILITY,
+	]:
+		if required_capability not in _server_capabilities:
+			_teardown_peer()
+			_fail("This server is missing required network capabilities.")
+			return
 	if (
 		received_world_layout == WorldLayout.STARTER_ISLAND
 		and NetworkProtocol.WORLD_LAYOUT_CAPABILITY not in _server_capabilities
@@ -1456,6 +1573,9 @@ func receive_server_hello(data: Dictionary) -> void:
 			NetworkProtocol.BACKPACK_SHOP_CAPABILITY,
 			NetworkProtocol.WORLD_GENERATION_CAPABILITY,
 			NetworkProtocol.WORLD_LAYOUT_CAPABILITY,
+			NetworkProtocol.FRIENDS_CAPABILITY,
+			NetworkProtocol.MOVEMENT_RECONCILIATION_CAPABILITY,
+			NetworkProtocol.FISHING_REPLICATION_CAPABILITY,
 		]),
 	)
 	_registry.update_appearance(local_peer_id, _local_appearance_snapshot)
@@ -1501,6 +1621,8 @@ func receive_peer_despawn(peer_id: int) -> void:
 	if state != State.JOINED_CLIENT:
 		return
 	_operator_peer_ids.erase(peer_id)
+	_last_animation_state_by_peer.erase(peer_id)
+	_pending_animation_state_by_peer.erase(peer_id)
 	_registry.remove_peer(peer_id)
 	_spawn_service.remove_peer(peer_id)
 	peer_removed.emit(peer_id)
@@ -1542,6 +1664,10 @@ func _apply_spawn_entry(entry: Dictionary) -> void:
 		or typeof(entry.get("display_name")) != TYPE_STRING
 		or typeof(entry.get("position")) != TYPE_ARRAY
 		or typeof(entry.get("yaw")) not in [TYPE_FLOAT, TYPE_INT]
+		or typeof(entry.get("sitting")) != TYPE_BOOL
+		or not NetworkPlayerAnimationProtocol.validate_state(
+			entry.get("animation_state")
+		)
 	):
 		return
 	if not _verify_spawn_identity(entry):
@@ -1557,6 +1683,7 @@ func _apply_spawn_entry(entry: Dictionary) -> void:
 				)
 				own_snapshot["position"] = own_position
 				own_snapshot["visual_yaw"] = float(entry["yaw"])
+				own_snapshot["sitting"] = bool(entry["sitting"])
 				own_avatar.apply_network_teleport(own_snapshot)
 		return
 	var peer_was_added: bool = false
@@ -1603,6 +1730,18 @@ func _apply_spawn_entry(entry: Dictionary) -> void:
 	var record := _registry.get_peer(peer_id)
 	if avatar != null and record != null:
 		avatar.apply_appearance_snapshot(record.appearance_snapshot)
+		# Lifecycle and animation updates use separate reliable ENet channels.
+		# Include the current authoritative state in the spawn record so a peer
+		# joining during an action presents it immediately, regardless of which
+		# channel arrives first.
+		avatar.apply_network_animation_state(entry["animation_state"])
+		avatar.apply_network_sitting_state(bool(entry["sitting"]))
+		var pending_animation: Dictionary = (
+			_pending_animation_state_by_peer.get(peer_id, {})
+		)
+		if not pending_animation.is_empty():
+			avatar.apply_network_animation_state(pending_animation)
+			_pending_animation_state_by_peer.erase(peer_id)
 	if peer_was_added and record != null:
 		peer_authenticated.emit(peer_id, record.display_name)
 
@@ -1627,6 +1766,15 @@ func _make_spawn_entry(
 	transform: Transform3D,
 ) -> Dictionary:
 	var record: PeerRegistry.PeerRecord = _registry.get_peer(peer_id)
+	var avatar: Player = _spawn_service.get_avatar(peer_id)
+	var animation_state: Dictionary = (
+		avatar.make_network_animation_state()
+		if avatar != null
+		else NetworkPlayerAnimationProtocol.make_state(
+			NetworkPlayerAnimationProtocol.LOCOMOTION_IDLE,
+			true,
+		)
+	)
 	return {
 		"peer_id": peer_id,
 		"profile_id": record.profile_id if record != null else "",
@@ -1637,6 +1785,12 @@ func _make_spawn_entry(
 			transform.origin.z,
 		],
 		"yaw": transform.basis.get_euler().y,
+		"sitting": (
+			avatar.get_network_sitting_state()
+			if avatar != null
+			else false
+		),
+		"animation_state": animation_state,
 		"appearance": (
 			record.appearance_snapshot.duplicate(true)
 			if record != null
@@ -1862,14 +2016,69 @@ func _fail_identity(message: String) -> void:
 	_fail(message)
 
 
+func _maybe_send_local_input() -> void:
+	var peer_id: int = multiplayer.get_unique_id()
+	var avatar: Player = _spawn_service.get_avatar(peer_id)
+	if avatar == null:
+		return
+	var state_hash: int = avatar.get_network_input_state_hash()
+	var state_changed: bool = state_hash != _last_input_state_hash
+	if (
+		avatar.has_active_network_input()
+		or state_changed
+		or _idle_input_accumulator >= IDLE_INPUT_INTERVAL
+	):
+		_send_local_input()
+		_last_input_state_hash = state_hash
+		_idle_input_accumulator = 0.0
+
+
 func _send_local_input() -> void:
 	var peer_id: int = multiplayer.get_unique_id()
 	var avatar: Player = _spawn_service.get_avatar(peer_id)
 	if avatar == null:
 		return
-	_input_sequence += 1
+	_input_sequence = (
+		1
+		if _input_sequence >= MAX_MOVEMENT_INPUT_SEQUENCE
+		else _input_sequence + 1
+	)
 	var input: Dictionary = avatar.capture_network_input(_input_sequence)
-	submit_movement_input.rpc_id(1, input)
+	var encoded: Array = _encode_movement_input(input)
+	if encoded.is_empty():
+		return
+	_pending_movement_inputs.append(input.duplicate(true))
+	while _pending_movement_inputs.size() > MAX_PENDING_MOVEMENT_INPUTS:
+		_pending_movement_inputs.pop_front()
+	_movement_inputs_sent += 1
+	submit_movement_input.rpc_id(1, encoded)
+
+
+func _maybe_send_local_animation_action() -> void:
+	var peer_id: int = multiplayer.get_unique_id()
+	var avatar: Player = _spawn_service.get_avatar(peer_id)
+	if avatar == null:
+		return
+	var state: Dictionary = avatar.make_network_animation_state()
+	var action: Dictionary = state.get("action", {})
+	if not NetworkPlayerAnimationProtocol.validate_action_state(action):
+		return
+	var signature: Array = [
+		str(action["id"]),
+		int(action["sequence"]),
+		bool(action.get("paused", false)),
+		avatar.get_network_sitting_state(),
+	]
+	if signature == _last_local_animation_action_signature:
+		return
+	var encoded: Array = _encode_movement_animation_action(
+		action,
+		avatar.get_network_sitting_state(),
+	)
+	if encoded.is_empty():
+		return
+	_last_local_animation_action_signature = signature
+	submit_movement_animation_action.rpc_id(1, encoded)
 
 
 func submit_neutral_local_movement() -> void:
@@ -1879,17 +2088,112 @@ func submit_neutral_local_movement() -> void:
 
 
 @rpc("any_peer", "call_remote", "unreliable_ordered", 1)
-func submit_movement_input(data: Dictionary) -> void:
+func submit_movement_input(encoded: Array) -> void:
 	var sender_id: int = multiplayer.get_remote_sender_id()
 	if not is_host() or not _registry.has_peer(sender_id):
 		return
 	var avatar: Player = _spawn_service.get_avatar(sender_id)
-	if avatar == null or not _is_valid_movement_input(data):
+	var data: Dictionary = _decode_movement_input(encoded)
+	if avatar == null or data.is_empty():
 		return
-	avatar.apply_authoritative_network_input(data)
+	_movement_inputs_received += 1
+	avatar.apply_authoritative_network_input(
+		data,
+		Player.resolve_network_input_stale_timeout_seconds(
+			get_peer_rtt_ms(sender_id)
+		),
+	)
 
 
-func _is_valid_movement_input(data: Dictionary) -> bool:
+@rpc(
+	"any_peer",
+	"call_remote",
+	"reliable",
+	NetworkProtocol.MOVEMENT_ANIMATION_CHANNEL,
+)
+func submit_movement_animation_action(encoded: Array) -> void:
+	var sender_id: int = multiplayer.get_remote_sender_id()
+	if not is_host() or not _registry.has_peer(sender_id):
+		return
+	var avatar: Player = _spawn_service.get_avatar(sender_id)
+	var state: Dictionary = _decode_movement_animation_action(encoded)
+	if avatar == null or state.is_empty():
+		return
+	avatar.apply_authoritative_network_animation_action(state["action"])
+	avatar.apply_authoritative_network_sitting_state(bool(state["sitting"]))
+
+
+static func _encode_movement_input(data: Dictionary) -> Array:
+	if not _is_valid_movement_input(data):
+		return []
+	var axis: Array = data["axis"]
+	var flags: int = 0
+	if bool(data["jump"]):
+		flags |= MOVEMENT_FLAG_JUMP
+	if bool(data["sprint"]):
+		flags |= MOVEMENT_FLAG_SPRINT
+	if bool(data["sneak"]):
+		flags |= MOVEMENT_FLAG_SNEAK
+	if bool(data["slow_walk"]):
+		flags |= MOVEMENT_FLAG_SLOW_WALK
+	if bool(data["sitting"]):
+		flags |= MOVEMENT_FLAG_SITTING
+	if bool(data["casting"]):
+		flags |= MOVEMENT_FLAG_CASTING
+	var action: Dictionary = data["animation_action"]
+	return [
+		int(data["sequence"]),
+		Vector2(float(axis[0]), float(axis[1])),
+		float(data["camera_yaw"]),
+		flags,
+		str(action["id"]),
+		int(action["sequence"]),
+		float(action["elapsed"]),
+		bool(action.get("paused", false)),
+	]
+
+
+static func _decode_movement_input(value: Variant) -> Dictionary:
+	if typeof(value) != TYPE_ARRAY:
+		return {}
+	var fields: Array = value
+	if (
+		fields.size() != MOVEMENT_INPUT_FIELD_COUNT
+		or typeof(fields[0]) != TYPE_INT
+		or typeof(fields[1]) != TYPE_VECTOR2
+		or typeof(fields[2]) not in [TYPE_FLOAT, TYPE_INT]
+		or typeof(fields[3]) != TYPE_INT
+		or typeof(fields[4]) not in [TYPE_STRING, TYPE_STRING_NAME]
+		or typeof(fields[5]) != TYPE_INT
+		or typeof(fields[6]) not in [TYPE_FLOAT, TYPE_INT]
+		or typeof(fields[7]) != TYPE_BOOL
+	):
+		return {}
+	var axis: Vector2 = fields[1]
+	var flags: int = int(fields[3])
+	if flags < 0 or flags & ~MOVEMENT_ALLOWED_FLAGS != 0:
+		return {}
+	var decoded: Dictionary = {
+		"sequence": int(fields[0]),
+		"axis": [axis.x, axis.y],
+		"camera_yaw": float(fields[2]),
+		"jump": bool(flags & MOVEMENT_FLAG_JUMP),
+		"sprint": bool(flags & MOVEMENT_FLAG_SPRINT),
+		"sneak": bool(flags & MOVEMENT_FLAG_SNEAK),
+		"slow_walk": bool(flags & MOVEMENT_FLAG_SLOW_WALK),
+		"sitting": bool(flags & MOVEMENT_FLAG_SITTING),
+		"casting": bool(flags & MOVEMENT_FLAG_CASTING),
+		"animation_action": NetworkPlayerAnimationProtocol.make_action_state(
+			StringName(str(fields[4])),
+			int(fields[5]),
+			float(fields[6]),
+			bool(fields[7]),
+		),
+	}
+	return decoded if _is_valid_movement_input(decoded) else {}
+
+
+static func _is_valid_movement_input(data: Dictionary) -> bool:
 	if (
 		typeof(data.get("sequence")) != TYPE_INT
 		or typeof(data.get("axis")) != TYPE_ARRAY
@@ -1929,49 +2233,142 @@ func _is_valid_movement_input(data: Dictionary) -> bool:
 
 
 func _broadcast_movement_snapshots() -> void:
-	var snapshots: Array = []
-	for peer_id: int in _registry.get_peer_ids():
-		var avatar: Player = _spawn_service.get_avatar(peer_id)
-		if avatar == null:
+	_movement_snapshot_tick += 1
+	var peer_ids: Array[int] = _registry.get_peer_ids()
+	for recipient_id: int in peer_ids:
+		# Peer 1 is the local listen-server player. Dedicated servers do not
+		# register a peer 1, so every remaining record is a remote recipient.
+		if recipient_id == 1:
 			continue
-		var encoded: Array = _encode_movement_snapshot(
-			avatar.make_network_snapshot(peer_id)
-		)
-		if not encoded.is_empty():
-			snapshots.append(encoded)
-	# The compact v5 representation keeps a normal eight-player update near
-	# 1 KiB instead of the roughly 3.7 KiB dictionary representation. Rooms
-	# configured above the normal cap are divided into the same safe size.
+		var recipient_avatar: Player = _spawn_service.get_avatar(recipient_id)
+		if recipient_avatar == null:
+			continue
+		var snapshots: Array = []
+		for subject_id: int in peer_ids:
+			var subject_avatar: Player = _spawn_service.get_avatar(subject_id)
+			if subject_avatar == null:
+				continue
+			if (
+				subject_id != recipient_id
+				and not _should_send_remote_snapshot(
+					recipient_avatar.global_position,
+					subject_avatar.global_position,
+				)
+			):
+				continue
+			var encoded: Array = _encode_movement_snapshot(
+				subject_avatar.make_network_snapshot(subject_id)
+			)
+			if not encoded.is_empty():
+				snapshots.append(encoded)
+		_send_movement_snapshot_batches(recipient_id, snapshots)
+	_broadcast_movement_animation_updates(peer_ids)
+
+
+func _should_send_remote_snapshot(
+	recipient_position: Vector3,
+	subject_position: Vector3,
+) -> bool:
+	var distance_squared: float = recipient_position.distance_squared_to(
+		subject_position
+	)
+	var divisor: int = DISTANT_REMOTE_SNAPSHOT_DIVISOR
+	if distance_squared <= NEAR_REMOTE_DISTANCE * NEAR_REMOTE_DISTANCE:
+		divisor = NEAR_REMOTE_SNAPSHOT_DIVISOR
+	elif distance_squared <= FAR_REMOTE_DISTANCE * FAR_REMOTE_DISTANCE:
+		divisor = FAR_REMOTE_SNAPSHOT_DIVISOR
+	return _movement_snapshot_tick % divisor == 0
+
+
+func _send_movement_snapshot_batches(
+	recipient_id: int,
+	snapshots: Array,
+) -> void:
 	for start_index: int in range(
 		0,
 		snapshots.size(),
 		MOVEMENT_SNAPSHOT_BATCH_SIZE,
 	):
-		receive_movement_snapshots.rpc(
-			snapshots.slice(
-				start_index,
-				mini(
-					start_index + MOVEMENT_SNAPSHOT_BATCH_SIZE,
-					snapshots.size(),
-				),
-			)
+		var batch: Array = snapshots.slice(
+			start_index,
+			mini(
+				start_index + MOVEMENT_SNAPSHOT_BATCH_SIZE,
+				snapshots.size(),
+			),
 		)
+		receive_movement_snapshots.rpc_id(recipient_id, batch)
+		_movement_snapshot_packets_sent += 1
+		_movement_snapshot_states_sent += batch.size()
+
+
+func _broadcast_movement_animation_updates(peer_ids: Array[int]) -> void:
+	var refresh_all: bool = (
+		_animation_refresh_accumulator >= ANIMATION_REFRESH_INTERVAL
+	)
+	if refresh_all:
+		_animation_refresh_accumulator = fmod(
+			_animation_refresh_accumulator,
+			ANIMATION_REFRESH_INTERVAL,
+		)
+	var updates: Array = []
+	for subject_id: int in peer_ids:
+		var avatar: Player = _spawn_service.get_avatar(subject_id)
+		if avatar == null:
+			continue
+		var state: Dictionary = avatar.make_network_animation_state()
+		var previous: Dictionary = _last_animation_state_by_peer.get(
+			subject_id, {}
+		)
+		if (
+			not refresh_all
+			and _movement_animation_signature(state)
+				== _movement_animation_signature(previous)
+		):
+			continue
+		_last_animation_state_by_peer[subject_id] = state.duplicate(true)
+		var encoded: Array = _encode_movement_animation(subject_id, state)
+		if not encoded.is_empty():
+			updates.append(encoded)
+	if updates.is_empty():
+		return
+	for recipient_id: int in peer_ids:
+		if recipient_id == 1:
+			continue
+		receive_movement_animations.rpc_id(recipient_id, updates)
+		_movement_animation_packets_sent += 1
+		_movement_animation_states_sent += updates.size()
+
+
+static func _movement_animation_signature(state: Dictionary) -> Array:
+	if not NetworkPlayerAnimationProtocol.validate_state(state):
+		return []
+	var action: Dictionary = state["action"]
+	return [
+		str(state["locomotion_id"]),
+		bool(state["grounded"]),
+		str(action["id"]),
+		int(action["sequence"]),
+		bool(action.get("paused", false)),
+	]
 
 
 static func _encode_movement_snapshot(snapshot: Dictionary) -> Array:
 	var position: Variant = snapshot.get("position")
 	var snapshot_velocity: Variant = snapshot.get("velocity")
-	var animation_value: Variant = snapshot.get("animation_state")
 	if (
 		typeof(position) != TYPE_ARRAY
 		or position.size() != 3
 		or typeof(snapshot_velocity) != TYPE_ARRAY
 		or snapshot_velocity.size() != 3
-		or not NetworkPlayerAnimationProtocol.validate_state(animation_value)
 	):
 		return []
-	var animation: Dictionary = animation_value
-	var action: Dictionary = animation["action"]
+	var flags: int = 0
+	if bool(snapshot.get("grounded", false)):
+		flags |= SNAPSHOT_FLAG_GROUNDED
+	if bool(snapshot.get("sitting", false)):
+		flags |= SNAPSHOT_FLAG_SITTING
+	if bool(snapshot.get("casting", false)):
+		flags |= SNAPSHOT_FLAG_CASTING
 	return [
 		int(snapshot.get("peer_id", 0)),
 		int(snapshot.get("acknowledged_input", 0)),
@@ -1982,14 +2379,7 @@ static func _encode_movement_snapshot(snapshot: Dictionary) -> Array:
 			float(snapshot_velocity[2]),
 		),
 		float(snapshot.get("visual_yaw", 0.0)),
-		str(animation["locomotion_id"]),
-		bool(animation["grounded"]),
-		str(action["id"]),
-		int(action["sequence"]),
-		float(action["elapsed"]),
-		bool(action.get("paused", false)),
-		bool(snapshot.get("sitting", false)),
-		bool(snapshot.get("casting", false)),
+		flags,
 	]
 
 
@@ -2004,18 +2394,22 @@ static func _decode_movement_snapshot(value: Variant) -> Dictionary:
 		or typeof(fields[2]) != TYPE_VECTOR3
 		or typeof(fields[3]) != TYPE_VECTOR3
 		or typeof(fields[4]) not in [TYPE_FLOAT, TYPE_INT]
-		or typeof(fields[5]) not in [TYPE_STRING, TYPE_STRING_NAME]
-		or typeof(fields[6]) != TYPE_BOOL
-		or typeof(fields[7]) not in [TYPE_STRING, TYPE_STRING_NAME]
-		or typeof(fields[8]) != TYPE_INT
-		or typeof(fields[9]) not in [TYPE_FLOAT, TYPE_INT]
-		or typeof(fields[10]) != TYPE_BOOL
-		or typeof(fields[11]) != TYPE_BOOL
-		or typeof(fields[12]) != TYPE_BOOL
+		or typeof(fields[5]) != TYPE_INT
 	):
 		return {}
 	var position: Vector3 = fields[2]
 	var snapshot_velocity: Vector3 = fields[3]
+	var flags: int = int(fields[5])
+	if (
+		int(fields[0]) <= 0
+		or int(fields[1]) < 0
+		or not position.is_finite()
+		or not snapshot_velocity.is_finite()
+		or not is_finite(float(fields[4]))
+		or flags < 0
+		or flags & ~SNAPSHOT_ALLOWED_FLAGS != 0
+	):
+		return {}
 	return {
 		"peer_id": int(fields[0]),
 		"acknowledged_input": int(fields[1]),
@@ -2026,19 +2420,100 @@ static func _decode_movement_snapshot(value: Variant) -> Dictionary:
 			snapshot_velocity.z,
 		],
 		"visual_yaw": float(fields[4]),
-		"animation_state": NetworkPlayerAnimationProtocol.make_state(
-			StringName(str(fields[5])),
-			bool(fields[6]),
-			StringName(str(fields[7])),
-			int(fields[8]),
-			float(fields[9]),
-			bool(fields[10]),
-		),
-		"sitting": bool(fields[11]),
-		"casting": bool(fields[12]),
+		"grounded": bool(flags & SNAPSHOT_FLAG_GROUNDED),
+		"sitting": bool(flags & SNAPSHOT_FLAG_SITTING),
+		"casting": bool(flags & SNAPSHOT_FLAG_CASTING),
 	}
 
 
+static func _encode_movement_animation(
+	peer_id: int,
+	state: Dictionary,
+) -> Array:
+	if not NetworkPlayerAnimationProtocol.validate_state(state):
+		return []
+	var action: Dictionary = state["action"]
+	return [
+		peer_id,
+		str(state["locomotion_id"]),
+		bool(state["grounded"]),
+		str(action["id"]),
+		int(action["sequence"]),
+		float(action["elapsed"]),
+		bool(action.get("paused", false)),
+	]
+
+
+static func _decode_movement_animation(value: Variant) -> Dictionary:
+	if typeof(value) != TYPE_ARRAY:
+		return {}
+	var fields: Array = value
+	if (
+		fields.size() != MOVEMENT_ANIMATION_FIELD_COUNT
+		or typeof(fields[0]) != TYPE_INT
+		or typeof(fields[1]) not in [TYPE_STRING, TYPE_STRING_NAME]
+		or typeof(fields[2]) != TYPE_BOOL
+		or typeof(fields[3]) not in [TYPE_STRING, TYPE_STRING_NAME]
+		or typeof(fields[4]) != TYPE_INT
+		or typeof(fields[5]) not in [TYPE_FLOAT, TYPE_INT]
+		or typeof(fields[6]) != TYPE_BOOL
+	):
+		return {}
+	var state: Dictionary = NetworkPlayerAnimationProtocol.make_state(
+		StringName(str(fields[1])),
+		bool(fields[2]),
+		StringName(str(fields[3])),
+		int(fields[4]),
+		float(fields[5]),
+		bool(fields[6]),
+	)
+	if (
+		int(fields[0]) <= 0
+		or not NetworkPlayerAnimationProtocol.validate_state(state)
+	):
+		return {}
+	return {"peer_id": int(fields[0]), "state": state}
+
+
+static func _encode_movement_animation_action(
+	action: Dictionary,
+	sitting: bool,
+) -> Array:
+	if not NetworkPlayerAnimationProtocol.validate_action_state(action):
+		return []
+	return [
+		str(action["id"]),
+		int(action["sequence"]),
+		float(action["elapsed"]),
+		bool(action.get("paused", false)),
+		sitting,
+	]
+
+
+static func _decode_movement_animation_action(value: Variant) -> Dictionary:
+	if typeof(value) != TYPE_ARRAY:
+		return {}
+	var fields: Array = value
+	if (
+		fields.size() != MOVEMENT_ANIMATION_ACTION_FIELD_COUNT
+		or typeof(fields[0]) not in [TYPE_STRING, TYPE_STRING_NAME]
+		or typeof(fields[1]) != TYPE_INT
+		or typeof(fields[2]) not in [TYPE_FLOAT, TYPE_INT]
+		or typeof(fields[3]) != TYPE_BOOL
+		or typeof(fields[4]) != TYPE_BOOL
+	):
+		return {}
+	var action: Dictionary = NetworkPlayerAnimationProtocol.make_action_state(
+		StringName(str(fields[0])),
+		int(fields[1]),
+		float(fields[2]),
+		bool(fields[3]),
+	)
+	if not NetworkPlayerAnimationProtocol.validate_action_state(action):
+		return {}
+	return {"action": action, "sitting": bool(fields[4])}
+
+
 @rpc("authority", "call_remote", "unreliable_ordered", 2)
 func receive_movement_snapshots(encoded_snapshots: Array) -> void:
 	if state != State.JOINED_CLIENT:
@@ -2058,11 +2533,13 @@ func receive_movement_snapshots(encoded_snapshots: Array) -> void:
 		if avatar == null:
 			continue
 		if peer_id == local_peer_id:
+			_discard_acknowledged_movement_inputs(
+				int(snapshot.get("acknowledged_input", 0))
+			)
 			avatar.apply_local_prediction_correction(
 				snapshot,
-				_input_sequence,
+				_pending_movement_inputs,
 				INPUT_INTERVAL,
-				estimated_transit_seconds,
 			)
 		else:
 			avatar.push_network_snapshot(
@@ -2071,6 +2548,39 @@ func receive_movement_snapshots(encoded_snapshots: Array) -> void:
 			)
 
 
+func _discard_acknowledged_movement_inputs(acknowledged_sequence: int) -> void:
+	while (
+		not _pending_movement_inputs.is_empty()
+		and int(_pending_movement_inputs[0].get("sequence", 0))
+			<= acknowledged_sequence
+	):
+		_pending_movement_inputs.pop_front()
+
+
+@rpc(
+	"authority",
+	"call_remote",
+	"reliable",
+	NetworkProtocol.MOVEMENT_ANIMATION_CHANNEL,
+)
+func receive_movement_animations(encoded_states: Array) -> void:
+	if state != State.JOINED_CLIENT:
+		return
+	for value: Variant in encoded_states:
+		var decoded: Dictionary = _decode_movement_animation(value)
+		if decoded.is_empty():
+			continue
+		var avatar: Player = _spawn_service.get_avatar(
+			int(decoded["peer_id"])
+		)
+		if avatar == null:
+			_pending_animation_state_by_peer[int(decoded["peer_id"])] = (
+				Dictionary(decoded["state"]).duplicate(true)
+			)
+			continue
+		avatar.apply_network_animation_state(decoded["state"])
+
+
 func _estimated_movement_transit_seconds() -> float:
 	if state != State.JOINED_CLIENT:
 		return -1.0
@@ -2346,7 +2856,15 @@ func _teardown_peer() -> void:
 	_connection_deadline = 0.0
 	_input_sequence = 0
 	_input_accumulator = 0.0
+	_idle_input_accumulator = 0.0
+	_last_input_state_hash = 0
+	_pending_movement_inputs.clear()
 	_snapshot_accumulator = 0.0
+	_movement_snapshot_tick = 0
+	_animation_refresh_accumulator = 0.0
+	_last_animation_state_by_peer.clear()
+	_pending_animation_state_by_peer.clear()
+	_last_local_animation_action_signature.clear()
 	_server_capabilities = PackedStringArray()
 	_server_identity_fingerprint = ""
 	_server_identity_public_key = ""

network/network_surface_drawing_service.gd

@@ -52,6 +52,7 @@ var _canvas_states: Dictionary[String, Dictionary] = {}
 var _canvas_nodes: Dictionary[String, SurfaceDrawingCanvas] = {}
 var _selected_canvas_id: String = ""
 var _hovered_canvas_id: String = ""
+var _last_hovered_canvas_id: String = ""
 var _brush_preview: MultiMeshInstance3D
 var _brush_preview_multimesh: MultiMesh
 var _brush_preview_material: ShaderMaterial
@@ -236,6 +237,7 @@ func activate(
 		return
 	_active = true
 	_placing_grid = false
+	_last_hovered_canvas_id = ""
 	_clear_stamp_mode()
 	_eraser_mode = false
 	_clear_armed_guide_action(false)
@@ -267,6 +269,7 @@ func deactivate() -> void:
 	_reset_stroke()
 	_selected_canvas_id = ""
 	_hovered_canvas_id = ""
+	_last_hovered_canvas_id = ""
 	_aim_hit.clear()
 	_hide_previews()
 	_refresh_stencil_visibility()
@@ -849,10 +852,22 @@ func select_saved_stamp(path: String) -> bool:
 
 func export_aimed_canvas() -> String:
 	_update_aim()
-	if _hovered_canvas_id.is_empty():
+	var canvas_id: String = (
+		_hovered_canvas_id
+		if not _hovered_canvas_id.is_empty()
+		else _last_hovered_canvas_id
+	)
+	var canvas: SurfaceDrawingCanvas = _canvas_nodes.get(canvas_id)
+	if (
+		canvas_id.is_empty()
+		or canvas == null
+		or not is_instance_valid(canvas)
+		or canvas.is_hidden_by_relationship()
+	):
+		_last_hovered_canvas_id = ""
 		_emit_hud_state("aim at artwork before exporting")
 		return ""
-	return export_canvas_png(_hovered_canvas_id)
+	return export_canvas_png(canvas_id)
 
 
 func export_canvas_png(canvas_id: String) -> String:
@@ -1070,6 +1085,8 @@ func _update_aim() -> void:
 				selected_layer = layer
 	_selected_canvas_id = found_canvas_id
 	_hovered_canvas_id = found_hovered_canvas_id
+	if not _hovered_canvas_id.is_empty():
+		_last_hovered_canvas_id = _hovered_canvas_id
 	if previous_canvas_id != _selected_canvas_id:
 		_reset_stroke()
 	_update_previews()
@@ -2865,6 +2882,7 @@ func _clear_session_artwork_state() -> void:
 	_canvas_sequence = 0
 	_selected_canvas_id = ""
 	_hovered_canvas_id = ""
+	_last_hovered_canvas_id = ""
 	_stroke_history_by_peer.clear()
 	_cell_last_stroke.clear()
 	_last_local_stroke_id = ""
@@ -2900,6 +2918,8 @@ func _remove_canvas_state(canvas_id: String) -> void:
 		_selected_canvas_id = ""
 	if _hovered_canvas_id == canvas_id:
 		_hovered_canvas_id = ""
+	if _last_hovered_canvas_id == canvas_id:
+		_last_hovered_canvas_id = ""
 
 
 func _emit_artwork_changed() -> void:

network/network_transport.gd

@@ -22,6 +22,25 @@ func connect_to_route(_route: ConnectionRoute) -> Error:
 
 func disconnect_transport() -> void:
 	if _peer != null:
+		# Give ENet one poll cycle to emit its graceful disconnect packets before
+		# closing the socket. This keeps a host from retaining a departed player
+		# until the extended generated-world liveness timeout expires.
+		if (
+			_peer is ENetMultiplayerPeer
+			and _peer.get_connection_status()
+				== MultiplayerPeer.CONNECTION_CONNECTED
+		):
+			var enet_peer := _peer as ENetMultiplayerPeer
+			# A host explicitly notifies each still-live client before closing.
+			# Clients rely on ENetMultiplayerPeer.close(); their cached SceneTree
+			# peer list can briefly outlive the underlying ENet peer after a remote
+			# disconnect, making disconnect_peer() race and report an engine error.
+			if enet_peer.get_unique_id() == 1:
+				for peer_id: int in multiplayer.get_peers():
+					if enet_peer.get_peer(peer_id) != null:
+						enet_peer.disconnect_peer(peer_id, false)
+			if enet_peer.host != null:
+				enet_peer.host.flush()
 		_peer.close()
 	_peer = null
 	_route_description = ""

network/network_world_spawn_protocol.gd

@@ -3,8 +3,8 @@ extends RefCounted
 
 const CAPABILITY: StringName = &"world_spawn_envelope_v1"
 const ENVELOPE_VERSION: int = 1
-const RELIABLE_CHANNEL: int = NetworkProtocol.ITEM_RELIABLE_CHANNEL
-const SNAPSHOT_CHANNEL: int = 4
+const RELIABLE_CHANNEL: int = NetworkProtocol.WORLD_SPAWN_RELIABLE_CHANNEL
+const SNAPSHOT_CHANNEL: int = 15
 const MAX_SESSION_ID_LENGTH: int = 96
 const MAX_EVENT_ID_LENGTH: int = 48
 const MAX_ENTITY_ID_LENGTH: int = 128

network/network_world_spawn_service.gd

@@ -16,7 +16,9 @@ const CalendarSeasonType = preload("res://world/calendar_season.gd")
 signal local_capture_received(fish_catch: FishCatch)
 signal local_interaction_finished(accepted: bool, message: String)
 
-const SNAPSHOT_INTERVAL_SECONDS: float = 0.2
+const SIMULATION_INTERVAL_SECONDS: float = 1.0 / 15.0
+const SNAPSHOT_INTERVAL_SECONDS: float = 0.1
+const PLAYER_SPATIAL_CELL_SIZE: float = 8.0
 const CAPACITY_RESPONSE_TIMEOUT_SECONDS: float = 3.0
 const MAX_REQUEST_ID_LENGTH: int = 128
 const MAX_LEDGER_ENTRIES: int = 96
@@ -49,8 +51,16 @@ var _pending_captures: Dictionary = {}
 var _received_results: Dictionary = {}
 var _showcase_deadlines: Dictionary = {}
 var _envelope_sequence: int = 0
+var _simulation_elapsed: float = 0.0
 var _snapshot_elapsed: float = 0.0
 var _population_session_id: String = ""
+var _dirty_entity_ids: Dictionary[String, bool] = {}
+var _moving_players_by_cell: Dictionary[Vector2i, Array] = {}
+var _presentation_enabled: bool = true
+var _spawn_events_sent: int = 0
+var _despawn_events_sent: int = 0
+var _snapshot_packets_sent: int = 0
+var _snapshot_states_sent: int = 0
 var _rng := RandomNumberGenerator.new()
 
 
@@ -68,6 +78,7 @@ func setup(
 	save_manager: PlayerSaveManager,
 	item_use: NetworkItemUseService,
 	world_time: WorldTimeService = null,
+	presentation_enabled: bool = true,
 ) -> void:
 	_session = session
 	_spawn_service = spawn_service
@@ -82,6 +93,7 @@ func setup(
 	_save_manager = save_manager
 	_item_use = item_use
 	_world_time = world_time
+	_presentation_enabled = presentation_enabled
 	_rng.randomize()
 	_session.peer_authenticated.connect(_on_peer_authenticated)
 	_session.peer_removed.connect(_on_peer_removed)
@@ -136,6 +148,7 @@ func finish_local_interaction(
 		"target_position": NetworkWorldSpawnProtocol.vector3_to_array(
 			target_position
 		),
+		"movement_sequence": _session.get_latest_movement_input_sequence(),
 	}
 	if _session.is_host():
 		_handle_interaction_finish(_session.get_local_peer_id(), data)
@@ -217,6 +230,13 @@ func get_entry_for_entity(entity_id: String) -> GatherableDataType:
 	return state.get("data") as GatherableDataType
 
 
+func refresh_world_context() -> void:
+	_clear_world()
+	_population_session_id = ""
+	if _session != null and _session.is_gameplay_session_active():
+		_begin_population_if_ready.call_deferred()
+
+
 func _physics_process(delta: float) -> void:
 	if (
 		_session == null
@@ -225,11 +245,27 @@ func _physics_process(delta: float) -> void:
 	):
 		return
 	_begin_population_if_ready()
+	# There is no gameplay state to advance while a dedicated server has no
+	# spawned players. Pausing the roaming/respawn loop also prevents idle
+	# gatherables from producing a continuous stream of snapshots to nobody.
+	if _spawn_service == null or _spawn_service.get_peer_ids().is_empty():
+		_simulation_elapsed = 0.0
+		_snapshot_elapsed = 0.0
+		return
+	_simulation_elapsed += delta
+	_snapshot_elapsed += delta
+	if _simulation_elapsed < SIMULATION_INTERVAL_SECONDS:
+		return
+	var simulation_delta: float = minf(_simulation_elapsed, 0.2)
+	_simulation_elapsed = fmod(
+		_simulation_elapsed,
+		SIMULATION_INTERVAL_SECONDS,
+	)
 	_update_pending_capture_timeouts()
 	_update_respawns()
 	_update_showcase_deadlines()
-	_update_host_entities(delta)
-	_snapshot_elapsed += delta
+	_rebuild_moving_player_spatial_index()
+	_update_host_entities(simulation_delta)
 	if _snapshot_elapsed >= SNAPSHOT_INTERVAL_SECONDS:
 		_snapshot_elapsed = fmod(
 			_snapshot_elapsed,
@@ -246,6 +282,7 @@ func _begin_population_if_ready() -> void:
 		or _catalog == null
 		or _world == null
 		or _world_root == null
+		or not _world.is_world_ready()
 	):
 		return
 	var session_id: String = _session.get_session_id()
@@ -273,7 +310,10 @@ func _cache_spawn_surface(entry: GatherableDataType) -> void:
 			_world.get_gatherable_spawn_positions(entry.spawn_anchor_set_id)
 		)
 		_spawn_anchor_positions[entry.type_id] = anchor_positions
-		if anchor_positions.is_empty():
+		if (
+			anchor_positions.is_empty()
+			and _world.get_world_layout() == WorldLayout.GENERATED
+		):
 			push_warning(
 				"No gatherable spawn anchors were found for %s." % entry.type_id
 			)
@@ -299,7 +339,10 @@ func _cache_spawn_surface(entry: GatherableDataType) -> void:
 	_surface_triangles[entry.type_id] = triangles
 	_surface_areas[entry.type_id] = areas
 	_surface_total_areas[entry.type_id] = total_area
-	if triangles.is_empty():
+	if (
+		triangles.is_empty()
+		and _world.get_world_layout() == WorldLayout.GENERATED
+	):
 		push_warning(
 			"No valid spawn surface was found for %s." % entry.type_id
 		)
@@ -341,6 +384,7 @@ func _spawn_entity(entry: GatherableDataType) -> void:
 	)
 	_apply_envelope(envelope)
 	receive_world_envelope.rpc(envelope)
+	_spawn_events_sent += 1
 
 
 func _update_host_entities(delta: float) -> void:
@@ -392,6 +436,8 @@ func _update_host_entities(delta: float) -> void:
 				minf(step / maxf(horizontal_delta.length(), 0.001), 1.0),
 			)
 			state["yaw"] = atan2(-direction.x, -direction.z)
+			state["revision"] = int(state["revision"]) + 1
+			_dirty_entity_ids[entity_id] = true
 		state["position"] = position
 		_entities[entity_id] = state
 		if entry.can_be_scared() and _should_scare(entry, position, quality):
@@ -405,6 +451,34 @@ func _should_scare(
 ) -> bool:
 	var scare_radius: float = entry.get_scare_radius_for_quality(quality)
 	var radius_squared: float = scare_radius * scare_radius
+	var center_cell := _player_spatial_cell(position)
+	var cell_radius: int = ceili(scare_radius / PLAYER_SPATIAL_CELL_SIZE)
+	for cell_x: int in range(
+		center_cell.x - cell_radius,
+		center_cell.x + cell_radius + 1,
+	):
+		for cell_y: int in range(
+			center_cell.y - cell_radius,
+			center_cell.y + cell_radius + 1,
+		):
+			for value: Variant in _moving_players_by_cell.get(
+				Vector2i(cell_x, cell_y),
+				[],
+			):
+				var avatar := value as Player
+				if avatar == null:
+					continue
+				var player_delta := Vector2(
+					avatar.global_position.x - position.x,
+					avatar.global_position.z - position.z,
+				)
+				if player_delta.length_squared() <= radius_squared:
+					return true
+	return false
+
+
+func _rebuild_moving_player_spatial_index() -> void:
+	_moving_players_by_cell.clear()
 	for peer_id: int in _spawn_service.get_peer_ids():
 		var avatar: Player = _spawn_service.get_avatar(peer_id)
 		if (
@@ -413,13 +487,17 @@ func _should_scare(
 			or avatar.is_sneaking()
 		):
 			continue
-		var delta := Vector2(
-			avatar.global_position.x - position.x,
-			avatar.global_position.z - position.z,
-		)
-		if delta.length_squared() <= radius_squared:
-			return true
-	return false
+		var cell: Vector2i = _player_spatial_cell(avatar.global_position)
+		var players: Array = _moving_players_by_cell.get(cell, [])
+		players.append(avatar)
+		_moving_players_by_cell[cell] = players
+
+
+static func _player_spatial_cell(position: Vector3) -> Vector2i:
+	return Vector2i(
+		floori(position.x / PLAYER_SPATIAL_CELL_SIZE),
+		floori(position.z / PLAYER_SPATIAL_CELL_SIZE),
+	)
 
 
 func _sample_surface_position(
@@ -506,28 +584,69 @@ func _anchor_is_occupied(type_id: StringName, anchor: Vector3) -> bool:
 
 
 func _broadcast_entity_snapshots() -> void:
-	if _entities.is_empty():
+	if _dirty_entity_ids.is_empty() or not _can_broadcast_snapshots():
 		return
+	var has_remote_recipients: bool = _has_authenticated_remote_peers()
 	var snapshots: Array[Dictionary] = []
-	for entity_id: String in _entities:
-		var state: Dictionary = _entities[entity_id]
-		state["revision"] = int(state["revision"]) + 1
-		_entities[entity_id] = state
-		snapshots.append(_state_to_network(state))
-	while not snapshots.is_empty():
+	for entity_id: String in _dirty_entity_ids:
+		var state: Dictionary = _entities.get(entity_id, {})
+		if not state.is_empty():
+			snapshots.append(_state_to_network(state))
+	_dirty_entity_ids.clear()
+	var maximum_chunk_size: int = (
+		NetworkWorldSpawnProtocol.SNAPSHOT_ENTITIES_PER_ENVELOPE
+	)
+	for start_index: int in range(
+		0,
+		snapshots.size(),
+		maximum_chunk_size,
+	):
 		var chunk: Array[Dictionary] = []
-		var chunk_size: int = mini(
-			NetworkWorldSpawnProtocol.SNAPSHOT_ENTITIES_PER_ENVELOPE,
-			snapshots.size(),
-		)
-		for _index: int in chunk_size:
-			chunk.append(snapshots.pop_front())
+		for value: Dictionary in snapshots.slice(
+			start_index,
+			mini(start_index + maximum_chunk_size, snapshots.size()),
+		):
+			chunk.append(value)
 		var envelope: Dictionary = _make_envelope(
 			&"snapshot",
 			{"entities": chunk},
 		)
 		_apply_envelope(envelope)
+		if not has_remote_recipients:
+			continue
+		# A peer-disconnected callback can resume validation/gameplay code and
+		# close the host while this physics update is still unwinding. Never send
+		# through the replacement OfflineMultiplayerPeer in that narrow window.
+		if not _can_broadcast_snapshots():
+			return
 		receive_world_snapshot_envelope.rpc(envelope)
+		_snapshot_packets_sent += 1
+		_snapshot_states_sent += chunk.size()
+
+
+func _has_authenticated_remote_peers() -> bool:
+	if _session == null:
+		return false
+	var local_peer_id: int = _session.get_local_peer_id()
+	for peer_id: int in _session.get_authenticated_peer_ids():
+		if peer_id != local_peer_id:
+			return true
+	return false
+
+
+func _can_broadcast_snapshots() -> bool:
+	if (
+		_session == null
+		or not _session.is_host()
+		or not _session.is_gameplay_session_active()
+	):
+		return false
+	var peer: MultiplayerPeer = multiplayer.multiplayer_peer
+	return (
+		peer != null
+		and peer.get_connection_status()
+			== MultiplayerPeer.CONNECTION_CONNECTED
+	)
 
 
 func _despawn_entity(
@@ -541,6 +660,7 @@ func _despawn_entity(
 		return
 	var entry := state.get("data") as GatherableDataType
 	_entities.erase(entity_id)
+	_dirty_entity_ids.erase(entity_id)
 	var envelope: Dictionary = _make_envelope(
 		&"despawn",
 		{
@@ -551,6 +671,7 @@ func _despawn_entity(
 	)
 	_apply_envelope(envelope)
 	receive_world_envelope.rpc(envelope)
+	_despawn_events_sent += 1
 	if schedule_respawn and entry != null:
 		_respawns.append({
 			"type_id": entry.type_id,
@@ -633,6 +754,7 @@ func _handle_interaction_finish(peer_id: int, data: Dictionary) -> void:
 	var state: Dictionary = _entities.get(entity_id, {})
 	var entry := state.get("data") as GatherableDataType
 	var avatar: Player = _spawn_service.get_avatar(peer_id)
+	var movement_sequence_value: Variant = data.get("movement_sequence", null)
 	var error: String = ""
 	if (
 		str(data.get("session_id", "")) != _session.get_session_id()
@@ -640,6 +762,9 @@ func _handle_interaction_finish(peer_id: int, data: Dictionary) -> void:
 		or request_id.length() > MAX_REQUEST_ID_LENGTH
 		or str(charge.get("request_id", "")) != request_id
 		or not target_position.is_finite()
+		or typeof(movement_sequence_value) != TYPE_INT
+		or int(movement_sequence_value) < 0
+		or int(movement_sequence_value) > 2147483647
 	):
 		error = "The catch attempt was invalid."
 	elif state.is_empty() or entry == null or bool(state.get("locked", false)):
@@ -650,14 +775,28 @@ func _handle_interaction_finish(peer_id: int, data: Dictionary) -> void:
 		error = "Equip the correct gathering tool."
 	elif avatar == null:
 		error = "The player is unavailable."
-	elif entry.requires_sneaking and not avatar.is_sneaking():
-		error = "Sneak closer before using the tool."
 	else:
+		var movement_state: Dictionary = (
+			avatar.get_lag_compensated_movement_state(
+				int(movement_sequence_value)
+			)
+		)
+		if entry.requires_sneaking and not bool(
+			movement_state.get("sneaking", avatar.is_sneaking())
+		):
+			error = "Sneak closer before using the tool."
+		if not error.is_empty():
+			_send_interaction_result(peer_id, request_id, false, error)
+			return
 		var entity_position: Vector3 = state["position"]
 		var target_distance: float = entity_position.distance_to(target_position)
+		var compensated_position: Vector3 = movement_state.get(
+			"position",
+			avatar.global_position,
+		)
 		var player_distance: float = Vector2(
-			avatar.global_position.x - entity_position.x,
-			avatar.global_position.z - entity_position.z,
+			compensated_position.x - entity_position.x,
+			compensated_position.z - entity_position.z,
 		).length()
 		if target_distance > entry.capture_radius:
 			error = "The gathering tool missed."
@@ -903,7 +1042,7 @@ func _apply_capture_result(data: Dictionary) -> void:
 			_local_collection,
 		)
 		_local_inventory.add_catch(fish_catch)
-		_local_collection.mark_quality_discovered(
+		_local_collection.record_catch(
 			fish_catch.fish_id,
 			fish_catch.quality,
 		)
@@ -1078,6 +1217,9 @@ func _apply_entity_state(value: Variant, immediate: bool) -> void:
 	var position: Vector3 = NetworkWorldSpawnProtocol.array_to_vector3(
 		state["position"]
 	)
+	if not _presentation_enabled:
+		_entity_revisions[entity_id] = revision
+		return
 	var presentation := _presentations.get(entity_id) as WorldGatherableType
 	if presentation == null or not is_instance_valid(presentation):
 		presentation = WorldGatherableType.new()
@@ -1127,6 +1269,8 @@ func _remove_presentation(entity_id: String, with_dust: bool) -> void:
 
 
 func _apply_showcase(payload: Dictionary) -> void:
+	if not _presentation_enabled:
+		return
 	if (
 		typeof(payload.get("owner_peer_id")) != TYPE_INT
 		or typeof(payload.get("visible")) != TYPE_BOOL
@@ -1268,6 +1412,8 @@ func _clear_world() -> void:
 		if presentation != null and is_instance_valid(presentation):
 			presentation.queue_free()
 	_entities.clear()
+	_dirty_entity_ids.clear()
+	_moving_players_by_cell.clear()
 	_presentations.clear()
 	_entity_revisions.clear()
 	_surface_triangles.clear()
@@ -1279,9 +1425,22 @@ func _clear_world() -> void:
 	_charge_requests.clear()
 	_pending_captures.clear()
 	_showcase_deadlines.clear()
+	_simulation_elapsed = 0.0
 	_snapshot_elapsed = 0.0
 
 
+func get_network_metrics() -> Dictionary:
+	return {
+		"entities": _entities.size(),
+		"presentations": _presentations.size(),
+		"dirty_entities": _dirty_entity_ids.size(),
+		"spawn_events_sent": _spawn_events_sent,
+		"despawn_events_sent": _despawn_events_sent,
+		"snapshot_packets_sent": _snapshot_packets_sent,
+		"snapshot_states_sent": _snapshot_states_sent,
+	}
+
+
 func _bound(values: Dictionary) -> void:
 	while values.size() > MAX_LEDGER_ENTRIES:
 		values.erase(values.keys().front())

network/player_data_root.gd

@@ -189,6 +189,7 @@ func activate_process_root(path: String) -> bool:
 func path_for(store_owner: StringName) -> String:
 	var relative: String = {
 		&"player_save": "player/player_save.nfsave",
+		&"save_slots": "player/save_slots.json",
 		&"network_profile": "player/network_profile.json",
 		&"player_appearance": "player/player_appearance.json",
 		&"saved_servers": "social/saved_servers.json",
@@ -350,7 +351,7 @@ func _test_writable(path: String) -> bool:
 
 func _create_layout(path: String, id: String) -> bool:
 	for relative: String in [
-		"player", "social", "backups/saves", "backups/migrations",
+		"player", "player/saves", "social", "backups/saves", "backups/migrations",
 		"backups/conflicts", "identity-backups", "progression-backups",
 	]:
 		if DirAccess.make_dir_recursive_absolute(path.path_join(relative)) != OK:

network/player_list_entry.gd

@@ -12,6 +12,8 @@ var continuity_state := ""
 var ping_to_host_ms := -1
 var muted := false
 var blocked := false
+var is_friend := false
+var can_request_friend := false
 var can_kick := false
 var can_ban := false
 var can_clear_art := false

network/player_relationship_store.gd

@@ -3,8 +3,12 @@ extends Node
 
 signal relationship_changed(fingerprint: String)
 
-const FORMAT_VERSION := 1
+const FORMAT_VERSION := 2
 const MAX_RECORDS := 500
+const MAX_FRIENDS := 200
+const SOCIAL_TOKEN_LENGTH := 64
+const PRESENCE_CHANNEL_DOMAIN := "NETFISHING_PRESENCE_V1:"
+const INVITE_INBOX_DOMAIN := "NETFISHING_INVITE_V1:"
 
 var _records: Dictionary = {}
 var _loaded := false
@@ -29,6 +33,11 @@ func is_blocked(fingerprint: String) -> bool:
 	return bool(_records.get(fingerprint, {}).get("blocked", false))
 
 
+func is_friend(fingerprint: String) -> bool:
+	_ensure_loaded()
+	return bool(_records.get(fingerprint, {}).get("friend", false))
+
+
 func set_muted(fingerprint: String, display_name: String, value: bool) -> bool:
 	if not _valid_target(fingerprint, display_name):
 		return false
@@ -48,20 +57,101 @@ func set_blocked(fingerprint: String, display_name: String, value: bool) -> bool
 	# fresh visibility boundary for future messages; already suppressed
 	# messages retain their immutable local suppression flag.
 	record["muted"] = value
+	if value:
+		_clear_friend_fields(record)
 	return _commit(fingerprint, record)
 
 
+func create_friend_capabilities() -> Dictionary:
+	var presence_write_token := NetworkIdentityCrypto.secure_id(32)
+	var invite_token := NetworkIdentityCrypto.secure_id(32)
+	if not _valid_social_token(presence_write_token) or not _valid_social_token(invite_token):
+		return {}
+	return {
+		"local_presence_write_token": presence_write_token,
+		"local_invite_token": invite_token,
+		"presence_channel": presence_channel_for_write_token(
+			presence_write_token
+		),
+		"invite_token": invite_token,
+	}
+
+
+func add_friend(
+	fingerprint: String,
+	display_name: String,
+	local_capabilities: Dictionary,
+	remote_capabilities: Dictionary,
+) -> bool:
+	if (
+		not _valid_target(fingerprint, display_name)
+		or is_blocked(fingerprint)
+		or not _valid_local_capabilities(local_capabilities)
+		or not _valid_public_capabilities(remote_capabilities)
+	):
+		return false
+	_ensure_loaded()
+	if not is_friend(fingerprint) and get_friends().size() >= MAX_FRIENDS:
+		return false
+	var record := _record(fingerprint, display_name)
+	var now := int(Time.get_unix_time_from_system())
+	record["friend"] = true
+	record["friend_since_unix"] = int(record.get("friend_since_unix", now))
+	record["local_presence_write_token"] = str(
+		local_capabilities["local_presence_write_token"]
+	)
+	record["local_invite_token"] = str(
+		local_capabilities["local_invite_token"]
+	)
+	record["remote_presence_channel"] = str(
+		remote_capabilities["presence_channel"]
+	)
+	record["remote_invite_token"] = str(
+		remote_capabilities["invite_token"]
+	)
+	return _commit(fingerprint, record)
+
+
+func remove_friend(fingerprint: String, display_name: String) -> bool:
+	if not _valid_target(fingerprint, display_name):
+		return false
+	_ensure_loaded()
+	var record := _record(fingerprint, display_name)
+	_clear_friend_fields(record)
+	return _commit(fingerprint, record)
+
+
+func get_friend_record(fingerprint: String) -> Dictionary:
+	_ensure_loaded()
+	var record: Dictionary = _records.get(fingerprint, {})
+	return record.duplicate(true) if bool(record.get("friend", false)) else {}
+
+
+func get_friends() -> Array[Dictionary]:
+	_ensure_loaded()
+	var result: Array[Dictionary] = []
+	for value: Dictionary in _records.values():
+		if bool(value.get("friend", false)):
+			result.append(value.duplicate(true))
+	_sort_records(result)
+	return result
+
+
+static func presence_channel_for_write_token(write_token: String) -> String:
+	return (PRESENCE_CHANNEL_DOMAIN + write_token).sha256_text()
+
+
+static func invite_inbox_id(invite_token: String) -> String:
+	return (INVITE_INBOX_DOMAIN + invite_token).sha256_text()
+
+
 func get_records() -> Array[Dictionary]:
 	_ensure_loaded()
 	var result: Array[Dictionary] = []
 	for value: Dictionary in _records.values():
 		if bool(value.get("muted", false)) or bool(value.get("blocked", false)):
 			result.append(value.duplicate(true))
-	result.sort_custom(func(a: Dictionary, b: Dictionary) -> bool:
-		return str(a.get("last_known_display_name", "")).naturalnocasecmp_to(
-			str(b.get("last_known_display_name", ""))
-		) < 0
-	)
+	_sort_records(result)
 	return result
 
 
@@ -72,6 +162,7 @@ func _record(fingerprint: String, display_name: String) -> Dictionary:
 		"created_unix": now,
 		"muted": false,
 		"blocked": false,
+		"friend": false,
 	})
 	record["last_known_display_name"] = display_name
 	record["updated_unix"] = now
@@ -82,7 +173,11 @@ func _commit(fingerprint: String, record: Dictionary) -> bool:
 	if _write_blocked:
 		return false
 	var previous: Dictionary = _records.duplicate(true)
-	if not bool(record["muted"]) and not bool(record["blocked"]):
+	if (
+		not bool(record.get("muted", false))
+		and not bool(record.get("blocked", false))
+		and not bool(record.get("friend", false))
+	):
 		_records.erase(fingerprint)
 	else:
 		_records[fingerprint] = record
@@ -115,7 +210,8 @@ func _ensure_loaded() -> void:
 	if json.parse(file.get_as_text()) != OK or typeof(json.data) != TYPE_DICTIONARY:
 		return
 	var data: Dictionary = json.data
-	if data.get("format_version") != FORMAT_VERSION:
+	var format_version: int = int(data.get("format_version", 0))
+	if format_version not in [1, FORMAT_VERSION]:
 		_write_blocked = true
 		return
 	for value: Variant in data.get("records", []):
@@ -127,6 +223,13 @@ func _ensure_loaded() -> void:
 			continue
 		record["blocked"] = bool(record.get("blocked", false))
 		record["muted"] = bool(record.get("muted", false)) or record["blocked"]
+		record["friend"] = (
+			bool(record.get("friend", false))
+			and not record["blocked"]
+			and _valid_stored_friend_capabilities(record)
+		)
+		if not record["friend"]:
+			_clear_friend_fields(record)
 		_records[fingerprint] = record.duplicate(true)
 	_expected_hash = PortableFileGuard.hash_file(_store_path)
 
@@ -145,3 +248,69 @@ func _save() -> bool:
 	if bool(result.get("ok", false)):
 		_expected_hash = str(result["hash"])
 	return bool(result.get("ok", false))
+
+
+func _clear_friend_fields(record: Dictionary) -> void:
+	record["friend"] = false
+	for key: String in [
+		"friend_since_unix",
+		"local_presence_write_token",
+		"local_invite_token",
+		"remote_presence_channel",
+		"remote_invite_token",
+	]:
+		record.erase(key)
+
+
+func _valid_local_capabilities(value: Dictionary) -> bool:
+	var write_token := str(value.get("local_presence_write_token", ""))
+	var invite_token := str(value.get("local_invite_token", ""))
+	return (
+		_valid_social_token(write_token)
+		and _valid_social_token(invite_token)
+		and str(value.get("presence_channel", ""))
+			== presence_channel_for_write_token(write_token)
+		and str(value.get("invite_token", "")) == invite_token
+	)
+
+
+func _valid_public_capabilities(value: Dictionary) -> bool:
+	return (
+		_valid_social_token(str(value.get("presence_channel", "")))
+		and _valid_social_token(str(value.get("invite_token", "")))
+	)
+
+
+func _valid_stored_friend_capabilities(record: Dictionary) -> bool:
+	var local: Dictionary = {
+		"local_presence_write_token": str(
+			record.get("local_presence_write_token", "")
+		),
+		"local_invite_token": str(record.get("local_invite_token", "")),
+		"presence_channel": presence_channel_for_write_token(str(
+			record.get("local_presence_write_token", "")
+		)),
+		"invite_token": str(record.get("local_invite_token", "")),
+	}
+	var remote: Dictionary = {
+		"presence_channel": str(record.get("remote_presence_channel", "")),
+		"invite_token": str(record.get("remote_invite_token", "")),
+	}
+	return _valid_local_capabilities(local) and _valid_public_capabilities(remote)
+
+
+func _valid_social_token(value: String) -> bool:
+	if value.length() != SOCIAL_TOKEN_LENGTH:
+		return false
+	for character: String in value:
+		if character not in "0123456789abcdef":
+			return false
+	return true
+
+
+func _sort_records(records: Array[Dictionary]) -> void:
+	records.sort_custom(func(a: Dictionary, b: Dictionary) -> bool:
+		return str(a.get("last_known_display_name", "")).naturalnocasecmp_to(
+			str(b.get("last_known_display_name", ""))
+		) < 0
+	)

network/player_spawn_service.gd

@@ -11,16 +11,19 @@ var _local_player: Player
 var _spawn_transform: Transform3D
 var _avatars: Dictionary[int, Player] = {}
 var _local_peer_id: int = 1
+var _dedicated_runtime: bool = false
 
 
 func setup(
 	players_root: Node3D,
 	local_player: Player,
 	spawn_transform: Transform3D,
+	dedicated_runtime: bool = false,
 ) -> void:
 	_players_root = players_root
 	_local_player = local_player
 	_spawn_transform = spawn_transform
+	_dedicated_runtime = dedicated_runtime
 
 
 func set_spawn_transform(spawn_transform: Transform3D) -> void:
@@ -56,7 +59,10 @@ func spawn_remote_player(
 	avatar.set_network_peer_id(peer_id)
 	_players_root.add_child(avatar)
 	avatar.global_transform = transform
-	avatar.configure_network_remote(authoritative_simulation)
+	avatar.configure_network_remote(
+		authoritative_simulation,
+		_dedicated_runtime and authoritative_simulation,
+	)
 	_avatars[peer_id] = avatar
 	avatar_spawned.emit(peer_id, avatar)
 	return avatar

network/portable_data_migration.gd

@@ -124,6 +124,7 @@ static func migrate_active_to(
 	for relative: String in [
 		"player/player_save.nfsave",
 		"player/player_save.json",
+		"player/save_slots.json",
 		"player/network_profile.json",
 		"player/player_appearance.json",
 		"social/saved_servers.json",
@@ -142,6 +143,16 @@ static func migrate_active_to(
 		):
 			_remove_tree(staging)
 			return {"ok": false, "message": "Migration validation failed for %s." % relative}
+	var source_slots: String = data_root.root_path.path_join("player/saves")
+	if (
+		DirAccess.dir_exists_absolute(source_slots)
+		and not _copy_progression_slots(
+			source_slots,
+			staging.path_join("player/saves"),
+		)
+	):
+		_remove_tree(staging)
+		return {"ok": false, "message": "Migration validation failed for save slots."}
 	if DirAccess.dir_exists_absolute(normalized):
 		DirAccess.remove_absolute(normalized)
 	if DirAccess.rename_absolute(staging, normalized) != OK:
@@ -257,7 +268,7 @@ static func _copy_verified_owned_file(
 	source: String,
 	destination: String,
 ) -> Dictionary:
-	if source.get_file() != "player_save.nfsave":
+	if source.get_extension().to_lower() != "nfsave":
 		return _copy_verified_json(source, destination)
 	var decoded: Dictionary = ProgressionSaveCodec.read_local_save(source)
 	if not bool(decoded.get("ok", false)):
@@ -281,12 +292,42 @@ static func _valid_owned_data(filename: String, data: Dictionary) -> bool:
 	if filename == "player_save.json":
 		var version: int = int(data.get("save_version", -1))
 		return version >= 1 and version <= PlayerSaveManager.SAVE_VERSION
+	if filename == "save_slots.json":
+		return (
+			int(data.get("format_version", -1)) == PlayerSaveSlotCatalog.FORMAT_VERSION
+			and typeof(data.get("active_slot_id")) == TYPE_STRING
+			and typeof(data.get("slots")) == TYPE_ARRAY
+		)
 	return (
 		filename not in LEGACY_FILES
 		or int(data.get("format_version", -1)) == 1
 	)
 
 
+static func _copy_progression_slots(source: String, destination: String) -> bool:
+	if DirAccess.make_dir_recursive_absolute(destination) != OK:
+		return false
+	var access := DirAccess.open(source)
+	if access == null:
+		return false
+	access.list_dir_begin()
+	var filename: String = access.get_next()
+	while not filename.is_empty():
+		if (
+			not access.current_is_dir()
+			and filename.get_extension().to_lower() == "nfsave"
+			and not bool(_copy_verified_owned_file(
+				source.path_join(filename),
+				destination.path_join(filename),
+			).get("ok", false))
+		):
+			access.list_dir_end()
+			return false
+		filename = access.get_next()
+	access.list_dir_end()
+	return true
+
+
 static func _copy_bytes(source: String, destination: String) -> bool:
 	return _write_bytes(destination, PortableFileGuard.read_bytes(source))