refactor(M03): extract ADT water load pipeline state

RENDER.md

@@ -33,6 +33,7 @@ Paired run 2026-07-11 подтвердил крупный coordinate/placement g
 - `src/render/wmo/wmo_render_build_queue.gd` / `wmo_render_build_job.gd` - typed pending group jobs, FIFO placement keys and strong root/resource references without engine destruction.
 - `src/render/wmo/wmo_render_resource_cache_state.gd` - validated lightweight WMO render Resources, negative cache and pending cache paths without ResourceLoader I/O.
 - `src/render/wmo/wmo_scene_resource_cache_state.gd` - validated cached-WMO PackedScenes, negative cache and pending `.tscn` paths without file/I/O/Node ownership.
+- `src/render/liquid/adt_water_load_pipeline_state.gd` - ADT water pending FIFO/dedupe, active task IDs and worker-safe parsed-result mailbox without parser or Node ownership.
 - `src/scenes/streaming/eastern_kingdoms_streaming.tscn` - текущая сцена для проверки Azeroth/Eastern Kingdoms.
 - `addons/mpq_extractor/loaders/adt_builder.gd` - сборка ADT terrain, control splat материалов и MH2O liquids.
 - `addons/mpq_extractor/loaders/m2_builder.gd` - сборка статического M2 mesh/material.
@@ -1077,6 +1078,21 @@ $exe = Join-Path $env:TEMP 'godot-4.6.1-openwc\Godot_v4.6.1-stable_win64.exe'
 - Asset-backed oversize/stale fixtures, traversal/leak p95/p99 and paired fidelity
   evidence remain pending.
 
+## 2026-07-17 ADT Water Load Pipeline State Extraction
+
+- `AdtWaterLoadPipelineState` now owns ADT water pending FIFO/deduplication,
+  active WorkerThreadPool task IDs and the mutex-protected parsed-result mailbox.
+- `StreamingWorldLoader` still selects concurrency, starts/waits tasks, invokes
+  `ADTLoader`, consumes `WATER_FINALIZE` permits, rejects stale tile/path/root
+  results and builds/attaches water Nodes through `ADTBuilder` on the main thread.
+- Tile release still cancels only pending water requests; active work completes
+  and its result is discarded when tile ownership is stale. Shutdown waits task
+  IDs before clearing state; reset does not interrupt workers.
+- Pending-plus-active `water` metrics, parsing, formats, materials, profiles and
+  visible behavior are unchanged.
+- Asset-backed MH2O/MCLQ fidelity, long-traversal leak/p95/p99 and paired capture
+  evidence remain pending.
+
 ## 2026-07-16 World Environment Snapshot Facade
 
 - `WorldEnvironmentSnapshot` carries one immutable finite time-of-day value,

docs/modules/README.md

@@ -15,6 +15,7 @@
 | Terrain quality Mesh cache | Implemented extraction | [`terrain-quality-mesh-cache.md`](terrain-quality-mesh-cache.md) |
 | Terrain chunk LOD planner | Implemented extraction | [`terrain-chunk-lod-planner.md`](terrain-chunk-lod-planner.md) |
 | Terrain chunk geometry queue planner | Implemented extraction | [`terrain-chunk-geometry-queue-planner.md`](terrain-chunk-geometry-queue-planner.md) |
+| ADT water load pipeline state | Implemented extraction | [`adt-water-load-pipeline-state.md`](adt-water-load-pipeline-state.md) |
 | M2 unique placement registry | Implemented extraction | [`m2-unique-placement-registry.md`](m2-unique-placement-registry.md) |
 | M2 placement transform resolver | Implemented | [`m2-placement-transform-resolver.md`](m2-placement-transform-resolver.md) |
 | M2 placement grouper | Implemented extraction | [`m2-placement-grouper.md`](m2-placement-grouper.md) |

docs/modules/adt-water-load-pipeline-state.md

@@ -0,0 +1,247 @@
+# ADT Water Load Pipeline State
+
+## Metadata
+
+| Field | Value |
+|---|---|
+| Status | Implemented |
+| Target/work package | M03 / `M03-RND-ADT-WATER-LOAD-PIPELINE-001` |
+| Owners | ADT water pending FIFO/dedupe, active task IDs and worker-safe result mailbox |
+| Last verified | Worktree `work/sindo-main-codex/m03-adt-water-load-pipeline`, 2026-07-17 |
+| Profiles/capabilities | ADT MH2O/MCLQ water loading; profile-independent state |
+
+## Purpose
+
+Own the cross-frame and cross-thread bookkeeping for asynchronous ADT water
+loads outside `StreamingWorldLoader`: FIFO requests, queued-key deduplication,
+opaque active task IDs and parsed-result delivery. The loader retains task
+execution, parsing, budgets, stale-result decisions and scene finalization.
+
+## Non-goals
+
+- Start, cancel or wait WorkerThreadPool tasks.
+- Instantiate `ADTLoader` or interpret MH2O/MCLQ data.
+- Select concurrency limits or consume render permits.
+- Create, attach or free Nodes, Meshes, materials or RIDs.
+- Change liquid visuals, coordinates, tile acceptance or loaded-state timing.
+
+## Context and boundaries
+
+```mermaid
+flowchart LR
+    Tile[Loaded tile needing water] --> Loader[StreamingWorldLoader]
+    Loader --> State[AdtWaterLoadPipelineState]
+    State --> Request[Oldest tile/path request]
+    Request --> Loader
+    Loader --> Worker[WorkerThreadPool ADTLoader task]
+    Worker --> State
+    State --> Result[Parsed ADT result]
+    Result --> Loader
+    Loader --> Budget[WATER_FINALIZE permit]
+    Loader --> Builder[ADTBuilder water scene]
+    Builder --> Scene[Main-thread tile Node attach]
+```
+
+Allowed dependencies are RefCounted, Dictionary/Array/String and one Godot
+`Mutex` around results. WorkerThreadPool, ADTLoader, scheduler, tile state,
+scene-tree/render resources and gameplay/editor/network are forbidden.
+
+## Public API
+
+| Symbol | Kind | Purpose | Thread/lifetime | Errors |
+|---|---|---|---|---|
+| `enqueue_request(tile_key, adt_path)` | Command/query | Append deduplicated FIFO request | Renderer main thread; until dequeue/cancel/clear | Empty/queued/active returns false |
+| `take_next_request()` | Command/query | Pop oldest request and release queued marker | Renderer main thread | Empty returns `{}` |
+| `cancel_pending_requests(tile_key)` | Command/query | Remove all matching pending entries | Tile release/main thread | Empty/unknown removes zero; active unaffected |
+| `remember_active_task(tile_key, task_id)` | Command/query | Retain loader-created opaque task ID | Main thread; until completion/clear | Empty/duplicate returns false |
+| `has_active_task` / `active_task_id_for` | Query | Test/borrow active task identity | Main thread | Unknown ID is `-1` |
+| `complete_active_task(tile_key)` | Command/query | Remove and return completed task ID | Main thread after wait | Unknown returns `-1` |
+| `active_task_ids_snapshot()` | Query | Copy IDs for orderly shutdown waiting | Main thread | Detached Array |
+| `publish_result(tile_key, path, data)` | Command/query | Append parsed result under mutex | Worker or main thread | Empty key/path returns false |
+| `has_result()` / `pop_result()` | Query/command | Observe/pop oldest result under mutex | Main-thread drain | Empty pop returns `{}` |
+| `total_load_count()` | Query | Pending plus active metric | Main thread | Results intentionally excluded |
+| `pending_request_count()` / `active_task_count()` | Query | Support loader concurrency loop | Main thread | None |
+| `clear()` | Command | Release all bookkeeping and results | Main thread after wait or map reset | Does not interrupt workers |
+| `diagnostic_snapshot()` | Query | Return detached scalar requests/tasks/results | Main thread; result mailbox locked | Water data omitted |
+
+## Inputs and outputs
+
+| Direction | Contract/data | Producer | Consumer | Ownership | Thread/lifetime |
+|---|---|---|---|---|---|
+| Input | Tile key and ADT path | Loader tile state | Pending FIFO | Copied Strings | Until dequeue/cancel/clear |
+| Input | Opaque task ID | Loader WorkerThreadPool adapter | Active task map | Integer value | Until wait/completion/clear |
+| Input | Tile/path/parsed ADT Dictionary | Worker parser adapter | Result mailbox | Dictionary reference retained | Until pop/clear |
+| Output | Oldest request Dictionary | Pipeline state | Loader task-start adapter | Fresh record from FIFO | One start attempt |
+| Output | Active task ID snapshot | Pipeline state | Loader shutdown wait | Detached Array | Shutdown loop |
+| Output | Oldest result Dictionary | Pipeline state | Loader finalize adapter | Mailbox record transferred | One finalize attempt |
+| Output | Scalar diagnostics | Pipeline state | Verifier/future metrics | Detached records | Snapshot lifetime |
+
+Side effects are limited to collection mutation, reference retention and result
+mutex locking. Parsing, permits and scene mutation remain outside the module.
+
+## Data flow
+
+```mermaid
+flowchart TD
+    Enqueue[enqueue tile/path] --> Duplicate{Queued or active?}
+    Duplicate -->|yes| Reject[false; unchanged]
+    Duplicate -->|no| FIFO[Append FIFO and queued key]
+    FIFO --> Take[take next; release queued marker]
+    Take --> Validate[Loader checks tile/path]
+    Validate --> Start[Loader starts worker and remembers task ID]
+    Start --> Parse[Worker parses ADT]
+    Parse --> Publish[Mutex-protected publish result]
+    Publish --> Permit[Loader obtains WATER_FINALIZE permit]
+    Permit --> Pop[Pop oldest result]
+    Pop --> Wait[Loader waits task and completes active ID]
+    Wait --> Stale{Tile/path/root current?}
+    Stale -->|no| Discard[Discard result]
+    Stale -->|yes| Attach[Build and attach water scene]
+```
+
+## Lifecycle/state
+
+A tile request progresses through `Absent`, `Queued`, `Active` and
+`ResultReady`; the loader wait/completion returns bookkeeping to `Absent` before
+validating/attaching the result. Queued tile release cancels only `Queued`.
+Active work is not interrupted; a later stale result is drained and discarded.
+
+```mermaid
+stateDiagram-v2
+    [*] --> Absent
+    Absent --> Queued: accepted enqueue
+    Queued --> Absent: pending cancellation or invalid dequeue
+    Queued --> Active: loader starts task
+    Active --> ResultReady: worker publishes result
+    ResultReady --> Absent: loader waits and completes task
+    Queued --> Absent: clear
+    Active --> Absent: clear bookkeeping only
+    ResultReady --> Absent: clear
+```
+
+## Main sequence
+
+```mermaid
+sequenceDiagram
+    participant Loader as StreamingWorldLoader main thread
+    participant State as AdtWaterLoadPipelineState
+    participant Worker as WorkerThreadPool task
+    participant Budget as RenderBudgetScheduler
+    participant Builder as ADTBuilder
+    Loader->>State: enqueue_request(tile key, ADT path)
+    Loader->>State: take_next_request()
+    Loader->>Worker: add_task(parse binding)
+    Loader->>State: remember_active_task(key, task ID)
+    Worker->>Worker: ADTLoader.load_adt(path)
+    Worker->>State: publish_result(key, path, data)
+    Note over State: result mutex protects publication/pop
+    Loader->>Budget: try_consume_permit(WATER_FINALIZE)
+    Loader->>State: pop_result()
+    Loader->>Worker: wait_for_task_completion(task ID)
+    Loader->>State: complete_active_task(key)
+    Loader->>Loader: reject stale tile/path/root
+    Loader->>Builder: build_tile_water_scene(data, origin)
+    Loader->>Loader: attach Node and mark water_loaded
+```
+
+## Dependency diagram
+
+```mermaid
+flowchart TB
+    Loader[StreamingWorldLoader] --> State[AdtWaterLoadPipelineState]
+    Loader --> Scheduler[RenderBudgetScheduler]
+    Loader --> Worker[WorkerThreadPool / ADTLoader]
+    Loader --> Builder[ADTBuilder / SceneTree]
+    Worker -->|publish only| State
+    State -. no dependency .-> Scheduler
+    State -. no dependency .-> Builder
+```
+
+## Ownership, threading and resources
+
+- Main thread owns pending and active-task collection mutation.
+- Worker threads may call only `publish_result`; result observe/pop/clear also lock.
+- Parsed water Dictionary identity is retained until main-thread pop.
+- Loader owns WorkerThreadPool task lifetime and must wait before orderly shutdown clear.
+- Loader owns tile states, ADTBuilder, water Nodes and every render resource.
+- `clear` cannot cancel a worker; post-reset publication becomes a stale result
+  and is discarded because no active task/tile ownership remains.
+
+## Errors, cancellation and recovery
+
+| Failure | Detection | Behavior | Diagnostic | Recovery |
+|---|---|---|---|---|
+| Empty/duplicate request | Enqueue guards | Reject unchanged | Contract verifier | Correct caller state |
+| Tile removed while queued | Loader release | Remove matching pending requests | Source/lifecycle verifier | Reload can enqueue later |
+| Tile removed while active | Loader stale check after result | Wait task, remove active ID, discard result | Existing metrics | Reload can enqueue later |
+| Parser unavailable/empty | Worker publishes empty Dictionary | Loader marks `water_loaded` without Node | Existing behavior | Future tile reload |
+| Path changed | Loader compares result path | Discard after task completion | Source regression | Current tile can request again |
+| Shutdown | Loader snapshots/waits active IDs | Clear all bookkeeping/results | Shutdown/source verifier | New loader starts empty |
+
+## Configuration and capabilities
+
+| Setting/capability | Default | Profile | Runtime mutable | Effect |
+|---|---|---|---|---|
+| `enable_water` | Loader setting | Quality/custom | Yes | Gates enqueue/process, unchanged |
+| `max_concurrent_water_tasks` | Loader setting | Quality/custom | Yes | Bounds active tasks outside state |
+| `water_finalize_ops_per_tick` | Loader setting | Quality/custom | Yes | Bounds result pops via scheduler |
+
+## Persistence, cache and migration
+
+No state is serialized or cached. ADT source/cache formats and liquid material
+versions are unchanged; no migration or rebuild is required.
+
+## Diagnostics and observability
+
+- `total_load_count` preserves the prior pending-plus-active `water` metric.
+- Snapshot exposes FIFO tile/path, sorted active keys and result tile/path only.
+- Parsed liquid payloads, Nodes and task internals are not exposed.
+- No logs or correlation IDs are introduced.
+
+## Verification
+
+- `verify_adt_water_load_pipeline_state.gd`: invalid/dedupe/FIFO behavior,
+  active IDs, payload identity, real Thread publication, pending cancellation,
+  clear, detached diagnostics, loader boundaries and bounded timing.
+- Existing material/shutdown/facade/streaming regressions cover adjacent behavior.
+- Fidelity evidence is preservation of current scheduling/finalize transitions;
+  no MH2O/MCLQ visual or original-client parity claim is made.
+- Performance budget: 100 cycles of 256 request/task/result transitions under 1 second.
+
+## Extension points
+
+- ADT parse execution can later move behind a parser/job adapter without changing state.
+- Water scene materialization can be extracted separately because it owns Nodes
+  and main-thread effects rather than async bookkeeping.
+
+## Capability status
+
+| Capability | Status | Evidence | Gap/next step |
+|---|---|---|---|
+| ADT water load pipeline state | Implemented extraction | FIFO/task/thread/source/timing verifier | Asset-backed traversal/leak evidence pending |
+| ADT liquid parsing | Existing/loader-owned adapter | Native ADTLoader and dry-run paths | MH2O/MCLQ fidelity fixtures pending |
+| Water scene finalization | Existing/loader-owned | Scheduler/material/shutdown regressions | Separate materialization extraction possible |
+
+## Known gaps and risks
+
+- `clear` intentionally does not interrupt active WorkerThreadPool work.
+- Result payload Dictionaries are retained by reference and must not be mutated after publication.
+- Result availability and pop use separate mutex acquisitions; one serialized
+  main-thread consumer preserves ordering, but the API is not a multi-consumer queue.
+- No proprietary liquid corpus, traversal/leak run, p95/p99 or paired capture is included.
+
+## Source map
+
+| Path | Responsibility |
+|---|---|
+| `src/render/liquid/adt_water_load_pipeline_state.gd` | Request/task/result bookkeeping and result mutex |
+| `src/scenes/streaming/streaming_world_loader.gd` | Parse task execution, permits, stale checks and Node finalization |
+| `addons/mpq_extractor/loaders/adt_builder.gd` | ADT liquid geometry/material construction |
+| `src/tools/verify_adt_water_load_pipeline_state.gd` | Lifecycle/thread/boundary/timing regression |
+
+## Related decisions and references
+
+- [`world-renderer.md`](world-renderer.md)
+- [`render-budget-scheduler.md`](render-budget-scheduler.md)
+- [`../../RENDER.md`](../../RENDER.md)
+- [`../../targets/roadmap/02-rendering-and-graphics.md`](../../targets/roadmap/02-rendering-and-graphics.md)

docs/modules/world-renderer.md

@@ -7,7 +7,7 @@
 | Status | Partial |
 | Target/work package | M00 baseline; `M01-RND-STREAMING-FOCUS-001`; `M01-QAR-SERVER-SPAWN-RENDERER-001`; M03 facade/planner/scheduler/internal-access/ground/environment/entity packages; M03 terrain packages; M03 M2 packages; M03 WMO placement package |
 | Owners | Renderer workstream / milestone integrator |
-| Last verified | Worktree `work/sindo-main-codex/m03-wmo-scene-resource-cache`, 2026-07-17 |
+| Last verified | Worktree `work/sindo-main-codex/m03-adt-water-load-pipeline`, 2026-07-17 |
 | Profiles/capabilities | `Performance`, `Balanced`, `High`, `Custom`; Blizzlike fidelity incomplete |
 
 ## Purpose
@@ -138,6 +138,7 @@ from externally reading/writing loader-private queue, task, cache and tile-state
 | `WmoRenderBuildQueue` / `WmoRenderBuildJob` | Internal WMO pending-state service | Owns typed root/resource/cursor jobs and FIFO placement keys | Renderer main thread; map session | Invalid enqueue rejected; duplicate/stale behavior preserved |
 | `WmoRenderResourceCacheState` | Internal WMO cache-state service | Owns validated Resources, negative entries and pending cache paths | Renderer main thread; map/cache session | Invalid/occupied request and unknown completion are rejected |
 | `WmoSceneResourceCacheState` | Internal WMO cache-state service | Owns validated PackedScenes, negative entries and pending `.tscn` paths | Renderer main thread; map/cache session | Direct missing and terminal request transitions remain distinct |
+| `AdtWaterLoadPipelineState` | Internal liquid async-state service | Owns ADT water FIFO/dedupe, active task IDs and mutex result mailbox | Main-thread state; worker result publication | Invalid/duplicate requests rejected; clear does not interrupt workers |
 
 Публичным contract не считаются `StreamingWorldLoader` methods/properties,
 внутренние dictionaries, queues, job records и generated node names. Scene-owned
@@ -172,6 +173,7 @@ loader configuration remains transitional composition data, not a caller API.
 | Internal WMO pending build | Placement key, Node3D root, WMO Resource and cursors | Loader / `WmoRenderBuildQueue` | Loader drain and step planner adapter | Queue-owned job and strong references | Until cancel/clear/replacement |
 | Internal WMO render cache | Normalized path, cache path and validated Resource | Loader / `WmoRenderResourceCacheState` | Loader lookup, ResourceLoader poll and build queue | State-owned Resource/path references; detached request snapshots | Until transient/full clear |
 | Internal WMO scene cache | Normalized path, `.tscn` path and validated PackedScene | Loader / `WmoSceneResourceCacheState` | Loader lookup, request poll and scene instantiation | State-owned PackedScene/path references; detached request snapshots | Until transient/full clear |
+| Internal ADT water load | Tile key, ADT path, task ID and parsed Dictionary | Loader/worker / `AdtWaterLoadPipelineState` | Loader task start, budgeted drain and finalization | State-owned records; mutex result mailbox | Request through result completion/reset |
 | Output | Desired tile/detail operations | Streamer plan application | Finalize queues | Loader-owned | Cross-frame |
 | Output | Terrain/M2/WMO/liquid instances | Loader/builders | Godot world/renderer | Loader/world owner | Main-thread attach |
 | Output | Detached renderer metrics | `WorldRenderFacade` | Capture/baseline tools | Caller-owned deep copy | On demand |
@@ -356,6 +358,10 @@ sequenceDiagram
   PackedScenes, negative entries and pending `.tscn` paths. The loader retains
   ResourceLoader/FileAccess I/O, size and cache-version validation, live fallback,
   materialization, permits, validity reactions and every Node lifecycle action.
+- `AdtWaterLoadPipelineState` owns pending request order/deduplication, opaque
+  active task IDs and the worker-safe parsed-result mailbox. The loader retains
+  WorkerThreadPool start/wait, ADTLoader parsing, concurrency/finalize permits,
+  stale tile/path/root checks and ADTBuilder/Node finalization.
 - Rendered-ground query results and diagnostic snapshots are caller-owned values;
   the facade never returns Mesh, Node, tile-state or queue references.
 - Loaded-mesh ground sampling is renderer diagnostics, not authoritative gameplay
@@ -461,6 +467,9 @@ Exact exported settings and cache versions remain documented in [`../../RENDER.m
 - WMO scene Resource cache contract: direct missing sources, pending terminal
   transitions, transient/full reset, detached diagnostics, loader-owned size/
   metadata validation and bounded timing.
+- ADT water load pipeline contract: FIFO/dedupe, active task identity, real
+  worker-thread publication, pending-only cancellation, result payload/order,
+  clear/source boundaries and bounded timing.
 - Budget scheduler contract: exact lane exhaustion, shared chunk removal/create
   priority, independent lanes, frame reset, invalid limits, terminal cancellation,
   dependency boundaries and bounded permit timing without loading a world scene.
@@ -501,6 +510,7 @@ Exact exported settings and cache versions remain documented in [`../../RENDER.m
 | WMO render Resource cache state | Implemented extraction | Scene-free lifecycle/exclusivity/source/timing plus shutdown contract | ResourceLoader I/O and asset-backed traversal/leak evidence pending |
 | WMO scene Resource cache state | Implemented extraction | Scene-free lifecycle/direct-missing/source/timing plus shutdown contract | ResourceLoader/live-fallback extraction and asset-backed traversal/leak evidence pending |
 | WMO rendering | Partial | Cached group rendering | Portals/rooms/material parity |
+| ADT water load pipeline state | Implemented extraction | Scene-free FIFO/task/thread/source/timing contract | Parse/finalization and asset-backed traversal/leak evidence pending |
 | Liquids | Partial | MH2O/MLIQ paths | LiquidType/depth/shore fidelity |
 | Sky/light | Partial | DBC controller/logs | Zone transition paired validation |
 | Character presentation | Prototype | Character creator/player experiment | Full customization/equipment states |
@@ -550,6 +560,7 @@ Exact exported settings and cache versions remain documented in [`../../RENDER.m
 | `src/render/entities/world_entity_presenter.gd` | Main-thread entity visual subtree ownership and lifecycle |
 | `src/render/terrain/rendered_ground_sample.gd` | Immutable renderer-owned available/unavailable ground query result |
 | `src/render/terrain/terrain_quality_mesh_cache.gd` | Full-quality terrain revisit Mesh retention and LRU ownership |
+| `src/render/liquid/adt_water_load_pipeline_state.gd` | ADT water request/task/result bookkeeping and worker-safe mailbox |
 | `src/render/streaming/streaming_target_planner.gd` | Scene-free wanted/retained ADT target calculation |
 | `src/render/streaming/streaming_target_policy.gd` | Immutable renderer radius/prefetch policy |
 | `src/render/streaming/streaming_target_plan.gd` | Immutable planner result with read-only tile-key sets |
@@ -562,6 +573,7 @@ Exact exported settings and cache versions remain documented in [`../../RENDER.m
 | `src/tools/verify_world_entity_presentation.gd` | Entity snapshot, create/update/replace/remove and ownership regression |
 | `src/tools/verify_rendered_ground_sample.gd` | Cold-start renderer ground result invariant regression |
 | `src/tools/verify_terrain_quality_mesh_cache.gd` | Terrain cache admission, LRU, ownership and loader-boundary regression |
+| `src/tools/verify_adt_water_load_pipeline_state.gd` | ADT water FIFO/task/thread/boundary/timing regression |
 | `src/tools/verify_streaming_target_planner.gd` | Planner behavior, dependency and bounded timing regression |
 | `src/tools/verify_render_budget_scheduler.gd` | Scheduler bounds, shared-lane priority, cancellation and timing regression |
 | `src/tools/verify_renderer_internal_access.gd` | Gameplay/EditorPlugin/registered renderer-tool boundary gate derived from private streamer fields |

src/render/liquid/adt_water_load_pipeline_state.gd

@@ -0,0 +1,167 @@
+class_name AdtWaterLoadPipelineState
+extends RefCounted
+
+## Owns ADT water request, active-task and worker-result bookkeeping. The caller
+## owns WorkerThreadPool operations, ADT parsing, permits and scene finalization.
+
+var _pending_requests: Array[Dictionary] = []
+var _queued_tile_keys: Dictionary = {}
+var _active_task_ids_by_tile_key: Dictionary = {}
+var _result_mutex := Mutex.new()
+var _results: Array[Dictionary] = []
+
+
+## Appends one tile/path request in FIFO order. Invalid, queued or active tile
+## keys are rejected without mutation.
+func enqueue_request(tile_key: String, adt_path: String) -> bool:
+	if tile_key.is_empty() or adt_path.is_empty():
+		return false
+	if _queued_tile_keys.has(tile_key) or _active_task_ids_by_tile_key.has(tile_key):
+		return false
+	_pending_requests.append({
+		"key": tile_key,
+		"path": adt_path,
+	})
+	_queued_tile_keys[tile_key] = true
+	return true
+
+
+## Removes and returns the oldest request, also releasing its queued-key marker.
+## An empty Dictionary means no pending request exists.
+func take_next_request() -> Dictionary:
+	if _pending_requests.is_empty():
+		return {}
+	var request: Dictionary = _pending_requests.pop_front()
+	_queued_tile_keys.erase(String(request.get("key", "")))
+	return request
+
+
+## Removes all pending requests for one tile without affecting active work.
+## Returns the number of removed FIFO entries.
+func cancel_pending_requests(tile_key: String) -> int:
+	if tile_key.is_empty():
+		return 0
+	_queued_tile_keys.erase(tile_key)
+	var removed_count := 0
+	for request_index in range(_pending_requests.size() - 1, -1, -1):
+		var request: Dictionary = _pending_requests[request_index]
+		if String(request.get("key", "")) != tile_key:
+			continue
+		_pending_requests.remove_at(request_index)
+		removed_count += 1
+	return removed_count
+
+
+## Records one caller-started worker task. The opaque task ID is retained until
+## the caller waits for and completes it.
+func remember_active_task(tile_key: String, task_id: int) -> bool:
+	if tile_key.is_empty() or _active_task_ids_by_tile_key.has(tile_key):
+		return false
+	_active_task_ids_by_tile_key[tile_key] = task_id
+	return true
+
+
+## Returns whether a worker task is active for the tile.
+func has_active_task(tile_key: String) -> bool:
+	return not tile_key.is_empty() and _active_task_ids_by_tile_key.has(tile_key)
+
+
+## Returns the opaque active task ID, or -1 for an unknown tile.
+func active_task_id_for(tile_key: String) -> int:
+	return int(_active_task_ids_by_tile_key.get(tile_key, -1))
+
+
+## Removes and returns one completed task ID, or -1 for an unknown tile.
+func complete_active_task(tile_key: String) -> int:
+	if tile_key.is_empty() or not _active_task_ids_by_tile_key.has(tile_key):
+		return -1
+	var task_id := int(_active_task_ids_by_tile_key[tile_key])
+	_active_task_ids_by_tile_key.erase(tile_key)
+	return task_id
+
+
+## Returns a detached snapshot of opaque task IDs for orderly shutdown waiting.
+func active_task_ids_snapshot() -> Array:
+	return _active_task_ids_by_tile_key.values().duplicate()
+
+
+## Publishes one parsed ADT result from a worker thread. Payload identity is
+## retained; the caller must not mutate it after publication.
+func publish_result(tile_key: String, adt_path: String, water_data: Dictionary) -> bool:
+	if tile_key.is_empty() or adt_path.is_empty():
+		return false
+	_result_mutex.lock()
+	_results.append({
+		"key": tile_key,
+		"path": adt_path,
+		"data": water_data,
+	})
+	_result_mutex.unlock()
+	return true
+
+
+## Returns whether a worker result is currently ready.
+func has_result() -> bool:
+	_result_mutex.lock()
+	var result_available := not _results.is_empty()
+	_result_mutex.unlock()
+	return result_available
+
+
+## Removes and returns the oldest worker result. An empty Dictionary means no
+## result was ready. Intended for the serialized renderer main thread.
+func pop_result() -> Dictionary:
+	_result_mutex.lock()
+	var result: Dictionary = {}
+	if not _results.is_empty():
+		result = _results.pop_front()
+	_result_mutex.unlock()
+	return result
+
+
+## Returns pending plus active load count for existing renderer metrics.
+func total_load_count() -> int:
+	return _pending_requests.size() + _active_task_ids_by_tile_key.size()
+
+
+## Returns the pending request count.
+func pending_request_count() -> int:
+	return _pending_requests.size()
+
+
+## Returns the active worker-task count.
+func active_task_count() -> int:
+	return _active_task_ids_by_tile_key.size()
+
+
+## Clears bookkeeping and queued results without interrupting worker tasks.
+## Orderly shutdown must wait the task-ID snapshot before calling this method.
+func clear() -> void:
+	_pending_requests.clear()
+	_queued_tile_keys.clear()
+	_active_task_ids_by_tile_key.clear()
+	_result_mutex.lock()
+	_results.clear()
+	_result_mutex.unlock()
+
+
+## Returns detached FIFO/task/result scalar diagnostics without water payloads.
+func diagnostic_snapshot() -> Dictionary:
+	var pending_requests: Array[Dictionary] = []
+	for request in _pending_requests:
+		pending_requests.append(request.duplicate())
+	var active_tile_keys: Array = _active_task_ids_by_tile_key.keys()
+	active_tile_keys.sort()
+	_result_mutex.lock()
+	var result_records: Array[Dictionary] = []
+	for result in _results:
+		result_records.append({
+			"key": String(result.get("key", "")),
+			"path": String(result.get("path", "")),
+		})
+	_result_mutex.unlock()
+	return {
+		"pending_requests": pending_requests,
+		"active_tile_keys": active_tile_keys.duplicate(),
+		"result_records": result_records,
+	}

src/render/liquid/adt_water_load_pipeline_state.gd.uid

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

src/scenes/streaming/streaming_world_loader.gd

@@ -42,6 +42,9 @@ const TERRAIN_CHUNK_LOD_PLANNER_SCRIPT := preload("res://src/render/terrain/terr
 const TERRAIN_CHUNK_GEOMETRY_QUEUE_PLANNER_SCRIPT := preload(
 	"res://src/render/terrain/terrain_chunk_geometry_queue_planner.gd"
 )
+const ADT_WATER_LOAD_PIPELINE_STATE_SCRIPT := preload(
+	"res://src/render/liquid/adt_water_load_pipeline_state.gd"
+)
 const M2_UNIQUE_PLACEMENT_REGISTRY_SCRIPT := preload(
 	"res://src/render/m2/m2_unique_placement_registry.gd"
 )
@@ -256,11 +259,7 @@ var _terrain_splat_cache_tasks: Dictionary = {}
 var _terrain_splat_tasks: Dictionary = {}
 var _terrain_splat_result_mutex := Mutex.new()
 var _terrain_splat_result_queue: Array = []
-var _water_load_queue: Array = []
+var _adt_water_load_pipeline_state := ADT_WATER_LOAD_PIPELINE_STATE_SCRIPT.new()
-var _water_load_queued: Dictionary = {}
-var _water_load_tasks: Dictionary = {}
-var _water_result_mutex := Mutex.new()
-var _water_result_queue: Array = []
 var _tile_result_mutex := Mutex.new()
 var _tile_result_queue: Array = []
 var _shared_tex_cache: Dictionary = {}
@@ -999,7 +998,7 @@ func _log_hitch_profile(start_usec: int, timings: Array[String], did_refresh: bo
 		_terrain_upgrade_tasks.size(),
 		_terrain_control_splat_cache_tasks.size(),
 		_terrain_splat_queue_size(),
-		_water_load_queue.size() + _water_load_tasks.size(),
+		_adt_water_load_pipeline_state.total_load_count(),
 		_detail_asset_queue.size(),
 		_m2_group_tasks.size(),
 		_m2_animation_finalize_queue.size() + _m2_animation_load_requests.size(),
@@ -1030,7 +1029,7 @@ func render_baseline_snapshot() -> Dictionary:
 			"terrain_upgrade": _terrain_upgrade_tasks.size(),
 			"terrain_control": _terrain_control_splat_cache_tasks.size(),
 			"terrain_splat": _terrain_splat_queue_size(),
-			"water": _water_load_queue.size() + _water_load_tasks.size(),
+			"water": _adt_water_load_pipeline_state.total_load_count(),
 			"detail": _detail_asset_queue.size(),
 			"m2_task": _m2_group_tasks.size(),
 			"m2_animation": _m2_animation_finalize_queue.size() + _m2_animation_load_requests.size(),
@@ -1305,7 +1304,7 @@ func _tick_runtime_stats(delta: float) -> void:
 		_terrain_upgrade_tasks.size(),
 		_terrain_control_splat_cache_tasks.size(),
 		_terrain_splat_queue_size(),
-		_water_load_queue.size() + _water_load_tasks.size(),
+		_adt_water_load_pipeline_state.total_load_count(),
 		terrain_baked_full,
 		terrain_control_splat,
 		terrain_splat,
@@ -2230,24 +2229,20 @@ func _request_tile_water_load(state: Dictionary) -> void:
 	var path := String(state.get("path", ""))
 	if key.is_empty() or path.is_empty():
 		return
-	if _water_load_tasks.has(key) or _water_load_queued.has(key):
+	_adt_water_load_pipeline_state.enqueue_request(key, path)
-		return
-	_water_load_queue.append({
-		"key": key,
-		"path": path,
-	})
-	_water_load_queued[key] = true
 
 
 func _process_water_load_queue() -> void:
 	if not enable_water:
 		return
 	var limit := maxi(1, max_concurrent_water_tasks)
-	while _water_load_tasks.size() < limit and not _water_load_queue.is_empty():
+	while (
-		var request: Dictionary = _water_load_queue.pop_front()
+			_adt_water_load_pipeline_state.active_task_count() < limit
+			and _adt_water_load_pipeline_state.pending_request_count() > 0
+	):
+		var request: Dictionary = _adt_water_load_pipeline_state.take_next_request()
 		var key := String(request.get("key", ""))
-		_water_load_queued.erase(key)
+		if key.is_empty() or _adt_water_load_pipeline_state.has_active_task(key):
-		if key.is_empty() or _water_load_tasks.has(key):
 			continue
 		if not _tile_states.has(key):
 			continue
@@ -2255,7 +2250,7 @@ func _process_water_load_queue() -> void:
 		if path.is_empty():
 			continue
 		var task_id: int = WorkerThreadPool.add_task(_load_tile_water_task.bind(key, path))
-		_water_load_tasks[key] = task_id
+		_adt_water_load_pipeline_state.remember_active_task(key, task_id)
 
 
 func _load_tile_water_task(key: String, adt_path: String) -> void:
@@ -2264,13 +2259,7 @@ func _load_tile_water_task(key: String, adt_path: String) -> void:
 	if loader:
 		data = loader.call("load_adt", adt_path)
 
-	_water_result_mutex.lock()
+	_adt_water_load_pipeline_state.publish_result(key, adt_path, data)
-	_water_result_queue.append({
-		"key": key,
-		"path": adt_path,
-		"data": data,
-	})
-	_water_result_mutex.unlock()
 
 
 func _drain_water_load_results() -> void:
@@ -2278,20 +2267,20 @@ func _drain_water_load_results() -> void:
 	if not _render_budget_scheduler.has_remaining_permit(RENDER_BUDGET_SCHEDULER_SCRIPT.WATER_FINALIZE):
 		return
 
-	_water_result_mutex.lock()
+	while _adt_water_load_pipeline_state.has_result() and _render_budget_scheduler.try_consume_permit(
-	while not _water_result_queue.is_empty() and _render_budget_scheduler.try_consume_permit(
 			RENDER_BUDGET_SCHEDULER_SCRIPT.WATER_FINALIZE):
-		results.append(_water_result_queue.pop_front())
+		results.append(_adt_water_load_pipeline_state.pop_result())
-	_water_result_mutex.unlock()
 
 	for result in results:
 		var key := String(result.get("key", ""))
 		if key.is_empty():
 			continue
-		if not _water_load_tasks.has(key):
+		if not _adt_water_load_pipeline_state.has_active_task(key):
 			continue
-		WorkerThreadPool.wait_for_task_completion(int(_water_load_tasks[key]))
+		WorkerThreadPool.wait_for_task_completion(
-		_water_load_tasks.erase(key)
+			_adt_water_load_pipeline_state.active_task_id_for(key)
+		)
+		_adt_water_load_pipeline_state.complete_active_task(key)
 		if _shutting_down or not _tile_states.has(key):
 			continue
 
@@ -2419,14 +2408,9 @@ func _wait_for_tile_tasks() -> void:
 	_terrain_splat_result_queue.clear()
 	_terrain_splat_result_mutex.unlock()
 
-	for task_id in _water_load_tasks.values():
+	for task_id in _adt_water_load_pipeline_state.active_task_ids_snapshot():
 		WorkerThreadPool.wait_for_task_completion(int(task_id))
-	_water_load_tasks.clear()
+	_adt_water_load_pipeline_state.clear()
-	_water_load_queue.clear()
-	_water_load_queued.clear()
-	_water_result_mutex.lock()
-	_water_result_queue.clear()
-	_water_result_mutex.unlock()
 
 
 func _should_accept_loaded_tile(request: Dictionary) -> bool:
@@ -2961,11 +2945,7 @@ func _release_tile(key: String) -> void:
 	_terrain_control_splat_cache_tasks.erase(key)
 	_terrain_splat_cache_tasks.erase(key)
 	_terrain_splat_tasks.erase(key)
-	_water_load_queued.erase(key)
+	_adt_water_load_pipeline_state.cancel_pending_requests(key)
-	for i in range(_water_load_queue.size() - 1, -1, -1):
-		var pending: Dictionary = _water_load_queue[i]
-		if String(pending.get("key", "")) == key:
-			_water_load_queue.remove_at(i)
 	var tile_lod_rid: RID = state.get("tile_lod_rid", RID())
 	if tile_lod_rid.is_valid():
 		_free_render_instance(tile_lod_rid)
@@ -2996,12 +2976,7 @@ func _clear_streamed_world() -> void:
 	_terrain_splat_result_mutex.lock()
 	_terrain_splat_result_queue.clear()
 	_terrain_splat_result_mutex.unlock()
-	_water_load_queue.clear()
+	_adt_water_load_pipeline_state.clear()
-	_water_load_queued.clear()
-	_water_load_tasks.clear()
-	_water_result_mutex.lock()
-	_water_result_queue.clear()
-	_water_result_mutex.unlock()
 	_detail_asset_queue.clear()
 	_detail_asset_queued.clear()
 	for key in _m2_build_jobs.keys():

src/tools/verify_adt_water_load_pipeline_state.gd

@@ -0,0 +1,351 @@
+extends SceneTree
+
+## Asset-free lifecycle, threading, boundary and timing regression for ADT water loads.
+
+const PIPELINE_SCRIPT := preload(
+	"res://src/render/liquid/adt_water_load_pipeline_state.gd"
+)
+const PIPELINE_PATH := "res://src/render/liquid/adt_water_load_pipeline_state.gd"
+const LOADER_PATH := "res://src/scenes/streaming/streaming_world_loader.gd"
+
+
+func _initialize() -> void:
+	var failures: Array[String] = []
+	_verify_invalid_dedupe_and_fifo(failures)
+	_verify_active_task_lifecycle(failures)
+	_verify_result_mailbox_and_payload_identity(failures)
+	_verify_worker_publication(failures)
+	_verify_pending_cancellation(failures)
+	_verify_clear_and_detached_diagnostics(failures)
+	_verify_ownership_boundaries(failures)
+	var elapsed_milliseconds := _verify_bounded_timing(failures)
+	if not failures.is_empty():
+		for failure in failures:
+			push_error("ADT_WATER_LOAD_PIPELINE_STATE: %s" % failure)
+		quit(1)
+		return
+	print(
+		"ADT_WATER_LOAD_PIPELINE_STATE PASS cases=12 iterations=100 elapsed_ms=%.3f"
+		% elapsed_milliseconds
+	)
+	quit(0)
+
+
+func _verify_invalid_dedupe_and_fifo(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	_expect_false(
+		pipeline.call("enqueue_request", "", "res://tile.adt"),
+		"empty tile key rejected",
+		failures
+	)
+	_expect_false(
+		pipeline.call("enqueue_request", "31:49", ""),
+		"empty path rejected",
+		failures
+	)
+	_expect_true(
+		pipeline.call("enqueue_request", "31:49", "res://a.adt"),
+		"first request accepted",
+		failures
+	)
+	_expect_false(
+		pipeline.call("enqueue_request", "31:49", "res://duplicate.adt"),
+		"queued duplicate rejected",
+		failures
+	)
+	pipeline.call("enqueue_request", "32:49", "res://b.adt")
+	_expect_equal(int(pipeline.call("pending_request_count")), 2, "pending count", failures)
+	var first_request: Dictionary = pipeline.call("take_next_request")
+	var second_request: Dictionary = pipeline.call("take_next_request")
+	_expect_string_equal(String(first_request["key"]), "31:49", "FIFO first key", failures)
+	_expect_string_equal(String(second_request["key"]), "32:49", "FIFO second key", failures)
+	_expect_true(
+		pipeline.call("enqueue_request", "31:49", "res://retry.adt"),
+		"dequeued key can be requested again",
+		failures
+	)
+
+
+func _verify_active_task_lifecycle(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	_expect_false(
+		pipeline.call("remember_active_task", "", 7),
+		"empty active key rejected",
+		failures
+	)
+	_expect_true(
+		pipeline.call("remember_active_task", "31:49", 101),
+		"active task accepted",
+		failures
+	)
+	_expect_false(
+		pipeline.call("remember_active_task", "31:49", 202),
+		"duplicate active task rejected",
+		failures
+	)
+	_expect_true(
+		pipeline.call("has_active_task", "31:49"),
+		"active task observable",
+		failures
+	)
+	_expect_equal(
+		int(pipeline.call("active_task_id_for", "31:49")),
+		101,
+		"opaque task ID retained",
+		failures
+	)
+	_expect_false(
+		pipeline.call("enqueue_request", "31:49", "res://active.adt"),
+		"active tile rejects enqueue",
+		failures
+	)
+	_expect_equal(
+		int(pipeline.call("complete_active_task", "31:49")),
+		101,
+		"completion returns task ID",
+		failures
+	)
+	_expect_equal(
+		int(pipeline.call("complete_active_task", "31:49")),
+		-1,
+		"unknown completion rejected",
+		failures
+	)
+
+
+func _verify_result_mailbox_and_payload_identity(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	var water_data := {"chunks": [{"liquid": true}]}
+	_expect_false(
+		pipeline.call("publish_result", "", "res://a.adt", water_data),
+		"empty result key rejected",
+		failures
+	)
+	_expect_true(
+		pipeline.call("publish_result", "31:49", "res://a.adt", water_data),
+		"valid result published",
+		failures
+	)
+	_expect_true(pipeline.call("has_result"), "published result observable", failures)
+	var result: Dictionary = pipeline.call("pop_result")
+	_expect_string_equal(String(result["key"]), "31:49", "result key", failures)
+	_expect_string_equal(String(result["path"]), "res://a.adt", "result path", failures)
+	_expect_true(result["data"] == water_data, "water payload identity retained", failures)
+	_expect_false(pipeline.call("has_result"), "result pop empties mailbox", failures)
+	_expect_true(
+		(pipeline.call("pop_result") as Dictionary).is_empty(),
+		"empty result pop is safe",
+		failures
+	)
+
+
+func _verify_worker_publication(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	var worker := Thread.new()
+	var start_error := worker.start(
+		Callable(pipeline, "publish_result").bind(
+			"worker",
+			"res://worker.adt",
+			{"worker": true}
+		)
+	)
+	_expect_equal(int(start_error), OK, "worker publication starts", failures)
+	if start_error == OK:
+		worker.wait_to_finish()
+	_expect_true(pipeline.call("has_result"), "worker result safely visible", failures)
+	var result: Dictionary = pipeline.call("pop_result")
+	_expect_string_equal(String(result.get("key", "")), "worker", "worker result key", failures)
+
+
+func _verify_pending_cancellation(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	pipeline.call("enqueue_request", "31:49", "res://a.adt")
+	pipeline.call("enqueue_request", "32:49", "res://b.adt")
+	pipeline.call("remember_active_task", "31:49-active", 12)
+	_expect_equal(
+		int(pipeline.call("cancel_pending_requests", "31:49")),
+		1,
+		"tile cancellation removes pending request",
+		failures
+	)
+	_expect_equal(int(pipeline.call("pending_request_count")), 1, "other pending retained", failures)
+	_expect_true(
+		pipeline.call("has_active_task", "31:49-active"),
+		"pending cancellation leaves active work",
+		failures
+	)
+	_expect_equal(
+		int(pipeline.call("cancel_pending_requests", "unknown")),
+		0,
+		"unknown cancellation unchanged",
+		failures
+	)
+
+
+func _verify_clear_and_detached_diagnostics(failures: Array[String]) -> void:
+	var pipeline: RefCounted = PIPELINE_SCRIPT.new()
+	pipeline.call("enqueue_request", "32:49", "res://b.adt")
+	pipeline.call("enqueue_request", "31:49", "res://a.adt")
+	pipeline.call("remember_active_task", "z", 2)
+	pipeline.call("remember_active_task", "a", 1)
+	pipeline.call("publish_result", "result", "res://result.adt", {"secret": "payload"})
+	var snapshot: Dictionary = pipeline.call("diagnostic_snapshot")
+	_expect_string_array(
+		snapshot["active_tile_keys"],
+		["a", "z"],
+		"active diagnostics sorted",
+		failures
+	)
+	_expect_equal(
+		(snapshot["pending_requests"] as Array).size(),
+		2,
+		"pending diagnostics count",
+		failures
+	)
+	_expect_false(
+		(snapshot["result_records"][0] as Dictionary).has("data"),
+		"diagnostics omit water payload",
+		failures
+	)
+	(snapshot["pending_requests"] as Array).clear()
+	var fresh_snapshot: Dictionary = pipeline.call("diagnostic_snapshot")
+	_expect_equal(
+		(fresh_snapshot["pending_requests"] as Array).size(),
+		2,
+		"diagnostics detached",
+		failures
+	)
+	pipeline.call("clear")
+	pipeline.call("clear")
+	_expect_equal(int(pipeline.call("total_load_count")), 0, "clear load count", failures)
+	_expect_false(pipeline.call("has_result"), "clear result mailbox", failures)
+
+
+func _verify_ownership_boundaries(failures: Array[String]) -> void:
+	var loader_source := _read_text(LOADER_PATH, failures)
+	var pipeline_source := _read_text(PIPELINE_PATH, failures)
+	_expect_true(
+		loader_source.contains("ADT_WATER_LOAD_PIPELINE_STATE_SCRIPT.new()"),
+		"loader composes pipeline state",
+		failures
+	)
+	for legacy_field in [
+		"var _water_load_queue:",
+		"var _water_load_queued:",
+		"var _water_load_tasks:",
+		"var _water_result_mutex",
+		"var _water_result_queue:",
+	]:
+		_expect_false(loader_source.contains(legacy_field), "loader removes %s" % legacy_field, failures)
+	for retained_loader_rule in [
+		"ClassDB.instantiate(\"ADTLoader\")",
+		"WorkerThreadPool.add_task(_load_tile_water_task.bind(key, path))",
+		"RENDER_BUDGET_SCHEDULER_SCRIPT.WATER_FINALIZE",
+		"_builder.build_tile_water_scene(data, origin)",
+		"tile_root.add_child(water_root)",
+		"state[\"water_loaded\"] = true",
+	]:
+		_expect_true(loader_source.contains(retained_loader_rule), "loader retains %s" % retained_loader_rule, failures)
+	for forbidden_token in [
+		"WorkerThreadPool.",
+		"ADTLoader",
+		"RenderBudgetScheduler",
+		"Node3D",
+		"Mesh",
+		"RID",
+		"_tile_states",
+	]:
+		_expect_false(
+			pipeline_source.contains(forbidden_token),
+			"pipeline omits %s ownership" % forbidden_token,
+			failures
+		)
+	_expect_equal(
+		_count_occurrences(loader_source, "_adt_water_load_pipeline_state.clear()"),
+		2,
+		"loader retains two clear sites",
+		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 tile_index in range(256):
+			var tile_key := "tile:%d" % tile_index
+			pipeline.call("enqueue_request", tile_key, "res://tile_%d.adt" % tile_index)
+		while int(pipeline.call("pending_request_count")) > 0:
+			var request: Dictionary = pipeline.call("take_next_request")
+			var tile_key := String(request["key"])
+			pipeline.call("remember_active_task", tile_key, tile_key.hash())
+			pipeline.call("publish_result", tile_key, String(request["path"]), {})
+		while pipeline.call("has_result"):
+			var result: Dictionary = pipeline.call("pop_result")
+			pipeline.call("complete_active_task", String(result["key"]))
+		pipeline.call("clear")
+		_expect_equal(
+			int(pipeline.call("total_load_count")),
+			0,
+			"timing cycle %d clears" % iteration,
+			failures
+		)
+	var elapsed_milliseconds := float(Time.get_ticks_usec() - started_microseconds) / 1000.0
+	_expect_true(
+		elapsed_milliseconds < 1000.0,
+		"100 request/task/result cycles remain bounded",
+		failures
+	)
+	return elapsed_milliseconds
+
+
+func _read_text(path: String, failures: Array[String]) -> String:
+	var file := FileAccess.open(path, FileAccess.READ)
+	if file == null:
+		failures.append("cannot open %s" % path)
+		return ""
+	return file.get_as_text()
+
+
+func _count_occurrences(text: String, token: String) -> int:
+	return text.split(token).size() - 1
+
+
+func _expect_equal(
+		actual_value: int,
+		expected_value: int,
+		label: String,
+		failures: Array[String]
+) -> void:
+	if actual_value != expected_value:
+		failures.append("%s expected %d, got %d" % [label, expected_value, actual_value])
+
+
+func _expect_string_equal(
+		actual_value: String,
+		expected_value: String,
+		label: String,
+		failures: Array[String]
+) -> void:
+	if actual_value != expected_value:
+		failures.append("%s expected %s, got %s" % [label, expected_value, actual_value])
+
+
+func _expect_string_array(
+		actual_values: Array,
+		expected_values: Array,
+		label: String,
+		failures: Array[String]
+) -> void:
+	if actual_values != expected_values:
+		failures.append("%s expected %s, got %s" % [label, expected_values, actual_values])
+
+
+func _expect_true(actual_value: bool, label: String, failures: Array[String]) -> void:
+	if not actual_value:
+		failures.append("%s expected true" % label)
+
+
+func _expect_false(actual_value: bool, label: String, failures: Array[String]) -> void:
+	if actual_value:
+		failures.append("%s expected false" % label)

src/tools/verify_adt_water_load_pipeline_state.gd.uid

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