refactor(M03): extract M2 mesh load pipeline state

Work-Package: M03-RND-M2-MESH-LOAD-PIPELINE-001
Agent: sindo-main-codex
Tests: 34 headless contract regressions pass; documentation and coordination gates pass; checkpoint dry-run 7/7
Fidelity: preserves request insertion polling order, terminal completion FIFO/statuses, metrics and clear/shutdown behavior; no visual parity claim

src/render/m2/m2_mesh_load_pipeline_state.gd

@@ -0,0 +1,103 @@
+class_name M2MeshLoadPipelineState
+extends RefCounted
+
+## Owns static M2 threaded-load request records and the terminal finalize FIFO.
+## The caller owns ResourceLoader I/O, permits, Mesh caches and materialization.
+
+var _request_by_normalized_path: Dictionary = {}
+var _finalize_records: Array[Dictionary] = []
+
+
+## Records one normalized M2 path and ResourceLoader path. Empty or duplicate
+## values are rejected without mutation.
+func remember_request(normalized_relative_path: String, resource_path: String) -> bool:
+	if normalized_relative_path.is_empty() or resource_path.is_empty():
+		return false
+	if _request_by_normalized_path.has(normalized_relative_path):
+		return false
+	_request_by_normalized_path[normalized_relative_path] = {
+		"normalized": normalized_relative_path,
+		"path": resource_path,
+	}
+	return true
+
+
+## Returns whether one normalized path currently has a pending threaded request.
+func has_request(normalized_relative_path: String) -> bool:
+	return (
+		not normalized_relative_path.is_empty()
+		and _request_by_normalized_path.has(normalized_relative_path)
+	)
+
+
+## Returns detached pending records in their existing Dictionary insertion order.
+func request_records_snapshot() -> Array[Dictionary]:
+	var records: Array[Dictionary] = []
+	for request_variant in _request_by_normalized_path.values():
+		var request: Dictionary = request_variant
+		records.append(request.duplicate())
+	return records
+
+
+## Removes one pending request and appends its terminal status to the finalize
+## FIFO. Unknown paths are rejected without mutation.
+func complete_request(normalized_relative_path: String, terminal_status: int) -> bool:
+	if not _request_by_normalized_path.has(normalized_relative_path):
+		return false
+	var request: Dictionary = _request_by_normalized_path[normalized_relative_path]
+	_request_by_normalized_path.erase(normalized_relative_path)
+	var finalize_record := request.duplicate()
+	finalize_record["status"] = terminal_status
+	_finalize_records.append(finalize_record)
+	return true
+
+
+## Removes one pending request without enqueuing finalization. Returns whether a
+## record was removed.
+func discard_request(normalized_relative_path: String) -> bool:
+	return _request_by_normalized_path.erase(normalized_relative_path)
+
+
+## Returns whether a terminal record is waiting for budgeted finalization.
+func has_finalize_record() -> bool:
+	return not _finalize_records.is_empty()
+
+
+## Removes and returns the oldest terminal record. Empty means none is ready.
+func pop_finalize_record() -> Dictionary:
+	if _finalize_records.is_empty():
+		return {}
+	return _finalize_records.pop_front()
+
+
+## Returns pending plus terminal-finalize work for existing renderer metrics.
+func total_work_count() -> int:
+	return _request_by_normalized_path.size() + _finalize_records.size()
+
+
+## Returns the pending request count.
+func pending_request_count() -> int:
+	return _request_by_normalized_path.size()
+
+
+## Returns the terminal finalize FIFO count.
+func finalize_record_count() -> int:
+	return _finalize_records.size()
+
+
+## Clears request/finalize bookkeeping. The caller must drain ResourceLoader
+## requests before orderly shutdown clear.
+func clear() -> void:
+	_request_by_normalized_path.clear()
+	_finalize_records.clear()
+
+
+## Returns detached request and finalize records without loaded Resources/Meshes.
+func diagnostic_snapshot() -> Dictionary:
+	var finalize_records: Array[Dictionary] = []
+	for record in _finalize_records:
+		finalize_records.append(record.duplicate())
+	return {
+		"requests": request_records_snapshot(),
+		"finalize_records": finalize_records,
+	}

src/render/m2/m2_mesh_load_pipeline_state.gd.uid

@@ -0,0 +1 @@
+uid://dadsxuswq0tnd

src/scenes/streaming/streaming_world_loader.gd

@@ -63,6 +63,9 @@ const M2_BUILD_BATCH_PLANNER_SCRIPT := preload(
 const M2_RUNTIME_MESH_REBUILD_CLASSIFIER_SCRIPT := preload(
 	"res://src/render/m2/m2_runtime_mesh_rebuild_classifier.gd"
 )
+const M2_MESH_LOAD_PIPELINE_STATE_SCRIPT := preload(
+	"res://src/render/m2/m2_mesh_load_pipeline_state.gd"
+)
 const STREAMING_TARGET_PLANNER_SCRIPT := preload("res://src/render/streaming/streaming_target_planner.gd")
 const STREAMING_TARGET_POLICY_SCRIPT := preload("res://src/render/streaming/streaming_target_policy.gd")
 const RENDER_BUDGET_SCHEDULER_SCRIPT := preload("res://src/render/streaming/render_budget_scheduler.gd")
@@ -254,8 +257,7 @@ var _m2_runtime_mesh_rebuild_classifier := (
 	M2_RUNTIME_MESH_REBUILD_CLASSIFIER_SCRIPT.new()
 )
 var _m2_mesh_cache: Dictionary = {}
-var _m2_mesh_load_requests: Dictionary = {}
-var _m2_mesh_finalize_queue: Array = []
+var _m2_mesh_load_pipeline_state := M2_MESH_LOAD_PIPELINE_STATE_SCRIPT.new()
 var _m2_animation_load_requests: Dictionary = {}
 var _m2_animation_finalize_queue: Array = []
 var _wmo_build_jobs: Dictionary = {}
@@ -1011,7 +1013,7 @@ func _log_hitch_profile(start_usec: int, timings: Array[String], did_refresh: bo
 		_detail_asset_queue.size(),
 		_m2_group_tasks.size(),
 		_m2_animation_finalize_queue.size() + _m2_animation_load_requests.size(),
-		_m2_mesh_finalize_queue.size() + _m2_mesh_load_requests.size(),
+		_m2_mesh_load_pipeline_state.total_work_count(),
 		_m2_build_queue.size(),
 		(
 			_wmo_build_queue.size()
@@ -1042,7 +1044,7 @@ func render_baseline_snapshot() -> Dictionary:
 			"detail": _detail_asset_queue.size(),
 			"m2_task": _m2_group_tasks.size(),
 			"m2_animation": _m2_animation_finalize_queue.size() + _m2_animation_load_requests.size(),
-			"m2_mesh": _m2_mesh_finalize_queue.size() + _m2_mesh_load_requests.size(),
+			"m2_mesh": _m2_mesh_load_pipeline_state.total_work_count(),
 			"m2_build": _m2_build_queue.size(),
 			"wmo_build": (
 				_wmo_build_queue.size()
@@ -1328,7 +1330,7 @@ func _tick_runtime_stats(delta: float) -> void:
 		_detail_asset_queue.size(),
 		_m2_group_tasks.size(),
 		_m2_animation_finalize_queue.size() + _m2_animation_load_requests.size(),
-		_m2_mesh_finalize_queue.size() + _m2_mesh_load_requests.size(),
+		_m2_mesh_load_pipeline_state.total_work_count(),
 		_m2_build_queue.size(),
 		(
 			_wmo_build_queue.size()
@@ -2338,15 +2340,14 @@ func _wait_for_tile_tasks() -> void:
 	_m2_group_result_queue.clear()
 	_m2_group_result_mutex.unlock()
 
-	for pending in _m2_mesh_load_requests.values():
+	for pending in _m2_mesh_load_pipeline_state.request_records_snapshot():
 		var path: String = String(pending.get("path", ""))
 		if path.is_empty():
 			continue
 		var status := ResourceLoader.load_threaded_get_status(path)
 		if status == ResourceLoader.THREAD_LOAD_IN_PROGRESS or status == ResourceLoader.THREAD_LOAD_LOADED:
 			ResourceLoader.load_threaded_get(path)
-	_m2_mesh_load_requests.clear()
-	_m2_mesh_finalize_queue.clear()
+	_m2_mesh_load_pipeline_state.clear()
 	_m2_runtime_mesh_rebuild_classifier.clear()
 
 	for pending in _m2_animation_load_requests.values():
@@ -2996,8 +2997,7 @@ func _clear_streamed_world() -> void:
 		_cancel_m2_build_job(String(key))
 	_m2_build_queue.clear()
 	_m2_unique_placement_registry.clear()
-	_m2_mesh_load_requests.clear()
-	_m2_mesh_finalize_queue.clear()
+	_m2_mesh_load_pipeline_state.clear()
 	_m2_runtime_mesh_rebuild_classifier.clear()
 	_m2_animation_load_requests.clear()
 	_m2_animation_finalize_queue.clear()
@@ -4247,24 +4247,21 @@ func _drain_m2_animation_loads() -> void:
 
 
 func _drain_m2_mesh_loads() -> void:
-	for normalized_rel_variant in _m2_mesh_load_requests.keys():
-		var normalized_rel := String(normalized_rel_variant)
-		var pending: Dictionary = _m2_mesh_load_requests[normalized_rel]
+	for pending in _m2_mesh_load_pipeline_state.request_records_snapshot():
+		var normalized_rel := String(pending.get("normalized", ""))
 		var path: String = String(pending.get("path", ""))
 		if path.is_empty():
-			_m2_mesh_load_requests.erase(normalized_rel)
+			_m2_mesh_load_pipeline_state.discard_request(normalized_rel)
 			_m2_missing_cache[normalized_rel] = true
 			continue
 		var status := ResourceLoader.load_threaded_get_status(path)
 		if status != ResourceLoader.THREAD_LOAD_LOADED and status != ResourceLoader.THREAD_LOAD_FAILED:
 			continue
-		pending["status"] = status
-		_m2_mesh_load_requests.erase(normalized_rel)
-		_m2_mesh_finalize_queue.append(pending)
+		_m2_mesh_load_pipeline_state.complete_request(normalized_rel, status)
 
-	while not _m2_mesh_finalize_queue.is_empty() and _render_budget_scheduler.try_consume_permit(
+	while _m2_mesh_load_pipeline_state.has_finalize_record() and _render_budget_scheduler.try_consume_permit(
 			RENDER_BUDGET_SCHEDULER_SCRIPT.M2_MESH_FINALIZE):
-		var pending: Dictionary = _m2_mesh_finalize_queue.pop_front()
+		var pending: Dictionary = _m2_mesh_load_pipeline_state.pop_finalize_record()
 		var normalized_rel := String(pending.get("normalized", ""))
 		if normalized_rel.is_empty() or _m2_mesh_cache.has(normalized_rel):
 			continue
@@ -4620,7 +4617,7 @@ func _get_m2_mesh_or_request(rel_path: String) -> Mesh:
 
 
 func _request_m2_mesh_load(normalized_rel: String) -> void:
-	if normalized_rel.is_empty() or _m2_mesh_load_requests.has(normalized_rel):
+	if normalized_rel.is_empty() or _m2_mesh_load_pipeline_state.has_request(normalized_rel):
 		return
 
 	for cache_res_path in _get_m2_cache_resource_paths(normalized_rel, [".tscn", ".glb"]):
@@ -4634,10 +4631,7 @@ func _request_m2_mesh_load(normalized_rel: String) -> void:
 			false,
 			ResourceLoader.CACHE_MODE_REUSE)
 		if err == OK or err == ERR_BUSY:
-			_m2_mesh_load_requests[normalized_rel] = {
-				"normalized": normalized_rel,
-				"path": cache_res_path,
-			}
+			_m2_mesh_load_pipeline_state.remember_request(normalized_rel, cache_res_path)
 			return
 		break
 

src/tools/verify_m2_mesh_load_pipeline_state.gd

@@ -0,0 +1,166 @@
+extends SceneTree
+
+## Synthetic request/terminal/FIFO/lifecycle/boundary/timing regression for
+## static M2 threaded mesh loading.
+
+const PIPELINE_SCRIPT := preload("res://src/render/m2/m2_mesh_load_pipeline_state.gd")
+const PIPELINE_PATH := "res://src/render/m2/m2_mesh_load_pipeline_state.gd"
+const LOADER_PATH := "res://src/scenes/streaming/streaming_world_loader.gd"
+
+
+func _initialize() -> void:
+	var failures: Array[String] = []
+	_verify_validation_dedupe_and_order(failures)
+	_verify_terminal_transition_and_fifo(failures)
+	_verify_discard_metrics_clear_and_diagnostics(failures)
+	_verify_ownership_boundaries(failures)
+	var elapsed_milliseconds := _verify_bounded_timing(failures)
+	if not failures.is_empty():
+		for failure in failures:
+			push_error("M2_MESH_LOAD_PIPELINE_STATE: %s" % failure)
+		quit(1)
+		return
+	print(
+		"M2_MESH_LOAD_PIPELINE_STATE PASS cases=11 iterations=100 elapsed_ms=%.3f"
+		% elapsed_milliseconds
+	)
+	quit(0)
+
+
+func _verify_validation_dedupe_and_order(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	_expect_false(pipeline.call("remember_request", "", "res://a.tscn"), "empty normalized path rejected", failures)
+	_expect_false(pipeline.call("remember_request", "world/a.m2", ""), "empty resource path rejected", failures)
+	_expect_true(pipeline.call("remember_request", "world/a.m2", "res://a.tscn"), "first request accepted", failures)
+	_expect_false(pipeline.call("remember_request", "world/a.m2", "res://duplicate.glb"), "duplicate request rejected", failures)
+	_expect_true(pipeline.call("remember_request", "world/b.m2", "res://b.glb"), "second request accepted", failures)
+	_expect_true(pipeline.call("has_request", "world/a.m2"), "request observable", failures)
+	var records: Array[Dictionary] = pipeline.call("request_records_snapshot")
+	_expect_equal(records.size(), 2, "request record count", failures)
+	_expect_string_equal(String(records[0]["normalized"]), "world/a.m2", "request insertion order first", failures)
+	_expect_string_equal(String(records[1]["normalized"]), "world/b.m2", "request insertion order second", failures)
+
+
+func _verify_terminal_transition_and_fifo(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	pipeline.call("remember_request", "world/a.m2", "res://a.tscn")
+	pipeline.call("remember_request", "world/b.m2", "res://b.glb")
+	_expect_false(pipeline.call("complete_request", "world/missing.m2", 3), "unknown completion rejected", failures)
+	_expect_true(pipeline.call("complete_request", "world/b.m2", 4), "second request completes first", failures)
+	_expect_true(pipeline.call("complete_request", "world/a.m2", 3), "first request completes second", failures)
+	_expect_false(pipeline.call("has_request", "world/a.m2"), "completed request removed", failures)
+	_expect_equal(int(pipeline.call("pending_request_count")), 0, "all requests completed", failures)
+	_expect_equal(int(pipeline.call("finalize_record_count")), 2, "two terminal records", failures)
+	var first: Dictionary = pipeline.call("pop_finalize_record")
+	var second: Dictionary = pipeline.call("pop_finalize_record")
+	_expect_string_equal(String(first["normalized"]), "world/b.m2", "completion FIFO first", failures)
+	_expect_equal(int(first["status"]), 4, "first opaque status retained", failures)
+	_expect_string_equal(String(second["normalized"]), "world/a.m2", "completion FIFO second", failures)
+	_expect_equal(int(second["status"]), 3, "second opaque status retained", failures)
+	_expect_false(pipeline.call("has_finalize_record"), "finalize FIFO drained", failures)
+	_expect_true((pipeline.call("pop_finalize_record") as Dictionary).is_empty(), "empty pop returns Dictionary", failures)
+
+
+func _verify_discard_metrics_clear_and_diagnostics(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	pipeline.call("remember_request", "world/a.m2", "res://a.tscn")
+	pipeline.call("remember_request", "world/b.m2", "res://b.glb")
+	_expect_true(pipeline.call("discard_request", "world/a.m2"), "pending request discarded", failures)
+	_expect_false(pipeline.call("discard_request", "world/a.m2"), "discard is idempotent", failures)
+	pipeline.call("complete_request", "world/b.m2", 3)
+	_expect_equal(int(pipeline.call("total_work_count")), 1, "total includes finalize only", failures)
+	var snapshot: Dictionary = pipeline.call("diagnostic_snapshot")
+	(snapshot["finalize_records"] as Array)[0]["path"] = "mutated"
+	var fresh_snapshot: Dictionary = pipeline.call("diagnostic_snapshot")
+	_expect_string_equal(
+		String((fresh_snapshot["finalize_records"] as Array)[0]["path"]),
+		"res://b.glb",
+		"diagnostics detached",
+		failures
+	)
+	pipeline.call("clear")
+	pipeline.call("clear")
+	_expect_equal(int(pipeline.call("total_work_count")), 0, "clear removes all work", failures)
+
+
+func _verify_ownership_boundaries(failures: Array[String]) -> void:
+	var pipeline_source := FileAccess.get_file_as_string(PIPELINE_PATH)
+	var loader_source := FileAccess.get_file_as_string(LOADER_PATH)
+	_expect_true(
+		loader_source.contains("M2_MESH_LOAD_PIPELINE_STATE_SCRIPT.new()"),
+		"loader composes pipeline state",
+		failures
+	)
+	_expect_false(loader_source.contains("var _m2_mesh_load_requests:"), "legacy request field removed", failures)
+	_expect_false(loader_source.contains("var _m2_mesh_finalize_queue:"), "legacy finalize field removed", failures)
+	_expect_equal(
+		loader_source.count("_m2_mesh_load_pipeline_state.total_work_count()"),
+		3,
+		"three existing metrics delegate",
+		failures
+	)
+	_expect_equal(
+		loader_source.count("_m2_mesh_load_pipeline_state.clear()"),
+		2,
+		"two existing clear sites delegate",
+		failures
+	)
+	for retained_loader_rule in [
+		"ResourceLoader.load_threaded_request(",
+		"ResourceLoader.load_threaded_get_status(path)",
+		"ResourceLoader.load_threaded_get(path)",
+		"RENDER_BUDGET_SCHEDULER_SCRIPT.M2_MESH_FINALIZE",
+		"_m2_mesh_cache[normalized_rel]",
+		"_m2_missing_cache[normalized_rel]",
+	]:
+		_expect_true(loader_source.contains(retained_loader_rule), "loader retains %s" % retained_loader_rule, failures)
+	for forbidden_dependency in [
+		"ResourceLoader.",
+		"WorkerThreadPool.",
+		": Mesh",
+		"Node3D",
+		"RID(",
+		"_m2_mesh_cache",
+		"_m2_missing_cache",
+	]:
+		_expect_false(
+			pipeline_source.contains(forbidden_dependency),
+			"pipeline omits %s ownership" % forbidden_dependency,
+			failures
+		)
+
+
+func _verify_bounded_timing(failures: Array[String]) -> float:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	var started_microseconds := Time.get_ticks_usec()
+	for _iteration in range(100):
+		for path_index in range(256):
+			var normalized_path := "world/model_%d.m2" % path_index
+			pipeline.call("remember_request", normalized_path, "res://model_%d.tscn" % path_index)
+		for path_index in range(256):
+			pipeline.call("complete_request", "world/model_%d.m2" % path_index, 3)
+		while pipeline.call("has_finalize_record"):
+			pipeline.call("pop_finalize_record")
+		pipeline.call("clear")
+	var elapsed_milliseconds := float(Time.get_ticks_usec() - started_microseconds) / 1000.0
+	_expect_true(elapsed_milliseconds < 1000.0, "100 by 256 transitions under 1 second", failures)
+	return elapsed_milliseconds
+
+
+func _expect_true(condition: bool, label: String, failures: Array[String]) -> void:
+	if not condition:
+		failures.append(label)
+
+
+func _expect_false(condition: bool, label: String, failures: Array[String]) -> void:
+	_expect_true(not condition, label, failures)
+
+
+func _expect_equal(actual: int, expected: int, label: String, failures: Array[String]) -> void:
+	if actual != expected:
+		failures.append("%s expected=%d actual=%d" % [label, expected, actual])
+
+
+func _expect_string_equal(actual: String, expected: String, label: String, failures: Array[String]) -> void:
+	if actual != expected:
+		failures.append("%s expected=%s actual=%s" % [label, expected, actual])

src/tools/verify_m2_mesh_load_pipeline_state.gd.uid

@@ -0,0 +1 @@
+uid://bdn3pdtg62ou1