using System;
using System.Collections.Generic;
using System.Diagnostics;
using AcDream.App.Rendering.Wb;
using AcDream.Core.Terrain;
using AcDream.Core.World;

namespace AcDream.App.Streaming;

/// <summary>
/// Called once per frame from <c>GameWindow.OnUpdate</c>. Owns the
/// <see cref="StreamingRegion"/> and uses delegates into
/// <see cref="LandblockStreamer"/> so tests can inject fakes. All work
/// happens on the render thread; the streamer itself is background.
///
/// <remarks>
/// Threading: not thread-safe. All calls must happen on the render thread.
/// </remarks>
/// </summary>
public sealed class StreamingController
    : IStreamingFrameBackend,
      IWorldRevealStreamingScheduler
{
    private readonly record struct DestinationReservation(
        long RevealGeneration,
        uint LandblockId,
        int Radius);

    private sealed class OriginRecenterRetirement
    {
        public bool RadiiConverged;
        public bool GenerationAdvanced;
        public bool PendingLoadsCleared;
        public bool CompletionQueueCleared;
        public bool RegionCleared;
        public bool SpatialGenerationDetached;
        public bool PreparationCommitted;
        public (int X, int Y, bool IsSealedDungeon)? Destination;
        public bool DestinationConfigured;
        public bool DestinationLoadEnqueued;
    }

    private sealed class FullWindowRetirement
    {
        public bool GenerationAdvanced;
        public bool PendingLoadsCleared;
        public bool CompletionQueueCleared;
        public bool RegionCleared;
        public List<uint>? ResidentIds;
        public IEnumerator<uint>? ResidentEnumerator;
        public int RetirementCursor;
        public bool PreparationCommitted;
    }

    private readonly Action<uint, LandblockStreamJobKind, ulong> _enqueueLoad;
    private readonly Action<uint, ulong> _enqueueUnload;
    private readonly ILandblockCompletionSource _completionSource;
    private readonly Action? _clearPendingLoads;
    private readonly LandblockPresentationPipeline _presentation;
    private readonly Func<LandblockStreamResult, bool>
        _isPublicationBlockedByRetirement;
    private readonly StreamingWorkBudgetOptions _configuredWorkBudgetOptions;
    private StreamingWorkBudget _workBudget;
    private StreamingWorkMeter? _activeWorkMeter;
    private StreamingWorkMeterSnapshot _lastWorkMeter;
    private readonly StreamingCompletionQueue _completionQueue = new();
    private long _nextCompletionSequence;
    private int _legacyCompletionProfile = 4;
    private DestinationReservation? _destinationReservation;
    private long _lifetimeWorkOverruns;
    private long _lifetimeOversizedProgress;
    private double _maximumWorkFrameMilliseconds;
    private string? _maximumWorkFrameStage;
    private double _maximumWorkOperationMilliseconds;
    private string? _maximumWorkOperationStage;
    private readonly GpuWorldState _state;
    private StreamingRegion? _region;
    private RadiiReconfiguration? _pendingRadiiReconfiguration;
    private bool _advancingRadiiReconfiguration;
    private (int NearRadius, int FarRadius)? _deferredRadiiRequest;
    private OriginRecenterRetirement? _originRecenterRetirement;
    private FullWindowRetirement? _fullWindowRetirement;
    private bool _advancingOriginRecenter;
    // Hard streaming boundaries advance this token. Worker completions carry
    // the generation captured at enqueue time, so an old overlapping load or
    // unload cannot mutate the replacement window after a portal recenter.
    private ulong _generation;

    // True while streaming is collapsed to the single dungeon landblock the
    // player stands in (the dungeon gate, #133 FPS). AC dungeons have NO
    // adjacent landblocks — neighbors are unrelated ocean-grid dungeons that
    // are never visible, so we stop loading the 25×25 window entirely.
    private bool _collapsed;

    // The dungeon landblock id we collapsed onto. Once collapsed we key the
    // gate on this STABLE landblock, not the per-frame insideDungeon signal:
    // CurrCell can momentarily resolve to null/outdoor mid-frame, and gating
    // expand on that flicker thrashes collapse↔expand (reload storms + a light
    // leak). We only expand when the observer actually moves to a different
    // landblock (teleport/portal out).
    private uint _collapsedCenter;

    /// <summary>
    /// Near-tier radius (LBs from observer that load full detail: terrain +
    /// scenery + entities). Runtime changes go through
    /// <see cref="ReconfigureRadii"/>.
    /// </summary>
    public int NearRadius { get; private set; }

    /// <summary>
    /// Far-tier radius (LBs from observer that load terrain only).
    /// </summary>
    public int FarRadius { get; private set; }

    /// <summary>
    /// Compatibility quality selector for callers that still express the
    /// streaming profile as a completion count. Four preserves the configured
    /// High profile; other positive values scale every typed time/count/byte
    /// dimension together. It is not an independent execution throttle.
    /// </summary>
    public int MaxCompletionsPerFrame
    {
        get => _legacyCompletionProfile;
        set
        {
            StreamingWorkBudgetOptions profile =
                _configuredWorkBudgetOptions
                    .ScaleForLegacyCompletionCount(value);
            _legacyCompletionProfile = value;
            _workBudget = profile.ToBudget();
        }
    }

    internal long ActiveRevealGeneration =>
        _destinationReservation?.RevealGeneration ?? 0L;

    internal uint DestinationLandblockId =>
        _destinationReservation?.LandblockId ?? 0u;

    internal int DestinationRadius =>
        _destinationReservation?.Radius ?? 0;

    internal void BeginDestinationReservation(
        long revealGeneration,
        uint destinationCell,
        int requiredRenderRadius)
    {
        if (revealGeneration <= 0)
            throw new ArgumentOutOfRangeException(nameof(revealGeneration));
        if (destinationCell == 0u)
            throw new ArgumentOutOfRangeException(nameof(destinationCell));
        if (requiredRenderRadius < 0)
            throw new ArgumentOutOfRangeException(nameof(requiredRenderRadius));

        _destinationReservation = new DestinationReservation(
            revealGeneration,
            (destinationCell & 0xFFFF0000u) | 0xFFFFu,
            requiredRenderRadius);
    }

    internal void EndDestinationReservation(long revealGeneration)
    {
        if (_destinationReservation is { } active
            && active.RevealGeneration == revealGeneration)
        {
            _destinationReservation = null;
        }
    }

    void IWorldRevealStreamingScheduler.BeginDestinationReservation(
        long revealGeneration,
        uint destinationCell,
        int requiredRenderRadius) =>
        BeginDestinationReservation(
            revealGeneration,
            destinationCell,
            requiredRenderRadius);

    void IWorldRevealStreamingScheduler.EndDestinationReservation(
        long revealGeneration) =>
        EndDestinationReservation(revealGeneration);

    // [FRAME-DIAG] (read by GameWindow's ACDREAM_WB_DIAG rollup): the standing
    // accepted completion backlog. A non-zero value during/after a teleport is
    // the typed-budget publication tail. Removable compatibility probe surface.
    public int DeferredApplyBacklog => _completionQueue.Count;
    public int PendingRetirementCount => _presentation.PendingRetirementCount;
    public StreamingWorkDiagnostics WorkDiagnostics
    {
        get
        {
            StreamingCompletionQueueSnapshot queued =
                _completionQueue.CaptureSnapshot();
            return new StreamingWorkDiagnostics(
                _lastWorkMeter,
                _lifetimeWorkOverruns,
                _lifetimeOversizedProgress,
                _maximumWorkFrameMilliseconds,
                _maximumWorkFrameStage,
                _maximumWorkOperationMilliseconds,
                _maximumWorkOperationStage,
                queued.Count,
                queued.RetainedCpuBytes,
                queued.OldestAgeMilliseconds,
                _presentation.PendingPublicationCount,
                _presentation.PendingRetirementCount,
                _completionSource.BacklogCount,
                queued.Destination,
                queued.Control,
                queued.Unload,
                queued.Near,
                queued.Far);
        }
    }
    internal bool IsCollapsedToDungeon => _collapsed;

    /// <summary>
    /// True once every in-bounds landblock in the requested Chebyshev ring has
    /// crossed the render-thread publication barrier. Worker completion and
    /// world-state registration are not sufficient: all static GfxObj and
    /// EnvCell shell meshes must have completed their render-thread upload.
    /// Portal-space exit uses this alongside physics residency so the world
    /// cannot be revealed while its render slots are still absent.
    /// </summary>
    public bool IsRenderNeighborhoodResident(uint cellOrLandblockId, int radius)
    {
        if (radius < 0)
            throw new ArgumentOutOfRangeException(nameof(radius));

        // LandDefs::InboundValidCellId validates both map axes and the low-word
        // class (outdoor cell, EnvCell, or canonical landblock sentinel).
        if (!AcDream.Core.Physics.LandDefs.InboundValidCellId(cellOrLandblockId))
            return false;
        int cx = (int)((cellOrLandblockId >> 24) & 0xFFu);
        int cy = (int)((cellOrLandblockId >> 16) & 0xFFu);
        for (int dx = -radius; dx <= radius; dx++)
        for (int dy = -radius; dy <= radius; dy++)
        {
            int nx = cx + dx;
            int ny = cy + dy;
            // Match PhysicsEngine.IsNeighborhoodTerrainResident: the outer
            // 0xFF coordinate has no loadable neighbour beyond it.
            if (nx < 0 || nx > 254 || ny < 0 || ny > 254)
                continue;

            uint canonical = ((uint)nx << 24) | ((uint)ny << 16) | 0xFFFFu;
            if (!_state.IsNearTier(canonical) || !_state.IsRenderReady(canonical))
                return false;
        }

        return true;
    }

    // [FRAME-DIAG]: full-window presentation retirements include explicit
    // reloads and shared-origin recenter barriers. Both retire and later
    // re-upload the complete resident window.
    public int FullWindowRetirementCount { get; private set; }
    public int LastFullWindowRetirementLandblockCount { get; private set; }

    /// <summary>
    /// Internal compatibility seam for hermetic controller policy tests. Live
    /// composition cannot supply presentation callbacks; it must use the
    /// concrete pipeline constructor below.
    /// </summary>
    internal StreamingController(
        Action<uint, LandblockStreamJobKind, ulong> enqueueLoad,
        Action<uint, ulong> enqueueUnload,
        Func<int, IReadOnlyList<LandblockStreamResult>> drainCompletions,
        Action<LandblockBuild, LandblockMeshData> applyTerrain,
        GpuWorldState state,
        int nearRadius,
        int farRadius,
        Action<uint>? removeTerrain = null,
        Action<uint>? demoteNearLayer = null,
        Action? clearPendingLoads = null,
        Action<uint>? onLandblockLoaded = null,
        Action<EnvCellLandblockBuild>? ensureEnvCellMeshes = null,
        LandblockRetirementCoordinator? retirementCoordinator = null,
        StreamingWorkBudgetOptions? workBudgetOptions = null)
        : this(
            enqueueLoad,
            enqueueUnload,
            drainCompletions,
            state,
            nearRadius,
            farRadius,
            new LandblockPresentationPipeline(
                applyTerrain,
                state,
                onLandblockLoaded,
                ensureEnvCellMeshes,
                retirementCoordinator,
                removeTerrain,
                demoteNearLayer),
            clearPendingLoads,
            workBudgetOptions)
    {
    }

    /// <summary>
    /// Concrete presentation path. Its pipeline is the sole presentation and
    /// retirement owner, so no legacy publication delegate can be supplied or
    /// silently ignored.
    /// </summary>
    internal StreamingController(
        Action<uint, LandblockStreamJobKind, ulong> enqueueLoad,
        Action<uint, ulong> enqueueUnload,
        Func<int, IReadOnlyList<LandblockStreamResult>> drainCompletions,
        GpuWorldState state,
        int nearRadius,
        int farRadius,
        LandblockPresentationPipeline presentationPipeline,
        Action? clearPendingLoads = null,
        StreamingWorkBudgetOptions? workBudgetOptions = null)
        : this(
            enqueueLoad,
            enqueueUnload,
            new DelegateLandblockCompletionSource(drainCompletions),
            state,
            nearRadius,
            farRadius,
            presentationPipeline,
            clearPendingLoads,
            workBudgetOptions)
    {
    }

    /// <summary>
    /// Allocation-free production completion path. Peek/read is single-consumer
    /// and lets the scheduler price a result before adopting it.
    /// </summary>
    public StreamingController(
        Action<uint, LandblockStreamJobKind, ulong> enqueueLoad,
        Action<uint, ulong> enqueueUnload,
        ILandblockCompletionSource completionSource,
        GpuWorldState state,
        int nearRadius,
        int farRadius,
        LandblockPresentationPipeline presentationPipeline,
        Action? clearPendingLoads = null,
        StreamingWorkBudgetOptions? workBudgetOptions = null)
    {
        _enqueueLoad = enqueueLoad;
        _enqueueUnload = enqueueUnload;
        _completionSource = completionSource
            ?? throw new ArgumentNullException(nameof(completionSource));
        _clearPendingLoads = clearPendingLoads;
        _state = state;
        _presentation = presentationPipeline
            ?? throw new ArgumentNullException(nameof(presentationPipeline));
        _isPublicationBlockedByRetirement =
            IsPublicationBlockedByRetirement;
        _configuredWorkBudgetOptions =
            workBudgetOptions ?? StreamingWorkBudgetOptions.Default;
        _workBudget = _configuredWorkBudgetOptions.ToBudget();
        if (!_presentation.MatchesState(_state))
        {
            throw new ArgumentException(
                "The presentation pipeline must own the controller's world state.",
                nameof(presentationPipeline));
        }
        NearRadius = nearRadius;
        FarRadius  = farRadius;
    }

    /// <summary>
    /// Reconciles an active outdoor streaming window to new quality radii
    /// without replacing the worker/controller or blindly bootstrapping data
    /// already published in <see cref="GpuWorldState"/>. Pending jobs from the
    /// old window are invalidated by generation; resident blocks are retained,
    /// promoted, demoted, loaded, or unloaded from their actual current tier.
    /// A sealed dungeon remains radius-zero until its normal exit edge while
    /// remembering the new outdoor radii for that exit.
    /// </summary>
    public void ReconfigureRadii(int nearRadius, int farRadius)
    {
        if (nearRadius < 0)
            throw new ArgumentOutOfRangeException(nameof(nearRadius));
        if (farRadius < nearRadius)
            throw new ArgumentOutOfRangeException(
                nameof(farRadius),
                "Far radius must be greater than or equal to near radius.");

        // A radius transaction may not admit work after an origin-recenter
        // snapshot has been captured. Retain only the latest requested radii;
        // the first Tick after origin commit applies them to the new frame.
        if (_originRecenterRetirement is not null)
        {
            _deferredRadiiRequest = (nearRadius, farRadius);
            return;
        }

        // Queue callbacks are injected seams and can synchronously reenter the
        // controller. Last-request-wins deferral keeps the active mutation
        // cursor stable; the request is applied after the current transaction
        // converges instead of recursively replaying its active step.
        if (_advancingRadiiReconfiguration)
        {
            _deferredRadiiRequest = (nearRadius, farRadius);
            return;
        }

        // A prior callback may have failed after this controller accepted the
        // request. Resume that exact transaction before considering another
        // target; replacing it would orphan already-admitted generation work.
        if (_pendingRadiiReconfiguration is not null)
            AdvanceRadiiReconfiguration();

        // This explicit call is newer than any request deferred by a prior
        // failed attempt and therefore supersedes it.
        _deferredRadiiRequest = null;

        if (nearRadius == NearRadius && farRadius == FarRadius)
            return;

        // A staged old-window publication must either finish or fail before
        // the desired-tier snapshot below is computed. Otherwise a landblock
        // that becomes spatially resident during convergence is absent from
        // the reconfiguration mutation ledger.
        if (!ConvergePendingPublications())
        {
            _deferredRadiiRequest = (nearRadius, farRadius);
            return;
        }

        if (_collapsed || _region is null)
        {
            NearRadius = nearRadius;
            FarRadius = farRadius;
            return;
        }

        var rebuilt = new StreamingRegion(
            _region.CenterX,
            _region.CenterY,
            nearRadius,
            farRadius);
        TwoTierDiff bootstrap = rebuilt.ComputeFirstTickDiff();
        var desiredNear = new HashSet<uint>(bootstrap.ToLoadNear);
        var desiredFar = new HashSet<uint>(bootstrap.ToLoadFar);
        var mutations = new List<RadiiMutation>();

        uint[] loaded = [.. _state.LoadedLandblockIds];
        for (int i = 0; i < loaded.Length; i++)
        {
            uint id = loaded[i];
            if (desiredNear.Contains(id))
            {
                if (!_state.IsNearTier(id))
                    mutations.Add(new RadiiMutation(
                        () => EnqueueLoad(id, LandblockStreamJobKind.PromoteToNear)));
            }
            else if (desiredFar.Contains(id))
            {
                if (_state.IsNearTier(id))
                    mutations.Add(new RadiiMutation(() => DemoteLandblock(id)));
            }
            else
            {
                mutations.Add(new RadiiMutation(() => EnqueueUnload(id)));
            }
        }

        foreach (uint id in desiredNear)
            if (!_state.IsLoaded(id))
                mutations.Add(new RadiiMutation(
                    () => EnqueueLoad(id, LandblockStreamJobKind.LoadNear)));
        foreach (uint id in desiredFar)
            if (!_state.IsLoaded(id))
                mutations.Add(new RadiiMutation(
                    () => EnqueueLoad(id, LandblockStreamJobKind.LoadFar)));

        _pendingRadiiReconfiguration = new RadiiReconfiguration(
            nearRadius,
            farRadius,
            rebuilt,
            mutations);
        AdvanceRadiiReconfiguration();
    }

    /// <summary>
    /// Compatibility constructor for deterministic tests and callers that do
    /// not own an asynchronous worker. Production must use the generation-aware
    /// overload above so hard-recenter results can be rejected by incarnation.
    /// </summary>
    internal StreamingController(
        Action<uint, LandblockStreamJobKind> enqueueLoad,
        Action<uint> enqueueUnload,
        Func<int, IReadOnlyList<LandblockStreamResult>> drainCompletions,
        Action<LandblockBuild, LandblockMeshData> applyTerrain,
        GpuWorldState state,
        int nearRadius,
        int farRadius,
        Action<uint>? removeTerrain = null,
        Action<uint>? demoteNearLayer = null,
        Action? clearPendingLoads = null,
        Action<uint>? onLandblockLoaded = null,
        Action<EnvCellLandblockBuild>? ensureEnvCellMeshes = null,
        LandblockRetirementCoordinator? retirementCoordinator = null,
        StreamingWorkBudgetOptions? workBudgetOptions = null)
        : this(
            (id, kind, _) => enqueueLoad(id, kind),
            (id, _) => enqueueUnload(id),
            drainCompletions,
            applyTerrain,
            state,
            nearRadius,
            farRadius,
            removeTerrain,
            demoteNearLayer,
            clearPendingLoads,
            onLandblockLoaded,
            ensureEnvCellMeshes,
            retirementCoordinator,
            workBudgetOptions)
    {
    }

    private void EnqueueLoad(uint id, LandblockStreamJobKind kind) =>
        _enqueueLoad(id, kind, _generation);

    private void EnqueueUnload(uint id) => _enqueueUnload(id, _generation);

    private void DemoteLandblock(uint id)
    {
        uint canonical = (id & 0xFFFF0000u) | 0xFFFFu;
        _presentation.EnqueueNearLayerRetirement(canonical);
    }

    private bool AdvanceGeneration()
    {
        if (!ConvergePendingPublications())
            return false;
        _generation = unchecked(_generation + 1);
        return true;
    }

    private bool ConvergePendingPublications(
        bool preferDestination = false)
    {
        IReadOnlyList<LandblockStreamResult> pending =
            _presentation.GetPendingPublicationResults();
        if (pending.Count == 0)
            return true;
        if (_activeWorkMeter is null)
        {
            // Settings/native callbacks may request policy changes between
            // frames, but publication remains owned by the frame-scoped meter.
            return false;
        }

        try
        {
            bool progressed = false;
            bool destinationPending = false;
            if (preferDestination)
            {
                for (int i = 0; i < pending.Count; i++)
                {
                    if (!IsDestinationWork(pending[i].LandblockId))
                        continue;
                    destinationPending = true;
                    break;
                }
            }
            // A partially prepared background landblock must not serialize the
            // reveal-critical destination behind dictionary insertion order.
            // Each publication owns independent receipts, so destination
            // transactions can safely resume first without replaying or
            // discarding the non-destination cursor.
            for (int destinationPass = 1; destinationPass >= 0; destinationPass--)
            {
                bool requireDestination = destinationPass != 0;
                for (int i = 0; i < pending.Count; i++)
                {
                    LandblockStreamResult result = pending[i];
                    bool isDestination = IsDestinationWork(result.LandblockId);
                    if (destinationPending && !isDestination)
                        continue;
                    if (isDestination != requireDestination)
                        continue;

                    using StreamingWorkMeter.LaneScope lane =
                        _activeWorkMeter.EnterLane(
                            isDestination
                                ? StreamingWorkLane.Destination
                                : StreamingWorkLane.NonDestination);
                    LandblockPublicationAdvance advance =
                        _presentation.ResumePublication(
                            result,
                            _activeWorkMeter,
                            ensureProgress: !progressed);
                    progressed |= advance.Progressed;
                    if (!advance.Completed)
                        return false;
                }
            }
            return true;
        }
        finally
        {
            // A completed staged result can still be present in the deferred
            // FIFO that originally retained it. Remove only those exact
            // objects; unrelated accepted completions keep their order.
            _completionQueue.RemoveResults(
                pending,
                result => !_presentation.HasPendingPublication(result));
        }
    }

    /// <summary>
    /// Advance one frame. <paramref name="observerCx"/>/<paramref name="observerCy"/>
    /// are landblock coordinates (0..255) of the current viewer — the camera
    /// in offline mode, the server-sent player position in live.
    ///
    /// <para>Two-tier model (Phase A.5 T13):</para>
    /// <list type="bullet">
    ///   <item><see cref="TwoTierDiff.ToLoadFar"/> → enqueue LoadFar (terrain only, no entities)</item>
    ///   <item><see cref="TwoTierDiff.ToLoadNear"/> → enqueue LoadNear (terrain + entities)</item>
    ///   <item><see cref="TwoTierDiff.ToPromote"/> → enqueue PromoteToNear (entity layer for already-loaded terrain)</item>
    ///   <item><see cref="TwoTierDiff.ToDemote"/> → drop entities on render thread immediately (terrain stays)</item>
    ///   <item><see cref="TwoTierDiff.ToUnload"/> → enqueue full unload</item>
    /// </list>
    /// </summary>
    public void Tick(int observerCx, int observerCy, bool insideDungeon = false)
    {
        if (_activeWorkMeter is not null)
            throw new InvalidOperationException(
                "StreamingController.Tick cannot be reentered.");

        var meter = new StreamingWorkMeter(
            _workBudget,
            destinationReservationActive: _destinationReservation is not null);
        _activeWorkMeter = meter;
        try
        {
            if (_fullWindowRetirement is not null)
            {
                bool retirementWasPending =
                    _presentation.PendingRetirementCount != 0;
                TryAdvanceFullWindowRetirement(meter);
                AdvanceDestinationRetirementDependency(meter);
                if (_presentation.UsesBudgetedRetirementSteps
                    || retirementWasPending)
                {
                    _presentation.AdvanceRetirements(meter);
                }
                if (_fullWindowRetirement is
                    {
                        PreparationCommitted: true,
                    }
                    && _presentation.PendingRetirementCount == 0)
                {
                    _fullWindowRetirement = null;
                }
                return;
            }

            if (_originRecenterRetirement is not null)
            {
                bool retirementWasPending =
                    _presentation.PendingRetirementCount != 0;
                TryAdvanceOriginRecenterPreparation(meter);
                AdvanceDestinationRetirementDependency(meter);
                if (_presentation.UsesBudgetedRetirementSteps
                    || retirementWasPending)
                {
                    _presentation.AdvanceRetirements(meter);
                }
                return;
            }

            if (_pendingRadiiReconfiguration is not null)
                AdvanceRadiiReconfiguration();
            else if (_deferredRadiiRequest is { } deferred)
            {
                _deferredRadiiRequest = null;
                ReconfigureRadii(deferred.NearRadius, deferred.FarRadius);
            }

            // Complete an admitted publication before this frame can mutate its
            // desired tier or spatial residence. A later recenter may then demote
            // or retire the fully known owner set through the normal ledger.
            bool retirementWasPendingAtFrameStart =
                _presentation.PendingRetirementCount != 0;
            AdvanceDestinationRetirementDependency(meter);
            _presentation.AdvanceRetirements(meter);
            bool destinationPublicationIncomplete =
                _destinationReservation is not null
                && !IsRenderNeighborhoodResident(
                    DestinationLandblockId,
                    DestinationRadius);
            if (!ConvergePendingPublications(
                    preferDestination: destinationPublicationIncomplete))
                return;

            uint centerId = StreamingRegion.EncodeLandblockId(observerCx, observerCy);

            if (_collapsed)
            {
                // Hysteresis. Cases:
                //  - Still in the SAME dungeon landblock → hold (sweep stragglers).
                //  - In a DIFFERENT dungeon cell (multi-landblock dungeon / new dungeon)
                //    → re-collapse onto it.
                //  - CurrCell flickered null but the player hasn't gone anywhere: the
                //    observer landblock reverts to the position-derived value, which for a
                //    dungeon is only ever the ADJACENT off-by-one landblock (negative cell-
                //    local Y). Hold — never expand on an adjacent flicker.
                //  - Genuinely left to a DISTANT landblock (portal/teleport out, always far
                //    from the ocean-grid dungeon block) → expand.
                if (insideDungeon && centerId != _collapsedCenter)
                    EnterDungeonCollapse(observerCx, observerCy, centerId);
                else if (!insideDungeon && ChebyshevLandblocks(centerId, _collapsedCenter) > 1)
                    ExitDungeonExpand(observerCx, observerCy);
                else
                    SweepCollapsed();
            }
            else if (insideDungeon)
            {
                EnterDungeonCollapse(observerCx, observerCy, centerId);
            }
            else
            {
                NormalTick(observerCx, observerCy);
            }

            DrainAndApply(
                preferDestination: destinationPublicationIncomplete);
            // Retirement is cleanup after immediate spatial withdrawal. Spend
            // remaining capacity only after visible/destination publication
            // work has had access to this frame's shared meter.
            if (_presentation.UsesBudgetedRetirementSteps
                && !retirementWasPendingAtFrameStart)
                _presentation.AdvanceRetirements(meter);
        }
        finally
        {
            meter.FinishFrame();
            StreamingWorkMeterSnapshot snapshot = meter.Snapshot;
            ObserveWorkLifetime(snapshot);
            _lastWorkMeter = snapshot;
            _activeWorkMeter = null;
        }
    }

    private void AdvanceDestinationRetirementDependency(
        StreamingWorkMeter meter)
    {
        if (_destinationReservation is null
            || !_presentation.IsRetirementPending(DestinationLandblockId))
        {
            return;
        }

        using StreamingWorkMeter.LaneScope lane =
            meter.EnterLane(StreamingWorkLane.Destination);
        _presentation.AdvancePriorityRetirement(
            DestinationLandblockId,
            meter);
    }

    private void ObserveWorkLifetime(StreamingWorkMeterSnapshot snapshot)
    {
        _lifetimeWorkOverruns = SaturatingAdd(
            _lifetimeWorkOverruns,
            snapshot.OverrunCount);
        _lifetimeOversizedProgress = SaturatingAdd(
            _lifetimeOversizedProgress,
            snapshot.OversizedProgressCount);

        if (_maximumWorkFrameStage is null
            || snapshot.ElapsedMilliseconds > _maximumWorkFrameMilliseconds)
        {
            _maximumWorkFrameMilliseconds = snapshot.ElapsedMilliseconds;
            _maximumWorkFrameStage = snapshot.LastStage;
        }

        if (_maximumWorkOperationStage is null
            || snapshot.MaximumOperationMilliseconds
                > _maximumWorkOperationMilliseconds)
        {
            _maximumWorkOperationMilliseconds =
                snapshot.MaximumOperationMilliseconds;
            _maximumWorkOperationStage = snapshot.MaximumOperationStage;
        }
    }

    private static long SaturatingAdd(long left, int right) =>
        left > long.MaxValue - right ? long.MaxValue : left + right;

    private void AdvanceRadiiReconfiguration()
    {
        if (_advancingRadiiReconfiguration)
            return;

        RadiiReconfiguration pending = _pendingRadiiReconfiguration
            ?? throw new InvalidOperationException("No radii reconfiguration is pending.");
        var failures = new List<Exception>();
        _advancingRadiiReconfiguration = true;
        try
        {

            if (!pending.GenerationAdvanced)
            {
                if (!AdvanceGeneration())
                    return;
                pending.GenerationAdvanced = true;
            }

            if (!pending.PendingLoadsCleared)
            {
                try
                {
                    _clearPendingLoads?.Invoke();
                    pending.PendingLoadsCleared = true;
                }
                catch (Exception error)
                {
                    if (error is StreamingMutationException { MutationCommitted: true })
                        pending.PendingLoadsCleared = true;
                    failures.Add(error);
                }
            }

            // Do not admit new-generation work until cancellation of the old
            // inbox is durable. Otherwise a successful retry of ClearPendingLoads
            // would also erase mutations already marked complete by this ledger.
            for (int i = 0; pending.PendingLoadsCleared && i < pending.Mutations.Count; i++)
            {
                RadiiMutation mutation = pending.Mutations[i];
                if (mutation.Completed)
                    continue;
                try
                {
                    mutation.Apply();
                    mutation.Completed = true;
                }
                catch (Exception error)
                {
                    if (error is StreamingMutationException { MutationCommitted: true })
                        mutation.Completed = true;
                    failures.Add(error);
                }
            }

            bool converged = pending.PendingLoadsCleared
                && pending.Mutations.All(static mutation => mutation.Completed);
            if (converged)
            {
                // Publish the new logical window only after every queue/retirement
                // admission is durable. A later Tick can now trust Resident as the
                // complete desired set and will never strand a hole.
                pending.Region.MarkResidentFromBootstrap();
                _region = pending.Region;
                NearRadius = pending.NearRadius;
                FarRadius = pending.FarRadius;
                _completionQueue.Clear();
                _pendingRadiiReconfiguration = null;
            }
        }
        finally
        {
            _advancingRadiiReconfiguration = false;
        }

        if (failures.Count != 0)
        {
            throw new AggregateException(
                "Streaming quality reconfiguration did not complete cleanly.",
                failures);
        }

        if (_pendingRadiiReconfiguration is null
            && _deferredRadiiRequest is { } deferred)
        {
            _deferredRadiiRequest = null;
            ReconfigureRadii(deferred.NearRadius, deferred.FarRadius);
        }
    }

    private sealed class RadiiReconfiguration
    {
        public RadiiReconfiguration(
            int nearRadius,
            int farRadius,
            StreamingRegion region,
            List<RadiiMutation> mutations)
        {
            NearRadius = nearRadius;
            FarRadius = farRadius;
            Region = region;
            Mutations = mutations;
        }

        public int NearRadius { get; }
        public int FarRadius { get; }
        public StreamingRegion Region { get; }
        public List<RadiiMutation> Mutations { get; }
        public bool GenerationAdvanced { get; set; }
        public bool PendingLoadsCleared { get; set; }
    }

    private sealed class RadiiMutation
    {
        public RadiiMutation(Action apply) => Apply = apply;

        public Action Apply { get; }
        public bool Completed { get; set; }
    }

    /// <summary>
    /// #135: collapse to a single dungeon landblock IMMEDIATELY, before the first
    /// <see cref="Tick"/> has a chance to bootstrap the full 25×25 window. Called
    /// from the login / teleport spawn path the instant the streaming center is
    /// recentered onto a SEALED dungeon landblock.
    ///
    /// <para>The per-frame <c>insideDungeon</c> gate keys on the physics
    /// <c>CurrCell</c>, which is only set once the player is PLACED — and placement
    /// waits for the dungeon landblock to hydrate. So for the whole hydration window
    /// (tens of seconds for a ~200-cell dungeon) the gate reads false and
    /// <see cref="NormalTick"/> would enqueue the ~24 unrelated ocean-grid neighbor
    /// dungeons (+ ~19k entities each); the collapse then only mops them up after
    /// placement. That mop-up is the 10→high FPS ramp users see at a dungeon login.</para>
    ///
    /// <para>Pre-collapsing means the EXPENSIVE dungeon-neighbour window is never
    /// enqueued. On teleport nothing is enqueued at all (this fires before the next
    /// Tick recenters). On login a brief Holtburg outdoor window may be enqueued by the
    /// frame-1 NormalTick (before the player's spawn arrives) and is immediately
    /// cancelled by <c>_clearPendingLoads</c> here — cheap outdoor terrain, not the
    /// ocean-grid dungeons, and a handful of already-dequeued loads get swept next
    /// frame. Idempotent: a no-op when already collapsed onto this same landblock, so a
    /// re-sent spawn or a same-frame double call costs nothing. Render-thread only,
    /// same as <see cref="Tick"/>.</para>
    /// </summary>
    public void InitializeKnownLoginCenter(int cx, int cy, bool isSealedDungeon)
    {
        // StreamingReadinessGate keeps the worker stopped until the player's
        // real spawn supplies this center. There is therefore no guessed-center
        // window to force-reload here. A sealed dungeon is the one special case:
        // pin radius zero before the first Tick so no ocean-grid neighbours are
        // ever enqueued.
        if (isSealedDungeon)
            PreCollapseToDungeon(cx, cy);
    }

    /// <summary>
    /// Pins streaming to a sealed dungeon before the first normal Tick.
    /// </summary>
    public void PreCollapseToDungeon(int cx, int cy)
    {
        uint centerId = StreamingRegion.EncodeLandblockId(cx, cy);
        if (_collapsed && _collapsedCenter == centerId) return;
        EnterDungeonCollapse(cx, cy, centerId);
    }

    /// <summary>
    /// Outdoor / building-interior streaming — the original two-tier model.
    /// </summary>
    private void NormalTick(int observerCx, int observerCy)
    {
        if (_region is null)
        {
            _region = new StreamingRegion(observerCx, observerCy, NearRadius, FarRadius);
            var bootstrap = _region.ComputeFirstTickDiff();
            EnqueueLoadsByRevealPriority(
                bootstrap.ToLoadNear,
                LandblockStreamJobKind.LoadNear);
            foreach (var id in bootstrap.ToLoadFar)  EnqueueLoad(id, LandblockStreamJobKind.LoadFar);
            _region.MarkResidentFromBootstrap();
        }
        else if (_region.CenterX != observerCx || _region.CenterY != observerCy)
        {
            var diff = _region.RecenterTo(observerCx, observerCy);
            EnqueueLoadsByRevealPriority(
                diff.ToPromote,
                LandblockStreamJobKind.PromoteToNear);
            EnqueueLoadsByRevealPriority(
                diff.ToLoadNear,
                LandblockStreamJobKind.LoadNear);
            foreach (var id in diff.ToLoadFar)  EnqueueLoad(id, LandblockStreamJobKind.LoadFar);
            foreach (var id in diff.ToDemote)   DemoteLandblock(id);
            foreach (var id in diff.ToUnload)   EnqueueUnload(id);
        }
    }

    /// <summary>
    /// Places the canonical reveal neighborhood at the front of the worker's
    /// existing Near FIFO. Completion priority alone is too late: a
    /// non-destination build can otherwise begin a multi-frame publication
    /// before the destination result has even reached the render thread.
    /// </summary>
    private void EnqueueLoadsByRevealPriority(
        IReadOnlyList<uint> landblockIds,
        LandblockStreamJobKind kind,
        bool skipLoaded = false)
    {
        for (int destinationPass = 1; destinationPass >= 0; destinationPass--)
        {
            bool requireDestination = destinationPass != 0;
            for (int i = 0; i < landblockIds.Count; i++)
            {
                uint id = landblockIds[i];
                if ((!skipLoaded || !_state.IsLoaded(id))
                    && IsDestinationWork(id) == requireDestination)
                {
                    EnqueueLoad(id, kind);
                }
            }
        }
    }

    /// <summary>
    /// Dungeon-entry edge: cancel the in-flight window load, unload every
    /// resident neighbor, and pin streaming to the player's single dungeon
    /// landblock. Retail-faithful — AC dungeons have no adjacent landblocks
    /// (ACE <c>LandblockManager.GetAdjacentIDs</c> returns empty for a dungeon);
    /// the 25×25 window was pulling in ~129 unrelated ocean-grid dungeons and
    /// their thousands of emitters (#133 FPS). Unloading them also tears down
    /// their lights, shrinking the static-light set toward retail's ≤40.
    /// </summary>
    private void EnterDungeonCollapse(int cx, int cy, uint centerId)
    {
        bool logTransition = !_collapsed || _collapsedCenter != centerId;
        if (logTransition)
            Console.WriteLine($"streaming: dungeon collapse -> 0x{centerId:X8}");
        if (!AdvanceGeneration())
            return;
        _collapsed = true;
        _collapsedCenter = centerId;
        _clearPendingLoads?.Invoke();
        _completionQueue.Clear();
        _region = null;

        foreach (var id in _state.LoadedLandblockIds)
            if (id != centerId) EnqueueUnload(id);

        // Pin a radius-0 region so RecenterTo never re-expands while inside,
        // and so the post-exit rebuild starts from a clean, consistent state.
        _region = new StreamingRegion(cx, cy, 0, 0);
        _region.MarkResidentFromBootstrap();

        // The dungeon landblock itself must be (or become) loaded. If a prior
        // ClearPendingLoads cancelled its queued load, re-enqueue it.
        if (!_state.IsLoaded(centerId))
            EnqueueLoad(centerId, LandblockStreamJobKind.LoadNear);
        else if (!_state.IsNearTier(centerId))
            EnqueueLoad(centerId, LandblockStreamJobKind.PromoteToNear);
    }

    /// <summary>
    /// While collapsed, unload any landblock that finished loading after the
    /// collapse edge — a Load the worker had already dequeued before the
    /// <see cref="LandblockStreamer.ClearPendingLoads"/> control job took
    /// effect. At steady state only the dungeon landblock is resident, so this
    /// is a no-op.
    /// </summary>
    private void SweepCollapsed()
    {
        // Always preserve the true dungeon landblock (_collapsedCenter), never the
        // per-frame observer landblock — a CurrCell flicker must not unload the dungeon.
        foreach (var id in _state.LoadedLandblockIds)
            if (id != _collapsedCenter) EnqueueUnload(id);
    }

    /// <summary>Chebyshev distance in landblock cells between two landblock ids.</summary>
    private static int ChebyshevLandblocks(uint a, uint b)
    {
        int ax = (int)((a >> 24) & 0xFFu), ay = (int)((a >> 16) & 0xFFu);
        int bx = (int)((b >> 24) & 0xFFu), by = (int)((b >> 16) & 0xFFu);
        return Math.Max(Math.Abs(ax - bx), Math.Abs(ay - by));
    }

    /// <summary>
    /// True when <paramref name="id"/> belongs to the canonical reveal
    /// generation's destination neighborhood.
    /// </summary>
    private bool IsDestinationWork(uint id)
        => _destinationReservation is { } reservation
           && ChebyshevLandblocks(id, reservation.LandblockId)
                <= reservation.Radius;

    /// <summary>
    /// Dungeon-exit edge (portal to outdoors / teleport): rebuild the full
    /// two-tier window at the new center and unload anything resident from the
    /// collapsed state that falls outside it.
    /// </summary>
    private void ExitDungeonExpand(int observerCx, int observerCy)
    {
        Console.WriteLine(
            $"streaming: dungeon EXIT-expand -> ({observerCx},{observerCy}) " +
            $"(was collapsed on 0x{_collapsedCenter:X8})");
        if (!AdvanceGeneration())
            return;
        _collapsed = false;
        var rebuilt = new StreamingRegion(observerCx, observerCy, NearRadius, FarRadius);

        foreach (var id in _state.LoadedLandblockIds)
            if (!rebuilt.Resident.Contains(id)) EnqueueUnload(id);

        var boot = rebuilt.ComputeFirstTickDiff();
        EnqueueLoadsByRevealPriority(
            boot.ToLoadNear,
            LandblockStreamJobKind.LoadNear,
            skipLoaded: true);
        foreach (var id in boot.ToLoadFar)
            if (!_state.IsLoaded(id)) EnqueueLoad(id, LandblockStreamJobKind.LoadFar);
        rebuilt.MarkResidentFromBootstrap();
        _region = rebuilt;
    }

    /// <summary>
    /// Starts a reload of the current streaming window. The next and later
    /// <see cref="Tick"/> calls invalidate accepted work, capture and detach
    /// stable residents, and converge render, physics, and static-lighting
    /// owners under the shared frame budget before <see cref="NormalTick"/>
    /// bootstraps the window again. Hard login/portal transitions separately
    /// publish their immediate observable edge through
    /// <see cref="WorldGenerationQuiescence"/>.
    /// Shared-origin teleports use <see cref="BeginOriginRecenter"/> instead so
    /// every old-window owner retires while the old coordinate frame is active.
    /// </summary>
    public void ForceReloadWindow()
    {
        BeginFullWindowRetirement();
    }

    /// <summary>
    /// Starts the old-window half of a shared-origin recenter. All accepted
    /// worker work is invalidated and every resident landblock is detached
    /// through the retryable presentation ledger while the old coordinate
    /// frame is still active. <see cref="Tick"/> remains blocked until
    /// <see cref="TryCommitOriginRecenter"/> succeeds.
    /// </summary>
    internal void BeginOriginRecenter()
    {
        _originRecenterRetirement ??= new OriginRecenterRetirement();
    }

    /// <summary>
    /// Reports whether every old resident has been detached and captured by
    /// an exact retirement receipt. Deferred receipt cleanup may continue
    /// after the shared origin changes; same-landblock publication remains
    /// fenced by <see cref="IsPublicationBlockedByRetirement"/>.
    /// </summary>
    internal bool IsOriginRecenterRetirementComplete()
    {
        if (_originRecenterRetirement is null)
            throw new InvalidOperationException(
                "No streaming-origin recenter transaction is pending.");

        return _originRecenterRetirement.PreparationCommitted;
    }

    /// <summary>
    /// Releases the streaming bootstrap gate after the composition root has
    /// committed the new shared origin. Old detached receipts continue under
    /// the shared frame meter and fence only a replacement publication with
    /// the same canonical landblock key.
    /// </summary>
    internal bool TryCommitOriginRecenter(
        int destinationX,
        int destinationY,
        bool isSealedDungeon)
    {
        if (_advancingOriginRecenter)
            return false;

        _advancingOriginRecenter = true;
        try
        {
            return TryCommitOriginRecenterCore(
                destinationX,
                destinationY,
                isSealedDungeon);
        }
        finally
        {
            _advancingOriginRecenter = false;
        }
    }

    private bool TryCommitOriginRecenterCore(
        int destinationX,
        int destinationY,
        bool isSealedDungeon)
    {
        OriginRecenterRetirement transaction = _originRecenterRetirement
            ?? throw new InvalidOperationException(
                "No streaming-origin recenter transaction is pending.");
        if (!transaction.PreparationCommitted)
            throw new InvalidOperationException(
                "Streaming-origin retirement preparation has not completed.");
        var destination = (destinationX, destinationY, isSealedDungeon);
        if (transaction.Destination is { } retained && retained != destination)
        {
            throw new InvalidOperationException(
                "A recenter transaction cannot commit two different destinations.");
        }
        transaction.Destination ??= destination;

        if (!transaction.DestinationConfigured)
        {
            _collapsed = isSealedDungeon;
            _collapsedCenter = isSealedDungeon
                ? StreamingRegion.EncodeLandblockId(destinationX, destinationY)
                : 0u;
            if (isSealedDungeon)
            {
                _region = new StreamingRegion(
                    destinationX,
                    destinationY,
                    nearRadius: 0,
                    farRadius: 0);
                _region.MarkResidentFromBootstrap();
            }
            else
            {
                _region = null;
            }
            transaction.DestinationConfigured = true;
        }

        if (isSealedDungeon && !transaction.DestinationLoadEnqueued)
        {
            try
            {
                EnqueueLoad(
                    StreamingRegion.EncodeLandblockId(destinationX, destinationY),
                    LandblockStreamJobKind.LoadNear);
                transaction.DestinationLoadEnqueued = true;
            }
            catch (StreamingMutationException error) when (error.MutationCommitted)
            {
                transaction.DestinationLoadEnqueued = true;
                Console.WriteLine(
                    $"streaming: committed dungeon recenter enqueue reported failure: {error}");
                return false;
            }
            catch (Exception error)
            {
                Console.WriteLine(
                    $"streaming: dungeon recenter enqueue will resume: {error}");
                return false;
            }
        }

        _originRecenterRetirement = null;
        return true;
    }

    /// <summary>
    /// Releases a fully retired origin transaction at a session boundary
    /// without bootstrapping a destination from the ending session.
    /// </summary>
    internal bool TryCancelOriginRecenter()
    {
        if (_originRecenterRetirement is not { PreparationCommitted: true })
            return false;

        _collapsed = false;
        _collapsedCenter = 0u;
        _region = null;
        _originRecenterRetirement = null;
        return true;
    }

    private bool TryAdvanceOriginRecenterPreparation(StreamingWorkMeter meter)
    {
        ArgumentNullException.ThrowIfNull(meter);
        OriginRecenterRetirement transaction = _originRecenterRetirement
            ?? throw new InvalidOperationException(
                "No streaming-origin recenter transaction is pending.");
        if (transaction.PreparationCommitted)
            return true;
        if (_advancingOriginRecenter)
            return false;

        _advancingOriginRecenter = true;
        try
        {
            if (!transaction.RadiiConverged)
            {
                if (_pendingRadiiReconfiguration is not null
                    && !TryRunStreamingWork(
                        meter,
                        new StreamingWorkCost(
                            EntityOperations: Math.Max(
                                1,
                                _pendingRadiiReconfiguration.Mutations.Count)),
                        "recenter-radii-convergence",
                        () =>
                        {
                            AdvanceRadiiReconfiguration();
                            return _pendingRadiiReconfiguration is null;
                        }))
                {
                    return false;
                }
                if (_pendingRadiiReconfiguration is not null)
                    return false;
                transaction.RadiiConverged = true;
            }
            if (!transaction.GenerationAdvanced)
            {
                // AdvanceGeneration meters any retained publication it must
                // converge through the active frame meter. The token increment
                // itself is constant-time and must not create a nested meter
                // reservation around that retry.
                if (!AdvanceGeneration())
                    return false;
                transaction.GenerationAdvanced = true;
            }
            if (!transaction.PendingLoadsCleared)
            {
                bool ClearPendingLoads()
                {
                    try
                    {
                        _clearPendingLoads?.Invoke();
                        transaction.PendingLoadsCleared = true;
                        return true;
                    }
                    catch (StreamingMutationException error) when (error.MutationCommitted)
                    {
                        transaction.PendingLoadsCleared = true;
                        Console.WriteLine(
                            $"streaming: committed pending-load clear reported failure: {error}");
                        return false;
                    }
                }

                if (!TryRunStreamingWork(
                        meter,
                        new StreamingWorkCost(EntityOperations: 1),
                        "recenter-clear-worker-inbox",
                        ClearPendingLoads))
                    return false;
            }
            if (!transaction.CompletionQueueCleared)
            {
                while (_completionQueue.Count != 0)
                {
                    StreamingWorkAdmission admission = meter.TryReserve(
                        new StreamingWorkCost(EntityOperations: 1),
                        "recenter-release-completion");
                    if (admission == StreamingWorkAdmission.Yielded)
                        return false;

                    if (!_completionQueue.TryRemoveOne())
                    {
                        meter.Fail();
                        throw new InvalidOperationException(
                            "Completion queue count changed during recenter release.");
                    }
                    meter.Complete();
                }
                transaction.CompletionQueueCleared = true;
            }
            if (!transaction.RegionCleared)
            {
                if (!TryRunStreamingWork(
                        meter,
                        new StreamingWorkCost(EntityOperations: 1),
                        "recenter-clear-region",
                        () =>
                        {
                            _collapsed = false;
                            _region = null;
                            return true;
                        }))
                {
                    return false;
                }
                transaction.RegionCleared = true;
            }
            if (!transaction.SpatialGenerationDetached)
            {
                StreamingWorkAdmission admission = meter.TryReserve(
                    new StreamingWorkCost(
                        EntityOperations:
                            _state.OriginRecenterSpatialOperationCount),
                    "recenter-detach-spatial-generation",
                    ensureProgress: true);
                if (admission == StreamingWorkAdmission.Yielded)
                    return false;

                try
                {
                    GpuWorldRecenterRetirement detached =
                        _presentation.DetachAllForOriginRecenter();
                    transaction.SpatialGenerationDetached = true;
                    FullWindowRetirementCount++;
                    LastFullWindowRetirementLandblockCount =
                        detached.Landblocks.Count;
                    meter.Complete();
                    if (detached.ObserverFailure is not null)
                    {
                        Console.WriteLine(
                            "streaming: committed origin-recenter spatial " +
                            $"generation reported failure: {detached.ObserverFailure}");
                        return false;
                    }
                }
                catch
                {
                    meter.Fail();
                    throw;
                }
            }

            transaction.PreparationCommitted = true;
            return true;
        }
        catch (Exception error)
        {
            Console.WriteLine(
                $"streaming: origin-recenter preparation will resume: {error}");
            return false;
        }
        finally
        {
            _advancingOriginRecenter = false;
        }
    }

    private void BeginFullWindowRetirement()
    {
        _fullWindowRetirement ??= new FullWindowRetirement();
    }

    private bool TryAdvanceFullWindowRetirement(StreamingWorkMeter meter)
    {
        FullWindowRetirement transaction = _fullWindowRetirement
            ?? throw new InvalidOperationException(
                "No full-window retirement transaction is pending.");
        if (transaction.PreparationCommitted)
            return true;

        try
        {
            if (!transaction.GenerationAdvanced)
            {
                if (!AdvanceGeneration())
                {
                    return false;
                }
                transaction.GenerationAdvanced = true;
            }

            if (!transaction.PendingLoadsCleared)
            {
                bool ClearPendingLoads()
                {
                    try
                    {
                        _clearPendingLoads?.Invoke();
                        transaction.PendingLoadsCleared = true;
                        return true;
                    }
                    catch (StreamingMutationException error) when (error.MutationCommitted)
                    {
                        transaction.PendingLoadsCleared = true;
                        Console.WriteLine(
                            $"streaming: committed reload pending-load clear reported failure: {error}");
                        return false;
                    }
                }

                if (!TryRunStreamingWork(
                        meter,
                        new StreamingWorkCost(EntityOperations: 1),
                        "reload-clear-worker-inbox",
                        ClearPendingLoads))
                {
                    return false;
                }
            }

            if (!transaction.CompletionQueueCleared)
            {
                while (_completionQueue.Count != 0)
                {
                    StreamingWorkAdmission admission = meter.TryReserve(
                        new StreamingWorkCost(EntityOperations: 1),
                        "reload-release-completion");
                    if (admission == StreamingWorkAdmission.Yielded)
                        return false;

                    if (!_completionQueue.TryRemoveOne())
                    {
                        meter.Fail();
                        throw new InvalidOperationException(
                            "Completion queue count changed during reload release.");
                    }
                    meter.Complete();
                }
                transaction.CompletionQueueCleared = true;
            }

            if (!transaction.RegionCleared)
            {
                if (!TryRunStreamingWork(
                        meter,
                        new StreamingWorkCost(EntityOperations: 1),
                        "reload-clear-region",
                        () =>
                        {
                            _collapsed = false;
                            _region = null;
                            return true;
                        }))
                {
                    return false;
                }
                transaction.RegionCleared = true;
            }

            if (transaction.ResidentIds is null)
            {
                transaction.ResidentIds = [];
                transaction.ResidentEnumerator =
                    _state.LoadedLandblockIds.GetEnumerator();
            }
            while (transaction.ResidentEnumerator is { } residentEnumerator)
            {
                StreamingWorkAdmission admission = meter.TryReserve(
                    new StreamingWorkCost(EntityOperations: 1),
                    "reload-capture-resident-id");
                if (admission == StreamingWorkAdmission.Yielded)
                    return false;

                bool moved;
                try
                {
                    moved = residentEnumerator.MoveNext();
                    if (moved)
                        transaction.ResidentIds.Add(residentEnumerator.Current);
                    else
                    {
                        residentEnumerator.Dispose();
                        transaction.ResidentEnumerator = null;
                        FullWindowRetirementCount++;
                        LastFullWindowRetirementLandblockCount =
                            transaction.ResidentIds.Count;
                    }
                    meter.Complete();
                }
                catch
                {
                    meter.Fail();
                    throw;
                }
            }

            List<uint> residentIds = transaction.ResidentIds
                ?? throw new InvalidOperationException(
                    "Full-window resident capture did not commit.");
            while (transaction.RetirementCursor < residentIds.Count)
            {
                uint id = residentIds[transaction.RetirementCursor];
                int entityCount = _state.TryGetLandblock(
                        id,
                        out LoadedLandblock? loaded)
                    ? loaded!.Entities.Count
                    : 0;
                StreamingWorkAdmission admission = meter.TryReserve(
                    new StreamingWorkCost(
                        EntityOperations: Math.Max(1, entityCount)),
                    $"reload-detach-0x{id:X8}");
                if (admission == StreamingWorkAdmission.Yielded)
                    return false;

                try
                {
                    _presentation.EnqueueFullRetirement(id);
                    transaction.RetirementCursor++;
                    meter.Complete();
                }
                catch (Exception error)
                {
                    if (!_state.IsLoaded(id))
                        transaction.RetirementCursor++;
                    meter.Fail();
                    Console.WriteLine(
                        $"streaming: full-window retirement for 0x{id:X8} " +
                        $"will resume: {error}");
                    return false;
                }
            }

            transaction.PreparationCommitted = true;
            return true;
        }
        catch (Exception error)
        {
            Console.WriteLine(
                $"streaming: full-window preparation will resume: {error}");
            return false;
        }
    }

    /// <summary>
    /// Adopts immutable worker results and advances explicit priority FIFOs
    /// under the one typed frame meter. Destination and unload work win queue
    /// order but no class bypasses time, bytes, entities, uploads, or retire
    /// operation limits.
    /// </summary>
    private void DrainAndApply(bool preferDestination = false)
    {
        StreamingWorkMeter meter = _activeWorkMeter
            ?? throw new InvalidOperationException(
                "Completion scheduling requires an active frame meter.");

        AdmitCompletions(meter);
        bool destinationQueued =
            preferDestination
            && _completionQueue.HasPriority(
                StreamingCompletionPriority.Destination);
        bool executed = false;
        while (_completionQueue.TryPeekNext(
            _isPublicationBlockedByRetirement,
            out StreamingQueuedCompletion? completion,
            destinationQueued
                ? StreamingCompletionPriority.Unload
                : StreamingCompletionPriority.Far))
        {
            StreamingQueuedCompletion work = completion
                ?? throw new InvalidOperationException(
                    "The completion queue returned a null head.");
            try
            {
                using StreamingWorkMeter.LaneScope lane = meter.EnterLane(
                    work.Priority == StreamingCompletionPriority.Destination
                        && work.RevealGeneration == ActiveRevealGeneration
                            ? StreamingWorkLane.Destination
                            : StreamingWorkLane.NonDestination);
                LandblockPublicationAdvance advance = ApplyResult(
                    work,
                    meter,
                    ensureProgress: !executed);
                executed |= advance.Progressed;
                if (!advance.Completed)
                    break;
                _completionQueue.RemoveHead(work);
            }
            catch
            {
                // Concrete publication failures retain their exact owner
                // receipt and queue head. A compatibility callback without a
                // receipt cannot safely replay, so consume only that result.
                if (!_presentation.HasPendingPublication(work.Result))
                    _completionQueue.RemoveHead(work);
                throw;
            }
        }
    }

    private static bool TryRunStreamingWork(
        StreamingWorkMeter meter,
        StreamingWorkCost cost,
        string stage,
        Func<bool> operation)
    {
        StreamingWorkAdmission admission = meter.TryReserve(cost, stage);
        if (admission == StreamingWorkAdmission.Yielded)
            return false;

        try
        {
            bool completed = operation();
            if (completed)
                meter.Complete();
            else
                meter.Fail();
            return completed;
        }
        catch
        {
            meter.Fail();
            throw;
        }
    }

    private void AdmitCompletions(StreamingWorkMeter meter)
    {
        while (_completionSource.TryPeek(out LandblockStreamResult? peeked))
        {
            LandblockStreamResult result = peeked
                ?? throw new InvalidOperationException(
                    "The completion source returned a null peek.");
            bool stale = IsStaleGeneration(result)
                && !_presentation.HasPendingPublication(result);
            StreamingCompletionPriority priority = stale
                ? StreamingCompletionPriority.Control
                : ClassifyCompletion(result);
            LandblockStreamCostEstimate estimate =
                LandblockStreamResultCost.Estimate(result);
            // A stale result belongs to a generation that was already
            // cancelled. Reading it releases worker-outbox ownership; it does
            // not admit payload into the current world. Charge only elapsed
            // time so a large completed old window cannot consume the
            // destination generation's completion quota for many frames.
            StreamingWorkCost admissionCost = stale
                ? default
                : new StreamingWorkCost(
                    CompletionAdmissions:
                        estimate.Work.CompletionAdmissions,
                    AdoptedCpuBytes:
                        estimate.Work.AdoptedCpuBytes);
            using StreamingWorkMeter.LaneScope lane = meter.EnterLane(
                priority == StreamingCompletionPriority.Destination
                    ? StreamingWorkLane.Destination
                    : StreamingWorkLane.NonDestination);
            StreamingWorkAdmission admission = meter.TryReserve(
                admissionCost,
                "completion-admission");
            if (admission == StreamingWorkAdmission.Yielded)
                return;

            try
            {
                if (!_completionSource.TryRead(
                        out LandblockStreamResult? consumed)
                    || !ReferenceEquals(result, consumed))
                {
                    throw new InvalidOperationException(
                        "The single-consumer completion source changed between peek and read.");
                }

                if (!stale)
                {
                    _completionQueue.Enqueue(new StreamingQueuedCompletion(
                        result,
                        estimate,
                        priority,
                        priority == StreamingCompletionPriority.Destination
                            ? ActiveRevealGeneration
                            : 0L,
                        result.Generation,
                        checked(_nextCompletionSequence++),
                        Stopwatch.GetTimestamp()));
                }
                meter.Complete();
            }
            catch
            {
                meter.Fail();
                throw;
            }
        }
    }

    private StreamingCompletionPriority ClassifyCompletion(
        LandblockStreamResult result)
    {
        if (result is LandblockStreamResult.Loaded
                or LandblockStreamResult.Promoted
            && IsDestinationWork(result.LandblockId))
        {
            return StreamingCompletionPriority.Destination;
        }

        return result switch
        {
            LandblockStreamResult.Failed
                or LandblockStreamResult.WorkerCrashed =>
                StreamingCompletionPriority.Control,
            LandblockStreamResult.Unloaded =>
                StreamingCompletionPriority.Unload,
            LandblockStreamResult.Promoted
                or LandblockStreamResult.Loaded
                {
                    Tier: LandblockStreamTier.Near,
                } =>
                StreamingCompletionPriority.Near,
            _ => StreamingCompletionPriority.Far,
        };
    }

    private static LandblockPublicationAdvance RunSimpleResultOperation(
        StreamingWorkMeter meter,
        string stage,
        bool ensureProgress,
        Action operation)
    {
        StreamingWorkAdmission admission = meter.TryReserve(
            default,
            stage,
            ensureProgress);
        if (admission == StreamingWorkAdmission.Yielded)
            return new LandblockPublicationAdvance(false, false);

        try
        {
            operation();
            meter.Complete();
            return new LandblockPublicationAdvance(true, true);
        }
        catch
        {
            meter.Fail();
            throw;
        }
    }

    /// <summary>
    /// Apply a single <see cref="LandblockStreamResult"/> with the full side-
    /// effects: terrain upload, GPU state, and the re-hydration callback.
    /// All priority queues route through this one publication path.
    /// </summary>
    private LandblockPublicationAdvance ApplyResult(
        StreamingQueuedCompletion work,
        StreamingWorkMeter meter,
        bool ensureProgress)
    {
        LandblockStreamResult result = work.Result;
        if (_presentation.HasPendingPublication(result))
        {
            LandblockPublicationAdvance resumed =
                _presentation.ResumePublication(
                    result,
                    meter,
                    ensureProgress);
            if (resumed.Completed)
                ReconcileCompletedPendingPublication(result.LandblockId);
            return resumed;
        }

        if (IsStaleGeneration(result))
        {
            // A worker can finish one old load/unload after a hard recenter.
            // Landblock id membership cannot distinguish overlapping windows;
            // generation is the logical streaming incarnation boundary.
            return RunSimpleResultOperation(
                meter,
                "execute-stale-generation",
                ensureProgress,
                static () => { });
        }

        if (result is LandblockStreamResult.Unloaded
            && _region?.TryGetDesiredTier(result.LandblockId, out _) == true)
        {
            // Normal recenter does not advance the hard-window generation. A
            // rapid away->back can therefore re-own an id while its same-
            // generation unload is already in the worker outbox. Current
            // region ownership wins; the old unload must not tear it down.
            return RunSimpleResultOperation(
                meter,
                "execute-reowned-unload",
                ensureProgress,
                static () => { });
        }

        if (result is LandblockStreamResult.Loaded
                or LandblockStreamResult.Promoted)
        {
            if (_region is null
                || !_region.TryGetDesiredTier(result.LandblockId, out var desiredTier))
            {
                // A hard recenter/collapse can leave one worker job already in
                // flight. Its completion belongs to the old region and must not
                // resurrect a landblock the new StreamingRegion never owns.
                return RunSimpleResultOperation(
                    meter,
                    "execute-undesired-load",
                    ensureProgress,
                    static () => { });
            }

            bool isNearCompletion = result is LandblockStreamResult.Promoted
                or LandblockStreamResult.Loaded { Tier: LandblockStreamTier.Near };
            if (isNearCompletion
                && _state.IsNearTier(result.LandblockId)
                && !_presentation.HasPendingPublication(result))
            {
                // Normal recenter can enqueue a second high-priority Near job
                // while an earlier one is already in flight. The streamer
                // supersedes Far/Unload work but intentionally does not dedupe
                // high-priority jobs. Publication is idempotent at this owner:
                // never append statics, replay defaults, or reapply the complete
                // Near transaction to an already-Near landblock.
                return RunSimpleResultOperation(
                    meter,
                    "execute-duplicate-near",
                    ensureProgress,
                    static () => { });
            }
            if (desiredTier == LandblockStreamTier.Far && isNearCompletion)
            {
                // Recenter can demote a Near job before its first completion.
                // StreamingRegion already considers that landblock Far, so it
                // deliberately emits no second LoadFar job. If no terrain is
                // resident yet, publish the completed heightmap/mesh as a Far
                // result while discarding its entity and EnvCell layers. This
                // is the same payload a fresh LoadFar would have produced and
                // closes the in-flight Near -> desired Far lifecycle without a
                // duplicate worker read or a permanent terrain hole.
                if (!_state.IsLoaded(result.LandblockId))
                {
                    switch (result)
                    {
                        case LandblockStreamResult.Loaded loaded:
                            return _presentation.PublishAsFar(
                                loaded,
                                loaded.Build,
                                loaded.MeshData,
                                meter,
                                ensureProgress);
                        case LandblockStreamResult.Promoted promoted:
                            return _presentation.PublishAsFar(
                                promoted,
                                promoted.Build,
                                promoted.MeshData,
                                meter,
                                ensureProgress);
                    }
                }
                return RunSimpleResultOperation(
                    meter,
                    "execute-already-far",
                    ensureProgress,
                    static () => { });
            }
        }

        switch (result)
        {
            case LandblockStreamResult.Loaded loaded:
                if (loaded.Tier == LandblockStreamTier.Far
                    && _state.IsNearTier(loaded.LandblockId))
                {
                    // This Far completion was queued before a newer Near load
                    // or promotion. Applying it would erase the entity/cell
                    // layer that now owns the landblock.
                    return RunSimpleResultOperation(
                        meter,
                        "execute-stale-far-tier",
                        ensureProgress,
                        static () => { });
                }
                return _presentation.PublishLoaded(
                    loaded,
                    work.Estimate,
                    meter,
                    ensureProgress);
            case LandblockStreamResult.Promoted promoted:
                // PromoteToNear carries a complete build and mesh because the
                // streamer deliberately lets it supersede a queued LoadFar. If
                // that Far job never started, publish this as the real Near
                // landblock; if the base is already resident, merge only the
                // Near layer so existing live projections retain identity.
                return _presentation.PublishPromoted(
                    promoted,
                    mergeIntoExistingLandblock:
                        _state.IsLoaded(promoted.LandblockId),
                    work.Estimate,
                    meter,
                    ensureProgress);
            case LandblockStreamResult.Unloaded unloaded:
                return RunSimpleResultOperation(
                    meter,
                    "execute-unload",
                    ensureProgress,
                    () => _presentation.EnqueueFullRetirement(
                        unloaded.LandblockId));
            case LandblockStreamResult.Failed failed:
                return RunSimpleResultOperation(
                    meter,
                    "execute-load-failure",
                    ensureProgress,
                    () => Console.WriteLine(
                        $"streaming: load failed for 0x{failed.LandblockId:X8}: {failed.Error}"));
            case LandblockStreamResult.WorkerCrashed crashed:
                return RunSimpleResultOperation(
                    meter,
                    "execute-worker-crash",
                    ensureProgress,
                    () => Console.WriteLine(
                        $"streaming: worker CRASHED: {crashed.Error}"));
            default:
                throw new InvalidOperationException(
                    $"Unsupported streaming result {result.GetType().Name}.");
        }
    }

    private void ReconcileCompletedPendingPublication(uint landblockId)
    {
        if (_region is null
            || !_region.TryGetDesiredTier(landblockId, out LandblockStreamTier desiredTier))
        {
            if (_state.IsLoaded(landblockId))
                _presentation.EnqueueFullRetirement(landblockId);
            return;
        }

        if (desiredTier == LandblockStreamTier.Far
            && _state.IsNearTier(landblockId))
        {
            _presentation.EnqueueNearLayerRetirement(landblockId);
        }
    }

    private bool IsStaleGeneration(LandblockStreamResult result) =>
        result is not LandblockStreamResult.WorkerCrashed
        && result.Generation != _generation;

    private bool IsPublicationBlockedByRetirement(LandblockStreamResult result) =>
        result is LandblockStreamResult.Loaded or LandblockStreamResult.Promoted
        && _presentation.IsRetirementPending(result.LandblockId);

    /// <summary>
    /// Returns the landblock id associated with <paramref name="result"/>.
    /// For <see cref="LandblockStreamResult.WorkerCrashed"/> this is 0 by
    /// convention (not tied to a specific landblock).
    /// </summary>
    private static uint ResultLandblockId(LandblockStreamResult result) => result.LandblockId;
}