fix(strict): carry the eager walk's validated capacities into the traced refresh

jaccpot/runtime/_interaction_cache.py

@@ -952,9 +952,29 @@ def _build_dual_tree_artifacts_split_strict_streamed(
     traversal_config: Optional[DualTreeTraversalConfig],
     pair_policy: Optional[PairPolicy],
     policy_state: Optional[AdaptivePolicyState],
+    capacity_report: Optional[Callable[[dict], None]] = None,
+    max_neighbors_per_leaf_override: Optional[int] = None,
 ) -> _DualTreeArtifacts:
     """Strict static fast-lane: single compact shared far+near build call.
 
+    ``capacity_report`` receives, after a successful build, the capacities the
+    walk actually ran with plus (eager only) what it observed: the queue the
+    yggdrax ladder settled on, the compact far-pair cap, the far-pair count, the
+    longest neighbour row and the total edge count.  The fused traced refresh
+    cannot grow capacities -- under ``jit`` the overflow flags are tracers and
+    yggdrax returns the truncated result -- so it must be told what the eager
+    prepare needed.  ``max_neighbors_per_leaf_override`` is how it is told, and it
+    only ever RAISES the cap.  (Found 2026-09-06: at N=200k / leaf 256 the preset
+    cap of 256 neighbours per leaf against rows of 781 cut 85 % of the near field
+    out of every step after the first, with no diagnostic firing.)
+
+    Only the neighbour cap is carried over, because it is the only one that fails
+    SILENTLY: the compact far-pair cap is checked under tracing too
+    (``_raw_to_compact_far_pairs`` raises through a debug callback), so a
+    too-small one is already loud, and raising a cap the caller named in
+    ``JACCPOT_STATIC_STRICT_FUSED_COMPACT_FAR_PAIR_CAP`` would turn a deliberate
+    memory bound into a no-op.
+
     This path intentionally avoids generic split-builder host branching and
     callback plumbing. It is valid only for streamed compact far-pairs with no
     dense/grouped/interactions payload requests.
@@ -987,6 +1007,12 @@ def _build_dual_tree_artifacts_split_strict_streamed(
         MAC alone.
     policy_state : Optional[AdaptivePolicyState]
         State the pair policy reads. Meaningless without ``pair_policy``.
+    capacity_report : Optional[Callable[[dict], None]]
+        Called once after a successful build with the capacities the walk ran
+        with (queue, far-pair cap, neighbour cap) and, eagerly, what it observed
+        (far-pair count, longest neighbour row, total edges). ``None`` skips it.
+    max_neighbors_per_leaf_override : Optional[int]
+        Floor for the per-leaf neighbour cap; only ever raises it.
 
     Returns
     -------
@@ -1021,6 +1047,10 @@ def _build_dual_tree_artifacts_split_strict_streamed(
         process_block_resolved = (
             None if pair_process_block is None else int(pair_process_block)
         )
+    if max_neighbors_per_leaf_override is not None:
+        max_neighbors_per_leaf = max(
+            int(max_neighbors_per_leaf), int(max_neighbors_per_leaf_override)
+        )
 
     flat_compact_enabled = os.environ.get(
         "JACCPOT_STATIC_STRICT_FUSED_FLAT_COMPACT_FAR_PAIRS", "1"
@@ -1081,6 +1111,20 @@ def _build_dual_tree_artifacts_split_strict_streamed(
     attempt_queue = max_pair_queue_resolved
     attempt_far_cap = compact_far_pair_capacity
     grew: list[str] = []
+    # The eager yggdrax ladder may settle on a LARGER queue than requested; the
+    # "success" event is the only place that number is reported, and the traced
+    # refresh needs it (see ``capacity_report``).
+    ladder_success: dict = {}
+
+    def _capture_ladder(event: DualTreeRetryEvent) -> None:
+        if str(event.status) == "success":
+            ladder_success.update(
+                queue_capacity=int(event.queue_capacity),
+                interaction_capacity=int(event.interaction_capacity),
+                far_pair_count=int(event.far_pair_count),
+                near_pair_count=int(event.near_pair_count),
+            )
+
     for attempt in range(_STRICT_STREAMED_RETRY_ATTEMPTS):
         try:
             (
@@ -1097,7 +1141,7 @@ def _build_dual_tree_artifacts_split_strict_streamed(
                 max_pair_queue=attempt_queue,
                 process_block=process_block_resolved,
                 traversal_config=None,
-                retry_logger=None,
+                retry_logger=_capture_ladder,
                 timing_callback=None,
                 compact_far_pair_capacity=attempt_far_cap,
                 pair_policy=pair_policy,
@@ -1145,6 +1189,48 @@ def _build_dual_tree_artifacts_split_strict_streamed(
                 attempt_far_cap = grown
     else:  # pragma: no cover - the loop always breaks or raises
         raise RuntimeError("strict streamed dual-tree walk did not run")
+    if capacity_report is not None:
+        counts_arr = getattr(neighbor_list, "counts", None)
+        offsets_arr = getattr(neighbor_list, "offsets", None)
+        fp_count = getattr(compact_far_pairs, "far_pair_count", None)
+        traced = isinstance(counts_arr, Tracer) or isinstance(fp_count, Tracer)
+        report = dict(
+            traced=bool(traced),
+            max_pair_queue_requested=(
+                None if attempt_queue is None else int(attempt_queue)
+            ),
+            queue_capacity=int(
+                ladder_success.get(
+                    "queue_capacity",
+                    (
+                        _STRICT_STREAMED_QUEUE_FLOOR
+                        if attempt_queue is None
+                        else int(attempt_queue)
+                    ),
+                )
+            ),
+            compact_far_pair_capacity=(
+                None if attempt_far_cap is None else int(attempt_far_cap)
+            ),
+            max_neighbors_per_leaf_used=int(max_neighbors_per_leaf),
+            grew=list(grew),
+        )
+        if not traced:
+            try:
+                report["far_pair_count"] = (
+                    int(fp_count)
+                    if fp_count is not None
+                    else int(compact_far_pairs.sources.shape[0])
+                )
+                report["max_neighbors_observed"] = (
+                    int(jnp.max(counts_arr)) if int(counts_arr.shape[0]) else 0
+                )
+                report["total_neighbors"] = (
+                    int(offsets_arr[-1]) if int(offsets_arr.shape[0]) else 0
+                )
+            except Exception:  # pragma: no cover - diagnostics only
+                pass
+        capacity_report(report)
     return _DualTreeArtifacts(
         interactions=None,
         neighbor_list=neighbor_list,
@@ -2169,6 +2255,8 @@ def _build_dual_tree_artifacts(
     jit_traversal: bool = True,
     timing_callback: Optional[Callable[[str, float], None]] = None,
     planner_hint: Optional[_RefreshDualPlannerHint] = None,
+    strict_capacity_report: Optional[Callable[[dict], None]] = None,
+    strict_max_neighbors_per_leaf_override: Optional[int] = None,
 ) -> tuple[_DualTreeArtifacts, Optional[_InteractionCacheEntry]]:
     """Construct or reuse dual-tree traversal products for a tree.
 
@@ -2240,6 +2328,10 @@ def _build_dual_tree_artifacts(
     planner_hint : Optional[_RefreshDualPlannerHint]
         Hint from a previous refresh, letting the planner skip work whose answer
         is already known.
+    strict_capacity_report : Optional[Callable[[dict], None]]
+        Forwarded to the strict streamed builder as ``capacity_report``.
+    strict_max_neighbors_per_leaf_override : Optional[int]
+        Forwarded to the strict streamed builder; only ever raises the cap.
 
     Returns
     -------
@@ -2316,6 +2408,10 @@ def _build_dual_tree_artifacts(
                     traversal_config=traversal_config,
                     pair_policy=pair_policy,
                     policy_state=policy_state,
+                    capacity_report=strict_capacity_report,
+                    max_neighbors_per_leaf_override=(
+                        strict_max_neighbors_per_leaf_override
+                    ),
                 )
                 if strict_streamed_split
                 else _build_dual_tree_artifacts_split(

jaccpot/runtime/fmm_prepare.py

@@ -1464,10 +1464,20 @@ def _record_dual_artifact_substage(name: str, elapsed: float) -> None:
                 max_leaf_size=int(tree_artifacts.leaf_cap),
             )
         )
+        (
+            runtime_traversal_config,
+            strict_nbr_override,
+            _strict_capacity_report,
+        ) = self._strict_fused_capacity_handoff(
+            runtime_traversal_config=runtime_traversal_config,
+            suppress_host_side_effects=suppress_host_side_effects,
+        )
         dual_artifacts, cache_entry = _build_dual_tree_artifacts(
             tree_artifacts.tree,
             tree_artifacts.upward.geometry,
             geometry_factory=geometry_factory,
+            strict_capacity_report=_strict_capacity_report,
+            strict_max_neighbors_per_leaf_override=strict_nbr_override,
             theta=theta_val,
             mac_type=mac_type_val,
             dehnen_radius_scale=dehnen_radius_scale,
@@ -3323,6 +3333,99 @@ def _prepare_downward_with_artifacts(
             p_gears=p_gears,
         )
 
+    def _strict_fused_capacity_handoff(
+        self,
+        *,
+        runtime_traversal_config: Optional[DualTreeTraversalConfig],
+        suppress_host_side_effects: bool,
+    ) -> tuple[
+        Optional[DualTreeTraversalConfig],
+        Optional[int],
+        Callable[[dict], None],
+    ]:
+        """Carry the eager walk's validated capacities into the traced refresh.
+
+        Eagerly, yggdrax's retry ladder grows the pair queue and the per-leaf
+        neighbour cap until the walk fits, so the eager state is exact whatever
+        the preset caps say.  Under the fused velocity-Verlet scan the SAME walk
+        runs traced: the overflow flags are tracers, the ladder has one attempt,
+        and yggdrax returns the truncated result.  Measured 2026-09-06 at
+        N=200k, leaf 256: the preset caps (queue 65536, 256 neighbours per leaf)
+        against rows of up to 781 cut the near field to 15 % from step 2 on --
+        relative force error ~60 % at theta 0.6, 5.8 % at theta 1.0 -- with no
+        diagnostic.
+
+        So the eager pass records what it needed (``_strict_fused_validated_caps``,
+        via the returned report callback) and the traced pass raises its caps to
+        cover that with headroom: the neighbour cap to 1.5x the observed longest
+        row (next power of two), the queue to 2x the ladder's answer.  The queue
+        cannot be verified after the fact -- its overflow flag never leaves the
+        trace -- hence the larger margin; the neighbour cap IS re-checked inside
+        the scan by ``strict_run_v2`` from ``_strict_fused_traced_caps``, which
+        the same callback records on the traced build.
+
+        The compact far-pair cap is deliberately NOT carried over.  It is checked
+        under tracing already (``_raw_to_compact_far_pairs`` raises through a
+        debug callback), so a too-small one fails loudly instead of truncating
+        silently, and it is set explicitly by
+        ``JACCPOT_STATIC_STRICT_FUSED_COMPACT_FAR_PAIR_CAP`` -- widening a cap the
+        caller named would make a deliberate memory bound a no-op.
+
+        Parameters
+        ----------
+        runtime_traversal_config : Optional[DualTreeTraversalConfig]
+            Capacities resolved for this build; widened (never shrunk) on the
+            traced path when the eager pass recorded a larger validated queue.
+        suppress_host_side_effects : bool
+            ``True`` on the traced hot path (the refresh inside the compiled
+            scan); selects which of the two attributes the report lands in.
+
+        Returns
+        -------
+        Optional[DualTreeTraversalConfig]
+            The traversal config to build with.
+        Optional[int]
+            ``max_neighbors_per_leaf`` floor for the strict streamed builder.
+        Callable[[dict], None]
+            Report callback that stores the builder's capacities on the engine.
+        """
+
+        def _pow2_ceil(value: int) -> int:
+            value = max(1, int(value))
+            return 1 << (value - 1).bit_length()
+
+        nbr_override: Optional[int] = None
+        validated = getattr(self, "_strict_fused_validated_caps", None)
+        if bool(suppress_host_side_effects) and isinstance(validated, dict):
+            observed_rows = validated.get("max_neighbors_observed")
+            if observed_rows is not None:
+                nbr_override = _pow2_ceil(int(1.5 * int(observed_rows)) + 1)
+            validated_queue = validated.get("queue_capacity")
+            if validated_queue is not None and runtime_traversal_config is not None:
+                widened_queue = max(
+                    int(runtime_traversal_config.max_pair_queue),
+                    _pow2_ceil(2 * int(validated_queue)),
+                )
+                if widened_queue != int(runtime_traversal_config.max_pair_queue):
+                    runtime_traversal_config = DualTreeTraversalConfig(
+                        max_pair_queue=int(widened_queue),
+                        process_block=int(runtime_traversal_config.process_block),
+                        max_interactions_per_node=int(
+                            runtime_traversal_config.max_interactions_per_node
+                        ),
+                        max_neighbors_per_leaf=int(
+                            runtime_traversal_config.max_neighbors_per_leaf
+                        ),
+                    )
+
+        def _report(report: dict) -> None:
+            if bool(report.get("traced")):
+                self._strict_fused_traced_caps = dict(report)
+            else:
+                self._strict_fused_validated_caps = dict(report)
+
+        return runtime_traversal_config, nbr_override, _report
+
     def _prepare_state_dual_and_downward_strict_streamed_fast(
         self,
         *,
@@ -3385,10 +3488,20 @@ def _prepare_state_dual_and_downward_strict_streamed_fast(
                 max_leaf_size=int(tree_artifacts.leaf_cap),
             )
         )
+        (
+            runtime_traversal_config,
+            strict_nbr_override,
+            _strict_capacity_report,
+        ) = self._strict_fused_capacity_handoff(
+            runtime_traversal_config=runtime_traversal_config,
+            suppress_host_side_effects=suppress_host_side_effects,
+        )
         dual_artifacts, cache_entry = _build_dual_tree_artifacts(
             tree_artifacts.tree,
             tree_artifacts.upward.geometry,
             geometry_factory=geometry_factory,
+            strict_capacity_report=_strict_capacity_report,
+            strict_max_neighbors_per_leaf_override=strict_nbr_override,
             theta=theta_val,
             mac_type=mac_type_val,
             dehnen_radius_scale=dehnen_radius_scale,

jaccpot/runtime/fmm_strict_run.py

@@ -819,20 +819,45 @@ def _evaluate_self(prepared_in: PreparedStateLike, state_in: Array) -> Array:
         def _static_target_block_capacity_ok(
             prepared_in: PreparedStateLike,
         ) -> Array:
+            offsets = jnp.asarray(prepared_in.neighbor_list.offsets)
+            counts = offsets[1:] - offsets[:-1]
+            ok = jnp.asarray(True)
             padded = getattr(
                 prepared_in,
                 "nearfield_target_block_source_leaf_ids_padded",
                 None,
             )
-            if padded is None:
-                return jnp.asarray(True)
-            padded_arr = jnp.asarray(padded)
-            if padded_arr.ndim != 3 or int(padded_arr.shape[1]) == 0:
-                return jnp.asarray(True)
-            offsets = jnp.asarray(prepared_in.neighbor_list.offsets)
-            counts = offsets[1:] - offsets[:-1]
-            capacity = int(padded_arr.shape[1]) * int(padded_arr.shape[2])
-            return jnp.all(counts <= jnp.asarray(capacity, dtype=counts.dtype))
+            if padded is not None:
+                padded_arr = jnp.asarray(padded)
+                if padded_arr.ndim == 3 and int(padded_arr.shape[1]) > 0:
+                    capacity = int(padded_arr.shape[1]) * int(padded_arr.shape[2])
+                    ok = ok & jnp.all(
+                        counts <= jnp.asarray(capacity, dtype=counts.dtype)
+                    )
+            # Traversal-capacity saturation guard.  The traced refresh walk runs
+            # with fixed caps, and a neighbour row that fills its cap -- or a
+            # far-pair list that fills its buffer -- means entries were dropped
+            # and the force is wrong.  The neighbour cap is the one that fails
+            # SILENTLY: yggdrax's near-overflow flag is a tracer under jit and
+            # nothing reads it, which is the defect this guard exists for.  The
+            # far-pair cap already raises through a debug callback in
+            # ``_raw_to_compact_far_pairs``, so its arm here is a second, cheap
+            # line of defence rather than the only one.  Both caps are host
+            # constants recorded while the refresh traced
+            # (``_strict_fused_traced_caps``), so this is a static comparison.
+            traced_caps = getattr(self, "_strict_fused_traced_caps", None)
+            if isinstance(traced_caps, dict):
+                nbr_cap = traced_caps.get("max_neighbors_per_leaf_used")
+                if nbr_cap is not None and int(counts.shape[0]) > 0:
+                    ok = ok & (
+                        jnp.max(counts) < jnp.asarray(int(nbr_cap), counts.dtype)
+                    )
+                far_cap = traced_caps.get("compact_far_pair_capacity")
+                far_pairs = getattr(prepared_in, "compact_far_pairs", None)
+                far_count = getattr(far_pairs, "far_pair_count", None)
+                if far_cap is not None and far_count is not None:
+                    ok = ok & (jnp.asarray(far_count) < jnp.asarray(int(far_cap)))
+            return ok
 
         def _refresh_and_evaluate_endpoint(
             prepared_in: PreparedStateLike,
@@ -1017,11 +1042,17 @@ def _skip(_):
                         "JACCPOT_LARGE_N_STATIC_TARGET_BLOCKS_MAX_PER_LEAF",
                         "32",
                     )
+                    traced_caps = getattr(self, "_strict_fused_traced_caps", None) or {}
                     raise RuntimeError(
-                        "fused payload static target-block cap exceeded during "
-                        "compiled velocity-Verlet scan: max_blocks_per_leaf="
-                        f"{max_blocks}. Increase "
-                        "JACCPOT_LARGE_N_STATIC_TARGET_BLOCKS_MAX_PER_LEAF."
+                        "a fixed capacity saturated inside the compiled velocity-Verlet "
+                        "scan, so the refreshed interaction lists are truncated and the "
+                        "forces from that step on are wrong. Checked: static target-block "
+                        f"cap (max_blocks_per_leaf={max_blocks}), traced neighbour cap "
+                        f"({traced_caps.get('max_neighbors_per_leaf_used')} per leaf) and "
+                        f"compact far-pair cap ({traced_caps.get('compact_far_pair_capacity')}). "
+                        "Raise JACCPOT_LARGE_N_STATIC_TARGET_BLOCKS_MAX_PER_LEAF, pass "
+                        "jaccpot.TraversalOverrides(max_neighbors_per_leaf=...), or raise "
+                        "JACCPOT_STATIC_STRICT_FUSED_COMPACT_FAR_PAIR_CAP."
                     )
             except Exception as exc:
                 if bool(