perf #418: publish landblocks under the meter, not one per streaming tick

Phase-1 measurement (new [publish-timing]/[stream-tick] probe surfaces,
ACDREAM_PROBE_REVEAL_TIMING=1) refuted the ~31 ms-per-admission
hypothesis: the hold runs at ~64 fps with the streaming tick at ~32 Hz,
the whole 625-block window costs only ~500 ms of publication CPU (far
blocks ~0.17 ms, near 2-43 ms), and steady state showed ZERO meter
yields with ~0.22 ms of the 2 ms budget used - yet exactly one block
published per tick against a ~400-deep completion queue.

The real limiter: Runtime's collision-generation activation is a
deliberate two-poll transaction (the first
TryAcquireCollisionPrefixMutationPermission poll parks residents and
refuses by design), and LandblockPresentationPipeline.Advance's metered
arm returned Completed=false on ANY nonterminal commit, which
DrainAndApply treats as end-of-frame. One landblock per 32 Hz tick =
the flat 32/s, with the authored budget ~90% idle.

Fix: the metered arm now uses the same Runtime-owned gate the unmetered
arm and the synchronous CompletePublication API always used
(CanContinueMutationSynchronously). The second poll runs in the same
frame under the same meter, so the unchanged 2 ms elapsed-time ceiling
is now genuinely the authoritative per-frame bound; with any real debt
(live residents parked mid-game, pending withdrawals, dispatch backlog)
publication defers to the next frame exactly as before. No budget
values change, no reveal-gate/readiness change, and the streamed result
is byte-identical - only the frame scheduling of identical operations.

Measured A/B (this binary, two runs): totalMs 12689 / 12734 vs baseline
26728/27395/27503; loaded slope 32/s -> bursts of 100-360/s, 625/625 in
~6-7 s vs ~23 s. The remaining ~12.7 s floor is fully attributed in
docs/ISSUES.md: ~8 s of real budgeted readiness work plus retail's
authored tunnel exit (TunnelContinue 2-5 s + two 1 s fades, golden
constants), so the <12 s acceptance needs a lead decision on the
hold-time budget, not another hidden limiter.

New regression pin:
MeteredLoaded_NonterminalCommitWithoutDebt_CompletesInOneMeteredAdvance.
Gates: Release build 0 errors; App tests 5576/3 skips/0 failed;
Runtime tests 1756/0.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/AcDream.App/Streaming/LandblockPresentationPipeline.cs

@@ -64,6 +64,13 @@ public sealed class LandblockPresentationPipeline
         public bool SpatialPresentationCommitted;
         public bool EnvCellReplayCommitted;
         public bool LiveRecoveryCommitted;
+
+        /// <summary>
+        /// #418 measurement only (<c>ACDREAM_PROBE_REVEAL_TIMING=1</c>): the
+        /// per-stage wall-clock attribution of this one publication. Null
+        /// whenever the probe env is unset.
+        /// </summary>
+        public PublicationStageTimings? Timing;
     }
 
     private readonly Action<LandblockBuild, LandblockMeshData>?
@@ -460,6 +467,7 @@ public sealed class LandblockPresentationPipeline
                         Origin: completedBuild.Origin),
                     meshData),
                 Tier = LandblockStreamTier.Far,
+                Timing = PublicationTimingProbe.CreateTimings(),
             };
             _publications.Add(acceptedResult, transaction);
         }
@@ -509,6 +517,7 @@ public sealed class LandblockPresentationPipeline
                 LandblockId = farLandblock.LandblockId,
                 Cost = LandblockStreamResultCost.Estimate(farBuild, meshData),
                 Tier = LandblockStreamTier.Far,
+                Timing = PublicationTimingProbe.CreateTimings(),
             };
             _publications.Add(acceptedResult, transaction);
         }
@@ -553,6 +562,7 @@ public sealed class LandblockPresentationPipeline
             Cost = estimate
                 ?? LandblockStreamResultCost.Estimate(build, meshData),
             Tier = tier,
+            Timing = PublicationTimingProbe.CreateTimings(),
         };
         _publications.Add(result, created);
         return created;
@@ -565,6 +575,26 @@ public sealed class LandblockPresentationPipeline
         bool ensureProgress)
     {
         bool progressed = false;
+        void RunTimed(string stage, Action operation)
+        {
+            if (transaction.Timing is not { } timing)
+            {
+                operation();
+                return;
+            }
+
+            long start = System.Diagnostics.Stopwatch.GetTimestamp();
+            try
+            {
+                operation();
+            }
+            finally
+            {
+                timing.Add(
+                    stage,
+                    System.Diagnostics.Stopwatch.GetTimestamp() - start);
+            }
+        }
         bool TryRun(
             StreamingWorkCost cost,
             string stage,
@@ -572,7 +602,7 @@ public sealed class LandblockPresentationPipeline
         {
             if (meter is null)
             {
-                operation();
+                RunTimed(stage, operation);
                 progressed = true;
                 return true;
             }
@@ -585,7 +615,7 @@ public sealed class LandblockPresentationPipeline
                 return false;
             try
             {
-                operation();
+                RunTimed(stage, operation);
                 meter.Complete();
                 progressed = true;
                 return true;
@@ -810,8 +840,33 @@ public sealed class LandblockPresentationPipeline
                     }
                     if (transaction.PhysicsPublication.RuntimeMutationPending)
                     {
-                        if (meter is not null
-                            || !_physicsPublisher
+                        // #418: Runtime's collision-generation activation is a
+                        // deliberate two-poll transaction — the first
+                        // CommitCollisionGeneration poll closes the prefix
+                        // quiescence boundary (parking any affected residents)
+                        // and returns nonterminal; only a later poll may
+                        // consume mutation permission. The metered path used
+                        // to defer that later poll to the NEXT frame
+                        // unconditionally, which serialized the whole
+                        // completion drain to ONE landblock per streaming tick
+                        // (DrainAndApply breaks on an incomplete publication):
+                        // the measured login hold published exactly 32
+                        // blocks/s at a 32 Hz tick rate with ZERO meter
+                        // yields and ~0.2 ms of its 2 ms budget used. Runtime
+                        // explicitly supports consuming that finite suffix
+                        // synchronously when it reports no cross-cutting debt
+                        // (CanContinueMutationSynchronously — the same gate
+                        // the unmetered path and the synchronous
+                        // CompletePublication API already use), so the
+                        // metered path now polls again within the same frame
+                        // under the SAME meter: every extra poll still passes
+                        // TryRun/TryReserve, so the elapsed-time budget
+                        // remains the authoritative per-frame bound. With any
+                        // real debt (live residents parked mid-game, pending
+                        // withdrawals, dispatch backlog) Runtime keeps
+                        // reporting nonterminal-with-debt and this defers to
+                        // the next frame exactly as before.
+                        if (!_physicsPublisher
                                 .CanContinueMutationSynchronously())
                         {
                             return new LandblockPublicationAdvance(
@@ -976,6 +1031,13 @@ public sealed class LandblockPresentationPipeline
         }
 
         _publications.Remove(result);
+        if (transaction.Timing is { } completedTiming)
+        {
+            PublicationTimingProbe.EmitPublication(
+                transaction.LandblockId,
+                transaction.Kind.ToString(),
+                completedTiming);
+        }
         return new LandblockPublicationAdvance(true, progressed);
     }
 }

src/AcDream.App/Streaming/PublicationTimingProbe.cs

@@ -0,0 +1,227 @@
+using System.Diagnostics;
+using System.Text;
+
+namespace AcDream.App.Streaming;
+
+/// <summary>
+/// Per-stage wall-clock accumulator for ONE landblock publication transaction
+/// (#418 measurement). Created only when
+/// <see cref="StreamingDiagnostics.ProbeRevealTiming"/> is set; a null
+/// instance on the transaction means the probe is off and costs one null
+/// check per stage operation.
+/// </summary>
+internal sealed class PublicationStageTimings
+{
+    private readonly Dictionary<string, (long Ticks, int Count)> _stages = new();
+    private long _totalTicks;
+
+    public long TotalTicks => _totalTicks;
+
+    public void Add(string stage, long ticks)
+    {
+        if (ticks < 0)
+            ticks = 0;
+        _totalTicks += ticks;
+        _stages[stage] = _stages.TryGetValue(stage, out (long Ticks, int Count) prior)
+            ? (prior.Ticks + ticks, prior.Count + 1)
+            : (ticks, 1);
+    }
+
+    public IReadOnlyDictionary<string, (long Ticks, int Count)> Stages => _stages;
+}
+
+/// <summary>
+/// <c>ACDREAM_PROBE_REVEAL_TIMING=1</c> sibling surface for issue #418:
+/// attributes ONE admitted landblock's update-thread publication cost per
+/// stage. Emits:
+///
+/// <list type="bullet">
+/// <item><c>[publish-timing] lb=…</c> — one line per completed landblock
+/// publication with the total milliseconds and a per-stage
+/// <c>name:ms/count</c> breakdown (stages sorted by cost).</item>
+/// <item><c>[publish-timing] CUMULATIVE …</c> — every 64 publications, the
+/// running per-stage totals across all completed publications.</item>
+/// </list>
+///
+/// It also accumulates the per-frame streaming-tick elapsed time so
+/// <see cref="RevealTimingProbe"/>'s 1 Hz progress line can report how much
+/// of each wall-clock second the streaming tick consumed (with the observed
+/// render-frame count, this verifies or refutes the ~31 ms/frame
+/// prediction). Diagnostic-only: no behavior change, never active unless the
+/// probe env is set. All access happens on the window thread (the streaming
+/// controller and render frame graph share it).
+/// </summary>
+internal static class PublicationTimingProbe
+{
+    private const int CumulativeEmitInterval = 64;
+
+    private static readonly Dictionary<string, (long Ticks, int Count)>
+        s_cumulativeStages = new();
+    private static readonly Dictionary<string, int> s_tickWindowYieldReasons =
+        new();
+    private static long s_cumulativeTicks;
+    private static int s_publications;
+    private static double s_tickWindowSumMs;
+    private static double s_tickWindowMaxMs;
+    private static int s_tickWindowCount;
+    private static int s_tickWindowYields;
+    private static int s_tickWindowOperations;
+    private static int s_tickWindowAdmissions;
+    private static int s_lastWorkerBacklog;
+    private static int s_lastQueuedCompletions;
+
+    public static bool Enabled => StreamingDiagnostics.ProbeRevealTiming;
+
+    /// <summary>Per-transaction accumulator, or null when the probe is off.</summary>
+    public static PublicationStageTimings? CreateTimings() =>
+        Enabled ? new PublicationStageTimings() : null;
+
+    /// <summary>
+    /// Reports one completed landblock publication and folds its stages into
+    /// the cumulative rollup.
+    /// </summary>
+    public static void EmitPublication(
+        uint landblockId,
+        string kind,
+        PublicationStageTimings timings)
+    {
+        s_publications++;
+        s_cumulativeTicks += timings.TotalTicks;
+        var line = new StringBuilder(256);
+        line.Append("[publish-timing] lb=0x")
+            .Append(landblockId.ToString("X8"))
+            .Append(" kind=").Append(kind)
+            .Append(" totalMs=")
+            .Append(ToMs(timings.TotalTicks).ToString("F2"))
+            .Append(" stages=");
+        AppendStagesByCost(line, timings.Stages);
+        Console.WriteLine(line.ToString());
+
+        foreach ((string stage, (long ticks, int count)) in timings.Stages)
+        {
+            s_cumulativeStages[stage] = s_cumulativeStages.TryGetValue(
+                stage,
+                out (long Ticks, int Count) prior)
+                ? (prior.Ticks + ticks, prior.Count + count)
+                : (ticks, count);
+        }
+
+        if (s_publications % CumulativeEmitInterval == 0)
+            EmitCumulative();
+    }
+
+    private static void EmitCumulative()
+    {
+        var line = new StringBuilder(256);
+        line.Append("[publish-timing] CUMULATIVE landblocks=")
+            .Append(s_publications)
+            .Append(" totalMs=")
+            .Append(ToMs(s_cumulativeTicks).ToString("F0"))
+            .Append(" stages=");
+        AppendStagesByCost(line, s_cumulativeStages);
+        Console.WriteLine(line.ToString());
+    }
+
+    private static void AppendStagesByCost(
+        StringBuilder line,
+        IReadOnlyDictionary<string, (long Ticks, int Count)> stages)
+    {
+        bool first = true;
+        foreach ((string stage, (long ticks, int count)) in
+            stages.OrderByDescending(static pair => pair.Value.Ticks))
+        {
+            if (!first)
+                line.Append(',');
+            first = false;
+            line.Append(TrimStagePrefix(stage))
+                .Append(':')
+                .Append(ToMs(ticks).ToString("F2"))
+                .Append('/')
+                .Append(count);
+        }
+    }
+
+    /// <summary>
+    /// Every pipeline stage name starts with <c>publication-</c>; dropping the
+    /// shared prefix keeps the per-landblock line readable.
+    /// </summary>
+    private static string TrimStagePrefix(string stage) =>
+        stage.StartsWith("publication-", StringComparison.Ordinal)
+            ? stage["publication-".Length..]
+            : stage;
+
+    /// <summary>
+    /// Records one <see cref="StreamingController.Tick"/> frame into the
+    /// current 1 Hz reporting window: elapsed time, meter operation/yield
+    /// counts, admissions, the reason the meter last refused work, and the
+    /// end-of-tick worker/queue backlogs.
+    /// </summary>
+    public static void ObserveStreamingTick(
+        in StreamingWorkMeterSnapshot snapshot,
+        int workerBacklog,
+        int queuedCompletions)
+    {
+        s_tickWindowCount++;
+        s_tickWindowSumMs += snapshot.ElapsedMilliseconds;
+        if (snapshot.ElapsedMilliseconds > s_tickWindowMaxMs)
+            s_tickWindowMaxMs = snapshot.ElapsedMilliseconds;
+        s_tickWindowYields += snapshot.YieldCount;
+        s_tickWindowOperations += snapshot.Operations;
+        s_tickWindowAdmissions += snapshot.Used.CompletionAdmissions;
+        if (snapshot.YieldCount > 0
+            && snapshot.LastLimit != StreamingWorkLimit.None
+            && snapshot.LastStage is { } stage)
+        {
+            string reason = $"{stage}/{snapshot.LastLimit}";
+            s_tickWindowYieldReasons[reason] =
+                s_tickWindowYieldReasons.TryGetValue(reason, out int prior)
+                    ? prior + 1
+                    : 1;
+        }
+        s_lastWorkerBacklog = workerBacklog;
+        s_lastQueuedCompletions = queuedCompletions;
+    }
+
+    /// <summary>
+    /// Emits and resets the current streaming-tick reporting window as one
+    /// <c>[stream-tick]</c> line. Called by <see cref="RevealTimingProbe"/>'s
+    /// 1 Hz progress path.
+    /// </summary>
+    public static void EmitStreamingTickWindow()
+    {
+        var line = new StringBuilder(192);
+        line.Append("[stream-tick] ticks=").Append(s_tickWindowCount)
+            .Append(" sumMs=").Append(s_tickWindowSumMs.ToString("F1"))
+            .Append(" maxMs=").Append(s_tickWindowMaxMs.ToString("F1"))
+            .Append(" ops=").Append(s_tickWindowOperations)
+            .Append(" yields=").Append(s_tickWindowYields)
+            .Append(" admissions=").Append(s_tickWindowAdmissions)
+            .Append(" workerBacklog=").Append(s_lastWorkerBacklog)
+            .Append(" queuedCompletions=").Append(s_lastQueuedCompletions)
+            .Append(" yieldReasons=");
+        bool first = true;
+        foreach ((string reason, int count) in
+            s_tickWindowYieldReasons.OrderByDescending(
+                static pair => pair.Value))
+        {
+            if (!first)
+                line.Append(',');
+            first = false;
+            line.Append(reason).Append('x').Append(count);
+        }
+        if (first)
+            line.Append("none");
+        Console.WriteLine(line.ToString());
+
+        s_tickWindowCount = 0;
+        s_tickWindowSumMs = 0;
+        s_tickWindowMaxMs = 0;
+        s_tickWindowYields = 0;
+        s_tickWindowOperations = 0;
+        s_tickWindowAdmissions = 0;
+        s_tickWindowYieldReasons.Clear();
+    }
+
+    private static double ToMs(long stopwatchTicks) =>
+        stopwatchTicks * 1000.0 / Stopwatch.Frequency;
+}

src/AcDream.App/Streaming/RevealTimingProbe.cs

@@ -47,6 +47,7 @@ internal sealed class RevealTimingProbe
     private long _gateReadyMs = -1;
     private long _materializedMs = -1;
     private long _lastProgressMs;
+    private int _framesSinceProgress;
 
     public RevealTimingProbe(Func<int>? loadedLandblockCount) =>
         _loadedLandblockCount = loadedLandblockCount;
@@ -88,6 +89,7 @@ internal sealed class RevealTimingProbe
         if (_generation == 0 || portal.Generation != _generation)
             return;
 
+        _framesSinceProgress++;
         long elapsed = _clock.ElapsedMilliseconds;
         if (!_render && readiness.IsRenderNeighborhoodReady)
         {
@@ -136,13 +138,21 @@ internal sealed class RevealTimingProbe
         if (!_summarized && elapsed - _lastProgressMs >= 1000)
         {
             _lastProgressMs = elapsed;
+            // #418: frames = render frames observed since the previous
+            // progress line (the hold's effective frame rate). The paired
+            // [stream-tick] line carries the same window's streaming-tick
+            // wall clock, meter operation/yield/admission counts, backlogs,
+            // and the reasons the meter refused work.
             Console.WriteLine(
                 $"[reveal-timing] elapsedMs={elapsed} "
                 + $"render={(_render ? 1 : 0)} "
                 + $"composites={(_composites ? 1 : 0)} "
                 + $"collision={(_collision ? 1 : 0)} "
                 + $"loaded={_loadedLandblockCount?.Invoke() ?? -1}"
-                + $"/{_windowLandblocks}");
+                + $"/{_windowLandblocks} "
+                + $"frames={_framesSinceProgress}");
+            PublicationTimingProbe.EmitStreamingTickWindow();
+            _framesSinceProgress = 0;
         }
     }
 

src/AcDream.App/Streaming/StreamingController.cs

@@ -778,6 +778,13 @@ public sealed class StreamingController
             ObserveWorkLifetime(snapshot);
             _lastWorkMeter = snapshot;
             _activeWorkMeter = null;
+            if (PublicationTimingProbe.Enabled)
+            {
+                PublicationTimingProbe.ObserveStreamingTick(
+                    snapshot,
+                    _completionSource.BacklogCount,
+                    _completionQueue.Count);
+            }
         }
     }