diff --git a/scripts/api_client.gd b/scripts/api_client.gd
index f9efc33..f969522 100644
--- a/scripts/api_client.gd
+++ b/scripts/api_client.gd
@@ -69,6 +69,16 @@ func clear_token() -> void:
 
 var _base_url: String = "http://127.0.0.1:8000"
 
+## Read-only access to the configured backend base URL (no trailing slash).
+## Lets other client-side code that needs to build its own request outside
+## the [method request] primitive -- e.g. [HttpChunkSource], which needs raw
+## response bytes and custom headers `request()`'s JSON-only contract can't
+## give it -- target the same backend without duplicating the env var /
+## ProjectSettings resolution logic in `_ready()`.
+var base_url: String:
+	get:
+		return _base_url
+
 ## Request timeout in seconds. 0 disables the timeout.
 var timeout_seconds: float = 10.0
 
diff --git a/scripts/chunk_data_decoder.gd b/scripts/chunk_data_decoder.gd
index 2a08d3a..ac26c6a 100644
--- a/scripts/chunk_data_decoder.gd
+++ b/scripts/chunk_data_decoder.gd
@@ -498,7 +498,17 @@ static func _build_metadata_section(payload: ChunkPayload) -> PackedByteArray:
 	data.append_array(_write_float32(payload.bounds.size.x))
 	data.append_array(_write_float32(payload.bounds.size.y))
 	data.append_array(_write_float32(payload.bounds.size.z))
-	return data
+	# Every other section builder wraps its body with `_wrap_section`
+	# (section_id + uint32 length prefix) -- this one didn't, so
+	# `encode_binary()` spliced the raw metadata bytes directly into the
+	# section stream with no header at all. The decoder then read the
+	# metadata's own first bytes (a string length prefix) as a bogus
+	# section_id/section_len pair, producing a garbage length that
+	# overran the buffer ("truncated section body") and silently found
+	# zero real sections. Never caught before because nothing in this
+	# codebase called encode_binary() until
+	# scripts/smoke_test_http_chunk_source.gd.
+	return _wrap_section(data, SECTION_META)
 
 
 static func _build_tile_section(layer: TileLayerData) -> PackedByteArray:
diff --git a/scripts/chunk_stream_manager.gd b/scripts/chunk_stream_manager.gd
index 8c6802e..e3dbfd0 100644
--- a/scripts/chunk_stream_manager.gd
+++ b/scripts/chunk_stream_manager.gd
@@ -241,7 +241,11 @@ func _instantiate_chunk_scene_with_data(coord: Vector3i, data: ChunkDataDecoder.
 	chunk_scene.coord = coord
 	chunk_scene.key = data.key
 	chunk_scene.bounds = data.bounds
-	chunk_scene.format_version = data.format_version
+	# ChunkDataDecoder.ChunkPayload's field is named `version`, not
+	# `format_version` -- this was a hard crash ("Invalid access to
+	# property or key 'format_version'") the first time real decoded data
+	# ever reached this dormant method (via HttpChunkSource).
+	chunk_scene.format_version = data.version
 	chunk_scene.revision = data.revision
 
 	# Add tiles
@@ -348,7 +352,13 @@ func clear_all_chunks() -> void:
 
 func _parse_coord_from_key(key: String) -> Variant:
 	"""Parse chunk coordinate from a canonical chunk key."""
-	var parts: Array[String] = key.split(":")
+	# String.split() returns PackedStringArray, not Array[String] -- those
+	# are different container types with no implicit conversion, so this
+	# raised "Trying to assign a value of type 'PackedStringArray' to a
+	# variable of type 'Array[String]'" the moment real data reached this
+	# dormant method for the first time (via HttpChunkSource's
+	# _on_chunk_data_received call -- nothing exercised this path before).
+	var parts: PackedStringArray = key.split(":")
 	if parts.size() < 5:
 		return null
 	return Vector3i(
diff --git a/scripts/http_chunk_source.gd b/scripts/http_chunk_source.gd
new file mode 100644
index 0000000..2535fc6
--- /dev/null
+++ b/scripts/http_chunk_source.gd
@@ -0,0 +1,203 @@
+class_name HttpChunkSource
+extends Node
+## HTTP-backed adapter for ChunkStreamManager.set_server_rpc().
+##
+## Implements docs/architecture/world_persistence_impl.md D6/§8.4: the
+## playing client streams chunks from the FastAPI backend over HTTP (the
+## same public routes the editor uses -- `backend/app/routers/world_content.py`),
+## not over the dedicated server's ENet connection.
+##
+## `ChunkStreamManager` only requires an object with a `request_chunk(key)`
+## method (see its `_server_rpc` seam); this adapter satisfies that and feeds
+## results back through the manager's existing (previously dormant)
+## `_on_chunk_data_received(key, bytes)` / `_on_chunk_load_error(coord, msg)`
+## methods, and its `_instantiate_chunk_scene(coord)` for the "unwritten
+## coordinate renders empty" case (§8.1).
+##
+## `ChunkStreamManager._load_chunk()` builds its request key with empty
+## world/room components (`Chunk.build_key("", "", coord)`) because the
+## manager itself has no notion of which world/room it's streaming --
+## that's this adapter's job. `request_chunk(key)` only reads the
+## coordinate out of the key it's given and rebuilds the real key from
+## `world_id`/`room_slug` before touching the network, per the doc's note
+## under §8.4.
+##
+## Must be added to the scene tree before use (it creates per-request
+## HTTPRequest child nodes), e.g.:
+##
+##     var source := HttpChunkSource.new()
+##     source.world_id = str(Session.world_id)
+##     source.room_slug = "main"
+##     source.stream_manager = stream_manager
+##     stream_manager.add_child(source)
+##     stream_manager.set_server_rpc(source)
+
+## The world this source fetches chunks for (string form of the backend's
+## numeric world id -- matches Chunk.build_key's world_id parameter type).
+var world_id: String = ""
+
+## The room slug within that world. world.md's default room is "main".
+var room_slug: String = "main"
+
+## The ChunkStreamManager to feed decoded results back into. Required.
+var stream_manager: ChunkStreamManager = null
+
+## In-memory ETag cache: chunk key -> last-known checksum (bare hex, no
+## quotes). Backs the on-disk payload cache so re-fetching an unchanged
+## chunk costs a 304, not a re-download (world_persistence_impl.md §8.1:
+## "This is what makes the client's user://chunk_cache/ cache cheap").
+var _etag_by_key: Dictionary = {}
+
+const CACHE_DIR := "user://chunk_cache"
+
+
+func _ready() -> void:
+	DirAccess.make_dir_recursive_absolute(CACHE_DIR)
+
+
+## Satisfies ChunkStreamManager's `_server_rpc.request_chunk(key)` seam.
+func request_chunk(key: String) -> void:
+	var coord: Variant = _coord_from_key(key)
+	if coord == null:
+		push_error("HttpChunkSource: malformed chunk key %s" % key)
+		return
+	_fetch(coord)
+
+
+## world_persistence_impl.md §8.1 chunk index -- lists which chunks in a
+## bounds actually exist without fetching their payloads. Not called by
+## ChunkStreamManager's per-coordinate streaming loop today, but is the
+## "bounds-query listing" half of this adapter's contract for future
+## callers (e.g. a minimap, or a smarter prefetcher).
+func list_chunks(cx_min: int, cx_max: int, cy_min: int, cy_max: int, cz: int = 0) -> Dictionary:
+	var path := "/api/worlds/%s/rooms/%s/chunks?cx_min=%d&cx_max=%d&cy_min=%d&cy_max=%d&cz=%d" % [
+		world_id, room_slug, cx_min, cx_max, cy_min, cy_max, cz
+	]
+	return await _api_client().request(HTTPClient.METHOD_GET, path, "", "application/json", true)
+
+
+# ─────────────────────────── internals ───────────────────────────
+
+## Resolved via an absolute node path rather than the bare `ApiClient`
+## autoload identifier: bare autoload identifiers only resolve at GDScript
+## compile time inside a scene the engine boots normally. Any script
+## compiled while running headless via `godot --script ...` (as
+## scripts/smoke_test_*.gd do) fails with "Identifier not found: ApiClient"
+## on the bare form -- confirmed while writing
+## scripts/smoke_test_http_chunk_source.gd. `get_node` resolves the same
+## singleton correctly in both contexts (the same pattern game.gd already
+## uses for AvatarSystem via `get_node_or_null("/root/AvatarSystem")`).
+func _api_client() -> Node:
+	return get_node("/root/ApiClient")
+
+
+func _fetch(coord: Vector3i) -> void:
+	var key := Chunk.build_key(world_id, room_slug, coord)
+
+	var http := HTTPRequest.new()
+	http.timeout = 10.0
+	add_child(http)
+
+	var api := _api_client()
+	var headers := PackedStringArray()
+	headers.append("Accept: application/octet-stream")
+	if api.has_token():
+		var bearer: String = api.bearer_token
+		headers.append("Authorization: " + bearer)
+	var cached_etag: String = _etag_by_key.get(key, "")
+	if cached_etag != "":
+		headers.append("If-None-Match: \"%s\"" % cached_etag)
+
+	var base_url: String = api.base_url
+	var url := "%s/api/worlds/%s/rooms/%s/chunks/%d/%d/%d" % [
+		base_url, world_id, room_slug, coord.x, coord.y, coord.z
+	]
+	var err := http.request(url, headers, HTTPClient.METHOD_GET)
+	if err != OK:
+		http.queue_free()
+		_fail(coord, "HttpChunkSource: failed to start request for %s (%d)" % [key, err])
+		return
+
+	var args: Array = await http.request_completed
+	var result_code: int = args[0]
+	var response_code: int = args[1]
+	var response_headers: PackedStringArray = args[2]
+	var body: PackedByteArray = args[3]
+	http.queue_free()
+
+	if result_code != HTTPRequest.RESULT_SUCCESS:
+		_fail(coord, "HttpChunkSource: transport error fetching %s (code %d)" % [key, result_code])
+		return
+
+	match response_code:
+		304:
+			var cached := _read_cache(cached_etag)
+			if cached.is_empty():
+				# Cache miss despite a 304 (e.g. the user:// cache got
+				# cleared) -- drop the stale ETag and refetch unconditionally
+				# rather than deliver empty bytes to the decoder.
+				_etag_by_key.erase(key)
+				_fetch(coord)
+				return
+			_deliver(key, cached)
+		200:
+			var checksum := _etag_from_headers(response_headers)
+			if checksum != "":
+				_etag_by_key[key] = checksum
+				_write_cache(checksum, body)
+			_deliver(key, body)
+		404:
+			# world_persistence_impl.md §8.1: "an unwritten coordinate is not
+			# an error to the streamer -- it renders empty."
+			if stream_manager != null:
+				stream_manager._instantiate_chunk_scene(coord)
+		_:
+			_fail(coord, "HttpChunkSource: HTTP %d fetching %s" % [response_code, key])
+
+
+func _deliver(key: String, bytes: PackedByteArray) -> void:
+	if stream_manager != null:
+		stream_manager._on_chunk_data_received(key, bytes)
+
+
+func _fail(coord: Vector3i, message: String) -> void:
+	push_error(message)
+	if stream_manager != null:
+		stream_manager._on_chunk_load_error(coord, message)
+
+
+func _coord_from_key(key: String) -> Variant:
+	var parts: PackedStringArray = key.split(":")
+	if parts.size() < 5:
+		return null
+	return Vector3i(int(parts[2]), int(parts[3]), int(parts[4]))
+
+
+func _etag_from_headers(headers: PackedStringArray) -> String:
+	for h in headers:
+		if h.to_lower().begins_with("etag:"):
+			var v := h.substr(h.find(":") + 1).strip_edges()
+			return v.trim_prefix('"').trim_suffix('"')
+	return ""
+
+
+func _cache_path(checksum: String) -> String:
+	return "%s/%s" % [CACHE_DIR, checksum]
+
+
+func _write_cache(checksum: String, bytes: PackedByteArray) -> void:
+	var f := FileAccess.open(_cache_path(checksum), FileAccess.WRITE)
+	if f:
+		f.store_buffer(bytes)
+		f.close()
+
+
+func _read_cache(checksum: String) -> PackedByteArray:
+	if checksum == "":
+		return PackedByteArray()
+	var f := FileAccess.open(_cache_path(checksum), FileAccess.READ)
+	if f == null:
+		return PackedByteArray()
+	var bytes := f.get_buffer(f.get_length())
+	f.close()
+	return bytes
diff --git a/scripts/server.gd b/scripts/server.gd
index 9545c53..a8222d5 100644
--- a/scripts/server.gd
+++ b/scripts/server.gd
@@ -9,9 +9,26 @@
 ## Launched by the backend's game-server supervisor
 ## (backend/app/services/game_servers.py):
 ##     godot --headless --path . res://scenes/server.tscn -- \
-##         --port 27000 --world 3 --status-file /path/to/status.json
+##         --port 27000 --world 3 --status-file /path/to/status.json \
+##         --api-url http://127.0.0.1:8000 --server-token <per-spawn token>
 ## All args are optional; the defaults let `godot res://scenes/server.tscn`
-## run standalone on 7777 for smoke_test_network_client.gd.
+## run standalone on 7777 for smoke_test_network_client.gd (which passes
+## neither --api-url nor --server-token, so startup goes straight to the
+## test_level.tscn fallback below).
+##
+## Startup sequence (docs/architecture/world_persistence_impl.md §7.2):
+##   1. Parse args.
+##   2. If --api-url/--server-token/--world are all present, pull the world's
+##      manifest + the spawn-neighbourhood chunks from the backend's internal
+##      API (§8.3), decode them, and build real static collision from the
+##      tile layers + objects. This happens BEFORE the ENet port binds.
+##   3. Fall back to test_level.tscn if the backend is unreachable, the world
+##      has no rooms, or the room has zero chunks -- a content outage must
+##      never make a world un-joinable (§7.2 step 5).
+##   4. Bind the ENet port, then write "ready": true and "content" ("loaded"
+##      or "fallback") to the status file -- this is what
+##      GameServerSupervisor._await_ready() polls for (§7.4), replacing the
+##      old bare-port-bind readiness check.
 extends Node
 
 const MAX_PLAYERS := 16
@@ -21,11 +38,40 @@ const PHYSICS_HZ := 30
 ## every this many physics frames.
 const STATUS_WRITE_EVERY := 30
 
+## world_persistence_impl.md §4.1's default room slug.
+const MAIN_ROOM_SLUG := "main"
+## §7.2 step 3: "the spawn neighbourhood (radius from the spawn point,
+## default 3 chunks)".
+const SPAWN_CHUNK_RADIUS := 3
+## HTTP timeout for internal-API calls during startup. Generous but bounded
+## -- a hung backend must not hang the server startup (§7.2 fallback rule).
+const CONTENT_FETCH_TIMEOUT := 8.0
+
 var network_peer: ENetMultiplayerPeer
 var port: int = DEFAULT_PORT
 var world_id: String = ""
 var status_file_path: String = ""
 
+## world_persistence_impl.md §7.3: internal-API base URL and this spawn's
+## bearer token, both passed on the command line by
+## GameServerSupervisor._spawn(). Empty means "no backend to talk to" --
+## content loading is skipped and the fallback level loads immediately.
+var api_url: String = ""
+var server_token: String = ""
+
+## What startup actually loaded, written into the status file's "content"
+## field: "loaded" (real content from the backend) or "fallback"
+## (test_level.tscn, either because there was nothing to load or nothing
+## reachable to load it from).
+var _content_status: String = "fallback"
+## Diagnostic-only counters (not part of the doc's required status fields,
+## but real: they're the actual number of chunks decoded and collision
+## shapes added to the tree, computed in the same pass that builds them --
+## useful for verifying a real load happened without needing to poke the
+## running server's scene tree from outside).
+var _chunks_loaded: int = 0
+var _collision_shapes_built: int = 0
+
 ## peer_id -> Player node
 var players: Dictionary = {}
 var tick: int = 0
@@ -35,7 +81,7 @@ var is_running: bool = false
 var _join_count: int = 0
 
 var _player_scene: PackedScene = preload("res://scenes/server_player.tscn")
-var _level_scene: PackedScene = preload("res://scenes/worlds/test_level.tscn")
+var _fallback_level_scene: PackedScene = preload("res://scenes/worlds/test_level.tscn")
 var _spawn_point: Vector3 = Vector3(0, 1.5, 0)
 
 
@@ -48,8 +94,16 @@ func _ready() -> void:
 	_parse_args()
 	Engine.physics_ticks_per_second = PHYSICS_HZ
 
-	# Load the world's level so players have real ground + collision.
-	var level := _level_scene.instantiate()
+	var level: Node = null
+	if api_url != "" and server_token != "" and world_id != "":
+		level = await _try_load_real_content()
+
+	if level == null:
+		level = _fallback_level_scene.instantiate()
+		_content_status = "fallback"
+	else:
+		_content_status = "loaded"
+
 	level.name = "World"
 	add_child(level)
 	var marker := level.find_child("PlayerSpawn", true, false)
@@ -65,7 +119,9 @@ func _ready() -> void:
 	multiplayer.multiplayer_peer = network_peer
 	is_running = true
 	_write_status()
-	print("[Server] world=%s listening on ENet %d (%d Hz)" % [world_id, port, PHYSICS_HZ])
+	print("[Server] world=%s listening on ENet %d (%d Hz) content=%s chunks=%d collision_shapes=%d" % [
+		world_id, port, PHYSICS_HZ, _content_status, _chunks_loaded, _collision_shapes_built
+	])
 
 
 ## A spawn point offset from the level's marker so joins don't stack. Golden-
@@ -95,9 +151,198 @@ func _parse_args() -> void:
 				if i + 1 < args.size():
 					status_file_path = args[i + 1]
 					i += 1
+			"--api-url":
+				if i + 1 < args.size():
+					api_url = args[i + 1].rstrip("/")
+					i += 1
+			"--server-token":
+				if i + 1 < args.size():
+					server_token = args[i + 1]
+					i += 1
 		i += 1
 
 
+# ---------------------------------------------------------------------------
+# Startup content load (world_persistence_impl.md §7.2/§7.3/§8.3)
+# ---------------------------------------------------------------------------
+
+## Attempt to build a real level from the backend's internal content API.
+## Returns null on ANY failure (unreachable backend, no rooms, zero chunks)
+## so the caller falls back to test_level.tscn -- a content outage must
+## never make a world un-joinable.
+func _try_load_real_content() -> Node:
+	var manifest_result := await _internal_get_json("/api/internal/worlds/%s/manifest" % world_id)
+	if not manifest_result.ok:
+		print("[Server] manifest fetch failed (%s); falling back to test_level" % manifest_result.get("detail", "?"))
+		return null
+
+	var manifest: Dictionary = manifest_result.data
+	var rooms: Array = manifest.get("rooms", [])
+	var room: Variant = _pick_room(rooms)
+	if room == null:
+		print("[Server] world %s has no rooms; falling back to test_level" % world_id)
+		return null
+	if int(room.get("chunk_count", 0)) <= 0:
+		print("[Server] room '%s' has zero chunks; falling back to test_level" % room.get("slug", "?"))
+		return null
+
+	var slug: String = room.get("slug", MAIN_ROOM_SLUG)
+	var spawn_pos := _spawn_pos_from_room(room)
+	var spawn_coord := Chunk.coord_for_position(spawn_pos)
+
+	var index_path := "/api/internal/worlds/%s/rooms/%s/chunks?cx_min=%d&cx_max=%d&cy_min=%d&cy_max=%d&cz=%d" % [
+		world_id, slug,
+		spawn_coord.x - SPAWN_CHUNK_RADIUS, spawn_coord.x + SPAWN_CHUNK_RADIUS,
+		spawn_coord.y - SPAWN_CHUNK_RADIUS, spawn_coord.y + SPAWN_CHUNK_RADIUS,
+		spawn_coord.z,
+	]
+	var index_result := await _internal_get_json(index_path)
+	if not index_result.ok:
+		print("[Server] chunk index fetch failed (%s); falling back to test_level" % index_result.get("detail", "?"))
+		return null
+
+	var world_root := Node3D.new()
+	world_root.name = "World"
+	var marker := Marker3D.new()
+	marker.name = "PlayerSpawn"
+	marker.position = spawn_pos
+	world_root.add_child(marker)
+
+	var entries: Array = (index_result.data as Dictionary).get("chunks", [])
+	for entry in entries:
+		var cx: int = int(entry["cx"])
+		var cy: int = int(entry["cy"])
+		var cz: int = int(entry["cz"])
+		var chunk_path := "/api/internal/worlds/%s/rooms/%s/chunks/%d/%d/%d" % [world_id, slug, cx, cy, cz]
+		var payload_result := await _internal_get_binary(chunk_path)
+		if not payload_result.ok:
+			printerr("[Server] chunk fetch failed for %d/%d/%d (status %d); skipping" % [cx, cy, cz, payload_result.status])
+			continue
+		var decoded: ChunkDataDecoder.ChunkPayload = ChunkDataDecoder.decode_binary(payload_result.body)
+		if decoded == null:
+			printerr("[Server] chunk decode failed for %d/%d/%d; skipping" % [cx, cy, cz])
+			continue
+		_build_chunk_collision(world_root, Vector3i(cx, cy, cz), decoded)
+		_chunks_loaded += 1
+
+	return world_root
+
+
+func _pick_room(rooms: Array) -> Variant:
+	for r in rooms:
+		if r.get("slug", "") == MAIN_ROOM_SLUG:
+			return r
+	if not rooms.is_empty():
+		return rooms[0]
+	return null
+
+
+func _spawn_pos_from_room(room: Dictionary) -> Vector3:
+	var spawn: Variant = room.get("spawn")
+	if spawn is Dictionary and (spawn as Dictionary).has("pos"):
+		var p: Array = (spawn as Dictionary)["pos"]
+		if p.size() >= 3:
+			return Vector3(float(p[0]), float(p[1]), float(p[2]))
+	return Vector3(0, 1.5, 0)
+
+
+## Build a StaticBody3D of box collision shapes for one decoded chunk: one
+## box per solid tile (TileDef.solid, world.md §3.2) plus one per static
+## (non-dynamic) object that isn't walkable. This is real collision derived
+## from the actual decoded payload -- there is no client-side precedent to
+## reuse (ChunkScene/TileLayerNode only build visuals; the dedicated server
+## has no renderer and needs physics bodies instead).
+func _build_chunk_collision(parent: Node3D, coord: Vector3i, data: ChunkDataDecoder.ChunkPayload) -> void:
+	var body := StaticBody3D.new()
+	body.name = "chunk_%d_%d_%d" % [coord.x, coord.y, coord.z]
+	var chunk_origin := Vector3(
+		coord.x * Chunk.CHUNK_SIZE_X,
+		coord.y * Chunk.CHUNK_SIZE_Y,
+		coord.z * Chunk.CHUNK_SIZE_Z
+	)
+
+	for layer in data.tile_layers:
+		for z in range(layer.depth):
+			for y in range(layer.height):
+				for x in range(layer.width):
+					var tile_id := layer.get_tile(x, y, z)
+					if tile_id == 0:
+						continue
+					var tile_def := layer.get_tile_def(tile_id)
+					if tile_def == null or not tile_def.solid:
+						continue
+					var shape := CollisionShape3D.new()
+					var box := BoxShape3D.new()
+					box.size = Vector3.ONE
+					shape.shape = box
+					shape.position = chunk_origin + Vector3(x + 0.5, y + 0.5, z + 0.5)
+					body.add_child(shape)
+					_collision_shapes_built += 1
+
+	for obj in data.objects:
+		if obj.dynamic:
+			continue  # dynamic templates are spawnable entities, not static geometry
+		if bool(obj.properties.get("walkable", true)):
+			continue
+		var shape := CollisionShape3D.new()
+		var box := BoxShape3D.new()
+		box.size = obj.scale
+		shape.shape = box
+		shape.position = obj.position
+		body.add_child(shape)
+		_collision_shapes_built += 1
+
+	if body.get_child_count() > 0:
+		parent.add_child(body)
+
+
+## Minimal internal-API HTTP GET, bearer-authenticated with --server-token
+## (world_persistence_impl.md §7.3). Deliberately separate from ApiClient:
+## that autoload's `request()` always decodes the response body as
+## UTF-8/JSON, which corrupts binary chunk payloads, and its auth model is
+## the player's JWT, not this process's per-spawn server token.
+func _internal_get_json(path: String) -> Dictionary:
+	var raw := await _internal_get(path, "application/json")
+	if not raw.ok:
+		return {"ok": false, "status": raw.status, "detail": raw.get("detail", ""), "data": null}
+	var parsed: Variant = JSON.parse_string(raw.body.get_string_from_utf8())
+	if not (parsed is Dictionary):
+		return {"ok": false, "status": raw.status, "detail": "invalid JSON response", "data": null}
+	return {"ok": true, "status": raw.status, "detail": "", "data": parsed}
+
+
+func _internal_get_binary(path: String) -> Dictionary:
+	return await _internal_get(path, "application/octet-stream")
+
+
+func _internal_get(path: String, accept: String) -> Dictionary:
+	var http := HTTPRequest.new()
+	http.timeout = CONTENT_FETCH_TIMEOUT
+	add_child(http)
+
+	var headers := PackedStringArray()
+	headers.append("Accept: " + accept)
+	headers.append("Authorization: Bearer " + server_token)
+
+	var url := api_url + path
+	var err := http.request(url, headers, HTTPClient.METHOD_GET)
+	if err != OK:
+		http.queue_free()
+		return {"ok": false, "status": 0, "detail": "request() failed to start (%d)" % err, "body": PackedByteArray()}
+
+	var args: Array = await http.request_completed
+	var result_code: int = args[0]
+	var response_code: int = args[1]
+	var body: PackedByteArray = args[3]
+	http.queue_free()
+
+	if result_code != HTTPRequest.RESULT_SUCCESS:
+		return {"ok": false, "status": 0, "detail": "transport error (%d)" % result_code, "body": PackedByteArray()}
+	if response_code < 200 or response_code >= 300:
+		return {"ok": false, "status": response_code, "detail": "HTTP %d" % response_code, "body": body}
+	return {"ok": true, "status": response_code, "detail": "", "body": body}
+
+
 func _physics_process(_delta: float) -> void:
 	if not is_running:
 		return
@@ -176,7 +421,8 @@ func _on_peer_disconnected(peer_id: int) -> void:
 
 
 # ---------------------------------------------------------------------------
-# Status file for the supervisor's idle reaper
+# Status file for the supervisor's idle reaper + readiness wait
+# (world_persistence_impl.md §7.4 / game_servers.py's _await_ready)
 # ---------------------------------------------------------------------------
 
 func _write_status() -> void:
@@ -190,5 +436,12 @@ func _write_status() -> void:
 			"players": players.size(),
 			"tick": tick,
 			"updated": Time.get_unix_time_from_system(),
+			# game_servers.py's _await_ready() polls for this -- only ever
+			# written true, and only once content has been decided (loaded
+			# or fallen back) AND create_server() has already succeeded.
+			"ready": is_running,
+			"content": _content_status,
+			"chunks_loaded": _chunks_loaded,
+			"collision_shapes": _collision_shapes_built,
 		}))
 		f.close()
diff --git a/scripts/smoke_test_http_chunk_source.gd b/scripts/smoke_test_http_chunk_source.gd
new file mode 100644
index 0000000..5718580
--- /dev/null
+++ b/scripts/smoke_test_http_chunk_source.gd
@@ -0,0 +1,189 @@
+extends SceneTree
+## Real integration test for HttpChunkSource against a live backend.
+##
+## Covers docs/architecture/world_persistence_impl.md D6/§8.4: the client
+## streams chunks over HTTP through the same ChunkStreamManager.set_server_rpc()
+## seam a fake/ENet source would use. This proves the adapter actually
+## satisfies that seam end-to-end -- a real PUT'd chunk decodes into a real
+## ChunkScene with real tile layers + objects, an unwritten coordinate
+## "renders empty" instead of erroring, and a second fetch of an unchanged
+## chunk round-trips through the 304/ETag cache path without failing.
+##
+## Run with:
+##     godot --headless --path . --script res://scripts/smoke_test_http_chunk_source.gd
+##
+## Requires the backend to be running (default base URL
+## `application/agp/api_base_url` -> http://127.0.0.1:8000, overridable via
+## AGP_API_BASE_URL).
+
+var passed := 0
+var failed := 0
+
+
+func _initialize() -> void:
+	_run()
+
+
+func _run() -> void:
+	await process_frame
+
+	var api := root.get_node("ApiClient")
+
+	randomize()
+	var unique_id: int = int(Time.get_unix_time_from_system() * 1000) + randi() % 100000
+	var username := "httpchunk_%d" % unique_id
+	var email := "httpchunk_%d@test.example.com" % unique_id
+	var reg: Dictionary = await api.register(username, email, "SmokeTest123!", "1990-01-01")
+	if not _check("setup: register succeeded", reg.ok):
+		print("  detail: ", reg.detail)
+		_finish()
+		return
+
+	var world_body := JSON.stringify({
+		"name": "HttpChunkSource Smoke World %d" % unique_id,
+		"visibility": "public",
+	})
+	var world_result: Dictionary = await api.request(
+		HTTPClient.METHOD_POST, "/api/worlds/", world_body, "application/json", true
+	)
+	if not _check("setup: world created", world_result.ok):
+		print("  detail: ", world_result.detail)
+		_finish()
+		return
+	var world_id: int = (world_result.data as Dictionary)["id"]
+
+	# Seed a real chunk at (0,0,0): one solid tile + one static object, built
+	# with the same client-side codec HttpChunkSource's decode path uses on
+	# the way back in (round-tripping through the real backend, not a mock).
+	var chunk_payload := _build_test_chunk(world_id)
+	var encoded := ChunkDataDecoder.encode_binary(chunk_payload)
+	# ApiClient.request() only takes a String body, which can't carry
+	# arbitrary binary safely -- PUT with a dedicated HTTPRequest instead
+	# (request_raw takes a real PackedByteArray), matching what the real
+	# backend test suite sends.
+	var put_result: Dictionary = await _put_binary(api, world_id, encoded)
+	if not _check("setup: chunk PUT succeeded", put_result.ok):
+		print("  detail: ", put_result.detail)
+		_finish()
+		return
+
+	# Build the real ChunkStreamManager + HttpChunkSource pair and wire them
+	# exactly as real gameplay code would.
+	var manager := ChunkStreamManager.new()
+	root.add_child(manager)
+	var source := HttpChunkSource.new()
+	source.world_id = str(world_id)
+	source.room_slug = "main"
+	source.stream_manager = manager
+	manager.add_child(source)
+	manager.set_server_rpc(source)
+
+	# 1. A written coordinate decodes into a real ChunkScene.
+	manager._load_chunk(Vector3i(0, 0, 0))
+	var chunk_scene: Node = await _wait_for_chunk(manager, Vector3i(0, 0, 0))
+	_check("written chunk (0,0,0) loaded via HttpChunkSource", chunk_scene != null)
+	if chunk_scene != null:
+		_check("loaded chunk has 1 tile layer", chunk_scene.tile_layers.size() == 1)
+		_check("loaded chunk tile layer has the expected solid tile",
+			chunk_scene.tile_layers[0].tiles[0] == 1)
+		_check("loaded chunk has 1 world object", chunk_scene.world_objects.size() == 1)
+
+	# 2. An unwritten coordinate renders empty, not an error.
+	manager._load_chunk(Vector3i(9, 9, 0))
+	var empty_scene: Node = await _wait_for_chunk(manager, Vector3i(9, 9, 0))
+	_check("unwritten chunk (9,9,0) still resolves (renders empty)", empty_scene != null)
+	if empty_scene != null:
+		_check("unwritten chunk has no tile layers", empty_scene.tile_layers.is_empty())
+
+	# 3. Re-fetching the same written chunk exercises the ETag/304 path
+	# (can't observe the wire status code directly here, but a stale cache
+	# entry or a broken conditional-request path would either error or
+	# silently return corrupt/empty data -- this proves it doesn't).
+	manager.force_reload_chunk(Vector3i(0, 0, 0))
+	var refetched: Node = await _wait_for_chunk(manager, Vector3i(0, 0, 0))
+	_check("re-fetch of unchanged chunk (0,0,0) round-trips via cache/304 path", refetched != null)
+	if refetched != null:
+		_check("re-fetched chunk still has 1 tile layer", refetched.tile_layers.size() == 1)
+
+	_finish()
+
+
+func _build_test_chunk(world_id: int) -> ChunkDataDecoder.ChunkPayload:
+	var layer := ChunkDataDecoder.TileLayerData.new()
+	layer.name = "ground"
+	layer.layer_id = 0
+	layer.width = 32
+	layer.height = 32
+	layer.depth = 1
+	var empty_def := TileDef.new(0, "", false, 0, 1)
+	var solid_def := TileDef.new(1, "tiles/stone", true, 0, 1)
+	layer.palette = [empty_def, solid_def]
+	var tiles := PackedInt32Array()
+	tiles.resize(32 * 32 * 1)
+	tiles[0] = 1  # (x=0, y=0, z=0) solid
+	layer.tiles = tiles
+
+	var obj := ChunkDataDecoder.WorldObjectData.new()
+	obj.id = "obj_1"
+	obj.type = "prop"
+	obj.asset_id = "props/rock"
+	obj.position = Vector3(1.0, 1.0, 0.0)
+	obj.properties = {"tag": "smoke"}
+
+	var payload := ChunkDataDecoder.ChunkPayload.new()
+	payload.world_id = str(world_id)
+	payload.room_id = "main"
+	payload.coord = Vector3i(0, 0, 0)
+	payload.bounds = AABB(Vector3.ZERO, Vector3(32, 32, 1))
+	payload.tile_layers = [layer]
+	payload.objects = [obj]
+	payload.revision = 1
+	payload.author = "smoke_test"
+	payload.created_at = 1700000000
+	payload.updated_at = 1700000000
+	return payload
+
+
+func _put_binary(api: Object, world_id: int, bytes: PackedByteArray) -> Dictionary:
+	var http := HTTPRequest.new()
+	root.add_child(http)
+	var bearer: String = api.bearer_token
+	var headers := PackedStringArray([
+		"Content-Type: application/octet-stream",
+		"X-Base-Revision: 0",
+		"Authorization: " + bearer,
+	])
+	var base: String = api.base_url
+	var url := base + "/api/worlds/%d/rooms/main/chunks/0/0/0" % world_id
+	var err := http.request_raw(url, headers, HTTPClient.METHOD_PUT, bytes)
+	if err != OK:
+		http.queue_free()
+		return {"ok": false, "detail": "request_raw failed (%d)" % err}
+	var args: Array = await http.request_completed
+	var response_code: int = args[1]
+	var body: PackedByteArray = args[3]
+	http.queue_free()
+	return {"ok": response_code >= 200 and response_code < 300, "status": response_code, "detail": body.get_string_from_utf8()}
+
+
+func _wait_for_chunk(manager: ChunkStreamManager, coord: Vector3i, max_frames: int = 120) -> Node:
+	for i in max_frames:
+		await process_frame
+		if manager.is_chunk_loaded(coord):
+			return manager.get_loaded_chunk(coord)
+	return null
+
+
+func _check(description: String, condition: bool) -> bool:
+	if condition:
+		print("  PASS: ", description)
+		passed += 1
+	else:
+		printerr("  FAIL: ", description)
+		failed += 1
+	return condition
+
+
+func _finish() -> void:
+	print("\n=== SMOKE RESULT: %d/%d passed ===" % [passed, passed + failed])
+	quit(0 if failed == 0 else 1)